LangGraph+MCP+Harness:构建生产级AI Agent的工程实践
搭建一个真正能落地的 AI Agent不只在笔记本上跑一个“记住聊天上下文”的 demo。这次我们用 LangGraph 做状态编排、MCP 接外部工具、再套一层 Harness 约束执行边界最后把安全架构补上。你会看到 Agent 怎么拆图、怎么接工具、怎么控制循环和并行分支以及哪些坑是框架本身不会替你解决的。这篇文章不是纯概念梳理我会按“环境准备 → 启动项目 → 实现 Agent → 接 MCP → 加安全控制 → 验证效果”的顺序走一遍。老规矩先给规格结论再讲操作细节。1. 核心能力速览能力项说明项目类型AI Agent 编排框架基于状态图实现复杂任务流程核心框架LangGraphLangChain 生态里的图编排方案工具接入支持 MCPModel Context Protocol标准可对接文件、数据库、浏览器、开发工具等执行控制Harness 层可约束工具调用范围、控制迭代轮数、限制敏感操作运行方式Python 脚本、LangGraph CLI 开发模式、服务化 API语言要求Python 3.9建议 3.11 及以上硬件要求开发调试 CPU 即可LLM 推理按模型选择 CPU/GPU无强制要求API 能力LangGraph Server / SDK 暴露接口支持异步任务和人工介入批量任务图天然支持循环、并行分支、条件路由适合流程化批量处理适合人群后端工程师、AI 应用开发者、想从 demo 走向工程的 Agent 项目负责人这里的核心结论是LangGraph 解决的是“流程可控、状态可恢复、分支可预测”MCP 解决的是“工具接入方式统一”Harness 解决的是“Agent 的行为边界”。三者拼起来才是生产环境能用的 Agent。2. 适用场景与使用边界2.1 适合什么场景复杂任务编排需要多步推理、多次调用工具、根据中间结果决定下一步走向的任务。RAG 流程改造查库、打分、筛选、重写提示词整体流程用图来管比用 if-else 要好维护。自动化流程定时从接口拉数据、清洗、调用模型生成摘要、写回业务系统。终端智能体让 Agent 操作命令行、读写文件、调用 CI/CD 接口但必须加权限控制。多 Agent 协作子图拆分后每个 Agent 负责独立领域主图统一调度。2.2 不适合什么场景单轮问答、无状态任务用普通 Python LLM 调用即可不需要引入图框架。极低延迟场景Agent 多次串行调用 LLM延迟会叠加不适合对每次请求都要求毫秒级返回的业务。无法接受工具副作用的业务Agent 无法保证 100% 按预期调用外部系统必须有权限收口和人工审核。2.3 合规与安全边界构建 AI Agent 时必须优先确认数据合规。业务数据、用户隐私、版权内容进入 LLM 上下文之前要有脱敏、授权和访问控制机制。调用外部工具时应遵循最小权限原则工具行为要能被审计和回滚。涉及人脸、声音、版权素材等敏感能力时必须确认授权范围和合规边界不能把未经授权的数据直接送入模型或写入公开服务。测试环境使用仿真工具生产环境才开放真实系统权限。3. 环境准备与前置条件3.1 基础环境操作系统Windows / Linux / macOS 均可。Python 版本建议 3.11。LangGraph 对 Python 3.9 也可以运行但 3.11 的类型提示、异步生态和性能体验更好。包管理推荐使用uv速度比 pip 快且能直接配合 LangGraph CLI 使用。LLM 调用准备一个 OpenAI 兼容的 API Key或者部署本地模型如 Ollama、vLLM 服务保证本地网络可以访问。3.2 开发工具编辑器VS Code 或 JetBrains 系。可选Docker用于隔离 MCP 服务和数据库依赖。LangGraph Studio桌面可视化调试工具可选不是必须。3.3 磁盘与网络源码项目几百 MB 足够。如果使用本地模型按模型大小预留磁盘空间。需要访问 Python 包索引和 LangGraph 相关依赖国内环境建议配置 pip 镜像源。4. 安装部署与启动方式4.1 安装 uv 与初始化项目# 安装 uvmacOS/Linux 示例 curl -LsSf https://astral.sh/uv/install.sh | shWindows 可以下载 uv 的安装包或使用 pip 安装pip install uv初始化项目目录mkdir my-agent cd my-agent uv init4.2 安装 LangGraph 依赖uv add langgraph langchain-openai如果需要使用 LangGraph 的官方开发命令可以安装langgraph-cliuv tool install langgraph-cli或者直接在项目里添加uv add langgraph-cli[inmem]4.3 创建最小 Agent 图先写一个最简单的 LangGraph 任务验证环境能跑通。项目根目录下创建agent.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list[str] def node_a(state: AgentState) - dict: return {messages: state[messages] [node_a 执行]} def node_b(state: AgentState) - dict: return {messages: state[messages] [node_b 执行]} builder StateGraph(AgentState) builder.add_node(node_a, node_a) builder.add_node(node_b, node_b) builder.add_edge(START, node_a) builder.add_edge(node_a, node_b) builder.add_edge(node_b, END) graph builder.compile() if __name__ __main__: result graph.invoke({messages: []}) print(result[messages])启动测试python agent.py预期输出[node_a 执行, node_b 执行]这个项目能跑通说明 LangGraph 环境没有问题。4.4 LangGraph CLI 开发模式如果安装了 CLI可以直接以开发模式启动一个本地服务方便调试uv run langgraph dev服务启动后会输出本地调试地址默认在127.0.0.1:2024或127.0.0.1:8123附近实际端口以启动日志为准。CLI 模式主要用于图结构可视化和接口调试适合在开发阶段使用。注意如果uv run langgraph dev和uv run uvicorn app两套启动方式混用容易遇到端口占用和配置读取差异。CLI 内部封装了配置加载逻辑建议开发期统一用 CLI。5. LangGraph 核心概念与实战编排5.1 状态图模型LangGraph 的核心思想是 StateGraph。整个 Agent 流程被定义成一个有向图每个节点是一个处理函数每个边定义节点之间的流转关系。from langgraph.graph import StateGraph, START, END一个图至少包含State全局共享的状态通常是TypedDict或 Pydantic 模型。Node一个可调用函数输入整个 State返回 State 的增量更新。Edge节点之间的连接。Conditional Edge条件路由根据状态值决定进入哪个节点。START / END图的入口和出口。这种设计的价值在于流程里任何一步中断都能根据当前 State 恢复或重试。5.2 打造一个能调用工具的基础 Agent下面是一个更接近真实场景的示例Agent 能调用“查询天气”和“写文件”两个工具。我们用最简单的函数工具演示后续再替换成 MCP。import json from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: list def get_weather(city: str) - str: 查询城市天气返回模拟结果。 return json.dumps({city: city, weather: sunny, temperature: 26}) def write_note(content: str) - str: 写入一条笔记返回写入结果。 with open(./note.txt, a, encodingutf-8) as f: f.write(content \n) return note saved tools [get_weather, write_note] tool_node ToolNode(tools) llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) def call_model(state: AgentState) - dict: response llm_with_tools.invoke(state[messages]) return {messages: [response]} def build_agent_graph(): builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges( agent, tools_condition, { tools: tools, __end__: END, }, ) builder.add_edge(tools, agent) return builder.compile()这里的关键点是bind_tools让模型知道有哪些工具可用并输出结构化调用参数。ToolNode负责真正执行工具。tools_condition根据模型输出判断是继续调用工具还是结束。只用一个框架内置的条件判断就实现了一个完整的 ReAct 循环模型思考 → 调用工具 → 拿结果 → 继续思考。5.3 条件路由动态决定下一步条件路由是 LangGraph 最常用的能力之一。我们可以根据状态字段决定分支走向。from typing import Literal def classify_intent(state: AgentState) - Literal[weather, file, end]: last_message state[messages][-1] content last_message if isinstance(last_message, str) else last_message.content if 天气 in content: return weather if 文件 in content: return file return end builder StateGraph(AgentState) builder.add_node(weather, weather_node) builder.add_node(file, file_node) builder.add_conditional_edges(agent, classify_intent, { weather: weather, file: file, end: END, })条件路由的逻辑完全由开发者控制意味着可以自己对模型输出做一次二次判断而不是无条件相信模型选的那条路。这在工具栏和状态机设计里是核心控制点。5.4 循环检测与迭代上限Agent 的循环本质是图里出现“从某个节点又回到自己”的边。没有上限的循环会变成失控的 API 费用消耗。LangGraph 提供了recursion_limit参数result graph.invoke( {messages: [{role: user, content: 帮我查天气然后写一条笔记}]}, config{recursion_limit: 20}, )如果图的执行步数超过recursion_limit会抛出GraphRecursionError。生产环境必须设置这个上限同时在业务层记录每一步的工具调用日志方便回溯。5.5 子图与并行分支子图适合做领域拆分。比如“数据分析 Agent”可以拆成“数据读取子图”“清洗子图”“报告生成子图”子图可以单独测试、单独复用。from langgraph.graph import StateGraph, START, END # 子图 analysis_builder StateGraph(AnalysisState) analysis_builder.add_node(load, load_node) analysis_builder.add_node(clean, clean_node) analysis_builder.add_edge(START, load) analysis_builder.add_edge(load, clean) analysis_builder.add_edge(clean, END) analysis_graph analysis_builder.compile() # 主图引入子图 main_builder StateGraph(MainState) main_builder.add_node(analysis, analysis_graph)并行分支用graph.add_edge从同一个节点扩展到多个节点LangGraph 会根据依赖关系并发执行无依赖的节点。builder.add_node(task_a, task_a_node) builder.add_node(task_b, task_b_node) builder.add_node(aggregate, aggregate_node) builder.add_edge(start_node, task_a) builder.add_edge(start_node, task_b) builder.add_edge(task_a, aggregate) builder.add_edge(task_b, aggregate)实际执行时task_a和task_b会并行运行等两个都完成后进入aggregate。需要注意它们是并发执行不一定是多线程并行。CPU 密集任务要考虑线程池和 GIL 的影响IO 密集任务比如模型调用并发收益明显。6. MCP 接入与工具服务化6.1 MCP 是什么MCPModel Context Protocol是模型与外部工具系统之间的统一协议。它的价值不是让模型“变聪明”而是让工具接入方式标准化。过去每个 Agent 都要单独写一套工具调用逻辑连接数据库要写数据库工具连浏览器要写浏览器工具。有了 MCP 标准后工具作为 Server 暴露Agent 通过 Client 连接传输层走 JSON-RPC。目前很多开发工具比如 Figma、浏览器控制、ES 日志平台、本地文件系统都提供 MCP Server。好消息是LangGraph 社区已经提供 MCP 适配支持不需要自己手写协议层。6.2 用 LangGraph 连接 MCP Server先安装适配包uv add langgraph-mcp代码示例from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters( commandpython, args[./mcp_server.py], ) async def load_tools(): 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这里mcp_server.py是 MCP Server 入口文件。实际使用时要替换为自己的脚本路径和参数。6.3 写一个最小 MCP Server# mcp_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-server) mcp.tool() def get_server_time() - str: 返回当前服务器时间。 import time return time.strftime(%Y-%m-%d %H:%M:%S) mcp.tool() def read_localfile(path: str) - str: 读取本地文本文件注意控制访问范围。 allowed_dir /data/agent_files if not path.startswith(allowed_dir): return Permission denied with open(path, r, encodingutf-8) as f: return f.read() if __name__ __main__: mcp.run()这个 Server 文件有两个工具。注意read_localfile里做了路径前缀校验这是 MCP Server 必须的一步工具端不能无条件信任 Agent 传进来的参数否则任意路径读取会成为灾难。6.4 MCP 与 Agent Script 的区别社区里经常问“Agent Skill 和 MCP 有什么区别”。从实践角度看两者解决的问题不同MCP 是工具接入协议描述的是“Agent 怎么调用外部能力”。Skill 是行为模板描述的是“Agent 在特定任务里怎么组织步骤”。Skill 可以内部使用 MCP 调工具也可以完全不用工具只做提示词编排。工具协议和能力编排是两个维度不是二选一。6.5 MCP 工具注册失败排查用 MCP 接入第三方工具比如 Figma MCP时常见问题是“工具已经连接但 Agent 注册不到工具”。优先检查MCP Server 是否真正启动成功日志里有没有报错。工具名是否被模型绑定成功可以通过bind_tools后的模型输出确认。网络接口访问权限是否受限有些平台接口需要 token 认证。Server 返回值是否符合工具签名类型不匹配会导致解析失败。7. Harness 层约束 Agent 执行边界7.1 Harness 解决什么问题Agent 直接调用工具的失控风险比多数人预想的更严重。一个没有约束的 Agent 在收到模糊指令时可能连续触发几十个工具调用甚至执行高权限操作。Harness 层就是给 Agent 套上执行框架强制约束工具使用。Harness 不是某一个开源软件的名字而是一类设计模式。典型实现包括工具白名单与黑名单。调用参数校验。敏感操作二次确认。单次任务工具调用次数上限。完整调用链路日志审计。运行环境沙箱化。7.2 Python 层的 Harness 拦截示例在 LangGraph 的工具调用节点前面加一个 wrapperALLOWED_TOOLS {get_weather, get_server_time} DENIED_PARAM_PATTERN [password, token, secret] def tool_harness(tool_name: str, args: dict) - dict: if tool_name not in ALLOWED_TOOLS: raise PermissionError(ftool {tool_name} is not allowed) for key in args: if any(denied in key.lower() for denied in DENIED_PARAM_PATTERN): raise PermissionError(fparam {key} is blocked) # 记录日志 print(f[harness] calling {tool_name} with {args}) return {allowed: True, args: args}然后重写ToolNode的调用逻辑from langgraph.prebuilt import ToolNode class GuardedToolNode(ToolNode): def _func(self, name, args): tool_harness(name, args) return super()._func(name, args)实际实现不用和上面的代码一致但关键点是所有工具调用必须经过一个统一闸口不能直接让模型输出参数去执行原生函数。7.3 Harness 工程化的三个层面Harness 应该在三个层面落地参数层校验工具参数类型、长度、范围过滤危险字段。权限层按用户维度区分工具白名单比如普通用户不能触发写生产库的工具。审计层每个工具调用都记录事件包括时间、用户、Agent 任务 ID、参数、结果状态。如果只是写 demo参数层就够了。生产环境一定要补审计层否则出了事故看不到原因。8. 安全架构设计要点8.1 认证与授权Agent 服务必须明确“谁在调用、能调谁、能调用哪些工具”。最简单的实现是外部请求必带 API Token 或 OAuth 鉴权。不同角色用户映射到不同的工具白名单。每个任务的 Agent 上下文里带上用户身份。8.2 输入与输出过滤输入侧对用户消息做长度限制、内容检查防止恶意指令注入。输出侧避免把内部工具信息、系统路径、密钥直接暴露给用户。LLM 的输出不能直接作为 SQL、Shell 命令或文件路径执行必须经过程序化校验。8.3 网络安全服务监听地址默认只绑127.0.0.1需要远程访问时再考虑网关代理和 HTTPS。MCP Server 如果通过网络暴露必须做认证本地 MCP 建议走 stdio 模式减少攻击面。日志系统不要打印完整密钥和敏感数据。8.4 数据隔离每个用户的对话上下文和工具文件建议隔离目录。跨用户的数据访问必须经过授权服务校验不能依赖模型自身判断。涉及日志分析、ES 查询等场景Agent 的查询权限应遵循最小权限原则。9. 接口 API 与批量任务落地9.1 服务化部署LangGraph 图是纯 Python 对象暴露成服务的方式有很多最简单的是用 FastAPI 包一层from fastapi import FastAPI from pydantic import BaseModel from agent import build_agent_graph app FastAPI() graph build_agent_graph() class ChatRequest(BaseModel): message: str user_id: str class ChatResponse(BaseModel): result: str app.post(/chat, response_modelChatResponse) def chat_endpoint(req: ChatRequest): result graph.invoke( { messages: [ {role: system, content: 你是业务助手}, {role: user, content: req.message}, ] }, config{recursion_limit: 15}, ) final_message result[messages][-1] content final_message.content if hasattr(final_message, content) else str(final_message) return ChatResponse(resultcontent)启动命令uvicorn app:app --host 127.0.0.1 --port 80109.2 curl 调用接口curl -X POST http://127.0.0.1:8010/chat \ -H Content-Type: application/json \ -d {message: 帮我查一下北京天气并写笔记, user_id: test_user}返回预期{ result: 已查询北京天气晴26℃并保存为笔记。 }9.3 Python 调用接口import requests url http://127.0.0.1:8010/chat payload { message: 帮我查一下北京天气并写笔记, user_id: test_user, } response requests.post(url, jsonpayload, timeout60) print(response.json())9.4 批量任务设计Agent 接口天然适合异步批量任务。推荐的队列模型输入文件/列表 → 任务入队 → Worker 逐个调用 Agent 图 → 结果落盘 → 失败重试实现批量任务时要注意给每次任务分配唯一task_id。所有中间状态写入独立文件或数据库避免进程重启丢任务。对 LLM 调用做失败重试但重试只针对超时和临时网络错误不要重试参数非法类错误。批量任务的并发数要按 LLM 服务 QPS 上限设定防止被打爆。10. 资源占用与性能观察AI Agent 的核心资源消耗来自 LLM 推理而不是 LangGraph 框架本身。10.1 观察哪些指标GPU 或 API 服务侧查看单次请求延迟、Tokens 消耗。本地 Python 进程观察 CPU、内存占用。工具侧外部服务调用次数、响应时间。图执行步数每个任务实际跑了多少节点和工具调用。10.2 性能调优方向减少 LLM 调用能用规则判断的节点就写规则不调模型。压缩上下文不是所有历史消息都要喂给模型可以用摘要节点替代长历史。缓存工具结果相同参数的查询结果可以缓存降低外部服务压力。提高并行度无依赖的节点放在并行分支里例如同时查询多个数据源。10.3 显存与推理资源如果使用本地模型显存消耗取决于模型参数和量化方式。开发调试图结构阶段建议先用远程 API 模型跑通流程后再换本地模型这样可以避免把流程问题和推理资源问题混在一起排查。11. 常见问题与排查方法问题现象可能原因排查方式解决方案GraphRecursionError循环没有终止条件或超过迭代上限检查条件路由逻辑查看执行日志增大recursion_limit或修复循环条件模型没有调用工具没有bind_tools或工具描述不清晰打印模型返回内容确认 tool_calls 字段正确绑定工具优化工具描述MCP 工具注册不上Server 未启动、token 失效、协议不匹配检查 MCP Server 日志单独测试工具调用修复 Server确认版本兼容性端口被占用开发服务或旧进程残留查端口占用换端口或杀旧进程API 调用超时LLM 响应慢或工具卡住查看任务日志和工具日志增加超时时间为工具加超时机制并行节点不生效节点间存在隐性依赖或共享可变量检查节点函数是否修改同一对象让节点函数返回新值避免共享内存直接修改Agent 执行了危险操作缺少 Harness 拦截检查工具调用日志增加白名单和参数校验12. 最佳实践与使用建议12.1 图设计要克制不要试图把所有业务逻辑都塞进一个巨大的图里。优先用子图拆分领域主图只做路由和编排。节点函数保持幂等同一个状态进来输出要可预期。12.2 工具要可观测所有工具函数都要有日志。至少记录调用时间。参数摘要敏感字段脱敏。执行耗时。返回状态。没有日志的 Agent出了问题只能靠猜。12.3 用评估驱动迭代不要只看一两个例子就认为 Agent 可用。准备一套评测集合包含常规问题。边界问题超长输入、空输入、特殊字符。工具调用错误场景。敏感输入。每次修改图结构或工具逻辑后跑一遍评测集观察通过率变化。12.4 先最小闭环再扩展工具第一个版本只保留两个工具跑通“模型 → 工具 → 状态更新 → 结束”的闭环。加工具时一个个加每次加完都重新评测避免一次接入多个工具后无法定位问题。12.5 安全是产品问题不是技术问题安全架构能不能落地取决于产品是否愿意为“二次确认”和“最小权限”付出体验成本。开发者在设计阶段就要把权限模型定清楚不要在 Agent 上线后才发现工具白名单没有做。13. 总结与下一步这次我们把 AI Agent 从零到工程化拆了一遍LangGraph 负责图编排MCP 统一工具接入Harness 约束执行边界安全架构覆盖认证、权限、审计和数据隔离。从材料看LangGraph 生态已经覆盖了从单机脚本到服务化部署的完整链路和 LangChain 相比它的优势在于图结构和状态管理更清晰适合复杂任务。最容易踩的坑有三个MCP 工具注册不上、循环没有上限、工具调用没有权限拦截。建议第一次动手时先做最小闭环只接一个 MCP 工具加一个简单 Harness跑通后逐步扩展。后续可以继续深入的方向多 Agent 协作的子图调度、LLM 调用结果缓存、基于评测集自动回归、以及把 Harness 从工具拦截升级到全链路审计平台。整个方案值得持续迭代建议收藏备用。