就是那种打字机的效果，令用户感觉他/她问完问题以后，就有答案立刻出现。虽然整个答案完全出来也得好几分钟，但那种心理感受会不大一样。

在Streamlit上实现起来会略有一点点麻烦，通常streamlit上显示一段输出是完整的，比如

import streamlit as st
full_string="hello world"
st.markdown(full_string)
st.markdown("other content")

显示的结果是

hello world
other content

如果你想让hello world一个一个字母逐一出现，而又不破坏下方的other content的内容，方法应该是这样的：

  • 先设定一个st.empty()的占位
  • 然后让这个占位不断输出逐渐加长的字符
import streamlit as st
full_string="hello world"
display_string=""
show_string=st.empty()
st.markdown("other content")	
for c in full_string:
	display_string+=c
	show_string.markdown(diskplay_string)

这样就可以看到打字机效果了。

现在考虑GPT回复时的调用，我这里使用封装好的langchain。langchain中使用了一种叫callbacks的技术，当GPT设定为流式回复时，每返回一个字符，就会调用一次callback。那么要实现在streamlit里显示流式回复就和上面的动作类似：

  • 设定一个st.empty()占位
  • 写一个callback，让这个占位不断输出逐渐加长的字符
class StreamDisplayHandler(BaseCallbackHandler):
    def __init__(self, container, initial_text="", display_method='markdown'):
        self.container = container
        self.text = initial_text
        self.display_method = display_method

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.text += token

        display_function = getattr(self.container, self.display_method, None)
        if display_function is not None:
            display_function(self.text)
        else:
            raise ValueError(f"Invalid display_method: {self.display_method}")

    def on_llm_end(self, response, **kwargs) -> None:
        self.text=""

这个callback在初始化的时候应当把占位符container传递给这个StreamDisplayHandler。其他的动作都和上面的简单打字机示例是一样的。

这里有一个小小的奇技淫巧display_function = getattr(self.container, self.display_method, None) 这句话的意义是我不确定打算使用streamlit哪种显示文本的方式，可以是st.markdown，也可以是其他，比如st.write，那么就可以通过一个字符串来说明，再去看st里面是不是有这个显示文本的方法，如果有，就显示，没有就报错。

这个callback的调用

chat_box = st.empty()
display_handler = StreamDisplayHandler(
    chat_box,
    display_method='write')
chat = ChatOpenAI(
    max_tokens=100, streaming=True,
    callbacks=display_handler)

更进一步，既然每次GPT回复一个字符都会调用，那么就可以用来处理一些“实时”的东西，比如几乎同步的语音，当然语音如果一个字一个字地说会不好听，于是可以累积成一句话以后一次播报，方法和上面也基本相同。详细的代码请参考我写的gist