LangGraph：用 StateGraph 把手搓 while 画成图

前两篇先把这一套拆开：LangGraph 是运行时，create_agent 是日常入口；再划清了什么时候不必上框架。这篇从底下看起，把手搓系列里那个 while 画成图。

从零手搓里，循环是：带 tools 问模型，有 tool_calls 就执行、回灌，再问，直到没有 tool_calls 给出终答。天气仍走 Open-Meteo，拆成 search_location 与 get_current_weather，坐标不能编。订房对上手搓第二篇，四个工具挂同一张图。目标是先看西安天气，再订 10 月 1 日套房。模型看见的还是 schema，只是这份 JSON 从 @tool 推出来，不是手写。这篇还没接持久化和人确认，跑完就结束。

Table of contents

一、两个节点对上那个 while

循环只做两件事：问模型；模型要工具时，执行并回灌。还要工具就再问一轮，不要了，问到的就是终答。手搓把这两步写在同一个 for 里，这边拆成两个节点。

agent 只问，对照手搓里的 create(..., tools=...)，不查天气。tools 只执行并按 id 回灌，对照 dispatch，不再问。节点之间传一份 State，这篇只有 messages。

先把图里的四个词对上，后面几篇就不会把它们混在一起：

这张图每个 superstep 只有一个节点，所以看起来仍像串行的 while：先执行 agent，合并它返回的 AIMessage；下一个 superstep 执行 tools，合并 ToolMessage；再回到 agent。等后面用 Send 同时展开多座城市时，一个 superstep 才会出现多份并行任务，但「先执行本轮、合并 State、再进入下一轮」的节奏不变。

二、图上这几个调用

自己画图时，就是下面这段：

# StateGraph 用来把那两件事画成图，按「问 → 执行 → 再问」串起来，对照手搓里写在 while / for 里的顺序。
builder = StateGraph(State)  # 就是先立一张空白图，并规定节点之间传这种 State

# add_node 是在图上创建一个节点，指定这个节点执行哪个函数
builder.add_node("agent", agent)  # 创建节点名叫 agent，跑问模型那个函数。对照手搓里 create(..., tools=...)
builder.add_node("tools", ToolNode(TOOLS))  # 创建节点名叫 tools，跑执行并按 id 回灌，对照手搓里的 dispatch

# add_edge 是在两个节点之间的连线：START 不是你写的节点，是图的入口
builder.add_edge(START, "agent")  # 一开始一定先问模型

# add_conditional_edges 是一条活线，下一站由函数当场决定
# agent 问完模型之后，用 tools_condition 看出这轮有没有 tool_calls
builder.add_conditional_edges("agent", tools_condition)  # 有 tool_calls → tools；没有 → END，对照手搓里 if not tool_calls: return

# add_edge 是一条死线：tools 做完，一定回到 agent，对照手搓里执行完继续 for
builder.add_edge("tools", "agent")

# compile() 把上面的节点和连线收成一张能跑的图
graph = builder.compile()

# invoke 对照手搓里的 run()：把初始 messages 丢进图，从 START 走到 END。
# 每到一个节点就跑一次，tools 回来再进 agent，直到没有 tool_calls。
# 返回的是走完之后的整份 State，轨迹都在 result["messages"] 里。
def main() -> None:
    result = graph.invoke(
        {
            "messages": [
                SystemMessage(content=SYSTEM),
                HumanMessage(content=GOAL),
            ]
        }
    )

compile 之后图才能跑。invoke 把初始 messages 从 START 送到 END，对照手搓里的 run()。返回的是整份 State，轨迹在 result["messages"] 里。

tools_condition 本身不会再问一次模型，它只是查看最后一条 AIMessage：存在 tool_calls 就返回 "tools"，否则返回 END。真正执行函数的是 ToolNode；它按工具名找到 Python 函数，再用每个 tool_call_id 把结果包装成对应的 ToolMessage。这个 id 不能丢，否则下一轮模型无法判断哪份结果回应了哪次调用。

还要注意，节点通常不返回一份手动复制的完整 State。agent 只返回新生成的消息，tools 也只返回工具结果；LangGraph 再通过 messages 通道上的 add_messages 把局部更新并入现有历史。为什么这不是普通的列表赋值、并行写入时又为什么必须声明 reducer，留到状态契约篇展开。

这篇只有两个节点、三条边，是最小的一张。后面要加能力，步骤不变，多的是节点和边：

• 订房、确认、压缩，各加一个 add_node
• 下雨走一条、没确认走另一条，多写几条活线
• State 里除了 messages，还可以放记忆、计划
• compile 时传入 checkpointer，就能落盘、停住再续

立图、加节点、连线、compile、invoke，是图 API 的常规顺序。日常循环不必从空白图画起，官方建议使用 create_agent，问和执行已经画好。@task / @entrypoint 是另一种函数式写法，这篇用不到。要对上手搓里的 while，需要把图画开，所以从这张最小的图写起。

三、@tool 怎么变成那份 schema

四个工具都挂了 @tool。装饰器做的事只有一件：根据函数，自动生成手搓里那段 TOOL_SCHEMAS。发给模型的不是 Python，是推出来的 JSON。

天气只有 name: str、latitude: float 和一句 docstring，对照看不出来。订房把三项都对上：docstring 对照 description，Literal[...] 对照 enum，类型和必填从签名来。

@tool
def search_hotels(city: str, check_in: str) -> dict:
    """按城市和入住日期搜索可订酒店，返回 id、房型和每晚价格。
    用户只给了城市名、还没有 hotel_id 时先调它。不要用它下单。
    check_in 格式 YYYY-MM-DD，例如 2026-10-01。"""

@tool
def book_hotel(
    hotel_id: str,
    room_type: Literal["standard", "deluxe", "suite"],
    check_in: str,
    nights: int,
) -> dict:
    """用 search_hotels 返回的 hotel_id 预订房间。
    没有 hotel_id 时不要编一个，先搜。
    用户只是询问有没有房或多少钱时不要调用。"""

docstring 不是代码注释。读它的是模型：干什么、什么时候用、参数从哪来。日期格式写进描述，函数体里再验一次；验失败回灌期望格式，它下一轮会改。

schema 仍是软约束。缺字段、日期写错、room_type 填「豪华套房」，要在函数里拦，回灌「期望是什么」，不要只回 invalid。Literal 能收窄取值，拦不住它编一个从没搜过的 HT-999。

search_hotels 每命中一家，就把 id 记进本次的 seen_hotel_ids。book_hotel 先查这份名单，不在里面就回灌 error，不改库存。白名单不能写进 schema——写进去等于告诉模型「合法 id 是这些」，它反而会抄。

TOOLS 里执行类故意排前面。模型应靠 docstring 先搜再查、先搜后订，而不是按数组下标。搜地点和搜酒店没有互相依赖，可以同一轮一起调。查天气等坐标，订房等 hotel_id；两份回灌都到了，就可以同一轮一起调。库存是本地假数据，和手搓第二篇同一份，不打外部订房接口。确认下单也还没接，停住留给后面那层。

四、对照手搓看差在哪

循环形状没变。模型仍然只负责说出下一个工具；执行和回灌是你的代码，装在 ToolNode 里。先搜后订、id 从哪来、房型只能三项，都靠 schema 和函数体。这篇只留一个出口：没有 tool_calls 就终答。轮数上限和累计 token，后面再加。

五、跑起来看轨迹

完整代码在 agent-in-action/part2-agent-frameworks/langgraph/01-min-loop/。读环境和接模型在 env.py，各层 from env import chat_model。

cd part2-agent-frameworks/langgraph/01-min-loop
python agent.py

目标是「西安现在天气怎么样？再帮我订一间豪华套房，2026 年 10 月 1 日住两晚。」一次跑出来的轨迹如下。温度会随天气变，形状应是先搜地点和酒店，坐标和 hotel_id 都回灌后再同一轮查天气、下单：

—— 轨迹 ——
SystemMessage        只能通过工具获取事实，不要编温度、坐标、房价或 hotel_id。没有坐标时先 search_location，再 get_current_weather。订房没有 hotel_id 时先 search_hotels，再 book_hotel。有依赖的下一步，等回灌之后再调，不要同一轮一起调。用户只是询问有没有房或多少钱时不要下单。给出终答，带上天气、酒店名、房型、入住日期和总价。
HumanMessage         西安现在天气怎么样？再帮我订一间豪华套房，2026 年 10 月 1 日住两晚。
AIMessage            要调 ['search_location', 'search_hotels']
ToolMessage          {"name": "西安", "country": "中国", "latitude": 34.25833, "longitude": 108.92861}
ToolMessage          {"check_in": "2026-10-01", "hotels": [{"id": "HT-001", "name": "西安钟楼饭店", "city": "西安", "room_types": ["standard", "deluxe"], "price_per_night": {"standard": 180, "deluxe": 260}}, {"id": "HT-002", "name": "西安香格里拉", "city": "西安", "room_types": ["standard", "deluxe", "suite"], "price_per_night": {"standard": 150, "deluxe": 220, "suite": 400}}, {"id": "HT-003", "name": "回民街文化酒店", "city": "西安", "room_types": ["standard"], "price_per_night": {"standard": 110}}]}
AIMessage            要调 ['get_current_weather', 'book_hotel']
ToolMessage          {"temperature": 18.0, "apparent_temperature": 19.1, "humidity": 81, "wind_speed": 2.6}
ToolMessage          {"status": "ok", "hotel_id": "HT-002", "hotel_name": "西安香格里拉", "room_type": "suite", "check_in": "2026-10-01", "nights": 2, "total": 800, "currency": "RMB"}
AIMessage            西安现在天气是 18.0 摄氏度，湿度 81%，风速 2.6 米/秒。我在西安香格里拉酒店为您预订了一间豪华套房，预订时间为 2026 年 10 月 1 日，住两晚。总价为 800 元人民币（RMB）。

终答里的温度应与天气回灌一致。hotel_id 来自搜索结果，room_type 是 suite。库存里只有 HT-002 有 suite。跳过搜索、自造 id、房型填中文，先看 docstring 和白名单回灌。

图能转、先搜后订之后，下一篇停住再续，对上手搓上线那一层。