ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

AI Agent核心原理:状态机、工具协议与记忆机制深度解析

2026/9/20 13:27:29 拓冰建站 浏览量
AI Agent核心原理:状态机、工具协议与记忆机制深度解析 1. 为什么“从零开始深入理解 AI Agent”不是一句空话——先拆掉三个最危险的认知地雷AI Agent 这个词最近半年在技术社区里像被扔进沸油的水滴滋滋作响、四处飞溅。你刷到的每第三条技术推文标题里都带着“Agent”招聘JD里“具备AI Agent开发经验”已从加分项悄悄挪到了“必须项”甚至非技术岗的周报里也开始出现“用Agent自动整理会议纪要”的字眼。但问题来了当所有人嘴上都在说Agent真正能画出一个带记忆、能调工具、会自我反思的Agent工作流草图的人可能连会议室投影仪的HDMI线都还没插稳。我去年带过两个真实项目一个是给某制造业客户做设备故障预判Agent另一个是帮教育公司搭课程内容生成Agent。两个项目启动前团队里90%的人对Agent的理解还停留在“就是让大模型多跑几轮prompt”。结果呢第一个项目卡在工具调用链路断裂上整整三周因为没人意识到LangChain的ToolExecutor默认不支持异步IO第二个项目上线后召回率暴跌复盘发现根本原因是把“记忆”简单等同于往prompt里塞历史对话完全忽略了向量数据库的chunk策略对上下文连贯性的致命影响。这些坑和代码语法无关和算力无关纯粹是认知错位导致的系统性返工。所以“从零开始深入理解”首先得把脑子里那几个根深蒂固的“常识”彻底炸掉。第一颗地雷叫“Agent 大模型 Prompt工程”。错。Prompt只是Agent的皮肤真正的骨架是状态机State Machine——它决定Agent在收到用户输入后是该查知识库、还是该调API、或是该自我反思上一轮决策是否合理。第二颗地雷是“LangChain/ LangGraph 就是Agent框架”。错。它们是工具箱不是蓝图。LangChain像一套万能螺丝刀LangGraph像一张电路布线图但你要造的是一台能自己换电池、能根据路况调整路线的智能车光有螺丝刀和布线图远远不够。第三颗地雷最隐蔽“MCP协议就是Agent通信标准”。错。MCPModel Communication Protocol目前连RFC草案都算不上它更像一群工程师在GitHub Issues里吵出来的临时共识核心价值在于定义了“工具描述怎么写”“调用结果怎么结构化返回”而不是规定Agent之间如何握手、如何建立信任链。把它当成OSI七层模型来学只会越学越晕。提示别急着装库、写代码。先拿出一张白纸画一个圆圈代表“用户输入”再画一个圆圈代表“最终输出”中间用箭头连接。现在在这条箭头上至少标出5个你认为Agent内部必须发生的、不可跳过的环节。如果标不出或者标出来的全是“调用LLM”“解析JSON”那说明你的认知地雷还没拆干净。这个练习比跑通第一个Hello World Demo重要十倍。这三颗地雷不拆后面所有实操都会变成在流沙上盖楼。接下来我们不再讲“是什么”而是直接进入“怎么动起来”——从一个最简但真实的Agent原型开始亲手把它从概念变成可执行的Python进程过程中每一个函数、每一行配置都对应着上面提到的状态机、工具链、协议层的真实需求。2. 构建第一个可运行Agent用纯Python手写状态机绕过所有框架幻觉市面上90%的Agent入门教程第一行代码就是pip install langchain。这就像教人修车第一课就发一把扳手却不告诉你发动机气缸有几个、活塞怎么运动。我们要做的是先用最原始的Python把Agent的“心跳”模拟出来——一个能接收输入、判断下一步动作、执行动作、更新内部状态、再等待下一次输入的闭环。这个过程不依赖任何第三方库只用Python内置的dataclass、enum和基础IO。2.1 定义Agent的“生命体征”状态与动作的最小集合Agent不是魔法它是一个有明确生命周期的程序实体。它的核心状态只有四个字段current_input: 当前收到的用户指令字符串memory: 一个列表存储过往交互的关键摘要不是原始对话而是“用户昨天问过设备A的维修手册已获取PDF”这类结构化事实tools: 一个字典键是工具名如search_manual值是该工具的可调用对象函数或类实例state: 一个枚举值表示当前所处阶段WAITING,PLANNING,EXECUTING,REFLECTINGfrom enum import Enum from dataclasses import dataclass from typing import List, Dict, Callable, Any, Optional class AgentState(Enum): WAITING waiting # 等待用户输入 PLANNING planning # 分析输入决定下一步行动 EXECUTING executing # 调用工具或LLM REFLECTING reflecting # 评估执行结果更新记忆 dataclass class SimpleAgent: current_input: str memory: List[str] None tools: Dict[str, Callable] None state: AgentState AgentState.WAITING def __post_init__(self): if self.memory is None: self.memory [] if self.tools is None: self.tools {}看到这里你可能会想“这不就是个空壳吗” 对这正是关键。框架如LangChain最大的陷阱就是用一层华丽的API包装让你误以为agent.invoke()这个函数调用本身就有智能。而手写这个空壳逼你直面最本质的问题状态如何流转谁来触发流转流转的条件是什么比如state从WAITING变成PLANNING触发条件必须是current_input非空从PLANNING变成EXECUTING触发条件必须是规划模块返回了一个有效的工具调用指令。这些条件不是框架自动给你而是你必须在代码里用if/else明确定义。2.2 实现“思考”环节一个极简但真实的PlannerPlanner是Agent的大脑皮层负责把模糊的用户输入翻译成精确的机器指令。很多初学者以为Planner必须用大模型这是巨大误区。在工业场景中80%的Planner逻辑是硬编码规则轻量级模型。我们先实现一个基于关键词匹配的Planner它足够简单却暴露了所有核心设计点def simple_planner(self) - Dict[str, Any]: 基于关键词的极简Planner返回工具调用指令 input_lower self.current_input.lower() # 规则1用户提到手册、说明书、PDF触发手册搜索 if any(word in input_lower for word in [手册, 说明书, pdf, 文档]): return { tool: search_manual, args: {query: self.current_input}, reason: 用户明确请求查找设备操作手册 } # 规则2用户提到温度、压力、报警触发实时数据查询 if any(word in input_lower for word in [温度, 压力, 报警, 异常]): return { tool: query_sensor_data, args: {device_id: self._extract_device_id()}, reason: 用户关注设备实时运行参数 } # 默认fallback交给LLM做通用问答 return { tool: llm_fallback, args: {prompt: self.current_input}, reason: 输入未匹配预设规则启用通用大模型响应 } # 注意_extract_device_id() 是一个虚构的辅助方法实际中可能从memory里提取 # 或从input中用正则匹配比如设备A的温度 - device_idA这个Planner的价值不在于它多聪明而在于它强制你思考三个问题第一Planner的输入是什么是原始用户输入还是经过清洗、标注后的版本第二Planner的输出格式必须严格定义因为后续的Executor要靠这个格式去调用工具。这里我们约定输出必须是{tool: ..., args: {...}, reason: ...}其中reason字段不是给人看的而是给后续的Reflector模块做自我评估用的。第三Planner必须有明确的fallback机制。没有哪个Planner能覆盖100%的case硬编码规则必然有盲区此时必须优雅降级而不是让整个Agent崩溃。2.3 执行器Executor让工具调用成为可验证的原子操作Executor是Agent的手和脚它把Planner的指令变成实实在在的函数调用。这里最容易犯的错误是把Executor写成一个黑盒——“调用工具返回结果”。但真实世界里工具调用失败是常态。我们的Executor必须包含重试、超时、错误分类三大能力import time import json from typing import Dict, Any def execute_tool(self, plan: Dict[str, Any]) - Dict[str, Any]: 执行Planner返回的工具调用指令带重试和错误处理 tool_name plan[tool] tool_args plan[args] # 检查工具是否存在 if tool_name not in self.tools: return { status: error, message: f工具 {tool_name} 未注册, raw_result: None } tool_func self.tools[tool_name] # 设置重试参数工业场景常见网络抖动、服务短暂不可用 max_retries 3 base_delay 1.0 # 秒 for attempt in range(max_retries): try: # 所有工具调用都加超时保护防止阻塞整个Agent result tool_func(**tool_args) return { status: success, message: 工具执行成功, raw_result: result, attempt: attempt 1 } except TimeoutError: if attempt max_retries - 1: return { status: error, message: f工具 {tool_name} 超时已重试{max_retries}次, raw_result: None } time.sleep(base_delay * (2 ** attempt)) # 指数退避 except Exception as e: # 关键捕获具体异常类型便于后续Reflector分析 error_type type(e).__name__ return { status: error, message: f工具 {tool_name} 执行异常: {error_type} - {str(e)}, raw_result: None, error_type: error_type } # 理论上不会走到这里但保险起见 return {status: error, message: 未知错误, raw_result: None}这段代码里藏着工业级Agent的生存法则。第一超时Timeout是必须的。一个HTTP请求卡死30秒整个Agent就瘫痪了。第二重试策略必须是指数退避Exponential Backoff而不是固定间隔。因为网络抖动通常是瞬时的第一次失败后立刻重试大概率再次失败等1秒再试成功率就高很多等2秒再试成功率更高。第三错误分类比错误信息更重要。ConnectionError和ValueError的处理方式天差地别前者应该重试后者应该立刻终止流程并提示用户修正输入。这个分类信息会直接喂给后面的Reflector模块。2.4 让Agent学会“记事”和“反思”Memory与Reflector的协同设计很多教程把Memory简单等同于“把对话历史存进Redis”。这是对Agent记忆机制的根本性误解。真正的Memory是为未来决策服务的结构化知识沉淀。而Reflector则是Agent的“元认知”能力——它不关心具体任务只关心“这次决策对不对”。我们设计一个极简但有效的Memory模块它只存储两类信息事实型记忆Fact Memory从工具调用结果中提取的、可复用的客观事实。例如search_manual(设备A)返回了手册PDF的URLMemory就存{device: A, manual_url: https://...}。经验型记忆Experience Memory对某类问题的处理模式总结。例如query_sensor_data(A)连续三次超时Reflector就会生成一条经验“设备A的传感器API不稳定下次优先尝试缓存数据”。def update_memory(self, plan: Dict[str, Any], execution_result: Dict[str, Any]): 根据Planner计划和Executor结果更新Memory if execution_result[status] success: # 从成功结果中提取结构化事实 if plan[tool] search_manual: # 假设result是{url: ..., title: ...} manual_info execution_result[raw_result] fact f设备{self._extract_device_id()}的操作手册位于{manual_info.get(url, 未知)} self.memory.append(fact) elif plan[tool] query_sensor_data: sensor_data execution_result[raw_result] # 存储最新读数用于后续趋势判断 self.memory.append(f设备{self._extract_device_id()}最新温度:{sensor_data.get(temp, N/A)}°C) # 经验记忆由Reflector生成此处只预留接口 pass def reflect(self, plan: Dict[str, Any], execution_result: Dict[str, Any]) - str: Reflector模块评估本次执行效果生成改进指令 if execution_result[status] error: error_type execution_result.get(error_type, Unknown) # 针对不同错误类型生成不同反思 if error_type ConnectionError: return 网络连接失败下次尝试前增加重试次数 elif error_type ValueError: return f工具参数错误{execution_result[message]}需检查用户输入是否包含必要设备ID else: return f未知错误{execution_result[message]}启用人工审核流程 # 成功情况下的反思检查结果是否满足Planner的预期 if reason in plan and 温度 in plan[reason]: # 如果Planner说要查温度但结果里没温度字段说明工具返回格式异常 if not isinstance(execution_result[raw_result], dict) or temp not in execution_result[raw_result]: return 工具返回数据格式不符合预期需联系工具提供方修正Schema return 执行符合预期无须调整看到这里你应该能感受到一个真正可用的Agent其复杂度不在“调用大模型”这个动作本身而在状态管理、错误恢复、记忆沉淀、自我校准这一整套闭环机制。LangChain的AgentExecutor之所以常被诟病“不透明”正是因为它的内部状态流转是黑盒你无法像上面这样清晰地看到state是如何从PLANNING变成EXECUTING再变成REFLECTING的。而手写这个过程就是把Agent的“灵魂”具象化。3. LangChain与LangGraph不是替代关系而是“乐高积木”与“建筑图纸”的分工当你的手写Agent原型跑通后下一个自然问题是既然我能手写为什么还要用LangChain这个问题的答案决定了你能否真正驾驭Agent开发。LangChain不是Agent的“操作系统”它是一套高度抽象的组件化工具集LangGraph也不是LangChain的“升级版”它是为了解决LangChain在复杂状态编排上的先天不足而生的流程图引擎。把它们混为一谈就像把钢筋LangChain和AutoCADLangGraph当成同一种建材。3.1 LangChain的核心价值标准化“零件”而非组装说明书LangChain最被低估也最被滥用的是它的Tool和BaseTool抽象。很多人以为tool装饰器只是让函数能被LLM调用这太浅了。它的真正威力在于统一了工具的描述、调用、错误处理、结果解析这四个维度。我们来看一个真实案例from langchain_core.tools import BaseTool from langchain_core.pydantic_v1 import BaseModel, Field class SearchManualInput(BaseModel): query: str Field(description要搜索的设备型号或关键词例如设备A) class SearchManualTool(BaseTool): name search_manual description 搜索设备操作手册。输入必须是具体的设备型号或关键词不能是模糊描述如我的设备 args_schema: Type[BaseModel] SearchManualInput def _run(self, query: str) - str: # 真实实现调用企业知识库API try: response requests.get(fhttps://kb-api/internal/manuals?query{query}) return response.json()[url] except Exception as e: # LangChain要求所有工具错误必须抛出ToolException raise ToolException(f手册搜索失败: {str(e)})这段代码里args_schema定义了LLM调用此工具时必须生成的JSON Schemadescription是LLM Planner的唯一输入依据_run方法里的ToolException是LangChain统一的错误信道确保所有工具错误都能被AgentExecutor捕获并按统一策略处理。这才是LangChain的护城河它用一套严格的契约Contract让千奇百怪的工具数据库、API、本地脚本能在一个Agent里无缝协作。如果你手写的Agent里每个工具都有自己的错误码、自己的参数校验逻辑、自己的返回格式那维护成本会指数级上升。LangChain的BaseTool就是帮你把这种混乱变成可预测、可测试、可替换的标准件。3.2 LangGraph的诞生逻辑当状态机变得太复杂你需要一张“交通管制图”LangChain的AgentExecutor有一个致命缺陷它只能处理线性状态流——input - plan - execute - output。但真实业务中Agent经常需要分支、循环、并行、状态持久化。比如一个客服Agent流程可能是接收用户投诉并行执行a) 查询订单状态 b) 检查物流轨迹 c) 调取用户历史投诉记录根据a/b/c的结果决定走“退款流程”还是“补发流程”在退款流程中需要循环确认用户银行卡信息直到格式正确这种流程用LangChain的AgentExecutor写出来会是一团嵌套的if/else和回调地狱。LangGraph的出现就是为了解决这个问题。它把Agent的整个生命周期建模成一个有向图Directed Graph每个节点是一个函数Node每条边是一个条件Edge。我们用LangGraph重写上面的客服Agent流程from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any class AgentState(TypedDict): messages: List[dict] order_status: Dict[str, Any] logistics: Dict[str, Any] history: List[Dict[str, Any]] bank_info: str retry_count: int # 定义节点函数 def fetch_order_status(state: AgentState) - AgentState: # 调用订单API state[order_status] call_order_api(state[messages][-1][content]) return state def fetch_logistics(state: AgentState) - AgentState: state[logistics] call_logistics_api(state[messages][-1][content]) return state def fetch_history(state: AgentState) - AgentState: state[history] call_history_api(state[messages][-1][content]) return state def decide_next_step(state: AgentState) - str: 这是一个Edge函数决定下一步走向哪个节点 if state[order_status][status] shipped: return handle_refund else: return handle_reship def handle_refund(state: AgentState) - AgentState: # 处理退款逻辑 if validate_bank_info(state[bank_info]): process_refund(state[bank_info]) return {messages: [{role: assistant, content: 退款已发起}]} else: state[retry_count] 1 if state[retry_count] 3: return {messages: [{role: assistant, content: 请提供正确的银行卡号}]} return {messages: [{role: assistant, content: 银行卡号格式有误请重新输入}]} # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(fetch_order_status, fetch_order_status) workflow.add_node(fetch_logistics, fetch_logistics) workflow.add_node(fetch_history, fetch_history) workflow.add_node(decide_next_step, decide_next_step) # 这个节点不修改state只返回edge名 workflow.add_node(handle_refund, handle_refund) workflow.add_node(handle_reship, handle_reship) # 添加边边的条件由函数返回的字符串决定 workflow.add_edge(fetch_order_status, decide_next_step) workflow.add_edge(fetch_logistics, decide_next_step) workflow.add_edge(fetch_history, decide_next_step) workflow.add_conditional_edges( decide_next_step, lambda x: x, # 直接返回decide_next_step的返回值作为edge名 { handle_refund: handle_refund, handle_reship: handle_reship } ) workflow.add_edge(handle_refund, END) workflow.add_edge(handle_reship, END) # 编译图 app workflow.compile()LangGraph的革命性在于它把“状态流转”这件事从代码逻辑里解放出来变成了可视化、可调试、可版本控制的图结构。你可以用app.get_graph().draw_mermaid_png()生成流程图虽然我们禁用mermaid但这个能力本身说明了它的定位可以清晰地看到fetch_order_status和fetch_logistics是并行执行的decide_next_step是一个决策点handle_refund内部有重试循环。这种表达力是任何线性代码都无法比拟的。LangGraph不是LangChain的替代品而是当LangChain的组件Tool、LLM组合起来后用来指挥它们“怎么协作”的总控台。3.3 CrewAI当Agent需要“团队作战”它提供的是组织架构而非技术栈CrewAI的定位常常被严重误读。很多人以为它是“LangChain的竞品”其实它解决的是一个完全不同的问题多Agent协同的组织管理。单个Agent再强大也难以胜任需要“市场调研产品设计文案撰写视觉设计”四步联动的完整项目。CrewAI提供的是一套模拟人类团队协作的抽象层Agent定义单个角色的能力、目标、工具底层仍用LangChain的ToolTask定义一个具体、可交付的工作单元如“撰写一份关于XX功能的用户手册”Crew将多个Agent组织起来定义它们之间的协作流程串行、并行、评审制from crewai import Agent, Task, Crew, Process # 定义市场研究员Agent researcher Agent( roleMarket Researcher, goal收集并分析竞品功能文档提炼核心差异点, tools[serpapi_search, pdf_reader], # 这些tool仍是LangChain的BaseTool backstory拥有5年SaaS产品分析经验擅长从海量文档中提取关键信息 ) # 定义产品经理Agent product_manager Agent( roleProduct Manager, goal基于调研结果设计新功能的PRD文档, tools[doc_writer], backstory曾主导3款百万级用户产品的功能设计 ) # 定义任务 research_task Task( description搜索并分析竞品A、B、C的最新版本功能文档重点对比其AI Agent相关功能, agentresearcher, expected_output一份包含3个竞品功能对比表格的Markdown报告 ) write_prd_task Task( description根据调研报告撰写新功能PRD包含用户故事、验收标准、技术约束, agentproduct_manager, context[research_task], # 明确指定前置任务形成依赖 expected_output一份完整的PRD文档 ) # 组建团队 crew Crew( agents[researcher, product_manager], tasks[research_task, write_prd_task], processProcess.sequential, # 可选sequential, hierarchical, consensus verboseTrue ) # 启动团队协作 result crew.kickoff()CrewAI的价值在于它把“跨Agent的知识传递”、“任务依赖管理”、“结果质量评审”这些软性协作问题变成了硬编码的配置。它不关心底层Agent用的是LangChain还是LlamaIndex它只关心“这个Agent有没有权限访问那个Task的输出”。如果你的项目需要一个Agent负责查数据另一个Agent负责写报告第三个Agent负责审核报告那么CrewAI就是你不可或缺的“项目经理”。但它绝不是“不用LangChain就能做Agent”的捷径——它的Agent类内部依然重度依赖LangChain的LLM和Tool。4. MCP协议不是银弹而是Agent世界的“USB-C接口标准”在所有热词中“MCP”Model Communication Protocol被赋予了过多不切实际的期待。很多人以为它是一套能让所有Agent“即插即用”的万能协议仿佛有了MCPLangChain Agent就能无缝调用CrewAI Agent。这种想法源于对协议本质的误解。MCP不是操作系统它只是一个聚焦于“工具描述”和“调用结果”这两个环节的轻量级规范。它的核心目标是解决Agent生态中最痛的“方言”问题同一个“搜索手册”的功能在LangChain里叫search_manual在CrewAI里叫kb_search在自研框架里叫get_doc而它们的参数、返回格式更是五花八门。4.1 MCP的“三板斧”只管好工具的“出生证”和“成绩单”MCP协议的全部规范可以浓缩为三个核心概念Tool Specification工具规格书用JSON Schema定义一个工具的“身份证”。它强制要求包含name: 工具的全局唯一标识符如mcp://kb/search-manualdescription: 人类可读的用途说明input_schema: 一个严格的JSON Schema定义调用时必须传入的参数如{type: object, properties: {query: {type: string}}}output_schema: 一个严格的JSON Schema定义成功返回时的结构如{type: object, properties: {url: {type: string}, title: {type: string}}}Tool Call工具调用定义了一种标准化的调用消息格式。无论底层是HTTP、gRPC还是本地函数MCP要求调用方必须发送一个包含tool工具名、args参数对象、id调用唯一ID的JSON对象。Tool Result工具结果定义了一种标准化的结果消息格式。无论工具内部如何实现返回给Agent的消息必须包含id对应调用ID、result成功时的结构化数据、error失败时的错误信息。// MCP Tool Specification 示例 { name: mcp://kb/search-manual, description: 在企业知识库中搜索设备操作手册, input_schema: { type: object, properties: { query: { type: string, description: 要搜索的设备型号或关键词 } }, required: [query] }, output_schema: { type: object, properties: { url: {type: string}, title: {type: string}, page_count: {type: integer} } } } // MCP Tool Call 消息 { id: call_abc123, tool: mcp://kb/search-manual, args: {query: 设备A} } // MCP Tool Result 消息成功 { id: call_abc123, result: { url: https://kb.example.com/manuals/device_a.pdf, title: 设备A操作手册V3.2, page_count: 42 } } // MCP Tool Result 消息失败 { id: call_abc123, error: { code: NOT_FOUND, message: 未找到与设备A匹配的手册 } }看到这里你应该明白MCP解决的是“如何让不同框架的Agent能互相理解对方的工具”这个问题。它不规定Agent内部的状态机怎么设计不规定Planner用什么模型不规定Memory存在哪里。它只像USB-C接口一样保证“只要你的工具贴了MCP认证标签我的Agent主机就能认出你并且知道怎么给你供电、怎么读取你的数据”。MCP的价值在于降低集成成本而不是提升单个Agent的能力。一个不遵循MCP的、功能强大的Agent依然是强大的一个严格遵循MCP的、功能孱弱的Agent依然是孱弱的。4.2 在实践中落地MCP一个兼容LangChain和自研框架的双模工具要真正理解MCP最好的方式是动手写一个同时支持LangChain和原生Python Agent的工具。我们以“搜索手册”为例展示如何用同一份MCP Spec驱动两种完全不同的实现# mcp_spec.py - MCP规范文件独立于任何框架 MCP_SEARCH_MANUAL_SPEC { name: mcp://kb/search-manual, description: 在企业知识库中搜索设备操作手册, input_schema: { type: object, properties: {query: {type: string}}, required: [query] }, output_schema: { type: object, properties: {url: {type: string}, title: {type: string}}, required: [url, title] } } # langchain_tool.py - LangChain兼容实现 from langchain_core.tools import BaseTool from langchain_core.pydantic_v1 import BaseModel, Field from mcp_spec import MCP_SEARCH_MANUAL_SPEC class SearchManualInput(BaseModel): query: str Field(description要搜索的设备型号或关键词) class MCPSearchManualTool(BaseTool): name MCP_SEARCH_MANUAL_SPEC[name] description MCP_SEARCH_MANUAL_SPEC[description] args_schema: Type[BaseModel] SearchManualInput def _run(self, query: str) - Dict[str, str]: # 实际调用逻辑 result call_kb_api(query) # 严格按MCP output_schema返回 return { url: result[url], title: result[title] } # native_agent_tool.py - 原生Agent兼容实现 from mcp_spec import MCP_SEARCH_MANUAL_SPEC def search_manual_mcp(query: str) - Dict[str, Any]: 一个纯函数符合MCP规范的工具实现 # 参数校验必须符合input_schema if not isinstance(query, str) or not query.strip(): raise ValueError(query must be a non-empty string) # 实际调用 result call_kb_api(query) # 结果校验必须符合output_schema if not isinstance(result, dict) or url not in result or title not in result: raise ValueError(KB API returned invalid format) return { url: result[url], title: result[title] } # 在原生Agent中注册 my_agent.tools[mcp://kb/search-manual] search_manual_mcp这个例子揭示了MCP的精髓它是一份契约不是一套代码。工具提供方call_kb_api只需写一次业务逻辑然后为LangChain和原生Agent分别写一个薄薄的适配层Adapter这个适配层的唯一职责就是把业务逻辑的输入/输出转换成符合MCP Spec的格式。对于Agent使用者来说他只需要知道mcp://kb/search-manual这个URI以及它接受什么参数、返回什么结构至于背后是LangChain的BaseTool还是一个裸函数完全不重要。这就是MCP带来的“解耦”力量。4.3 MCP的现实边界它不解决也不该解决的五个问题尽管MCP前景广阔但必须清醒认识其局限性。以下五个问题MCP明确不解决试图用它来解决只会南辕北辙Agent内部状态管理MCP不管你的Agent是用LangGraph的StateGraph还是手写的dataclass还是某种自研状态机。它只管“工具调用”这一瞬间的输入输出。Planner的智能水平MCP不规定Planner用GPT-4还是本地小模型也不规定它用RAG还是微调。它只规定Planner在决定调用mcp://kb/search-manual时必须传入符合input_schema的参数。Memory的存储与检索MCP不涉及向量数据库选型、embedding模型选择、chunk策略。它只规定当Agent需要把一次成功的mcp://kb/search-manual调用结果存入Memory时应该提取url和title这两个字段。安全与权限控制MCP不定义“谁有权调用这个工具”。权限控制必须在Agent框架层或网关层实现。MCP Spec里可以加permissions字段作为提示但这只是建议不是强制。跨网络通信协议MCP不规定工具调用是走HTTP、WebSocket还是gRPC。它只定义消息的JSON结构。一个HTTP服务