很多同学现在缺的不是一个大模型 API而是缺一套能把模型转成“能干活”的系统能力。这次我们不写玩具 Demo直接从零拆一个 AI Agent安全架构、Harness、LangGraph、MCP、底层执行链路全部串起来。你会看到 Agent 由哪些模块组成为什么 LangGraph 适合做编排MCP 怎么把外部工具接进来Harness 如何给模型套上“安全围栏”最后怎么把整个 Agent 封装成 API 和批量任务。这篇文章不是某个具体开源项目的安装教程而是一条“造轮子”的实战路线。核心特点有三点第一偏工程落地重点讲架构和安全边界第二代码示例围绕 LangGraph MCP 展开但不是照抄文档第三没有固定测试机配置所以不写“实测显存占用 7G”这种没有依据的数字资源占用怎么观察、怎么调参会给你通用方法。适合的读者是后端工程师、算法工程师、以及被老板一句话要求“搞个 Agent 接入业务”的同学。看完之后你能掌握一套最小可用架构并且知道最容易踩的坑在哪里。1. 技术全景与核心概念速览先把概念对齐。AI Agent 不是一个魔法模型而是一套系统模型负责推理编排层负责控制流程工具层负责执行动作安全层负责限制边界。下面这张表对应了本次实战的核心技术项。技术项作用核心关注点常见门槛LLM 推理服务理解用户意图、生成计划和回复上下文长度、推理成本、响应延迟显存或 API 成本LangGraph编排 Agent 状态机维护节点、边、循环、子图状态结构、条件路由、循环终止条件需要理解图执行模型MCP统一工具接入协议让模型调用外部工具工具定义、传输方式、认证方式工具注册和参数校验Harness约束模型执行过程的沙箱/框架层工具白名单、文件系统限制、输入输出过滤安全实践往往被忽略安全架构加密、鉴权、审计、数据隔离API Key 管理、提示注入防护、权限最小化缺少全局视角从材料信息看LangGraph 和 MCP 是当前 AI Agent 方向的搜索热点同时 Harness、安全架构也开始被频繁提到。这不是偶然模型能力越强越需要流程控制和执行边界。先把这两个点想清楚再谈“Agent 能自动写代码、能搜索网页、能操作数据库”。2. Agent 系统到底由哪几部分组成一个可用的 Agent 系统至少包括五层。第一层是模型层。无论是调用云端大模型还是本地部署开源模型都需要一个可重复调用的推理入口。模型层只负责文本生成不负责决定最终动作。第二层是记忆层。Agent 需要短期记忆当前对话上下文和长期记忆向量库、业务数据库。没有记忆模型每次都是“陌生人”。第三层是规划层。规划层决定下一步做什么。最简单的是 ReAct 模式思考 - 行动 - 观察 - 再思考。LangGraph 这一类编排框架本质上是把 ReAct 变成了一张可以控制的状态图。第四层是工具层。工具层让模型能够真正影响外部世界查数据库、调 API、写文件、发邮件。MCP 的价值在于把工具调用标准化不用每个工具都写一套私有协议。第五层是安全与治理层。这一层把模型限制在“允许的范围内”哪些工具可以调用、哪些目录可以写、输入输出是否包含敏感信息、所有执行是否有审计日志。所以不要一上来就写调用链prompt - LLM - output。工程化的第一步是画出数据的流向用户输入 - 安全过滤 - 规划器 - 工具调用 - 结果回收 - 再次规划 - 最终输出。后面的 LangGraph、MCP、Harness 都是为了让这条链路可控。3. Harness把模型跑进“安全围栏”的执行框架“Harness” 这个词在最近 AI coding agent 方向出现频率很高。简单说Harness 是包在模型外层的执行框架负责约束模型运行环境。模型不是一个能随意接触网络和文件系统的进程而是跑在一个由 Harness 控制的沙箱里。一个 Harness 通常包含四个能力工具白名单只暴露当前任务需要的工具其他工具一律不可见。文件系统隔离给任务一个临时目录模型只能读写这个目录不能碰宿主机的系统文件。执行动作审计模型发起的每一步工具调用都记录日志方便回溯。循环终止控制Agent 不能无限“思考”下去达到最大步数或满足终止条件必须停止。下面是一个简化版的 Harness 伪代码演示了“模型请求 - 工具调用 - 返回结果”的流程。class Harness: def __init__(self, allowed_tools: list[str], workspace: str): self.allowed_tools set(allowed_tools) self.workspace workspace self.step_count 0 self.max_steps 10 def execute(self, llm, user_input: str) - str: messages [{role: user, content: user_input}] while self.step_count self.max_steps: response llm.chat(messages) action self._parse_action(response) if action is None: return response.content if action[tool] not in self.allowed_tools: raise PermissionError(ftool not allowed: {action[tool]}) result self._safe_call(action[tool], action[args]) messages.append({role: tool, content: result}) self.step_count 1 return max steps reached def _safe_call(self, tool: str, args: dict): # 这里只能访问 workspace 内的路径外部路径一律拒绝 if tool read_file: path args.get(path, ) if not path.startswith(self.workspace): raise PermissionError(path outside workspace) # 实际工具执行逻辑省略 return ftool {tool} executed看到重点了吗Harness 不是“调模型的工具”而是“限制模型动作的工具”。实际生产环境里Harness 往往还会包含网络代理层、依赖安装白名单、命令执行超时和资源限制。只要模型还能直接执行任意系统命令这个 Agent 就还不适合上线。4. LangGraph 实战状态图、条件路由与子图编排4.1 为什么用 LangGraph 而不是 LangChainLangChain 的定位是“模型应用的开发框架”提供了很多封装好的链式调用。LangGraph 则是建立在 LangChain 之上的图执行框架核心是 StateGraph状态在节点之间传递节点是函数边是状态转移。和“一条链走到底”的方式相比LangGraph 的优势是可控制性更强。它支持条件路由、循环、子图和并行分支。这正好匹配 Agent 的典型场景模型先决定一个动作执行完看结果决定继续还是结束。4.2 一个最简 Agent 图下面这个示例定义了一个带工具调用的 Agent 图。节点分别是agent和tools条件边根据模型输出决定跳转到tools还是结束。from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langgraph.graph.state import CompiledStateGraph class AgentState(TypedDict): messages: list[dict] need_tool: bool final_answer: str def agent_node(state: AgentState) - dict: # 这里调用 LLM根据 messages 判断是否需要工具 # 实际项目里替换为自己的模型调用 response llm.invoke(state[messages]) if response.tool_calls: return {messages: state[messages] [response], need_tool: True} return {messages: state[messages] [response], final_answer: response.content} def tools_node(state: AgentState) - dict: # 执行模型请求的工具把结果追加到 messages tool_results run_tools(state[messages][-1].tool_calls) return {messages: state[messages] tool_results, need_tool: False} def route(state: AgentState) - Literal[tools, end]: return tools if state[need_tool] else end graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(agent, tools, conditionroute) graph.add_edge(tools, agent) graph.add_edge(agent, END, conditionroute) app: CompiledStateGraph graph.compile()这个例子虽然简单但已经包含三个关键点状态对象AgentState是整个图的共享内存。条件边让流程可以“循环”这也是 Agent 与普通链式调用的根本区别。终止条件必须显式给出否则图会一直循环。4.3 条件路由、循环检测与子图真实业务里条件路由不只有“要不要调用工具”这一种。你可能会遇到根据意图走不同分支查询天气走天气工具查数据库走 SQL 工具。循环检测一旦某条路径走了 N 次都没有结束强制终止。子图将“多轮工具调用”封装成一张子图主图只关心子图的最终输出。并行分支同时查两个独立源再把结果合并。LangGraph 对这几类场景都有对应原语。实现时要注意的一点是在图中增加状态判断字段之前先想清楚所有节点可能需要的公共状态。状态字段一旦定好后续改动会让图结构变得混乱。如果你要读源码重点关注StateGraph的add_conditional_edges和CompiledStateGraph的invoke实现。理解图如何遍历、如何传递状态比背 API 重要得多。5. MCP 实战把外部工具接入 Agent5.1 MCP 协议解决了什么问题MCPModel Context Protocol是一个开放协议目标是把“模型需要的外部数据/工具”标准化。传统做法是给 Agent 写一堆函数调用每个工具都有自己的参数格式。MCP 出现之后工具以统一接口暴露给模型一组工具定义、一个传输通道、一套请求响应格式。很多人会问 MCP 和 Agent Skill 有什么区别。一个粗略的区分是MCP 主要解决“工具怎么被调用”的协议问题Skill 更偏向“模型在特定场景下的经验/提示词/工具组合”。实际项目中两者可以同时存在不建议对立。常见的 MCP Server 包括文件系统操作数据库查询Playwright / 浏览器自动化Figma 设计稿读取Elasticsearch API 日志分析。5.2 MCP Server 配置示例很多 MCP 客户端支持通过.mcp文件声明服务器。下面是一个通用配置结构实际命令和路径需要按你选用的包替换。{ mcpServers: { filesystem: { command: npx, args: [-y, modelcontextprotocol/server-filesystem, /tmp/workspace], env: {} }, database: { command: python, args: [-m, my_mcp_db_server], env: { DB_CONN: postgresql://user:passlocalhost:5432/db } } } }重点看两个字段command指定启动方式env注入敏感配置。不要把数据库密码直接写进args否则在进程列表里就泄露了。5.3 Python 端调用 MCP 的通用思路在 Agent 代码里通常通过 MCP Client 连接 Server拿到工具列表然后再把工具执行结果交给 LLM。下面的示例是通用连接逻辑具体 SDK 包名和初始化方式以官方文档为准。import anyio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def load_tools_from_mcp(): server_params StdioServerParameters( commandnpx, args[-y, modelcontextprotocol/server-filesystem, /tmp/workspace], envNone, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() return tools def run_mcp_tool_pipeline(): tools anyio.run(load_tools_from_mcp) for tool in tools: print(tool.name, tool.description)真实项目中你会把list_tools返回的工具定义转换成 LangGraph 节点的输入模型决定调用哪一个tool然后通过 MCP Client 执行call_tool。这里有个容易踩的坑工具描述写得太简单模型不会主动调用。每个工具的描述里要写清楚“什么时候用、参数是什么、返回什么”。5.4 工具注册失败的排查思路搜索热词里经常出现“Figma MCP 在 Codex 中总是工具注册不上”。这类问题通常不是 MCP 协议的问题而是本地环境问题。排查顺序是先单独运行 MCP Server 命令确认能正常启动。再检查配置里command是否能被 MCP 客户端找到尤其 Node 环境用 npx 时需要设置 PATH。然后看客户端日志里工具是否已经加载出来。最后确认工具描述和参数 JSON Schema 是否符合协议要求。6. 安全架构落地加密、鉴权、沙箱与数据隔离安全架构不是上线前再补的“检查项”而是 Agent 系统的地基。下面从四个方向展开。6.1 传输与存储加密模型输入输出、工具调用结果、日志会包含大量业务数据。传输层至少启用 TLS。存储层涉及敏感样本时优先使用国密算法或行业通用加密算法做落盘加密。需要强调的是算法选型要结合所在行业的合规要求不要自己发明“加密方案”。6.2 鉴权与最小权限Agent 服务必须校验调用者身份。内部服务可以用 Token 或 mTLS对外服务建议走统一 API 网关。关键原则是“最小权限”Agent 调用数据库时不要直接复用运维账号Agent 操作文件时只给它一个空目录Agent 调用第三方 API 时用独立 Key 并设置额度上限。生产实践里可以把模型能使用的工具账号单独创建只授予业务需要的表或接口权限。这样即使提示词注入成功攻击面也是有限的。6.3 提示注入与输出过滤提示注入是 Agent 特有的安全问题。攻击者可能通过网页内容、文档、外部接口返回值诱导模型执行非预期操作。对策有几类对模型输入进行来源标记区分用户指令和外部内容。工具参数的取值范围做白名单校验。模型输出在返回用户前过滤掉内部 IP、密钥、手机号等敏感信息。高风险动作要求二次确认比如删除操作、转账操作。6.4 沙箱与审计前面说的 Harness 在安全架构里的位置就是执行沙箱。凡是模型可以影响的系统资源包括文件、进程、网络都应该在沙箱里执行。容器是最简单的一种方式加上只读根文件系统、无特权模式、资源限制基本能满足大部分场景。审计日志必须完整记录“谁在什么时间、用什么工具、传了什么参数、得到了什么结果”。这里要特别提醒日志本身可能包含敏感数据存储日志前要做脱敏。7. 从底层源码视角看 Agent 执行链路很多同学用 LangGraph 只是调invoke对内部发生了什么没有感知。下面以自研简化版执行循环为例把源码层面的关键路径拆开。def run_agent(llm, tools, user_input, max_rounds5): state { messages: [{role: user, content: user_input}], rounds: 0, trace: [], } while state[rounds] max_rounds: # 1. 模型根据当前状态生成回复 reply llm.chat(messagesstate[messages]) # 2. 解析模型输出的工具调用 action parse_tool_call(reply) if action is None: return reply[content] # 3. 在 Harness 里执行工具 result harness.execute(action) # 4. 把工具结果追加回状态 state[messages].append({role: tool, content: result}) state[rounds] 1 state[trace].append({action: action, result: result}) raise RuntimeError(agent exceeded max rounds)这段代码把 Agent 的本质说清楚了一个“循环思考-行动-观察”的状态机。LangGraph 做的事情是把while循环改造成显式图让路由、分支、并发都变得可控。所以你看 LangGraph 源码时不要被 API 名称绕晕抓住三个核心概念状态如何被传递和更新图节点何时执行条件边如何决定下一步。看懂了这三条底层源码的大部分代码都只是在处理这三种机制的边界情况并发、重试、持久化、时间旅行。8. 服务化接口 API 与批量任务Agent 不是脚本要能被外部系统调用。最直接的方式是封装成 HTTP API同时把“单个请求”和“批量任务”分开设计。8.1 用 FastAPI 封装 Agent下面是一个调用 LangGraph 编译结果的示例接口。实际项目中需要把run_agent替换成你自己的调用逻辑并增加鉴权、限流和超时控制。from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class AgentRequest(BaseModel): user_input: str session_id: str default class AgentResponse(BaseModel): answer: str trace: list app.post(/agent/invoke, response_modelAgentResponse) async def invoke_agent(req: AgentRequest): try: result run_agent(req.user_input, req.session_id) return AgentResponse(answerresult[answer], traceresult[trace]) except Exception as exc: raise HTTPException(status_code500, detailstr(exc))启动命令通用模板如下实际端口和参数按项目调整。uvicorn app:app --host 0.0.0.0 --port 80008.2 批量任务处理批量任务和在线请求不同不能因为单条任务耗时几十秒就把用户请求全部阻塞。通用做法是提交任务后立刻返回任务 ID后台 Worker 消费队列前端轮询状态。import uuid from celery import Celery celery_app Celery(agent_tasks, brokerredis://localhost:6379/0) celery_app.task def agent_batch_task(user_input: str): result run_agent(user_input) return {status: done, result: result} def submit_agent_task(user_input: str): task_id str(uuid.uuid4()) agent_batch_task.apply_async(args[user_input], task_idtask_id) return task_id批量任务的失败重试也很关键。建议每个任务记录三个阶段提交、执行、完成。执行阶段失败要留存堆栈和输入快照便于复现。不要一失败就重试无限次否则下游系统会被打爆。8.3 curl 调用示例curl -X POST http://127.0.0.1:8000/agent/invoke \ -H Content-Type: application/json \ -d {user_input: 查询今天北京天气, session_id: test-001}预期返回一个 JSON包含最终回答和调用轨迹。判断接口是否成功的标准不是 HTTP 200而是trace里的工具调用是否符合预期。如果模型没有调用该调用的工具优先检查工具描述和 MCP 注册状态。9. 资源占用与性能观察AI Agent 的资源消耗比单次模型推理更复杂因为它可能包含多轮模型调用、工具调用、上下文拼接。9.1 显存和 CPU 占用如果使用本地大模型显存占用主要取决于模型参数量、量化位数和上下文长度。同样一个模型8K 上下文和 32K 上下文的显存差异很大。最稳妥的方法是在推理服务端开启指标监控观察显存、GPU 利用率、请求延迟三个指标。不要凭印象“估”要跑一次多轮任务记录曲线。9.2 降低资源占用的通用手段减少上下文每轮只保留关键消息及时压缩历史。控制最大步数Agent 循环步数越高耗时越长。使用流式输出首字延迟降低用户体验更好。工具结果截断数据库返回几千行时只保留前 N 条。批量任务并发限制防止一次性发出太多 LLM 请求导致限流。9.3 延迟观测Agent 的性能瓶颈通常在外部工具调用上SQL 查得慢、网页打不开、MCP Server 启动慢。建议整套链路接入 Tracing记录每一步耗时。否则生产环境出现一次“Agent 卡死”你不知道是模型没返回还是工具没响应。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 不调用任何工具工具定义不清晰或模型未收到工具描述查看请求日志中 tools 参数优化工具 description给出使用示例工具调用后报参数错误MCP 工具参数 Schema 与模型输出不匹配检查 MCP Server 返回的 JSON Schema修正参数类型严格校验必填字段同一流程反复循环不结束终止条件缺失或判断字段状态未更新查看 tracing 中循环路径增加最大步数更新状态字段本地 MCP Server 启动失败缺少依赖或 PATH 未配置单独执行 command 测试安装依赖、配置环境变量API 调用超时Agent 多轮推理耗时过长查看接口调用耗时曲线开启流式输出增加超时时间限制最大步数批量任务大量失败工具调用限流或数据异常查看 Worker 日志和任务重试次数增加退避重试失败任务隔离显存不足上下文过长或并发过高观察推理服务指标压缩上下文降低并发减小 batch size输出包含敏感信息缺少输出过滤检查最终输出与日志部署敏感信息脱敏服务11. 最佳实践与下一步最后给几条实战经验。第一条先把最小链路跑通。不要一开始就接十个 MCP Server先让“用户输入 - LLM - 一个工具 - 输出”成立再加复杂路由。第二条安全边界前置设计。Agent 能访问什么、不能访问什么在画图阶段就定清楚。如果模型能调用数据库那数据库账号的权限一定要单独收窄。第三条审计日志不是可选项。把每一次工具调用记录下来既能排查问题也能在发生安全事故时快速定位影响范围。日志中涉及敏感字段务必脱敏。第四条不要盲目追新。LangGraph、MCP、Harness 这些概念确实有用但最终目标还是业务价值。如果一个场景用普通链式调用就能解决不需要强行套复杂的 Agent 编排。下一阶段你可以继续探索三个方向一是给 LangGraph 加持久化让 Agent 能在多轮会话间恢复状态二是把 Harness 从文件沙箱扩展到网络策略控制模型能不能访问特定域名三是沉淀统一工具接入规范把 MCP Server 做成内部平台让业务团队自助接入工具。这套路线走完你已经不是“套 API 生成文本”而是真正在造一个可控、可审计、可上线的 AI Agent 系统。建议把文中的最小架构先落到本地跑通再逐步扩展。