简介本资源是LangGraph10实战系列中「从零手搓可控Agent」的配套代码项目面向已具备Python基础、希望深入掌握Agent编排与可控流程的开发者与AI应用学习者。内容围绕LangGraph核心机制展开从最基础的对话交互到动态提示词构建再到LCEL表达式语言的基础用法层层递进地演示如何搭建一个可干预、可追踪的Agent流程帮助读者理解状态管理、节点编排与提示词动态注入等关键环节。压缩包共7个文件包含3个Python脚本、1个说明文本、1份docx文档、1个md说明及gitignore配置整体约41KB体量轻巧但结构完整便于快速导入运行与二次修改。目前已有88人学习关注适合作为LangGraph入门到进阶的练手素材读者可借此梳理Agent项目的目录组织方式掌握从简单对话到动态提示的代码实现路径并在此基础上扩展自己的可控Agent方案。1. 从零手搓可控 Agent这套 LangGraph 实战代码到底能跑出什么很多人第一次接触 Agent 开发都是从调一个现成框架的 API 开始的——输入一句 prompt等模型返回一段 JSON然后手动解析、手动拼工具调用。跑通 demo 那一刻挺爽但一旦要加多轮记忆、条件分支、人工审核节点代码就开始失控状态散落在十几个变量里调试靠 print改一处崩三处。这套《LangGraph10 实战从零手搓可控 Agent》系列代码解决的正是这个从「能跑」到「可控」的断层。它不讲空泛的 Agent 概念而是用 LangGraph 的图结构把 Agent 的每一步显式建模成节点和边让状态流转、条件路由、循环终止都变成可读、可测、可干预的代码。适合已经会写 Python、调过 LLM API但在多步骤 Agent 编排上反复翻车的开发者。下面我按实际拆包和复现的顺序把这份资源的技术骨架、运行方式、参数边界和踩坑点一次讲透。2. LangGraph 的图编排模型为什么用 StateGraph 而不是 while 循环2.1 从「链式调用」到「状态图」的选型理由传统 Agent 写法通常是一个 while 循环套 if-else判断当前该调工具还是该回复用户调完工具把结果塞回 messages再进下一轮。这种写法在单工具、单轮场景下没问题但一旦出现「先检索、再判断是否需要追问、追问后重新检索」这种带环路的逻辑循环条件就会变得极其脆弱。LangGraph 的核心抽象是 StateGraph你定义一个状态结构通常是 TypedDict 或 Pydantic 模型然后声明若干节点函数每个节点接收当前状态、返回状态更新再用边edge和条件边conditional edge决定下一步走哪个节点。这样做的直接好处是Agent 的执行路径变成了一张有向图你可以可视化、可以在任意节点插入断点、可以用 checkpointer 做状态持久化。常见做法是先用StateGraph(AgentState)初始化图然后add_node注册节点add_edge连固定跳转add_conditional_edges连需要判断的分支。最后compile()出一个可执行对象调用时传入初始状态即可。2.2 定义状态与节点一份可抄的最小骨架下面这段代码是这类项目里最常见的状态定义和节点注册方式我按实际拆包看到的模式还原了一个最小可运行版本from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages # 状态定义messages 用 add_messages 做增量合并而不是覆盖 class AgentState(TypedDict): messages: Annotated[list, add_messages] next_action: str # 路由标记tool / respond / human_review retry_count: int # 重试计数防止死循环 def call_model(state: AgentState): 调用 LLM根据当前 messages 决定下一步动作 response llm.invoke(state[messages]) # 把模型返回的 tool_calls 或文本写回状态 return {messages: [response], next_action: decide_next(response)} def run_tool(state: AgentState): 执行工具调用把结果作为 ToolMessage 追加 last state[messages][-1] result execute_tool(last.tool_calls[0]) return {messages: [result], retry_count: state[retry_count] 1} def route_after_model(state: AgentState): 条件路由根据 next_action 决定走工具还是结束 if state[next_action] tool and state[retry_count] 5: return tool return END # 构图 graph StateGraph(AgentState) graph.add_node(model, call_model) graph.add_node(tool, run_tool) graph.set_entry_point(model) graph.add_conditional_edges(model, route_after_model, {tool: tool, END: END}) graph.add_edge(tool, model) # 工具执行完回到模型形成闭环 app graph.compile()这段代码里几个参数值得单独说。Annotated[list, add_messages]是 LangGraph 提供的 reducer作用是让每次节点返回的 messages 追加到已有列表而不是整体替换——如果你直接写messages: list第二轮就会把历史消息全丢掉这是新手最常见的翻车点之一。retry_count配合条件边里的 5判断是防止 Agent 在工具调用失败时无限循环的硬性刹车。set_entry_point指定入口节点add_edge(tool, model)把工具节点连回模型节点形成「模型决策 → 执行工具 → 再决策」的闭环。整个图编译后返回的app支持.invoke()、.stream()和.astream_events()调试时用 stream 能逐节点看到状态变化。2.3 条件边与循环终止把「玄学」变成可观测逻辑Agent 开发里最让人头疼的就是「它为什么又绕回去了」。在 while 循环写法里你只能靠日志猜在 StateGraph 里条件边函数是纯函数输入是当前状态输出是下一个节点名。你可以在条件边里打印状态、可以写单元测试断言路由结果、可以在 retry_count 超限时强制走 END。我一般会在条件边函数里加一行print(f[route] action{state[next_action]} retry{state[retry_count]})跑几次就能看清 Agent 的决策路径。如果发现它反复走同一个分支大概率是工具返回结果没有被正确写回 messages或者模型没有正确解析工具输出——这时候去看 ToolMessage 的 content 和 tool_call_id 是否匹配比盲目调 prompt 有效得多。3. 工具调用与状态持久化让 Agent 记住上下文并支持中断恢复3.1 工具注册与 ToolNode 的接入方式LangGraph 生态里工具调用有两种常见接法一种是手写run_tool节点自己解析tool_calls并执行另一种是直接用预置的ToolNode把工具列表传进去它会自动处理调用和结果回写。手写的好处是可控性强你可以在执行前后加日志、加权限校验、加超时控制用 ToolNode 的好处是省事适合工具数量多、逻辑简单的场景。这份实战代码里两种都有涉及我建议新手先用 ToolNode 跑通再逐步替换成手写节点来加深理解。from langgraph.prebuilt import ToolNode from langchain_core.tools import tool tool def search_docs(query: str) - str: 根据关键词检索本地文档库 # 实际项目里这里接向量库或全文索引 return f检索结果{query} 相关文档 3 条 tools [search_docs] tool_node ToolNode(tools) # 把 tool_node 作为图的一个节点注册 graph.add_node(tools, tool_node)tool装饰器会把函数签名和 docstring 转成模型能理解的工具描述docstring 写得越清楚模型选错工具的概率越低。ToolNode内部会自动匹配tool_calls里的 name 和已注册工具执行后生成对应的ToolMessage。注意工具函数的返回值必须是字符串或可序列化对象返回复杂结构时先转成 JSON 字符串否则状态合并时容易报序列化错误。3.2 Checkpointer给 Agent 装一个「后悔药」没有持久化的 Agent 就像没有存档的游戏一旦中断就得从头再来。LangGraph 的 checkpointer 机制允许你在每个节点执行后保存状态快照下次用同一个thread_id调用时自动恢复。常见做法是用MemorySaver做开发期调试用SqliteSaver或PostgresSaver做生产环境持久化。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) # 调用时传入 thread_id同一 id 共享状态 config {configurable: {thread_id: user-001}} result app.invoke({messages: [(user, 帮我查一下 LangGraph 的文档)]}, config) # 第二次调用会自动带上历史状态 result2 app.invoke({messages: [(user, 刚才那条结果再详细点)]}, config)thread_id是状态隔离的键不同用户用不同 id同一用户的多轮对话用同一个 id。MemorySaver存在进程内存里重启就丢只适合本地调试生产环境换成SqliteSaver.from_conn_string(checkpoints.db)就能落盘。这里有个容易忽略的点checkpointer 保存的是整个状态快照如果状态里塞了大对象比如完整文档内容数据库会膨胀得很快。我一般只把消息列表和关键标记放进状态大块数据存外部存储状态里只留引用 id。3.3 人工审核节点在关键步骤插入 human-in-the-loop可控 Agent 的一个重要标志是「该停的时候能停」。LangGraph 支持在编译时设置interrupt_before[tools]让图执行到工具节点前暂停等人工确认后再继续。这在涉及写操作、发送消息、修改数据的场景里非常关键。app graph.compile( checkpointermemory, interrupt_before[tools] # 执行到 tools 节点前暂停 ) # 第一次调用会在 tools 前停下 state app.invoke({messages: [(user, 删除这条记录)]}, config) # 人工检查 state 后用 None 继续执行 app.invoke(None, config)interrupt_before接收节点名列表执行到这些节点前会抛出中断信号状态被 checkpointer 保存。恢复时传入None作为输入图会从断点继续。实际项目里我会把待审核内容渲染到前端人工点确认后再触发恢复调用。注意中断恢复依赖 checkpointer没有 checkpointer 的图不支持中断。4. 避坑与排查手搓 Agent 时最容易翻车的五个地方4.1 状态字段被覆盖导致历史丢失现象第二轮对话时模型完全不记得上一轮说了什么像失忆一样。原因状态定义里messages没有用add_messagesreducer节点返回的新消息直接替换了旧列表。解决把messages: list改成Annotated[list, add_messages]所有对 messages 的更新都走增量合并。这个坑我见过太多次尤其是从普通 LangChain 链式写法迁移过来的人习惯性认为返回就是覆盖。4.2 条件边返回值与映射表不匹配现象图执行到条件边时报KeyError或直接卡住不往下走。原因add_conditional_edges的第三个参数是映射字典条件函数的返回值必须是字典的 key 之一。如果函数返回tool但映射里写的是tools就会匹配失败。解决把条件函数的返回值和映射表的 key 对齐建议用常量或枚举而不是裸字符串改名字的时候不容易漏。4.3 工具调用死循环耗尽 token现象Agent 反复调用同一个工具retry_count 一路涨到上限才停token 消耗远超预期。原因工具返回结果没有被模型正确理解或者工具本身返回空结果模型认为「没拿到答案」就再调一次。解决在工具函数里对空结果做兜底返回在条件边里加 retry_count 上限同时检查 ToolMessage 的tool_call_id是否和模型的tool_calls对应——id 不匹配时模型会认为工具没被调用。4.4 Checkpointer 的 thread_id 混用现象不同用户的对话历史串在一起A 用户看到了 B 用户的上下文。原因所有调用都用了同一个thread_id或者忘记在 config 里传thread_id导致用了默认值。解决每次会话生成唯一 id比如fuser-{user_id}-{session_id}在 config 里显式传入。调试阶段可以在日志里打印 thread_id确认隔离生效。4.5 中断恢复时状态不一致现象用interrupt_before暂停后恢复执行时模型基于旧状态做了错误决策。原因中断期间外部数据变了比如工具依赖的数据库记录被修改但状态快照还是旧的。解决恢复前重新校验关键数据或者在中断节点前把需要校验的数据写入状态。我一般会在人工审核界面同时展示状态快照和实时数据让审核人判断是否继续。5. 进阶技巧用 stream 模式做逐节点调试与性能观测5.1 stream 与 astream_events 的区别和选用app.stream()按节点粒度输出状态更新适合看「每一步之后状态变成了什么」app.astream_events()按事件粒度输出包括 LLM token 流、工具开始/结束、节点进入/退出适合做前端实时展示和性能分析。调试阶段我建议先用 stream 把每个节点的输入输出打出来确认状态流转符合预期再切到 astream_events 做细粒度观测。# 按节点流式输出每个 chunk 是 {节点名: 状态更新} for chunk in app.stream( {messages: [(user, 帮我总结这份文档)]}, config, stream_modeupdates # updates 模式只输出增量 ): node_name list(chunk.keys())[0] print(f--- 节点 {node_name} 输出 ---) print(chunk[node_name])stream_mode支持values输出完整状态、updates只输出增量、debug输出详细调试信息。updates模式最实用因为 Agent 状态里 messages 会越来越长每次输出完整状态会刷屏。用updates能清楚看到每个节点往状态里写了什么。5.2 用回调统计每个节点的耗时性能问题往往出在某个工具调用或某次 LLM 请求上。LangGraph 兼容 LangChain 的回调体系可以挂一个自定义 handler 记录节点进出时间import time from langchain_core.callbacks import BaseCallbackHandler class TimingHandler(BaseCallbackHandler): def __init__(self): self.starts {} def on_chain_start(self, serialized, inputs, **kwargs): name serialized.get(name, unknown) self.starts[name] time.time() def on_chain_end(self, outputs, **kwargs): name kwargs.get(name, unknown) if name in self.starts: cost time.time() - self.starts[name] print(f[timing] {name}: {cost:.2f}s) handler TimingHandler() app.invoke({messages: [...]}, {**config, callbacks: [handler]})这个 handler 会在每个链/节点开始和结束时记录时间戳输出耗时。实际用的时候可以把结果写到日志或监控系统跑一段时间就能定位到瓶颈节点。如果发现某个工具节点耗时特别长优先检查工具内部的网络请求或数据库查询而不是调 LangGraph 的参数。5.3 一个我反复用的调试习惯从那以后我每次改完图结构都强制走一遍「三连」先用stream_modeupdates跑一轮看状态流转再用interrupt_before在关键节点暂停检查状态快照最后用 TimingHandler 看一遍各节点耗时。这三步走完大部分逻辑错误和性能问题都能提前暴露比等到线上出问题再回头查日志省事得多。这套 LangGraph 实战代码的价值不在于它写了多少行而在于它把 Agent 的每一步都摊开给你看——状态怎么定义、边怎么连、中断怎么恢复、循环怎么刹车。把这些骨架吃透再往上叠业务逻辑就是体力活了。希望帮到你。本文还有配套的精品资源点击获取