根据Stream来看LangGraph过程

这是是LangChain官网的带checkpoint的“获取当地天气”的案例，这里头原本是做了结构化输出，但是这里我用agent.stream，来探求Agent Graph是怎么跑的：

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from dataclasses import dataclass
from langchain.agents.structured_output import ToolStrategy
from langgraph.checkpoint.memory import InMemorySaver

model = ChatOpenAI(
    model="Pro/MiniMaxAI/MiniMax-M2.5",
    api_key="",
    base_url = "https://api.siliconflow.cn/v1"
)

checkpointer = InMemorySaver()

SYSTEM_PROMPT = """You are an expert weather forecaster, who speaks in puns.

You have access to two tools:

- get_weather_for_location: use this to get the weather for a specific location
- get_user_location: use this to get the user's location

If a user asks you for the weather, make sure you know the location. If you can tell from the question that they mean wherever they are, use the get_user_location tool to find their location."""

@tool
def get_weather_for_location(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

@dataclass
class Context:
    """Custom runtime context schema."""
    user_id: str

# We use a dataclass here, but Pydantic models are also supported.
@dataclass
class ResponseFormat:
    """Response schema for the agent."""
    # A punny response (always required)
    punny_response: str
    # Any interesting information about the weather if available
    weather_conditions: str | None = None

@tool
def get_user_location(runtime: ToolRuntime[Context]) -> str:
    """Retrieve user information based on user ID."""
    user_id = runtime.context.user_id
    return "Florida" if user_id == "1" else "SF"

agent = create_agent(
    model,
    system_prompt=SYSTEM_PROMPT,
    tools=[get_user_location, get_weather_for_location],
    context_schema=Context,
    response_format=ToolStrategy(ResponseFormat),
    checkpointer=checkpointer
)

# `thread_id` is a unique identifier for a given conversation.
config = {"configurable": {"thread_id": "1"}}

# # Run the agent
# response = agent.invoke(
#     {"messages": [{"role": "user", "content": "what is the weather outside?"}]},
#     config=config,
#     context=Context(user_id="1")
# )

# print(response['structured_response'])

# # Note that we can continue the conversation using the same `thread_id`.
# response = agent.invoke(
#     {"messages": [{"role": "user", "content": "Are you sure?"}]},
#     config=config,
#     context=Context(user_id="1")
# )

# print(response['structured_response'])

for event in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather outside?"}]},
    config=config,
    context=Context(user_id="1")
):
    print(event)

for event in agent.stream(
    {"messages": [{"role": "user", "content": "Are you sure?"}]},
    config=config,
    context=Context(user_id="1")
):
    print(event)

再跑完之后，我们根据输出，探求了以下的问题：

Agent Graph是怎么跑的？

把流程抽象成图是这样的：

    ┌─────────┐
    │  model  │
    └────┬────┘
         │
  (if tool_calls)
         ▼
    ┌─────────┐
    │  tools  │
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │  model  │
    └────┬────┘
         │
  (no more tools)
         ▼
       END

我们再根据日志json来看每一步。由于用户问了两轮，先看

第一轮提问

1. model

{
    'model': {
        'messages': [AIMessage(content = '', additional_kwargs = {
            'refusal': None
        },
        response_metadata = {
            'token_usage': {
                'completion_tokens': 15,
                'prompt_tokens': 400,
                'total_tokens': 415,
                'completion_tokens_details': {
                    'accepted_prediction_tokens': None,
                    'audio_tokens': None,
                    'reasoning_tokens': 0,
                    'rejected_prediction_tokens': None
                },
                'prompt_tokens_details': {
                    'audio_tokens': None,
                    'cached_tokens': 0
                },
                'prompt_cache_hit_tokens': 0,
                'prompt_cache_miss_tokens': 400
            },
            'model_provider': 'openai',
            'model_name': 'Pro/MiniMaxAI/MiniMax-M2.5',
            'system_fingerprint': '',
            'id': '019c97b05c5a90021b52d88003bfc3ea',
            'finish_reason': 'tool_calls',
            'logprobs': None
        },
        id = 'lc_run--019c97b0-5a8e-7d30-b08a-20ec9eb41ad9-0', tool_calls = [{
            'name': 'get_user_location',
            'args': {},
            'id': '019c97b05f0724176eeeece72c29fa94',
            'type': 'tool_call'
        }], invalid_tool_calls = [], usage_metadata = {
            'input_tokens': 400,
            'output_tokens': 15,
            'total_tokens': 415,
            'input_token_details': {
                'cache_read': 0
            },
            'output_token_details': {
                'reasoning': 0
            }
        })]
    }
}

这一步表示Agent

  • 进入了model结点
  • content是空，因为这里模型决定只调用工具
  • 决定调用get_user_location这个方法
  • 结束原因是准备调用tool_call

2.tools

{
    'tools': {
        'messages': [ToolMessage(content = 'Florida', name = 'get_user_location', id = '89791976-f63c-41fc-887d-a89db8e87ace', tool_call_id = '019c97b05f0724176eeeece72c29fa94')]
    }
}
  • tools 节点执行函数
  • 得到结果Florida
  • 构造ToolMessage

3.model

{
    'model': {
        'messages': [AIMessage(content = '', additional_kwargs = {
            'refusal': None
        },
        response_metadata = {
            'token_usage': {
                'completion_tokens': 31,
                'prompt_tokens': 431,
                'total_tokens': 462,
                'completion_tokens_details': {
                    'accepted_prediction_tokens': None,
                    'audio_tokens': None,
                    'reasoning_tokens': 0,
                    'rejected_prediction_tokens': None
                },
                'prompt_tokens_details': {
                    'audio_tokens': None,
                    'cached_tokens': 0
                },
                'prompt_cache_hit_tokens': 0,
                'prompt_cache_miss_tokens': 431
            },
            'model_provider': 'openai',
            'model_name': 'Pro/MiniMaxAI/MiniMax-M2.5',
            'system_fingerprint': '',
            'id': '019c97b05f8ba84ac2fc0190f7ee608d',
            'finish_reason': 'tool_calls',
            'logprobs': None
        },
        id = 'lc_run--019c97b0-5e8e-70a3-8ed2-17b1dcc80351-0', tool_calls = [{
            'name': 'get_weather_for_location',
            'args': {
                'city': 'Florida'
            },
            'id': '019c97b062b9fe28aa226986c139f96c',
            'type': 'tool_call'
        }], invalid_tool_calls = [], usage_metadata = {
            'input_tokens': 431,
            'output_tokens': 31,
            'total_tokens': 462,
            'input_token_details': {
                'cache_read': 0
            },
            'output_token_details': {
                'reasoning': 0
            }
        })]
    }
}

和之前的model结点环节一样：

  • content是空，因为这里模型决定只调用工具
  • 决定调用get_weather_for_location这个方法
  • 结束原因是准备调用tool_call

4.tools

{
    'tools': {
        'messages': [ToolMessage(content = "It's always sunny in Florida!", name = 'get_weather_for_location', id = 'e3dd6a37-f836-4d5d-b3fc-b8fb2800abba', tool_call_id = '019c97b062b9fe28aa226986c139f96c')]
    }
}
  • tools 节点执行函数
  • 得到结果It's always sunny in Florida!
  • 构造ToolMessage

5.model

{
    'model': {
        'messages': [AIMessage(content = '', additional_kwargs = {
            'refusal': None
        },
        response_metadata = {
            'token_usage': {
                'completion_tokens': 69,
                'prompt_tokens': 478,
                'total_tokens': 547,
                'completion_tokens_details': {
                    'accepted_prediction_tokens': None,
                    'audio_tokens': None,
                    'reasoning_tokens': 0,
                    'rejected_prediction_tokens': None
                },
                'prompt_tokens_details': {
                    'audio_tokens': None,
                    'cached_tokens': 384
                },
                'prompt_cache_hit_tokens': 384,
                'prompt_cache_miss_tokens': 94
            },
            'model_provider': 'openai',
            'model_name': 'Pro/MiniMaxAI/MiniMax-M2.5',
            'system_fingerprint': '',
            'id': '019c97b0639e53330722b10e55e7cea5',
            'finish_reason': 'tool_calls',
            'logprobs': None
        },
        id = 'lc_run--019c97b0-620b-7271-aed2-0fb845a15b8f-0', tool_calls = [{
            'name': 'ResponseFormat',
            'args': {
                'punny_response': "Well folks, it's always sunny in Florida! Looks like the weather is having a ball - talk about a bright outlook! Don't forget your sunglasses, or you might get a little burnt by all this rays-istence. Stay cool and keep on shining!"
            },
            'id': '019c97b069d7c4e81de3233f2868e860',
            'type': 'tool_call'
        }], invalid_tool_calls = [], usage_metadata = {
            'input_tokens': 478,
            'output_tokens': 69,
            'total_tokens': 547,
            'input_token_details': {
                'cache_read': 384
            },
            'output_token_details': {
                'reasoning': 0
            }
        }), ToolMessage(content = 'Returning structured response: ResponseFormat(punny_response="Well folks, it\'s always sunny in Florida! Looks like the weather is having a ball - talk about a bright outlook! Don\'t forget your sunglasses, or you might get a little burnt by all this rays-istence. Stay cool and keep on shining!", weather_conditions=None)', name = 'ResponseFormat', id = 'd55a7412-389e-447d-ad6b-12c648849611', tool_call_id = '019c97b069d7c4e81de3233f2868e860')],
        'structured_response': ResponseFormat(punny_response = "Well folks, it's always sunny in Florida! Looks like the weather is having a ball - talk about a bright outlook! Don't forget your sunglasses, or you might get a little burnt by all this rays-istence. Stay cool and keep on shining!", weather_conditions = None)
    }
}

这里是关键，它没有直接输出文本，因为我用了：

response_format=ToolStrategy(ResponseFormat)

所以结构化输出被当成“一个特殊工具”，模型实际上在“调用 ResponseFormat 这个 tool

好，我们接着你这个「逐节点拆解」风格往后写 👇
下面是 第二轮：Are you sure? 的完整执行流。

第二轮提问

这里我们再次调用：

for event in agent.stream(
    {"messages": [{"role": "user", "content": "Are you sure?"}]},
    config=config,
    context=Context(user_id="1")
):
    print(event)

因为 user_id="1" 相同 + 有 checkpointer，LangGraph 会：

  • 读取上一轮完整 state
  • 把 "Are you sure?" 追加进去
  • 重新从入口执行 graph

6. model

{
  "model": {
    "messages": [
      AIMessage(
        content = "",
        additional_kwargs = { "refusal": None },
        response_metadata = {
          "token_usage": {
            "completion_tokens": 31,
            "prompt_tokens": 640,
            "total_tokens": 671,
            "completion_tokens_details": {
              "accepted_prediction_tokens": null,
              "audio_tokens": null,
              "reasoning_tokens": 0,
              "rejected_prediction_tokens": null
            },
            "prompt_tokens_details": {
              "audio_tokens": null,
              "cached_tokens": 384
            },
            "prompt_cache_hit_tokens": 384,
            "prompt_cache_miss_tokens": 256
          },
          "model_provider": "openai",
          "model_name": "Pro/MiniMaxAI/MiniMax-M2.5",
          "system_fingerprint": "",
          "id": "019c97b06aa309bf320ec03268fea806",
          "finish_reason": "tool_calls",
          "logprobs": null
        },
        tool_calls = [{
          "name": "get_weather_for_location",
          "args": { "city": "Florida" },
          "id": "019c97b06f7f658367f4ab18f1b08d44",
          "type": "tool_call"
        }],
        invalid_tool_calls = [],
        usage_metadata = {
          "input_tokens": 640,
          "output_tokens": 31,
          "total_tokens": 671
        }
      )
    ]
  }
}
  • 进入 model 结点
  • content 仍然是空（因为模型决定调用工具）
  • 它根据历史上下文判断：用户在质疑刚才 Florida 的天气
  • 所以再次调用： 
    get_weather_for_location(city="Florida")
  • finish_reason = tool_calls

7. tools

{
  "tools": {
    "messages": [
      ToolMessage(
        content = "It's always sunny in Florida!",
        name = "get_weather_for_location",
        id = "e1a638c6-649c-4ae1-883a-a2c05566873f",
        tool_call_id = "019c97b06f7f658367f4ab18f1b08d44"
      )
    ]
  }
}

这一节点说明：

  • tools 结点执行函数

  • 再次得到结果：

    It's always sunny in Florida!

  • 构造 ToolMessage

  • 把结果写回 state

8. model

{
  "model": {
    "messages": [
      AIMessage(
        content = "",
        additional_kwargs = { "refusal": None },
        response_metadata = {
          "token_usage": {
            "completion_tokens": 68,
            "prompt_tokens": 687,
            "total_tokens": 755
          },
          "model_provider": "openai",
          "model_name": "Pro/MiniMaxAI/MiniMax-M2.5",
          "finish_reason": "tool_calls"
        },
        tool_calls = [{
          "name": "ResponseFormat",
          "args": {
            "punny_response": "I'm absolutely sure - the weather report is sunny! I promise I'm not just blowing hot air here. Florida's living up to its sunny reputation - it's a real ray of hope! ☀️"
          },
          "id": "019c97b07974ccdf3968d9f81699c78d",
          "type": "tool_call"
        }]
      ),
      ToolMessage(
        content = "Returning structured response: ResponseFormat(...)",
        name = "ResponseFormat",
        id = "d0b80db9-d306-4754-b22d-022a50d55bb7",
        tool_call_id = "019c97b07974ccdf3968d9f81699c78d"
      )
    ],
    "structured_response": ResponseFormat(
      punny_response = "I'm absolutely sure - the weather report is sunny! I promise I'm not just blowing hot air here. Florida's living up to its sunny reputation - it's a real ray of hope! ☀️",
      weather_conditions = None
    )
  }
}

和第一轮一样：

  • 模型没有直接输出文本
  • 而是调用 ResponseFormat
  • 因为你设置了：
response_format = ToolStrategy(ResponseFormat)

所以：

结构化输出被当成“最后一个特殊工具”

模型本质是在：

call ResponseFormat(...)

然后由 ToolStrategy 解析成：

structured_response = ResponseFormat(...)

🔥 现在把完整两轮总结成一条执行链

第一轮：

model → tools → model → tools → model(ResponseFormat)

第二轮：

load old state
        ↓
append new user message
        ↓
model → tools → model(ResponseFormat)

二、Tool Message 是怎么插入的？

看上面我打印出来的stream输出，可以发现如下结构体：

ToolMessage(content = 'Florida', name = 'get_user_location', id = '89791976-f63c-41fc-887d-a89db8e87ace', tool_call_id = '019c97b05f0724176eeeece72c29fa94')

它本质上等价于

{
  role: "tool",
  name: "get_user_location",
  content: "Florida"
}

LangGraph 做了：

  1. 执行工具函数
  2. 构造 ToolMessage
  3. 插入到 state["messages"] 里
  4. 再次调用 model

所以模型第二次看到的 prompt 实际是：

system: ...
user: what is the weather outside?
assistant: (tool_call get_user_location)
tool: Florida

这一步的本质就是tool作为一个role，进入了这个对话历史记录中，使得 LLM 可以看到完整对话后继续推理。

三、中间状态长什么样？

看到的stream输出大致如下：

{'model': {...}}
{'tools': {...}}
{'model': {...}}
{'tools': {...}}
{'model': {...}}

其实是：LangGraph state 的局部 diff。这句话怎么理解呢？说的准确一点就是：

agent.stream() 里每次 yield 出来的 event
不是完整 state，而是「这一步对 state 做了什么修改」。

真实完整 state 长这样（抽象化）：

{
  "messages": [
      SystemMessage(...),
      HumanMessage("what is the weather outside?"),

      AIMessage(tool_call=get_user_location),
      ToolMessage("Florida"),

      AIMessage(tool_call=get_weather_for_location),
      ToolMessage("It's always sunny in Florida!"),

      AIMessage(tool_call=ResponseFormat)
  ],
  "structured_response": ...
}

可以看到这个完整的state是一个大字典，所谓局部diff，就是我输出的比如：

{
  "model": {
    "messages": [AIMessage(... tool_calls=get_user_location)]
  }
}

他不是一个大字典，而知识一条message，意思是往这个完整state里添加一条message。

每个节点执行：

  • 读取 state
  • 修改 state
  • 返回新的 state

stream 只是把每次 state 变化吐出来。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成，浏览时请结合常识与多方信息审慎甄别。
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容