很多团队做了半年 Agent最后落地的 Demo 还是“一个 while 循环 一个 OpenAI 函数调用”。代码能跑但一碰到线上流量、权限边界、外部系统接入、问题回溯立刻捉襟见肘。本文不聊概念直接给你一套 LangChain 1.0LangGraph 1.0 的企业级 Agent 搭建思路覆盖状态图设计、条件路由、MCP 工具接入、全链路可观测和常见生产问题附完整可运行代码。之前我在业务迭代中被两件事逼疯一是 Agent 行为不可控模型一旦多轮调用就容易绕圈、丢状态二是接外部系统时每个系统都要自研一套“函数调用协议”接口层次不齐联调效率极低。后来把 Agent 的编排层从 LangChain 原生链式调用切换到了 LangGraph 状态图再用 MCPModel Context Protocol统一外部工具接入最后在关键路径上补了结构化日志和链路追踪才真正感受到“工程化 Agent”和“Demo Agent”的区别。这篇文章适合三类读者已经会用 LangChain 写简单工具调用但想进阶到带状态、带分支、带人工审批的 Agent。后端开发想把 Agent 接入企业内部工单、数据库、RAG 服务但对 MCP 协议还不熟悉。正在做 Agent 生产落地的技术负责人想找一份可观测、有审计、能出问题的排查清单。读完你会掌握LangChain 和 LangGraph 的分工边界、LangGraph 状态图核心建模方法、MCP 在 LangChain 生态里的接入方式、以及一套从“记录日志”升级到“全链路可观测”的落地路径。1. 为什么企业级 Agent 不能再手搓 Demo1.1 LangChain 并没有过时而是“拆得越来越清楚”很多开发者看到 LangGraph 之后会问一句话LangChain 是不是要被 LangGraph 替代了我的结论LangChain 没有过时它和 LangGraph 是不同抽象层级的组件。LangChain 更像一个组件库大模型统一封装、Prompt 模板、向量库集成、各类文档加载器、输出解析器。LangGraph 是一个编排引擎负责 Agent 的流程控制、状态管理、分支路由、循环终止、持久化和人工介入。简单理解LangChain 负责“怎么跟模型和外部资源打交道”LangGraph 负责“整个任务怎么一步步跑完”。两者是配合关系不是替代关系。从 1.0 版本之后LangChain 官方也明显把重心往 LangGraph 这个“Agent 运行时”上移。你写的业务逻辑应该尽量是“状态图里的节点”而不是一条线性的 Chain。这样后续加审批、加重试、加多分支改动成本会低很多。1.2 Demo Agent 与企业级 Agent 的核心差距手搓 Demo 的时候Agent 通常长这样用户提问。把问题、工具列表、历史记录一股脑塞给大模型。大模型返回一个工具调用。代码里exec()或者 if-else 执行函数。把结果拼回去再调用一次大模型。循环直到模型说“完成”。这套流程在小范围验证时没问题但企业级场景下会暴露五个短板短板Demo 表现企业级要求状态管理所有状态存在一个 dict 里覆盖即丢失明确 State 结构支持多轮累积流程控制while 循环难以精确控制退出图结构节点、边、条件路由清晰人工介入无法暂停、审批、回退human-in-the-loop执行到审批节点暂停工具协议每个系统一套 SDK难维护MCP 统一工具协议动态加载可观测性print 日志出问题靠猜全链路 trace关键节点可回溯1.3 企业级 Agent 的三个核心维度结合我自己的工程经验下面三个维度是判断一个 Agent 能不能上生产的底线第一安全可控。Agent 不能是一个“模型自由发挥的黑盒”。要有最大步数限制要有工具白名单要有数据脱敏要有敏感操作审批。LangGraph 的状态图天然适合做这些每个节点都是一个函数函数内部可以做权限校验每条边都可以加条件条件不满足就走进度分支。第二标准化接入。内部系统千奇百怪如果每个系统都单独开发函数调用接口Agent 的工具层会很快腐化。MCP 的价值在于它给“外部工具”定了一套统一协议工具描述、参数 Schema、调用返回结果格式都是标准化的。新系统接入时只需要实现一个 MCP ServerLangChain 侧就能动态加载工具。第三全链路可观测。一个 Agent 任务可能涉及一次模型调用、三次工具调用、两次状态变更。如果只有最终结果出了问题无法定位是模型判断错了、工具返回错了还是状态被覆盖了。可观测性的做法是要让每一步“可回放”。2. 技术底座与环境准备2.1 版本背景说明本文标题写的是 LangChain 1.0LangGraph 1.0主要是因为从 1.x 开始这两个项目从包名、API 到思维模型都有不少调整。由于不同时间安装的版本可能有差异文中的代码以常见安装方式为例重点演示设计思路不写死具体版本号。你安装时建议用以下命令拉取最新稳定版pip install -U langchain langgraph langchain-openai langchain-mcp-adapters mcp部分环境可能要把langchain和langgraph分开安装这在 1.x 版本中很正常两者已经是独立发布的包了。如果你要用本地模型可以替换langchain-openai为langchain-ollama或langchain-community里对应模型封装。2.2 推荐项目结构企业级项目不建议把所有代码堆在一个main.py里推荐下面这种结构agent_project/ ├── pyproject.toml ├── .env.example ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口接收 HTTP 请求 │ ├── agent/ │ │ ├── __init__.py │ │ ├── graph.py # LangGraph 状态图构建 │ │ ├── state.py # 状态定义 │ │ ├── nodes.py # 各节点业务逻辑 │ │ └── tools.py # 普通工具注册 │ ├── mcp/ │ │ ├── __init__.py │ │ ├── client.py # MCP Client 连接管理 │ │ └── servers.py # 本地 MCP Server 配置 │ ├── observer/ │ │ ├── __init__.py │ │ └── tracing.py # 日志与链路埋点 │ └── config.py # 配置读取 └── tests/ └── test_agent.py这种分离的好处是状态、节点、图、MCP、可观测性各司其职。后面加功能时不用在一个文件里反复打补丁。2.3 环境变量与密钥管理大模型 API Key、MCP Server 地址这些敏感信息不要写死在代码里。# .env.example OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 MCP_TICKET_SERVER_URLhttp://internal-mcp-server:9000在config.py里统一读取import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) MCP_TICKET_SERVER_URL os.getenv(MCP_TICKET_SERVER_URL, http://localhost:9000) MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini) MAX_RECURSION_LIMIT int(os.getenv(MAX_RECURSION_LIMIT, 10))注意生产环境不要用.env管理密钥建议接入 Vault、KMS 或云厂商的密钥管理服务。.env只适合本地开发。3. LangGraph 状态图核心建模从线性链到可控 Agent3.1 State、Node、Edge 的最小理解LangGraph 的核心抽象只有四个概念State整个 Agent 运行期间共享的数据结构。可以理解成“工作记忆”但要显式定义。Node一个普通的 Python 函数。输入是 State输出是 State 的增量更新。Edge从一个 Node 到另一个 Node 的连接。Conditional Edge根据 State 的值动态选择下一步去哪个 Node。先看一个最小例子from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] step_count: int def node_a(state: AgentState): return {messages: [A 执行完毕], step_count: state[step_count] 1} def node_b(state: AgentState): return {messages: [B 执行完毕], step_count: state[step_count] 1} graph StateGraph(AgentState) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) app graph.compile() result app.invoke({ messages: [], step_count: 0, }) print(result)在这个例子中Annotated[list, operator.add]表示每次节点返回的messages都会追加到原来的列表里而不是直接覆盖。这是 LangGraph 做多轮对话和工具结果累积的核心机制。3.2 用条件路由实现“模型决定下一步”真实 Agent 里模型可能选择调用工具也可能直接回复用户。这个分支判断用add_conditional_edges来实现def route_after_agent(state: AgentState): last_message state[messages][-1] # 判断模型返回的内容如果有 tool_calls就路由到工具节点 if getattr(last_message, tool_calls, None): return tools return END graph.add_conditional_edges(agent, route_after_agent, { tools: tools, END: END, })这里的关键是路由函数返回一个字符串第三个参数是一个映射表把这个字符串映射到实际的 Node 名。很多新手会问为什么不用 if-else 直接调函数因为在 LangGraph 里Node 之间的跳转是图结构的一部分只有通过 Edge 连接整个执行流程才是可追踪、可持久化、可回放的。如果你在 Node 内部直接调用另一个函数图就“断开”了。3.3 循环检测与终止条件Agent 最怕两类问题模型反复调用同一个工具陷入死循环。工具出错后不断重试无法跳出。LangGraph 提供了recursion_limit来限制整张图的执行步数config {recursion_limit: 10} result app.invoke({messages: []}, configconfig)如果超限会抛出类似GraphRecursionError的异常。企业级实践里我会在两步做防护在入口处设置recursion_limit全局兜底。在路由函数里记录step_count达到阈值后强制路由到“兜底回复节点”。def route_after_agent(state: AgentState): if state[step_count] MAX_RECURSION_LIMIT: return fallback if getattr(state[messages][-1], tool_calls, None): return tools return END这样 Agent 永远有一条“退出路径”不会让用户等一个永远跑不完的任务。3.4 子图与并行分支当业务复杂以后可以把一组节点封装成一个 Subgraph作为父图的一个 Node 使用。应用场景比较典型的是一个“数据查询子图”内部有“查库 → 判断是否需要查明细 → 返回结果”多个步骤。父图只需要知道“调用数据查询子图”然后根据结果决定下一步。并行分支则适用于“同时查多个数据源”的场景比如同时查库存、查物流、查价格。可以用SendAPI 实现动态并行。这两块属于进阶内容我建议你先掌握单层状态图等真正遇到流程嵌套再引入 Subgraph不要一开始就把图画复杂。4. MCP 接入外部系统一次接入处处可用4.1 先搞懂 MCP 是什么MCPModel Context Protocol是 Anthropic 在 2024 年底开源的一套“模型上下文协议”。它的作用是定义了大模型应用与外部工具、数据源之间的通信标准。在没有 MCP 之前Agent 接入一个外部系统要经历看对方 API 文档。写一个 Python/Java 函数封装 HTTP 请求。用tool装饰器装饰成 LangChain Tool。手动维护参数类型和描述。每接一个系统重复一遍。系统多了以后工具变成一座“屎山”。有了 MCP 之后外部系统只需要提供一个 MCP Server把能力暴露成标准工具。LangChain 侧通过 MCP Client 加载工具就能获得工具名。参数 Schema。工具描述。调用执行能力。这样 LangChain Agent 与具体系统之间不再强耦合。新系统接入不需要改 Agent 主流程只需要配置一个新的 MCP Server 地址然后动态加载即可。4.2 在 LangChain 中接入 MCP Server官方适配包是langchain-mcp-adapters。核心函数是load_mcp_tools。下面演示如何连接一个远程 MCP Serverfrom contextlib import asynccontextmanager from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.client.streamable_http import streamablehttp_client async def load_remote_mcp_tools(server_url: str): # 连接远程 MCP Server async with streamablehttp_client(server_url) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools如果你的 MCP Server 是本地子进程方式启动用stdio_clientasync def load_local_mcp_tools(command: str, args: list[str]): server_params StdioServerParameters( commandcommand, argsargs, envNone, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools注意load_mcp_tools是基于 async 的而 LangGraph 的普通 Node 多数是 sync 函数。你可以借助asyncio.run()在同步节点里加载工具或者在启动时预先加载好工具列表再注入图节点。import asyncio def get_mcp_tools_sync(server_url: str): return asyncio.run(load_remote_mcp_tools(server_url))4.3 自己写一个 MCP Server以工单系统为例为了演示完整闭环这里写一个最小 MCP Server。企业内部如果要把一个老系统接进来就是这个套路# 文件路径mcp_servers/ticket_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(ticket-service) mcp.tool() def create_ticket(title: str, content: str, requester: str) - str: 创建一条工单记录。 Args: title: 工单标题。 content: 工单详细描述。 requester: 提交人工号。 Returns: 工单编号。 # 这里是模拟实现真实场景会调用企业内部工单系统 API ticket_id fTICKET-{hash(title requester) % 10000:04d} return ticket_id mcp.tool() def query_ticket_status(ticket_id: str) - str: 查询工单状态。 Args: ticket_id: 工单编号。 Returns: 工单当前状态。 status_map {TICKET-0001: 处理中, TICKET-0002: 已完成} return status_map.get(ticket_id, 未知工单) if __name__ __main__: mcp.run(transportsse, host0.0.0.0, port9000, sse_path/sse)这个 Server 启动后LangChain 侧就可以通过streamablehttp_client或 SSE 地址加载它的工具。关于“MCP 工具注册不上”的问题我在后面常见问题里专门展开。4.4 MCP Tool 与 LangChain Tool 如何共存不是所有工具都适合走 MCP。以下情况直接用 LangChaintool更轻量纯内部函数比如密码校验、字符串处理。数据访问层已有的 Service 方法。与外部系统无关的本地逻辑。而以下情况建议走 MCP多个 Agent 共享一套工具能力。工具由不同团队维护接口需要统一治理。需要在运行时动态注册/卸载工具。工具调用需要独立的鉴权、限流、审计。实际项目中通常两种方式混合使用。LangGraph 节点里可以把普通工具和 MCP 工具合并到一个列表中一起绑定给模型all_tools local_tools mcp_tools5. 全链路可观测从 print 到 trace5.1 为什么 Agent 的可观测性比普通后端更难普通后端接口的可观测性往往只需要记录“入参、出参、耗时、错误”。但 Agent 是“多轮决策系统”一个问题可能触发多次模型调用和多次工具调用。有一个线上案例让我印象很深用户反馈“Agent 查不到库存数据”。从界面看最终回答是“暂无可售库存”看起来没有问题。但实际是模型第一次选择了“查询商品ID”工具但参数传错了查出一个空列表。模型第二次没有继续查而是直接根据空列表得出结论。整个链路毫无异常没有报错。如果只有结果日志这个问题根本无法定位。必须记录每一步的工具入参、工具出参、模型决策理由才能还原现场。5.2 给 Agent 加结构化日志不要在 Node 里随意print生产环境需要结构化 JSON 日志方便接入 ELK、Loki 或其他日志平台。import json import logging import time from datetime import datetime logger logging.getLogger(agent_trace) def log_node_enter(node_name: str, state: dict): logger.info(json.dumps({ event: node_enter, node: node_name, timestamp: datetime.utcnow().isoformat(), step_count: state.get(step_count), trace_id: state.get(trace_id), }, ensure_asciiFalse)) def log_tool_call(tool_name: str, tool_args: dict, result: str, duration_ms: int): logger.info(json.dumps({ event: tool_call, tool: tool_name, args: tool_args, result_preview: result[:200], duration_ms: duration_ms, timestamp: datetime.utcnow().isoformat(), }, ensure_asciiFalse))注意工具出参不要全量打日志截断到前 200 到 500 字符即可。否则大模型返回的长文本会把日志存储打爆。5.3 在 LangGraph 节点里嵌入链路追踪LangGraph 允许在节点函数内部通过RunnableConfig拿到运行时配置。可以在启动时往 config 里塞一个全局唯一的trace_id然后所有节点日志都携带这个 ID。from langchain_core.runnables import RunnableConfig def agent_node(state: AgentState, config: RunnableConfig): trace_id config.get(configurable, {}).get(trace_id, unknown) log_node_enter(agent, {**state, trace_id: trace_id}) # 组装 tools 和 messages # 调用大模型注意记录耗时 start time.time() response llm_with_tools.invoke(state[messages]) duration int((time.time() - start) * 1000) logger.info(json.dumps({ event: llm_call, trace_id: trace_id, duration_ms: duration, response_preview: response.content[:200] if response.content else , tool_calls: response.tool_calls, }, ensure_asciiFalse)) return {messages: [response], step_count: state[step_count] 1}通过trace_id就能把“用户请求 → 模型调用 → 工具调用 → 最终回复”串成一条完整链路。遇到问题时只需要按trace_id搜索日志。5.4 关于 LangSmith 和其他可观测平台如果团队条件允许可以直接接入 LangChain 官方推出的 LangSmith它提供了非常完善的 trace、评价、数据集管理能力。配置方式很简单LANGSMITH_TRACINGtrue LANGSMITH_API_KEYyour_api_key LANGSMITH_PROJECTyour_project_name不过如果你所在团队的网络环境或数据合规要求不允许使用外部 SaaS建议按照上面的思路自建结构化日志再配合 OpenTelemetry 将 trace 打到内部链路平台。自建方案的成本并不高核心是要在关键节点埋点。6. 完整实战构建一个带 MCP 工具与可观测性的企业级 Agent6.1 需求拆解用一个“内部工单助手”作为案例需求如下用户通过 HTTP 发起问题例如“刚才我提交的 TICKET-0001 现在什么状态如果还没完成请帮我催一下”。Agent 需要查询工单状态。如果状态是“处理中”则创建一个“催办工单”并返回。所有调用需要记录日志带trace_id能回溯到每次模型决策和工具调用。这个案例很能说明问题它不是简单的“问答”而是涉及条件判断、工具调用、再次调用模型总结还需要可观测。6.2 定义状态先定义完整的 AgentState# 文件路径app/agent/state.py from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] step_count: int trace_id: str注意messages用operator.add这样每个节点返回的messages都会追加不会覆盖。6.3 构建图# 文件路径app/agent/graph.py from langgraph.graph import StateGraph, START, END from app.agent.state import AgentState from app.agent.nodes import agent_node, tools_node, fallback_node def route_after_agent(state: AgentState): if state[step_count] MAX_RECURSION_LIMIT: return fallback last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return END def build_graph(llm, tools): graph StateGraph(AgentState) graph.add_node(agent, agent_node(llm, tools)) graph.add_node(tools, tools_node(tools)) graph.add_node(fallback, fallback_node) graph.add_edge(START, agent) graph.add_conditional_edges( agent, route_after_agent, { tools: tools, fallback: fallback, END: END, } ) graph.add_edge(tools, agent) graph.add_edge(fallback, END) return graph.compile()这里要注意tools_nodeLangGraph 有一个内置的ToolNode可以直接用。# 文件路径app/agent/nodes.py from langgraph.prebuilt import ToolNode from langchain_core.runnables import RunnableConfig import json import logging import time logger logging.getLogger(agent_trace) def agent_node(llm, tools): llm_with_tools llm.bind_tools(tools) def node(state: AgentState, config: RunnableConfig): trace_id config.get(configurable, {}).get(trace_id, unknown) logger.info(json.dumps({ event: node_enter, node: agent, trace_id: trace_id, step_count: state.get(step_count), }, ensure_asciiFalse)) start time.time() response llm_with_tools.invoke(state[messages]) duration_ms int((time.time() - start) * 1000) logger.info(json.dumps({ event: llm_call, trace_id: trace_id, duration_ms: duration_ms, tool_calls: response.tool_calls, response_preview: response.content[:200] if response.content else , }, ensure_asciiFalse)) return {messages: [response], step_count: state[step_count] 1} return node def tools_node(tools): tool_node ToolNode(tools) def node(state: AgentState, config: RunnableConfig): trace_id config.get(configurable, {}).get(trace_id, unknown) start time.time() result tool_node.invoke(state, config) duration_ms int((time.time() - start) * 1000) logger.info(json.dumps({ event: tools_node, trace_id: trace_id, duration_ms: duration_ms, result_preview: str(result)[:300], }, ensure_asciiFalse)) return {messages: result} return node def fallback_node(state: AgentState): return { messages: [{ role: assistant, content: 抱歉当前任务复杂度超过限制请稍后重试或转人工处理。 }] }代码里用ToolNode(tools)来执行工具它内部会解析模型返回的tool_calls逐个执行并返回 ToolMessage。6.4 装配 MCP 工具与本地工具在实际接入中把 MCP 工具加载出来再和本地工具合并# 文件路径app/agent/create_agent.py import asyncio from langchain_openai import ChatOpenAI from app.agent.graph import build_graph from app.agent.tools import local_tools from app.mcp.client import load_remote_mcp_tools def create_agent(): llm ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyOPENAI_API_KEY, base_urlOPENAI_BASE_URL, ) # 加载 MCP 工具 mcp_tools asyncio.run(load_remote_mcp_tools(MCP_TICKET_SERVER_URL)) all_tools local_tools mcp_tools return build_graph(llm, all_tools)注意生产环境不建议在启动时asyncio.run()加载工具更推荐在应用启动生命周期里一次性加载并缓存避免每个请求都重复建立 MCP 连接那是巨大的性能浪费。6.5 FastAPI 入口与调用验证# 文件路径app/main.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from uuid import uuid4 from langchain_core.runnables import RunnableConfig from app.agent.create_agent import create_agent app FastAPI() # 应用启动时构建 Agent并缓存在全局 app.on_event(startup) async def startup(): app.state.agent create_agent() app.post(/chat) async def chat(request: Request): payload await request.json() user_message payload.get(message) trace_id payload.get(trace_id) or str(uuid4()) config RunnableConfig( configurable{trace_id: trace_id}, recursion_limit10, ) result await app.state.agent.ainvoke( { messages: [{role: user, content: user_message}], step_count: 0, trace_id: trace_id, }, configconfig, ) final_answer result[messages][-1].content return JSONResponse({ trace_id: trace_id, answer: final_answer, steps: result[step_count], })启动服务uvicorn app.main:app --reload --port 8000调用接口curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {message: 帮我查一下 TICKET-0001 的状态如果还没完成就催一下}实际运行中你会看到日志里出现了完整的链路agent 节点进入、LLM 调用、tools 节点执行、再次 LLM 调用总结。有了 trace_id整条链路都是可回溯的。6.6 运行结果说明预期行为是这样一条决策链模型判断需要查工单状态调用query_ticket_status。ToolNode执行 MCP 工具返回“处理中”。模型再次判断因为状态是“处理中”决定创建催办工单调用create_ticket。ToolNode执行完成返回工单编号。模型总结“TICKET-0001 仍在处理中我已帮你创建催办工单 TICKET-xxxx。”整个过程没有报错但如果最终答案不是用户想要的你可以通过 trace_id 查看是哪一环模型判断出错还是工具返回了错误结果。这就是全链路可观测的价值。7. 常见问题与排查思路7.1 高频问题排查表问题现象常见原因解决思路Agent 无限循环不结束没有设置 recursion_limit或路由条件一直返回 tools设置recursion_limit在路由函数中加入 step_count 判断工具执行超时外部系统响应慢或 MCP Server 连接未复用对工具调用增加超时控制MCP 连接使用连接池MCP 工具加载为空MCP Server 未启动、地址错误、鉴权失败先用mcp.__main__或直接调用 Server 接口验证模型不调用工具工具描述不清晰或模型版本能力不足优化工具 description明确“什么时候该调用”State 里消息被覆盖没在 TypedDict 中用Annotated[list, operator.add]检查 State 定义确保累积字段使用 reducerLangChain 和 LangGraph 版本不匹配包版本差异导致 API 变更固定版本用pip freeze锁定依赖7.2 “MCP 工具注册不上”的排查思路这是很多开发者遇到的高频问题。现象是MCP Server 正常启动但 LangChain 加载出来的工具列表是空的或者 Agent 始终不调用 MCP 工具。按以下顺序排查验证 MCP Server 本身是否有问题。直接用 MCP Inspector 或写一个最小 Client 连接先确认工具能拉取到。检查 transport 是否匹配。Server 是 SSEClient 就不要用 stdio反之亦然。检查 Session 是否正确初始化。必须调用await session.initialize()否则拉取不到工具列表。检查工具描述是否规范。有些模型对空描述的工具会“选择不调用”尽量给每个工具写清楚触发场景。检查鉴权。如果 MCP Server 有鉴权Client 必须在 header 中带上 token。检查加载时机。不要在每次请求时重新加载全部工具启动时加载一次并缓存避免连接泄漏。7.3 “The agent execution provider did not respond in time” 类超时问题这类报错的核心是Agent 执行链路中某一步没有在预期时间内返回通常是模型调用超时或工具调用超时。处理方式先看日志里最后一步是llm_call还是tools_node缩小范围。如果是模型调用超时考虑换模型、降输入长度、或者给 LLM 客户端设置更长 timeout。如果是工具调用超时优先检查 MCP Server 的响应时间。llm ChatOpenAI( modelgpt-4o-mini, temperature0, timeout60, # 单位秒按实际情况调整 )在设计上外部工具调用建议设置“软超时”和“降级方案”。比如工单系统查询超过 5 秒直接返回“系统繁忙请稍后重试”而不是让整个 Agent 挂起。8. 工程化最佳实践8.1 状态设计只保留必要字段AgentState 不是垃圾桶。不要让所有临时变量都塞进 State因为 State 会随着执行过程不断传递字段越多模型上下文被污染的风险越大。我的习惯是运行时临时量放到节点内部局部变量。跨节点共享的关键量消息、步数、业务对象 ID才放 State。敏感数据密钥、token永远不进 State。8.2 安全边界与权限控制企业级 Agent 必须把“模型能做什么”和“模型应该做什么”分开。工具层做白名单不是一个模型能调用所有工具而是根据用户角色动态注入可用工具列表。节点层做鉴权涉及“创建工单”“修改数据”等敏感操作时节点函数内部先校验用户是否具有权限再执行。敏感操作加审批在 LangGraph 中可以设计一个approval节点必要时暂停等待人工确认。LangGraph 对人工介入human-in-the-loop有专门支持核心是interrupt机制可以在执行到某个节点时暂停图等人工确认后继续。这在生产环境里是做“半自动”Agent 的关键能力强烈建议深入学习。8.3 性能与资源控制对每个请求设置超时时间避免 Agent 无限等待。MCP 连接要复用不要在每次请求里重新建连。大量工具时可以按照工具分组不要让模型每次看到全部工具降低上下文长度。日志要设置采样率重点链路全量记录一般链路按比例采样。8.4 测试与回滚Agent 不是传统代码不能只靠单测。建议建立“回归评测集”准备 50 到 200 条典型用户问题覆盖主要业务分支。每条问题标注期望行为调用什么工具、最终答案方向。每次修改 Prompt、工具、模型版本后跑一遍评测集。对比通过率通过后再发布。这样能最大限度避免“改了 A 场景破坏了 B 场景”的回归问题。8.5 从 Demo 到生产的演进清单如果你正在把 Demo Agent 往生产推可以按这个顺序检查是否设置了recursion_limit是否有兜底回复节点工具是否做了权限校验是否记录 trace_id 和关键节点日志MCP 连接是否复用是否有回归评测集模型 API Key 和 MCP Server 密钥是否在密钥管理系统里每一项目前都可以用很小的工作量补齐但缺了任何一项线上都可能出事故。9. 总结与下一步学习路线这篇文章的核心是把 Agent 从“Demo 思维”拉回到“工程思维”。你掌握了三块核心能力用 LangGraph 状态图控制 AgentState、Node、Conditional Edge、循环控制、Subgraph 建模。用 MCP 标准化外部工具接入MCP Server 实现、MCP Client 加载、LangChain Tool 共存策略。用结构化日志实现全链路可观测trace_id 贯穿节点、LLM 调用、工具调用支持线上问题回溯。下一步建议按这个顺序继续深入先把本文的代码复制到本地跑通一个最小 Agent。然后尝试加上interrupt人工审批节点把所有敏感写操作包一层人工确认。接着把 LangSmith 或自建 trace 平台接上用 trace_id 排查一次真实问题。最后着手搭建回归评测集把 Agent 的 Prompt 和工具调整从“拍脑袋”变成“数据驱动”。如果本文对你有帮助可以收藏备用。后面我还会继续拆 LangGraph 的条件路由深度变体、子图复用、Checkpointer 持久化以及 MCP Server 在生产环境的高可用部署方案可以关注后续更新。
LangGraph+MCP构建企业级Agent：状态图与全链路可观测实践
很多团队做了半年 Agent最后落地的 Demo 还是“一个 while 循环 一个 OpenAI 函数调用”。代码能跑但一碰到线上流量、权限边界、外部系统接入、问题回溯立刻捉襟见肘。本文不聊概念直接给你一套 LangChain 1.0LangGraph 1.0 的企业级 Agent 搭建思路覆盖状态图设计、条件路由、MCP 工具接入、全链路可观测和常见生产问题附完整可运行代码。之前我在业务迭代中被两件事逼疯一是 Agent 行为不可控模型一旦多轮调用就容易绕圈、丢状态二是接外部系统时每个系统都要自研一套“函数调用协议”接口层次不齐联调效率极低。后来把 Agent 的编排层从 LangChain 原生链式调用切换到了 LangGraph 状态图再用 MCPModel Context Protocol统一外部工具接入最后在关键路径上补了结构化日志和链路追踪才真正感受到“工程化 Agent”和“Demo Agent”的区别。这篇文章适合三类读者已经会用 LangChain 写简单工具调用但想进阶到带状态、带分支、带人工审批的 Agent。后端开发想把 Agent 接入企业内部工单、数据库、RAG 服务但对 MCP 协议还不熟悉。正在做 Agent 生产落地的技术负责人想找一份可观测、有审计、能出问题的排查清单。读完你会掌握LangChain 和 LangGraph 的分工边界、LangGraph 状态图核心建模方法、MCP 在 LangChain 生态里的接入方式、以及一套从“记录日志”升级到“全链路可观测”的落地路径。1. 为什么企业级 Agent 不能再手搓 Demo1.1 LangChain 并没有过时而是“拆得越来越清楚”很多开发者看到 LangGraph 之后会问一句话LangChain 是不是要被 LangGraph 替代了我的结论LangChain 没有过时它和 LangGraph 是不同抽象层级的组件。LangChain 更像一个组件库大模型统一封装、Prompt 模板、向量库集成、各类文档加载器、输出解析器。LangGraph 是一个编排引擎负责 Agent 的流程控制、状态管理、分支路由、循环终止、持久化和人工介入。简单理解LangChain 负责“怎么跟模型和外部资源打交道”LangGraph 负责“整个任务怎么一步步跑完”。两者是配合关系不是替代关系。从 1.0 版本之后LangChain 官方也明显把重心往 LangGraph 这个“Agent 运行时”上移。你写的业务逻辑应该尽量是“状态图里的节点”而不是一条线性的 Chain。这样后续加审批、加重试、加多分支改动成本会低很多。1.2 Demo Agent 与企业级 Agent 的核心差距手搓 Demo 的时候Agent 通常长这样用户提问。把问题、工具列表、历史记录一股脑塞给大模型。大模型返回一个工具调用。代码里exec()或者 if-else 执行函数。把结果拼回去再调用一次大模型。循环直到模型说“完成”。这套流程在小范围验证时没问题但企业级场景下会暴露五个短板短板Demo 表现企业级要求状态管理所有状态存在一个 dict 里覆盖即丢失明确 State 结构支持多轮累积流程控制while 循环难以精确控制退出图结构节点、边、条件路由清晰人工介入无法暂停、审批、回退human-in-the-loop执行到审批节点暂停工具协议每个系统一套 SDK难维护MCP 统一工具协议动态加载可观测性print 日志出问题靠猜全链路 trace关键节点可回溯1.3 企业级 Agent 的三个核心维度结合我自己的工程经验下面三个维度是判断一个 Agent 能不能上生产的底线第一安全可控。Agent 不能是一个“模型自由发挥的黑盒”。要有最大步数限制要有工具白名单要有数据脱敏要有敏感操作审批。LangGraph 的状态图天然适合做这些每个节点都是一个函数函数内部可以做权限校验每条边都可以加条件条件不满足就走进度分支。第二标准化接入。内部系统千奇百怪如果每个系统都单独开发函数调用接口Agent 的工具层会很快腐化。MCP 的价值在于它给“外部工具”定了一套统一协议工具描述、参数 Schema、调用返回结果格式都是标准化的。新系统接入时只需要实现一个 MCP ServerLangChain 侧就能动态加载工具。第三全链路可观测。一个 Agent 任务可能涉及一次模型调用、三次工具调用、两次状态变更。如果只有最终结果出了问题无法定位是模型判断错了、工具返回错了还是状态被覆盖了。可观测性的做法是要让每一步“可回放”。2. 技术底座与环境准备2.1 版本背景说明本文标题写的是 LangChain 1.0LangGraph 1.0主要是因为从 1.x 开始这两个项目从包名、API 到思维模型都有不少调整。由于不同时间安装的版本可能有差异文中的代码以常见安装方式为例重点演示设计思路不写死具体版本号。你安装时建议用以下命令拉取最新稳定版pip install -U langchain langgraph langchain-openai langchain-mcp-adapters mcp部分环境可能要把langchain和langgraph分开安装这在 1.x 版本中很正常两者已经是独立发布的包了。如果你要用本地模型可以替换langchain-openai为langchain-ollama或langchain-community里对应模型封装。2.2 推荐项目结构企业级项目不建议把所有代码堆在一个main.py里推荐下面这种结构agent_project/ ├── pyproject.toml ├── .env.example ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口接收 HTTP 请求 │ ├── agent/ │ │ ├── __init__.py │ │ ├── graph.py # LangGraph 状态图构建 │ │ ├── state.py # 状态定义 │ │ ├── nodes.py # 各节点业务逻辑 │ │ └── tools.py # 普通工具注册 │ ├── mcp/ │ │ ├── __init__.py │ │ ├── client.py # MCP Client 连接管理 │ │ └── servers.py # 本地 MCP Server 配置 │ ├── observer/ │ │ ├── __init__.py │ │ └── tracing.py # 日志与链路埋点 │ └── config.py # 配置读取 └── tests/ └── test_agent.py这种分离的好处是状态、节点、图、MCP、可观测性各司其职。后面加功能时不用在一个文件里反复打补丁。2.3 环境变量与密钥管理大模型 API Key、MCP Server 地址这些敏感信息不要写死在代码里。# .env.example OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 MCP_TICKET_SERVER_URLhttp://internal-mcp-server:9000在config.py里统一读取import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) MCP_TICKET_SERVER_URL os.getenv(MCP_TICKET_SERVER_URL, http://localhost:9000) MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini) MAX_RECURSION_LIMIT int(os.getenv(MAX_RECURSION_LIMIT, 10))注意生产环境不要用.env管理密钥建议接入 Vault、KMS 或云厂商的密钥管理服务。.env只适合本地开发。3. LangGraph 状态图核心建模从线性链到可控 Agent3.1 State、Node、Edge 的最小理解LangGraph 的核心抽象只有四个概念State整个 Agent 运行期间共享的数据结构。可以理解成“工作记忆”但要显式定义。Node一个普通的 Python 函数。输入是 State输出是 State 的增量更新。Edge从一个 Node 到另一个 Node 的连接。Conditional Edge根据 State 的值动态选择下一步去哪个 Node。先看一个最小例子from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] step_count: int def node_a(state: AgentState): return {messages: [A 执行完毕], step_count: state[step_count] 1} def node_b(state: AgentState): return {messages: [B 执行完毕], step_count: state[step_count] 1} graph StateGraph(AgentState) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) app graph.compile() result app.invoke({ messages: [], step_count: 0, }) print(result)在这个例子中Annotated[list, operator.add]表示每次节点返回的messages都会追加到原来的列表里而不是直接覆盖。这是 LangGraph 做多轮对话和工具结果累积的核心机制。3.2 用条件路由实现“模型决定下一步”真实 Agent 里模型可能选择调用工具也可能直接回复用户。这个分支判断用add_conditional_edges来实现def route_after_agent(state: AgentState): last_message state[messages][-1] # 判断模型返回的内容如果有 tool_calls就路由到工具节点 if getattr(last_message, tool_calls, None): return tools return END graph.add_conditional_edges(agent, route_after_agent, { tools: tools, END: END, })这里的关键是路由函数返回一个字符串第三个参数是一个映射表把这个字符串映射到实际的 Node 名。很多新手会问为什么不用 if-else 直接调函数因为在 LangGraph 里Node 之间的跳转是图结构的一部分只有通过 Edge 连接整个执行流程才是可追踪、可持久化、可回放的。如果你在 Node 内部直接调用另一个函数图就“断开”了。3.3 循环检测与终止条件Agent 最怕两类问题模型反复调用同一个工具陷入死循环。工具出错后不断重试无法跳出。LangGraph 提供了recursion_limit来限制整张图的执行步数config {recursion_limit: 10} result app.invoke({messages: []}, configconfig)如果超限会抛出类似GraphRecursionError的异常。企业级实践里我会在两步做防护在入口处设置recursion_limit全局兜底。在路由函数里记录step_count达到阈值后强制路由到“兜底回复节点”。def route_after_agent(state: AgentState): if state[step_count] MAX_RECURSION_LIMIT: return fallback if getattr(state[messages][-1], tool_calls, None): return tools return END这样 Agent 永远有一条“退出路径”不会让用户等一个永远跑不完的任务。3.4 子图与并行分支当业务复杂以后可以把一组节点封装成一个 Subgraph作为父图的一个 Node 使用。应用场景比较典型的是一个“数据查询子图”内部有“查库 → 判断是否需要查明细 → 返回结果”多个步骤。父图只需要知道“调用数据查询子图”然后根据结果决定下一步。并行分支则适用于“同时查多个数据源”的场景比如同时查库存、查物流、查价格。可以用SendAPI 实现动态并行。这两块属于进阶内容我建议你先掌握单层状态图等真正遇到流程嵌套再引入 Subgraph不要一开始就把图画复杂。4. MCP 接入外部系统一次接入处处可用4.1 先搞懂 MCP 是什么MCPModel Context Protocol是 Anthropic 在 2024 年底开源的一套“模型上下文协议”。它的作用是定义了大模型应用与外部工具、数据源之间的通信标准。在没有 MCP 之前Agent 接入一个外部系统要经历看对方 API 文档。写一个 Python/Java 函数封装 HTTP 请求。用tool装饰器装饰成 LangChain Tool。手动维护参数类型和描述。每接一个系统重复一遍。系统多了以后工具变成一座“屎山”。有了 MCP 之后外部系统只需要提供一个 MCP Server把能力暴露成标准工具。LangChain 侧通过 MCP Client 加载工具就能获得工具名。参数 Schema。工具描述。调用执行能力。这样 LangChain Agent 与具体系统之间不再强耦合。新系统接入不需要改 Agent 主流程只需要配置一个新的 MCP Server 地址然后动态加载即可。4.2 在 LangChain 中接入 MCP Server官方适配包是langchain-mcp-adapters。核心函数是load_mcp_tools。下面演示如何连接一个远程 MCP Serverfrom contextlib import asynccontextmanager from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.client.streamable_http import streamablehttp_client async def load_remote_mcp_tools(server_url: str): # 连接远程 MCP Server async with streamablehttp_client(server_url) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools如果你的 MCP Server 是本地子进程方式启动用stdio_clientasync def load_local_mcp_tools(command: str, args: list[str]): server_params StdioServerParameters( commandcommand, argsargs, envNone, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools注意load_mcp_tools是基于 async 的而 LangGraph 的普通 Node 多数是 sync 函数。你可以借助asyncio.run()在同步节点里加载工具或者在启动时预先加载好工具列表再注入图节点。import asyncio def get_mcp_tools_sync(server_url: str): return asyncio.run(load_remote_mcp_tools(server_url))4.3 自己写一个 MCP Server以工单系统为例为了演示完整闭环这里写一个最小 MCP Server。企业内部如果要把一个老系统接进来就是这个套路# 文件路径mcp_servers/ticket_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(ticket-service) mcp.tool() def create_ticket(title: str, content: str, requester: str) - str: 创建一条工单记录。 Args: title: 工单标题。 content: 工单详细描述。 requester: 提交人工号。 Returns: 工单编号。 # 这里是模拟实现真实场景会调用企业内部工单系统 API ticket_id fTICKET-{hash(title requester) % 10000:04d} return ticket_id mcp.tool() def query_ticket_status(ticket_id: str) - str: 查询工单状态。 Args: ticket_id: 工单编号。 Returns: 工单当前状态。 status_map {TICKET-0001: 处理中, TICKET-0002: 已完成} return status_map.get(ticket_id, 未知工单) if __name__ __main__: mcp.run(transportsse, host0.0.0.0, port9000, sse_path/sse)这个 Server 启动后LangChain 侧就可以通过streamablehttp_client或 SSE 地址加载它的工具。关于“MCP 工具注册不上”的问题我在后面常见问题里专门展开。4.4 MCP Tool 与 LangChain Tool 如何共存不是所有工具都适合走 MCP。以下情况直接用 LangChaintool更轻量纯内部函数比如密码校验、字符串处理。数据访问层已有的 Service 方法。与外部系统无关的本地逻辑。而以下情况建议走 MCP多个 Agent 共享一套工具能力。工具由不同团队维护接口需要统一治理。需要在运行时动态注册/卸载工具。工具调用需要独立的鉴权、限流、审计。实际项目中通常两种方式混合使用。LangGraph 节点里可以把普通工具和 MCP 工具合并到一个列表中一起绑定给模型all_tools local_tools mcp_tools5. 全链路可观测从 print 到 trace5.1 为什么 Agent 的可观测性比普通后端更难普通后端接口的可观测性往往只需要记录“入参、出参、耗时、错误”。但 Agent 是“多轮决策系统”一个问题可能触发多次模型调用和多次工具调用。有一个线上案例让我印象很深用户反馈“Agent 查不到库存数据”。从界面看最终回答是“暂无可售库存”看起来没有问题。但实际是模型第一次选择了“查询商品ID”工具但参数传错了查出一个空列表。模型第二次没有继续查而是直接根据空列表得出结论。整个链路毫无异常没有报错。如果只有结果日志这个问题根本无法定位。必须记录每一步的工具入参、工具出参、模型决策理由才能还原现场。5.2 给 Agent 加结构化日志不要在 Node 里随意print生产环境需要结构化 JSON 日志方便接入 ELK、Loki 或其他日志平台。import json import logging import time from datetime import datetime logger logging.getLogger(agent_trace) def log_node_enter(node_name: str, state: dict): logger.info(json.dumps({ event: node_enter, node: node_name, timestamp: datetime.utcnow().isoformat(), step_count: state.get(step_count), trace_id: state.get(trace_id), }, ensure_asciiFalse)) def log_tool_call(tool_name: str, tool_args: dict, result: str, duration_ms: int): logger.info(json.dumps({ event: tool_call, tool: tool_name, args: tool_args, result_preview: result[:200], duration_ms: duration_ms, timestamp: datetime.utcnow().isoformat(), }, ensure_asciiFalse))注意工具出参不要全量打日志截断到前 200 到 500 字符即可。否则大模型返回的长文本会把日志存储打爆。5.3 在 LangGraph 节点里嵌入链路追踪LangGraph 允许在节点函数内部通过RunnableConfig拿到运行时配置。可以在启动时往 config 里塞一个全局唯一的trace_id然后所有节点日志都携带这个 ID。from langchain_core.runnables import RunnableConfig def agent_node(state: AgentState, config: RunnableConfig): trace_id config.get(configurable, {}).get(trace_id, unknown) log_node_enter(agent, {**state, trace_id: trace_id}) # 组装 tools 和 messages # 调用大模型注意记录耗时 start time.time() response llm_with_tools.invoke(state[messages]) duration int((time.time() - start) * 1000) logger.info(json.dumps({ event: llm_call, trace_id: trace_id, duration_ms: duration, response_preview: response.content[:200] if response.content else , tool_calls: response.tool_calls, }, ensure_asciiFalse)) return {messages: [response], step_count: state[step_count] 1}通过trace_id就能把“用户请求 → 模型调用 → 工具调用 → 最终回复”串成一条完整链路。遇到问题时只需要按trace_id搜索日志。5.4 关于 LangSmith 和其他可观测平台如果团队条件允许可以直接接入 LangChain 官方推出的 LangSmith它提供了非常完善的 trace、评价、数据集管理能力。配置方式很简单LANGSMITH_TRACINGtrue LANGSMITH_API_KEYyour_api_key LANGSMITH_PROJECTyour_project_name不过如果你所在团队的网络环境或数据合规要求不允许使用外部 SaaS建议按照上面的思路自建结构化日志再配合 OpenTelemetry 将 trace 打到内部链路平台。自建方案的成本并不高核心是要在关键节点埋点。6. 完整实战构建一个带 MCP 工具与可观测性的企业级 Agent6.1 需求拆解用一个“内部工单助手”作为案例需求如下用户通过 HTTP 发起问题例如“刚才我提交的 TICKET-0001 现在什么状态如果还没完成请帮我催一下”。Agent 需要查询工单状态。如果状态是“处理中”则创建一个“催办工单”并返回。所有调用需要记录日志带trace_id能回溯到每次模型决策和工具调用。这个案例很能说明问题它不是简单的“问答”而是涉及条件判断、工具调用、再次调用模型总结还需要可观测。6.2 定义状态先定义完整的 AgentState# 文件路径app/agent/state.py from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] step_count: int trace_id: str注意messages用operator.add这样每个节点返回的messages都会追加不会覆盖。6.3 构建图# 文件路径app/agent/graph.py from langgraph.graph import StateGraph, START, END from app.agent.state import AgentState from app.agent.nodes import agent_node, tools_node, fallback_node def route_after_agent(state: AgentState): if state[step_count] MAX_RECURSION_LIMIT: return fallback last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return END def build_graph(llm, tools): graph StateGraph(AgentState) graph.add_node(agent, agent_node(llm, tools)) graph.add_node(tools, tools_node(tools)) graph.add_node(fallback, fallback_node) graph.add_edge(START, agent) graph.add_conditional_edges( agent, route_after_agent, { tools: tools, fallback: fallback, END: END, } ) graph.add_edge(tools, agent) graph.add_edge(fallback, END) return graph.compile()这里要注意tools_nodeLangGraph 有一个内置的ToolNode可以直接用。# 文件路径app/agent/nodes.py from langgraph.prebuilt import ToolNode from langchain_core.runnables import RunnableConfig import json import logging import time logger logging.getLogger(agent_trace) def agent_node(llm, tools): llm_with_tools llm.bind_tools(tools) def node(state: AgentState, config: RunnableConfig): trace_id config.get(configurable, {}).get(trace_id, unknown) logger.info(json.dumps({ event: node_enter, node: agent, trace_id: trace_id, step_count: state.get(step_count), }, ensure_asciiFalse)) start time.time() response llm_with_tools.invoke(state[messages]) duration_ms int((time.time() - start) * 1000) logger.info(json.dumps({ event: llm_call, trace_id: trace_id, duration_ms: duration_ms, tool_calls: response.tool_calls, response_preview: response.content[:200] if response.content else , }, ensure_asciiFalse)) return {messages: [response], step_count: state[step_count] 1} return node def tools_node(tools): tool_node ToolNode(tools) def node(state: AgentState, config: RunnableConfig): trace_id config.get(configurable, {}).get(trace_id, unknown) start time.time() result tool_node.invoke(state, config) duration_ms int((time.time() - start) * 1000) logger.info(json.dumps({ event: tools_node, trace_id: trace_id, duration_ms: duration_ms, result_preview: str(result)[:300], }, ensure_asciiFalse)) return {messages: result} return node def fallback_node(state: AgentState): return { messages: [{ role: assistant, content: 抱歉当前任务复杂度超过限制请稍后重试或转人工处理。 }] }代码里用ToolNode(tools)来执行工具它内部会解析模型返回的tool_calls逐个执行并返回 ToolMessage。6.4 装配 MCP 工具与本地工具在实际接入中把 MCP 工具加载出来再和本地工具合并# 文件路径app/agent/create_agent.py import asyncio from langchain_openai import ChatOpenAI from app.agent.graph import build_graph from app.agent.tools import local_tools from app.mcp.client import load_remote_mcp_tools def create_agent(): llm ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyOPENAI_API_KEY, base_urlOPENAI_BASE_URL, ) # 加载 MCP 工具 mcp_tools asyncio.run(load_remote_mcp_tools(MCP_TICKET_SERVER_URL)) all_tools local_tools mcp_tools return build_graph(llm, all_tools)注意生产环境不建议在启动时asyncio.run()加载工具更推荐在应用启动生命周期里一次性加载并缓存避免每个请求都重复建立 MCP 连接那是巨大的性能浪费。6.5 FastAPI 入口与调用验证# 文件路径app/main.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from uuid import uuid4 from langchain_core.runnables import RunnableConfig from app.agent.create_agent import create_agent app FastAPI() # 应用启动时构建 Agent并缓存在全局 app.on_event(startup) async def startup(): app.state.agent create_agent() app.post(/chat) async def chat(request: Request): payload await request.json() user_message payload.get(message) trace_id payload.get(trace_id) or str(uuid4()) config RunnableConfig( configurable{trace_id: trace_id}, recursion_limit10, ) result await app.state.agent.ainvoke( { messages: [{role: user, content: user_message}], step_count: 0, trace_id: trace_id, }, configconfig, ) final_answer result[messages][-1].content return JSONResponse({ trace_id: trace_id, answer: final_answer, steps: result[step_count], })启动服务uvicorn app.main:app --reload --port 8000调用接口curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {message: 帮我查一下 TICKET-0001 的状态如果还没完成就催一下}实际运行中你会看到日志里出现了完整的链路agent 节点进入、LLM 调用、tools 节点执行、再次 LLM 调用总结。有了 trace_id整条链路都是可回溯的。6.6 运行结果说明预期行为是这样一条决策链模型判断需要查工单状态调用query_ticket_status。ToolNode执行 MCP 工具返回“处理中”。模型再次判断因为状态是“处理中”决定创建催办工单调用create_ticket。ToolNode执行完成返回工单编号。模型总结“TICKET-0001 仍在处理中我已帮你创建催办工单 TICKET-xxxx。”整个过程没有报错但如果最终答案不是用户想要的你可以通过 trace_id 查看是哪一环模型判断出错还是工具返回了错误结果。这就是全链路可观测的价值。7. 常见问题与排查思路7.1 高频问题排查表问题现象常见原因解决思路Agent 无限循环不结束没有设置 recursion_limit或路由条件一直返回 tools设置recursion_limit在路由函数中加入 step_count 判断工具执行超时外部系统响应慢或 MCP Server 连接未复用对工具调用增加超时控制MCP 连接使用连接池MCP 工具加载为空MCP Server 未启动、地址错误、鉴权失败先用mcp.__main__或直接调用 Server 接口验证模型不调用工具工具描述不清晰或模型版本能力不足优化工具 description明确“什么时候该调用”State 里消息被覆盖没在 TypedDict 中用Annotated[list, operator.add]检查 State 定义确保累积字段使用 reducerLangChain 和 LangGraph 版本不匹配包版本差异导致 API 变更固定版本用pip freeze锁定依赖7.2 “MCP 工具注册不上”的排查思路这是很多开发者遇到的高频问题。现象是MCP Server 正常启动但 LangChain 加载出来的工具列表是空的或者 Agent 始终不调用 MCP 工具。按以下顺序排查验证 MCP Server 本身是否有问题。直接用 MCP Inspector 或写一个最小 Client 连接先确认工具能拉取到。检查 transport 是否匹配。Server 是 SSEClient 就不要用 stdio反之亦然。检查 Session 是否正确初始化。必须调用await session.initialize()否则拉取不到工具列表。检查工具描述是否规范。有些模型对空描述的工具会“选择不调用”尽量给每个工具写清楚触发场景。检查鉴权。如果 MCP Server 有鉴权Client 必须在 header 中带上 token。检查加载时机。不要在每次请求时重新加载全部工具启动时加载一次并缓存避免连接泄漏。7.3 “The agent execution provider did not respond in time” 类超时问题这类报错的核心是Agent 执行链路中某一步没有在预期时间内返回通常是模型调用超时或工具调用超时。处理方式先看日志里最后一步是llm_call还是tools_node缩小范围。如果是模型调用超时考虑换模型、降输入长度、或者给 LLM 客户端设置更长 timeout。如果是工具调用超时优先检查 MCP Server 的响应时间。llm ChatOpenAI( modelgpt-4o-mini, temperature0, timeout60, # 单位秒按实际情况调整 )在设计上外部工具调用建议设置“软超时”和“降级方案”。比如工单系统查询超过 5 秒直接返回“系统繁忙请稍后重试”而不是让整个 Agent 挂起。8. 工程化最佳实践8.1 状态设计只保留必要字段AgentState 不是垃圾桶。不要让所有临时变量都塞进 State因为 State 会随着执行过程不断传递字段越多模型上下文被污染的风险越大。我的习惯是运行时临时量放到节点内部局部变量。跨节点共享的关键量消息、步数、业务对象 ID才放 State。敏感数据密钥、token永远不进 State。8.2 安全边界与权限控制企业级 Agent 必须把“模型能做什么”和“模型应该做什么”分开。工具层做白名单不是一个模型能调用所有工具而是根据用户角色动态注入可用工具列表。节点层做鉴权涉及“创建工单”“修改数据”等敏感操作时节点函数内部先校验用户是否具有权限再执行。敏感操作加审批在 LangGraph 中可以设计一个approval节点必要时暂停等待人工确认。LangGraph 对人工介入human-in-the-loop有专门支持核心是interrupt机制可以在执行到某个节点时暂停图等人工确认后继续。这在生产环境里是做“半自动”Agent 的关键能力强烈建议深入学习。8.3 性能与资源控制对每个请求设置超时时间避免 Agent 无限等待。MCP 连接要复用不要在每次请求里重新建连。大量工具时可以按照工具分组不要让模型每次看到全部工具降低上下文长度。日志要设置采样率重点链路全量记录一般链路按比例采样。8.4 测试与回滚Agent 不是传统代码不能只靠单测。建议建立“回归评测集”准备 50 到 200 条典型用户问题覆盖主要业务分支。每条问题标注期望行为调用什么工具、最终答案方向。每次修改 Prompt、工具、模型版本后跑一遍评测集。对比通过率通过后再发布。这样能最大限度避免“改了 A 场景破坏了 B 场景”的回归问题。8.5 从 Demo 到生产的演进清单如果你正在把 Demo Agent 往生产推可以按这个顺序检查是否设置了recursion_limit是否有兜底回复节点工具是否做了权限校验是否记录 trace_id 和关键节点日志MCP 连接是否复用是否有回归评测集模型 API Key 和 MCP Server 密钥是否在密钥管理系统里每一项目前都可以用很小的工作量补齐但缺了任何一项线上都可能出事故。9. 总结与下一步学习路线这篇文章的核心是把 Agent 从“Demo 思维”拉回到“工程思维”。你掌握了三块核心能力用 LangGraph 状态图控制 AgentState、Node、Conditional Edge、循环控制、Subgraph 建模。用 MCP 标准化外部工具接入MCP Server 实现、MCP Client 加载、LangChain Tool 共存策略。用结构化日志实现全链路可观测trace_id 贯穿节点、LLM 调用、工具调用支持线上问题回溯。下一步建议按这个顺序继续深入先把本文的代码复制到本地跑通一个最小 Agent。然后尝试加上interrupt人工审批节点把所有敏感写操作包一层人工确认。接着把 LangSmith 或自建 trace 平台接上用 trace_id 排查一次真实问题。最后着手搭建回归评测集把 Agent 的 Prompt 和工具调整从“拍脑袋”变成“数据驱动”。如果本文对你有帮助可以收藏备用。后面我还会继续拆 LangGraph 的条件路由深度变体、子图复用、Checkpointer 持久化以及 MCP Server 在生产环境的高可用部署方案可以关注后续更新。