炸裂副标题：从"模型说我要用工具"到"工具真的跑起来"——手把手拆解Tool Calls的生死一跃，看完这篇你终于能写出让AI真正干活的代码了！

全文总结：Tool Calls是大模型Agent的"手"和"脚"，但90%的新手都卡在这道坎上——模型返回了工具调用指令，你却不知道怎么接、怎么跑、怎么把结果喂回去。本文将用6个关键要点，从协议解析到消息组装，从错误兜底到并发优化，带你走完Tool Calls的完整闭环，让你的Agent从"嘴炮王者"变身"实干家"。

文字目录：

• 要点一：协议解析——读懂模型的工具调用语言
• 要点二：工具路由——让正确的函数找到正确的入口
• 要点三：参数校验——别让脏数据毁了你的函数
• 要点四：执行引擎——安全地跑起来，优雅地收结果
• 要点五：结果封装——把工具的输出变成模型能懂的话
• 要点六：消息闭环——构建新消息，开启下一轮对话

---

嗨，大家好呀，我是你的老朋友精通代码大仙。接下来我们一起学习 《大模型应用开发_动手做AI_Agent》，震撼你的学习轨迹！

---

“知道很多道理，却依然过不好这一生”——这句话放在AI Agent开发上，简直扎心到骨子里。

你是不是也这样：看了无数教程，知道大模型能调用工具，甚至能对着API文档把function calling的参数背得滚瓜烂熟。可一旦真动手写代码，模型明明说了"我要调用天气查询"，你却愣在原地——这串JSON我从哪取？取完往哪送？送完怎么告诉模型结果？

这种"知道概念，卡在落地"的窒息感，我懂。Tool Calls就像一道隐形的门槛，前面是"会调API的新手"，后面是"能写Agent的老司机"。今天这篇，我们就把这道门槛踏平，从协议解析到消息闭环，走完Tool Calls的完整链路。

---

要点一：协议解析——读懂模型的工具调用语言

点题

Tool Calls的第一步，是读懂模型在说什么。当大模型决定使用工具时，它不会直接执行代码，而是返回一段结构化的"指令"——通常包含工具名、参数、以及一个唯一标识。你的任务，就是把这个"天书"翻译成程序能处理的结构。

痛点分析

新手最容易栽在三个坑里：

坑一：以为模型会直接执行工具

很多人第一次用OpenAI的API，看到返回里有tool_calls字段，就以为工具已经自动跑完了。错！模型只是声明意图，执行权完全在你手里。

# 错误示范：以为finish_reason='tool_calls'时工具已执行
response = client.chat.completions.create(
    model="gpt-4",
    messages=messages,
    tools=tools
)

# 错！这里只是模型"说"要调用，根本没执行
if response.choices[0].finish_reason == "tool_calls":
    print("工具已执行！")  # 大乌龙！
    return response.choices[0].message.content  # 返回的是空或None

坑二：忽略tool_calls是数组

复杂任务里，模型可能一次要调多个工具。新手只取第一个，后面的全丢了。

# 错误示范：只取第一个工具调用
tool_call = response.choices[0].message.tool_calls[0]  # 后面的呢？丢了！

坑三：JSON解析不做异常处理

模型偶尔会吐出格式不合法的JSON（比如多了个逗号、少了引号），直接json.loads()就崩溃。

解决方案/正确做法

正确的解析流程要防御性编程，假设模型可能犯各种错：

import json
from typing import List, Dict, Any, Optional

class ToolCallParser:
    """安全解析模型的工具调用指令"""
    
    @staticmethod
    def parse(response) -> List[Dict[str, Any]]:
        """
        从模型响应中提取所有工具调用
        返回: [{"id": "...", "name": "...", "arguments": {...}}, ...]
        """
        result = []
        
        # 第一步：检查finish_reason
        choice = response.choices[0]
        if choice.finish_reason != "tool_calls":
            return result  # 没有工具调用，安全返回空列表
        
        message = choice.message
        
        # 第二步：检查tool_calls字段存在且是列表
        if not hasattr(message, 'tool_calls') or message.tool_calls is None:
            return result
        
        if not isinstance(message.tool_calls, list):
            print(f"警告：tool_calls不是列表，类型为{type(message.tool_calls)}")
            return result
        
        # 第三步：逐个解析每个工具调用
        for tc in message.tool_calls:
            try:
                # 提取基础字段
                call_id = getattr(tc, 'id', None)
                function = getattr(tc, 'function', None)
                
                if not function:
                    continue
                
                name = getattr(function, 'name', None)
                arguments_str = getattr(function, 'arguments', '{}')
                
                # 安全解析JSON参数
                try:
                    arguments = json.loads(arguments_str)
                except json.JSONDecodeError as e:
                    print(f"工具 {name} 的参数JSON解析失败: {e}")
                    # 尝试修复常见错误
                    arguments = ToolCallParser._fix_json(arguments_str)
                
                result.append({
                    "id": call_id,
                    "name": name,
                    "arguments": arguments,
                    "raw": tc  # 保留原始对象备用
                })
                
            except Exception as e:
                print(f"解析单个tool_call失败: {e}")
                continue  # 跳过坏的，继续处理其他的
        
        return result
    
    @staticmethod
    def _fix_json(bad_json: str) -> Dict:
        """应急修复常见JSON格式错误"""
        # 去掉尾部逗号
        bad_json = bad_json.strip().rstrip(',')
        # 尝试再次解析
        try:
            return json.loads(bad_json)
        except:
            # 实在修不好，返回空对象
            print(f"无法修复的JSON: {bad_json[:100]}...")
            return {}

使用示例：

# 正确示范：完整解析流程
parser = ToolCallParser()
tool_calls = parser.parse(response)

if not tool_calls:
    # 没有工具调用，直接拿文本回复
    answer = response.choices[0].message.content
    print(f"模型直接回答: {answer}")
else:
    print(f"模型请求调用 {len(tool_calls)} 个工具")
    for call in tool_calls:
        print(f"  - {call['name']}: {call['arguments']}")

小结

解析Tool Calls要像拆快递——先检查有没有（finish_reason），再检查有几个（遍历数组），最后小心拆开（异常处理）。模型是合作伙伴，不是可靠的数据库，永远做防御性编程。

---

要点二：工具路由——让正确的函数找到正确的入口

点题

拿到工具调用指令后，你需要一个路由系统，把name="get_weather"映射到实际的get_weather(city)函数。这看似简单的"字符串匹配"，实则暗藏工程设计的智慧。

痛点分析

坑一：硬编码if-else地狱

新手最容易写的代码：

# 错误示范：硬编码路由，维护噩梦
def execute_tool(name, arguments):
    if name == "get_weather":
        return get_weather(**arguments)
    elif name == "search_db":
        return search_db(**arguments)
    elif name == "send_email":
        return send_email(**arguments)
    # ... 每加一个工具就要改这里
    else:
        return "未知工具"

工具一多，这函数膨胀到没法看。而且字符串拼写错误编译期发现不了，运行时才发现get_wather不存在。

坑二：函数签名和schema不一致

你注册时写的参数是city: str，但模型传过来{"city_name": "北京"}，直接**arguments就炸。

坑三：没有权限控制

所有工具平铺直叙，敏感操作（如删除数据）和普通查询混在一起，没有分级管控。

解决方案/正确做法

设计一个可注册、可校验、可扩展的工具路由系统：

from functools import wraps
from typing import Callable, Dict, Any, Optional
from dataclasses import dataclass
import inspect

@dataclass
class ToolMetadata:
    """工具元数据"""
    name: str
    description: str
    func: Callable
    parameters_schema: Dict[str, Any]
    required_permission: str = "basic"  # 权限分级

class ToolRouter:
    """工具路由中心"""
    
    def __init__(self):
        self._registry: Dict[str, ToolMetadata] = {}
        self._permission_levels = {
            "basic": 0,
            "advanced": 1,
            "admin": 2
        }
    
    def register(self, 
                 name: Optional[str] = None,
                 description: Optional[str] = None,
                 permission: str = "basic"):
        """装饰器：注册工具到路由系统"""
        def decorator(func: Callable):
            tool_name = name or func.__name__
            tool_desc = description or func.__doc__ or "No description"
            
            # 自动提取函数签名生成schema
            sig = inspect.signature(func)
            schema = {
                "type": "object",
                "properties": {},
                "required": []
            }
            
            for param_name, param in sig.parameters.items():
                param_info = {"type": "string"}  # 简化处理，实际可更精细
                
                # 检查是否有类型注解
                if param.annotation != inspect.Parameter.empty:
                    if param.annotation == int:
                        param_info["type"] = "integer"
                    elif param.annotation == float:
                        param_info["type"] = "number"
                    elif param.annotation == bool:
                        param_info["type"] = "boolean"
                
                # 检查是否有默认值
                if param.default == inspect.Parameter.empty:
                    schema["required"].append(param_name)
                
                schema["properties"][param_name] = param_info
            
            # 注册到路由表
            self._registry[tool_name] = ToolMetadata(
                name=tool_name,
                description=tool_desc,
                func=func,
                parameters_schema=schema,
                required_permission=permission
            )
            
            print(f"✓ 注册工具: {tool_name}")
            return func
        return decorator
    
    def get_openai_tools_format(self) -> List[Dict]:
        """生成OpenAI API需要的tools参数格式"""
        return [
            {
                "type": "function",
                "function": {
                    "name": meta.name,
                    "description": meta.description,
                    "parameters": meta.parameters_schema
                }
            }
            for meta in self._registry.values()
        ]
    
    def execute(self, 
                name: str, 
                arguments: Dict[str, Any],
                user_permission: str = "basic") -> Dict[str, Any]:
        """执行指定工具"""
        # 1. 检查工具存在
        if name not in self._registry:
            return {
                "success": False,
                "error": f"工具 '{name}' 未注册。可用工具: {list(self._registry.keys())}"
            }
        
        meta = self._registry[name]
        
        # 2. 权限检查
        if self._permission_levels.get(user_permission, 0) < \
           self._permission_levels.get(meta.required_permission, 0):
            return {
                "success": False,
                "error": f"权限不足，需要 {meta.required_permission} 级别"
            }
        
        # 3. 参数校验（基础版）
        required = meta.parameters_schema.get("required", [])
        missing = [p for p in required if p not in arguments]
        if missing:
            return {
                "success": False,
                "error": f"缺少必需参数: {missing}"
            }
        
        # 4. 执行并捕获异常
        try:
            result = meta.func(**arguments)
            return {
                "success": True,
                "result": result,
                "tool_name": name
            }
        except Exception as e:
            return {
                "success": False,
                "error": f"工具执行异常: {str(e)}"
            }
    
    def list_tools(self) -> List[str]:
        """列出所有可用工具"""
        return [f"{m.name} [{m.required_permission}]" 
                for m in self._registry.values()]


# ========== 使用示例 ==========

router = ToolRouter()

@router.register(description="查询指定城市的天气")
def get_weather(city: str, date: Optional[str] = None) -> str:
    """模拟天气查询"""
    return f"{city} {'今天' if date is None else date}天气晴朗，25°C"

@router.register(description="发送邮件", permission="advanced")
def send_email(to: str, subject: str, body: str) -> str:
    """模拟发送邮件"""
    return f"邮件已发送至 {to}: {subject}"

# 查看注册的工具
print(router.list_tools())
# ['get_weather [basic]', 'send_email [advanced]']

# 获取OpenAI格式的工具描述
tools_for_api = router.get_openai_tools_format()

# 执行工具调用
result = router.execute(
    name="get_weather",
    arguments={"city": "北京"},
    user_permission="basic"
)
print(result)
# {'success': True, 'result': '北京 今天天气晴朗，25°C', 'tool_name': 'get_weather'}

小结

好的路由系统像乐高积木——每个工具独立封装，通过装饰器自动注册，权限和校验内建其中。告别if-else地狱，新增工具只需写函数+加装饰器，零侵入原有代码。

---

要点三：参数校验——别让脏数据毁了你的函数

点题

模型生成的参数，永远不要直接信任。它可能漏传必填字段，可能把数字写成字符串，可能给出完全不存在的枚举值。参数校验是Tool Calls的"防火墙"，守住最后一道数据质量关。

痛点分析

坑一：类型假设崩溃

# 错误示范：假设模型给的price是数字
def query_products(price: float):
    # 模型可能传 "100"（字符串）或 "便宜点"（无法解析）
    return db.filter(price__lt=price)  # 字符串比较会出奇葩结果

# 调用时
query_products(price="100")  # 可能能跑，但逻辑错了！

坑二：注入攻击风险

# 危险！如果arguments来自模型，且直接拼SQL
def search_user(name: str):
    query = f"SELECT * FROM users WHERE name = '{name}'"
    # 模型可能生成: name = "'; DROP TABLE users; --"
    return db.execute(query)  # 完犊子

坑三：业务规则被绕过

# 模型生成的日期可能是"2023-13-45"这种无效日期
# 或者预约时间已经过期，但函数照样执行

解决方案/正确做法

使用Pydantic做声明式校验，把脏数据挡在函数外：

from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime, timedelta
import re

class WeatherQuery(BaseModel):
    """天气查询参数模型"""
    city: str = Field(
        ...,  # 必填
        min_length=2,
        max_length=50,
        description="城市名称"
    )
    date: Optional[str] = Field(
        None,
        pattern=r"^\d{4}-\d{2}-\d{2}$",  # YYYY-MM-DD格式
        description="日期，默认为今天"
    )
    temperature_unit: str = Field(
        "celsius",
        pattern=r"^(celsius|fahrenheit)$"
    )
    
    @validator('city')
    def city_must_be_chinese_or_english(cls, v):
        """校验城市名只包含中英文"""
        if not re.match(r'^[\u4e00-\u9fa5a-zA-Z\s]+$', v):
            raise ValueError('城市名只能包含中文或英文字符')
        return v.strip()
    
    @validator('date')
    def date_must_be_valid(cls, v):
        """校验日期有效性"""
        if v is None:
            return v
        try:
            dt = datetime.strptime(v, "%Y-%m-%d")
            # 不能查询超过30天后的天气
            if dt > datetime.now() + timedelta(days=30):
                raise ValueError('只能查询未来30天内的天气')
            # 不能查询过去的天气（除非你的API支持）
            if dt < datetime.now().replace(hour=0, minute=0, second=0, microsecond=0):
                raise ValueError('不能查询过去的天气')
            return v
        except ValueError as e:
            if "unconverted data remains" in str(e):
                raise ValueError('日期格式错误，应为YYYY-MM-DD')
            raise


class SafeToolExecutor:
    """带校验的安全执行器"""
    
    def __init__(self, router: ToolRouter):
        self.router = router
        self._validators = {}  # 工具名 -> Pydantic模型
    
    def register_validator(self, tool_name: str, model_class: type[BaseModel]):
        """为工具注册参数校验模型"""
        self._validators[tool_name] = model_class
    
    def safe_execute(self, 
                     tool_name: str, 
                     raw_arguments: Dict[str, Any],
                     user_permission: str = "basic") -> Dict[str, Any]:
        """带完整校验的执行流程"""
        
        # 第一步：基础路由检查
        if tool_name not in self.router._registry:
            return {
                "success": False,
                "error": f"未知工具: {tool_name}",
                "error_type": "TOOL_NOT_FOUND"
            }
        
        # 第二步：Pydantic校验（如果有注册）
        validator_class = self._validators.get(tool_name)
        if validator_class:
            try:
                validated = validator_class(**raw_arguments)
                clean_arguments = validated.model_dump(exclude_none=True)
                print(f"✓ 参数校验通过: {clean_arguments}")
            except Exception as e:
                # 提取友好的错误信息
                error_msg = self._format_pydantic_error(e)
                return {
                    "success": False,
                    "error": f"参数校验失败: {error_msg}",
                    "error_type": "VALIDATION_ERROR",
                    "raw_arguments": raw_arguments
                }
        else:
            # 没有专用校验器，做基础清理
            clean_arguments = self._basic_sanitize(raw_arguments)
        
        # 第三步：执行
        return self.router.execute(
            tool_name, 
            clean_arguments, 
            user_permission
        )
    
    def _format_pydantic_error(self, error) -> str:
        """格式化Pydantic错误为可读文本"""
        errors = error.errors() if hasattr(error, 'errors') else [str(error)]
        messages = []
        for e in errors:
            if isinstance(e, dict):
                loc = ".".join(str(x) for x in e.get('loc', []))
                msg = e.get('msg', '未知错误')
                messages.append(f"{loc}: {msg}")
            else:
                messages.append(str(e))
        return "; ".join(messages)
    
    def _basic_sanitize(self, args: Dict) -> Dict:
        """基础清理：去空值、截断长字符串"""
        result = {}
        for k, v in args.items():
            if v is None:
                continue
            if isinstance(v, str):
                v = v.strip()[:1000]  # 截断超长字符串
            result[k] = v
        return result


# ========== 使用示例 ==========

safe_executor = SafeToolExecutor(router)

# 为天气工具注册校验器
safe_executor.register_validator("get_weather", WeatherQuery)

# 测试各种脏数据
test_cases = [
    {"city": "北京", "date": "2024-06-15"},  # 正常
    {"city": "北", "date": "2024-06-15"},    # 城市名太短
    {"city": "北京", "date": "2024-13-45"},  # 无效日期
    {"city": "北京'; DROP TABLE cities; --"},  # 注入尝试
    {"city": "上海", "date": "2023-01-01"},  # 过去日期
]

for args in test_cases:
    print(f"\n输入: {args}")
    result = safe_executor.safe_execute("get_weather", args)
    print(f"结果: {result.get('success') and '成功' or result.get('error')}")

小结

信任模型，但验证一切。Pydantic让参数校验从"体力活"变成"声明式配置"，类型、格式、业务规则一层层过滤，脏数据进不来，你的函数才能睡得安稳。

---

要点四：执行引擎——安全地跑起来，优雅地收结果

点题

参数准备好了，真正的执行环节却危机四伏：工具可能卡住不动，可能抛出异常，可能返回几MB的超大结果把上下文撑爆。执行引擎要跑得稳、收得住、断得了。

痛点分析

坑一：同步阻塞导致整个Agent卡死

# 错误示范：同步调用可能阻塞数分钟
def search_knowledge_base(query: str):
    # 这个操作可能耗时5分钟！
    return heavy_vector_search(query)

# Agent线程被卡死，用户只能干瞪眼

坑二：异常吞掉或崩溃

# 错误示范：裸奔执行
result = some_external_api(**args)  # 一炸全炸，没有降级

坑三：结果太大撑爆上下文

# 错误示范：直接把原始结果塞给模型
tool_result = search_db("python")  # 返回1000条记录
messages.append({
    "role": "tool",
    "content": str(tool_result)  # 几万字塞进去，token爆炸
})

解决方案/正确做法

构建异步、超时、降级、截断四位一体的执行引擎：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from dataclasses import dataclass
from enum import Enum
import traceback

class ExecutionStatus(Enum):
    SUCCESS = "success"
    TIMEOUT = "timeout"
    ERROR = "error"
    PARTIAL = "partial"  # 结果截断

@dataclass
class ExecutionResult:
    status: ExecutionStatus
    data: Any
    execution_time_ms: float
    message: str = ""
    original_length: Optional[int] = None  # 截断前的长度

class SafeExecutionEngine:
    """安全执行引擎"""
    
    def __init__(self, 
                 default_timeout: float = 30.0,
                 max_result_length: int = 2000,
                 max_workers: int = 10):
        self.default_timeout = default_timeout
        self.max_result_length = max_result_length
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self._execution_stats = []  # 用于监控
    
    def execute(self,
                func: Callable,
                args: Dict[str, Any],
                timeout: Optional[float] = None,
                tool_name: str = "unknown") -> ExecutionResult:
        """
        同步接口：在后台线程执行，主线程等待结果
        """
        timeout = timeout or self.default_timeout
        start_time = time.time()
        
        future = self.executor.submit(func, **args)
        
        try:
            raw_result = future.result(timeout=timeout)
            execution_time = (time.time() - start_time) * 1000
            
            # 截断处理
            processed, was_truncated, original_len = self._truncate_result(raw_result)
            
            self._record_stats(tool_name, ExecutionStatus.SUCCESS, execution_time)
            
            return ExecutionResult(
                status=ExecutionStatus.PARTIAL if was_truncated else ExecutionStatus.SUCCESS,
                data=processed,
                execution_time_ms=execution_time,
                message="结果已截断" if was_truncated else "执行成功",
                original_length=original_len
            )
            
        except FutureTimeout:
            execution_time = (time.time() - start_time) * 1000
            self._record_stats(tool_name, ExecutionStatus.TIMEOUT, execution_time)
            
            # 尝试取消任务
            future.cancel()
            
            return ExecutionResult(
                status=ExecutionStatus.TIMEOUT,
                data=None,
                execution_time_ms=execution_time,
                message=f"执行超时（>{timeout}秒），请简化请求或稍后重试"
            )
            
        except Exception as e:
            execution_time = (time.time() - start_time) * 1000
            self._record_stats(tool_name, ExecutionStatus.ERROR, execution_time)
            
            error_detail = traceback.format_exc()
            print(f"工具执行异常:\n{error_detail[:500]}")
            
            return ExecutionResult(
                status=ExecutionStatus.ERROR,
                data=None,
                execution_time_ms=execution_time,
                message=f"执行错误: {str(e)}"
            )
    
    async def execute_async(self,
                           func: Callable,
                           args: Dict[str, Any],
                           timeout: Optional[float] = None) -> ExecutionResult:
        """
        异步接口：用于asyncio环境
        """
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.executor,
            self.execute,
            func, args, timeout
        )
    
    def _truncate_result(self, raw: Any) -> tuple[Any, bool, Optional[int]]:
        """
        智能截断结果
        """
        # 统一转为字符串处理
        if not isinstance(raw, str):
            try:
                text = json.dumps(raw, ensure_ascii=False, indent=2)
            except:
                text = str(raw)
        else:
            text = raw
        
        original_len = len(text)
        
        if len(text) <= self.max_result_length:
            return raw, False, original_len
        
        # 智能截断策略
        truncated = text[:self.max_result_length]
        
        # 尝试在句子边界截断
        last_period = truncated.rfind('。')
        last_newline = truncated.rfind('\n')
        cut_point = max(last_period, last_newline, self.max_result_length - 100)
        
        if cut_point > self.max_result_length * 0.8:
            truncated = truncated[:cut_point + 1]
        
        truncated += f"\n\n[结果已截断，原始长度{original_len}字符]"
        
        return truncated, True, original_len
    
    def _record_stats(self, tool_name: str, status: ExecutionStatus, time_ms: float):
        """记录执行统计"""
        self._execution_stats.append({
            "tool": tool_name,
            "status": status.value,
            "time_ms": time_ms,
            "timestamp": time.time()
        })
        # 只保留最近100条
        self._execution_stats = self._execution_stats[-100:]
    
    def get_stats(self) -> Dict:
        """获取执行统计"""
        if not self._execution_stats:
            return {"message": "暂无统计数据"}
        
        total = len(self._execution_stats)
        success = sum(1 for s in self._execution_stats if s["status"] == "success")
        timeout = sum(1 for s in self._execution_stats if s["status"] == "timeout")
        error = sum(1 for s in self._execution_stats if s["status"] == "error")
        
        avg_time = sum(s["time_ms"] for s in self._execution_stats) / total
        
        return {
            "total_executions": total,
            "success_rate": f"{success/total*100:.1f}%",
            "timeout_rate": f"{timeout/total*100:.1f}%",
            "error_rate": f"{error/total*100:.1f}%",
            "avg_time_ms": f"{avg_time:.1f}"
        }


# ========== 使用示例 ==========

engine = SafeExecutionEngine(
    default_timeout=10.0,      # 默认10秒超时
    max_result_length=1500     # 最大结果1500字符
)

# 模拟一个慢工具
def slow_search(query: str):
    time.sleep(15)  # 模拟慢查询
    return f"搜索结果: {query}"

# 模拟一个返回大结果的工具有
def big_data_query(table: str):
    return {"records": [{"id": i, "data": "x" * 100} for i in range(100)]}

# 测试超时
print("测试超时...")
result = engine.execute(slow_search, {"query": "test"}, tool_name="slow_search")
print(f"状态: {result.status.value}, 消息: {result.message}")

# 测试截断
print("\n测试截断...")
result = engine.execute(big_data_query, {"table": "users"}, tool_name="big_data_query")
print(f"状态: {result.status.value}, 截断前: {result.original_length}, 截断后: {len(str(result.data))}")

# 查看统计
print(f"\n执行统计: {engine.get_stats()}")

小结

执行引擎是Tool Calls的减震器——超时防止卡死，异常捕获防崩溃，截断保护上下文。记住：对外部工具的调用，要像调用第三方API一样谨慎，因为你控制不了它的内部实现。

---

要点五：结果封装——把工具的输出变成模型能懂的话

点题

工具执行完了，但结果不能直接塞给模型。你需要格式化、结构化、添加上下文，让模型理解"这个结果是干嘛的"，并能基于它继续推理。

痛点分析

坑一：直接塞原始对象

# 错误示范：把Python对象直接当content
messages.append({
    "role": "tool",
    "tool_call_id": call_id,
    "content": {"temperature": 25, "humidity": 60}  # 错了！必须是字符串
})

坑二：错误信息太技术化

# 错误示范：把异常堆栈直接给模型看
error_msg = traceback.format_exc()  # 几百行堆栈
messages.append({
    "role": "tool",
    "content": f"错误: {error_msg}"  # 模型看懵了，用户也看懵了
})

坑三：丢失工具调用的关联

# 错误示范：忘记tool_call_id
messages.append({
    "role": "tool",
    # 没有tool_call_id！模型不知道这个结果对应哪个调用
    "content": "天气晴朗"
})

解决方案/正确做法

设计智能封装层，根据结果类型自动选择最佳呈现方式：

from typing import Union

class ResultFormatter:
    """结果格式化器"""
    
    # 不同结果类型的格式化模板
    TEMPLATES = {
        "weather": "【天气查询结果】\n城市: {city}\n天气: {condition}\n温度: {temperature}°C\n湿度: {humidity}%\n更新时间: {update_time}",
        "search": "【搜索结果】共找到{count}条相关记录：\n{items}\n{truncation_notice}",
        "database": "【数据库查询】\n影响行数: {affected_rows}\n返回数据: {preview}",
        "error": "【执行失败】{error_type}\n问题描述: {message}\n建议: {suggestion}"
    }
    
    def __init__(self, default_template: str = "generic"):
        self.default_template = default_template
    
    def format_for_model(self,
                         execution_result: ExecutionResult,
                         tool_call_id: str,
                         tool_name: str,
                         original_arguments: Dict) -> Dict[str, str]:
        """
        将执行结果格式化为标准的tool消息格式
        """
        if execution_result.status == ExecutionStatus.ERROR:
            content = self._format_error(
                execution_result, 
                tool_name, 
                original_arguments
            )
        elif execution_result.status == ExecutionStatus.TIMEOUT:
            content = self._format_timeout(
                execution_result, 
                tool_name
            )
        else:
            content = self._format_success(
                execution_result,
                tool_name,
                original_arguments
            )
        
        # 构建标准tool消息
        return {
            "role": "tool",
            "tool_call_id": tool_call_id,  # 关键！必须对应上
            "name": tool_name,
            "content": content
        }
    
    def _format_success(self,
                        result: ExecutionResult,
                        tool_name: str,
                        arguments: Dict) -> str:
        """格式化成功结果"""
        data = result.data
        
        # 尝试匹配专用模板
        template = self.TEMPLATES.get(tool_name)
        if template and isinstance(data, dict):
            try:
                return template.format(**data)
            except KeyError:
                pass  # 字段不匹配，回退到通用格式
        
        # 通用格式化
        lines = [
            f"【{tool_name} 执行成功】",
            f"执行时间: {result.execution_time_ms:.0f}ms",
        ]
        
        if result.original_length and result.original_length != len(str(data)):
            lines.append(f"数据说明: 原始{result.original_length}字符，已智能截断")
        
        lines.append("---")
        
        # 根据数据类型选择展示方式
        if isinstance(data, (list, dict)):
            lines.append(json.dumps(data, ensure_ascii=False, indent=2))
        else:
            lines.append(str(data))
        
        return "\n".join(lines)
    
    def _format_error(self,
                      result: ExecutionResult,
                      tool_name: str,
                      arguments: Dict) -> str:
        """格式化错误结果——给模型看的友好版本"""
        error_msg = result.message
        
        # 分类错误并给出建议
        if "VALIDATION_ERROR" in error_msg:
            error_type = "参数校验失败"
            suggestion = "请检查参数格式，特别是日期、数字等字段。如需帮助，可以询问用户确认。"
        elif "TIMEOUT" in error_msg:
            error_type = "执行超时"
            suggestion = "请求过于复杂或外部服务响应慢，建议简化查询条件或稍后重试。"
        elif "权限" in error_msg:
            error_type = "权限不足"
            suggestion = "当前操作需要更高权限，请引导用户联系管理员。"
        else:
            error_type = "执行异常"
            suggestion = "系统暂时无法处理该请求，建议尝试其他方式或稍后重试。"
        
        # 提取关键信息，避免堆栈吓坏模型
        clean_message = error_msg.split('\n')[0][:200]
        
        template = self.TEMPLATES["error"]
        return template.format(
            error_type=error_type,
            message=clean_message,
            suggestion=suggestion
        )
    
    def _format_timeout(self,
                        result: ExecutionResult,
                        tool_name: str) -> str:
        """格式化超时结果"""
        return (
            f"【{tool_name} 执行超时】\n"
            f"该操作预计需要较长时间（>{result.execution_time_ms/1000:.0f}秒），"
            f"已自动中断以保护响应速度。\n"
            f"建议：请缩小查询范围，或改用异步方式处理此请求。"
        )
    
    def format_for_display(self,
                          execution_result: ExecutionResult,
                          tool_name: str) -> str:
        """
        给用户看的简化版本（用于UI展示）
        """
        if execution_result.status != ExecutionStatus.SUCCESS:
            return f"⚠️ {tool_name}: {execution_result.message}"
        
        # 成功时只展示关键信息
        data = execution_result.data
        if isinstance(data, str) and len(data) > 100:
            return f"✓ {tool_name}: {data[:100]}..."
        return f"✓ {tool_name}: 执行成功"


# ========== 使用示例 ==========

formatter = ResultFormatter()

# 模拟一个成功的天气查询
weather_result = ExecutionResult(
    status=ExecutionStatus.SUCCESS,
    data={
        "city": "北京",
        "condition": "晴朗",
        "temperature": 28,
        "humidity": 45,
        "update_time": "2024-06-15 14:30"
    },
    execution_time_ms=234,
    message="执行成功"
)

tool_message = formatter.format_for_model(
    weather_result,
    tool_call_id="call_abc123",
    tool_name="get_weather",
    original_arguments={"city": "北京"}
)

print("=== 给模型的消息 ===")
print(json.dumps(tool_message, ensure_ascii=False, indent=2))

# 模拟一个错误
error_result = ExecutionResult(
    status=ExecutionStatus.ERROR,
    data=None,
    execution_time_ms=56,
    message="VALIDATION_ERROR: date: 日期格式错误，应为YYYY-MM-DD"
)

error_message = formatter.format_for_model(
    error_result,
    tool_call_id="call_def456",
    tool_name="get_weather",
    original_arguments={"city": "北京", "date": "昨天"}
)

print("\n=== 错误情况的消息 ===")
print(json.dumps(error_message, ensure_ascii=False, indent=2))

小结

结果封装是翻译工作——把机器的执行结果，翻译成模型能理解的上下文。记住三个关键：tool_call_id必须对上、错误信息要"说人话"、成功结果要结构化。这样模型才能基于工具输出，做出正确的下一步决策。

---

要点六：消息闭环——构建新消息，开启下一轮对话

点题

Tool Calls不是终点，而是新一轮对话的起点。你需要把模型的工具调用请求、工具的执行结果，按正确顺序组装成消息历史，再送回给模型，让它基于新信息继续推理或给出最终回答。

痛点分析

坑一：消息顺序错乱

# 错误示范：先加tool消息，后加assistant消息
messages.append({"role": "tool", ...})  # 错了！assistant消息必须在前面
messages.append({"role": "assistant", "tool_calls": [...]})

坑二：忘记保留assistant的原始内容

# 错误示范：只取tool_calls，丢了assistant的其他话
assistant_msg = response.choices[0].message
messages.append({
    "role": "assistant",
    "tool_calls": assistant_msg.tool_calls  # 如果assistant还有content，丢了！
})

坑三：无限循环不收敛

# 错误示范：没有终止条件，模型一直调工具
while True:
    response = client.chat.completions.create(...)
    if response.choices[0].finish_reason == "tool_calls":
        # 执行工具，继续循环
        # 但模型可能永远调下去！

解决方案/正确做法

实现消息管理器，确保消息顺序正确、支持多轮迭代、防止无限循环：

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field

@dataclass
class ConversationTurn:
    """单轮对话记录"""
    turn_number: int
    assistant_message: Dict  # 模型的回复（可能含tool_calls）
    tool_results: List[Dict] = field(default_factory=list)  # 工具执行结果
    final_response: Optional[str] = None  # 模型基于工具结果后的最终回复

class MessageManager:
    """消息历史管理器"""
    
    def __init__(self, 
                 max_turns: int = 5,  # 最大工具调用轮数
                 max_total_messages: int = 20):
        self.max_turns = max_turns
        self.max_total_messages = max_total_messages
        self.history: List[Dict] = []  # 完整消息历史
        self.turns: List[ConversationTurn] = []  # 按轮次组织的记录
        self.current_turn = 0
    
    def add_user_message(self, content: str):
        """添加用户消息"""
        self.history.append({
            "role": "user",
            "content": content
        })
        self._trim_history()
    
    def process_model_response(self, 
                                response,
                                tool_executor: SafeExecutionEngine,
                                router: ToolRouter,
                                formatter: ResultFormatter) -> Dict[str, Any]:
        """
        处理模型响应：如果有tool_calls则执行，构建完整消息闭环
        """
        self.current_turn += 1
        if self.current_turn > self.max_turns:
            return {
                "status": "max_turns_reached",
                "message": f"已达到最大工具调用轮数({self.max_turns})，强制终止",
                "history": self.history
            }
        
        choice = response.choices[0]
        message = choice.message
        
        # 提取assistant的完整消息（包括content和tool_calls）
        assistant_msg = self._extract_assistant_message(message)
        
        # 记录本轮
        turn = ConversationTurn(
            turn_number=self.current_turn,
            assistant_message=assistant_msg
        )
        
        # 添加到历史
        self.history.append(assistant_msg)
        
        # 检查是否需要工具调用
        if choice.finish_reason != "tool_calls":
            # 直接结束
            turn.final_response = message.content
            self.turns.append(turn)
            return {
                "status": "complete",
                "final_answer": message.content,
                "turns_used": self.current_turn
            }
        
        # 需要执行工具
        tool_calls = message.tool_calls if hasattr(message, 'tool_calls') else []
        tool_messages = []
        
        for tc in tool_calls:
            call_id = tc.id
            func = tc.function
            tool_name = func.name
            
            # 解析参数
            try:
                arguments = json.loads(func.arguments)
            except json.JSONDecodeError:
                arguments = {}
            
            print(f"  执行工具: {tool_name}({arguments})")
            
            # 执行
            exec_result = tool_executor.execute(
                router._registry[tool_name].func,
                arguments,
                tool_name=tool_name
            )
            
            # 格式化结果
            tool_msg = formatter.format_for_model(
                exec_result,
                call_id,
                tool_name,
                arguments
            )
            
            tool_messages.append(tool_msg)
            turn.tool_results.append({
                "tool_name": tool_name,
                "arguments": arguments,
                "result": exec_result
            })
        
        # 添加所有tool消息到历史
        self.history.extend(tool_messages)
        self.turns.append(turn)
        self._trim_history()
        
        return {
            "status": "need_continue",
            "message": f"已完成{len(tool_messages)}个工具调用，需要继续请求模型",
            "tool_messages": tool_messages,
            "current_history": self.history.copy()
        }
    
    def _extract_assistant_message(self, message) -> Dict:
        """完整提取assistant消息的所有字段"""
        msg = {
            "role": "assistant"
        }
        
        # content可能为null
        if hasattr(message, 'content') and message.content:
            msg["content"] = message.content
        
        # tool_calls
        if hasattr(message, 'tool_calls') and message.tool_calls:
            msg["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": tc.type,
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments
                    }
                }
                for tc in message.tool_calls
            ]
        
        return msg
    
    def _trim_history(self):
        """智能截断历史，保留关键信息"""
        if len(self.history) <= self.max_total_messages:
            return
        
        # 保留：系统消息、最近的用户消息、所有工具相关消息
        # 简化策略：保留后N条
        keep_count = self.max_total_messages - 2  # 留点余量
        
        # 但必须保留工具调用的完整链条
        # 找到最后一个user消息的位置
        last_user_idx = None
        for i in range(len(self.history) - 1, -1, -1):
            if self.history[i]["role"] == "user":
                last_user_idx = i
                break
        
        if last_user_idx is not None:
            # 保留从last_user_idx开始的所有消息
            self.history = self.history[max(0, last_user_idx - 2):]
    
    def get_history_for_api(self) -> List[Dict]:
        """获取适合API调用的历史（去掉内部字段）"""
        # API不需要某些内部字段
        clean_history = []
        for msg in self.history:
            clean_msg = {k: v for k, v in msg.items() 
                        if k in ["role", "content", "tool_calls", "tool_call_id", "name"]}
            clean_history.append(clean_msg)
        return clean_history
    
    def get_conversation_summary(self) -> str:
        """获取对话摘要（用于调试）"""
        lines = [f"=== 对话摘要 ({len(self.turns)}轮工具调用) ==="]
        for turn in self.turns:
            lines.append(f"\n第{turn.turn_number}轮:")
            lines.append(f"  Assistant: {turn.assistant_message.get('content', '[纯工具调用]')[:100]}")
            for tr in turn.tool_results:
                status = "✓" if tr["result"].status == ExecutionStatus.SUCCESS else "✗"
                lines.append(f"  {status} {tr['tool_name']}: {tr['result'].message[:80]}")
            if turn.final_response:
                lines.append(f"  最终回复: {turn.final_response[:100]}")
        return "\n".join(lines)


# ========== 完整使用示例 ==========

def run_agent_conversation(user_query: str, 
                           client,
                           router: ToolRouter,
                           engine: SafeExecutionEngine,
                           formatter: ResultFormatter):
    """运行完整的Agent对话"""
    
    # 初始化
    manager = MessageManager(max_turns=3)
    manager.add_user_message(user_query)
    
    # 系统提示
    system_msg = {
        "role": "system",
        "content": "你是一个 helpful 的助手，可以使用工具帮助用户。如果工具返回错误，请向用户解释原因。"
    }
    
    for iteration in range(5):  # 安全上限
        print(f"\n{'='*40}")
        print(f"迭代 {iteration + 1}")
        
        # 准备消息
        api_messages = [system_msg] + manager.get_history_for_api()
        
        # 调用模型
        response = client.chat.completions.create(
            model="gpt-4",
            messages=api_messages,
            tools=router.get_openai_tools_format(),
            tool_choice="auto"
        )
        
        # 处理响应
        result = manager.process_model_response(
            response, engine, router, formatter
        )
        
        print(f"状态: {result['status']}")
        
        if result['status'] == 'complete':
            print(f"\n🎉 最终答案: {result['final_answer']}")
            break
            
        elif result['status'] == 'max_turns_reached':
            print(f"\n⚠️ {result['message']}")
            break
            
        elif result['status'] == 'need_continue':
            print(f"工具调用完成，继续下一轮...")
            continue
    
    print(f"\n{manager.get_conversation_summary()}")
    return manager


# 模拟运行（实际需要有真实的client）
print("""
# 使用示例：
client = OpenAI(api_key="your-key")
manager = run_agent_conversation(
    "北京今天天气怎么样？",
    client,
    router,
    engine,
    formatter
)
""")

小结

消息闭环是Tool Calls的最后一公里——顺序要对（assistant在前，tool在后）、内容要全（别丢字段）、迭代要控（防无限循环）。好的消息管理器让多轮工具调用像搭积木一样清晰可控。

---

写在最后

Tool Calls这道坎，我踩过，你正在踩，后来的新人还会踩。但好消息是：它真的只是一道坎，不是一堵墙。

今天我们一起走完了完整链路——从解析模型的"工具意图"，到路由找到正确函数，到校验参数、安全执行、格式化结果，最后组装消息开启新一轮对话。这六个环节，环环相扣，缺一不可。

你可能会觉得，“不就调个工具吗，怎么这么复杂？” 但这就是工程化的代价，也是工程化的价值。简单的demo可以裸奔，但生产环境的Agent需要容错、可观测、可维护。你今天写的每一行防御代码，都是在给未来的自己省debug时间。

编程之路不易，但每一步成长都算数。Tool Calls从"卡住"到"打通"的那一刻，你会突然觉得自己和模型之间建立了真正的协作关系——不再是单向的问答，而是双向的、动态的、能解决问题的工作流。

保持好奇，持续动手。Agent的世界很大，Tool Calls只是入门，后面还有多Agent协作、长期记忆、规划推理等着你去探索。但别急，先把这一课吃透，下一篇我们聊更刺激的。

你，已经比昨天的自己更强了。

---

关注私信备注：“资料代找获取”，全网计算机学习资料代找：例如:
《课程：2026 年多模态大模型实战训练营》
《课程：AI 大模型工程师系统课程 (22 章完整版 持续更新)》
《课程：AI 大模型系统实战课第四期 (2026 年开课 持续更新)》
《课程：2026 年 AGI 大模型系统课 23 期》
《课程：2026 年 AGI 大模型系统课 21 期》
《课程：AI 大模型实战课 8 期 (2026 年 2 月最新完结版)》
《课程：AI 大模型系统实战课三期》
《课程：AI 大模型系统课程 (2026 年 2 月开课 持续更新)》
《课程：AI 大模型全阶课程 (2025 年 12 月开课 2026 年 6 月结课)》
《课程：AI 大模型工程师全阶课程 (2025 年 10 月开课 2026 年 4 月结课)》
《课程：2026 年最新大模型 Agent 开发系统课 (持续更新)》
《课程：LLM 多模态视觉大模型系统课》
《课程：大模型 AI 应用开发企业级项目实战课 (2026 年 1 月开课)》
《课程：大模型智能体线上速成班 V2.0》
《课程：Java+AI 大模型智能应用开发全阶课》
《课程：Python+AI 大模型实战视频教程》
《书籍：软件工程 3.0: 大模型驱动的研发新范式.pdf》
《课程：人工智能大模型系统课 (2026 年 1 月底完结版)》
《课程：AI 大模型零基础到商业实战全栈课第五期》
《课程：Vue3.5+Electron + 大模型跨平台 AI 桌面聊天应用实战 (2025)》
《课程：AI 大模型实战训练营 从入门到实战轻松上手》
《课程：2026 年 AI 大模型 RAG 与 Agent 智能体项目实战开发课》
《课程：大模型训练营配套补充资料》