尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

LangGraph实战:从零手搓可控Agent的完整指南

LangGraph实战:从零手搓可控Agent的完整指南 简介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 langgraph0.2.60 pip install langchain0.3.7 pip install langchain-openai0.2.14网上很多报错追根溯源都是langgraph和langchain版本不匹配。尤其langgraph对langchain-core的版本比较敏感装的时候我建议用pip install langgraph[all]一把梭让它自己把配套依赖拉齐不要手动一个个装那样极容易装出个满汉全席版本的依赖冲突。小技巧装完用pip check命令检查依赖是否冲突如果有冲突优先升级langchain-corepip 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实现的长期记忆最简单的落地方式是使用SqliteSaverfrom langgraph.checkpoint.sqlite import SqliteSaver # Python 3.7注意用with语句确保连接正确关闭 with SqliteSaver.from_conn_string(checkpoints.sqlite) as saver: app graph.compile( checkpointersaver, 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和Ollama6.1 为什么要接Ollama项目默认用的是OpenAI兼容接口但很多读者的机器上跑的是本地模型。热词里一直有人搜基于langgraph、ollama构建本地ai智能体我就在这个项目里额外留了一组Ollama的配置。LangGraph用的大语言模型本质上是实现了LangChainBaseChatModel接口的对象。所以接Ollama只需要用langchain_ollama包pip install langchain-ollamafrom langchain_ollama import ChatOllama llm ChatOllama( modelqwen2.5:7b, temperature0.3, base_urlhttp://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(checkpointerMemorySaver())MemorySaver纯内存实现没有文件锁问题适合本地开发调试。确认整条链路OK之后再切到SqliteSaver做持久化。8. 手搓可控Agent的五个经验总结最后我想从项目经验角度说几个个人体会。第一可控Agent的核心不在模型在图。我一开始以为做一个好用的Agent主要靠调prompt后来发现prompt再精巧也扛不住模型随机性带来的不可控。把流程拆成节点、用边和条件边固定决策路径才是真正的可控。第二每个节点只做一件事且函数签名越简单越好。有些代码写出来一个节点里干三件事看似省了步骤实际上出了问题你都不知道是哪一段引发的。第三强烈建议给每个节点加上current_step的状态标记。它相当于程序的日志出了问题你能立刻知道整个图卡在哪一步。没有这个调试LangGraph就像在没有仪表盘的飞机上找故障。第四人工审批不要滥用。interrupt_before虽然好用但每一步都中断用户会被搞疯。我建议只对高影响、不可逆的动作做人工确认比如扣费、删数据、发消息普通的查询、读取没必要。第五先把不带记忆的图跑通再上checkpointer。长期记忆是锦上添花但也是排查问题时最大的干扰源。先把图逻辑调稳再开记忆配合SQLite一步步来遇到问题更容易定位。这个系列的项目文件虽然是个zip但里面的核心不是那一堆代码文件而是如何把Agent放进一个可控的框里这一套方法论。从State设计、Node切分、条件边路由到人工审批和数据持久化串起来就是一个完整的、可以直接对接业务的Agent应用。如果你正在做的项目对Agent的确定性有要求LangGraph这套东西值得你花一个周末好好跑一遍。本文还有配套的精品资源点击获取
返回列表