企业官网建设流程全解析

很多团队做了半年 Agent，最后落地的 Demo 还是“一个 while 循环 + 一个 OpenAI 函数调用”。代码能跑，但一碰到线上流量、权限边界、外部系统接入、问题回溯，立刻捉襟见肘。本文不聊概念，直接给你一套 LangChain 1.0+LangGraph 1.0 的企业级 Agent 搭建思路，覆盖状态图设计、条件路由、MCP 工具接入、全链路可观测和常见生产问题，附完整可运行代码。

之前我在业务迭代中被两件事逼疯：一是 Agent 行为不可控，模型一旦多轮调用就容易绕圈、丢状态；二是接外部系统时，每个系统都要自研一套“函数调用协议”，接口层次不齐，联调效率极低。后来把 Agent 的编排层从 LangChain 原生链式调用切换到了 LangGraph 状态图，再用 MCP（Model Context Protocol）统一外部工具接入，最后在关键路径上补了结构化日志和链路追踪，才真正感受到“工程化 Agent”和“Demo Agent”的区别。

这篇文章适合三类读者：

• 已经会用 LangChain 写简单工具调用，但想进阶到带状态、带分支、带人工审批的 Agent。
• 后端开发，想把 Agent 接入企业内部工单、数据库、RAG 服务，但对 MCP 协议还不熟悉。
• 正在做 Agent 生产落地的技术负责人，想找一份可观测、有审计、能出问题的排查清单。

读完你会掌握：LangChain 和 LangGraph 的分工边界、LangGraph 状态图核心建模方法、MCP 在 LangChain 生态里的接入方式、以及一套从“记录日志”升级到“全链路可观测”的落地路径。

1. 为什么企业级 Agent 不能再手搓 Demo

1.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 通常长这样：

1. 用户提问。
2. 把问题、工具列表、历史记录一股脑塞给大模型。
3. 大模型返回一个工具调用。
4. 代码里exec()或者 if-else 执行函数。
5. 把结果拼回去，再调用一次大模型。
6. 循环，直到模型说“完成”。

这套流程在小范围验证时没问题，但企业级场景下会暴露五个短板：

1.3 企业级 Agent 的三个核心维度

结合我自己的工程经验，下面三个维度是判断一个 Agent 能不能上生产的底线：

第一，安全可控。Agent 不能是一个“模型自由发挥的黑盒”。要有最大步数限制，要有工具白名单，要有数据脱敏，要有敏感操作审批。LangGraph 的状态图天然适合做这些：每个节点都是一个函数，函数内部可以做权限校验；每条边都可以加条件，条件不满足就走进度分支。

第二，标准化接入。内部系统千奇百怪，如果每个系统都单独开发函数调用接口，Agent 的工具层会很快腐化。MCP 的价值在于，它给“外部工具”定了一套统一协议：工具描述、参数 Schema、调用返回结果格式都是标准化的。新系统接入时，只需要实现一个 MCP Server，LangChain 侧就能动态加载工具。

第三，全链路可观测。一个 Agent 任务可能涉及一次模型调用、三次工具调用、两次状态变更。如果只有最终结果，出了问题无法定位是模型判断错了、工具返回错了，还是状态被覆盖了。可观测性的做法，是要让每一步“可回放”。

2. 技术底座与环境准备

2.1 版本背景说明

本文标题写的是 LangChain 1.0+LangGraph 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_KEY=sk-xxxx OPENAI_BASE_URL=https://api.openai.com/v1 MCP_TICKET_SERVER_URL=http://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 状态图核心建模：从线性链到可控 Agent

3.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 最怕两类问题：

1. 模型反复调用同一个工具，陷入死循环。
2. 工具出错后不断重试，无法跳出。

LangGraph 提供了recursion_limit来限制整张图的执行步数：

config = {"recursion_limit": 10} result = app.invoke({"messages": []}, config=config)

如果超限，会抛出类似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 是什么

MCP（Model Context Protocol）是 Anthropic 在 2024 年底开源的一套“模型上下文协议”。它的作用，是定义了大模型应用与外部工具、数据源之间的通信标准。

在没有 MCP 之前，Agent 接入一个外部系统要经历：

1. 看对方 API 文档。
2. 写一个 Python/Java 函数封装 HTTP 请求。
3. 用@tool装饰器装饰成 LangChain Tool。
4. 手动维护参数类型和描述。

每接一个系统，重复一遍。系统多了以后，工具变成一座“屎山”。

有了 MCP 之后，外部系统只需要提供一个 MCP Server，把能力暴露成标准工具。LangChain 侧通过 MCP Client 加载工具，就能获得：

• 工具名。
• 参数 Schema。
• 工具描述。
• 调用执行能力。

这样 LangChain Agent 与具体系统之间，不再强耦合。新系统接入，不需要改 Agent 主流程，只需要配置一个新的 MCP Server 地址，然后动态加载即可。

4.2 在 LangChain 中接入 MCP Server

官方适配包是langchain-mcp-adapters。核心函数是load_mcp_tools。

下面演示如何连接一个远程 MCP Server：

from 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_client：

async def load_local_mcp_tools(command: str, args: list[str]): server_params = StdioServerParameters( command=command, args=args, env=None, ) 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 = f"TICKET-{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(transport="sse", host="0.0.0.0", port=9000, sse_path="/sse")

这个 Server 启动后，LangChain 侧就可以通过streamablehttp_client或 SSE 地址加载它的工具。

关于“MCP 工具注册不上”的问题，我在后面常见问题里专门展开。

4.4 MCP Tool 与 LangChain Tool 如何共存

不是所有工具都适合走 MCP。以下情况直接用 LangChain@tool更轻量：

• 纯内部函数，比如密码校验、字符串处理。
• 数据访问层已有的 Service 方法。
• 与外部系统无关的本地逻辑。

而以下情况建议走 MCP：

• 多个 Agent 共享一套工具能力。
• 工具由不同团队维护，接口需要统一治理。
• 需要在运行时动态注册/卸载工具。
• 工具调用需要独立的鉴权、限流、审计。

实际项目中，通常两种方式混合使用。LangGraph 节点里可以把普通工具和 MCP 工具合并到一个列表中，一起绑定给模型：

all_tools = local_tools + mcp_tools

5. 全链路可观测：从 print 到 trace

5.1 为什么 Agent 的可观测性比普通后端更难

普通后端接口的可观测性，往往只需要记录“入参、出参、耗时、错误”。但 Agent 是“多轮决策系统”，一个问题可能触发多次模型调用和多次工具调用。

有一个线上案例让我印象很深：用户反馈“Agent 查不到库存数据”。从界面看，最终回答是“暂无可售库存”，看起来没有问题。但实际是：

1. 模型第一次选择了“查询商品ID”工具，但参数传错了，查出一个空列表。
2. 模型第二次没有继续查，而是直接根据空列表得出结论。
3. 整个链路毫无异常，没有报错。

如果只有结果日志，这个问题根本无法定位。必须记录每一步的工具入参、工具出参、模型决策理由，才能还原现场。

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_ascii=False)) 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_ascii=False))

注意：工具出参不要全量打日志，截断到前 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_ascii=False)) return {"messages": [response], "step_count": state["step_count"] + 1}

通过trace_id，就能把“用户请求 → 模型调用 → 工具调用 → 最终回复”串成一条完整链路。遇到问题时，只需要按trace_id搜索日志。

5.4 关于 LangSmith 和其他可观测平台

如果团队条件允许，可以直接接入 LangChain 官方推出的 LangSmith，它提供了非常完善的 trace、评价、数据集管理能力。配置方式很简单：

LANGSMITH_TRACING=true LANGSMITH_API_KEY=your_api_key LANGSMITH_PROJECT=your_project_name

不过，如果你所在团队的网络环境或数据合规要求不允许使用外部 SaaS，建议按照上面的思路自建结构化日志，再配合 OpenTelemetry 将 trace 打到内部链路平台。自建方案的成本并不高，核心是要在关键节点埋点。

6. 完整实战：构建一个带 MCP 工具与可观测性的企业级 Agent

6.1 需求拆解

用一个“内部工单助手”作为案例，需求如下：

1. 用户通过 HTTP 发起问题，例如“刚才我提交的 TICKET-0001 现在什么状态？如果还没完成，请帮我催一下”。
2. Agent 需要查询工单状态。如果状态是“处理中”，则创建一个“催办工单”并返回。
3. 所有调用需要记录日志，带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_node：LangGraph 有一个内置的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_ascii=False)) 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_ascii=False)) 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_ascii=False)) 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( model="gpt-4o-mini", temperature=0, api_key=OPENAI_API_KEY, base_url=OPENAI_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_limit=10, ) result = await app.state.agent.ainvoke( { "messages": [{"role": "user", "content": user_message}], "step_count": 0, "trace_id": trace_id, }, config=config, ) 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 运行结果说明

预期行为是这样一条决策链：

1. 模型判断需要查工单状态，调用query_ticket_status。
2. ToolNode执行 MCP 工具，返回“处理中”。
3. 模型再次判断，因为状态是“处理中”，决定创建催办工单，调用create_ticket。
4. ToolNode执行完成，返回工单编号。
5. 模型总结：“TICKET-0001 仍在处理中，我已帮你创建催办工单 TICKET-xxxx。”

整个过程没有报错，但如果最终答案不是用户想要的，你可以通过 trace_id 查看是哪一环模型判断出错，还是工具返回了错误结果。这就是全链路可观测的价值。

7. 常见问题与排查思路

7.1 高频问题排查表

7.2 “MCP 工具注册不上”的排查思路

这是很多开发者遇到的高频问题。现象是：MCP Server 正常启动，但 LangChain 加载出来的工具列表是空的，或者 Agent 始终不调用 MCP 工具。

按以下顺序排查：

1. 验证 MCP Server 本身是否有问题。直接用 MCP Inspector 或写一个最小 Client 连接，先确认工具能拉取到。
2. 检查 transport 是否匹配。Server 是 SSE，Client 就不要用 stdio；反之亦然。
3. 检查 Session 是否正确初始化。必须调用await session.initialize()，否则拉取不到工具列表。
4. 检查工具描述是否规范。有些模型对空描述的工具会“选择不调用”，尽量给每个工具写清楚触发场景。
5. 检查鉴权。如果 MCP Server 有鉴权，Client 必须在 header 中带上 token。
6. 检查加载时机。不要在每次请求时重新加载全部工具，启动时加载一次并缓存，避免连接泄漏。

7.3 “The agent execution provider did not respond in time” 类超时问题

这类报错的核心是：Agent 执行链路中某一步没有在预期时间内返回，通常是模型调用超时或工具调用超时。

处理方式：

1. 先看日志里最后一步是llm_call还是tools_node，缩小范围。
2. 如果是模型调用超时，考虑换模型、降输入长度、或者给 LLM 客户端设置更长 timeout。
3. 如果是工具调用超时，优先检查 MCP Server 的响应时间。

llm = ChatOpenAI( model="gpt-4o-mini", temperature=0, timeout=60, # 单位秒，按实际情况调整 )

4. 在设计上，外部工具调用建议设置“软超时”和“降级方案”。比如工单系统查询超过 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 不是传统代码，不能只靠单测。建议建立“回归评测集”：

1. 准备 50 到 200 条典型用户问题，覆盖主要业务分支。
2. 每条问题标注期望行为（调用什么工具、最终答案方向）。
3. 每次修改 Prompt、工具、模型版本后，跑一遍评测集。
4. 对比通过率，通过后再发布。

这样能最大限度避免“改了 A 场景，破坏了 B 场景”的回归问题。

8.5 从 Demo 到生产的演进清单

如果你正在把 Demo Agent 往生产推，可以按这个顺序检查：

1. 是否设置了recursion_limit？
2. 是否有兜底回复节点？
3. 工具是否做了权限校验？
4. 是否记录 trace_id 和关键节点日志？
5. MCP 连接是否复用？
6. 是否有回归评测集？
7. 模型 API Key 和 MCP Server 密钥是否在密钥管理系统里？

每一项目前都可以用很小的工作量补齐，但缺了任何一项，线上都可能出事故。

9. 总结与下一步学习路线

这篇文章的核心，是把 Agent 从“Demo 思维”拉回到“工程思维”。

你掌握了三块核心能力：

1. 用 LangGraph 状态图控制 Agent：State、Node、Conditional Edge、循环控制、Subgraph 建模。
2. 用 MCP 标准化外部工具接入：MCP Server 实现、MCP Client 加载、LangChain Tool 共存策略。
3. 用结构化日志实现全链路可观测：trace_id 贯穿节点、LLM 调用、工具调用，支持线上问题回溯。

下一步建议按这个顺序继续深入：

• 先把本文的代码复制到本地，跑通一个最小 Agent。
• 然后尝试加上interrupt人工审批节点，把所有敏感写操作包一层人工确认。
• 接着把 LangSmith 或自建 trace 平台接上，用 trace_id 排查一次真实问题。
• 最后着手搭建回归评测集，把 Agent 的 Prompt 和工具调整从“拍脑袋”变成“数据驱动”。

如果本文对你有帮助，可以收藏备用。后面我还会继续拆 LangGraph 的条件路由深度变体、子图复用、Checkpointer 持久化，以及 MCP Server 在生产环境的高可用部署方案，可以关注后续更新。