
简介这份资源是公众号「AI喵智能体」发布的 LangGraph10 实战系列配套代码包面向希望从零手搓可控 Agent 的开发者与学习者尤其适合已具备 Python 基础、想深入理解 LangGraph 编排与 LCEL 表达式的进阶人群。包内共 7 个文件以 3 个 py 脚本为核心分别对应简单对话、动态提示词与 LCEL 基础等章节示例另附说明文档、readme 与 gitignore 等辅助文件压缩包约 41KB体量轻便、结构清晰便于按章节顺序阅读与运行。目前已有 88 人学习下载。通过这份代码读者可以对照章节逐步搭建可控 Agent 的最小可运行骨架理解节点、边与状态流转的写法并借助 LCEL 基础示例掌握链式组合思路为后续扩展更复杂的多步推理与工具调用打下基础适合作为系列实战的起步参考。1. 从零手搓可控 Agent为什么我不建议你直接套现成框架很多人第一次接触 LangGraph是因为听说它能做「可控 Agent」——状态机、循环、条件边、人工介入听起来比黑盒式的 Agent 框架靠谱得多。但真正上手时大部分教程给的是「调 API 拼节点」的玩具 demo跑通一次就再也复现不了。这个系列代码项目要解决的恰恰是这个问题用 LangGraph 从零搭一个可控 Agent把状态、路由、工具调用、记忆、中断恢复这几件事拆开讲清楚而不是丢一个封装好的 AgentExecutor 让你猜里面发生了什么。如果你已经会写 Python、用过至少一个 LLM API、对「Agent 是什么」有模糊概念但说不清它和普通链式调用的区别这篇就是给你写的。我会按「先立住概念 → 再跑通最小闭环 → 再补工具和记忆 → 最后处理中断和踩坑」的顺序推进每一步都给出可抄的代码和参数说明。读完你应该能自己判断什么时候该用 LangGraph什么时候用普通函数调用就够了。2. LangGraph 的可控 Agent 到底控制了什么2.1 状态图模型Agent 的本质是一个带循环的有向图普通 LLM 调用是「输入 → 输出」的一次性映射。Agent 的区别在于它要多轮决策先想一步决定调不调工具调完看结果再决定下一步。LangGraph 把这个过程建模成一张状态图StateGraph节点是函数边是流转条件整张图共享一个状态对象。这个模型的价值在于「可控」二字。传统 Agent 框架把「思考-行动-观察」循环藏在内部你只能通过 prompt 和回调去影响它出了问题很难定位。LangGraph 把循环显式画出来你能看到每一步状态怎么变、为什么走这条边、在哪一步卡住。这就是热搜里常说的 agent 架构和 agent 编排的区别——编排是让多个 Agent 协作架构是单个 Agent 内部怎么组织。核心概念只有四个概念作用对应代码State全局共享的数据结构TypedDict / PydanticNode处理状态的函数普通 Python 函数Edge节点间的固定流转add_edgeConditional Edge按条件选择下一节点add_conditional_edges理解这四样LangGraph 就没有玄学了。剩下的都是在这四个概念上做工程。2.2 最小可运行闭环三个节点跑通一次工具调用先不追求功能完整把「LLM 决策 → 调工具 → 回填结果 → 再决策」这个最小闭环跑通。下面这段代码是整篇的基础后面所有扩展都从这里长出来。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool # 1. 定义状态messages 用 add_messages 做累加而不是覆盖 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 2. 定义一个最简单的工具 tool def get_weather(city: str) - str: 查询指定城市的天气city 为城市名 fake_db {北京: 晴 12℃, 上海: 多云 18℃} return fake_db.get(city, 暂无数据) tools [get_weather] llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) # 3. 节点一LLM 决策 def call_model(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} # 4. 节点二执行工具 def call_tools(state: AgentState): last state[messages][-1] results [] for call in last.tool_calls: tool_fn {t.name: t for t in tools}[call[name]] output tool_fn.invoke(call[args]) results.append(ToolMessage(contentstr(output), tool_call_idcall[id])) return {messages: results} # 5. 路由函数有工具调用就去执行否则结束 def should_continue(state: AgentState): last state[messages][-1] return tools if getattr(last, tool_calls, None) else END # 6. 组装图 graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_node(tools, call_tools) graph.set_entry_point(agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) graph.add_edge(tools, agent) # 工具执行完回到 agent 再决策 app graph.compile() result app.invoke({messages: [HumanMessage(content北京天气怎么样)]}) print(result[messages][-1].content)逻辑说明add_messages这个 reducer 是关键它让每次节点返回的新消息追加到列表而不是替换否则多轮对话历史会丢。should_continue是条件边的路由函数返回字符串决定走哪条边。tools → agent这条回边构成了循环这是 Agent 区别于普通链的本质。参数说明temperature0在 Agent 场景几乎必设决策需要稳定bind_tools把工具 schema 注入 LLM模型返回的tool_calls里带name、args、idid必须原样回填到ToolMessage否则下一轮会报错。这套最小闭环大约 60 行跑通它比读十篇概念文有用。3. 把工具、记忆和中断恢复接进状态图3.1 工具注册的三个必调参数工具不是随便写个函数加tool就行。生产里翻车最多的三个点描述、参数类型、错误处理。from langchain_core.tools import tool from pydantic import BaseModel, Field class SearchInput(BaseModel): query: str Field(description搜索关键词尽量具体) top_k: int Field(default3, description返回条数1-10) tool(args_schemaSearchInput) def search_docs(query: str, top_k: int 3) - str: 在内部知识库中检索文档。当用户询问产品细节、政策条款时使用。 try: # 实际检索逻辑 return f检索到 {top_k} 条关于 {query} 的结果 except Exception as e: return f检索失败{e}请换个关键词重试第一个参数是description函数 docstring。LLM 靠它判断什么时候用这个工具写得含糊模型就乱调。第二个是args_schema用 Pydantic 明确每个字段的类型和说明比裸函数签名可靠得多尤其是带默认值的可选参数。第三个是错误处理工具内部抛异常会直接中断整个图返回错误字符串让 LLM 自己决定重试还是换策略这才是 Agent 该有的韧性。3.2 短期记忆与长期记忆checkpointer 怎么选Agent 记忆分两层。短期记忆是当前会话的消息历史靠 checkpointer 持久化长期记忆是跨会话的事实通常存向量库或 KV。from langgraph.checkpoint.memory import MemorySaver # 生产环境换成 PostgresSaver / SqliteSaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-001}} app.invoke({messages: [HumanMessage(content我叫小明)]}, config) # 同一 thread_id 的第二次调用能记住上文 app.invoke({messages: [HumanMessage(content我叫什么)]}, config)thread_id是记忆的隔离键同一个 id 共享历史不同 id 互不干扰。开发阶段用MemorySaver够了进程重启就丢上线必须换SqliteSaver或PostgresSaver否则用户刷新页面记忆就没了。这里有个血泪经验checkpointer 存的是整个状态快照如果状态里塞了大对象比如完整文档数据库会迅速膨胀该裁剪的要裁剪。3.3 中断与人工介入interrupt_before 的正确用法可控 Agent 最有价值的能力之一是「关键步骤前暂停等人确认」。LangGraph 用interrupt_before实现。app graph.compile( checkpointermemory, interrupt_before[tools] # 执行工具前暂停 ) config {configurable: {thread_id: t1}} app.invoke({messages: [HumanMessage(content删除所有日志)]}, config) # 此时图停在 tools 节点前人工审查 state app.get_state(config) print(state.next) # (tools,) # 确认后继续 app.invoke(None, config) # 传 None 表示从断点恢复interrupt_before接受节点名列表图会在进入这些节点前停下状态被 checkpointer 保存。恢复时invoke(None, config)即可。注意interrupt_before和interrupt_after的区别前者在节点执行前停适合「危险操作前确认」后者在执行后停适合「结果审查」。两者都需要 checkpointer没有持久化就没法恢复。4. 可控 Agent 的避坑与排查清单4.1 无限循环Agent 停不下来现象Agent 反复调用同一个工具或者两个节点来回跳直到 token 烧完。原因路由函数没有终止条件或者工具返回的结果让 LLM 认为「还没完成」。常见于工具返回空字符串、报错信息模糊、或者 prompt 里没告诉模型「拿到结果就总结」。解决给状态加一个steps计数器路由函数里判断超过阈值强制走 END工具返回必须包含明确的结果或错误不要返回空system prompt 里写清「工具返回结果后直接用自然语言回答用户不要重复调用」。class AgentState(TypedDict): messages: Annotated[list, add_messages] steps: int def should_continue(state: AgentState): if state.get(steps, 0) 8: return END last state[messages][-1] return tools if getattr(last, tool_calls, None) else END4.2 状态被覆盖消息历史莫名丢失现象多轮对话后Agent 忘了前面说过的话或者工具结果没进历史。原因状态字段没用 reducer。TypedDict里如果直接写messages: list每次节点返回都会替换整个列表。必须用Annotated[list, add_messages]声明累加语义。解决所有需要累积的字段都加 reducer。自定义 reducer 也可以比如去重、按时间排序。检查方法在节点里 print 一下len(state[messages])如果一直是 1就是 reducer 没生效。4.3 工具调用报 tool_call_id 不匹配现象报错tool_call_id not found或Message does not have a tool_call_id。原因ToolMessage的tool_call_id必须和 LLM 返回的tool_calls[i][id]完全一致。手动构造消息时容易写错或者一次返回多个工具调用只回填了一个。解决遍历last.tool_calls逐个构造ToolMessage不要硬编码 id。如果模型一次调多个工具必须全部回填缺一个下一轮就报错。4.4 checkpointer 导致状态膨胀现象用 SqliteSaver 跑几天后数据库几个 G查询变慢。原因每次 invoke 都存一份完整状态快照消息历史越滚越长。解决定期裁剪历史只保留最近 N 轮或者用trim_messages在进 LLM 前截断但注意裁剪只影响传给模型的内容不影响 checkpointer 存储。更彻底的做法是分离「对话历史」和「工作状态」只持久化必要的部分。4.5 流式输出和中断恢复冲突现象用stream流式输出时interrupt_before不生效或恢复后重复输出。原因流式模式下事件是逐 token 推送的中断点判断和普通 invoke 不同。解决需要人工介入的场景用invoke或stream_modevalues不要用 token 级流式恢复时从 checkpointer 读状态不要重放整个输入。这个坑比较隐蔽建议先在非流式下把中断逻辑调通再考虑流式。5. 进阶用子图拆分复杂 Agent 与验证可控性当单个 Agent 的工具超过十个、逻辑分支超过五条状态图会变成一团乱麻。这时候该上子图subgraph把一组相关节点打包成独立图作为父图的一个节点。# 子图专门处理订单相关操作 order_graph StateGraph(OrderState) order_graph.add_node(parse, parse_order) order_graph.add_node(execute, execute_order) order_graph.set_entry_point(parse) order_graph.add_edge(parse, execute) order_graph.add_edge(execute, END) order_subgraph order_graph.compile() # 父图把子图当一个节点用 main_graph StateGraph(MainState) main_graph.add_node(router, route_intent) main_graph.add_node(order, order_subgraph) # 子图直接作为节点 main_graph.add_node(chat, general_chat) main_graph.add_conditional_edges(router, pick_branch, { order: order, chat: chat })子图的好处是状态隔离订单处理的状态不会污染主对话状态调试时也能单独跑子图。注意父子图的状态 schema 要兼容父图传给子图的字段子图必须能接住否则会静默丢数据。验证可控性有个实用方法给每个节点加日志记录输入状态的关键字段和输出然后跑一批测试用例看路由路径是否符合预期。可控不是嘴上说的是能画出每次执行的完整路径图。我一般会在 compile 前给每个节点包一层装饰器统一打点def traced(name): def deco(fn): def wrapper(state): print(f[{name}] in: {list(state.keys())}) out fn(state) print(f[{name}] out: {list(out.keys())}) return out return wrapper return deco graph.add_node(agent, traced(agent)(call_model))这套打点在生产里救过我很多次——Agent 行为异常时第一件事就是看路径而不是改 prompt。路径对了再调 prompt路径错了改的是图结构两者别混。最后说个习惯我搭任何 Agent 都先写「最小闭环 打点」跑通再加工具和记忆绝不一次性堆完。LangGraph 的图结构一旦复杂出问题很难二分定位增量搭建是唯一靠谱的后悔药。希望帮到你。本文还有配套的精品资源点击获取