LangChain 1.0 和 LangGraph 1.0 的组合现在最值得讨论的不是“能不能调通”而是“怎么按企业级标准去落地”。如果你还在写单节点的 LLM 调用脚本或者停留在“包一层 HTTP API 就算 Agent”的阶段这篇实战文章可以直接收藏。这次我们会把重点放在三件事上安全可控、MCP 接入、全链路可观测。先说结论LangChain 负责把模型、Prompt、工具、记忆这些组件串起来LangGraph 负责把执行流程变成有状态、可编排、可恢复的图。当你需要构建多轮对话、条件路由、子图拆解、循环检测、断点恢复这类企业级 Agent 能力时LangGraph 的价值会非常明显。而 MCP 则解决了 Agent“接工具难”的问题用统一协议把外部系统接进来。最后再挂上 Trace、日志和指标整个 Agent 的状态才能被看见、被审计、被优化。这篇文章会带你走完整套方法论从环境准备开始到用 LangGraph 搭出 Agent 骨架再到接入 MCP Server加入安全控制策略最后配置全链路可观测性并给出接口 API、批量任务、资源占用和排错思路。内容偏工程落地不是理论科普。1. LangChain LangGraph 企业级 Agent 核心能力速览能力项说明技术栈LangChain 负责模型、Prompt、记忆与工具抽象LangGraph 负责节点编排、状态流转、分支与循环控制核心能力多节点工作流、条件路由、循环检测、子图拆分、状态持久化、并行分支工具接入支持 MCP 协议可接入数据库、HTTP 服务、内部系统、第三方 API 等安全控制可做工具白名单、参数校验、敏感数据脱敏、人工审核节点、审计日志可观测性支持集成 LangSmith 或自研 Trace 方案记录调用链、耗时、Token 消耗、错误信息部署形态本地开发进程、Docker 容器、FastAPI 服务、LangGraph 平台服务需按项目选型批量任务可设计异步队列、逐条执行、失败重试、幂等记录适合批量客服问答、定时分析等场景硬件要求开发机即可运行实际取决于接入的 LLM 后端云端 API 或本地模型适合场景企业知识库问答、智能客服、办公自动化、数据分析助手、流程审批助手、多工具调用 Agent需要注意这里的表格是基于 LangChain 1.0 LangGraph 1.0 的常规企业落地能力整理具体到某一版本的 API 细节请以你本机安装的 SDK 和官方文档为准。2. 适用场景与使用边界这套组合适合三类团队。第一类正在做企业内部知识库和客服系统的团队。这类系统通常需要多轮对话、历史记忆、权限控制、敏感信息审计LangGraph 的状态管理和可观测性能直接覆盖这些需求。第二类正在把 AI 能力接入到业务流程中的团队。比如订单查询、库存查询、工单填写、数据报表生成Agent 需要调用多个内部系统。MCP 协议可以避免每个系统都手写一套工具适配统一接入成本会低很多。第三类已经有 LLM 应用但觉得“单次调用提示词”不够用想升级成真正有状态、可编排的 Agent 架构的团队。但同时也要明确使用边界。不推荐把它当成“零代码平台”需要一定 Python 工程能力。不推荐在前置业务逻辑都没理清时直接上 Agent 编排图结构会越做越复杂。不推荐在未经安全评审的情况下让 Agent 直接操作核心系统比如转账、删除数据、修改权限。工具调用必须有闸门。版权和安全边界也要提前定清楚涉及用户隐私数据、商业机密、受版权保护的文本、图片、音频时必须确认授权范围并在设计上做脱敏和权限隔离。任何 Agent 生成内容在正式发布或商用前都要经过人工复核。这套技术解决的问题是“复杂流程管理”不是“模型能力替代”。模型本身能力不足时框架再好也补不上推理质量。3. 本地部署环境准备开始写代码之前先把环境理清楚。以下是通用检查清单具体版本号需要以你本机安装时能查到的最新稳定版为准。3.1 操作系统与运行环境操作系统Windows 10/11、Ubuntu 20.04、macOS 均可推荐 Linux 或 Windows WSL2 做生产级开发。Python建议 3.10 及以上版本。LangChain 和 LangGraph 生态对新版本 Python 支持更及时。包管理工具推荐uv或pip建议在独立虚拟环境中安装依赖避免和系统 Python 环境互相污染。3.2 安装核心依赖# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate # 安装核心库版本号请以当前官方最新版为准 pip install langchain langgraph langchain-openai langchain-community # 如果要用 MCP 对接工具需要按项目的 MCP SDK 要求安装 # pip install langchain-mcp mcp # 如果要做 API 服务安装 FastAPI pip install fastapi uvicorn注意LangChain 和 LangGraph 现在是独立演进的项目安装时不要混用老版本的依赖锁定文件。建议在requirements.txt或pyproject.toml中固定版本方便复现。3.3 配置模型访问企业级 Agent 通常需要接入一个或多个 LLM 后端。以 OpenAI 兼容接口为例配置环境变量# OpenAI 兼容接口 export OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://your-llm-endpoint/v1如果你是用本地推理服务比如 vLLM、Ollama则改成对应的 Base URL。密钥管理建议使用环境变量或密钥管理服务不要写死在代码里。3.4 端口与磁盘本地 API 服务默认可用 8000、9000 之类的端口启动前先确认端口没有被占用。需要预留足够磁盘空间依赖包、日志、向量存储、临时文件都会占空间。如果要用向量检索还要预留模型存储和向量库容量。4. 用 LangGraph 1.0 搭出 Agent 骨架LangGraph 的核心思想是把 Agent 执行过程抽象成一张有向图。图里面有节点、边、条件边、状态。每个节点代表一个执行步骤边定义流转方向条件边根据状态决定下一步走向。这个模式比手写while True循环清晰得多也方便做断点恢复和审计。4.1 最小可运行图结构下面是一个通用模板。实际项目中decide节点会调用 LLM 判断下一步run_tool节点会执行具体工具check_safety节点会在执行前做安全校验。from typing import Annotated from typing_extensions import TypedDict import operator from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: Annotated[list, operator.add] next_action: str safe_to_proceed: bool def decide(state: AgentState) - dict: # 此处应调用大模型判断 user query 需要走哪个 action return {next_action: run_tool} def check_safety(state: AgentState) - dict: # 对 next_action 做权限校验不允许直接返回 False allowed {run_tool, finish} return {safe_to_proceed: state[next_action] in allowed} def run_tool(state: AgentState) - dict: # 执行实际工具调用比如查询数据库、调用内部 API return {messages: [tool executed]} def router(state: AgentState) - str: if not state[safe_to_proceed]: return __end__ return state[next_action] if state[next_action] in (run_tool,) else __end__ graph StateGraph(AgentState) graph.add_node(decide, decide) graph.add_node(check_safety, check_safety) graph.add_node(run_tool, run_tool) graph.add_edge(START, decide) graph.add_edge(decide, check_safety) graph.add_conditional_edges(check_safety, router, { run_tool: run_tool, __end__: END, }) graph.add_edge(run_tool, END) app graph.compile()这段代码演示了三个关键企业级设计状态集中管理AgentState里所有信息在执行过程中可见。条件路由放在check_safety节点之后权限不过直接终止。每一步之间通过节点身份和边关系清晰串联审计时可以精确还原执行链路。启动以后调用方式类似result app.invoke({messages: [查询本周订单]}) print(result)4.2 条件路由怎么设计条件路由是企业级 Agent 的刚需。上一轮用户提问可能只需要查知识库下一轮可能就需要调用数据库。在实际图里可以在decide节点返回多种next_action然后在条件边里映射到不同子图或不同工具节点。def router(state: AgentState) - str: return state[next_action] graph.add_conditional_edges(decide, router, { query_kb: query_kb_node, query_db: query_db_node, finish: END, })这种设计的好处是把业务分支变成显式配置而不是散落在代码里的if-else。分支多了以后可读性和可维护性仍然在线。4.3 子图和并行分支当业务变复杂比如一个 Agent 既要查订单、又要查库存、还要生成报表就可以拆子图。每个子图负责一个独立子任务主图负责编排。LangGraph 对子图的支持很成熟能让复杂流程模块化。并行分支适用于“互相没有依赖的多路查询”比如同时查询数据库和调用 Web 搜索。并行能提速但要注意底层 API 的限流策略并发数不要直接打满。5. MCP 接入让 Agent 真正接进企业系统MCPModel Context Protocol解决的是模型与外部工具之间的互联标准问题。以前接一个企业内部系统就要写一套自定义工具封装现在通过 MCP 协议可以用统一方式把工具、数据源、内部服务暴露给 Agent。5.1 MCP Server 配置在接入之前需要先有可用的 MCP Server。这个 Server 可以由后端团队提供也可以把现有的 HTTP 接口包一层 MCP。下面是一个通用配置示意{ mcpServers: [ { name: order-service, transport: http, url: http://127.0.0.1:8080/mcp, headers: { Authorization: Bearer token } }, { name: internal-kb, transport: stdio, command: python, args: [-m, kb_mcp_server] } ] }不同 MCP SDK 的加载方式有差异具体以项目选择的客户端 SDK 为准。核心是Agent 不需要感知每个工具的内部实现只需要知道 MCP Server 暴露了哪些工具以及工具参数。5.2 把 MCP 工具绑定到 LangGraph 节点MCP Server 中的工具可以加载到 LangChain 的工具列表中然后在 LangGraph 的节点中调用。示意流程如下# 伪代码不同包版本的 API 有差异以官方文档为准 mcp_tools await load_mcp_tools(order-service) # 将工具列表交给 Agent 节点 async def call_agent(state: AgentState) - dict: llm_with_tools llm.bind_tools(mcp_tools) response await llm_with_tools.ainvoke(state[messages]) return {messages: [response]}5.3 企业接入 MCP 的两个建议第一权限收口。不是所有 MCP 工具都对所有用户开放。工具加载之后建议在节点层做权限判断而不是让模型自由选择所有工具。ALLOWED_TOOLS { query_order: [admin, sales], create_ticket: [all], delete_record: [], }第二参数校验。MCP 工具暴露后模型生成的参数可能不合法。在执行前增加参数校验节点或者在各工具内部做校验避免脏数据进入核心系统。6. 安全可控企业级 Agent 的底线设计企业级 Agent 和 Demo 的最大区别就是“失败是否可控”。如果说不可控只是生成结果不好看那可以接受如果不可控会触发误操作、泄露数据、越权调用那就不能上线。6.1 工具调用白名单与黑名单在 LangGraph 中可以在节点之间插入“权限校验节点”。核心规则是默认拒绝白名单放行。用户输入 - 意图识别 - 权限校验节点 - 工具调用节点 - 输出 | -- 无权限时走拒绝分支白名单建议做两维设计一是工具级别二是数据级别。工具级别控制“能不能调用这个工具”数据级别控制“能传哪些参数、能查哪些范围”。比如一个订单查询工具普通用户只能查自己的订单管理员可以查全量订单。这个逻辑不要只靠模型自觉要在工具层强校验。6.2 敏感数据脱敏输入到模型的数据可能包含手机号、身份证、地址、密钥等敏感信息。在发往模型之前先做脱敏处理返回结果时再做恢复或保留脱敏状态。def mask_sensitive(text: str) - str: # 示意函数 text text.replace(13800000000, 138****0000) return text对需要长期存档的对话记录建议不保存明文敏感字段只保留脱敏后的日志。6.3 人工审核节点对高风险操作比如转账、删除数据、修改权限、批量发送消息必须在图上加人工审核节点。LangGraph 支持图执行中断等待审核通过后继续执行。这种 human-in-the-loop 设计是企业级系统最可靠的安全兜底。6.4 审计日志每个 Agent 调用都应记录请求用户 ID 或会话 ID调用时间输入内容摘要脱敏后触发的节点链路调用的工具名称工具返回状态最终输出摘要异常信息这份日志既是排查问题的依据也是合规审计的证据链。7. 全链路可观测性从 Trace 到指标Agent 应用最头疼的问题是“为什么这轮回答这么慢”“为什么这句话用了这个工具”“为什么调用失败了”。没有可观测性就只能靠猜。企业级落地必须有从 Trace 到日志再到指标的一整套方案。7.1 接入 LangSmith 或自研 TraceLangSmith 是 LangChain 生态里比较成熟的可观测性平台可以记录 Agent 每一步的输入、输出、延迟、Token 消耗。接入方式是设置环境变量export LANGCHAIN_TRACING_V2true export LANGCHAIN_PROJECTenterprise-agent export LANGCHAIN_API_KEYyour-langsmith-api-key如果企业内部不能使用外部 SaaS 服务也完全可以自研 Trace。核心是自己定义统一的 Trace ID然后在每个节点抛出trace_id、step_id、parent_id再落到日志或数据库。7.2 结构化日志不要打印一段完整的 JSON 字符串当日志应该按字段记录关键信息。推荐格式是 JSON 结构化日志方便后续接入日志平台检索。{ timestamp: 2025-04-12T10:00:00.001Z, trace_id: a1b2c3d4, step: tool_call, tool_name: query_order, latency_ms: 230, token_usage: 1200, status: success }7.3 指标监控需要关注的核心指标包括每秒请求数QPS单次 Agent 执行平均耗时各节点耗时分布Token 消耗工具调用成功率超时率异常分支触发率这些指标可以接入 Prometheus Grafana或者自建监控面板。当 Agent 耗时异常上升时能第一时间定位是模型推理慢还是工具调用慢还是状态积压导致。8. 接口 API 与批量任务Agent 搭好之后不能只在本地调试要提供一个稳定的调用入口。8.1 用 FastAPI 封装 Agent 服务from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AgentRequest(BaseModel): user_id: str query: str conversation_id: str default class AgentResponse(BaseModel): conversation_id: str reply: str trace_id: str app.post(/api/agent/invoke) async def invoke(req: AgentRequest): result await agent_app.ainvoke({ messages: [req.query], user_id: req.user_id, }) return AgentResponse( conversation_idreq.conversation_id, replyresult.get(messages, [])[-1].content, trace_idreq.conversation_id, )启动方式uvicorn main:app --host 0.0.0.0 --port 8000启动后可以直接用curl验证curl -X POST http://127.0.0.1:8000/api/agent/invoke \ -H Content-Type: application/json \ -d {user_id:u-001,query:查询本周订单数量,conversation_id:c-001}实际接口字段需要按项目调整示例只是证明调用链路能通。8.2 批量任务设计批量任务建议单独设计不要直接在高并发 HTTP 接口里同步跑长链路。可以先用消息队列接收任务Worker 逐个处理再回写结果。LangGraph 的ainvoke支持异步调用适合接到异步框架里。任务状态至少要有pending、running、success、failed。每个任务都要记录trace_id方便失败后重跑和排查。失败重试要注意幂等。比如已经成功生成了一条工单重试不能再次生成可以在核心写操作前加唯一请求 ID 校验。9. 资源占用与性能观察LangChain 和 LangGraph 本身的资源占用不算高主要资源消耗集中在 LLM 推理和工具调用上。如果接云端大模型 API本地只需要内存和少量 CPU普通开发机足够。如果接本地模型比如 7B、13B 参数模型显存占用要看模型大小和量化方式。具体多少 G 显存需要按实际模型和推理后端测试没有统一答案。并行分支会增加瞬时内存占用因为多个节点可能同时持有上下文。长时间运行的 Agent 服务要注意状态累积。如果State中不断追加对话历史和工具结果内存会逐渐上涨。建议定期清理过期会话状态或把长对话拆分成多个短任务。性能观察建议从三个维度入手模型调用耗时可以看 LangSmith Trace 或自研日志。工具调用耗时单独打点统计每个 MCP 工具的响应时间。图编排开销节点少时基本可忽略但如果图特别复杂、状态特别大还是要做压测。压测建议用大模型的 mock 服务把模型耗时固定在一个值这样能更准确地评估编排层本身的性能损耗。10. 常见问题与排查方法问题现象可能原因排查方式解决方案启动后 API 服务访问不到端口被占用或服务未启动成功查看进程日志检查端口占用情况更换端口或重启服务依赖安装失败Python 版本过低或依赖版本冲突检查 Python 版本查看 pip 错误信息升级 Python固定依赖版本模型调用一直超时API Key 失效、网络不通、模型后端负载高用 curl 直接测模型接口检查网络、密钥和模型后端状态工具调用报错MCP Server 服务未启动或参数不合法先单独测 MCP Server 工具启动 MCP Server完善参数校验条件路由走到错误分支状态字段拼写不一致或条件边映射错误打印每次节点返回的状态检查next_action的值与映射条件输出质量不稳定模型本身能力不足或提示词太简单对比不同提示词效果优化 Prompt或换更强模型批量任务卡住并发太高触发限流或队列消费异常查看队列和任务日志降低并发增加重试机制显存不足报错本地模型并行任务过多查看推理服务日志和显存占用降低并发数、换量化模型或改用云端 API11. 最佳实践与合规建议这里整理几条实战中直接能用的建议。11.1 第一批先跑最小闭环不要第一次就把所有工具都接进来。先让 Agent 跑通“用户提问 - 模型决策 - 一个工具调用 - 返回结果”的最小闭环再逐步加分支、加子图、加并发。11.2 保留可复现配置环境依赖锁定版本模型配置写进环境变量或配置中心图结构用代码版本管理。这样出了问题能回滚能对比不同版本的效果差异。11.3 目录和命名规范化模型配置文件、输入素材、日志、输出结果分目录管理。建议结构agent_project/ ├── app/ │ ├── graph.py │ ├── tools/ │ └── security.py ├── configs/ │ └── mcp_config.json ├── logs/ └── data/11.4 发布前做安全审查涉及人脸、声音、肖像、版权素材的场景必须确认授权。Agent 生成内容在商用前要做人工复核。工具调用必须做权限校验和操作审计。API 服务要限制访问来源不要在公网无鉴权暴露。11.5 监控与告警要提前接入不要等上线后再补监控。第一天就接 Trace第二天就配告警一旦线上出问题你才有足够的线索快速定位。12. 总结与下一步这次的核心思路很明确不要满足于手搓 DemoLangChain 1.0 LangGraph 1.0 在复杂流程编排、MCP 工具接入、安全审计、可观测性这几个方向上有很强的工程基础。最值得先验证的是状态管理和条件路由这是 LangGraph 和普通 LLM 脚本最大的区别。最容易踩的坑是工具权限没有收口、Trace 没有提前接入、状态无限增长没有清理。下一步建议按这个顺序推进先跑通最小 Agent 图再把一个内部系统包成 MCP Server 接进来然后加上安全校验和人工审核节点最后接上可观测性平台做成 API 服务对外提供。企业级 Agent 不是一次写出来的是一层层补出来的。这篇文章的内容可以当作你们的第一份企业级 Agent Checklist建议收藏备用。
LangChain 1.0 和 LangGraph 1.0 的组合现在最值得讨论的不是“能不能调通”而是“怎么按企业级标准去落地”。如果你还在写单节点的 LLM 调用脚本或者停留在“包一层 HTTP API 就算 Agent”的阶段这篇实战文章可以直接收藏。这次我们会把重点放在三件事上安全可控、MCP 接入、全链路可观测。先说结论LangChain 负责把模型、Prompt、工具、记忆这些组件串起来LangGraph 负责把执行流程变成有状态、可编排、可恢复的图。当你需要构建多轮对话、条件路由、子图拆解、循环检测、断点恢复这类企业级 Agent 能力时LangGraph 的价值会非常明显。而 MCP 则解决了 Agent“接工具难”的问题用统一协议把外部系统接进来。最后再挂上 Trace、日志和指标整个 Agent 的状态才能被看见、被审计、被优化。这篇文章会带你走完整套方法论从环境准备开始到用 LangGraph 搭出 Agent 骨架再到接入 MCP Server加入安全控制策略最后配置全链路可观测性并给出接口 API、批量任务、资源占用和排错思路。内容偏工程落地不是理论科普。1. LangChain LangGraph 企业级 Agent 核心能力速览能力项说明技术栈LangChain 负责模型、Prompt、记忆与工具抽象LangGraph 负责节点编排、状态流转、分支与循环控制核心能力多节点工作流、条件路由、循环检测、子图拆分、状态持久化、并行分支工具接入支持 MCP 协议可接入数据库、HTTP 服务、内部系统、第三方 API 等安全控制可做工具白名单、参数校验、敏感数据脱敏、人工审核节点、审计日志可观测性支持集成 LangSmith 或自研 Trace 方案记录调用链、耗时、Token 消耗、错误信息部署形态本地开发进程、Docker 容器、FastAPI 服务、LangGraph 平台服务需按项目选型批量任务可设计异步队列、逐条执行、失败重试、幂等记录适合批量客服问答、定时分析等场景硬件要求开发机即可运行实际取决于接入的 LLM 后端云端 API 或本地模型适合场景企业知识库问答、智能客服、办公自动化、数据分析助手、流程审批助手、多工具调用 Agent需要注意这里的表格是基于 LangChain 1.0 LangGraph 1.0 的常规企业落地能力整理具体到某一版本的 API 细节请以你本机安装的 SDK 和官方文档为准。2. 适用场景与使用边界这套组合适合三类团队。第一类正在做企业内部知识库和客服系统的团队。这类系统通常需要多轮对话、历史记忆、权限控制、敏感信息审计LangGraph 的状态管理和可观测性能直接覆盖这些需求。第二类正在把 AI 能力接入到业务流程中的团队。比如订单查询、库存查询、工单填写、数据报表生成Agent 需要调用多个内部系统。MCP 协议可以避免每个系统都手写一套工具适配统一接入成本会低很多。第三类已经有 LLM 应用但觉得“单次调用提示词”不够用想升级成真正有状态、可编排的 Agent 架构的团队。但同时也要明确使用边界。不推荐把它当成“零代码平台”需要一定 Python 工程能力。不推荐在前置业务逻辑都没理清时直接上 Agent 编排图结构会越做越复杂。不推荐在未经安全评审的情况下让 Agent 直接操作核心系统比如转账、删除数据、修改权限。工具调用必须有闸门。版权和安全边界也要提前定清楚涉及用户隐私数据、商业机密、受版权保护的文本、图片、音频时必须确认授权范围并在设计上做脱敏和权限隔离。任何 Agent 生成内容在正式发布或商用前都要经过人工复核。这套技术解决的问题是“复杂流程管理”不是“模型能力替代”。模型本身能力不足时框架再好也补不上推理质量。3. 本地部署环境准备开始写代码之前先把环境理清楚。以下是通用检查清单具体版本号需要以你本机安装时能查到的最新稳定版为准。3.1 操作系统与运行环境操作系统Windows 10/11、Ubuntu 20.04、macOS 均可推荐 Linux 或 Windows WSL2 做生产级开发。Python建议 3.10 及以上版本。LangChain 和 LangGraph 生态对新版本 Python 支持更及时。包管理工具推荐uv或pip建议在独立虚拟环境中安装依赖避免和系统 Python 环境互相污染。3.2 安装核心依赖# 创建虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate # 安装核心库版本号请以当前官方最新版为准 pip install langchain langgraph langchain-openai langchain-community # 如果要用 MCP 对接工具需要按项目的 MCP SDK 要求安装 # pip install langchain-mcp mcp # 如果要做 API 服务安装 FastAPI pip install fastapi uvicorn注意LangChain 和 LangGraph 现在是独立演进的项目安装时不要混用老版本的依赖锁定文件。建议在requirements.txt或pyproject.toml中固定版本方便复现。3.3 配置模型访问企业级 Agent 通常需要接入一个或多个 LLM 后端。以 OpenAI 兼容接口为例配置环境变量# OpenAI 兼容接口 export OPENAI_API_KEYyour-api-key export OPENAI_BASE_URLhttps://your-llm-endpoint/v1如果你是用本地推理服务比如 vLLM、Ollama则改成对应的 Base URL。密钥管理建议使用环境变量或密钥管理服务不要写死在代码里。3.4 端口与磁盘本地 API 服务默认可用 8000、9000 之类的端口启动前先确认端口没有被占用。需要预留足够磁盘空间依赖包、日志、向量存储、临时文件都会占空间。如果要用向量检索还要预留模型存储和向量库容量。4. 用 LangGraph 1.0 搭出 Agent 骨架LangGraph 的核心思想是把 Agent 执行过程抽象成一张有向图。图里面有节点、边、条件边、状态。每个节点代表一个执行步骤边定义流转方向条件边根据状态决定下一步走向。这个模式比手写while True循环清晰得多也方便做断点恢复和审计。4.1 最小可运行图结构下面是一个通用模板。实际项目中decide节点会调用 LLM 判断下一步run_tool节点会执行具体工具check_safety节点会在执行前做安全校验。from typing import Annotated from typing_extensions import TypedDict import operator from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: Annotated[list, operator.add] next_action: str safe_to_proceed: bool def decide(state: AgentState) - dict: # 此处应调用大模型判断 user query 需要走哪个 action return {next_action: run_tool} def check_safety(state: AgentState) - dict: # 对 next_action 做权限校验不允许直接返回 False allowed {run_tool, finish} return {safe_to_proceed: state[next_action] in allowed} def run_tool(state: AgentState) - dict: # 执行实际工具调用比如查询数据库、调用内部 API return {messages: [tool executed]} def router(state: AgentState) - str: if not state[safe_to_proceed]: return __end__ return state[next_action] if state[next_action] in (run_tool,) else __end__ graph StateGraph(AgentState) graph.add_node(decide, decide) graph.add_node(check_safety, check_safety) graph.add_node(run_tool, run_tool) graph.add_edge(START, decide) graph.add_edge(decide, check_safety) graph.add_conditional_edges(check_safety, router, { run_tool: run_tool, __end__: END, }) graph.add_edge(run_tool, END) app graph.compile()这段代码演示了三个关键企业级设计状态集中管理AgentState里所有信息在执行过程中可见。条件路由放在check_safety节点之后权限不过直接终止。每一步之间通过节点身份和边关系清晰串联审计时可以精确还原执行链路。启动以后调用方式类似result app.invoke({messages: [查询本周订单]}) print(result)4.2 条件路由怎么设计条件路由是企业级 Agent 的刚需。上一轮用户提问可能只需要查知识库下一轮可能就需要调用数据库。在实际图里可以在decide节点返回多种next_action然后在条件边里映射到不同子图或不同工具节点。def router(state: AgentState) - str: return state[next_action] graph.add_conditional_edges(decide, router, { query_kb: query_kb_node, query_db: query_db_node, finish: END, })这种设计的好处是把业务分支变成显式配置而不是散落在代码里的if-else。分支多了以后可读性和可维护性仍然在线。4.3 子图和并行分支当业务变复杂比如一个 Agent 既要查订单、又要查库存、还要生成报表就可以拆子图。每个子图负责一个独立子任务主图负责编排。LangGraph 对子图的支持很成熟能让复杂流程模块化。并行分支适用于“互相没有依赖的多路查询”比如同时查询数据库和调用 Web 搜索。并行能提速但要注意底层 API 的限流策略并发数不要直接打满。5. MCP 接入让 Agent 真正接进企业系统MCPModel Context Protocol解决的是模型与外部工具之间的互联标准问题。以前接一个企业内部系统就要写一套自定义工具封装现在通过 MCP 协议可以用统一方式把工具、数据源、内部服务暴露给 Agent。5.1 MCP Server 配置在接入之前需要先有可用的 MCP Server。这个 Server 可以由后端团队提供也可以把现有的 HTTP 接口包一层 MCP。下面是一个通用配置示意{ mcpServers: [ { name: order-service, transport: http, url: http://127.0.0.1:8080/mcp, headers: { Authorization: Bearer token } }, { name: internal-kb, transport: stdio, command: python, args: [-m, kb_mcp_server] } ] }不同 MCP SDK 的加载方式有差异具体以项目选择的客户端 SDK 为准。核心是Agent 不需要感知每个工具的内部实现只需要知道 MCP Server 暴露了哪些工具以及工具参数。5.2 把 MCP 工具绑定到 LangGraph 节点MCP Server 中的工具可以加载到 LangChain 的工具列表中然后在 LangGraph 的节点中调用。示意流程如下# 伪代码不同包版本的 API 有差异以官方文档为准 mcp_tools await load_mcp_tools(order-service) # 将工具列表交给 Agent 节点 async def call_agent(state: AgentState) - dict: llm_with_tools llm.bind_tools(mcp_tools) response await llm_with_tools.ainvoke(state[messages]) return {messages: [response]}5.3 企业接入 MCP 的两个建议第一权限收口。不是所有 MCP 工具都对所有用户开放。工具加载之后建议在节点层做权限判断而不是让模型自由选择所有工具。ALLOWED_TOOLS { query_order: [admin, sales], create_ticket: [all], delete_record: [], }第二参数校验。MCP 工具暴露后模型生成的参数可能不合法。在执行前增加参数校验节点或者在各工具内部做校验避免脏数据进入核心系统。6. 安全可控企业级 Agent 的底线设计企业级 Agent 和 Demo 的最大区别就是“失败是否可控”。如果说不可控只是生成结果不好看那可以接受如果不可控会触发误操作、泄露数据、越权调用那就不能上线。6.1 工具调用白名单与黑名单在 LangGraph 中可以在节点之间插入“权限校验节点”。核心规则是默认拒绝白名单放行。用户输入 - 意图识别 - 权限校验节点 - 工具调用节点 - 输出 | -- 无权限时走拒绝分支白名单建议做两维设计一是工具级别二是数据级别。工具级别控制“能不能调用这个工具”数据级别控制“能传哪些参数、能查哪些范围”。比如一个订单查询工具普通用户只能查自己的订单管理员可以查全量订单。这个逻辑不要只靠模型自觉要在工具层强校验。6.2 敏感数据脱敏输入到模型的数据可能包含手机号、身份证、地址、密钥等敏感信息。在发往模型之前先做脱敏处理返回结果时再做恢复或保留脱敏状态。def mask_sensitive(text: str) - str: # 示意函数 text text.replace(13800000000, 138****0000) return text对需要长期存档的对话记录建议不保存明文敏感字段只保留脱敏后的日志。6.3 人工审核节点对高风险操作比如转账、删除数据、修改权限、批量发送消息必须在图上加人工审核节点。LangGraph 支持图执行中断等待审核通过后继续执行。这种 human-in-the-loop 设计是企业级系统最可靠的安全兜底。6.4 审计日志每个 Agent 调用都应记录请求用户 ID 或会话 ID调用时间输入内容摘要脱敏后触发的节点链路调用的工具名称工具返回状态最终输出摘要异常信息这份日志既是排查问题的依据也是合规审计的证据链。7. 全链路可观测性从 Trace 到指标Agent 应用最头疼的问题是“为什么这轮回答这么慢”“为什么这句话用了这个工具”“为什么调用失败了”。没有可观测性就只能靠猜。企业级落地必须有从 Trace 到日志再到指标的一整套方案。7.1 接入 LangSmith 或自研 TraceLangSmith 是 LangChain 生态里比较成熟的可观测性平台可以记录 Agent 每一步的输入、输出、延迟、Token 消耗。接入方式是设置环境变量export LANGCHAIN_TRACING_V2true export LANGCHAIN_PROJECTenterprise-agent export LANGCHAIN_API_KEYyour-langsmith-api-key如果企业内部不能使用外部 SaaS 服务也完全可以自研 Trace。核心是自己定义统一的 Trace ID然后在每个节点抛出trace_id、step_id、parent_id再落到日志或数据库。7.2 结构化日志不要打印一段完整的 JSON 字符串当日志应该按字段记录关键信息。推荐格式是 JSON 结构化日志方便后续接入日志平台检索。{ timestamp: 2025-04-12T10:00:00.001Z, trace_id: a1b2c3d4, step: tool_call, tool_name: query_order, latency_ms: 230, token_usage: 1200, status: success }7.3 指标监控需要关注的核心指标包括每秒请求数QPS单次 Agent 执行平均耗时各节点耗时分布Token 消耗工具调用成功率超时率异常分支触发率这些指标可以接入 Prometheus Grafana或者自建监控面板。当 Agent 耗时异常上升时能第一时间定位是模型推理慢还是工具调用慢还是状态积压导致。8. 接口 API 与批量任务Agent 搭好之后不能只在本地调试要提供一个稳定的调用入口。8.1 用 FastAPI 封装 Agent 服务from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AgentRequest(BaseModel): user_id: str query: str conversation_id: str default class AgentResponse(BaseModel): conversation_id: str reply: str trace_id: str app.post(/api/agent/invoke) async def invoke(req: AgentRequest): result await agent_app.ainvoke({ messages: [req.query], user_id: req.user_id, }) return AgentResponse( conversation_idreq.conversation_id, replyresult.get(messages, [])[-1].content, trace_idreq.conversation_id, )启动方式uvicorn main:app --host 0.0.0.0 --port 8000启动后可以直接用curl验证curl -X POST http://127.0.0.1:8000/api/agent/invoke \ -H Content-Type: application/json \ -d {user_id:u-001,query:查询本周订单数量,conversation_id:c-001}实际接口字段需要按项目调整示例只是证明调用链路能通。8.2 批量任务设计批量任务建议单独设计不要直接在高并发 HTTP 接口里同步跑长链路。可以先用消息队列接收任务Worker 逐个处理再回写结果。LangGraph 的ainvoke支持异步调用适合接到异步框架里。任务状态至少要有pending、running、success、failed。每个任务都要记录trace_id方便失败后重跑和排查。失败重试要注意幂等。比如已经成功生成了一条工单重试不能再次生成可以在核心写操作前加唯一请求 ID 校验。9. 资源占用与性能观察LangChain 和 LangGraph 本身的资源占用不算高主要资源消耗集中在 LLM 推理和工具调用上。如果接云端大模型 API本地只需要内存和少量 CPU普通开发机足够。如果接本地模型比如 7B、13B 参数模型显存占用要看模型大小和量化方式。具体多少 G 显存需要按实际模型和推理后端测试没有统一答案。并行分支会增加瞬时内存占用因为多个节点可能同时持有上下文。长时间运行的 Agent 服务要注意状态累积。如果State中不断追加对话历史和工具结果内存会逐渐上涨。建议定期清理过期会话状态或把长对话拆分成多个短任务。性能观察建议从三个维度入手模型调用耗时可以看 LangSmith Trace 或自研日志。工具调用耗时单独打点统计每个 MCP 工具的响应时间。图编排开销节点少时基本可忽略但如果图特别复杂、状态特别大还是要做压测。压测建议用大模型的 mock 服务把模型耗时固定在一个值这样能更准确地评估编排层本身的性能损耗。10. 常见问题与排查方法问题现象可能原因排查方式解决方案启动后 API 服务访问不到端口被占用或服务未启动成功查看进程日志检查端口占用情况更换端口或重启服务依赖安装失败Python 版本过低或依赖版本冲突检查 Python 版本查看 pip 错误信息升级 Python固定依赖版本模型调用一直超时API Key 失效、网络不通、模型后端负载高用 curl 直接测模型接口检查网络、密钥和模型后端状态工具调用报错MCP Server 服务未启动或参数不合法先单独测 MCP Server 工具启动 MCP Server完善参数校验条件路由走到错误分支状态字段拼写不一致或条件边映射错误打印每次节点返回的状态检查next_action的值与映射条件输出质量不稳定模型本身能力不足或提示词太简单对比不同提示词效果优化 Prompt或换更强模型批量任务卡住并发太高触发限流或队列消费异常查看队列和任务日志降低并发增加重试机制显存不足报错本地模型并行任务过多查看推理服务日志和显存占用降低并发数、换量化模型或改用云端 API11. 最佳实践与合规建议这里整理几条实战中直接能用的建议。11.1 第一批先跑最小闭环不要第一次就把所有工具都接进来。先让 Agent 跑通“用户提问 - 模型决策 - 一个工具调用 - 返回结果”的最小闭环再逐步加分支、加子图、加并发。11.2 保留可复现配置环境依赖锁定版本模型配置写进环境变量或配置中心图结构用代码版本管理。这样出了问题能回滚能对比不同版本的效果差异。11.3 目录和命名规范化模型配置文件、输入素材、日志、输出结果分目录管理。建议结构agent_project/ ├── app/ │ ├── graph.py │ ├── tools/ │ └── security.py ├── configs/ │ └── mcp_config.json ├── logs/ └── data/11.4 发布前做安全审查涉及人脸、声音、肖像、版权素材的场景必须确认授权。Agent 生成内容在商用前要做人工复核。工具调用必须做权限校验和操作审计。API 服务要限制访问来源不要在公网无鉴权暴露。11.5 监控与告警要提前接入不要等上线后再补监控。第一天就接 Trace第二天就配告警一旦线上出问题你才有足够的线索快速定位。12. 总结与下一步这次的核心思路很明确不要满足于手搓 DemoLangChain 1.0 LangGraph 1.0 在复杂流程编排、MCP 工具接入、安全审计、可观测性这几个方向上有很强的工程基础。最值得先验证的是状态管理和条件路由这是 LangGraph 和普通 LLM 脚本最大的区别。最容易踩的坑是工具权限没有收口、Trace 没有提前接入、状态无限增长没有清理。下一步建议按这个顺序推进先跑通最小 Agent 图再把一个内部系统包成 MCP Server 接进来然后加上安全校验和人工审核节点最后接上可观测性平台做成 API 服务对外提供。企业级 Agent 不是一次写出来的是一层层补出来的。这篇文章的内容可以当作你们的第一份企业级 Agent Checklist建议收藏备用。
)
