如何在LangChain中为Gradio界面实现Agent响应流式输出？

要让Agent的响应流式输出到Gradio聊天界面，核心是利用LangChain的stream方法获取逐块输出，并通过Gradio的生成器函数（yield）实时更新界面。以下是具体修改步骤和代码：

1. 核心修改思路 #

将原同步调用chain.run()的逻辑，替换为流式迭代chain.stream()的输出结果，通过yield实时推送更新后的聊天历史到Gradio界面。

2. 完整修改后的代码 #

from langchain.callbacks.base import BaseCallbackHandler
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import AzureChatOpenAI
from langchain.agents import initialize_agent
from langchain.agents.conversational_chat.base import ConversationalChatAgent
from threading import Lock
from typing import Optional, List, Tuple

# 自定义回调（可选，用于额外捕获token，保留终端输出可保留原回调）
class GradioStreamingCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        self.tokens = []
    
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.tokens.append(token)

# Loading the LLM
def load_llm():
    return AzureChatOpenAI(
        temperature=hparams["temperature"],
        top_p=hparams["top_p"],
        max_tokens=hparams["max_tokens"],
        presence_penalty=hparams["presence_penalty"],
        frequency_penalty=hparams["freq_penaulty"],
        streaming=True, 
        callback_manager=CallbackManager([StreamingStdOutCallbackHandler(), GradioStreamingCallbackHandler()]), 
        verbose=True,
        model_name=hparams["model"],
        deployment_name=models_dict[hparams["model"]],
        )

# Loading the agent
def load_chain(memory, sys_msg, llm):
    agent_chain = initialize_agent(tools, 
                                   llm, 
                                   agent="conversational-react-description", 
                                   verbose=True, 
                                   memory=memory, 
                                   agent_kwargs={"added_prompt": sys_msg},
                                   streaming=True, 
                                   )
    return agent_chain

# Creating the chatbot to be used in Gradio.
class ChatWrapper:

    def __init__(self, sys_msg):
        self.lock = Lock()
        self.memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True,)
        self.chain = load_chain(self.memory, sys_msg, load_llm())
        self.sysmsg = sys_msg
    
    def __call__(
        self, api_key: str, inp: str, history: Optional[List[Tuple[str, str]]], chain: Optional[ConversationalChatAgent]
    ):
        self.lock.acquire()
        try:
            history = history or []
            # 初始化当前输出，先将用户输入加入历史
            current_output = ""
            history.append((inp, current_output))
            
            # 流式迭代Agent的输出块
            for chunk in self.chain.stream({"input": inp}):
                # 提取新增的响应内容（不同Agent的chunk结构可能略有差异，需按需调整）
                if "output" in chunk:
                    current_output += chunk["output"]
                    # 更新历史中的最新响应条目
                    history[-1] = (inp, current_output)
                    # 实时推送更新后的历史，Gradio会自动刷新界面
                    yield history, history
            
            # 确保最终完整输出同步到历史中
            history[-1] = (inp, current_output)
            yield history, history
        except Exception as e:
            raise e
        finally:
            self.lock.release()

3. 关键细节说明 #

• agent_chain.stream()：LangChain Agent的流式输出方法，返回生成器，每次迭代返回包含部分响应的chunk。
• 生成器函数（yield）：Gradio原生支持生成器作为接口函数，每次yield的结果会实时更新聊天界面。
• chunk结构适配：不同类型的Agent返回的chunk结构可能不同，conversational-react-description类型的Agent会返回含output键的字典，若调试时发现结构不符，可打印chunk内容后调整提取逻辑。

内容的提问来源于stack exchange，提问作者MRF