You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何从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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.13 19:08:04