如何在等待LangChain Agent React最终响应时实时获取其中间步骤？

问题背景 #

使用LangChain库中的AgentReact时，需要在等待最终响应的过程中实时获取并显示中间交互步骤，但当前代码仅在整个流程完成后才返回交互记录与最终结果，无法满足实时展示的需求。

现有代码 #

def question_answer(question: str, agents: AgentExecutor):
    
    list_logs_result = []
    answer = agents.invoke({'input': prompt_templates_100.format(question=question)}, )
    interactions = format_interactions(answer)

    
    for interaction in interactions:
        list_logs_result.append({"log": interaction["log"], "result": interaction["result"]})

    return answer["output"], list_logs_result

解决方案 #

LangChain的AgentExecutor支持通过**回调函数（Callback Handler）**实时捕获中间步骤，无需等待整个流程结束。具体实现步骤如下：

1. 自定义回调类 #

继承BaseCallbackHandler，重写关键方法来处理Agent的每一步动作和工具调用结果，实现实时输出或收集中间日志：

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

class RealTimeAgentCallback(BaseCallbackHandler):
    def __init__(self):
        self.interactions = []

    def on_agent_action(self, action: AgentAction, **kwargs) -> None:
        # 捕获Agent思考后的动作（比如调用工具的决策）
        log_entry = {"log": action.log, "result": f"准备调用工具: {action.tool}，参数: {action.tool_input}"}
        self.interactions.append(log_entry)
        # 这里可以直接打印实时显示，或者发送到前端等
        print(f"实时日志: {log_entry['log']}")
        print(f"实时结果: {log_entry['result']}")

    def on_tool_end(self, output: str, **kwargs) -> None:
        # 捕获工具调用后的返回结果
        log_entry = {"log": "工具调用完成", "result": output}
        self.interactions.append(log_entry)
        print(f"实时日志: {log_entry['log']}")
        print(f"实时结果: {log_entry['result']}")

    def on_agent_finish(self, finish: AgentFinish, **kwargs) -> None:
        # 捕获Agent最终完成的结果
        log_entry = {"log": "Agent流程结束", "result": finish.return_values['output']}
        self.interactions.append(log_entry)
        print(f"实时日志: {log_entry['log']}")
        print(f"实时结果: {log_entry['result']}")

2. 修改调用逻辑 #

在调用AgentExecutor.invoke时传入自定义的回调实例，同时可以实时收集或展示中间步骤：

def question_answer(question: str, agents: AgentExecutor):
    # 初始化回调实例
    callback = RealTimeAgentCallback()
    
    # 调用Agent时传入回调
    answer = agents.invoke(
        {'input': prompt_templates_100.format(question=question)},
        callbacks=[callback]
    )
    
    # 最终返回结果和所有交互记录（也可以在回调中实时处理，无需等到最后收集）
    return answer["output"], callback.interactions

关键说明 #

• on_agent_action：在Agent决定调用工具时触发，可获取Agent的思考日志和工具调用信息
• on_tool_end：在工具调用完成返回结果时触发，可获取工具输出
• on_agent_finish：在Agent完成所有步骤返回最终结果时触发
• 如果需要将实时日志发送到前端（比如Web应用），可以在回调方法中加入WebSocket推送、队列通知等逻辑，替代示例中的print

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