本文是《AI Agent 实战》系列第 2 篇。上篇我们讲了 Agent 与普通大模型的区别普通大模型主要会回答而 Agent 会执行。本篇解决其中最硬核的一个能力——任务规划。读完你会亲手跑通一个会拆任务 → 逐个执行 → 根据结果修正计划的 Agent全程代码可复制。写在前面先看一个真实对比。你问大模型帮我调研一下 A 产品近一个月的用户口碑给出结论。普通大模型输出一段调研方法论——建议你去看评论、做统计、注意样本偏差。然后就没有然后了。任务规划 Agent把目标拆成搜口碑关键词 → 读评论数据 → 计算占比 → 生成结论四个子任务逐个调用工具执行中途发现数据不够还会自己补一条任务最后带着数据把结论交给你。区别不在于模型聪明程度而在于有没有一个东西管着先做什么、后做什么、做错了怎么办。这个东西就是本篇的主角——LangGraph。先交代版本背景避免你抄代码踩坑LangGraph 1.0 已于 2025 年 10 月正式发布官方承诺 2.0 之前不再做破坏性变更。但网上大量 2024—2025 年初的教程基于 0.x API其中最流行的langgraph.prebuilt.create_react_agent已被官方标记废弃由 LangChain 1.0 的create_agent取代。本文全部按 1.x 写法这也是和大多数中文教程最大的区别。最终效果预览我的运行日志发布前建议自己也跑一遍截图[planner] 拆解出 3 个子任务搜索口碑数据 / 读取本地评论文件并统计 / 计算好评占比并总结 [executor] 完成子任务 1检索到近一月口碑关键词…… [executor] 完成子任务 2读取 reviews.txt共 46 条评论…… [executor] 完成子任务 3好评率 78.3%主要吐槽集中在续航和客服响应…… [replanner] 信息充分生成最终答复 [answer] 结论A 产品近一月口碑整体偏正面好评率约 78%核心卖点……但续航是最大风险点……这就是一个完整的感知—规划—执行—修正—输出闭环。下面开讲。一、为什么任务规划需要专门的编排框架让 LLM 输出一份计划并不难难的是执行计划。一次性问答是生成而任务规划是循环拆任务 → 执行一步 → 检查结果 → 继续/补任务/收尾 → 循环直到目标达成一旦引入循环你会立刻撞上三个工程问题状态管理走到第 3 步时前 2 步的产出存在哪里流程控制什么时候继续、什么时候结束、什么时候回头补做中断恢复任务跑到一半进程崩了或者中间需要人工审批怎么接着跑拿while循环手撕当然也能做但状态散落在变量里、断了就全丢、出错很难定位——这就是很多人写的 Agent 像意大利面条的原因。业界目前主流有两条路线路线做法优点缺点Prompt 路线一次让模型输出完整计划再逐条执行实现简单执行中不能变通计划错了只能重跑图编排路线把规划/执行/重规划建成图的节点用状态在节点间流转可控、可持久化、可插人工审批需要学习框架概念LangGraph 是图编排路线的代表底层采用 Pregel 式的超步执行模型每一步并行运行当前要执行的节点然后统一更新共享状态。Uber、LinkedIn、Klarna 等公司都把它用在生产环境。本篇采用经典的Plan-and-Execute结构【图 1架构示意图】flowchart LR S((开始)) -- P[Plannerbr/把目标拆成子任务] P -- E[Executorbr/执行队首子任务] E -- R[Replannerbr/继续执行/补充任务/收尾] R --|还有任务| E R --|信息已足够| F((结束br/输出结论))建议导出 PNG 作为正文配图和封面图比 mermaid 源码直接在 CSDN 渲染更稳。二、LangGraph 三个核心概念只讲本篇用得上的三个词State、Node、Edge。State状态整个图的共享工作记忆一个字典定义所有节点都要读写的字段。每个节点收到当前状态返回增量更新。Node节点一个普通 Python 函数输入状态、输出状态更新。节点内部想调 LLM 就调 LLM想查数据库就查数据库——LangGraph 不关心你节点里干什么它只管流转。Edge边与条件路由从直线工作流到自主循环的关键固定边写死A 之后走 Badd_conditional_edges则允许你放一个函数根据当前状态决定下一步去哪个节点——还剩任务就继续执行没剩任务就结束就是靠它实现的。【图 2三概念对照示意图——可自绘左边画 State 字段表中间画三个节点函数右边画连边规则】⚠️ 版本避坑如果你搜到的教程里有from langgraph.prebuilt import create_react_agent那是旧写法官方已宣布废弃。1.x 环境下一行创建执行型 Agent 的正确姿势是from langchain.agents import create_agent。三、环境准备python -m venv venv # Windows venv\Scripts\activate # macOS/Linux source venv/bin/activate pip install langgraph langchain langchain-openai要求 Python 3.10 以上。模型怎么选langchain-openai里的ChatOpenAI实际是OpenAI 兼容协议客户端DeepSeek、通义千问、Kimi 等国产模型都能直接接改一行base_url即可。本文以 DeepSeek 为例import os from langchain_openai import ChatOpenAI llm ChatOpenAI( modeldeepseek-chat, temperature0, base_urlhttps://api.deepseek.com/v1, # 通义/Kimi 同理换成各自兼容端点 api_key***DEEPSEEK_API_KEY), # 环境变量里放密钥别写死在代码里 )本篇用到的 3 个工具web_search搜索、read_file读本地评论数据文件、calculator纯本地计算。前两个用演示实现不依赖任何收费服务先把流程跑通。冒烟测试确认环境没问题from langgraph.graph import StateGraph, START, END from typing import TypedDict class Mini(TypedDict): msg: str g StateGraph(Mini) g.add_node(echo, lambda s: {msg: s[msg] OK}) g.add_edge(START, echo) g.add_edge(echo, END) print(g.compile().invoke({msg: hello})) # 预期输出 {msg: hello OK}四、实战从零搭一个任务规划 Agent4.1 定义状态 TaskStateimport operator from typing import Annotated, TypedDict class TaskState(TypedDict): objective: str # 用户的原始目标 plan: list[str] # 还排着队、没执行的子任务 past_steps: Annotated[list[tuple], operator.add] # 已完成的 (任务, 结果) response: str # 最终答复两个细节plan字段节点返回什么就整个替换什么而past_steps用了Annotated[..., operator.add]节点返回的更新会追加到列表而不是覆盖。这个 reducer 机制是新手最容易栽的地方后面报错环节还会遇到。节点返回的永远只是我要改哪些字段不是整个状态。4.2 Planner 节点把目标拆成子任务规划最怕模型输出自由散文没法解析。正确做法是结构化输出让模型按 Pydantic schema 返回 JSON。from pydantic import BaseModel, Field class Plan(BaseModel): 为完成总体目标需要按顺序执行的计划。 subtasks: list[str] Field(description2~5 个子任务每个具体到可单独完成) planner_llm llm.with_structured_output(Plan) planner_prompt ( 你是任务规划器。把用户目标拆成 2~5 个可执行的子任务 每个子任务具体、可单独完成、不重叠、不遗漏。 ) def plan_step(state: TaskState) - dict: plan planner_llm.invoke( [ (system, planner_prompt), (user, state[objective]), ] ) return {plan: plan.subtasks, past_steps: [], response: }实际拆分效果示例——输入目标调研 A 产品近一个月的用户口碑并给出结论得到{ subtasks: [ 检索 A 产品近一个月的口碑关键词和主流评价观点, 读取本地评论数据文件统计正负面评论分布, 计算好评占比结合检索结果给出综合结论 ] }记住一句话规划要靠结构约束不靠提示词玄学。with_structured_output才是工程做法。4.3 Executor 节点逐个执行子任务先定义 3 个工具。tool装饰器会根据函数签名和 docstring 自动生成工具调用的 JSON Schema所以docstring 要写清楚——模型就是靠它决定什么时候调用这个工具。from langchain.tools import tool import ast, operator as op tool def read_file(path: str) - str: 读取本地文本文件的完整内容。 with open(path, r, encodingutf-8) as f: return f.read() tool def web_search(query: str) - str: 联网检索指定关键词的最新信息。 # 演示实现接生产环境时替换为真实搜索 API如博查、Tavily 等 return (f[演示数据] 检索『{query}』近一月 A 产品好评率约 78% f praise 集中在影像和屏幕吐槽集中在续航与客服响应。) tool def calculator(expression: str) - str: 计算一个数学表达式支持加减乘除和括号。 OPS {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv} def ev(n): if isinstance(n, ast.Constant): return n.value if isinstance(n, ast.BinOp): return OPS[type(n.op)](ev(n.left), ev(n.right)) raise ValueError(不支持的表达式) return str(ev(ast.parse(expression, modeeval).body))calculator用 AST 白名单解析而不是裸eval避免任意代码执行——这是给读者的一个安全示范。执行器本身也是一个会思考—调工具—观察结果循环的小 Agent。这一层不用手撕官方预置就行from langchain.agents import create_agent executor_agent create_agent( modelllm, tools[read_file, web_search, calculator], system_prompt你是任务执行器。只完成交给你的这一个子任务基于工具真实结果作答不要臆测。, ) def execute_step(state: TaskState) - dict: current_task state[plan][0] result executor_agent.invoke( {messages: [( user, f总体目标{state[objective]}\n f当前子任务{current_task}\n f请完成该子任务输出简明结果。, )]}, {recursion_limit: 20}, ) return { past_steps: [(current_task, result[messages][-1].content)], plan: state[plan][1:], }注意职责分离执行器只看得到自己手里这一步全局观由 Planner 和 Replanner 负责。这是图编排的核心思想——每个节点职责单一流程才可测试、可替换。4.4 Replanner 节点 条件边决定继续还是收尾from typing import Literal class Action(BaseModel): action: Literal[continue, final] Field( descriptioncontinue继续执行队列里的下一个子任务final信息已足够可以作答) added_tasks: list[str] Field(default_factorylist, description发现计划缺口时补充的新子任务) response: str Field(default, descriptionaction 为 final 时的最终答复) replanner_llm llm.with_structured_output(Action) MAX_STEPS 6 # 防死循环兜底 def replan_step(state: TaskState) - dict: if len(state[past_steps]) MAX_STEPS: return {plan: [], response: 已达最大步数输出当前已有结果。} steps_text \n.join(f- {t}{r} for t, r in state[past_steps]) or 暂无 out replanner_llm.invoke( [ (system, 根据已完成步骤判断信息够回答就 final不够就 continue必要时补充新任务。), (user, f总体目标{state[objective]}\n已完成步骤\n{steps_text}\n剩余任务{state[plan]}), ] ) if out.action final: return {plan: [], response: out.response} return {plan: state[plan] out.added_tasks}配套的MAX_STEPS、补充任务数量上限是防止 Agent 陷入无限重规划死循环的工程兜底。生产上这三件套最大步数、最大子任务数、token 预算一个都不能少——这也是付费篇会展开的细节。然后是路由函数整个 Agent 的大脑回路就在这几行from langgraph.graph import END def route(state: TaskState) - str: if state[plan]: # 队列里还有任务 → 继续执行 return executor return END # 否则收工4.5 组装、编译、运行from langgraph.graph import StateGraph, START builder StateGraph(TaskState) builder.add_node(planner, plan_step) builder.add_node(executor, execute_step) builder.add_node(replanner, replan_step) builder.add_edge(START, planner) builder.add_edge(planner, executor) builder.add_edge(executor, replanner) builder.add_conditional_edges(replanner, route, {executor: executor, END: END}) graph builder.compile()【图 3本 demo 完整流程图——与图 1 相同结构可加上条件路由的菱形判断节点后导出 PNG】用stream模式跑能逐节点看到 Agent 的每一步动作objective 调研 A 产品近一个月的用户口碑并给出结论 for chunk in graph.stream({objective: objective, plan: [], past_steps: [], response: }): for node, update in chunk.items(): if node planner: print(f[{node}] 拆出子任务{update[plan]}) elif node executor: t, r update[past_steps][-1] print(f[{node}] 完成『{t}』→ {r[:60]}…) elif node replanner and update.get(response): print(f[{node}] {update[response]})发布前把这段的真实运行输出截图替换掉本文示例。到这里你的第一个任务规划 Agent 已经能跑了。但说实话它还只是个玩具——进程一断全丢敏感操作没人审批。下面加两块生产级的拼图。五、让它更像生产可用的两个升级升级 1状态持久化Checkpointer编译时挂一个 checkpointer再用thread_id标记会话任务状态就按检查点保存下来——中断后可以从任意一步恢复官方把这叫时间旅行回滚到某一步改状态重跑。from langgraph.checkpoint.memory import MemorySaver graph builder.compile(checkpointerMemorySaver()) # 生产可换 SQLite/Postgres 版本 config {configurable: {thread_id: task-001}} result graph.invoke({objective: objective, plan: [], past_steps: [], response: }, config) # 进程内继续同一个任务直接换个输入状态还在这就解决了第二节提的状态管理、中断恢复两个痛点。持久化策略有 sync / async / exit 三档按需选择。升级 2人工确认环节interrupt写文件、发消息、下单这类有副作用的动作绝不能让 Agent 说干就干。LangGraph 1.x 提供原生interrupt在任意节点暂停等人工给指令再用Command(resume...)恢复。from langgraph.types import interrupt def write_report(state: TaskState) - dict: decision interrupt({ question: 即将把结论写入报告文件是否批准, draft: state[response], }) if decision ! approve: return {response: 用户驳回未写入。} with open(report.md, w, encodingutf-8) as f: f.write(state[response]) return {response: 报告已写入。}from langgraph.types import Command # 第一次跑到 interrupt 处会自动暂停并抛出审批内容 # 人工看过 draft 之后 graph.invoke(Command(resumeapprove), config)呼应上篇讲的原则Agent 的能力边界要用人工闸门圈出来。会规划、会执行是本事知道哪一步停下来等人才是生产级。六、常见报错与调试清单按报错 → 原因 → 解法整理全部是我踩过的GraphRecursionError: maximum recursion depth exceeded原因条件边成环且路由函数没有终止分支。解法调用时设{recursion_limit: 25}并检查 route 是否存在必然到达 END 的路径规划类任务务必配 MAX_STEPS 兜底。past_steps只剩最后一步 / 报Please add a reducer原因列表字段没配 reducer节点返回值把旧数据整个覆盖了。解法Annotated[list, operator.add]或明确接受覆盖语义。ImportError: cannot import name create_react_agent/ 废弃警告原因旧教程代码直接搬进 1.x 环境。解法改用from langchain.agents import create_agent。执行器从不调用工具tool_calls为空原因所用模型或端点不支持 function calling 兼容模式。解法确认模型文档实在不支持就降级——把工具结果直接拼进提示词用结构化输出替代工具调用。结构化输出解析失败 / 输出跑偏解法三连温度调到 0、Field 描述写具体、外层加一次重试。Windows 下运行报事件循环错误原因asyncio 与部分库在 Windows 的兼容问题。解法优先用同步invoke/stream接口或改用 WSL。调试方法论三句话先graph.get_graph().draw_mermaid()看拓扑是否和你设计一致再用stream模式看每个节点更新后的完整状态循环和状态流转复杂时直接上 LangGraph Studio 可视化跑。七、成本估算很多人跑通 demo 就停了一上量发现 token 消耗惊人。这里给一个估算框架数字发布前以你的实测为准本 demo 完成一个 3 子任务目标Planner Replanner 3 次 Executor 循环合计约 ____ tokens按 DeepSeek 官方定价折合不到 ____ 元。实测口径建议记录在运行日志里。⚠️ 占位发布前请实跑一次回填真实数值并核对模型平台当时的实时价格成本大头不是单次调用而是Replanner 每次携带全量past_steps造成的上下文滚雪球。任务一多第 10 步的输入可能是第 1 步的十倍。想省钱跑流程可以换本地 Ollama 小参数模型。但要有预期小模型的规划和工具调用质量会明显下降——免费能跑和能干活是两回事。上下文压缩、状态裁剪、缓存复用这些降本手段我会整理进后续的进阶篇。总结一句话收束LangGraph 把 Agent 从一次生成变成一张可持久化、可插人审的状态图。本篇你完成了理解了任务规划为什么需要图编排而不是 while 循环手撕用 State / Node / Edge 三件套搭起 Plan-and-Execute 结构的任务规划 Agent掌握了结构化输出、条件路由、防死循环兜底三个关键工程手法加上 checkpointer 和 interrupt把玩具升级成具备恢复能力和人工闸门的生产雏形。下一篇预告《Dify 实战零代码搭建知识库问答 Agent》——不想写代码怎么落地 Agent我们下一篇见。完整的可运行工程含示例评论数据文件我打包好了评论区回复LangGraph领取。有问题也欢迎在评论区贴你的报错和模型选择我会挑典型问题统一解答。