LangChain+LangGraph实现企业级Agent:从Demo到工作流编排与可观测部署

发布时间:2026/9/1 13:00:49
LangChain+LangGraph实现企业级Agent:从Demo到工作流编排与可观测部署 这两年 AI Agent 已经从概念变成很多团队 KPI 里的关键词。但如果你真的用 LangChain 写过 Agent大概率会遇到这样一组问题本地 Demo 调得很顺模型回答也很聪明一接真实业务就发现流程不可控、结果不可复现、出了问题查不到上下文。这不是模型不够强而是编排层没跟上。这篇内容想聊清楚一件事用 LangChain LangGraph 做企业级 Agent真正的关键点不是“多会调 Prompt”而是把 Agent 从“自由发挥的对话模型”改造成“有状态、可路由、可观察、可回滚的工程系统”。本文将沿着一条完整路线展开先讲 LangChain 与 LangGraph 的分工与区别再带读者从 V1 版本的最小 Agent 骨架出发逐步重构为工作流编排的 StateGraph接着用一个 TextToSQL Agent 作为综合实战最后落到可观测部署与运维排查。读完你应该能回答三个问题第一LangGraph 到底解决 LangChain 的哪些短板第二如何把一个随手能跑的 Agent 改造成适合生产的工作流第三企业级 Agent 上线前哪些监控和验证环节不能跳过。1. 这篇文章真正要解决的问题现在网上关于 LangChain 和 LangGraph 的教程很多但大多数停留在“怎么装、怎么跑”的层面。很多团队照教程写完一个 Agent进入真实业务后遇到的第一波问题往往不是模型回答错误而是流程本身出了问题某个工具调用超时了整个对话卡死Agent 在某个节点反复循环Token 费用飙升一次回答依赖好几个中间状态但程序一重启状态就丢了。这些问题有一个共同根源Agent 没有明确的流程边界和状态管理。企业级 Agent 和普通 Demo 最大的区别体现在四个维度上可控性节点是否可编程、可跳转、可终止而不是完全交给模型自由发挥。可观测性每轮对话、每次工具调用、每次模型请求是否都有完整 Trace。状态持久化多轮对话、长任务、异步执行时状态能否被持久化和恢复。安全性Agent 访问数据、触发外部操作时有没有权限边界和风控手段。从 V1 到工作流再到 TextToSQL 和可观测部署正好覆盖了这四个维度。V1 解决的是“能不能跑”工作流解决的是“能不能控”TextToSQL 解决的是“能不能做真实的业务动作”可观测部署解决的是“上线后能不能维护”。这篇文章适合三类读者一是已经在用 LangChain 写 Agent但总觉得项目推进不下去的开发者二是准备在公司内部搭建 Agent 服务需要向团队解释技术选型的架构师三是刚接触 LLM 应用开发想把概念一次性理清、少走弯路的后端工程师。如果你属于其中一类读完之后可以沿着文中的代码结构直接改造自己的项目。2. LangChain 与 LangGraph先搞清楚分工再动手很多初学者第一次看到“LangChain 和 LangGraph 的区别”这个问题时第一反应是 LangGraph 是 LangChain 的升级版。这个理解不够准确。从官方定位来看LangChain 是一整套围绕 LLM 应用开发的基础工具集合包括模型封装、提示词模板、检索增强RAG、工具调用、输出解析和记忆组件而 LangGraph 是其中的有状态编排引擎专门用来把大模型、工具和外部系统组织成可控的图结构。可以把 LangChain 理解成一个“工具箱”LangGraph 则是“流水线控制系统”。工具箱里有很多好用的工具ChatOpenAI 封装了大模型调用DocumentLoader 负责文档加载有很多现成的检索器。但“每一步先做什么、再做什么、出错时怎么办”在传统 LangChain 代码里是没什么体现的。最常见的写法是这样用 AgentExecutor 挂上一堆工具把任务丢给模型让模型自己决定调用顺序。# 传统 LangChain Agent 写法简化示例 from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.agents import Tool llm ChatOpenAI(modelgpt-4o-mini, temperature0) tools [ Tool(namesearch, funcsearch_func, description搜索公开信息), Tool(namecalculator, funccalc_func, description数学计算), ] agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools) result executor.invoke({input: 帮我查一下上海今天的天气})这种方式写起来非常快但它的核心逻辑是模型在循环里自主决策。模型说“调用搜索”就调用搜索说“结束”就结束。在简单场景下这没问题但你会发现流程无法固定、状态无法持久化、中间步骤无法被外部系统观测和控制。模型返回出错时你很难知道它到底经过了哪些节点。LangGraph 改变了这个思路。它把一次 Agent 任务建模成一个带状态的图State状态在图执行过程中共享的数据结构类似 Web 服务里的请求上下文每一轮节点更新后都写入新的状态。Node节点一个具体的处理步骤可以是大模型调用、工具执行、规则判断或数据库读写。Edge边节点之间的连接关系表示执行顺序。Conditional Edge条件边根据当前状态决定走哪个分支的路由器。Subgraph子图一个大 Agent 内部嵌套的独立小图用于模块化复用。和传统 AgentExecutor 最大的区别是LangGraph 把执行路径从“模型自由决定”改成了“开发者定义路由规则模型在规则内做选择”。这意味着你可以在关键节点上加上校验、分支和熔断逻辑Agent 的每一步都变得可预期、可追踪。下表可以帮你快速区分两者对比维度LangChain AgentExecutorLangGraph StateGraph执行模型模型自主循环开发者定义图结构模型在节点内推理状态管理依赖外部 Memory易丢失内置 State 数据结构支持持久化流程控制弱难以介入中间步骤强支持条件路由、分支、循环、子图可观测性依赖回调提升基础字段有限原生 Trace节点级监控适用场景原型验证、简单问答生产级工作流、复杂多工具编排所以更稳妥的判断是如果你只是做原型或内部小工具LangChain 的 AgentExecutor 仍然够用如果你的 Agent 要接真实业务、要服务多用户、要做多轮长任务LangGraph 才是那个把工程问题交给开发者的关键组件。3. 环境准备与项目基础结构进入实操之前先准备好运行环境。下面是本文建议的最小环境操作系统macOS / Linux / WindowsWSL 更推荐Python 版本建议 3.10 及以上包管理pip 或 poetry推荐创建独立虚拟环境大模型 APIOpenAI 兼容接口即可也可以用国内大模型厂商的 OpenAI 兼容服务关键依赖langchain、langchain-openai、langgraph、langsmith建议先创建虚拟环境再安装依赖# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate # 安装核心依赖 pip install --upgrade pip pip install langchain langchain-openai langgraph langsmith如果你的服务器无法访问外部模型服务也可以在本地用 vLLM 或 Ollama 启动 OpenAI 兼容的服务然后在环境变量中替换 base_url。本文示例代码集中在 LangGraph 编排逻辑上所以模型 API 选用不影响核心结构。项目目录建议如下agent_project/ ├── .env # API Key 与环境配置 ├── requirements.txt # 依赖锁定文件 ├── app/ │ ├── __init__.py │ ├── state.py # Agent 状态定义 │ ├── nodes.py # 图节点实现 │ ├── router.py # 条件路由函数 │ ├── graph.py # 图构建与编译 │ └── tools.py # 工具函数与工具注册 ├── examples/ │ └── v1_simple_agent.py # V1 最小示例 └── tests/ └── test_basic_flow.py在项目根目录准备.env文件主要包含模型服务配置# .env OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 LANGCHAIN_TRACING_V2true LANGCHAIN_API_KEYlsv2-xxxx LANGCHAIN_PROJECTenterprise-agent-demo这里要特别提醒一句不要把 API Key 提交到 Git 仓库。生产环境的密钥管理应该放到密钥管理服务或 CI/CD 的 Secret 中。4. V1 Agent先跑通最小可用的 Agent 骨架先写一个最朴素的 V1 Agent。目标是模型能调用工具、能返回最终答案代码尽量短先跑通链路。后面再基于这个版本逐步改造。4.1 定义一个简单的计算工具# 文件路径app/tools.py def add(a: float, b: float) - float: 两个数字相加 return a b def multiply(a: float, b: float) - float: 两个数字相乘 return a * b TOOLS [ { type: function, function: { name: add, description: 计算两个数字的和, parameters: { type: object, properties: { a: {type: number, description: 第一个数字}, b: {type: number, description: 第二个数字} }, required: [a, b] } } }, { type: function, function: { name: multiply, description: 计算两个数字的乘积, parameters: { type: object, properties: { a: {type: number, description: 第一个数字}, b: {type: number, description: 第二个数字} }, required: [a, b] } } } ]工具定义采用 OpenAI Function Calling 风格LangChain 0.2 以上版本可以直接把这种格式传给模型模型会在需要时返回工具名称和参数。4.2 用 LangChain 封装 AgentExecutor# 文件路径examples/v1_simple_agent.py import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate from app.tools import TOOLS llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) tools [ { name: add, description: 计算两个数字的和, func: lambda a, b: a b, }, { name: multiply, description: 计算两个数字的乘积, func: lambda a, b: a * b, }, ] # 这里使用工具描述让模型知道有哪些函数可调用 prompt ChatPromptTemplate.from_messages([ (system, 你是一个能调用计算工具的助手。请根据用户问题选择合适的工具。), (human, {input}), ]) # 为了简化示例我们直接把工具名称映射到本地函数 def call_tool(tool_name: str, args: dict): fn_map { add: lambda: tools[0][func](args[a], args[b]), multiply: lambda: tools[1][func](args[a], args[b]), } return fn_map[tool_name]() model_with_tools llm.bind_tools(TOOLS) # 自定义 Agent 循环调用模型 - 如果返回工具则执行 - 继续调用模型 def simple_agent(user_input: str, max_iterations: int 5): messages [ {role: system, content: 你是一个能调用计算工具的助手。请根据用户问题选择合适的工具。}, {role: user, content: user_input}, ] for _ in range(max_iterations): response model_with_tools.invoke(messages) # 如果模型没有返回工具调用说明已经给出最终答案 if not response.tool_calls: return response.content # 逐个执行工具调用 for tool_call in response.tool_calls: tool_name tool_call[name] tool_args tool_call[args] result call_tool(tool_name, tool_args) # 把工具执行结果追加到对话历史 messages.append({ role: assistant, content: response.content, tool_calls: [tool_call], }) messages.append({ role: tool, tool_call_id: tool_call[id], content: str(result), }) return 达到最大迭代次数未得到最终结果。 if __name__ __main__: print(simple_agent(请计算 12 和 5 的和再乘以 3))这个示例没有直接依赖 AgentExecutor而是手动实现了一个最小的“模型-工具-模型”循环目的是把 Agent 的内部机制暴露出来。你会发现V1 Agent 的问题非常明显循环条件固定只能靠 max_iterations 硬限制模型失败后没有分支兜底。状态不透明整个 messages 列表是隐式状态外部看不到中间过程。无结构化输出如果某个工具调用失败你只能在字符串里等待模型自我纠错。不可观测没有 Trace出问题只能靠 print 猜测。这个版本适合用来验证模型能力和工具逻辑但不适合进生产。接下来的章节我们就用它做基底改造成 LangGraph 工作流。5. 从 V1 到工作流StateGraph 条件路由与子图编排LangGraph 的核心价值在于把 Agent 执行过程变成一张图开发者可以定义状态、节点、边和条件路由。这里我们用同一套“计算工具”场景演示怎么把 V1 改造成 StateGraph。5.1 定义状态结构# 文件路径app/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): # 消息列表add_messages 是 LangGraph 提供的增量合并器 messages: Annotated[list, add_messages] # 当前已经尝试过的工具执行次数用于循环检测 tool_attempts: int # 最终结果 answer: str这里Annotated[list, add_messages]是一个关键设计它告诉 LangGraph当多个节点更新 messages 字段时不是直接覆盖而是把新消息追加到旧消息后面。这个机制对应 Agent 多轮对话的上下文累积需求。5.2 编写节点函数# 文件路径app/nodes.py from app.state import AgentState from app.tools import TOOLS def llm_node(state: AgentState): 调用大模型返回最终回答或工具调用指令 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools llm.bind_tools(TOOLS) response model_with_tools.invoke(state[messages]) return {messages: [response]} def tool_node(state: AgentState): 执行模型请求的工具调用并把结果写回状态 from langchain_core.messages import ToolMessage tool_map { add: lambda args: args[a] args[b], multiply: lambda args: args[a] * args[b], } messages [] attempts state.get(tool_attempts, 0) for tool_call in state[messages][-1].tool_calls: tool_name tool_call[name] tool_args tool_call[args] try: result tool_map[tool_name](tool_args) content str(result) except Exception as e: content f工具执行失败: {str(e)} messages.append(ToolMessage( contentcontent, tool_call_idtool_call[id], )) attempts 1 return {messages: messages, tool_attempts: attempts}关键逻辑说明llm_node负责推理它会基于当前所有历史消息决定是返回最终答案还是调用工具。tool_node负责执行它读取模型返回的最后一个消息中的 tool_calls 字段执行对应工具并把结果作为 ToolMessage 追加到消息列表。如果工具执行出错我们不是让程序崩溃而是把错误信息作为 ToolMessage 返回给模型让模型自行决定如何修正。这是 Agent 容错的关键设计。5.3 条件路由模型下一步该去哪里# 文件路径app/router.py def should_continue(state: AgentState) - str: 根据模型返回判断走工具节点还是结束 返回字符串是路由分支的名称 last_message state[messages][-1] if last_message.tool_calls: # 还有工具调用进入工具执行节点 return tool # 没有工具调用说明模型已经给出最终回答 return end条件路由是 LangGraph 最重要的能力。它并不复杂本质就是一个普通 Python 函数输入当前状态输出一个分支名称。这个函数可以包含任意业务规则比如当工具执行次数超过阈值时强制结束当某个关键工具失败时进入人工审核节点。5.4 组装图结构# 文件路径app/graph.py from langgraph.graph import StateGraph, START, END from app.state import AgentState from app.nodes import llm_node, tool_node from app.router import should_continue builder StateGraph(AgentState) # 添加节点 builder.add_node(llm, llm_node) builder.add_node(tool, tool_node) # 设置入口 builder.add_edge(START, llm) # 条件路由llm 节点执行后决定下一个节点 builder.add_conditional_edges( llm, should_continue, { tool: tool, end: END, } ) # 工具节点执行完成后回到 llm让模型看到工具结果后继续推理 builder.add_edge(tool, llm) # 编译成可执行的图 agent_graph builder.compile()这段代码展示了一个经典的 Agent 循环START - llm - (条件路由) - tool - llm - ... - END。模型每次回答都会先经过条件路由判断如果返回工具调用就执行工具并把结果喂回去直到模型给出最终答案为止。调用方式如下from app.graph import agent_graph result agent_graph.invoke( {messages: [{role: user, content: 请计算 12 和 5 的和再乘以 3}], tool_attempts: 0} ) print(result[messages][-1].content)执行过程可以通过result[messages]完整复盘第一步模型调用add(12, 5)返回 17第二步模型调用multiply(17, 3)返回 51第三步模型给出最终答案“结果是 51”。5.5 循环检测与子图LangGraph 图结构本身允许环的存在但这也会带来失控风险。实际工程中一般会在路由函数里加上循环次数限制# 在 router.py 中加入次数控制 def should_continue_with_limit(state: AgentState) - str: last_message state[messages][-1] # 超过 5 次工具调用强制结束避免无限循环和费用失控 if state.get(tool_attempts, 0) 5: return end if last_message.tool_calls: return tool return end子图则是另一种常见需求。当 Agent 的业务变复杂比如一个客服 Agent 内部还包含“订单查询 Agent”“退款流程 Agent”可以把内部流程封装成子图作为一个节点挂进主图。在 LangGraph 中子图的编译结果可以直接作为父图的节点使用# 子图封装为节点示例 order_subgraph order_builder.compile() chat_builder.add_node(order_query, order_subgraph)这种分层设计能有效控制代码复杂度让主图只承担业务路由子图承担细化执行。团队协作时不同子图可以由不同成员独立开发和测试。6. TextToSQL Agent 实战从自然语言到可执行 SQLTextToSQL 是 Agent 在企业场景中落地价值很高的方向之一。业务人员不写 SQL用自然语言提问“上个月华东区销售额排名前五的产品是什么”系统自动生成 SQL、查询数据库、返回答案。听起来很美好但真实项目里有几个绕不开的坑。6.1 难点拆解第一是Schema 上下文。数据库表可能几十张每张表几十个字段全塞进 Prompt 会让模型失去注意力也浪费 Token。合理的做法是动态选择相关表只把相关表结构发给模型。第二是SQL 正确性。模型生成的 SQL 可能语法正确但逻辑错误比如 join 条件写错、聚合函数用错。所以不能只让模型生成 SQL 就结束必须实际执行或至少做语法校验。第三是安全性。TextToSQL 涉及数据库访问必须约束为只读连接、限制可访问的表和列、设置查询超时防止模型生成危险操作。6.2 架构设计用户输入 - 1. 选择相关表根据用户问题从表清单中筛选 - 2. 构建 Schema Prompt - 3. LLM 生成 SQL - 4. SQL 校验与执行只读 - 5. 结果格式化 - 6. 返回给用户用 LangGraph 实现可以设计四个节点select_tables、generate_sql、execute_sql、format_answer并在generate_sql之后加入一个条件路由SQL 校验失败时回到生成节点重试重试超过两次则进入人工处理。6.3 核心代码示例# 文件路径app/text2sql_nodes.py import sqlite3 # 模拟数据库连接信息 DB_PATH ./demo.db # 表清单和字段说明实际项目中可以从 information_schema 动态获取 TABLE_METADATA { products: { description: 产品信息表, columns: { id: 主键, name: 产品名称, category: 产品分类, price: 单价元, } }, orders: { description: 订单表, columns: { id: 订单ID, product_id: 产品ID关联 products.id, quantity: 数量, order_date: 下单日期, region: 区域, } } } def select_tables_node(state): 根据用户问题选择相关表简化示例直接返回所有表 user_question state[messages][-1].content # 实际项目中可以用 embedding 或关键词匹配筛选相关表 related_tables list(TABLE_METADATA.keys()) return {related_tables: related_tables} def generate_sql_node(state): 调用模型生成 SQL from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage user_question state[messages][-1].content related_tables state[related_tables] schema_text for table in related_tables: meta TABLE_METADATA[table] schema_text f表 {table}{meta[description]}字段\n for col, desc in meta[columns].items(): schema_text f - {col}: {desc}\n llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f你是一个专业的 SQL 工程师。根据用户问题生成 SQLite 查询 SQL。 可用表结构 {schema_text} 要求 1. 只做 SELECT 查询不能生成 INSERT/UPDATE/DELETE 等语句 2. 只使用上述表结构中的表和字段 3. 如果有时间条件使用订单表的 order_date 字段 4. 结果只输出 SQL 语句本身不要额外解释 用户问题{user_question} response llm.invoke([HumanMessage(contentprompt)]) sql response.content.strip() # 简单提取 SQL去掉可能出现的 markdown 代码块标记 if sql.startswith(): sql sql.split()[1] if sql.startswith(sql): sql sql[2:] sql sql.strip() return {sql: sql, sql_attempts: state.get(sql_attempts, 0) 1} def execute_sql_node(state): 执行 SQL只读模式并返回结果 import sqlite3 sql state[sql] # 生产环境请使用只读账号这里演示打开只读连接 conn sqlite3.connect(ffile:{DB_PATH}?modero, uriTrue) cursor conn.execute(sql) # 只取前 20 行防止返回数据过大 columns [desc[0] for desc in cursor.description] rows cursor.fetchmany(20) conn.close() return {query_result: {columns: columns, rows: rows}} def validate_sql_node(state): 校验 SQL 是否安全 sql state[sql].strip().lower() # 黑名单机制强制禁止非 SELECT 语句 forbidden_keywords [insert, update, delete, drop, alter, create] for kw in forbidden_keywords: if sql.startswith(kw) or f {kw} in sql: return invalid return valid6.4 组装 TextToSQL 图# 文件路径app/text2sql_graph.py from langgraph.graph import StateGraph, START, END from app.text2sql_nodes import ( select_tables_node, generate_sql_node, validate_sql_node, execute_sql_node, ) from app.state import AgentState builder StateGraph(AgentState) builder.add_node(select_tables, select_tables_node) builder.add_node(generate_sql, generate_sql_node) builder.add_node(validate_sql, validate_sql_node) builder.add_node(execute_sql, execute_sql_node) builder.add_edge(START, select_tables) builder.add_edge(select_tables, generate_sql) builder.add_edge(generate_sql, validate_sql) # SQL 校验失败时重新生成最多重试 2 次 def route_after_validate(state): if state.get(sql_attempts, 0) 2: return execute_sql # 实际项目中可改为进入人工处理节点 if state[validation_result] invalid: return generate_sql return execute_sql builder.add_conditional_edges( validate_sql, route_after_validate, { generate_sql: generate_sql, execute_sql: execute_sql, } ) builder.add_edge(execute_sql, END) text2sql_graph builder.compile()这里的重点是validate_sql之后的条件路由它实现了“生成 SQL - 校验 - 不合格就重新生成”的闭环。同时也体现了 LangGraph 的价值模型负责生成和修正规则负责兜底和拦截。6.5 安全提示TextToSQL 的安全性再怎么强调都不过分。生产环境至少要做到使用独立数据库账号只授予 SELECT 权限。配置查询超时时间比如单条 SQL 超过 10 秒直接终止。敏感表、敏感字段做脱敏或禁止访问。SQL 关键字黑名单只是最低要求更可靠的是用 SQL Parser 解析出 AST确认只包含查询语句。所有生成过的 SQL 都写入审计日志便于事后追踪。7. 可观测部署Agent 上生产前必须补的运维课Agent 比普通 API 服务更难运维因为一次用户请求会触发多次模型调用、工具调用和路由决策任何一个环节出问题都可能导致最终回答质量下降。如果没有 Trace排查问题几乎等于大海捞针。7.1 可观测的三个层次企业级 Agent 可观测性至少要覆盖三层服务层请求量、响应时间、错误率、Token 用量这是传统监控就能覆盖的。执行层一次 Agent 请求经过了哪些节点、每个节点耗时多少、路由到了哪个分支这就是 Trace。模型层每次 LLM 调用的 Prompt 是什么、响应是什么、延迟多少、消耗了多少 Token。本地开发时最简单的可观测配置是开启 LangSmith。LangSmith 是 LangChain 生态中的可观测平台可以自动捕获每次 Agent 运行的完整 Trace包括节点调用顺序、工具输入输出、模型耗时和 Token 统计。# 在代码中启用 LangSmith import os from dotenv import load_dotenv load_dotenv() # 确保 .env 中已经配置 # LANGCHAIN_TRACING_V2true # LANGCHAIN_API_KEYlsv2-xxxx # LANGCHAIN_PROJECTenterprise-agent-demo from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 之后所有 LangChain 调用和 LangGraph 执行都会自动上报 Trace如果公司内网环境不允许使用云端服务也可以考虑私有化部署 LangSmith或者直接把执行日志写成结构化 JSON。7.2 结构化日志自己的兜底方案不是所有团队都愿意把数据上报到第三方平台更通用的是在代码里埋点输出结构化日志。LangGraph 节点本身就是普通 Python 函数所以非常容易在节点入口和出口打印关键信息# 文件路径app/observability.py import json import time from datetime import datetime def log_node_start(node_name: str, state: dict): print(json.dumps({ event: node_start, node: node_name, ts: datetime.utcnow().isoformat(), message_count: len(state.get(messages, [])), }, ensure_asciiFalse)) def log_node_end(node_name: str, state: dict, duration_ms: float, extra: dict None): payload { event: node_end, node: node_name, ts: datetime.utcnow().isoformat(), duration_ms: round(duration_ms, 2), } if extra: payload.update(extra) print(json.dumps(payload, ensure_asciiFalse))在实际项目中可以把这些日志发送到 ELK、Loki 或 ClickHouse 等日志平台再配合 Grafana 展示。相比直接看 LangSmith结构化作法更可控适合对数据敏感的企业环境。7.3 部署形态与成本控制LangGraph 应用本身只是 Python 服务可以按普通后端服务方式部署。LangGraph 平台也提供了一键部署、任务队列、持久化会话等能力但如果你已经有一套成熟的微服务基础设施也可以直接把graph编译结果封装成 HTTP API跑在 Kubernetes 上。这里需要提醒一个成本问题Agent 的 Token 消耗比普通 RAG 问答高得多因为一次任务可能要经历“模型-工具-模型-工具”多轮调用。生产环境必须设置预算上限比如单次请求最大工具调用次数、单次请求最大 Token 数、单用户每日配额。LangGraph 的条件路由和recursion_limit就是最直接的兜底工具。8. 常见问题与排查思路以下是 LangChain LangGraph 开发中最高频的问题按经验整理了排查路径问题现象可能原因排查方式解决方案图执行时提示RecursionLimitAgent 循环次数超过默认限制查看 Trace 中哪个节点反复执行在路由函数中增加工具调用次数上限或调大recursion_limit工具调用返回后模型不继续推理ToolMessage 的tool_call_id与 assistant 消息不匹配检查 messages 列表结构严格使用模型返回的tool_call_id不要手动编造模型总是讨论而不调用工具工具描述不清晰或 Prompt 未约束查看模型响应日志优化工具description在系统提示词中明确“需要时直接调用工具”输出 SQL 被 markdown 包裹模型生成了 sql 代码块在生成节点后做字符串清洗用正则或字符串截取方式清理 markdown 标记TextToSQL 查询超时表数据量大或 SQL 缺少索引查看数据库慢查询日志增加数据库查询超时限制返回行数必要时建立索引调用 LangGraph 后无 Trace 上报环境变量未生效或 API Key 配置错误检查LANGCHAIN_TRACING_V2和LANGCHAIN_PROJECT确保在加载模型前完成环境变量配置重启进程线上模型与本地效果不一致使用了不同版本模型或 temperature 设置不同对比两边的模型参数和系统提示词使用固定模型版本保持 temperature 一致Agent 执行结果不稳定模型温度过高或 Prompt 信息不足查看 Trace 中同一问题的多次输出降低 temperature增加必要上下文建立回归测试集除了表格里的问题还有一个非常隐蔽的坑需要单独说不要在生产代码里把大模型请求放到用户请求的同步链路上。LangGraph 本身支持异步执行和持久化如果你的 Agent 任务比较耗时应该设计成“提交任务 - 后台执行 - 通过回调或轮询获取结果”的异步模式而不是让用户一直等待 HTTP 响应。9. 最佳实践与工程建议结合前文的项目经验这里整理一份企业级 Agent 的工程建议清单按代码开发、流程设计、运维部署三个维度展开。代码与状态设计State 结构不要在开发后期才定义。它是 LangGraph 应用的“接口协议”建议在项目一开始就和团队成员评审。状态里只放需要跨节点共享的数据临时变量不要塞进 State否则会污染 Trace 和调试体验。消息合并器add_messages很好用但请注意它只适合消息列表字段其他业务字段还是要显式控制合并逻辑。所有节点函数建议保持“纯函数”风格输入 State返回新的字段增量不修改外部全局变量。这样每个节点都可以单独测试也方便 mock。流程设计企业级 Agent 不鼓励把所有业务判断都交给模型。通用原则是需要灵活推理的地方用模型需要确定性结果的地方用规则。比如 SQL 生成让模型做SQL 校验就交给解析器工具选择让模型做工具输入参数校验就交给 JSON Schema。条件路由函数是业务逻辑最容易失控的地方建议为每个分支编写单元测试。比如“工具执行失败时是否走了降级分支”“超过重试次数后是否正确进入人工处理”。这些测试不需要真实调用大模型可以用 mock 数据验证路由逻辑。运维与安全Agent 上线前建议建立最小回归集准备 20 条典型业务问题每次模型版本或提示词变更后自动跑一遍记录回答质量和执行路径。这一步能显著降低“改了一个 Prompt另外九个场景全部受影响”的风险。数据库类 Agent 必须遵循最小权限原则。TextToSQL 用只读账号限制可访问的表和列配置超时和返回行数上限所有 SQL 记录审计日志。Agent 触发外部写操作时必须经过人工确认或额外的二次验证机制。后续学习方向如果这篇内容你已经完全消化下一步可以沿着三个方向深入学习 LangGraph 的持久化能力实现多轮会话状态恢复和长任务异步执行。为你的 Agent 接入评估体系用 LLM 作为裁判自动评估回答质量建立 Prompt 回归测试。把工作流从单 Agent 扩展到多 Agent 协作设计主管 Agent 和多个专用子 Agent 之间的任务分发与结果汇总。最后留一个建议不要一上来就追求复杂的图结构。先从 V1 骨架跑通用 StateGraph 替换循环再逐步增加条件路由和子图。Agent 复杂度的每一次上升都应该有明确的业务收益来支撑。工程上真正重要的不是把图画得多复杂而是让每条路径都可解释、可控制、可回滚。把这套基本功打扎实LangChain LangGraph 会成为你在企业级 AI 应用落地时非常趁手的工具组合。