如何让LangChain ZapierNLA Agent仅流式传输Final Answer

核心思路 #

LangChain的Agent流式输出时会触发多类回调事件，只需过滤掉中间思考步骤的输出，仅保留Final Answer相关内容即可解决问题。

具体实现方法 #

方法1：重写自定义Handler的token过滤逻辑 #

继承StreamingStdOutCallbackHandler后，仅在检测到进入Final Answer阶段时才发送token：

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

class CustomAgentHandler(StreamingStdOutCallbackHandler):
    def __init__(self):
        super().__init__()
        self.in_final_answer = False
        self.final_answer_prefix = "Final Answer:"  # 匹配Agent输出的Final Answer前缀

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        # 触发Final Answer阶段的判断
        if self.final_answer_prefix in token:
            self.in_final_answer = True
            # 截取前缀后的有效内容
            token = token.split(self.final_answer_prefix)[1]
        
        if self.in_final_answer:
            # 替换为你的WebSocket流式发送逻辑
            print(token, end="", flush=True)

方法2：自定义Agent输出解析器 #

替换默认解析器，仅提取Final Answer部分：

from langchain.agents import AgentOutputParser
from langchain.schema import AgentAction, AgentFinish

class FinalAnswerOnlyParser(AgentOutputParser):
    def parse(self, llm_output: str) -> AgentAction | AgentFinish:
        if "Final Answer:" in llm_output:
            return AgentFinish(
                return_values={"output": llm_output.split("Final Answer:")[-1].strip()},
                log=llm_output
            )
        # 无Final Answer时按默认逻辑处理（可选）
        return super().parse(llm_output)

# 初始化Agent时指定自定义解析器
agent = initialize_agent(
    tools=zapier_tools,
    llm=your_llm_instance,
    agent=AgentType.ZAPIER_NLA,
    output_parser=FinalAnswerOnlyParser(),
    verbose=False,
    callbacks=[CustomAgentHandler()]
)

方法3：监听AgentFinish事件发送结果 #

通过on_agent_finish回调直接获取最终结果，跳过中间思考步骤：

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentFinish

class FinalAnswerCallbackHandler(BaseCallbackHandler):
    def on_agent_finish(self, finish: AgentFinish, **kwargs) -> None:
        # 发送Final Answer到WebSocket
        self.send_to_websocket(finish.return_values["output"])

# 初始化Agent时添加该Handler
agent = initialize_agent(
    tools=zapier_tools,
    llm=your_llm_instance,
    agent=AgentType.ZAPIER_NLA,
    verbose=False,
    callbacks=[FinalAnswerCallbackHandler()]
)

关键注意事项 #

• 必须将Agent的verbose参数设为False，避免冗余日志输出。
• 不同LangChain版本的Agent输出格式可能有差异，需根据实际输出调整Final Answer的匹配前缀。
• WebSocket传输时需确保token发送逻辑适配连接状态与编码要求。

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