构建生产级AI工具调用：错误处理与可靠性五件套实战

1. 项目概述：从玩具到工具的蜕变

如果你用过Anthropic的Claude API或者类似的工具调用（ToolUse）功能，大概率经历过这样的场景：写了个简单的Demo，调用天气API或者查个数据库，在本地跑得挺欢。一旦想把它集成到真实的客服系统、数据分析流水线或者自动化流程里，马上就发现不对劲了。一个无关紧要的第三方API超时，导致整个对话线程卡死；用户输入了一个无法解析的日期，程序直接抛异常退出；甚至因为网络波动，同一个工具被重复调用了三次，给用户返回了三份一模一样的股票报价。

这就是“玩具级”和“生产级”ToolUse循环最核心的区别。前者只关心“能不能跑通”，后者则必须回答“在复杂、不可预测的真实环境里，能不能持续、稳定、正确地运行”。这个项目，就是要把后者落到实处。所谓“错误处理与可靠性五件套”，是我从多个线上AI应用项目中提炼出来的一套组合方案，它不局限于Anthropic SDK，其核心思想适用于任何将大语言模型作为“决策中枢”来调用外部工具的场景。目标很明确：让你的AI应用像一名经验丰富的老员工，遇到意外不崩溃，能自己尝试解决，解决不了也知道如何优雅地向上汇报（记录日志）并安抚用户（提供友好反馈），而不是像个实习生一样直接愣在原地或者把错误堆栈甩到用户脸上。

2. 核心设计思路：构建有弹性的AI工作流

把大语言模型（LLM）当作一个“黑盒函数调用器”是初级思路，而生产级应用需要将其视为一个“有状态的、可能出错的、需要被管理的业务流程执行引擎”。这个转变要求我们在架构层面进行重新设计。

2.1 从线性执行到韧性循环

最简单的ToolUse循环是线性的：用户输入 -> LLM思考并决定调用工具 -> 执行工具 -> 将结果返回给LLM -> LLM生成最终回复。这个链条上的任何一个环节断裂，整个流程就失败了。

生产级设计需要将这个线性链条改造成一个具有弹性的“循环系统”。这个系统的核心特征是：

1. 状态可观测：系统在任何时刻都知道流程进行到哪一步，LLM做出了什么决策，工具执行的结果是什么。
2. 故障可隔离：一个工具的失败不应导致整个会话或流程崩溃。失败的影响范围应该被严格控制。
3. 流程可恢复：对于某些类型的错误（如网络临时故障），系统应具备重试的能力。对于无法自动恢复的错误，应有明确的降级或补偿路径。
4. 行为可预测：即使发生错误，系统的应对方式（如重试策略、错误信息格式）也应该是统一和可配置的，而不是随机的。

基于这些原则，“五件套”方案围绕ToolUse循环的五个关键接触点进行加固，分别是：工具调用前（验证与防护）、工具执行中（超时与重试）、工具执行后（结果解析与清洗）、异常发生时（结构化捕获与转换）、以及系统层面（熔断与降级）。

2.2 “五件套”全景图

这五个组件并非彼此独立，它们共同构成一个防御纵深：

• 第一层（输入验证）：在错误发生前尽可能预防。确保递给工具的参数是合法、安全的。
• 第二层（执行保障）：承认外部依赖总会出错，为执行过程设定边界（超时）和补救措施（重试）。
• 第三层（输出清洗）：即使工具执行“成功”，返回的数据也可能脏乱、不符合预期，需要标准化处理。
• 第四层（异常处理）：当错误不可避免地发生时，用一种LLM能理解、业务逻辑能处理的统一方式封装它。
• 第五层（系统保护）：防止单一工具的持续故障拖垮整个系统，并在极端情况下提供保底服务。

接下来，我们深入每一件“套件”，看看具体如何实现。

3. 核心细节解析与实操要点

3.1 第一件套：工具调用前的参数验证与防护

LLM生成的工具调用参数是不可信的。即使你给出了最清晰的描述，它仍可能产生类型错误、范围错误、甚至安全上有风险的参数。让工具函数自己去处理这些无效输入，是一种糟糕的实践。

实操方案：使用Pydantic进行声明式验证不要在工具函数内部写一堆if-else进行参数检查。应该为每个工具定义一个对应的Pydantic模型（Schema），在调用工具前，先用这个模型验证和解析LLM生成的参数字典。

from pydantic import BaseModel, Field, validator from typing import Optional from datetime import datetime # 1. 定义工具参数模型 class GetStockQuoteParams(BaseModel): symbol: str = Field(..., description="股票代码，如 AAPL, 00700.HK") timeframe: str = Field("1d", description="时间范围：'1d', '1w', '1m'") indicators: Optional[list[str]] = Field(None, description="技术指标列表，如 ['MA5', 'RSI']") @validator('symbol') def symbol_uppercase(cls, v): return v.upper().strip() @validator('timeframe') def valid_timeframe(cls, v): if v not in ['1d', '1w', '1m', '3m', '1y']: raise ValueError(f"不支持的timeframe: {v}") return v @validator('indicators') def filter_indicators(cls, v): if v is None: return v allowed = ['MA5', 'MA10', 'MA20', 'RSI', 'MACD', 'BOLL'] return [i for i in v if i in allowed] # 2. 工具函数本身只处理“干净”的数据 def get_stock_quote(params: GetStockQuoteParams): # 此时，params.symbol 已经是大写且去除了空格 # params.timeframe 一定是合法值 # params.indicators 只包含允许的指标 # 你可以安全地进行后续API调用，无需再做检查 api_url = f"https://api.example.com/quote/{params.symbol}?range={params.timeframe}" # ... 调用逻辑 return {"price": 150.25, "currency": "USD"} # 3. 在ToolUse循环中，插入验证层 def safe_tool_invoke(tool_name: str, tool_input: dict): if tool_name == "get_stock_quote": try: # 关键步骤：验证和转换 validated_params = GetStockQuoteParams(**tool_input) # 调用真正的工具函数 result = get_stock_quote(validated_params) return {"status": "success", "data": result} except ValueError as e: # 验证失败，返回结构化的错误信息 return {"status": "error", "type": "VALIDATION_ERROR", "message": str(e)}

实操心得：Pydantic的@validator非常强大，除了类型检查，还能做数据清洗（如转大写、去空格）。验证失败抛出的ValidationError包含了详细的错误信息，你可以将其转化为给LLM的友好提示，例如：“您提供的股票代码格式有误，请确保是类似‘AAPL’的格式。”

3.2 第二件套：执行中的超时与智能重试

外部服务调用（HTTP请求、数据库查询）是主要的故障点。一个永不超时的请求会挂起你的工作线程，耗尽资源。

实操方案：分层超时与指数退避重试不要对所有工具使用相同的超时和重试策略。查询内部缓存的工具应该设置很短的超时（如2秒），而调用第三方支付网关的工具可能需要更长时间（如30秒）。重试策略也应区别对待。

import asyncio from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import aiohttp from typing import Any, Callable # 定义不同工具类别的配置 TOOL_POLICIES = { "internal_cache": {"timeout": 2.0, "max_retries": 1}, "internal_api": {"timeout": 5.0, "max_retries": 2}, "external_api_fast": {"timeout": 10.0, "max_retries": 3}, "external_api_slow": {"timeout": 30.0, "max_retries": 2}, } def get_policy_for_tool(tool_name: str) -> dict: # 根据工具名称映射到策略 if "cache" in tool_name: return TOOL_POLICIES["internal_cache"] elif "payment" in tool_name or "sms" in tool_name: return TOOL_POLICIES["external_api_slow"] else: return TOOL_POLICIES["external_api_fast"] async def execute_with_resilience(tool_func: Callable, *args, policy: dict, **kwargs) -> Any: """ 带有超时和重试的执行包装器 """ policy = policy or TOOL_POLICIES["external_api_fast"] # 定义重试装饰器 retry_decorator = retry( stop=stop_after_attempt(policy["max_retries"]), wait=wait_exponential(multiplier=1, min=1, max=10), # 指数退避：1s, 2s, 4s... retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)), reraise=True # 重试耗尽后抛出原异常 ) @retry_decorator async def _wrapped(): try: # 为单次执行设置超时 return await asyncio.wait_for( tool_func(*args, **kwargs), timeout=policy["timeout"] ) except asyncio.TimeoutError: # 记录超时日志，便于区分是单次超时还是重试后最终超时 print(f"工具 {tool_func.__name__} 单次执行超时（{policy['timeout']}s）") raise # 抛出，由tenacity决定是否重试 try: return await _wrapped() except Exception as e: # 所有重试尝试均失败后，在这里进行统一的错误处理转换 # 例如，将异常转换为结构化的错误信息 error_info = { "tool": tool_func.__name__, "error_type": e.__class__.__name__, "message": f"在执行{policy['max_retries']}次重试后仍失败: {str(e)}", "policy": policy } # 可以在这里触发告警 raise ToolExecutionError(error_info) from e

注意事项：指数退避（Exponential Backoff）是重试的核心策略。它让每次重试的等待时间逐渐加长（如1秒、2秒、4秒、8秒），避免在服务短暂故障时，大量客户端同时重试导致“惊群效应”，反而压垮正在恢复的服务。tenacity库让这种策略的实现变得非常简单。

3.3 第三件套：执行后的结果解析与清洗

工具执行“成功”（HTTP状态码200）不代表返回的数据就是可用的。API可能返回了{“data”: null}，或者数字被包裹在字符串里“price”: “150.25”，又或者多了一些LLM不需要的冗杂字段。

实操方案：定义标准化响应模型像处理输入参数一样，为每个工具的输出也定义一个Pydantic模型。这个模型负责：

1. 类型转换：确保数字、布尔值、日期等有正确的类型。
2. 数据脱敏：移除或加密不应暴露给LLM的敏感字段（如用户手机号、内部ID）。
3. 结构扁平化：将嵌套过深的API响应简化，便于LLM理解。
4. 提供默认值：为可能缺失的字段提供安全默认值。

class WeatherAPIResponse(BaseModel): """原始API响应模型（可能很复杂）""" location: dict current: dict forecast: list class CleanedWeatherData(BaseModel): """清洗后给LLM的标准化模型""" city: str temperature_c: float condition_text: str feels_like_c: float humidity: int wind_kph: float forecast_tomorrow: str # 简化为一句描述 @classmethod def from_raw_response(cls, raw: WeatherAPIResponse): # 复杂的清洗和转换逻辑集中在这里 return cls( city=raw.location["name"], temperature_c=float(raw.current["temp_c"]), condition_text=raw.current["condition"]["text"], feels_like_c=float(raw.current.get("feelslike_c", raw.current["temp_c"])), # 提供默认值 humidity=int(raw.current["humidity"]), wind_kph=float(raw.current["wind_kph"]), forecast_tomorrow=raw.forecast[0]["day"]["condition"]["text"] if raw.forecast else "无预报数据" ) def fetch_weather(city: str) -> CleanedWeatherData: raw_data = call_weather_api(city) # 返回原始字典 # 1. 解析原始响应 raw_model = WeatherAPIResponse(**raw_data) # 2. 清洗转换 cleaned = CleanedWeatherData.from_raw_response(raw_model) return cleaned # 在ToolUse循环中，你将返回cleaned.dict()给LLM，数据干净且结构一致。

实操心得：这个清洗层是提升LLM表现的关键。杂乱的数据会干扰LLM的判断，而干净、结构化的数据能极大提高后续生成回答的准确性和连贯性。这也是一个很好的数据脱敏点，确保不会意外将用户ID、内部错误码等敏感信息泄露给LLM和最终用户。

3.4 第四件套：异常的结构化捕获与LLM友好化

当错误发生时，直接把Python的Exception对象或HTTP错误堆栈扔给LLM是没用的。LLM需要能理解“发生了什么错误”以及“用户/系统接下来该怎么办”。

实操方案：自定义异常层级与错误码建立一个自定义的异常体系，将各种底层错误（网络错误、验证错误、业务逻辑错误）映射到有限的、预定义的“错误类型”和“友好消息”上。

from enum import Enum class ErrorType(Enum): VALIDATION = "参数验证失败" EXTERNAL_SERVICE_UNAVAILABLE = "外部服务暂时不可用" EXTERNAL_SERVICE_ERROR = "外部服务返回错误" RATE_LIMITED = "请求过于频繁，请稍后再试" NOT_FOUND = "请求的资源不存在" UNAUTHORIZED = "权限不足" INTERNAL = "系统内部错误" class ToolExecutionError(Exception): """工具执行错误的统一封装""" def __init__(self, error_type: ErrorType, message: str, details: dict = None, original_exception: Exception = None): self.error_type = error_type self.message = message # 给LLM/用户看的友好信息 self.details = details or {} # 内部调试的详细信息 self.original_exception = original_exception super().__init__(self.message) def to_llm_format(self): """转换为LLM能理解的标准化错误信息字典""" return { "status": "error", "error_code": self.error_type.name, "error_message": self.message, # 谨慎决定是否将details给LLM，通常不放 "suggestion_for_llm": self._get_llm_suggestion() } def _get_llm_suggestion(self): """根据错误类型，给LLM一些后续行动建议""" suggestions = { ErrorType.VALIDATION: "请用户检查并重新输入参数。", ErrorType.EXTERNAL_SERVICE_UNAVAILABLE: "告知用户服务暂时有问题，建议稍后重试。可以尝试使用备用方案或缓存数据。", ErrorType.NOT_FOUND: "告知用户未找到相关信息，并询问是否要搜索其他内容。", ErrorType.RATE_LIMITED: "告知用户操作过于频繁，请休息一分钟再试。", } return suggestions.get(self.error_type, "告知用户遇到了问题，建议稍后重试。") # 在工具执行层，捕获底层异常并转换 async def call_external_api(url): try: async with aiohttp.ClientSession() as session: async with session.get(url, timeout=10) as resp: if resp.status == 429: # 捕获速率限制错误，并向上抛出自定义错误 raise ToolExecutionError( ErrorType.RATE_LIMITED, "查询速度太快了，被限制啦。", details={"status_code": 429, "headers": dict(resp.headers)}, original_exception=None ) resp.raise_for_status() return await resp.json() except aiohttp.ClientConnectorError as e: # 网络连接错误 raise ToolExecutionError( ErrorType.EXTERNAL_SERVICE_UNAVAILABLE, "网络连接失败，请检查您的网络或稍后再试。", details={"url": url}, original_exception=e ) from e except asyncio.TimeoutError as e: # 超时错误 raise ToolExecutionError( ErrorType.EXTERNAL_SERVICE_UNAVAILABLE, "请求超时，服务响应可能过慢。", details={"url": url, "timeout": 10}, original_exception=e ) from e # 在ToolUse循环的主逻辑中 try: tool_result = await execute_with_resilience(call_external_api, some_url, policy=some_policy) return {"status": "success", "data": tool_result} except ToolExecutionError as e: # 捕获到我们定义的结构化错误 error_response = e.to_llm_format() # 将error_response返回给LLM，LLM就能根据error_code和suggestion_for_llm生成得体的用户回复 return error_response except Exception as e: # 捕获未预料的异常，转化为内部错误，避免泄露堆栈 unexpected_error = ToolExecutionError( ErrorType.INTERNAL, "系统处理时遇到了意外问题。", details={"exception_class": e.__class__.__name__}, original_exception=e ) # 记录完整的异常日志到监控系统 log_error(unexpected_error, exc_info=True) return unexpected_error.to_llm_format()

注意事项：错误消息分层至关重要。details字段用于内部调试和日志，包含原始异常、请求参数等敏感信息。error_message和suggestion_for_llm是经过处理的、对LLM和最终用户友好的信息，不应包含技术细节。这既保护了系统安全，也提升了用户体验。

3.5 第五件套：系统级保护与降级策略

当某个外部服务持续故障时，继续让所有用户请求去尝试调用它是没有意义的，只会浪费资源并增加系统负载。我们需要在系统层面进行保护。

实操方案：熔断器（Circuit Breaker）与静态降级熔断器模式模仿电路保险丝。当失败次数超过阈值，熔断器“跳闸”，在一段时间内直接拒绝所有对该服务的请求（快速失败），给服务恢复的时间。之后进入“半开”状态，试探性放行少量请求，如果成功则关闭熔断器，恢复调用；如果失败则继续保持熔断状态。

import time from dataclasses import dataclass from enum import Enum class CircuitState(Enum): CLOSED = "closed" # 正常状态，请求可通过 OPEN = "open" # 熔断状态，请求被快速拒绝 HALF_OPEN = "half_open" # 半开状态，试探性放行 @dataclass class CircuitBreaker: name: str failure_threshold: int = 5 # 连续失败多少次后熔断 reset_timeout: int = 60 # 熔断后多久进入半开状态（秒） half_open_success_threshold: int = 2 # 半开状态下成功多少次后关闭 def __post_init__(self): self.state = CircuitState.CLOSED self.failure_count = 0 self.last_failure_time = None self.half_open_success_count = 0 def record_success(self): if self.state == CircuitState.HALF_OPEN: self.half_open_success_count += 1 if self.half_open_success_count >= self.half_open_success_threshold: # 半开状态下连续成功，关闭熔断器 self._close() elif self.state == CircuitState.CLOSED: # 正常状态下成功，重置失败计数 self.failure_count = 0 def record_failure(self): self.failure_count += 1 self.last_failure_time = time.time() if self.state == CircuitState.CLOSED and self.failure_count >= self.failure_threshold: # 达到失败阈值，打开熔断器 self._open() elif self.state == CircuitState.HALF_OPEN: # 半开状态下失败，重新打开熔断器 self._open() def _open(self): self.state = CircuitState.OPEN print(f"[熔断器 {self.name}] 状态：OPEN。将在 {self.reset_timeout} 秒后进入半开状态。") def _close(self): self.state = CircuitState.CLOSED self.failure_count = 0 self.half_open_success_count = 0 self.last_failure_time = None print(f"[熔断器 {self.name}] 状态：CLOSED。恢复正常。") def allow_request(self) -> bool: """检查当前是否允许执行请求""" now = time.time() if self.state == CircuitState.OPEN: if now - self.last_failure_time > self.reset_timeout: # 超时后进入半开状态 self.state = CircuitState.HALF_OPEN self.half_open_success_count = 0 print(f"[熔断器 {self.name}] 状态：HALF_OPEN。开始试探。") return True else: # 仍在熔断期，快速失败 return False # CLOSED 或 HALF_OPEN 状态都允许请求 return True def __call__(self, func): """用作装饰器""" def wrapper(*args, **kwargs): if not self.allow_request(): # 快速失败，直接返回降级结果 raise ToolExecutionError( ErrorType.EXTERNAL_SERVICE_UNAVAILABLE, "相关服务暂时不可用（熔断保护），请稍后再试。", details={"circuit_breaker": self.name, "state": self.state.value} ) try: result = func(*args, **kwargs) self.record_success() return result except Exception as e: self.record_failure() raise return wrapper # 使用示例：为某个高风险工具添加熔断器 weather_circuit_breaker = CircuitBreaker(name="weather_api", failure_threshold=3, reset_timeout=30) @weather_circuit_breaker def call_weather_api_safe(city: str): # 这个函数现在被熔断器保护 return call_weather_api(city) # 在ToolUse循环中，直接调用 call_weather_api_safe # 如果熔断器打开，会立即抛出包含友好信息的ToolExecutionError，而不会真正发起网络请求。

降级策略（Fallback）是熔断的伴侣。当熔断器打开或工具调用失败时，不应该只是返回一个错误，而应该尽可能提供一个“降级”的、可用的结果。例如：

• 返回缓存中过期的数据，并提示“以下信息可能不是最新的”。
• 调用一个更稳定但功能较弱的备用API。
• 返回一个静态的、预定义的响应。
• 引导用户进行其他操作（如“天气服务暂时不可用，您可以先查询空气质量。”）。

def get_weather_with_fallback(city: str): try: return call_weather_api_safe(city) except ToolExecutionError as e: if e.error_type == ErrorType.EXTERNAL_SERVICE_UNAVAILABLE: # 尝试从缓存获取 cached = get_weather_from_cache(city) if cached: cached["_note"] = "提示：此数据来自缓存，可能不是实时信息。" return cached # 缓存也没有，返回一个友好的静态降级信息 return { "city": city, "temperature_c": None, "condition_text": "服务暂时无法获取实时天气。", "suggestion": "您可以稍后重试，或访问气象网站查询。" } else: # 其他错误，继续上抛 raise

4. 整合实战：构建生产级ToolUse执行引擎

现在，我们将“五件套”组合起来，形成一个完整的、高可用的ToolUse执行函数。这是整个系统的核心。

import asyncio from typing import Dict, Any, Callable from pydantic import BaseModel, ValidationError class ToolRegistry: """工具注册中心，管理所有可用工具及其元数据""" def __init__(self): self._tools: Dict[str, dict] = {} def register(self, name: str, func: Callable, param_model: BaseModel, result_model: BaseModel = None, policy: str = "default"): self._tools[name] = { "func": func, "param_model": param_model, "result_model": result_model, "policy": policy } async def execute(self, tool_name: str, tool_input: dict) -> Dict[str, Any]: if tool_name not in self._tools: raise ValueError(f"工具未注册: {tool_name}") tool_info = self._tools[tool_name] func = tool_info["func"] ParamModel = tool_info["param_model"] ResultModel = tool_info["result_model"] policy = get_policy_for_tool(tool_name) # 获取策略 # === 阶段1: 参数验证 (第一件套) === try: validated_params = ParamModel(**tool_input) except ValidationError as e: error = ToolExecutionError( ErrorType.VALIDATION, f"工具参数验证失败: {e.errors()[0]['msg']}", details={"validation_errors": e.errors()} ) return error.to_llm_format() # === 阶段2 & 3: 执行与重试 (第二件套) & 结果清洗 (第三件套) === async def _tool_call(): # 这里是实际的工具执行 raw_result = await func(validated_params) # 如果有结果模型，进行清洗 if ResultModel: if isinstance(raw_result, dict): cleaned_result = ResultModel(**raw_result) else: # 假设结果已经是模型实例 cleaned_result = raw_result return cleaned_result.dict() return raw_result try: # 使用带重试和超时的执行器 cleaned_data = await execute_with_resilience(_tool_call, policy=policy) return {"status": "success", "data": cleaned_data} except ToolExecutionError as e: # 已知的结构化错误，直接转换 return e.to_llm_format() except Exception as e: # 未知错误，封装为内部错误 unexpected_error = ToolExecutionError( ErrorType.INTERNAL, "工具执行过程中发生意外错误。", details={"exception": str(e)}, original_exception=e ) log_error(unexpected_error) return unexpected_error.to_llm_format() # 初始化注册中心 registry = ToolRegistry() # 注册工具 registry.register( name="get_weather", func=fetch_weather, # 这是已经包含了清洗和降级的函数 param_model=WeatherQueryParams, # 输入参数模型 result_model=CleanedWeatherData, # 输出结果模型（可选，用于二次确认） policy="external_api_slow" ) # 在Anthropic SDK的ToolUse循环中 async def handle_tool_use(tool_call): tool_name = tool_call.name tool_input = tool_call.input # 这是LLM生成的参数字典 # 调用我们的高可用执行引擎 result = await registry.execute(tool_name, tool_input) # 将结果返回给Anthropic的消息构建器 # result 已经是 {“status”: “success/error”, ...} 的标准格式 return result

这个ToolRegistry.execute方法就是一个生产级的执行引擎。它串联了验证、执行保障、错误处理的全流程，并返回LLM能直接使用的标准化响应。

5. 常见问题与排查技巧实录

在实际部署中，即使有了完善的框架，还是会遇到各种稀奇古怪的问题。下面是我踩过的一些坑和对应的排查思路。

5.1 LLM不按预期调用工具，或参数总是错误

问题现象：你定义了一个工具book_meeting(room: str, time: datetime, attendees: List[str])，但LLM总是用time: “明天下午两点”这样的字符串调用，或者attendees传成了一个字符串。

排查与解决：

1. 检查工具描述（Documentation）：Anthropic SDK中，工具的描述至关重要。确保描述清晰、无歧义，并明确说明参数格式。例如：

   tool = { "name": "book_meeting", "description": "预订会议室。time参数必须是ISO 8601格式的字符串，例如 '2023-10-27T14:30:00'。attendees是一个邮箱地址的列表。", "input_schema": { "type": "object", "properties": { "time": {"type": "string", "format": "date-time"}, # 使用标准格式提示 "attendees": { "type": "array", "items": {"type": "string", "format": "email"} # 提示是邮箱数组 } }, "required": ["room", "time", "attendees"] } }

2. 在System Prompt中强化规则：在发给Claude的System Prompt里，明确写出工具调用规范。例如：“当你需要预订会议时，请使用book_meeting工具。特别注意：time参数必须转换为‘YYYY-MM-DDTHH:MM:SS’格式的字符串；attendees必须是一个列表，即使只有一个人。”
3. 实施“后置修正”：如果LLM在某些格式上（如日期）持续犯错，可以在参数验证层（Pydantic模型）中加入一个“修正器”。例如，用一个更宽松的解析器先尝试解析“明天下午两点”，如果成功，再将其转换为标准格式。但这只是权宜之计，更好的方法是优化提示。

5.2 工具执行成功，但LLM无法理解返回结果

问题现象：工具返回了一大段复杂的JSON，LLM在后续回答中要么忽略了关键数据，要么错误解读。

排查与解决：

1. 强化结果清洗（第三件套）：这是最主要的手段。确保返回给LLM的数据极度简洁和结构化。只保留LLM生成回答所必需的字段。例如，天气API返回了湿度、压强、露点等10个字段，但你的对话场景可能只需要温度、体感温度和天气状况这三个。其他字段都是噪音。
2. 为结果添加自然语言摘要：在返回的JSON中，可以额外添加一个summary字段，用一句自然语言概括结果。例如：

   { "status": "success", "data": { "temperature_c": 22, "humidity": 65, "condition": "Sunny" }, "summary": "当前天气晴朗，气温22摄氏度，湿度65%。" }

   LLM有时会更倾向于直接使用summary来组织回答，这能显著提高回答的流畅性和准确性。
3. 在System Prompt中教导LLM：告诉LLM：“工具返回的结果中，data字段是主要信息，summary字段是可供你参考的总结，你可以直接引用或复述summary的内容。”

5.3 循环调用与超时失控

问题现象：LLM陷入了一个循环，反复调用同一个工具，或者工具调用链过长，导致整体响应时间超过用户等待极限。

排查与解决：

1. 设置会话级或工具级调用上限：在ToolUse循环的上下文状态中，维护一个计数器。

   class ConversationState: def __init__(self): self.tool_call_count = {} self.max_calls_per_tool = 3 # 单个工具最多调用3次 self.max_total_calls = 10 # 整个会话最多调用10次工具 def can_call_tool(self, tool_name: str) -> bool: if self.tool_call_count.get(tool_name, 0) >= self.max_calls_per_tool: return False if sum(self.tool_call_count.values()) >= self.max_total_calls: return False return True def record_call(self, tool_name: str): self.tool_call_count[tool_name] = self.tool_call_count.get(tool_name, 0) + 1

   在调用工具前检查can_call_tool，如果超过限制，直接返回一个错误信息给LLM：“该工具调用次数已达上限，请尝试其他方法或总结当前信息。”
2. 设置全局超时：为整个“用户提问 -> 多轮ToolUse -> 生成最终回答”的流程设置一个总超时（例如60秒）。可以使用asyncio.wait_for包裹整个处理协程，超时后强制中断，返回一个“处理超时”的友好提示。
3. 设计工具以终结循环：有些工具本身应该是“终结者”。例如，一个finalize_answer工具，它接收所有收集到的信息，并生成最终答案。在System Prompt中告诉LLM：“当你认为信息足够时，请使用finalize_answer工具来结束本次查询。”

5.4 监控与日志记录如何做

生产系统没有监控就是瞎子。你需要知道工具调用的成功率、延迟、哪些工具最容易出错。

实操建议：

1. 结构化日志：不要用print。使用structlog或json-logger，每一条日志都是一个JSON对象。

   import structlog logger = structlog.get_logger() async def execute_with_resilience_and_logging(tool_func, *args, **kwargs): start_time = time.time() tool_name = tool_func.__name__ log = logger.bind(tool=tool_name, attempt=1) try: result = await tool_func(*args, **kwargs) duration = time.time() - start_time log.info("tool_success", duration=duration, result_type=type(result).__name__) return result except ToolExecutionError as e: duration = time.time() - start_time log.warning("tool_error_known", duration=duration, error_code=e.error_type.name) raise except Exception as e: duration = time.time() - start_time log.error("tool_error_unknown", duration=duration, exception=str(e), exc_info=True) raise

2. 关键指标打点：在日志中记录duration（耗时）、status（成功/失败）、error_code。这些日志可以被日志收集系统（如Loki）抓取，并通过Grafana等工具绘制成仪表盘，监控成功率、P95/P99延迟、错误类型分布。
3. 链路追踪（Trace）：对于复杂的多工具调用链，引入OpenTelemetry这样的分布式追踪系统。为每个用户会话生成一个唯一的trace_id，并贯穿所有工具调用和LLM交互。这样当某个用户反馈问题时，你可以通过trace_id快速还原出完整的请求链路，看到每一步发生了什么，是哪个工具慢了或错了。

生产级ToolUse循环的构建，本质上是在LLM的“智能”和外部世界的“不确定性”之间，筑起一道道坚固的防线。这五件套——验证、重试、清洗、错误封装、熔断——就是你的核心防御工事。它们不会让你的应用变得百分百不出错，但能确保在出错时，系统行为是可控的、可观测的、对用户友好的。这套模式经过多个线上项目的锤炼，显著提升了AI应用的稳定性和用户体验。开始动手为你的工具函数穿上这层“铠甲”吧，你会发现，你的AI助手从此变得更加可靠和值得信赖。