如何从LangGraph多智能体图的stream输出中提取AI生成的响应
如何从LangGraph多智能体图的stream输出中提取AI生成的响应
我完全懂你的烦恼——LangGraph的stream输出结构确实有点“杂乱无章”,尤其是带嵌套子图和人类在环的时候,要从一堆元组、嵌套字典里精准揪出AI的自然语言回复,确实容易懵。不过别担心,咱们拆解一下结构,写个小函数就能搞定。
先理清楚输出的核心结构
你贴的示例已经很清楚了:graph.stream返回的每个事件都是一个元组,格式大概是:
((节点标识元组), {组件名称: 包含messages的结构})
其中关键的是messages列表,里面混着AIMessage(AI生成的消息)、HumanMessage(人类输入)、ToolMessage(工具返回)等类型。我们要抓的就是**AIMessage里有实际自然语言内容的条目**(那些空content但带function_call的,一般是AI的工具调用指令,如果你不需要可以过滤掉)。
提取AI回复的具体实现
我写了一个通用的提取函数,能递归遍历所有嵌套的消息结构,自动筛选出符合要求的AI回复:
from langchain_core.messages import AIMessage def extract_ai_responses(events): ai_responses = [] for event in events: # 每个事件的第二个元素是存储消息的核心数据 event_content = event[1] # 递归遍历所有层级,找AIMessage def traverse_data(data): if isinstance(data, dict): for val in data.values(): traverse_data(val) elif isinstance(data, list): for item in data: traverse_data(item) # 找到AIMessage,且content非空(排除仅工具调用的空回复) elif isinstance(data, AIMessage) and data.content.strip(): ai_responses.append({ "agent_source": event[0] or "supervisor", # 标记来自哪个节点 "response": data.content }) traverse_data(event_content) return ai_responses # 用你的events变量调用函数 ai_responses = extract_ai_responses(events) # 打印结果 for idx, resp in enumerate(ai_responses, 1): print(f"AI回复 {idx}(来自{resp['agent_source']}):\n{resp['response']}\n")
针对你的示例输出,运行结果会是:
AI回复 1(来自('agent:<id>', 'tool_manager:<id>')): Good evening! Here's your plan for today. AI回复 2(来自('agent:<id>', 'tool_manager:<id>')): Here's today's detailed plan: - Breakfast: Skipped - Lunch: Chicken salad - Dinner: Bhuna Ghost Would you like to make any changes?
灵活调整的小技巧
- 如果你需要保留带工具调用的空content AIMessage,只要把函数里的
and data.content.strip()去掉就行。 - 要是想区分AI的工具调用指令和自然语言回复,可以额外检查
data.additional_kwargs.get('function_call')是否存在。
这样不管你的子图嵌套多深,都能把所有AI生成的有效响应捞出来啦~
备注:内容来源于stack exchange,提问作者Sukumar Ganesan
相关产品推荐
相关产品推荐

