尧图网页设计 
企业AI全栈学习地图 · L3 Agent架构设计 | 构建自主决策的下一代AI助手 · 第三十六篇# LangGraph 保姆级教程用图把 AI Agent 的思考流程画出来再造一个能帮你规划旅游的机器人先说一个挺烦的事。很多教程讲 LangGraph一上来就甩一大堆术语State、Node、Edge、Reducer、Checkpointer、Subgraph……看完你脑子里一片浆糊只知道这是个「给大模型做工作流」的框架但到底怎么用、为什么要这么用还是蒙的。我换个讲法。你想过没有Deepseek 这类大模型你问一句话它答一句话本质是一条直线——输入进去输出出来结束。但你真正想要的 Agent是那种「能拆解问题、能调工具、能循环思考、能记住上次聊到哪」的东西。直线的模型做不到这些。LangGraph 干的事就是把这条直线变成一张「图」。图上有一个个节点Node节点之间有一条条边Edge数据在节点之间流转、积累、变形State。因为有了「图」你就能造出循环、分支、并行、暂停等直线模型做不到的控制流。▲ 图1LangGraph 把大模型的「直线问答」变成一张可控的「图」这篇文章我把 LangGraph 从头到尾给你讲明白一边讲一边给你代码让你能跑。讲完了我们真刀真枪搭一个「旅游规划机器人」——你告诉它想去哪它帮你查景点、查交通、查酒店餐厅、生成整套行程。完整项目我一行行拆给你看。先把丑话说在前面这篇文章很长但每一节都有用。你按顺序读读完就能自己搭一个能用的 Agent。先搞懂那三个最核心的词State、Node、EdgeLangGraph 的核心是把 Agent 的工作流程建模成一张「图」。这张图由三个关键组件定义第一个State状态这是整个图里所有节点共享的数据结构代表你应用「当前的状态」。它可以是任何 Python 类型但通常是TypedDict类型化字典或者 Pydantic 的BaseModel。你可以把它理解成一个「共享记事本」——每个节点都能往上面写也能从上面读。第二个Node节点一个 Python 函数。它接收当前状态作为输入执行一些计算或产生一些副作用最后返回「更新后的状态」。节点就是「干活的」。第三个Edge边决定「下一步去哪」的路径。它连接节点定义了信息也就是状态如何在节点之间流动以及图的执行顺序和逻辑分支。边可以是固定的正常边也可以是条件判断的条件边。一句话总结Node 干活Edge 指路State 是它们共享的记事本。▲ 图2LangGraph 三大核心要素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函数它决定了「节点返回的更新怎么作用到旧状态上」。如果你不指定 ReducerLangGraph 默认用「覆盖」——新值直接把旧值盖掉。这么说你可能没感觉看个实际场景。大模型LLM的聊天接口接受的是一个「消息列表」Message 对象列表。我们的图里通常要把对话历史当成消息列表存在状态里。如果你直接把messages设成一个普通 list不做任何处理那问题就来了节点 A 返回一条新消息旧的messages就被覆盖了——对话历史丢了Agent 就「失忆」了。所以对于messages这种需要「不断累积」的字段我们要给它指定一个 Reducer告诉图「每次更新是追加不是覆盖」。最简单的做法是用operator.addfrom operator import add from typing import TypedDict, Annotated class State(TypedDict): messages: Annotated[list, add] # add 表示追加当节点返回{messages: [新消息]}时旧的messages会上这个新列表实现累积。更规范的做法是用 LangGraph 自己提供的add_messagesfrom 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。▲ 图3Reducer 决定状态更新是「覆盖」还是「追加累积」这就是为什么旅游规划机器人项目里状态定义长这样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。▲ 图5Send 机制让节点扇出并行处理再扇入汇总Command一个节点同时更新状态 决定下一步有些时候你想在同一个节点里「既做状态更新又决定下一个要去哪个节点」。LangGraph 提供了Command对象来实现这个。你可以把Command理解成「加强版的条件边」。普通的条件边只能根据状态选节点而Command让节点函数自己返回一个「我下一步要去哪 我要更新什么状态」的完整指令把控制流边和状态更新节点揉在了一起。from langgraph.types import Command def my_node(state): return Command(update{some_key: some_value}, gotonext_node)这在做复杂路由、多智能体交接时特别有用。让 Agent 记住你Persistence 持久化与 Threads 线程这是 Agent 真正「像个人」的关键。一个没有记忆的 Agent每轮对话都是失忆的。LangGraph 用「检查点」Checkpoint来实现持久化。检查点器checkpointer会在每个超级步骤保存一份图状态的快照允许你在任何时间恢复。这就让「人机循环交互、内存管理、容错」这些特性成为可能。一个检查点StateSnapshot包含几个关键信息•config这个检查点关联的配置。•values此时各状态通道的值。•next下一个要执行的节点名称的元组。•tasks要执行的下个任务PregelTask 对象如果这步之前执行失败它会带上错误信息。那「线程」Threads又是什么线程就是检查点保存器给每个检查点分配的唯一 ID。你在调用带检查点的图时必须在配置的configurable部分指定thread_idconfig {configurable: {thread_id: user_123}} app.invoke({messages: [...]}, configconfig)一个thread_id代表「图与用户之间的一个独立会话」。同一个thread_id下的所有对话轮次、甚至单次图执行的步骤都会按这个 ID 组织起来。所以•同一个thread_id的多次调用 → 自动加载历史状态实现多轮记忆。•不同的thread_id→ 完全隔离的会话互不干扰。最常用的检查点器是InMemorySaver内存级进程重启就丢生产环境可以换成 Sqlite、RedisRedisSaver、MongoDB 等持久化存储。再往深一层LangGraph 还支持「跨线程持久化」——把用户信息姓名、偏好等存在共享内存里让新的对话线程也能复用。这就要用到「存储」Store的概念和检查点Checkpoint是两回事检查点是「会话内的状态快照」存储是「跨会话共享的数据」。配置持久化很简单编译图时传一个 checkpointer 进去即可▲ 图6检查点 thread_id 让 Agent 拥有「多轮记忆」from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() app graph.compile(checkpointermemory)两个让 Agent 更可控的高级特性Configuration把图做成「可插拔」LangGraph 允许你在创建图时把图的某些部分标记成「可配置的」。好处是运行时可以动态改这些部分的行为而不用改图的源代码。常见用例动态切换用哪个 LLM、改系统提示词内容、调整工具参数等。这让你能用一张「通用认知架构」也就是图本身通过不同配置实例化出多个行为略有不同的版本。Interrupt 与 HITL让人在关键时刻插一脚纯自动的 Agent 很危险尤其是涉及「花钱」「发消息」「对外操作」的动作。这时候你需要Human-in-the-loopHITL人机交互。interrupt函数允许你在图执行到特定点的时候「暂停」收集用户的输入、让用户验证或修改状态、或者做决策然后再继续执行。HITL 的三大关键用例1审查工具调用人在工具执行之前审查、编辑或批准 LLM 请求的工具调用。2验证 LLM 输出人审查、编辑或批准 LLM 生成的内容。3提供上下文让 LLM 明确请求人类输入来做澄清、补细节支持多轮对话。一个典型的场景是Agent 想调用「发送邮件」这个工具先在发之前interrupt把邮件内容给人看人点「批准」才真的发出去点「修改」就改一改再发或者输入一段反馈让 Agent 重新来。这跟另一个特性结合得很紧——▲ 图7HITL 让人在关键时刻暂停、审查、批准多智能体Multi-Agent让多个 Agent 分工协作当一个 LLM 应用太复杂把它拆成多个智能体每个负责一部分往往更清晰。在多智能体架构里每个智能体被表示成图里的一个节点。它们之间的交互最常见的模式是「交接」handoff——一个智能体把控制权转交给另一个智能体。交接时要指定两样东西•destination导航到目标 Agent也就是图里的节点名。•payload传递给那个 Agent 的信息也就是状态更新。交接可以用Command实现也可以用工具实现。更进一步可以构建「多智能体网络」——每个智能体都能跟所有其他智能体通信多对多连接自己决定下一个要调用谁。这在复杂协作场景里很有用。多智能体系统还要支持「多轮对话 HITL」关键机制是智能体需要用户输入时暂停图、收集输入、把输入作为新消息加入历史、再把控制权交还或路由到合适的智能体。▲ 图8多智能体通过「交接」把控制权转交给其他 AgentSubgraph 子图把图当乐高积木拼子图说白了就是「作为另一个图里的节点来使用的图」。这是「封装」这个古老思想在 LangGraph 上的应用。子图让你能构建「由多个自身就是图的组件」组成的复杂系统。一个常见用例是构建多智能体系统——每个智能体内部可能本身就是一个复杂的图。注意一个容易混淆的点有子图不一定是多智能体系统多智能体系统也不一定用了子图。这是两个正交的概念。用子图时核心问题是「父子图怎么通信」。有两种情况1父图和子图共享 Schema 的键直接加一个「包含编译后子图」的节点即可。2父图和子图 Schema 不同必须加一个节点函数来调用子图做状态的转换。子图同样能用Command把控制流和状态更新结合起来。另外给包含子图的图加持久性只需在编译父图时传一个 checkpointerLangGraph 会自动把它传播到所有子图。▲ 图9子图把「图」当作节点像乐高积木一样嵌套封装工具调用与 ToolNodeAI 应用通过自然语言和用户交互但光会聊天没意思我们期望 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 思考 → 调工具 → 看结果 → 再思考直到不再需要工具为止。▲ 图10ToolNode 构建起经典的 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旅游规划机器人的分层架构各司其职、环环相扣第一步定义状态 Statestates/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 知识在这里直接落地了。第二步造大模型 Modelmodels/factory.py里是一个工厂类from langchain_openai import ChatOpenAI class LLMFactory: staticmethod def get_llm(modelNone, temperatureNone): if model is None or temperature is None: raise ValueError(Both model and temperature must be specified.) if model.startswith(gpt): return ChatOpenAI(modelmodel, temperaturetemperature, streamingTrue) raise ValueError(fModel {model} is not supported.)这个工厂把「造模型」这件事封装起来根据传入的模型名和温度参数返回对应的 LLM 实例。streamingTrue表示启用流式输出逐块生成前端体验更好。以后想换成国产模型只需在这里改工厂逻辑别处不用动——这就是「Configuration 思想」的朴素实践。第三步定义 Agentagents/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(modelmodel_name, temperaturetemperature) 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_timeget_current_local_datetime(), first_user_messagestate[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_results10): 网络搜索工具。搜索关键词返回搜索结果列表 with DDGS() as ddgs: return [r for r in ddgs.text(keywordskeywords, regioncn-zh, max_resultsmax_results)]工具 2位置获取get_location_coordinate调高德地图地理编码 API根据地点名城市名查经纬度。因为可能有同名地点所以返回的是「所有同名地点的列表」让 Agent 自己判断选哪个。tool def get_location_coordinate(location, city): 位置获取工具。返回同名地点的经纬度列表 amap_key os.getenv(AMAP_API_KEY) url fhttps://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_mapstatic_map.py调高德静态地图 API根据经纬度生成一张地图图片。工具 7信息保存save_info_and_clear_history这个工具有点特别用到了response_formatcontent_and_artifacttool(response_formatcontent_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 保存实现了「保存但不污染对话」。第五步提示词 Promptprompts/main.py里是核心提示词模板。这是整个 Agent 的「大脑说明书」写得非常详细我挑几个关键点讲角色定义是「经验丰富的旅游规划师」。然后是约束比如•每次决策只用一种工具但可以用任意多次。•严格按工具参数说明生成 tool calls。•连续三次调用失败就求助不要死循环。•不得捏造信息DO NOT MAKE UP INFORMATION。•已得到的信息不反复查询。然后是「工作流程」把一次旅游规划拆成了 14 个步骤确定目的地 → 查景点 → 查经纬度 → 匹配同名地点 → 让用户选景点 → 规划每日行程 → 查交通 → 分析交通合理性 → 反思调整 → 出方案 → 查餐饮住宿 → 整合 → 出完整行程 → 用户确认。这个流程设计得非常讲究体现了前面学的高级概念反思、多步推理、HITL让用户选择、确认。它要求 Agent 每步输出「任务、回顾、思考、计划」四段思考过程严格区分「给用户的消息」和「调用工具的消息」。▲ 图12旅游规划机器人从需求到成型行程的完整工作流第六步把一切串起来 Graphgraph/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_asyncTrue): 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_namemodel_name, temperature0, prompt_templateagent_prompt_template, toolstools) 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_asyncTrue): graph create_graph(model_name, is_async) memory InMemorySaver() # 内存检查点实现多轮记忆 app graph.compile(checkpointermemory) # 编译 挂载检查点 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_namegpt-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 对你来说就不再是一个「听说过但没用过」的词了。 用图构建会思考的AI助手：LangGraph实战教程
企业AI全栈学习地图 · L3 Agent架构设计 | 构建自主决策的下一代AI助手 · 第三十六篇# LangGraph 保姆级教程用图把 AI Agent 的思考流程画出来再造一个能帮你规划旅游的机器人先说一个挺烦的事。很多教程讲 LangGraph一上来就甩一大堆术语State、Node、Edge、Reducer、Checkpointer、Subgraph……看完你脑子里一片浆糊只知道这是个「给大模型做工作流」的框架但到底怎么用、为什么要这么用还是蒙的。我换个讲法。你想过没有Deepseek 这类大模型你问一句话它答一句话本质是一条直线——输入进去输出出来结束。但你真正想要的 Agent是那种「能拆解问题、能调工具、能循环思考、能记住上次聊到哪」的东西。直线的模型做不到这些。LangGraph 干的事就是把这条直线变成一张「图」。图上有一个个节点Node节点之间有一条条边Edge数据在节点之间流转、积累、变形State。因为有了「图」你就能造出循环、分支、并行、暂停等直线模型做不到的控制流。▲ 图1LangGraph 把大模型的「直线问答」变成一张可控的「图」这篇文章我把 LangGraph 从头到尾给你讲明白一边讲一边给你代码让你能跑。讲完了我们真刀真枪搭一个「旅游规划机器人」——你告诉它想去哪它帮你查景点、查交通、查酒店餐厅、生成整套行程。完整项目我一行行拆给你看。先把丑话说在前面这篇文章很长但每一节都有用。你按顺序读读完就能自己搭一个能用的 Agent。先搞懂那三个最核心的词State、Node、EdgeLangGraph 的核心是把 Agent 的工作流程建模成一张「图」。这张图由三个关键组件定义第一个State状态这是整个图里所有节点共享的数据结构代表你应用「当前的状态」。它可以是任何 Python 类型但通常是TypedDict类型化字典或者 Pydantic 的BaseModel。你可以把它理解成一个「共享记事本」——每个节点都能往上面写也能从上面读。第二个Node节点一个 Python 函数。它接收当前状态作为输入执行一些计算或产生一些副作用最后返回「更新后的状态」。节点就是「干活的」。第三个Edge边决定「下一步去哪」的路径。它连接节点定义了信息也就是状态如何在节点之间流动以及图的执行顺序和逻辑分支。边可以是固定的正常边也可以是条件判断的条件边。一句话总结Node 干活Edge 指路State 是它们共享的记事本。▲ 图2LangGraph 三大核心要素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函数它决定了「节点返回的更新怎么作用到旧状态上」。如果你不指定 ReducerLangGraph 默认用「覆盖」——新值直接把旧值盖掉。这么说你可能没感觉看个实际场景。大模型LLM的聊天接口接受的是一个「消息列表」Message 对象列表。我们的图里通常要把对话历史当成消息列表存在状态里。如果你直接把messages设成一个普通 list不做任何处理那问题就来了节点 A 返回一条新消息旧的messages就被覆盖了——对话历史丢了Agent 就「失忆」了。所以对于messages这种需要「不断累积」的字段我们要给它指定一个 Reducer告诉图「每次更新是追加不是覆盖」。最简单的做法是用operator.addfrom operator import add from typing import TypedDict, Annotated class State(TypedDict): messages: Annotated[list, add] # add 表示追加当节点返回{messages: [新消息]}时旧的messages会上这个新列表实现累积。更规范的做法是用 LangGraph 自己提供的add_messagesfrom 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。▲ 图3Reducer 决定状态更新是「覆盖」还是「追加累积」这就是为什么旅游规划机器人项目里状态定义长这样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。▲ 图5Send 机制让节点扇出并行处理再扇入汇总Command一个节点同时更新状态 决定下一步有些时候你想在同一个节点里「既做状态更新又决定下一个要去哪个节点」。LangGraph 提供了Command对象来实现这个。你可以把Command理解成「加强版的条件边」。普通的条件边只能根据状态选节点而Command让节点函数自己返回一个「我下一步要去哪 我要更新什么状态」的完整指令把控制流边和状态更新节点揉在了一起。from langgraph.types import Command def my_node(state): return Command(update{some_key: some_value}, gotonext_node)这在做复杂路由、多智能体交接时特别有用。让 Agent 记住你Persistence 持久化与 Threads 线程这是 Agent 真正「像个人」的关键。一个没有记忆的 Agent每轮对话都是失忆的。LangGraph 用「检查点」Checkpoint来实现持久化。检查点器checkpointer会在每个超级步骤保存一份图状态的快照允许你在任何时间恢复。这就让「人机循环交互、内存管理、容错」这些特性成为可能。一个检查点StateSnapshot包含几个关键信息•config这个检查点关联的配置。•values此时各状态通道的值。•next下一个要执行的节点名称的元组。•tasks要执行的下个任务PregelTask 对象如果这步之前执行失败它会带上错误信息。那「线程」Threads又是什么线程就是检查点保存器给每个检查点分配的唯一 ID。你在调用带检查点的图时必须在配置的configurable部分指定thread_idconfig {configurable: {thread_id: user_123}} app.invoke({messages: [...]}, configconfig)一个thread_id代表「图与用户之间的一个独立会话」。同一个thread_id下的所有对话轮次、甚至单次图执行的步骤都会按这个 ID 组织起来。所以•同一个thread_id的多次调用 → 自动加载历史状态实现多轮记忆。•不同的thread_id→ 完全隔离的会话互不干扰。最常用的检查点器是InMemorySaver内存级进程重启就丢生产环境可以换成 Sqlite、RedisRedisSaver、MongoDB 等持久化存储。再往深一层LangGraph 还支持「跨线程持久化」——把用户信息姓名、偏好等存在共享内存里让新的对话线程也能复用。这就要用到「存储」Store的概念和检查点Checkpoint是两回事检查点是「会话内的状态快照」存储是「跨会话共享的数据」。配置持久化很简单编译图时传一个 checkpointer 进去即可▲ 图6检查点 thread_id 让 Agent 拥有「多轮记忆」from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() app graph.compile(checkpointermemory)两个让 Agent 更可控的高级特性Configuration把图做成「可插拔」LangGraph 允许你在创建图时把图的某些部分标记成「可配置的」。好处是运行时可以动态改这些部分的行为而不用改图的源代码。常见用例动态切换用哪个 LLM、改系统提示词内容、调整工具参数等。这让你能用一张「通用认知架构」也就是图本身通过不同配置实例化出多个行为略有不同的版本。Interrupt 与 HITL让人在关键时刻插一脚纯自动的 Agent 很危险尤其是涉及「花钱」「发消息」「对外操作」的动作。这时候你需要Human-in-the-loopHITL人机交互。interrupt函数允许你在图执行到特定点的时候「暂停」收集用户的输入、让用户验证或修改状态、或者做决策然后再继续执行。HITL 的三大关键用例1审查工具调用人在工具执行之前审查、编辑或批准 LLM 请求的工具调用。2验证 LLM 输出人审查、编辑或批准 LLM 生成的内容。3提供上下文让 LLM 明确请求人类输入来做澄清、补细节支持多轮对话。一个典型的场景是Agent 想调用「发送邮件」这个工具先在发之前interrupt把邮件内容给人看人点「批准」才真的发出去点「修改」就改一改再发或者输入一段反馈让 Agent 重新来。这跟另一个特性结合得很紧——▲ 图7HITL 让人在关键时刻暂停、审查、批准多智能体Multi-Agent让多个 Agent 分工协作当一个 LLM 应用太复杂把它拆成多个智能体每个负责一部分往往更清晰。在多智能体架构里每个智能体被表示成图里的一个节点。它们之间的交互最常见的模式是「交接」handoff——一个智能体把控制权转交给另一个智能体。交接时要指定两样东西•destination导航到目标 Agent也就是图里的节点名。•payload传递给那个 Agent 的信息也就是状态更新。交接可以用Command实现也可以用工具实现。更进一步可以构建「多智能体网络」——每个智能体都能跟所有其他智能体通信多对多连接自己决定下一个要调用谁。这在复杂协作场景里很有用。多智能体系统还要支持「多轮对话 HITL」关键机制是智能体需要用户输入时暂停图、收集输入、把输入作为新消息加入历史、再把控制权交还或路由到合适的智能体。▲ 图8多智能体通过「交接」把控制权转交给其他 AgentSubgraph 子图把图当乐高积木拼子图说白了就是「作为另一个图里的节点来使用的图」。这是「封装」这个古老思想在 LangGraph 上的应用。子图让你能构建「由多个自身就是图的组件」组成的复杂系统。一个常见用例是构建多智能体系统——每个智能体内部可能本身就是一个复杂的图。注意一个容易混淆的点有子图不一定是多智能体系统多智能体系统也不一定用了子图。这是两个正交的概念。用子图时核心问题是「父子图怎么通信」。有两种情况1父图和子图共享 Schema 的键直接加一个「包含编译后子图」的节点即可。2父图和子图 Schema 不同必须加一个节点函数来调用子图做状态的转换。子图同样能用Command把控制流和状态更新结合起来。另外给包含子图的图加持久性只需在编译父图时传一个 checkpointerLangGraph 会自动把它传播到所有子图。▲ 图9子图把「图」当作节点像乐高积木一样嵌套封装工具调用与 ToolNodeAI 应用通过自然语言和用户交互但光会聊天没意思我们期望 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 思考 → 调工具 → 看结果 → 再思考直到不再需要工具为止。▲ 图10ToolNode 构建起经典的 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旅游规划机器人的分层架构各司其职、环环相扣第一步定义状态 Statestates/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 知识在这里直接落地了。第二步造大模型 Modelmodels/factory.py里是一个工厂类from langchain_openai import ChatOpenAI class LLMFactory: staticmethod def get_llm(modelNone, temperatureNone): if model is None or temperature is None: raise ValueError(Both model and temperature must be specified.) if model.startswith(gpt): return ChatOpenAI(modelmodel, temperaturetemperature, streamingTrue) raise ValueError(fModel {model} is not supported.)这个工厂把「造模型」这件事封装起来根据传入的模型名和温度参数返回对应的 LLM 实例。streamingTrue表示启用流式输出逐块生成前端体验更好。以后想换成国产模型只需在这里改工厂逻辑别处不用动——这就是「Configuration 思想」的朴素实践。第三步定义 Agentagents/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(modelmodel_name, temperaturetemperature) 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_timeget_current_local_datetime(), first_user_messagestate[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_results10): 网络搜索工具。搜索关键词返回搜索结果列表 with DDGS() as ddgs: return [r for r in ddgs.text(keywordskeywords, regioncn-zh, max_resultsmax_results)]工具 2位置获取get_location_coordinate调高德地图地理编码 API根据地点名城市名查经纬度。因为可能有同名地点所以返回的是「所有同名地点的列表」让 Agent 自己判断选哪个。tool def get_location_coordinate(location, city): 位置获取工具。返回同名地点的经纬度列表 amap_key os.getenv(AMAP_API_KEY) url fhttps://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_mapstatic_map.py调高德静态地图 API根据经纬度生成一张地图图片。工具 7信息保存save_info_and_clear_history这个工具有点特别用到了response_formatcontent_and_artifacttool(response_formatcontent_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 保存实现了「保存但不污染对话」。第五步提示词 Promptprompts/main.py里是核心提示词模板。这是整个 Agent 的「大脑说明书」写得非常详细我挑几个关键点讲角色定义是「经验丰富的旅游规划师」。然后是约束比如•每次决策只用一种工具但可以用任意多次。•严格按工具参数说明生成 tool calls。•连续三次调用失败就求助不要死循环。•不得捏造信息DO NOT MAKE UP INFORMATION。•已得到的信息不反复查询。然后是「工作流程」把一次旅游规划拆成了 14 个步骤确定目的地 → 查景点 → 查经纬度 → 匹配同名地点 → 让用户选景点 → 规划每日行程 → 查交通 → 分析交通合理性 → 反思调整 → 出方案 → 查餐饮住宿 → 整合 → 出完整行程 → 用户确认。这个流程设计得非常讲究体现了前面学的高级概念反思、多步推理、HITL让用户选择、确认。它要求 Agent 每步输出「任务、回顾、思考、计划」四段思考过程严格区分「给用户的消息」和「调用工具的消息」。▲ 图12旅游规划机器人从需求到成型行程的完整工作流第六步把一切串起来 Graphgraph/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_asyncTrue): 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_namemodel_name, temperature0, prompt_templateagent_prompt_template, toolstools) 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_asyncTrue): graph create_graph(model_name, is_async) memory InMemorySaver() # 内存检查点实现多轮记忆 app graph.compile(checkpointermemory) # 编译 挂载检查点 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_namegpt-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 对你来说就不再是一个「听说过但没用过」的词了。