1. 项目概述从玩具到工具的蜕变如果你用过Anthropic的Claude API或者类似的工具调用ToolUse功能大概率经历过这样的场景写了个简单的Demo调用天气API或者查个数据库在本地跑得挺欢。一旦想把它集成到真实的客服系统、数据分析流水线或者自动化流程里马上就发现不对劲了。一个无关紧要的第三方API超时导致整个对话线程卡死用户输入了一个无法解析的日期程序直接抛异常退出甚至因为网络波动同一个工具被重复调用了三次给用户返回了三份一模一样的股票报价。这就是“玩具级”和“生产级”ToolUse循环最核心的区别。前者只关心“能不能跑通”后者则必须回答“在复杂、不可预测的真实环境里能不能持续、稳定、正确地运行”。这个项目就是要把后者落到实处。所谓“错误处理与可靠性五件套”是我从多个线上AI应用项目中提炼出来的一套组合方案它不局限于Anthropic SDK其核心思想适用于任何将大语言模型作为“决策中枢”来调用外部工具的场景。目标很明确让你的AI应用像一名经验丰富的老员工遇到意外不崩溃能自己尝试解决解决不了也知道如何优雅地向上汇报记录日志并安抚用户提供友好反馈而不是像个实习生一样直接愣在原地或者把错误堆栈甩到用户脸上。2. 核心设计思路构建有弹性的AI工作流把大语言模型LLM当作一个“黑盒函数调用器”是初级思路而生产级应用需要将其视为一个“有状态的、可能出错的、需要被管理的业务流程执行引擎”。这个转变要求我们在架构层面进行重新设计。2.1 从线性执行到韧性循环最简单的ToolUse循环是线性的用户输入 - LLM思考并决定调用工具 - 执行工具 - 将结果返回给LLM - LLM生成最终回复。这个链条上的任何一个环节断裂整个流程就失败了。生产级设计需要将这个线性链条改造成一个具有弹性的“循环系统”。这个系统的核心特征是状态可观测系统在任何时刻都知道流程进行到哪一步LLM做出了什么决策工具执行的结果是什么。故障可隔离一个工具的失败不应导致整个会话或流程崩溃。失败的影响范围应该被严格控制。流程可恢复对于某些类型的错误如网络临时故障系统应具备重试的能力。对于无法自动恢复的错误应有明确的降级或补偿路径。行为可预测即使发生错误系统的应对方式如重试策略、错误信息格式也应该是统一和可配置的而不是随机的。基于这些原则“五件套”方案围绕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 fhttps://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( stopstop_after_attempt(policy[max_retries]), waitwait_exponential(multiplier1, min1, max10), # 指数退避1s, 2s, 4s... retryretry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)), reraiseTrue # 重试耗尽后抛出原异常 ) retry_decorator async def _wrapped(): try: # 为单次执行设置超时 return await asyncio.wait_for( tool_func(*args, **kwargs), timeoutpolicy[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模型。这个模型负责类型转换确保数字、布尔值、日期等有正确的类型。数据脱敏移除或加密不应暴露给LLM的敏感字段如用户手机号、内部ID。结构扁平化将嵌套过深的API响应简化便于LLM理解。提供默认值为可能缺失的字段提供安全默认值。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( cityraw.location[name], temperature_cfloat(raw.current[temp_c]), condition_textraw.current[condition][text], feels_like_cfloat(raw.current.get(feelslike_c, raw.current[temp_c])), # 提供默认值 humidityint(raw.current[humidity]), wind_kphfloat(raw.current[wind_kph]), forecast_tomorrowraw.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, timeout10) as resp: if resp.status 429: # 捕获速率限制错误并向上抛出自定义错误 raise ToolExecutionError( ErrorType.RATE_LIMITED, 查询速度太快了被限制啦。, details{status_code: 429, headers: dict(resp.headers)}, original_exceptionNone ) resp.raise_for_status() return await resp.json() except aiohttp.ClientConnectorError as e: # 网络连接错误 raise ToolExecutionError( ErrorType.EXTERNAL_SERVICE_UNAVAILABLE, 网络连接失败请检查您的网络或稍后再试。, details{url: url}, original_exceptione ) from e except asyncio.TimeoutError as e: # 超时错误 raise ToolExecutionError( ErrorType.EXTERNAL_SERVICE_UNAVAILABLE, 请求超时服务响应可能过慢。, details{url: url, timeout: 10}, original_exceptione ) from e # 在ToolUse循环的主逻辑中 try: tool_result await execute_with_resilience(call_external_api, some_url, policysome_policy) return {status: success, data: tool_result} except ToolExecutionError as e: # 捕获到我们定义的结构化错误 error_response e.to_llm_format() # 将error_response返回给LLMLLM就能根据error_code和suggestion_for_llm生成得体的用户回复 return error_response except Exception as e: # 捕获未预料的异常转化为内部错误避免泄露堆栈 unexpected_error ToolExecutionError( ErrorType.INTERNAL, 系统处理时遇到了意外问题。, details{exception_class: e.__class__.__name__}, original_exceptione ) # 记录完整的异常日志到监控系统 log_error(unexpected_error, exc_infoTrue) 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(nameweather_api, failure_threshold3, reset_timeout30) 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: # 其他错误继续上抛 raise4. 整合实战构建生产级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, policypolicy) 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_exceptione ) log_error(unexpected_error) return unexpected_error.to_llm_format() # 初始化注册中心 registry ToolRegistry() # 注册工具 registry.register( nameget_weather, funcfetch_weather, # 这是已经包含了清洗和降级的函数 param_modelWeatherQueryParams, # 输入参数模型 result_modelCleanedWeatherData, # 输出结果模型可选用于二次确认 policyexternal_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传成了一个字符串。排查与解决检查工具描述DocumentationAnthropic 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] } }在System Prompt中强化规则在发给Claude的System Prompt里明确写出工具调用规范。例如“当你需要预订会议时请使用book_meeting工具。特别注意time参数必须转换为‘YYYY-MM-DDTHH:MM:SS’格式的字符串attendees必须是一个列表即使只有一个人。”实施“后置修正”如果LLM在某些格式上如日期持续犯错可以在参数验证层Pydantic模型中加入一个“修正器”。例如用一个更宽松的解析器先尝试解析“明天下午两点”如果成功再将其转换为标准格式。但这只是权宜之计更好的方法是优化提示。5.2 工具执行成功但LLM无法理解返回结果问题现象工具返回了一大段复杂的JSONLLM在后续回答中要么忽略了关键数据要么错误解读。排查与解决强化结果清洗第三件套这是最主要的手段。确保返回给LLM的数据极度简洁和结构化。只保留LLM生成回答所必需的字段。例如天气API返回了湿度、压强、露点等10个字段但你的对话场景可能只需要温度、体感温度和天气状况这三个。其他字段都是噪音。为结果添加自然语言摘要在返回的JSON中可以额外添加一个summary字段用一句自然语言概括结果。例如{ status: success, data: { temperature_c: 22, humidity: 65, condition: Sunny }, summary: 当前天气晴朗气温22摄氏度湿度65%。 }LLM有时会更倾向于直接使用summary来组织回答这能显著提高回答的流畅性和准确性。在System Prompt中教导LLM告诉LLM“工具返回的结果中data字段是主要信息summary字段是可供你参考的总结你可以直接引用或复述summary的内容。”5.3 循环调用与超时失控问题现象LLM陷入了一个循环反复调用同一个工具或者工具调用链过长导致整体响应时间超过用户等待极限。排查与解决设置会话级或工具级调用上限在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“该工具调用次数已达上限请尝试其他方法或总结当前信息。”设置全局超时为整个“用户提问 - 多轮ToolUse - 生成最终回答”的流程设置一个总超时例如60秒。可以使用asyncio.wait_for包裹整个处理协程超时后强制中断返回一个“处理超时”的友好提示。设计工具以终结循环有些工具本身应该是“终结者”。例如一个finalize_answer工具它接收所有收集到的信息并生成最终答案。在System Prompt中告诉LLM“当你认为信息足够时请使用finalize_answer工具来结束本次查询。”5.4 监控与日志记录如何做生产系统没有监控就是瞎子。你需要知道工具调用的成功率、延迟、哪些工具最容易出错。实操建议结构化日志不要用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(tooltool_name, attempt1) try: result await tool_func(*args, **kwargs) duration time.time() - start_time log.info(tool_success, durationduration, result_typetype(result).__name__) return result except ToolExecutionError as e: duration time.time() - start_time log.warning(tool_error_known, durationduration, error_codee.error_type.name) raise except Exception as e: duration time.time() - start_time log.error(tool_error_unknown, durationduration, exceptionstr(e), exc_infoTrue) raise关键指标打点在日志中记录duration耗时、status成功/失败、error_code。这些日志可以被日志收集系统如Loki抓取并通过Grafana等工具绘制成仪表盘监控成功率、P95/P99延迟、错误类型分布。链路追踪Trace对于复杂的多工具调用链引入OpenTelemetry这样的分布式追踪系统。为每个用户会话生成一个唯一的trace_id并贯穿所有工具调用和LLM交互。这样当某个用户反馈问题时你可以通过trace_id快速还原出完整的请求链路看到每一步发生了什么是哪个工具慢了或错了。生产级ToolUse循环的构建本质上是在LLM的“智能”和外部世界的“不确定性”之间筑起一道道坚固的防线。这五件套——验证、重试、清洗、错误封装、熔断——就是你的核心防御工事。它们不会让你的应用变得百分百不出错但能确保在出错时系统行为是可控的、可观测的、对用户友好的。这套模式经过多个线上项目的锤炼显著提升了AI应用的稳定性和用户体验。开始动手为你的工具函数穿上这层“铠甲”吧你会发现你的AI助手从此变得更加可靠和值得信赖。
)
