第 5 章:工程细节 —— 让工具调用达到生产级
5.1 工具描述写作军规(模型乱调工具时,90% 先改这里)
军规 | Bad | Good |
说清"何时用" | "查天气" | "查询城市当前实时天气。用户问天气/气温/穿衣/带伞时使用;历史天气不支持" |
参数给格式与示例 | "城市" | "城市中文名,如:北京。用户未提及城市时先追问,不要猜测" |
划清边界 | (无) | "只支持数学表达式,不含变量/函数;日期计算请改用 date_diff 工具" |
定义返回 | (无) | "返回格式:'城市: 天气 温度';失败时返回以'天气查询失败'开头的错误说明" |
补一条系统提示词侧的黄金句式(你已经在用了):"涉及 X 的问题必须使用工具,禁止凭记忆回答/心算"——把"该用工具时不用"的漏调率打下来。
5.2 用 Pydantic 自动生成 Schema(少写 30 行样板)
手写 JSON Schema 又长又易错。Pydantic 的model_json_schema()一键生成:
class WeatherArgs(BaseModel): """查询指定城市的当前实时天气。用户询问天气/气温/穿衣建议时使用。""" city: str = Field(..., description="城市中文名,如:北京") def pydantic_to_tool(model_cls) -> dict: schema = model_cls.model_json_schema() return {"type": "function", "function": { "name": schema.get("title", model_cls.__name__).lower(), "description": model_cls.__doc__.strip(), "parameters": {"type": "object", "properties": schema["properties"], "required": schema.get("required", [])}}}LangChain@tool装饰器把这一步也自动化了(读函数签名+docstring 直接生成)——你现在已看穿它的全部魔法。
5.3 敏感操作:先"申请",人来"批准"
绝不允许模型直调退款/转账/删除/群发类函数。生产范式——工具只返回"待确认单",真执行等人点头:
def request_refund(order_id: str, amount: float) -> str: """发起退款申请。注意:本工具只创建待确认的申请单,不会直接扣款。""" if amount > 2000: return f"已创建退款申请单 R-{order_id}(金额{amount}元,超过阈值),已转人工审核,模型无权继续操作。" return f"已创建退款申请单 R-{order_id}({amount}元)。请向用户复述金额并获得明确同意后,调用 confirm_refund 工具。"这就是human-in-the-loop的最小实现。
5.4 其余四条速记
- 工具结果限长:
return result[:2000]或先摘要——工具灌爆上下文是 Agent 变笨主因(第 2 批 17.5)。 - 工具历史要不要长留:本轮任务内必须留(模型靠它推理);跨任务的旧 tool 消息可在历史裁剪时优先清退(信息密度低、体积大)。
- 幂等设计:查询类随便重试;写操作类(下单)要防重复执行——传入唯一请求号。
- 给工具也加超时:
requests.get(..., timeout=10)你已在做;慢工具会卡死整个循环。
第 6 章:错误处理、重试与成本控制
6.1 SDK 异常家族(对号入座表)
import openai try: r = client.chat.completions.create(model=MODEL, messages=messages, timeout=30) except openai.AuthenticationError: # 401:Key错误/被吊销 → 查 .env,不要重试 ... except openai.RateLimitError: # 429:限流/欠费 → 指数退避重试(见下) ... except openai.BadRequestError as e: # 400:参数错/上下文超限 → 修请求,别重试 ... # 报错含 context length 字样 → 该裁历史了! except openai.APITimeoutError: # 超时 → 可重试 ... except openai.APIConnectionError: # 网络不通 → 可重试 ... except openai.APIStatusError as e: # 5xx 服务端故障 → 退避重试 print(e.status_code, e.response)分两类记:该重试的(429/超时/网络/5xx——问题在环境)与不该重试的(401/400——问题在你,重试一百次也一样)。
6.2 重试双方案
方案 A · SDK 自带(最省事,处理网络类错误):
client = OpenAI(api_key=..., base_url=..., max_retries=2, timeout=30.0)方案 B · tenacity 装饰器(可精确控制哪些异常、退避曲线;01 教程 17.2 装饰器知识变现):
# pip install tenacity from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type @retry(retry=retry_if_exception_type((openai.RateLimitError, openai.APITimeoutError, openai.APIConnectionError, openai.APIStatusError)), wait=wait_exponential_jitter(initial=1, max=30), # 指数退避+随机抖动 stop=stop_after_attempt(4)) def safe_chat(messages, **kw): return client.chat.completions.create(model=MODEL, messages=messages, **kw)为什么要抖动(jitter):并发的 100 个请求同时 429、又同时在第 2 秒重试 = 集体二次撞墙;各自随机等待错峰重试才是解。这是面试"429 怎么处理"的满分尾句。
6.3 成本记账器(挂进综合项目)
class CostTracker: # 价格随行就市,以官网为准;单位:元/百万token PRICES = {"deepseek-chat": (2.0, 8.0)} # (输入, 输出) 量级示意 def __init__(self): self.prompt_tokens = self.completion_tokens = self.calls = 0 def record(self, usage, model: str = "deepseek-chat"): self.calls += 1 self.prompt_tokens += usage.prompt_tokens self.completion_tokens += usage.completion_tokens def report(self, model: str = "deepseek-chat") -> str: pin, pout = self.PRICES[model] cost = self.prompt_tokens/1e6*pin + self.completion_tokens/1e6*pout return (f"调用{self.calls}次 | 输入{self.prompt_tokens} + " f"输出{self.completion_tokens} tokens | 约 ¥{cost:.4f}")省钱三板斧回顾:稳定前缀吃缓存(DeepSeek 响应里的prompt_cache_hit_tokens字段能直接看到命中量,好奇就打出来)、历史裁剪、小任务用小模型。