1. 为什么要用LangGraph构建AI智能体1.1 AI智能体到底“智能”在哪里先聊一个很多人都会混淆的问题普通的大模型调用和真正的AI智能体差别到底在哪单纯调一次LLM API输入Prompt、拿回结果这只是“单次推理”模型答完就结束了既不会查你的数据库也不会去调外部工具。而一个合格的Agent核心能力是自主决策 循环执行它拿到一个目标后能自己判断需要调用什么工具、分析返回结果、根据结果修正下一步动作直到最终目标完成或条件终止。举个例子。你让它“帮我查一下上个月所有未支付订单然后给金额超过一万的客户发一封催款邮件”。一个普通模型只能给你一封邮件的草稿而一个Agent应该能做到先调用订单查询接口拉数据筛选出满足条件的客户清单再调用邮件服务接口批量发送最后给你一份执行报告。这中间涉及了多轮“观察-思考-行动”的循环也就是业界常说的ReAct模式。LangGraph正是为了把这套循环逻辑工程化而生的。它把Agent的每一次决策和动作建模成一张“图”节点Node是处理逻辑边Edge是流转方向所有中间状态放在一个统一的State对象里。这种设计不是花架子而是解决了一个非常现实的问题——复杂Agent流程光靠写死代码容易崩靠手写状态机又维护不动必须有一个框架级别的编排层。1.2 LangGraph相比LangChain原生Agent的核心优势如果你是LangChain的老用户可能会问LangChain本身也有AgentExecutor为什么还要用LangGraph这个问题的答案也是我当年迁移框架的直接原因。第一可控性。LangChain早期的AgentExecutor是一个黑盒循环你很难在中间插入“需要人工审批”“最多执行5步就停止”“某个工具只能在工作时间调用”这类自定义逻辑。LangGraph把整条链路摊开成一张图每一步怎么走、什么时候停、要不要分支全由你定义透明得多。第二持久化与恢复。生产环境中最麻烦的问题就是进程崩溃。Agent跑到一半调了三个工具内存里的状态全丢了怎么恢复LangGraph支持检查点Checkpoint机制可以把每一步的状态保存到外部存储进程重启后直接从断点继续跑。这个能力在异步任务、后台任务、长耗时代理场景里是刚需。第三流式输出。用户不傻等你最好一边执行一边把中间日志和结果推给他。LangGraph对Streaming的支持比较完善节点输出、状态更新、工具调用过程都能实时推送。第四测试与仿真。标题里“生产就绪”四个字落到实操上很大程度体现在“可测试”。LangGraph允许你在一个受控的模拟状态上回放整个过程可以很方便地做回归测试、模拟异常场景这在Agent开发里特别重要——因为LLM的答案带有随机性没有可重复的测试手段你根本不敢往生产推。1.3 生产级Agent的硬性要求再补充一个我自己的判断。很多时候大家说某个Agent“能跑”其实只是原型。生产级和原型之间至少隔着五道坎稳定性同样的输入不能今天能跑、明天就崩。涉及模型的随机性必须有重试、兜底、降级策略。可观测性一旦出问题你能快速定位是模型幻觉、工具报错、还是状态机流转异常而不是只能重启服务。安全与权限Agent能调用的工具必须受限关键操作必须留痕不能出现“Agent自己给自己发了个彻底删库的请求就真的删了”这种事。成本可控一次Agent执行可能调十几次模型Token费用要能测算、能限制。并发与隔离多个用户同时触发Agent各自的状态不能互相污染。这五道坎LangGraph不能说全部替你解决但它提供了对应的基础设施和模式让你不用从零造轮子。下面我带你从零过一遍完整实现。2. 核心概念与基础架构2.1 状态机思维节点、边、状态LangGraph说到底就是一个带状态管理的图执行引擎。三个核心概念State状态、Node节点、Edge边。State是一个全局的数据结构用来保存Agent执行过程中所有重要的信息。最常见的设计是包含一个消息列表和当前步骤标识。你可以把它理解成一张“共享白板”每个节点执行完就往上写几笔下一个节点接着读。Node是一个普通的Python函数或异步函数输入是当前的State输出是一个字典字典里的键值会被更新到新的State中。节点内部可以干任何事情调LLM、调API、查数据库甚至sleep几秒。Edge负责决定执行顺序。它分两种——普通边和条件边。普通边就是“上一个节点跑完无脑去下一个”。条件边则是一个函数根据当前State的内容返回下一步应该跳转到哪个节点。Agent最核心的决策逻辑就是用条件边实现的。我用一个生活类比帮没接触过状态机的人理解你把Agent想象成一个外卖配送员。State就是他的手机地图当前位置、订单列表、剩余时间Node是他要做的具体行动取餐、送餐、联系顾客Edge是路线选择规则如果超时就先打电话如果顺路就一次送多单。LangGraph就是那个帮他规划路线、记录轨迹、随时可以回退重走的调度系统。2.2 与主流编排方案的对比市面上做Agent编排的框架不少我整理了一张对比表方便你选型时有个全局认知。方案核心思路优点短板适用场景LangGraph图状态机节点/边显式定义可控性强、支持持久化和复杂分支学习曲线稍陡生产级多步AgentLangChain AgentExecutor封装好的执行循环上手快灵活度低难插桩快速原型自研Python状态机全部手写无框架依赖维护成本高轮子重复造极简固定流程事件驱动框架消息队列驱动节点执行高并发好Agent逻辑分散排查困难大规模异步任务我的建议很直接如果你做的Agent只有“调一次模型返回结果”这么简单用什么都行但只要涉及多个工具决策、有严格的生产要求LangGraph是当前综合性价比最高的选择。2.3 先构建一个最小可运行的Agent骨架在学习任何新框架时我都建议先别管业务逻辑跑通最小的骨架。只要环境里装好LangGraph和相关依赖下面这段代码就是Agent的“Hello World”。from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定义全局状态结构 class AgentState(TypedDict): messages: list[str] steps: int # 2. 定义节点函数 async def call_model(state: AgentState): print(f模型收到消息: {state[messages]}) # 真实项目里这里会调用LLM先模拟一下 return {messages: state[messages] [模型回复已收到]} async def final_step(state: AgentState): print(流程结束) return {steps: state[steps] 1} # 3. 构建图 graph StateGraph(AgentState) graph.add_node(model, call_model) graph.add_node(final, final_step) graph.add_edge(START, model) graph.add_edge(model, final) graph.add_edge(final, END) app graph.compile()注意这里compile()之后返回的app就是可执行对象后续一切的补充逻辑——加工具、加条件分支、加持久化——都是在这个骨架基础上不断往图里添加节点和边。3. 实战搭建从零实现一个生产就绪的Agent接下来我会用一个具体的业务场景把整个过程串起来。假设我们要做一个“客户售后处理Agent”它需要具备三个能力读取用户问题、从内部订单系统查询订单信息、根据订单状态生成回复。过程中要考虑工具超时、数据不存在、重复调用等问题。3.1 环境准备与实际依赖安装Python版本我建议3.10以上3.9有时候会因为类型语法问题踩坑。核心依赖就两三个装完照着走就行。pip install langgraph langchain-openai如果你的工具调用需要用到LangChain的工具封装还需要装langchain-community或者对应的工具包。这里有个小建议生产项目里工具函数不必强行用LangChain的tool装饰器封装普通Python函数直接用也是完全可以的。我见过太多人为了“框架感”把简单逻辑复杂化没必要。安装完成后建议先跑一个简单的连通性测试确认依赖版本之间没有互相冲突。我建议把版本固定下来不要直接装最新版因为LangGraph迭代速度很快接口变动频繁——这一点后面会专门提醒。3.2 定义业务状态与工具集现在定义业务状态。除了一般的消息列表我还会加上order_id、query_status、tries这些字段。多一个字段就多一层状态管理的灵活性但也不建议一开始堆太多遵循“按需扩展”原则。class CustomerServiceState(TypedDict): messages: list[str] # 对话历史 order_id: str | None # 用户查询的订单号 order_info: dict | None # 查询到的订单数据 retries: int # 工具调用重试次数 finished: bool # 是否可结束工具集这里我用一个函数模拟订单系统查询。核心要点是工具函数必须自己捕获异常并返回结构化结果不要让异常直接抛到Agent的图执行引擎里否则状态机一中断整个流程就死了。import json def query_order_tool(order_id: str) - str: 模拟订单系统查询。返回JSON字符串内部处理所有异常。 try: # 真实项目这里可能是requests调用或数据库查询 mock_db { A1001: {status: 已发货, express: 顺丰, eta: 明天}, A1002: {status: 退款中, amount: 299.0}, } if order_id not in mock_db: return json.dumps({success: False, error: 订单不存在}, ensure_asciiFalse) return json.dumps({success: True, data: mock_db[order_id]}, ensure_asciiFalse) except Exception as exc: return json.dumps({success: False, error: f查询异常: {str(exc)}}, ensure_asciiFalse)注意将工具返回值设计成结构化字符串JSON是比较稳妥的做法。纯文本返回对LLM的理解更友好但丢失了确定性全部返回Python对象又不利于序列化。JSON字符串是两者兼顾的方案。3.3 编排LLM调用与工具循环接下来是做核心的Agent节点。这一步我把它拆成三个节点职责清晰分别负责“思考”、“调用工具”、“收尾判断”。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) def agent_node(state: CustomerServiceState): 负责分析当前状态决定走调用工具还是直接回复。 sys_prompt 你是客服助手。根据用户的对话历史判断是否需要查询订单。 如果用户提供了订单号输出ACTIONQUERY;ORDER_ID单号 如果不需要查单输出ACTIONREPLY;CONTENT回复内容 messages [{role: system, content: sys_prompt}] [ {role: user, content: m} for m in state[messages] ] resp llm.invoke(messages) content resp.content.strip() return {messages: state[messages] [content]} def tool_node(state: CustomerServiceState): 实际执行工具调用。 last_msg state[messages][-1] if ORDER_ID: in last_msg: order_id last_msg.split(ORDER_ID:)[-1].strip() result query_order_tool(order_id) return {order_info: result, retries: state[retries] 1} return {} def router_node(state: CustomerServiceState): 条件边使用的判断函数。 last_msg state[messages][-1] if last_msg.startswith(ACTIONQUERY): return tools return finish然后把它们组装进图里。workflow StateGraph(CustomerServiceState) workflow.add_node(agent, agent_node) workflow.add_node(tools, tool_node) workflow.add_node(finish, lambda state: {finished: True}) workflow.add_edge(START, agent) workflow.add_conditional_edge(agent, router_node, {tools: tools, finish: finish}) workflow.add_edge(tools, agent) # 调用完工具回到agent继续决策 workflow.add_edge(finish, END) app workflow.compile()这个循环结构就是Agent的灵魂agent节点负责决策tools节点负责行动执行完自动回到agent继续下一轮由条件边决定何时退出。调用一次result app.invoke({messages: [帮我查一下订单A1001的状态], order_id: None, order_info: None, retries: 0, finished: False}) print(result[messages])这里要特别提醒必须给Agent设置最大循环次数。否则遇到模型一直输出“调用工具”的故障情况你的服务将陷入死循环消耗大量Token。我的做法是在条件路由里加入次数判断def router_node(state: CustomerServiceState): if state[retries] 5: return finish last_msg state[messages][-1] if last_msg.startswith(ACTIONQUERY): return tools return finish这就是最大的“实战坑”在你把Agent推上线之前一定要先想清楚“什么情况下它会永远不结束”。3.4 加入记忆与上下文管理上面的实现里消息全部放在内存State里一旦服务重启就全丢了而且多轮对话场景下消息列表会越来越长很快触及上下文窗口上限。生产中我建议两个改进。第一接入消息持久化。LangGraph提供MemorySaver和SqliteSaver前者适合开发调试后者才是生产该用的方案。用起来也非常简单from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph # 使用SqliteSaver持久化状态 with SqliteSaver.from_conn_string(:memory:) as saver: app workflow.compile(checkpointersaver) # 每次调用传入thread_id来区分不同会话 config {configurable: {thread_id: user_123}} result app.invoke(input_data, configconfig)注意thread_id就是会话隔离的关键。不同的用户传入不同的thread_id他们的Agent执行状态互不干扰。第二做消息窗口截断。不要在State里无限累加消息。我习惯的做法是只保留最近10轮对话的摘要早期消息用LLM做一次压缩提炼塞进System Prompt里作为背景。这一步能极大稳定模型的表现也能防住Token超限。3.5 加入人工审批与权限控制面向生产环境的Agent往往需要引入Human-in-the-Loop尤其是在执行敏感操作发送邮件、扣款、删除数据之前。LangGraph把人工流程也建模成图节点。实现思路是在执行敏感工具前进入一个human_approval节点把待执行指令推送给管理员然后中断图执行等待外部传入批准或拒绝信号。from langgraph.types import interrupt def human_approval_node(state: CustomerServiceState): action state[pending_action] decision interrupt({action: action, tip: 请确认是否执行该操作}) if decision and decision.get(approved) is True: return {approved: True} return {approved: False} def after_approval_router(state: CustomerServiceState): return do_action if state[approved] else deny有几点经验之谈人工审批不能滥用。如果每个工具都挂审批Agent就失去了“智能”的意义变成半自动工具。建议只对高风险动作挂载。审批接口要超时自动拒绝默认安全原则。如果一个审批挂了半天没人看任务就应该失败走兜底流程而不是一直等着。审批记录要留存。谁批准的、什么时间、基于什么数据做出的决定全部打日志。出问题时这是审计证据。4. 生产就绪的关键细节4.1 可观测性从“玄学”到“可查证”我接手过一个项目线上Agent偶尔会给出错误答案但大家查不到原因因为没有任何Trace。后来我在整个图编排上挂了LangSmith或者自建日志体系把每次节点调用的输入输出全部落库才定位到是某个工具在特定入参下返回了错误格式。我的建议有两个方向可以配合使用。第一接入Managed Trace类服务。这类服务可以自动记录每一步LLM调用、Token数、延时、工具返回。不过要注意线上数据的隐私合规涉及敏感信息要在Trace里做脱敏处理。第二自建一份轻量级审计日志。不是在代码里到处print而是利用LangGraph的节点输入输出设计统一写日志。节点本身接受State并返回更新天然就是一条可审计的数据。我在代码里只要加一个装饰器就能自动记录每个节点的入参和出参。记住一个原则生产环境没有“日志不太好查”这个选项。排查问题靠的是日志不是猜。4.2 错误处理与重试策略LLM接口不像数据库那样稳定超时、限流、返回格式异常都是家常便饭。生产级的Agent必须具备分层重试机制。我的策略分三层第一层LLM调用级别。捕获API超时、RateLimit异常指数退避重试最多重试3次。第二层图执行节点级别。单节点内部逻辑加上try-except让节点自己消化异常并返回能反映错误的状态。第三层整个图执行级别。调用app.invoke()的外层包一层兜底超时或失败时返回用户友好的降级文案而不是让请求直接500。举个例子带重试的LLM调用import time import random def invoke_with_retry(llm, messages, max_retries3): for attempt in range(max_retries): try: return llm.invoke(messages) except Exception as exc: if attempt max_retries - 1: raise wait_time 2 ** attempt random.uniform(0, 0.5) time.sleep(wait_time)4.3 并发与隔离处理当Agent上线后会有成百上千个用户同时使用。如果不加处理每个请求都会占用线程等待LLM响应服务很快就会被打垮。LangGraph本身支持异步执行。把节点函数定义为async def编译后的app调用ainvoke()方法就可以跑在异步事件循环里。配合asyncio.Semaphore控制并发上限防止对一下冲爆模型API的配额。import asyncio semaphore asyncio.Semaphore(50) async def safe_invoke(app, input_data, config): async with semaphore: return await app.ainvoke(input_data, configconfig)状态隔离这块我在3.4提到过thread_id这里再强调一次千万不能在State里存放与特定用户相关的全局字段必须靠checkpointer按thread_id隔离。我见过有人图省事用一个全局dict模拟上下文用户A的订单信息被用户B看到这种事故一旦发生后果非常严重。4.4 成本控制与Token管理Agent是吞Token的大户。一次带工具调用的服务任务可能消耗1-2万Token这比单次ChatCompletion高出10倍以上。成本控制不是上线后才想的事而要从架构上落实。我的个人做法模型分级。简单路由判断用便宜的小模型复杂推理用能力强的模型。LangGraph的节点设计天然支持不同节点用不同模型成本能降不少。Prompt瘦身。System Prompt不是越详细越好冗余内容消耗的Token是持续的。定期审查每一次调用的实际Token消耗把不再需要的背景说明删掉。设置请求级别的Token上限。LangChain的max_tokens参数要设max_retries一次就够了不要让Agent无止境地来回调用。做一个简单的记账字段也是好习惯。在State里加一个total_tokens每次LLM调用后累加最后把消耗写入日志或数据库按天汇总。4.5 测试与评估体系最后这部分可能是最容易被人忽略、但恰恰是“生产就绪”和“演示demo”分水岭的内容Agent的测试。普通后端服务测试是确定性的——输入固定输出固定。Agent的输出有随机性所以必须建立一套不同的测试体系。我的方案分三层单元测试针对单个节点函数。把State构造好调用节点断言返回是否符合预期。比如给agent_node塞一个包含订单号的对话断言返回文本里包含ACTIONQUERY。集成测试针对整张图。写一批典型用户对话运行整个Agent流程断言最终能走到finish节点且不卡死不超过5轮。我维护了一个“金标数据集”每条数据标注期望行为和期望工具调用每次改模型或Prompt都用这批数据回归一遍。线上影子测试正式发布前把新版本Agent和旧版本同时跑一遍比较结果一致率。这个做法费些Token但能让你安心上线。5. 常见问题与排查技巧实录5.1 Agent卡在循环里不退出这是最常遇到的问题。症状任务日志显示agent节点和tools节点反复交替执行永远不进入finish。排查思路第一看条件路由函数router_node是否把所有可能的情况都覆盖了第二看模型是否持续输出“ACTIONQUERY”如果是多半是Prompt里没写清楚终止条件需要明确指示“当消息不再包含新的订单号时直接回复”第三检查工具结果是否真的被模型看到了有时候工具返回内容太长被截断模型误以为没查到继续反复查。根治方案就是我前面提到的条件路由里加retries上限达到阈值直接强行走finish。这个兜底必须加。5.2 工具返回的数据格式不稳定模型从工具结果里提取信息时如果工具返回的是大段非结构化的文本很容易提取不完整。我建议工具返回时尽量使用严格JSON结构并且在Prompt中明确告知“工具返回为JSON请基于其中的data字段生成回复”。另外工具解析时不要用正则去抠字段直接json.loads()之后索引取值取不到就返回明确错误不要猜测。猜测出来的数据往往是错的在客服场景里尤其致命。5.3 多轮对话的上下文越长模型表现越差和模型聊到第10轮模型开始忘记最初的订单号这是上下文窗口污染的典型表现。原因往往是State里累积了太多工具调用细节这些噪音干扰了模型注意力。我的处理办法是在消息列表入模型前做预处理把工具调用这一类过程性消息精简为一行摘要“已查询订单A1001状态已发货无异常情况”再加上过去两轮的对话原文。这样既保留了关键信息又滤掉了干扰。5.4 并发了但有点慢该怎么优化还有一类性能问题。同样的Agent任务单个用户跑2秒100个人并发跑就变成10秒。先排查是不是工具的IO成了瓶颈比如数据库查询。优化手段按顺序试工具调用改成异步IO对稳定数据做缓存再做并发数限流防止请求积压。LangGraph支持节点级别的异步函数把同步的requests调用换成httpx.AsyncClient效果立竿见影。5.5 LangGraph版本更新导致接口变动这个坑比较隐蔽。LangGraph当前版本迭代飞快接口偶尔会有Breaking Change。今天的代码下周可能就不能跑了。我现在的应对措施是固定依赖版本并且锁定精确版本号比如langgraph0.x.x不随便升级。升级前先读Changelog升级后必须完整跑一遍金标测试集。最后再分享一点实操体会如果你也在试着搭建Agent我的建议是从小处着手先实现一个只有两个节点的最小闭环跑通后再逐步加工具、加状态、加人工审批。别一开始就照着复杂架构图堆节点那只会让你面对一屏的报错无从下手。另一个体会是Agent的生产调试本质上是在调试“不确定性”。它的对手不只是代码Bug还有模型的随机行为。你要做的是用确定性工具状态、路由、日志、测试去约束和控制不确定性让Agent越来越“靠谱”。我踩过最大的坑就是急着加新功能反而忽略了边界条件。现在每做一个Agent项目我都会在开发早期就把“最大轮数”“失败兜底”“Token上限”这些护栏逻辑先写进去。能少走很多弯路。这些经验如果对你有用拿去做你的第一个生产级Agent应该能省下不少试错时间。