简介:Agent是人工智能应用的重要形态,但真正让Agent可靠落地的关键,在于“可控”。LangGraph以图结构重新定义了Agent的构建方式,将流程拆解为State(状态)、Node(节点)、Edge(边)三要素,使业务流程可定义、可审计、可干预。通过条件边实现决策可控,通过interrupt机制实现人工审批,再配合checkpoint实现跨会话的长期记忆,LangGraph为高风险的业务场景提供了完整的工程化方案。无论是意图识别、工具调用,还是接入Ollama本地模型,LangGraph都能在保持灵活性的同时,把Agent的行为约束在预设边界内。本文基于实际项目,从环境配置、核心三件套到human-in-the-loop落地,拆解从零手搓一个可控Agent的完整过程,帮助你避开版本、依赖和状态管理中的各种暗坑。 我在公众号AI喵智能体把这份LangGraph10实战从零手搓可控Agent.zip放出来之后,后台收到最多的问题,不是"图怎么搭",也不是"状态怎么写",而是"解压不了"和"装完跑不起来"。这让我挺意外的,但也说明一个事:LangGraph虽然已经不算新东西了,可对大部分人来说,从下载代码到真正跑通一个可控Agent,中间隔着的不是算法壁垒,而是一堆藏在环境、版本、数据流里的暗坑。
这篇东西不是官方文档的复读,也不是把项目README翻译一遍。我按自己从零手搓这个项目的完整过程,把LangGraph可控Agent的核心机制、关键代码、参数取舍和踩坑经历都拆开讲一遍。适合已经写过几段LangChain、想进一步掌控Agent行为的读者,也适合刚下完zip还不知道从哪下手的新手。
1. 为什么非要选LangGraph:可控Agent的核心痛点
1.1 LangChain链式调用解决不了的"失控"问题
先说结论:如果你只做"先调用一个模型、再调用一个工具、最后输出结果"的简单流程,LangChain的Chain足够用。但一旦遇到"根据用户输入决定下一步调用哪个工具、多轮对话中要保留上下文、业务上某个动作必须让人确认后才能执行"这些场景,Chain的线性结构就不行了。
我在实际项目里遇到过最典型的情况:一个客服Agent,用户说"我要退订套餐",模型直接调用了退订API,一气呵成,根本没有二次确认。这在真实业务里是不可接受的——退订涉及资费损失,必须有用户明确同意才能执行。
LangGraph解决这个问题的思路,是把Agent从"自由的链式调用"改成"有约束的图式流转"。图里每个节点就是一个明确的处理步骤,边指定了流转方向,条件边则让Agent根据规则(而不是完全靠模型心情)决定走哪条路。你把它理解成流水线:每个工位只做自己那件事,做完之后产品去哪个工位,由质检员(条件函数)根据规则决定,而不是由流水线末端的机器人(LLM)闭着眼睛乱扔。
1.2 "可控"到底控什么:三个层面的控制力
我见过很多人把"可控Agent"简单理解成"Agent能按步骤执行",其实不够。LangGraph给的"可控"是三个层面的:
流程可控:节点和边的拓扑结构预先定义好,Agent不可能跳出你画的图。它只能在你的流程图内运转,不会产生未定义的中间步骤。
决策可控:条件边允许你用规则或模型输出做路由判断。比如
if decision == "call_tool": go to tool_node else: go to chat_node,这个分支逻辑是显式的,可审计、可调试。执行可控:通过
interrupt_before之类的机制,在节点执行前暂停整个图,把控制权交给人。人确认后才继续,或者修改状态后继续。这就把"全自动"降级成了"人在回路"。
这三层加在一起,Agent才真正适合进入业务系统。所谓"从零手搓",本质就是在写这三层控制力的代码,而不是在调模型。
2. 环境准备:版本、解压、安装里的三座大山
2.1 拿到zip之后的第一步不是解压,而是确认文件完整
这个项目包命名里有zip后缀,但很多人在网盘下载后遇到"File is not a zip file"或者更诡异的"invalid zip archive: could not find eocd"错误。这类报错九成是压缩包没下载完整,或者下载工具把文件转成了HTML。
我用过的最稳妥的检查方式,是在终端里用file命令看一眼真实格式:
file LangGraph10实战从零手搓可控Agent.zip正常输出应该是Zip archive data,如果显示HTML document或ASCII text,基本可以确定下载出问题了。重新下载时尽量用浏览器直链,不要用某些下载工具的多线程加速,那个最容易把文件搞坏。
在Linux服务器上解压,我喜欢用unzip而不是图形化工具:
unzip LangGraph10实战从零手搓可控Agent.zip -d langgraph10注意-d参数指定解压目录,避免文件散落一地。
2.2 版本匹配是LangGraph社区最隐蔽的坑
我想重点强调一下版本匹配的问题。这个项目是langgraph10,对应的核心依赖我建议这样装:
pip install langgraph==0.2.60 pip install langchain==0.3.7 pip install langchain-openai==0.2.14网上很多报错,追根溯源都是langgraph和langchain版本不匹配。尤其langgraph对langchain-core的版本比较敏感,装的时候我建议用pip install "langgraph[all]"一把梭,让它自己把配套依赖拉齐,不要手动一个个装,那样极容易装出个"满汉全席"版本的依赖冲突。
小技巧:装完用pip check命令检查依赖是否冲突,如果有冲突,优先升级langchain-core:
pip check pip install --upgrade langchain-core2.3 项目目录结构解读:别一上来就改代码
解压之后先花两分钟看目录结构。这个项目的结构大致是这样:
langgraph10/ ├── graph/ │ ├── __init__.py │ ├── state.py # 状态定义 │ ├── nodes.py # 节点函数 │ ├── edges.py # 边的逻辑 │ └── graph.py # 图构建入口 ├── tools/ │ └── search.py # 自定义工具 ├── config/ │ └── settings.py # 模型配置、API Key ├── tests/ │ └── test_flow.py └── main.py这里我特别提醒:graph/目录下的__init__.py千万别删,哪怕它是个空文件。Python的包导入机制靠它识别目录,删了之后from graph.state import AgentState直接报ModuleNotFoundError。这种问题排查起来特别容易让人怀疑人生,因为看起来代码什么都没改错。
3. 核心三件套:State、Node、Edge是如何撑起一个Agent的
3.1 State:一个会"累积"的字典,Agent的临时记忆体
State是整个LangGraph最基础也最需要想清楚的概念。你可以把它理解成一个所有节点都能读写的共享笔记本。
这个项目里State的定义是这样的:
from typing_extensions import TypedDict from langgraph.graph.message import add_messages from typing import Annotated class AgentState(TypedDict): messages: Annotated[list, add_messages] current_step: str user_intent: str require_approval: bool tool_result: str注意messages字段用了Annotated[list, add_messages]。这个add_messages是LangGraph提供的一个更新策略:当多个节点往messages里追加内容时,不是粗暴替换,而是合并追加。如果不加这个注解,后写的节点会把先写的节点内容整个覆盖掉。
我一开始在这里栽过跟头:没有加add_messages,结果第二个节点写messages时,第一个节点的内容全没了,对话历史只剩最后一轮。查了半天发现是状态更新策略的问题。
current_step、user_intent这些字段则用来记录当前流程到了哪一步、用户意图是什么,供条件路由做判断。
3.2 Node:每个节点就是一个"工序"
在LangGraph里,节点就是普通的Python函数。这个项目的节点设计思路,是把每个能力边界切干净:
def parse_intent(state: AgentState) -> dict: """第一道工序:解析用户意图""" # 这里调用LLM做意图分类,返回结构化的意图标签 intent = classify_intent(state["messages"][-1].content) return {"user_intent": intent, "current_step": "intent_parsed"} def call_tool(state: AgentState) -> dict: """第二道工序:根据意图调用工具""" result = invoke_tool(state["user_intent"]) return {"tool_result": result, "current_step": "tool_called"} def generate_answer(state: AgentState) -> dict: """第三道工序:生成最终回复""" answer = llm.invoke(f"基于工具结果{state['tool_result']}回答用户") return {"messages": [answer], "current_step": "answered"}每个节点函数接收当前State,返回一个字典,字典里的键会更新到State上。关键细节:返回的字典里写哪些键,就只更新哪些键。比如parse_intent只返回user_intent和current_step,它不会动messages。这种局部更新的机制,保证了节点之间的隔离性——一个节点不会不小心改掉别的节点的数据。
3.3 Edge:边的定义方式决定了Agent的灵活度
边有三种:普通边、条件边、入口边。我重点说条件边,因为它是可控性的灵魂。
from langgraph.graph import StateGraph def should_call_tool(state: AgentState) -> str: """决策函数:返回下一个节点的名称""" if state["user_intent"] == "query": return "call_tool" elif state["user_intent"] == "chitchat": return "generate_answer" else: return "generate_answer" graph = StateGraph(AgentState) graph.add_node("parse_intent", parse_intent) graph.add_node("call_tool", call_tool) graph.add_node("generate_answer", generate_answer) # 入口:第一步从解析意图开始 graph.set_entry_point("parse_intent") # 普通边:解析完意图之后,走判断 graph.add_conditional_edges( "parse_intent", should_call_tool, { "call_tool": "call_tool", "generate_answer": "generate_answer" } ) # 普通边:调用工具后必须生成答案 graph.add_edge("call_tool", "generate_answer") graph.add_edge("generate_answer", "__end__")add_conditional_edges的第一个参数是起始节点,第二个参数是路由函数,第三个参数是路由结果到目标节点的映射。这个映射必须覆盖路由函数可能返回的所有值,否则运行时会报InvalidUpdateError之类的错误。
这里我想解释一下为什么要有"意图解析"这个节点。你完全可以直接让LLM在生成回答时决定要不要调工具,但如果这样,你无法保证这个决定是可靠的。单独拆一个意图识别节点出来,它只做分类,不做别的,逻辑就变得可测试、可观测。你甚至可以把它换成规则匹配、正则、或者一个小模型,完全不影响下游。
4. 从"全自动"到"人在回路":human-in-the-loop的落地姿势
4.1 用interrupt_before卡住关键节点
可控Agent和普通Agent最大的区别,是它允许人类在关键步骤插手。LangGraph实现这个能力很简单:编译图的时候指定interrupt_before。
interrupt_before的作用是:在某个节点执行之前暂停整张图,把图的执行状态保存在内存中,等待外部输入。外部确认后,调用invoke继续执行。
修改一下上面的图:
app = graph.compile(interrupt_before=["call_tool"])就这么一行,当图流转到call_tool节点前,会停下来。此时你可以在外部(比如Web应用的API层)检查当前状态:
config = {"configurable": {"thread_id": "user_session_001"}} result = app.invoke( {"messages": ["帮我查一下今天的天气"]}, config ) # 检查是否被中断了 if app.get_state(config).next: # 在这里做人工审批,展示待执行动作给用户确认 print("等待人工确认,下一步将执行:", app.get_state(config).next)get_state(config).next会返回下一个待执行的节点名称。你可以把它展示给用户:"即将查询天气,确认吗?",用户点确认后再继续。
4.2 恢复执行:继续而不是重跑
恢复执行的时候不要重新调用整个图的入口,而是用invoke传入None作为更新值,让图从暂停的位置继续:
# 用户确认后,继续执行 updated_result = app.invoke(None, config)如果你想在恢复前修改一下状态,比如用户说"其实不用查天气,帮我定个闹钟",你可以用update_state改掉状态里的意图,再继续:
app.update_state(config, {"user_intent": "set_alarm"}) updated_result = app.invoke(None, config)这个update_state非常实用。它让人可以在执行中途纠正Agent的方向,而不是一切重来。我在实际业务里,会把"待确认动作"通过WebSocket推给前端做一个弹窗,用户点了"允许"之后后端自动调用invoke(None),整个体验就像人工审批流。
4.3 三种中断方式怎么选
LangGraph给的中断方式不止interrupt_before一种,还有interrupt_after和动态中断(在节点内部用interrupt函数)。三者的区别是:
| 方式 | 中断时机 | 适用场景 |
|---|---|---|
interrupt_before | 节点执行前 | 高风险动作执行前审批 |
interrupt_after | 节点执行后 | 节点产出结果后人工复核 |
| 动态中断 | 节点内部某行代码处 | 根据运行数据动态决定是否中断 |
我个人用得最多的是interrupt_before,因为业务上"执行前审批"是最自然的交互。动态中断适合那种"只有函数运行到一半才知道要不要问人"的场景,但会增加代码复杂度,新手不建议一开始就用。
5. 长期记忆:可控Agent跨会话能力的底层支撑
5.1 记忆为什么要分"短期"和"长期"
如果你的Agent只在一个会话内工作,State里存messages就够了。但真实场景下,用户今天问"给我推荐一部电影",明天说"把昨天推荐的第一部加进我的片单",如果Agent不记得昨天的对话,这句指令就是无源之水。
LangGraph的记忆机制可以这样理解:短期记忆就是State里的大字典,一次会话内有效;长期记忆是保存在外部存储里、跨会话可读取的持久化状态。
5.2 用langgraph-checkpoint保存长期状态
这个项目里实现了通过langgraph-checkpoint实现的长期记忆,最简单的落地方式是使用SqliteSaver:
from langgraph.checkpoint.sqlite import SqliteSaver # Python 3.7+,注意用with语句确保连接正确关闭 with SqliteSaver.from_conn_string("checkpoints.sqlite") as saver: app = graph.compile( checkpointer=saver, interrupt_before=["call_tool"] )只要你给每次调用传了config["configurable"]["thread_id"],LangGraph就会把每一步的State快照存在SQLite里。下次用同一个thread_id恢复时,之前的messages还在,图的状态也还在。
这就实现了"长期记忆":用户的会话状态被持久化在磁盘上,而不是存在内存里一重启就没了。
5.3 长期记忆设计上的三个建议
用了几个项目之后,我对长期记忆的坑有三个体会:
不要一股脑把全量消息塞进长期记忆。SQLite存得下,但每次都要重新打包给LLM,成本高、响应慢。我习惯在
generate_answer节点做一次压缩:只保留关键结论和用户偏好。thread_id要有业务语义。用用户ID作为thread_id的维度,比如
customer_123,这样客服切换浏览器后,下一个客服依然能看到同一个用户的上下文。区分"记忆"和"状态"。记忆是给LLM看的上下文,状态是图运行的控制信息。
current_step这种控制字段不要暴露给LLM,否则模型容易被无关信息带偏。
6. 连接本地模型:让LangGraph同时支持云端API和Ollama
6.1 为什么要接Ollama
项目默认用的是OpenAI兼容接口,但很多读者的机器上跑的是本地模型。热词里一直有人搜"基于langgraph、ollama构建本地ai智能体",我就在这个项目里额外留了一组Ollama的配置。
LangGraph用的大语言模型,本质上是实现了LangChainBaseChatModel接口的对象。所以接Ollama只需要用langchain_ollama包:
pip install langchain-ollamafrom langchain_ollama import ChatOllama llm = ChatOllama( model="qwen2.5:7b", temperature=0.3, base_url="http://localhost:11434" )换成这个llm对象之后,上游的意图解析、下游的答案生成不用改任何代码。这就是LangGraph"模型无关"的好处——图的结构是固定的,模型只是图上执行具体任务的一个工具。
6.2 本地模型下可控性反而更重要
我实测下来,用本地7B模型跑这个项目时,条件路由的重要性会更明显。小模型的幻觉率更高,如果完全让它自由决定下一步动作,跑偏的概率大得多。但有了意图识别节点、条件边、人工审批三层约束之后,本地小模型也能在业务里稳定工作——因为图把它的"自由意志"限制在一个很窄的范围内。
这算是LangGraph一个反直觉的优势:你的模型越弱,越需要图这种强约束结构来兜底。
7. 踩坑实录:那些折磨了我一整晚的问题
7.1 zip解压类的坑
这个项目和zip相关的报错,网上搜索量最大的是"file is not a zip file"和"could not find eocd"。前面说了是下载不完整导致的,这里再补一个场景:在Windows下用自带资源管理器解压时,偶尔会提示"压缩文件已损坏",但同一个zip传到Linux服务器上用unzip却能正常解开。这大概率是Windows资源管理器对zip64格式的兼容性问题,换用7-Zip或者Bandizip解压就能绕过去。
7.2 图执行报错的坑
项目跑起来之后,最常见的报错是InvalidUpdateError: Expected list, got str。这个十有八九是State里的字段加了add_messages注解,但某个节点往这个字段里写的是字符串而不是消息对象。
比如你有这个字段:
messages: Annotated[list, add_messages]那在节点里返回时应该这样:
from langchain_core.messages import AIMessage # 正确 return {"messages": [AIMessage(content="你好")]} # 错误:直接返回字符串会被add_messages拆包,报错误,或静默覆盖 return {"messages": "你好"}这个问题排查的方法很简单:看报错堆栈里指向哪个节点,然后打开节点函数的return,确认返回类型和State定义一致。
7.3 检查点权限和文件锁的问题
SqliteSaver在Windows下偶尔会遇到文件锁问题,报database is locked。这是因为多线程同时写同一个SQLite文件导致的。我建议在编译图的时候给每个线程独立的连接,或者改用MemorySaver做测试、用SQLite做生产:
from langgraph.checkpoint.memory import MemorySaver # 测试用,重启即失忆 app = graph.compile(checkpointer=MemorySaver())MemorySaver纯内存实现,没有文件锁问题,适合本地开发调试。确认整条链路OK之后再切到SqliteSaver做持久化。
8. 手搓可控Agent的五个经验总结
最后我想从项目经验角度说几个个人体会。
第一,可控Agent的核心不在模型,在图。我一开始以为做一个好用的Agent主要靠调prompt,后来发现prompt再精巧,也扛不住模型随机性带来的不可控。把流程拆成节点、用边和条件边固定决策路径,才是真正的"可控"。
第二,每个节点只做一件事,且函数签名越简单越好。有些代码写出来一个节点里干三件事,看似省了步骤,实际上出了问题你都不知道是哪一段引发的。
第三,强烈建议给每个节点加上current_step的状态标记。它相当于程序的日志,出了问题你能立刻知道整个图卡在哪一步。没有这个,调试LangGraph就像在没有仪表盘的飞机上找故障。
第四,人工审批不要滥用。interrupt_before虽然好用,但每一步都中断,用户会被搞疯。我建议只对"高影响、不可逆"的动作做人工确认,比如扣费、删数据、发消息;普通的查询、读取没必要。
第五,先把不带记忆的图跑通,再上checkpointer。长期记忆是锦上添花,但也是排查问题时最大的干扰源。先把图逻辑调稳,再开记忆,配合SQLite一步步来,遇到问题更容易定位。
这个系列的项目文件虽然是个zip,但里面的核心不是那一堆代码文件,而是"如何把Agent放进一个可控的框里"这一套方法论。从State设计、Node切分、条件边路由,到人工审批和数据持久化,串起来就是一个完整的、可以直接对接业务的Agent应用。如果你正在做的项目对Agent的确定性有要求,LangGraph这套东西值得你花一个周末好好跑一遍。
本文还有配套的精品资源,点击获取