企业AI全栈学习地图 · L3 Agent架构设计 | 构建自主决策的下一代AI助手 · 第三十六篇：# LangGraph 保姆级教程：用图把 AI Agent 的思考流程画出来，再造一个能帮你规划旅游的机器人
先说一个挺烦的事。
很多教程讲 LangGraph，一上来就甩一大堆术语：State、Node、Edge、Reducer、Checkpointer、Subgraph……看完你脑子里一片浆糊，只知道这是个「给大模型做工作流」的框架，但到底怎么用、为什么要这么用，还是蒙的。
我换个讲法。
你想过没有，Deepseek 这类大模型，你问一句话它答一句话，本质是一条直线——输入进去，输出出来，结束。但你真正想要的 Agent，是那种「能拆解问题、能调工具、能循环思考、能记住上次聊到哪」的东西。直线的模型，做不到这些。
LangGraph 干的事，就是把这条直线，变成一张「图」。图上有一个个节点（Node），节点之间有一条条边（Edge），数据在节点之间流转、积累、变形（State）。因为有了「图」，你就能造出循环、分支、并行、暂停等直线模型做不到的控制流。
▲ 图1：LangGraph 把大模型的「直线问答」变成一张可控的「图」
这篇文章，我把 LangGraph 从头到尾给你讲明白，一边讲一边给你代码让你能跑。讲完了，我们真刀真枪搭一个「旅游规划机器人」——你告诉它想去哪，它帮你查景点、查交通、查酒店餐厅、生成整套行程。完整项目，我一行行拆给你看。
先把丑话说在前面：这篇文章很长，但每一节都有用。你按顺序读，读完就能自己搭一个能用的 Agent。
先搞懂那三个最核心的词：State、Node、Edge
LangGraph 的核心，是把 Agent 的工作流程建模成一张「图」。这张图由三个关键组件定义：
第一个，State（状态）：这是整个图里所有节点共享的数据结构，代表你应用「当前的状态」。它可以是任何 Python 类型，但通常是TypedDict（类型化字典）或者 Pydantic 的BaseModel。你可以把它理解成一个「共享记事本」——每个节点都能往上面写，也能从上面读。
第二个，Node（节点）：一个 Python 函数。它接收当前状态作为输入，执行一些计算或产生一些副作用，最后返回「更新后的状态」。节点就是「干活的」。
第三个，Edge（边）：决定「下一步去哪」的路径。它连接节点，定义了信息（也就是状态）如何在节点之间流动，以及图的执行顺序和逻辑分支。边可以是固定的（正常边），也可以是条件判断的（条件边）。
一句话总结：Node 干活，Edge 指路，State 是它们共享的记事本。
▲ 图2：LangGraph 三大核心要素：State 是记事本，Node 干活，Edge 指路
这么说可能还是抽象。我给你一个最小例子，看完就懂了。
from typing import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定义状态：一个包含消息列表的字典 class State(TypedDict): messages: list # 2. 定义节点：接收状态，返回更新 def node_a(state: State): return {"messages": state["messages"] + ["A 节点跑过了"]} def node_b(state: State): return {"messages": state["messages"] + ["B 节点跑过了"]} # 3. 构建图 graph = StateGraph(State) graph.add_node("a", node_a) graph.add_node("b", node_b) graph.add_edge(START, "a") # 从起点进入 a graph.add_edge("a", "b") # a 指向 b graph.add_edge("b", END) # b 指向终点 # 4. 编译（必须！） app = graph.compile() # 5. 运行 result = app.invoke({"messages": []}) print(result["messages"]) # 输出：['A 节点跑过了', 'B 节点跑过了']跑一遍，你会看到messages里依次累积了 A 和 B 的结果。这就是 LangGraph 最底层的模样：状态在节点间流转，边控制顺序。
这里有两个特殊节点你要记住：
•START：特殊节点，表示「用户输入进入图」的入口。加它主要是为了确定哪些节点应该最先被调用。
•END：特殊节点，表示终点。当你想表示「走到这就结束了」时，引用它。
还有一个铁律，我标红：构建完后必须调用.compile()编译，图才能真正用。编译这一步会做一些基本检查（比如有没有孤立节点），还可以指定检查点、断点这些运行时参数。
StateGraph，以及 State 里的门道
上面那个例子里，我用的是StateGraph这个类。它是 LangGraph 里最常用的图类，由一个你自己定义的 State 对象来参数化。
但 State 没看上去那么简单，它藏着两个进阶的门道，一个是Schema（模式），一个是Reducer（归约器）。这两个搞懂了，你才算真的会用 LangGraph。
Schema：图的输入输出长什么样
默认情况下，一张图的输入和输出用同一个 Schema（也就是同一个 State 结构）。但有时候你想更精细地控制：
•多模式（Multiple schemas）：图内部的节点之间，可能需要传递一些中间信息，而这些信息不一定是图的最终输入或输出。比如中间节点算出一个草稿，但最终只输出成品。
•约束图的输入输出：你可以明确指定「图只接收这几个字段作为输入，只输出这几个字段作为输出」，分开定义。
指定 Schema 主要用TypedDict，也支持用 Pydantic 的BaseModel（好处是能加默认值和数据校验，不过性能会受点影响）。
Reducer：状态更新是「覆盖」还是「追加」
这是 LangGraph 里最容易踩坑、但最重要的一环。
状态里每个键，都有自己的一个「归约器」（Reducer）函数，它决定了「节点返回的更新，怎么作用到旧状态上」。如果你不指定 Reducer，LangGraph 默认用「覆盖」——新值直接把旧值盖掉。
这么说你可能没感觉，看个实际场景。
大模型（LLM）的聊天接口，接受的是一个「消息列表」（Message 对象列表）。我们的图里，通常要把对话历史当成消息列表存在状态里。如果你直接把messages设成一个普通 list，不做任何处理，那问题就来了：
节点 A 返回一条新消息，旧的messages就被覆盖了——对话历史丢了，Agent 就「失忆」了。
所以，对于messages这种需要「不断累积」的字段，我们要给它指定一个 Reducer，告诉图「每次更新是追加，不是覆盖」。
最简单的做法是用operator.add：
from operator import add from typing import TypedDict, Annotated class State(TypedDict): messages: Annotated[list, add] # add 表示追加当节点返回{"messages": [新消息]}时，旧的messages会+上这个新列表，实现累积。
更规范的做法，是用 LangGraph 自己提供的add_messages：
from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages]add_messages比operator.add更聪明——它专门处理 Message 对象，能正确处理AIMessage、HumanMessage、ToolMessage这些类型，还能处理消息的合并、去重等细节。
你还可以自定义 Reducer 函数，做任意你想要的合并逻辑（比如只保留最新 N 条）。核心就一句话：State 里每个字段，想清楚它该「覆盖」还是「累积」，累积的话配上合适的 Reducer。
▲ 图3：Reducer 决定状态更新是「覆盖」还是「追加累积」
这就是为什么旅游规划机器人项目里，状态定义长这样：
from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class PublicState(TypedDict): messages: Annotated[list, add_messages]就一个字段messages，配上add_messages作为 Reducer，保证整个多轮对话的记忆不断累积、不会丢失。
边，和它背后的控制流魔法
前面说了，边分两种：正常边（Normal Edges）和条件边（Conditional Edges）。
正常边最简单，就是一个节点直接指向下一个节点。条件边就厉害了——它调用一个函数，根据当前状态动态决定下一个要访问的节点。这是 LangGraph 实现「循环」「分支」的关键。
from langgraph.graph import StateGraph, START, END def route(state) -> str: # 根据状态里的某个字段，决定下一步去哪 if state["messages"][-1] == "继续": return "node_b" return END graph.add_conditional_edges("node_a", route)route函数返回一个字符串，LangGraph 用它来路由到对应名字的节点（或 END）。这样你就有了「根据状态动态决定走向」的能力。
有了条件边，你就能造出循环。循环很常见——比如一个 Agent 要反复「思考 → 调工具 → 看结果 → 再思考」，直到得出最终答案为止。但循环有个风险：万一停不下来怎么办？
所以 LangGraph 里，创建包含循环的图时，必须有一种终止机制——通常是加一个条件边，一旦满足终止条件就路由到 END。还可以在调用图时设置「递归限制」（recursion limit），限制图在报错前允许执行的「超级步骤」数，防止死循环把资源耗尽。
这里提一个词：超级步骤（super-step）。这是理解 LangGraph 执行模型的关键。简单说，一次超级步骤，就是图里节点的一轮「并行执行 + 状态更新」。节点的并行能力，就靠这个机制。
▲ 图4：条件边根据状态动态分流，是「循环」和「分支」的关键
让节点跑起来：并行、Send、Command
讲到这，你已经能搭简单的图了。但 Agent 越来越复杂时，有几招能让你事半功倍。
并行执行：扇出和扇入
LangGraph 原生支持节点并行执行。当一个节点有多条边同时指向多个下游节点时，这些下游节点会并行跑，这就是「扇出」（fan-out）；它们各自算完后，结果再汇合到下一个节点，这叫「扇入」（fan-in）。
并行的意义在于加速——比如你要同时查三个景点的信息，与其一个个串行查，不如并行查，省时间。
Send 机制：边数量不确定时的并行
默认情况下，节点和边是事先固定好的，它们操作同一个共享状态。但有时候，「确切的边」在事先是未知的。
最典型的例子是map-reduce 模式：一个节点生成了一个对象列表，你想把某个下游节点应用到列表里的每一个对象上。但对象的数量事先不知道，所以「边的数量」也就不知道。
这时候就要用Send机制。Send 允许你动态地给同一个节点「派发」多个并行执行的分支，每个分支处理列表里的一个对象。数量不固定、输入状态也各不相同（每个对象一个）。
from langgraph.types import Send def distribute(state): # 假设 state["items"] 是一个列表 return [Send("process_item", {"item": item}) for item in state["items"]]这个 return 出现在一个节点里，LangGraph 会理解为「对这个process_item节点，并行派发 N 个任务」。这就是 map-reduce 的 map 部分——任务分解、并行处理，最后再汇总（reduce）。
▲ 图5：Send 机制让节点扇出并行处理，再扇入汇总
Command：一个节点同时更新状态 + 决定下一步
有些时候，你想在同一个节点里「既做状态更新，又决定下一个要去哪个节点」。LangGraph 提供了Command对象来实现这个。
你可以把Command理解成「加强版的条件边」。普通的条件边，只能根据状态选节点；而Command让节点函数自己返回一个「我下一步要去哪 + 我要更新什么状态」的完整指令，把控制流（边）和状态更新（节点）揉在了一起。
from langgraph.types import Command def my_node(state): return Command(update={"some_key": "some_value"}, goto="next_node")这在做复杂路由、多智能体交接时特别有用。
让 Agent 记住你：Persistence 持久化与 Threads 线程
这是 Agent 真正「像个人」的关键。一个没有记忆的 Agent，每轮对话都是失忆的。
LangGraph 用「检查点」（Checkpoint）来实现持久化。检查点器（checkpointer）会在每个超级步骤保存一份图状态的快照，允许你在任何时间恢复。这就让「人机循环交互、内存管理、容错」这些特性成为可能。
一个检查点（StateSnapshot）包含几个关键信息：
•config：这个检查点关联的配置。
•values：此时各状态通道的值。
•next：下一个要执行的节点名称的元组。
•tasks：要执行的下个任务（PregelTask 对象），如果这步之前执行失败，它会带上错误信息。
那「线程」（Threads）又是什么？
线程，就是检查点保存器给每个检查点分配的唯一 ID。你在调用带检查点的图时，必须在配置的configurable部分指定thread_id：
config = {"configurable": {"thread_id": "user_123"}} app.invoke({"messages": [...]}, config=config)一个thread_id代表「图与用户之间的一个独立会话」。同一个thread_id下的所有对话轮次、甚至单次图执行的步骤，都会按这个 ID 组织起来。所以：
•同一个thread_id的多次调用 → 自动加载历史状态，实现多轮记忆。
•不同的thread_id→ 完全隔离的会话，互不干扰。
最常用的检查点器是InMemorySaver（内存级，进程重启就丢），生产环境可以换成 Sqlite、Redis（RedisSaver）、MongoDB 等持久化存储。
再往深一层，LangGraph 还支持「跨线程持久化」——把用户信息（姓名、偏好等）存在共享内存里，让新的对话线程也能复用。这就要用到「存储」（Store）的概念，和检查点（Checkpoint）是两回事：检查点是「会话内的状态快照」，存储是「跨会话共享的数据」。
配置持久化很简单，编译图时传一个 checkpointer 进去即可：
▲ 图6：检查点 + thread_id 让 Agent 拥有「多轮记忆」
from langgraph.checkpoint.memory import InMemorySaver memory = InMemorySaver() app = graph.compile(checkpointer=memory)两个让 Agent 更可控的高级特性
Configuration：把图做成「可插拔」
LangGraph 允许你在创建图时，把图的某些部分标记成「可配置的」。好处是：运行时可以动态改这些部分的行为，而不用改图的源代码。
常见用例：动态切换用哪个 LLM、改系统提示词内容、调整工具参数等。这让你能用一张「通用认知架构」（也就是图本身），通过不同配置实例化出多个行为略有不同的版本。
Interrupt 与 HITL：让人在关键时刻插一脚
纯自动的 Agent 很危险，尤其是涉及「花钱」「发消息」「对外操作」的动作。这时候你需要Human-in-the-loop（HITL，人机交互）。
interrupt函数允许你在图执行到特定点的时候「暂停」，收集用户的输入、让用户验证或修改状态、或者做决策，然后再继续执行。
HITL 的三大关键用例：
1审查工具调用：人在工具执行之前，审查、编辑或批准 LLM 请求的工具调用。
2验证 LLM 输出：人审查、编辑或批准 LLM 生成的内容。
3提供上下文：让 LLM 明确请求人类输入来做澄清、补细节，支持多轮对话。
一个典型的场景是：Agent 想调用「发送邮件」这个工具，先在发之前interrupt，把邮件内容给人看，人点「批准」才真的发出去，点「修改」就改一改再发，或者输入一段反馈让 Agent 重新来。
这跟另一个特性结合得很紧——
▲ 图7：HITL 让人在关键时刻暂停、审查、批准
多智能体（Multi-Agent）：让多个 Agent 分工协作
当一个 LLM 应用太复杂，把它拆成多个智能体，每个负责一部分，往往更清晰。
在多智能体架构里，每个智能体被表示成图里的一个节点。它们之间的交互，最常见的模式是「交接」（handoff）——一个智能体把控制权转交给另一个智能体。交接时要指定两样东西：
•destination：导航到目标 Agent（也就是图里的节点名）。
•payload：传递给那个 Agent 的信息（也就是状态更新）。
交接可以用Command实现，也可以用工具实现。
更进一步，可以构建「多智能体网络」——每个智能体都能跟所有其他智能体通信（多对多连接），自己决定下一个要调用谁。这在复杂协作场景里很有用。
多智能体系统还要支持「多轮对话 + HITL」，关键机制是：智能体需要用户输入时暂停图、收集输入、把输入作为新消息加入历史、再把控制权交还（或路由到合适的智能体）。
▲ 图8：多智能体通过「交接」把控制权转交给其他 Agent
Subgraph 子图：把图当乐高积木拼
子图，说白了就是「作为另一个图里的节点来使用的图」。这是「封装」这个古老思想在 LangGraph 上的应用。
子图让你能构建「由多个自身就是图的组件」组成的复杂系统。一个常见用例是构建多智能体系统——每个智能体内部可能本身就是一个复杂的图。
注意一个容易混淆的点：有子图不一定是多智能体系统，多智能体系统也不一定用了子图。这是两个正交的概念。
用子图时，核心问题是「父子图怎么通信」。有两种情况：
1父图和子图共享 Schema 的键：直接加一个「包含编译后子图」的节点即可。
2父图和子图 Schema 不同：必须加一个节点函数来调用子图，做状态的转换。
子图同样能用Command把控制流和状态更新结合起来。另外，给包含子图的图加持久性，只需在编译父图时传一个 checkpointer，LangGraph 会自动把它传播到所有子图。
▲ 图9：子图把「图」当作节点，像乐高积木一样嵌套封装
工具调用与 ToolNode
AI 应用通过自然语言和用户交互，但光会聊天没意思，我们期望 Agent 能「用工具」。
工具（Tool）封装了一个可调用函数 + 它的输入模式（input schema）。把这些工具传给兼容的聊天模型，模型就能自主决定「要不要调工具、用什么参数」。
LangGraph 专门定义了一个ToolNode，作为工具节点：
from langgraph.prebuilt import ToolNode, tools_condition tool_node = ToolNode(tools) # 把工具们包成一个节点 graph.add_node("tools", tool_node) # agent 节点之后，用 tools_condition 做条件路由：要调工具就去 tools，否则结束 graph.add_conditional_edges("agent", tools_condition) graph.add_edge("tools", "agent") # 工具跑完，结果送回 agenttools_condition是一个内置的条件函数：如果模型输出了工具调用，就路由到对应的工具；否则路由到 END。这就构成了经典的 ReAct 循环——Agent 思考 → 调工具 → 看结果 → 再思考，直到不再需要工具为止。
▲ 图10：ToolNode 构建起经典的 ReAct「思考→调工具→再思考」循环
关于工具调用，还有几个实战要点你要知道：
•工具可能失败：模型可能调一个不存在的工具，或参数不对。要优雅地处理这些错误（异常 ToolNode、自定义策略、节点重试）。
•传运行时值给工具：有些参数不该由 LLM 填（比如用户 ID、出于安全考虑），要用InjectedArg注解，运行时由应用逻辑提供。
•工具更新图状态：可以从工具返回Command(update={...})来更新状态的某个键。
•处理大量工具：工具多了，可以像 RAG 一样，在调用模型前先「检索」出相关的工具子集，减少 token 消耗、降低模型选错工具的概率。
可视化：把图「画」出来
当图越来越复杂，能把图可视化出来会方便很多。LangGraph 提供了多种内置的可视化方式，能把你定义的图渲染成一张流程图（依赖 graphviz 等）。不过这功能有时候不太准，属于正常情况，不必迷信。
好，前半篇的「LangGraph 保姆级教程」到这里就讲完了。你可能已经有点感觉了：State、Node、Edge 是骨架，Reducer、Schema 是血肉，边和条件边是神经系统，持久化是记忆，Command/Send/并行是四肢，多智能体和子图是「团队协作」，HITL 是「安全刹车」。
现在，我们把所有这些，装进一个真实能跑的项目里。
实战：搭一个旅游规划机器人
需求是这样：做一个能根据用户需求，自动规划旅游行程的 Agent。它能推荐目的地、查景点、查交通、推荐餐厅酒店、保存结果。
技术上，这就是一个典型的「工具调用型 Agent」，正好用 LangGraph 的 StateGraph + ToolNode + Checkpointer 来搭。
项目结构长这样
旅游规划机器人/ ├── agents/ # 智能体，负责决策和调工具 ├── graph/ # LangGraph 工作流定义 ├── models/ # 大模型工厂 ├── prompts/ # 提示词模板 ├── states/ # 状态定义 ├── tools/ # 具体工具（景点、交通、周边、地图、搜索、保存） ├── utils/ # 辅助函数 └── webrun.py # Gradio Web 界面入口这个分层很清晰：states管状态结构，agents管「调用大模型」这个动作，tools管「具体干活的函数」，graph把这一切用图串起来，models负责造出大模型实例，prompts存提示词。
▲ 图11：旅游规划机器人的分层架构，各司其职、环环相扣
第一步：定义状态 State
states/state.py里，状态定义极其简单：
from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class PublicState(TypedDict): messages: Annotated[list, add_messages]就一个messages字段，用add_messages做 Reducer。这保证了整个多轮对话的记忆会不断累积。你前面学的 Reducer 知识，在这里直接落地了。
第二步：造大模型 Model
models/factory.py里是一个工厂类：
from langchain_openai import ChatOpenAI class LLMFactory: @staticmethod def get_llm(model=None, temperature=None): if model is None or temperature is None: raise ValueError("Both 'model' and 'temperature' must be specified.") if model.startswith('gpt'): return ChatOpenAI(model=model, temperature=temperature, streaming=True) raise ValueError(f"Model {model} is not supported.")这个工厂把「造模型」这件事封装起来，根据传入的模型名和温度参数，返回对应的 LLM 实例。streaming=True表示启用流式输出（逐块生成，前端体验更好）。以后想换成国产模型，只需在这里改工厂逻辑，别处不用动——这就是「Configuration 思想」的朴素实践。
第三步：定义 Agent
agents/agents.py里定义了 Agent。核心逻辑是：初始化时把工具绑定到 LLM 上，调用时把提示词+用户消息拼起来发给模型。
from langchain_core.tools import StructuredTool from langchain_core.messages import HumanMessage from models.factory import LLMFactory class Agent: def __init__(self, model_name, temperature, prompt_template, tools): # 图初始化时创建 LLM，不用每次调用都重建 self.llm = LLMFactory.get_llm(model=model_name, temperature=temperature) if tools: self.llm = self.llm.bind_tools(tools) # 把工具绑给模型 self.prompt_template = prompt_template class AsyncAgent(Agent): async def __call__(self, state): # 系统提示 + 第一条用户消息，拼成 first_prompt first_prompt = HumanMessage(self.prompt_template.format( current_time=get_current_local_datetime(), first_user_message=state['messages'][0].content )) prompt = [first_prompt] + state['messages'][1:] response = await self.llm.ainvoke(prompt) return {'messages': [response]}这里有三个细节很关键：
1bind_tools：这是把工具「告诉」模型的关键动作。绑定之后，模型才能知道有哪些工具、每个工具的参数，进而在合适时输出工具调用。
2first_prompt的构造：把系统提示词模板格式化（填入当前时间 + 第一条用户消息），再和剩余的历史消息拼成一个列表。之所以这么处理，是因为某些模型（如 Gemini）不支持单独的系统提示，用「把系统提示塞进第一条 HumanMessage」的方式绕过限制。
3异步__call__：节点函数签名是(state)，返回{'messages': [...]}，正好符合 LangGraph 节点的规范。ainvoke是异步调用，对应项目里的AsyncAgent；也有同步版的SyncAgent用invoke。
第四步：定义工具 Tools
这是项目的「能力层」，一共 7 个工具，每个都对应旅游规划的一类需求。工具都用@tool装饰器声明，靠Annotated[str, "参数说明"]给每个参数写清楚含义——这些说明会被转成 schema，是模型正确调用参数的依据。
工具 1：网络搜索web_search
用 DuckDuckGo 搜索，返回标题、链接、开头内容。用来在用户没给明确目的地时，搜资料、推荐候选目的地。
from langchain_core.tools import tool from duckduckgo_search import DDGS @tool def web_search(keywords, max_results=10): """网络搜索工具。搜索关键词，返回搜索结果列表""" with DDGS() as ddgs: return [r for r in ddgs.text(keywords=keywords, region='cn-zh', max_results=max_results)]工具 2：位置获取get_location_coordinate
调高德地图地理编码 API，根据地点名+城市名查经纬度。因为可能有同名地点，所以返回的是「所有同名地点的列表」，让 Agent 自己判断选哪个。
@tool def get_location_coordinate(location, city=""): """位置获取工具。返回同名地点的经纬度列表""" amap_key = os.getenv("AMAP_API_KEY") url = f"https://restapi.amap.com/v3/geocode/geo?key={amap_key}&address={location}&city={city}" result = requests.get(url).json() # 遍历 geocodes，返回 address/coordinate/citycode ...工具 3：景点信息get_attractions_information
这是最「重」的一个工具，用 Selenium + BeautifulSoup 爬马蜂窝网站，抓目的地的简介和景点列表（名称、简介、开放时间、游玩时间）。
@tool def get_attractions_information(destination): """景点搜索工具。获取目的地概览和景点信息列表""" driver = InitWebDriver() # 初始化 Chrome，伪装成正常浏览器 # 搜索目的地 -> 提取 mddid -> 进入攻略页 -> 抓 overview 和 scenic_list ... return {'overview': overview, 'scenic_list': scenic_spots}注意它用excludeSwitches、disable-blink-features、execute_cdp_cmd这些手段伪装浏览器，绕过网站的反爬检测。这是爬虫实战的细节，写出来是让你知道「LLM 应用的工具，本质还是传统 Python 代码」。
工具 4：路线规划route_planning
调高德地图的公共交通路线规划 API，返回步行距离、打车费用、公交方案列表。
@tool def route_planning(origin, destination, origin_city_code, dest_city_code): """路线规划工具。返回步行距离、打车费用、公交方案列表""" # 调高德 transit/integrated 接口 ... return routes # walking_distance, taxi_cost, public_transport_options_list工具 5：周边 POI 搜索search_nearby_poi
调高德周边搜索 API，按中心点坐标查餐厅、酒店，返回名称、类型、地址、距离、评分、价格。
工具 6：静态地图get_static_map（static_map.py）
调高德静态地图 API，根据经纬度生成一张地图图片。
工具 7：信息保存save_info_and_clear_history
这个工具有点特别，用到了response_format="content_and_artifact"：
@tool(response_format="content_and_artifact") def save_info_and_clear_history(infomation_to_save): """信息保存工具。保存当前已获取的有用信息""" content = "信息已保存。" return content, infomation_to_save它返回两个值：content（给模型看的提示「已保存」）和infomation_to_save（真正要保存的内容，存在 ToolMessage 的artifact字段里）。模型只看到content，真正内容通过 artifact 保存，实现了「保存但不污染对话」。
第五步：提示词 Prompt
prompts/main.py里是核心提示词模板。这是整个 Agent 的「大脑说明书」，写得非常详细，我挑几个关键点讲：
角色定义是「经验丰富的旅游规划师」。然后是约束，比如：
•每次决策只用一种工具，但可以用任意多次。
•严格按工具参数说明生成 tool calls。
•连续三次调用失败就求助，不要死循环。
•不得捏造信息（DO NOT MAKE UP INFORMATION）。
•已得到的信息不反复查询。
然后是「工作流程」，把一次旅游规划拆成了 14 个步骤：确定目的地 → 查景点 → 查经纬度 → 匹配同名地点 → 让用户选景点 → 规划每日行程 → 查交通 → 分析交通合理性 → 反思调整 → 出方案 → 查餐饮住宿 → 整合 → 出完整行程 → 用户确认。
这个流程设计得非常讲究，体现了前面学的高级概念：反思、多步推理、HITL（让用户选择、确认）。它要求 Agent 每步输出「任务、回顾、思考、计划」四段思考过程，严格区分「给用户的消息」和「调用工具的消息」。
▲ 图12：旅游规划机器人从需求到成型行程的完整工作流
第六步：把一切串起来 Graph
graph/graph.py是核心，用 StateGraph 把 Agent 和工具节点串成 ReAct 循环：
from langgraph.graph import StateGraph, START from langgraph.prebuilt import ToolNode, tools_condition from agents.agents import AsyncAgent def create_graph(model_name, is_async=True): graph = StateGraph(PublicState) tools = [web_search, get_location_coordinate, get_attractions_information, route_planning, search_nearby_poi, save_info_and_clear_history] travel_agent = AsyncAgent(model_name=model_name, temperature=0, prompt_template=agent_prompt_template, tools=tools) graph.add_node("agent", travel_agent) # agent 节点 graph.add_node("tools", ToolNode(tools)) # tools 节点 graph.add_edge(START, "agent") # 从 agent 开始 graph.add_conditional_edges("agent", tools_condition) # 要调工具→tools，否则→END graph.add_edge("tools", "agent") # 工具跑完→回 agent return graph def init_app(model_name, is_async=True): graph = create_graph(model_name, is_async) memory = InMemorySaver() # 内存检查点，实现多轮记忆 app = graph.compile(checkpointer=memory) # 编译 + 挂载检查点 return app这段代码，几乎涵盖了我前面讲的所有知识点：
•StateGraph：构建状态图。
•ToolNode + tools_condition：工具调用和条件路由，形成 ReAct 循环。
•Agent 作为节点：AsyncAgent.__call__(state)正好是节点函数签名。
•InMemorySaver + compile(checkpointer=...)：挂载检查点器，实现多轮记忆（也就是 Threads 机制）。
整个流程是：用户消息进入 → Agent 节点（大模型决策，输出工具调用或最终回答）→tools_condition判断 → 如果是工具调用就进 tools 节点执行，结果送回 agent 再思考 → 循环，直到模型不再调工具，给出最终答案。
第七步：前端界面 webrun.py
最后用 Gradio 搭了个 Web 界面，左栏显示调试信息（工具调用过程），右栏是聊天框。关键是用app.astream_events(...)做流式输出，实时展示模型逐字生成的内容，以及on_tool_start/on_tool_end事件展示工具调用过程。
import gradio as gr from graph.graph import init_app app = init_app(model_name="gpt-4o-mini") thread_id = get_thread_id() config = {"configurable": {"thread_id": thread_id}} async def process_message(user_message, chatbot_history, debug_history): async for event in app.astream_events(...): if kind == "on_chat_model_stream": # 流式更新 AI 回复 elif kind == "on_tool_start": # 展示工具开始调用 elif kind == "on_tool_end": # 展示工具调用结果运行python webrun.py，会自动打开 Gradio 网页，你就能直接和机器人对话了。
整个项目，从状态、模型、工具、提示词、图到前端，环环相扣，用的全是 LangGraph 的核心能力。这就是「把知识落地成项目」的完整样子。
复盘：这篇文章到底想让你带走什么
写到最后，我把 LangGraph 的学习路径和最关键的认知，留给你。
先说技术上的学习路径，按这个顺序走：
1先吃透 State、Node、Edge 三个词，这是地基。
2再弄懂 Schema 和 Reducer——尤其 Reducer，它决定了状态是「覆盖」还是「累积」，这是记忆能不能生效的根本。
3学会用边（尤其条件边）造循环和分支。
4掌握持久化（Checkpoint + Thread），让 Agent 有记忆。
5再进阶到 Send 并行、Command、多智能体、子图、HITL。
我特别想强调一个认知，比任何代码都重要：
LangGraph 的价值，不在「能用大模型」，而在「让你把 Agent 的思考流程，显式地画成一张可控的图」。直线的大模型调用，你没法控制它中间怎么想；但画成图之后，每一步、每一个分支、每一次循环、每一次暂停，都在你的掌控之中。这就是「可观测、可控制、可干预」的 Agent 和「黑盒聊天」的本质区别。
换句话说，能力的边界，从「模型多聪明」转移到了「你多会设计流程」。模型是发动机，LangGraph 是底盘和方向盘——发动机再好，没有底盘和方向盘，也只是一台踩不了刹车、转不了弯的裸机。
最后一个提醒：这篇文章里的代码，你都得亲手敲一遍、跑一遍。尤其是那个旅游规划机器人，你把InMemorySaver换成SqliteSaver，把爬虫工具换成你自己的 API，把gpt-4o-mini换成国产模型——改一改，它就从「教程里的例子」变成「你自己的项目」。
代码是敲出来的，不是看出来的。去装依赖，去配.env里的高德 API KEY，去运行python webrun.py，去问它一句「我想去长沙玩三天」。
跑通的那一刻，LangGraph 对你来说，就不再是一个「听说过但没用过」的词了。
