简介这份LangGraph10实战“从零手搓可控Agent”系列代码项目由公众号AI喵智能体发布面向正在学习LangGraph、希望亲手实现可控Agent的开发者。内容围绕基础对话、动态提示词、LCEL基础语法等实验模块展开兼顾概念讲解与可直接运行的代码示例适合动手实践、源码拆解及二次扩展。压缩包共7个文件以Python脚本为主体3个py实现多个关键示例同时包含txt说明文件、docx附赠资料、markdown说明文档及gitignore覆盖运行说明、拓展参考与项目配置。整个资源包仅41KB轻量精简下载后即可快速浏览。目前已有88人学习浏览适合作为入门LangGraph可控Agent的参考代码集。通过该资源可获得完整示例脚本、极简部署说明与拓展素材借助逐模块的代码组织理解Agent构建流程、提示词动态编排以及LCEL基础用法为继续开发更可控、更复杂的Agent应用打下基础。1. 从零手搓可控AgentLangGraph实战项目到底解决什么问题每次让大模型去调工具它要么答非所问要么在同一个函数上反复重试要么把已经确认过的结果又拿出来再问一遍——这是我在把LangGraph接入实际业务之前经常翻车的场景。这套「LangGraph10实战从零手搓可控Agent」系列代码项目的核心就是把「可控」两个字落到代码上用状态图把Agent的每一步决策铺开让开发者看得见、能干预、能限定边界。对已经写过LangChain链路、但受够了Agent黑匣子的从业者这套代码是值得照着敲一遍的最小骨架你可以直接抄节点写法也可以只借它的路由结构把它改造成自己的业务Agent。zip包里的东西本质上就是这一套可运行的最小工程。2. LangGraph核心设计拆解状态、节点、边如何构成可控Agent2.1 LangGraph与LangChain的区别同样写Agent为什么这里用图用LangChain写Agent最经典的形态是AgentExecutor一个内部循环由LLM决定调用哪个工具然后继续循环。这个循环对开发者来说是个黑匣子——你只知道输入和输出中间每一步的决策依据、工具返回、状态变化都不容易追踪。一旦输出结果不符合预期想定位是模型的问题还是工具的问题只能靠打日志一点点猜。LangGraph把这条路换掉了它把一次Agent执行建模成一张有向图节点是普通Python函数边是函数之间的跳转状态是独立的数据对象。节点从状态里读数据、计算结果、再写回状态边决定下一步去哪条件边还可以根据状态内容做分支。这样一拆所谓Agent就不再是「一个循环加一个LLM」而是一张可以逐段打断、逐段断言的图。你在这类实战代码项目里看到的风格也和LangChain完全不同很少见到AgentExecutor这类封装对象取而代之的是add_node、add_edge、add_conditional_edge这一组图操作。网上不少langgraph和langchain对比的面试题核心考的就是这个差异——LangChain的Chain是线性或少量分支的管道LangGraph的图是任意有向、可以成环的拓扑。成环才是Agent的本体调用工具、观察结果、再次决策本质就是一个环而链式编排天然表达不了环。LangGraph把环显式画出来这就是它和LangChain在架构层面最根本的分界。2.2 Agent状态设计messages、计数器和结果标记放哪里Agent区别于普通函数在于它有记忆和中间产物。LangGraph用State承担这个角色。定义State最常见的方式是TypedDict再配合Annotated和reducer。reducer的作用是决定「当多个节点都要写同一个键时到底是覆盖还是合并」。比如聊天消息我们希望每次追加而不是覆盖就给它挂add_messages这个reducer数值计数我们希望每次累加就挂operator.add反过来某个字段只想保留最新值那就不加reducer默认就是覆盖写。reducer选错了执行结果就是一个字乱。在设计手搓Agent的State时我一般至少放三个东西messages历史消息用add_messages追加、steps节点执行计数用operator.add累加、done结束标记默认覆盖写。steps这个字段在排障时价值极大——Agent死循环时你能从流式输出里直接看到steps在疯涨从而判断是哪个节点反复被触发。很多langgraph教程里只教你定义messages实战里建议把计数器也放进去。另外State不只是给模型看的它是全图共享的工作台任何节点需要的数据都应该在State里显式声明而不是靠闭包或全局变量偷偷传递否则一旦并发跑多个会话输出就会互相污染。2.3 节点与条件边的模型可控性从哪来以及怎么被打破可控性来自三个地方。第一节点是纯函数风格输入输出都在State上单测容易写mock工具调用也容易。第二边是显式的执行路径能完整画出来哪里可能有环、哪里可能死循环运行前就能预判。第三中间态可断LangGraph支持在节点之间设置中断点也支持配合checkpointer做暂停和恢复。标题项目里强调「从零手搓」本质就是让你自己定义节点边界而不是依赖框架里写死的ReAct循环。很多人以为用LangGraph就是换个API写同样的Agent实际差别在于你可以把「LLM生成」「工具执行」「结果校验」拆成三个独立节点各自负责一件事各自可以被单独测试。但这套灵活也有代价。图一旦设计得不合理Agent会变得比黑匣子还难查节点太多、边太密执行路径呈指数增长。常见的破坏可控性的误用有两个一是节点函数里直接修改入参dict不返回更新导致LangGraph拿不到变更二是在条件边里塞了太多逻辑路由函数又长又杂相当于把黑匣子搬了个位置。我自己踩过第二个坑后来硬性要求路由函数只做一件事根据State里的标记返回下一个节点名具体的判断逻辑放到节点内部完成。这样每条边都短每个节点都长图结构一眼能看懂。框架选型时LangGraph能排在agent框架前列靠的也是这套显式状态加显式边的设计它把控制权还给了开发者代价是你要自己维护图形的合理性。3. 从零手搓可控Agentzip代码解压后的最小实现3.1 解压与环境准备LangGraph项目文件怎么摆拿到这套zip包第一步自然是解压。注意解压后不要直接双击某个py文件就跑先规划好目录和虚拟环境。我的习惯是新建一个干净目录把zip解压进去然后创建venv。目录名不要带中文、不要带空格否则部分工具链在解析路径时会出莫名其妙的错误。mkdir langgraph-agent cd langgraph-agent python -m venv venv source venv/bin/activate pip install langgraph langchain-openai langchain-core说明一下langgraph是核心图引擎langchain-openai是模型接入层langchain-core提供消息类型和ToolMessage等基础数据结构。版本建议锁定不要装最新版就完事——这一点在第5章避坑里会展开。这类零基础上手项目解压后的代码结构我一般按四层来摆agent/ state.py # Agent状态的TypedDict定义 nodes.py # 模型调用节点、工具执行节点 routes.py # 条件路由函数的集中放置 config.py # 模型名、API Key、工具注册表 main.py # 构建图、编译、提供运行入口把路由函数单独放一个文件而不是散落在各个节点里是我反复踩坑后形成的习惯。路由是整个图的中枢神经集中在一处排查「Agent为什么走到这个节点」时只需打开一个文件。标题里「极简说明」的意思我理解就是把这类骨架代码压缩到最短可运行状态——不塞复杂的日志框架不接数据库先让你把图画出来、跑起来再谈业务增强。3.2 定义状态与节点首个可运行骨架先看state.py这是全图的契约。每个节点都要读它、写它所以字段类型要对齐。# agent/state.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): # 历史消息add_messages保证多个节点写入时是追加不是覆盖 messages: Annotated[list, add_messages] # 步骤计数器记录节点被调用的总轮次 steps: int # 完成标记路由读取这个字段决定是否结束 done: bool然后是节点函数。一个最简Agent至少有「模型调用」和「工具执行」两个节点。先写模型节点# agent/nodes.py from langchain_openai import ChatOpenAI from agent.state import AgentState # temperature设低一些让工具调用决策更稳定 llm ChatOpenAI(modelgpt-4o-mini, temperature0) def call_model(state: AgentState) - dict: 模型节点读取当前messages生成回复并累加步骤数。 response llm.invoke(state[messages]) # 返回dict只写需要变更的键LangGraph会自动合并到全局状态 return {messages: [response], steps: state[steps] 1}这里有两个关键点。第一Annotated[list, add_messages]让每次返回的消息都被追加到已有列表末尾而不是覆盖整个messages。如果没有这个reducer第二次调用模型时前一轮的对话历史会被清空Agent立刻失忆。第二节点返回的dict里只写要变更的字段LangGraph会把返回值与原state做一次合并没有返回的键保持原值。我见过新手把所有字段原封不动返回一遍结果静态字段被覆盖行为不可控。3.3 条件路由与工具编排让Agent自己决定下一步有了模型节点还需要一个「工具节点」和一个「路由函数」。工具节点负责执行模型要求调用的工具然后把工具结果追加回messages路由函数则根据最新消息判断下一步去哪。# agent/nodes.py继续追加 def call_tool(state: AgentState) - dict: 工具节点取出模型要求的工具调用参数执行并返回结果。 last_message state[messages][-1] # 模型可能一次请求多个工具这里简化取第一个 tool_call last_message.tool_calls[0] # 实际项目中按tool_call[name]分发到不同函数 result fake_search_tool(tool_call[args][query]) return { messages: [ToolMessage(contentresult, tool_call_idtool_call[id])], steps: state[steps] 1, }注意ToolMessage必须携带tool_call_id并且要和模型发出的tool_call中的id完全一致。这是LangChain消息协议里最容易出错的地方id对不上模型会认为工具没执行成功然后反复发起同一个调用。路由函数单独放routes.py# agent/routes.py from langgraph.graph import END from agent.state import AgentState def next_node(state: AgentState) - str: 条件路由根据最新消息和步骤数返回下一个节点名。 last_message state[messages][-1] # 模型要求调用工具时必须先去工具节点 if hasattr(last_message, tool_calls) and last_message.tool_calls: return tool # 步骤数超过阈值无论模型说什么都强制结束 if state[steps] 6: return END return END路由是可控Agent的命门。它决定了两件事什么条件下继续什么条件下中止。这套代码项目里最值得学的就是这个函数——它把你对Agent行为的约束显式写出来而不是让模型自己无限发挥。上面这段同时处理了「工具调用优先」和「步骤上限」两个约束属于最小可用的版本。如果想更可控可以加一个done字段当某个校验节点发现工具结果不合法时把done置True路由里优先判断done再决定是否结束。3.4 编译与运行从图对象到可调用的Agent节点、路由都准备好了接下来在main.py里把它们组装成图并编译。# main.py from langgraph.graph import StateGraph, START, END from agent.state import AgentState from agent.nodes import call_model, call_tool from agent.routes import next_node # 1. 创建图声明状态类型 graph StateGraph(AgentState) # 2. 添加节点节点名可以自定义后面路由返回的就是这个名字 graph.add_node(model, call_model) graph.add_node(tool, call_tool) # 3. 添加边入口到modeltool执行完必须回到model再做决策 graph.add_edge(START, model) graph.add_edge(tool, model) # 4. 添加条件边model之后根据路由函数决定是去tool还是结束 graph.add_conditional_edge(model, next_node, { tool: tool, END: END, }) # 5. 编译成可执行应用 app graph.compile()add_conditional_edge的第三个参数是一个字典把路由函数的返回值映射到具体节点名。路由返回的字符串如果在字典里找不到对应键运行时直接抛异常。早期我踩过这个坑路由函数拼写错误返回了tool 字典匹配不上排查了半天后来养成了把路由的返回值和节点名固定成常量而不是手写字面量的习惯。编译之后就是运行from langchain_core.messages import HumanMessage initial_state { messages: [HumanMessage(content帮我查一下今天的天气)], steps: 0, done: False, } final_state app.invoke(initial_state) print(final_state[messages][-1].content)这里必须显式给出完整的初始状态包括steps和done否则路由函数读取state[steps]时直接KeyError。整个流程跑通后你就有了一个最小可控Agent模型决定是否调用工具路由决定何时终止每次循环的步骤数都被计数。这套骨架往业务方向改无非是增加节点、增加路由分支、替换工具实现。4. 控制Agent的关键参数从默认值到生产级配置4.1 recursion_limit防死循环的保险丝怎么设LangGraph对每次执行设置了一个最大递归步数默认值是25。这个值对应的是图里「节点之间跳转的次数」也就是superstep数量并非LLM调用次数。一个模型调用加一次工具调用至少消耗2个superstep。业务Agent如果工具链路较长默认25很容易触顶表现是抛GraphRecursionError。通过config传入recursion_limit来调整config {recursion_limit: 50} final_state app.invoke(initial_state, configconfig)设置思路一般是按「预期最大工具调用轮次」乘2再加缓冲。比如设计上最多允许5轮工具调用每轮包含模型节点和工具节点两次跳转就是10次superstep再加5轮裕量设成20足够。不建议一上来就设100、200等于把保险丝拆了死循环时资源烧完你才察觉。线上环境我会额外包一层超时控制因为recursion_limit只管步数不管单步里LLM调用耗时。4.2 中断点与人工确认Human-in-the-loop的关键配置有些场景不能让Agent全自动跑完比如支付、审批、发送邮件。LangGraph提供了中断机制编译时可以指定在某个节点执行前或执行后停下。# 进入tool节点前暂停等待人工确认 app graph.compile( checkpointerMemorySaver(), interrupt_before[tool], ) thread_config {configurable: {thread_id: order-42}} # 第一次invoke会在tool节点前暂停返回未执行工具的状态 pending_state app.invoke(initial_state, configthread_config) print(暂停等待确认, pending_state) # 人工确认后用Command(resume)让图继续执行 from langgraph.types import Command confirmed app.invoke(Command(resumeTrue), configthread_config)这段代码里有两个关键点。第一中断必须配合checkpointer使用否则thread_id没有落点无法恢复执行。MemorySaver是内存版检查点适合测试生产需要换Postgres或SQLite的持久化实现。第二恢复是通过Command(resumeTrue)触发LangGraph会从上次中断的节点继续往下走。有的版本里也支持传一个dict进去直接作为暂停节点的额外输入。这个机制是我做生产Agent时最常用的「后悔药」——模型说它要调支付接口你在确认侧拦一道看它传的参数对不对再放行或终止。4.3 checkpoint持久化线程ID和序列化的边界checkpointer除了支撑中断恢复还有一个实用能力是时间旅行你可以拉出某一thread的历史快照回放某一步的状态排查线上问题。使用上最需要注意两个边界。一个是thread_id的粒度如果同一业务会话的多个调用用了不同thread_id每次都被当作全新会话状态不连续。另一个是序列化能力checkpointer要把state完整序列化后落盘如果state里塞了数据库连接、Socket、函数对象这类无法pickle的东西写入检查点时直接报错。我一般的做法是把这类资源对象放到config的configurable字段里而不是放进state。config { configurable: { thread_id: order-42, db_conn: db_conn, # 只在节点内部读取不进state } }节点里通过config参数获取资源def call_tool(state: AgentState, config: dict) - dict: conn config[configurable][db_conn] # 用conn执行查询 return {messages: [result_msg], steps: state[steps] 1}这里体现的是LangGraph的一个设计原则state只放需要被追踪、被回放的数据资源对象走config通道。很多从LangChain转过来的开发者不习惯老是往state里塞各种对象然后被序列化错误折磨。记住一句话state是给Agent看的config是给代码用的别混。5. LangGraph实战避坑五个必看问题与排查记录5.1 现象运行时提示找不到state里的某个键刚把第3章骨架跑起来时最容易报KeyError: steps。现象很统一invoke刚启动路由函数还没执行完就崩了。原因几乎都是初始state没有给全字段。路由函数state[steps] 6这一行在steps缺失时直接抛异常而且这个异常发生在图执行早期堆栈里只会看到next_node不会提示「你忘了初始化」。解决方式有两种一是每次invoke时显式给全字段二是路由函数里改用state.get(steps, 0)给缺失字段一个默认值。我两种都用了但更推荐后者因为后续新增字段时旧代码不容易被新逻辑搞挂。5.2 现象Agent陷入无限循环steps疯涨最经典的一幕模型反复调同一个工具工具每次都返回结果但模型还是继续发起调用日志里steps一路冲到GraphRecursionError。原因通常是ToolMessage的tool_call_id和模型请求的id不一致模型认为工具没有执行成功于是重试。另一个常见原因是工具返回的内容格式不符合模型预期比如空结果、异常结构模型读不懂就反复请求。解决分两层先把tool_call_id严格用tool_call[id]回填这是最容易忽略的然后在路由里加步骤上限超过阈值无论模型想干什么都强制END。还有第三层保险就是在工具节点里对结果做校验发现异常时写一个doneTrue标记路由里优先读取。顺序是done标记 → tool_calls判断 → steps上限三层检查Agent再难失控。5.3 现象多个节点写同一个state键数据互相覆盖图里有并行节点时常出现消息丢失或计数器被重置。原因是没有给该键配reducer。比如两个工具节点同时返回{messages: [...]}如果messages没挂add_messages后写入的节点会直接覆盖先写入的节点。解决方式是给所有需要累积的字段显式声明reducer消息用Annotated[list, add_messages]字典合并用Annotated[dict, operator.or_]计数器用Annotated[int, operator.add]。排查这个问题的技巧是开启stream模式观察每步事件里的state变化哪个键缺了reducer前后对比一眼就能看出来。5.4 现象langgraph和langchain版本错位一升级就报解析错误很常见。代码昨天还能跑今天pip install一升级突然报Pydantic校验失败或Message解析错误。原因很简单langgraph、langchain-core、langchain-openai三个包各自有发布节奏接口在快速演进某个包单方面升级就可能打破兼容。解决方式是锁版本区间我在第3章环境准备里特意提了这一点。实际项目里requirements.txt会写成这样pip install langgraph0.2.0,0.4.0 langchain-core0.3.0,0.4.0 langchain-openai0.2.0,0.3.0升级时先升级langchain-core再测langgraph不要一把梭全升。这类问题网上搜「langchain和langgraph区别」的帖子经常讨论到很多人以为是使用方法错了实际就是版本问题。5.5 现象开启checkpointer后invoke报序列化失败表现为某个节点执行正常但写入检查点时抛TypeError: cannot pickle ...。原因就是state里放了不可pickle对象比如数据库连接、Redis客户端、或某种lambda函数。checkpointer做快照时要把整个state序列化储存遇到这类对象就崩。解决方式是回到第4.3节说的原则资源走config数据进state。如果你发现在并发场景下资源带不走那也是因为放错了位置。把db_conn移进config[configurable]序列化问题立刻消失。另一个备选方案是在节点返回时主动剔除不可序列化字段但非常容易漏不建议。6. 进阶验证可控性的两个习惯6.1 把图结构打印出来再上线LangGraph自带图结构可视化能力编译后的对象可以直接打印ASCII结构的图print(app.get_graph().print_ascii())终端里会输出类似这样的结构START进入model节点model分出条件边指向tool或ENDtool又回到model。如果你画出来的图里有预期之外的边说明add_edge或add_conditional_edge的节点名写错了如果发现某个节点没有任何边指向它说明路由映射字典漏了分支。我每次改完图第一件事就是print_ascii看一眼确认拓扑符合设计。这比运行测试更快发现问题尤其是节点数量超过5个之后人脑已经不太容易记住所有连接关系。6.2 用流式事件逐节点断言invoke只返回最终状态不好定位是哪个节点出问题。把调用方式换成stream每个节点执行完都会吐一个事件for event in app.stream(initial_state, config{recursion_limit: 50}): for node_name, node_output in event.items(): print(node_name, node_output)event是一个字典键是刚执行完的节点名值是该节点返回的状态变更。逐条看输出节点执行顺序、状态变化过程一清二楚。我在带项目时要求每个Agent节点必须配一个独立测试用stream收到的节点名作为断言对象——如果某次业务期望走model→tool→modelstream里却出现model→model那要么路由函数写错要么tool_calls字段没被正确识别。这套代码项目教会我最重要的一件事是Agent失控不是玄学而是状态和路径没有显式化。我早年从某个agent框架里照搬过「循环加工具调用」的黑匣子实现上线后模型卡在一个工具上反复重试配额烧完才发现问题。后来把所有执行路径摊到LangGraph图上每个分支都能提前看、提前测才真正敢把Agent交给业务。如果你正在折腾自己的Agent建议先把第6.1节的图打印出来跑一遍再谈加更多智能。希望帮到你。本文还有配套的精品资源点击获取