找回密码
 立即注册
首页 业界区 业界 LLM 工具调用的范式演进与认知模型集成

LLM 工具调用的范式演进与认知模型集成

万妙音 5 小时前
引言:为什么 LLM 需要工具?

大语言模型的核心能力是文本生成,但其存在三个根本性限制:
限制类型具体表现工具调用如何解决知识时效性训练数据截止于某时间点调用搜索引擎/API 获取实时信息计算精确性浮点运算、复杂数学易出错调用计算器/代码执行器世界交互能力无法发送邮件、操作数据库调用外部服务完成副作用操作工具调用(Tool Calling)机制使 LLM 从"只能说"进化为"能说能做"。本文档将系统性地剖析这一机制的两种实现范式及其底层原理。
graph LR    subgraph LLM_Limitations["LLM 原生能力边界"]        A["文本理解"]        B["文本生成"]        C["模式识别"]    end        subgraph Tool_Extensions["工具扩展能力"]        D["实时数据获取"]        E["精确计算"]        F["外部系统操作"]    end        LLM_Limitations -->|"工具调用桥接"| Tool_Extensions        style LLM_Limitations fill:#e3f2fd,stroke:#1565c0    style Tool_Extensions fill:#e8f5e9,stroke:#2e7d32第一章:执行拓扑演进 —— 从进程内调用到协议化通信

在这一章节,我们拨开上层封装,探究 LLM 驱动外部代码的两种物理存在形式。我们首先确立一个核心共识:对于 LLM 而言,无论是本地函数还是远程 MCP,其交互接口(Interface)是完全同构的(Schema 输入 → Result 输出),但其执行拓扑(Implementation)存在异构性。
1.1 模式 A:原生调用架构 (@tool / Direct Function Calling)

这是 LangChain/LangGraph 的默认单体模式。
1.1.1 架构定义

紧耦合(Tightly Coupled):所有工具函数与 LLM 控制器(Runtime)运行在同一个操作系统进程内。
1.1.2 执行机制详解

原生调用的核心流程可分解为四个阶段:

  • 符号解析:Runtime 在 Python 的 locals() 或 globals() 命名空间中查找函数对象
  • 栈帧构建:参数通过调用栈(Call Stack)传递,遵循 Python 的调用约定
  • 同步执行:函数体在当前线程中执行,阻塞直至返回
  • 异常传播:任何未捕获的异常将沿调用栈向上传播
flowchart TB    subgraph Memory["ython 进程内存空间"]        direction TB                subgraph Namespace["命名空间"]            GlobalScope["globals()"]            LocalScope["locals()"]        end                subgraph CallStack["调用栈"]            Frame1["Runtime Frame"]            Frame2["Tool Function Frame"]            Frame1 --> Frame2        end                subgraph HeapObjects["堆对象"]            Args["参数对象"]            Result["返回值对象"]        end    end        Namespace -->|"符号查找"| CallStack    CallStack -->|"引用传递"| HeapObjects        style Memory fill:#fafafa,stroke:#424242    style Namespace fill:#fff3e0,stroke:#e65100    style CallStack fill:#e8eaf6,stroke:#3949ab    style HeapObjects fill:#f3e5f5,stroke:#7b1fa21.1.3 代码示例
  1. from langchain_core.tools import tool
  2. @tool
  3. def calculate_compound_interest(
  4.     principal: float,
  5.     rate: float,
  6.     years: int
  7. ) -> float:
  8.     """计算复利。
  9.    
  10.     Args:
  11.         principal: 本金
  12.         rate: 年利率(如 0.05 表示 5%)
  13.         years: 投资年数
  14.    
  15.     Returns:
  16.         最终金额
  17.     """
  18.     return principal * (1 + rate) ** years
  19. # 运行时直接通过函数指针调用
  20. result = calculate_compound_interest.invoke({
  21.     "principal": 10000,
  22.     "rate": 0.05,
  23.     "years": 10
  24. })
  25. # result = 16288.95(内存中的 float 对象)
复制代码
1.1.4 拓扑图解

graph TB    subgraph Process["单体宿主进程 (Python Runtime)"]        LLM["LLM 推理核心
生成 tool_calls"]        Runtime["LangGraph Runtime
工具调度器"]        SymbolTable[("符号表
locals / globals")]                subgraph ToolRegistry["工具注册表"]            Tool1["@tool
calculate_interest"]            Tool2["@tool
query_database"]            Tool3["@tool
send_email"]        end                LLM -->|"1. 输出结构化调用意图
tool_calls: [...]"| Runtime        Runtime -->|"2. 符号查找
getattr()"| SymbolTable        SymbolTable -->|"3. 函数指针"| ToolRegistry        ToolRegistry -->|"4. return 内存对象"| Runtime        Runtime -->|"5. 构造 ToolMessage"| LLM    end        style Process fill:#f5f5f5,stroke:#333,stroke-width:2px    style LLM fill:#bbdefb,stroke:#1976d2    style Runtime fill:#c8e6c9,stroke:#388e3c    style ToolRegistry fill:#fff9c4,stroke:#fbc02d    style SymbolTable fill:#ffecb3,stroke:#ffa0001.1.5 优势与局限

优势局限零网络延迟,调用速度极快工具故障可能导致整个进程崩溃调试简单,可直接断点跟踪必须使用相同编程语言无序列化开销无法独立扩展工具实例共享进程上下文(如数据库连接池)资源隔离困难1.2 模式 B:MCP 协议架构 (Model Context Protocol)

这是面向未来的组件化模式,由 Anthropic 于 2024 年提出并开源。
1.2.1 架构定义

松耦合(Loosely Coupled):Runtime 不再持有函数代码,而是持有"连接通道"。工具作为独立进程或服务运行,通过标准化协议通信。
1.2.2 MCP 协议栈详解

MCP 建立在 JSON-RPC 2.0 之上,定义了三层抽象:
graph TB    subgraph ProtocolStack["MCP 协议栈"]        direction TB                subgraph ApplicationLayer["应用层"]            Tools["tools/* 方法"]            Resources["resources/* 方法"]            Prompts["prompts/* 方法"]        end                subgraph MessageLayer["消息层"]            JSONRPC["JSON-RPC 2.0"]            Request["Request: id, method, params"]            Response["Response: id, result/error"]        end                subgraph TransportLayer["传输层"]            Stdio["Stdio
标准输入输出"]            SSE["SSE
Server-Sent Events"]            WebSocket["WebSocket
(计划中)"]        end                ApplicationLayer --> MessageLayer        MessageLayer --> TransportLayer    end        style ProtocolStack fill:#fafafa,stroke:#424242    style ApplicationLayer fill:#e3f2fd,stroke:#1565c0    style MessageLayer fill:#fff3e0,stroke:#e65100    style TransportLayer fill:#e8f5e9,stroke:#2e7d321.2.3 核心交互流程

工具发现阶段(Handshake)
sequenceDiagram    autonumber    participant Client as MCP Client
(LangGraph)    participant Server as MCP Server
(工具提供者)        Note over Client,Server: 连接建立阶段    Client->>Server: initialize
{"protocolVersion": "2024-11-05"}    Server-->>Client: {"capabilities": {...}, "serverInfo": {...}}    Client->>Server: initialized (通知)        Note over Client,Server: 工具发现阶段    Client->>Server: tools/list    Server-->>Client: {"tools": [
  {"name": "query_db", "description": "...", "inputSchema": {...}},
  {"name": "send_email", ...}
]}        Note over Client: Client 将工具 Schema
转换为 LLM 可理解的格式工具调用阶段(Invocation)
sequenceDiagram    autonumber    participant LLM as LLM    participant Runtime as LangGraph Runtime    participant Client as MCP Client Stub    participant Transport as 传输层
(Stdio/SSE)    participant Server as MCP Server    participant Logic as 业务逻辑        LLM->>Runtime: tool_calls: [{"name": "query_db", "args": {...}}]    Runtime->>Client: 路由到对应 MCP 连接        Note over Client,Transport: 序列化 (Marshaling)    Client->>Transport: JSON-RPC Request
{"jsonrpc": "2.0", "id": 1,
"method": "tools/call",
"params": {"name": "query_db", "arguments": {...}}}        Transport->>Server: 字节流传输    Server->>Logic: 反序列化 + 分发    Logic->>Logic: 执行 SQL 查询    Logic-->>Server: 返回结果        Note over Server,Transport: 响应序列化    Server-->>Transport: {"jsonrpc": "2.0", "id": 1,
"result": {"content": [{"type": "text", "text": "..."}]}}        Transport-->>Client: 字节流传输    Client-->>Runtime: 解析为 ToolMessage    Runtime-->>LLM: 注入对话上下文1.2.4 完整拓扑图解

graph LR    subgraph AgentSystem["AI Agent 主系统"]        LLM["LLM 推理核心"]        Runtime["LangGraph Runtime"]        ClientPool["MCP Client Pool"]                LLM --> Runtime        Runtime --> ClientPool    end        subgraph TransportBus["通信总线"]        Pipe1["JSON-RPC Channel 1
(Stdio)"]        Pipe2["JSON-RPC Channel 2
(SSE)"]    end        subgraph ExternalServices["独立工具进程群"]        subgraph Server1["MCP Server: 数据库"]            Dispatcher1["请求分发器"]            DB["SQLite/PostgreSQL"]            Dispatcher1 --> DB        end                subgraph Server2["MCP Server: 文件系统"]            Dispatcher2["请求分发器"]            FS["本地文件系统"]            Dispatcher2 --> FS        end    end        ClientPool -->|"序列化"| Pipe1    ClientPool -->|"序列化"| Pipe2    Pipe1 -->|"反序列化"| Dispatcher1    Pipe2 -->|"反序列化"| Dispatcher2        Dispatcher1 -.->|"Result"| Pipe1    Dispatcher2 -.->|"Result"| Pipe2    Pipe1 -.->|"ToolMessage"| ClientPool    Pipe2 -.->|"ToolMessage"| ClientPool        style AgentSystem fill:#e1f5fe,stroke:#01579b,stroke-width:2px    style ExternalServices fill:#e8f5e9,stroke:#2e7d32,stroke-width:2px    style TransportBus fill:#fff3e0,stroke:#e65100,stroke-dasharray: 5 51.2.5 MCP Server 实现示例
  1. # mcp_server_sqlite.py
  2. from mcp.server import Server
  3. from mcp.types import Tool, TextContent
  4. import sqlite3
  5. server = Server("sqlite-server")
  6. @server.list_tools()
  7. async def list_tools():
  8.     return [
  9.         Tool(
  10.             name="query_db",
  11.             description="执行 SQL 查询并返回结果",
  12.             inputSchema={
  13.                 "type": "object",
  14.                 "properties": {
  15.                     "sql": {"type": "string", "description": "SQL 查询语句"}
  16.                 },
  17.                 "required": ["sql"]
  18.             }
  19.         )
  20.     ]
  21. @server.call_tool()
  22. async def call_tool(name: str, arguments: dict):
  23.     if name == "query_db":
  24.         conn = sqlite3.connect("data.db")
  25.         cursor = conn.execute(arguments["sql"])
  26.         rows = cursor.fetchall()
  27.         conn.close()
  28.         return [TextContent(type="text", text=str(rows))]
  29.     raise ValueError(f"Unknown tool: {name}")
  30. # 启动:python -m mcp.server.stdio mcp_server_sqlite
复制代码
1.3 核心差异对比表

维度Native @toolMCP Protocol执行位置本地内存 (In-Process)远程/独立进程 (Out-of-Process)调用方式函数指针 (Function Pointer)消息传递 (Message Passing)上下文共享可访问全局变量、共享连接池完全隔离 (Context Blindness)故障影响宿主进程崩溃风险进程级沙箱隔离 (Sandboxed)技术栈约束必须与宿主语言一致多语言互通 (Polyglot)部署扩展随主进程扩缩容独立扩缩容,支持分布式部署版本管理与主应用耦合独立版本控制与发布延迟特征微秒级毫秒级(含序列化+传输)1.4 架构选型决策树

flowchart TD    Start["需要集成外部工具"] --> Q1{"工具是否需要
访问进程内状态?"}        Q1 -->|"是"| Native["选择 @tool
原生调用"]    Q1 -->|"否"| Q2{"是否需要
多语言支持?"}        Q2 -->|"是"| MCP["选择 MCP"]    Q2 -->|"否"| Q3{"是否需要
故障隔离?"}        Q3 -->|"是"| MCP    Q3 -->|"否"| Q4{"是否需要
独立扩缩容?"}        Q4 -->|"是"| MCP    Q4 -->|"否"| Native        Native --> NativeNote["适用场景:
- 原型开发
- 简单工具
- 性能敏感场景"]    MCP --> MCPNote["适用场景:
- 企业级部署
- 多团队协作
- 工具复用"]        style Start fill:#e3f2fd,stroke:#1565c0    style Native fill:#fff9c4,stroke:#fbc02d    style MCP fill:#c8e6c9,stroke:#388e3c    style NativeNote fill:#fffde7,stroke:#f9a825    style MCPNote fill:#e8f5e9,stroke:#43a047第二章:认知与执行 —— ReAct 在 MCP 上的运行机制

如果说第一章解决了"手脚如何连接"的问题,本章将解决"大脑如何驱动手脚"的问题。我们将经典的 ReAct (Reasoning + Acting) 认知框架映射到 MCP 的通信协议上。
2.1 ReAct 范式概述

ReAct 由 Yao et al. (2022) 提出,其核心思想是让 LLM 交替进行:

  • Reasoning(推理):生成自然语言的思考过程
  • Acting(行动):调用外部工具执行操作
  • Observing(观察):接收工具返回的结果
graph LR    subgraph ReActLoop["ReAct 认知循环"]        Thought["Thought
思考推理"]        Action["Action
工具调用"]        Observation["Observation
结果观察"]                Thought -->|"决定行动"| Action        Action -->|"执行后获得"| Observation        Observation -->|"基于结果再思考"| Thought    end        Input["用户输入"] --> Thought    Thought -->|"推理完成"| Output["最终回答"]        style ReActLoop fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px    style Thought fill:#e1bee7,stroke:#8e24aa    style Action fill:#bbdefb,stroke:#1976d2    style Observation fill:#c8e6c9,stroke:#388e3c2.2 架构分层原理

在我们的架构中,必须严格区分认知层执行层
graph TB    subgraph CognitiveLayer["认知层 (Cognitive Layer)"]        direction LR        State["状态管理
对话历史"]        Reasoning["推理引擎
LLM"]        Decision["决策输出
tool_calls / answer"]    end        subgraph ExecutionLayer["执行层 (Execution Layer)"]        direction LR        Routing["请求路由
工具分发"]        Transport["传输层
MCP Protocol"]        Execution["实际执行
外部系统"]    end        CognitiveLayer -->|"Action 意图"| ExecutionLayer    ExecutionLayer -->|"Observation 结果"| CognitiveLayer        Note1["负责 WHY
为什么这样做"] --> CognitiveLayer    Note2["负责 HOW
如何执行"] --> ExecutionLayer        style CognitiveLayer fill:#e8eaf6,stroke:#3949ab,stroke-width:2px    style ExecutionLayer fill:#fff3e0,stroke:#e65100,stroke-width:2px层级职责实现位置状态特征认知层 (ReAct)推理、决策、记忆LLM 上下文窗口有状态 (Stateful)执行层 (MCP)工具调用、结果传输网络通信管道无状态 (Stateless)2.3 ReAct 循环的协议映射

一个完整的 ReAct 步骤(思考-行动-观察)在 MCP 架构中被物理拆解为以下流程:
stateDiagram-v2   
  • --> Thought: 用户输入        Thought --> Action: 需要外部信息    Thought --> FinalAnswer: 可直接回答        state "Thought (思考)" as Thought {        
  • --> Analyze: 分析问题        Analyze --> Plan: 制定计划        Plan --> Decide: 决策    }        state "Action (行动)" as Action {        
  • --> Serialize: 序列化请求        Serialize --> Transport: MCP 传输        Transport --> Execute: 远程执行    }        state "Observation (观察)" as Observation {        
  • --> Receive: 接收结果        Receive --> Parse: 解析响应        Parse --> Inject: 注入上下文    }        Action --> Observation: 执行完成    Observation --> Thought: 继续推理        FinalAnswer -->
  • A. Thought (思考) → 纯神经活动


    • 场景:LLM 分析用户需求,规划解决方案
    • 系统行为:仅涉及 Prompt 计算,无网络流量
    • MCP 状态:Server 处于静默状态
    1. 用户: "查询库存中 Item B 的数量"
    2. LLM 内部思考:
    3. Thought: 用户想要查询库存信息。
    4.          我需要使用 query_db 工具执行 SQL 查询。
    5.          SQL 语句应该是: SELECT quantity FROM items WHERE name = 'Item B'
    复制代码
    B. Action (行动) → 协议请求 (Request)


    • ReAct 语义:模型决定采取行动
    • MCP 映射:意图被序列化为 JSON-RPC 请求
    flowchart LR    subgraph LLMOutput["LLM 输出"]        Intent["tool_calls: [{
      name: 'query_db',
      arguments: {sql: 'SELECT...'}
    }]"]    end        subgraph MCPClient["MCP Client"]        Serialize["序列化器"]    end        subgraph Transport["传输层"]        JSONPacket["JSON-RPC Packet
    {jsonrpc: '2.0',
    method: 'tools/call',
    params: {...}}"]    end        Intent --> Serialize    Serialize --> JSONPacket    JSONPacket -->|"Stdio / SSE"| Server["MCP Server"]        style LLMOutput fill:#e3f2fd,stroke:#1565c0    style MCPClient fill:#fff3e0,stroke:#e65100    style Transport fill:#e8f5e9,stroke:#2e7d32C. Observation (观察) → 协议响应 (Response)


    • ReAct 语义:模型通过观察结果来修正或确认下一步
    • MCP 映射:响应被反序列化并注入 LLM 上下文
    flowchart RL    subgraph Server["MCP Server"]        Execute["执行 SQL"]        Result["结果: [('Item B', 0)]"]    end        subgraph Transport["传输层"]        ResponsePacket["JSON-RPC Response
    {result: {content: [...]}}"]    end        subgraph MCPClient["MCP Client"]        Deserialize["反序列化器"]    end        subgraph LLMContext["LLM 上下文"]        ToolMessage["ToolMessage:
    'Item B 库存: 0'"]    end        Execute --> Result    Result --> ResponsePacket    ResponsePacket --> Deserialize    Deserialize --> ToolMessage        style Server fill:#e8f5e9,stroke:#2e7d32    style Transport fill:#fff3e0,stroke:#e65100    style MCPClient fill:#ffecb3,stroke:#ffa000    style LLMContext fill:#e3f2fd,stroke:#1565c02.4 全链路时序图 (The ReAct-MCP Loop)

    下图展示了时间维度上,Agent 如何利用 MCP 协议完成一次完整的认知闭环:
    sequenceDiagram    autonumber    participant User as 用户    participant Context as LLM Context
    (记忆流)    participant Runtime as LangGraph Runtime    participant Bus as MCP Protocol Bus    participant Server as MCP Server
    (SQLite)        User->>Context: "查询 Item B 的库存"        rect rgb(232, 234, 246)        Note over Context: 阶段 1: 思考 (Reasoning)        Context->>Context: Thought: "用户想查库存。
    我需要使用 query_db 工具。"    end        rect rgb(227, 242, 253)        Note over Context,Runtime: 阶段 2: 行动 (Acting)        Context->>Runtime: Action: {tool: "query_db",
    sql: "SELECT * FROM items WHERE name='Item B'"}        Runtime->>Bus: 序列化: JSON-RPC request
    {method: "tools/call", params: {...}}    end        rect rgb(232, 245, 233)        Note over Bus,Server: 阶段 3: 执行 (Execution)        Bus->>Server: 传输请求        Server->>Server: 执行 SQL 查询        Server-->>Bus: JSON-RPC response
    {result: "Item B: quantity=0"}    end        rect rgb(255, 243, 224)        Note over Runtime,Context: 阶段 4: 观察 (Observing)        Bus-->>Runtime: 传输响应        Runtime->>Context: 注入 Observation:
    ToolMessage("Item B: quantity=0")    end        rect rgb(243, 229, 245)        Note over Context: 阶段 5: 再思考 (Reasoning)        Context->>Context: Thought: "查询结果显示 Item B 库存为 0。
    我需要告知用户该商品缺货。"        Context->>User: Final Answer: "Item B 目前缺货,库存为 0。"    end2.5 多轮工具调用示例

    复杂任务通常需要多次工具调用,形成链式推理:
    sequenceDiagram    autonumber    participant User as 用户    participant Agent as Agent    participant DB as query_db    participant Email as send_email        User->>Agent: "检查库存不足的商品并通知采购部"        Note over Agent: Thought 1: 需要先查询库存    Agent->>DB: Action 1: SELECT * FROM items WHERE quantity < 10    DB-->>Agent: Observation 1: [("Item B", 0), ("Item D", 5)]        Note over Agent: Thought 2: 发现 2 个商品库存不足,需要发邮件    Agent->>Email: Action 2: send_email(to="purchase@...",
    subject="库存预警",
    body="Item B: 0, Item D: 5")    Email-->>Agent: Observation 2: "邮件发送成功"        Note over Agent: Thought 3: 任务完成    Agent->>User: "已检测到 2 件商品库存不足,
    已发送邮件通知采购部。"2.6 鲁棒性设计:错误即观察

    在 MCP 结合 ReAct 的架构中,Server 端的运行时错误被设计为一种合法的观察值,而非系统中断。
    flowchart TB    subgraph Traditional["传统模式"]        Error1["SQL 语法错误"]        Exception["抛出 Exception"]        Crash["Agent 进程崩溃 ❌"]                Error1 --> Exception --> Crash    end        subgraph MCPReAct["MCP + ReAct 模式"]        Error2["SQL 语法错误"]        ErrorResponse["返回错误响应
    {error: 'Syntax Error near...'}"]        Observation["包装为 Observation
    ToolMessage(error=...)"]        Thought["LLM 自愈思考
    'SQL 有误,需要修正'"]        FixAction["生成修复 Action
    新的正确 SQL"]        Success["成功执行 ✓"]                Error2 --> ErrorResponse --> Observation --> Thought --> FixAction --> Success    end        style Traditional fill:#ffebee,stroke:#c62828    style MCPReAct fill:#e8f5e9,stroke:#2e7d32    style Crash fill:#ef5350,stroke:#b71c1c,color:#fff    style Success fill:#66bb6a,stroke:#2e7d32,color:#fff错误自愈时序示例

    sequenceDiagram    autonumber    participant Agent as Agent    participant Server as MCP Server        Note over Agent: 首次尝试(含错误)    Agent->>Server: tools/call: "SELEC * FROM items"    Server-->>Agent: {error: "Syntax error: unknown keyword 'SELEC'"}        Note over Agent: Thought: SQL 关键字拼写错误,
    应该是 SELECT 而不是 SELEC        Note over Agent: 修正后重试    Agent->>Server: tools/call: "SELECT * FROM items"    Server-->>Agent: {result: "[('Item A', 50), ('Item B', 0)]"}        Note over Agent: Thought: 查询成功,可以回答用户了这种机制极大地提升了 Agent 系统的容错性和自主修复能力,将"异常处理"从工程问题转化为"认知问题"。
    第三章:LangGraph 集成实践

    本章展示如何在 LangGraph 中同时使用原生工具和 MCP 工具,实现混合架构。
    3.1 架构概览

    graph TB    subgraph LangGraphAgent["LangGraph Agent"]        StateGraph["StateGraph
    状态机"]        AgentNode["Agent Node
    LLM 推理"]        ToolNode["Tool Node
    工具执行"]                StateGraph --> AgentNode        AgentNode -->|"tool_calls"| ToolNode        ToolNode -->|"ToolMessage"| AgentNode    end        subgraph ToolSources["工具来源"]        Native["原生 @tool"]        MCPAdapter["MCP Adapter"]    end        subgraph MCPServers["MCP Servers"]        SQLite["sqlite-server"]        FileSystem["filesystem-server"]        GitHub["github-server"]    end        ToolNode --> Native    ToolNode --> MCPAdapter    MCPAdapter --> SQLite    MCPAdapter --> FileSystem    MCPAdapter --> GitHub        style LangGraphAgent fill:#e3f2fd,stroke:#1565c0,stroke-width:2px    style ToolSources fill:#fff3e0,stroke:#e65100    style MCPServers fill:#e8f5e9,stroke:#2e7d323.2 代码实现
    1. import asyncio
    2. from langchain_core.tools import tool
    3. from langchain_openai import ChatOpenAI
    4. from langgraph.prebuilt import create_react_agent
    5. from langchain_mcp_adapters.client import MultiServerMCPClient
    6. # 1. 定义原生工具
    7. @tool
    8. def calculate(expression: str) -> str:
    9.     """计算数学表达式。例如: '2 + 3 * 4'"""
    10.     try:
    11.         result = eval(expression, {"__builtins__": {}})
    12.         return f"计算结果: {result}"
    13.     except Exception as e:
    14.         return f"计算错误: {e}"
    15. # 2. 配置 MCP 服务器
    16. mcp_config = {
    17.     "sqlite": {
    18.         "command": "uvx",
    19.         "args": ["mcp-server-sqlite", "--db-path", "./data.db"]
    20.     },
    21.     "filesystem": {
    22.         "command": "npx",
    23.         "args": ["-y", "@anthropic/mcp-server-filesystem", "./workspace"]
    24.     }
    25. }
    26. # 3. 创建混合 Agent
    27. async def create_hybrid_agent():
    28.     async with MultiServerMCPClient(mcp_config) as mcp_client:
    29.         # 获取 MCP 工具
    30.         mcp_tools = mcp_client.get_tools()
    31.         
    32.         # 合并所有工具
    33.         all_tools = [calculate] + mcp_tools
    34.         
    35.         # 创建 Agent
    36.         llm = ChatOpenAI(model="gpt-4o")
    37.         agent = create_react_agent(llm, all_tools)
    38.         
    39.         # 执行查询
    40.         result = await agent.ainvoke({
    41.             "messages": [{"role": "user", "content": "计算 123 * 456,然后查询数据库中的商品总数"}]
    42.         })
    43.         
    44.         return result
    45. # 运行
    46. asyncio.run(create_hybrid_agent())
    复制代码
    3.3 工具路由机制

    flowchart TB    subgraph ToolDispatcher["Tool Node 分发逻辑"]        Input["接收 tool_calls"]        Check{"检查工具类型"}                Input --> Check                Check -->|"本地工具"| LocalPath["本地调用路径"]        Check -->|"MCP 工具"| MCPPath["MCP 调用路径"]                subgraph LocalPath["本地执行"]            LocalLookup["符号表查找"]            LocalExec["函数执行"]            LocalResult["返回结果"]            LocalLookup --> LocalExec --> LocalResult        end                subgraph MCPPath["MCP 执行"]            FindServer["定位 Server"]            Serialize["序列化请求"]            Transport["协议传输"]            Deserialize["反序列化响应"]            FindServer --> Serialize --> Transport --> Deserialize        end                LocalResult --> Merge["合并结果"]        Deserialize --> Merge        Merge --> Output["返回 ToolMessage 列表"]    end        style ToolDispatcher fill:#fafafa,stroke:#424242    style LocalPath fill:#fff9c4,stroke:#fbc02d    style MCPPath fill:#c8e6c9,stroke:#388e3c3.4 状态流转图

    stateDiagram-v2   
  • --> AgentNode: 用户输入        AgentNode --> ShouldContinue: LLM 响应        ShouldContinue --> ToolNode: 有 tool_calls    ShouldContinue -->
  • : 无 tool_calls (最终回答)        state ToolNode {        
  • --> Dispatch        Dispatch --> NativeExec: 本地工具        Dispatch --> MCPExec: MCP 工具        NativeExec --> Collect        MCPExec --> Collect        Collect -->
  •     }        ToolNode --> AgentNode: ToolMessage        note right of AgentNode        LLM 执行推理        生成 Thought + Action    end note        note right of ToolNode        执行工具调用        收集 Observation    end note第四章:生产环境考量

    4.1 性能对比分析

    graph LR    subgraph Latency["延迟构成对比"]        subgraph Native["原生调用 ~0.1ms"]            N1["函数查找: 0.01ms"]            N2["参数传递: 0.01ms"]            N3["执行: ~0.08ms"]        end                subgraph MCP["MCP 调用 ~5-50ms"]            M1["序列化: 0.5ms"]            M2["传输: 1-10ms"]            M3["反序列化: 0.5ms"]            M4["执行: ~3-40ms"]        end    end        style Native fill:#c8e6c9,stroke:#2e7d32    style MCP fill:#bbdefb,stroke:#1976d2指标原生 @toolMCP (Stdio)MCP (SSE/HTTP)冷启动无50-200ms100-500ms单次调用0.01-0.1ms2-10ms10-50ms并发能力受 GIL 限制进程级并行高度可扩展内存隔离无完全隔离完全隔离4.2 安全架构

    graph TB    subgraph SecurityLayers["安全防护层级"]        subgraph L1["Layer 1: 输入验证"]            Schema["JSON Schema 校验"]            Sanitize["输入消毒"]        end                subgraph L2["Layer 2: 传输安全"]            Auth["认证机制"]            Encrypt["传输加密 (TLS)"]        end                subgraph L3["Layer 3: 执行隔离"]            Sandbox["进程沙箱"]            Resource["资源限制 (CPU/Memory)"]            Timeout["超时控制"]        end                subgraph L4["Layer 4: 审计追踪"]            Logging["调用日志"]            Metrics["指标监控"]        end    end        L1 --> L2 --> L3 --> L4        style L1 fill:#ffcdd2,stroke:#c62828    style L2 fill:#fff9c4,stroke:#f9a825    style L3 fill:#c8e6c9,stroke:#2e7d32    style L4 fill:#bbdefb,stroke:#1976d24.3 部署拓扑选型

    graph TB    subgraph Development["开发环境"]        Dev["单机部署"]        DevAgent["Agent"]        DevMCP["MCP Server (Stdio)"]                DevAgent -->|"Stdio Pipe"| DevMCP    end        subgraph Staging["测试环境"]        StagingLB["负载均衡"]        StagingAgent1["Agent 1"]        StagingAgent2["Agent 2"]        StagingMCP["MCP Server Pool"]                StagingLB --> StagingAgent1        StagingLB --> StagingAgent2        StagingAgent1 -->|"SSE"| StagingMCP        StagingAgent2 -->|"SSE"| StagingMCP    end        subgraph Production["生产环境"]        ProdLB["负载均衡集群"]        ProdAgents["Agent 集群
    (K8s Deployment)"]        ProdMesh["Service Mesh"]        ProdMCPs["MCP Server 微服务群"]                ProdLB --> ProdAgents        ProdAgents --> ProdMesh        ProdMesh --> ProdMCPs    end        Development -->|"升级"| Staging    Staging -->|"升级"| Production        style Development fill:#e8f5e9,stroke:#2e7d32    style Staging fill:#fff3e0,stroke:#e65100    style Production fill:#e3f2fd,stroke:#1565c0第五章:总结与展望

    5.1 核心架构结论

    mindmap    root((LLM 工具调用架构))        原生调用            紧耦合            低延迟            适合原型开发        MCP 协议            松耦合            进程隔离            多语言支持            适合企业部署        ReAct 集成            认知层 + 执行层分离            错误即观察            自愈能力        最佳实践            混合架构            分层安全            渐进式迁移5.2 演进路线图

    timeline    title LLM 工具调用技术演进        section 2023        Function Calling : OpenAI 引入结构化工具调用        LangChain Tools : @tool 装饰器标准化        section 2024        MCP 发布 : Anthropic 开源 Model Context Protocol        LangGraph 集成 : MCP Adapter 发布        生态扩展 : 数十个官方 MCP Server        section 2025+        Streamable HTTP : 新传输层标准        工具市场 : Tool-as-a-Service 生态        安全标准 : 企业级认证授权框架5.3 技术选型指南

    场景推荐方案理由快速原型验证原生 @tool开发速度快,调试方便多语言团队协作MCP工具开发与 Agent 开发解耦高安全性要求MCP + 沙箱进程隔离,故障不扩散极致性能要求原生 @tool消除序列化和网络开销工具复用需求MCP一次开发,多处使用渐进式迁移混合架构两种模式可共存附录 :术语表

    术语定义Tool CallingLLM 生成结构化工具调用请求的能力MCPModel Context Protocol,工具调用的标准化协议ReActReasoning + Acting,交替推理与行动的认知框架JSON-RPC 2.0MCP 底层使用的远程过程调用协议Stdio Transport基于标准输入输出的进程间通信方式SSE Transport基于 Server-Sent Events 的 HTTP 长连接通信ToolMessageLangChain 中封装工具执行结果的消息类型
    来源:程序园用户自行投稿发布,如果侵权,请联系站长删除
    免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!
  • 相关推荐

    您需要登录后才可以回帖 登录 | 立即注册