Tool Use 与工具契约

from dataclasses import dataclass
from typing import Any, Callable
import json
@dataclass
class ToolResult:
    """统一返回协议：状态是显式的，绝不靠异常或空值暗示"""
    status: str            # ok | empty | bad_args | forbidden | not_found | timeout | upstream
    data: Any = None
    message: str = ""      # 给模型看的一句话说明
    retryable: bool = False
    side_effect_done: bool = False   # 副作用是否已经发生（决定能否安全重试）
    cost: float = 0.0
    def to_observation(self) -> str:
        """回喂给模型的字符串：状态 + 说明 + 精简数据"""
        head = f"[{self.status.upper()}] {self.message}"
        if self.data is None:
            return head
        body = json.dumps(self.data, ensure_ascii=False)
        return f"{head}\n{body[:1800]}"
def search_articles(q: str, limit: int = 10) -> ToolResult:
    # --- 第三层：执行前校验（不信任任何上游） ---
    if not isinstance(q, str) or not q.strip():
        return ToolResult("bad_args", message="参数 q 不能为空，请给出中文关键词。")
    if not (1 <= limit <= 20):
        return ToolResult("bad_args",
                          message=f"limit 必须在 1-20 之间，当前为 {limit}。",
                          data={"allowed": [1, 20]})
    try:
        rows = db_search(q, limit)                     # 真实查询
    except TimeoutError:
        return ToolResult("timeout", message="查询超时，这是瞬时问题，可原样重试。",
                          retryable=True)
    except PermissionError:
        return ToolResult("forbidden", message="当前身份无权访问该数据源，重试无用。")
    # --- 空结果不是错误 ---
    if not rows:
        return ToolResult("empty",
                          message=f"查询成功，但没有匹配「{q}」的文章。"
                                  f"建议换更宽的关键词，或确认该主题确实没有内容。",
                          data={"items": [], "hint": "可尝试同义词或去掉限定词"})
    return ToolResult("ok", data={"items": rows[:limit], "total": len(rows)},
                      message=f"命中 {len(rows)} 条。",
                      cost=0.001)
# ---------------------------------------------------------------
# 有副作用的工具必须带幂等键
# ---------------------------------------------------------------
IDEMPOTENCY_STORE: dict[str, ToolResult] = {}
def send_notification(user_id: str, text: str, idempotency_key: str) -> ToolResult:
    """
    不可逆操作的正确姿势：
      1. 要求调用方提供幂等键
      2. 同一个键重复调用直接返回上次结果，不重复执行
      3. 返回值必须说明「副作用是否已经发生」
    """
    if idempotency_key in IDEMPOTENCY_STORE:
        prev = IDEMPOTENCY_STORE[idempotency_key]
        return ToolResult("ok", data=prev.data,
                          message="该请求此前已成功执行过，本次未重复发送（幂等命中）。",
                          side_effect_done=True)
    if not text.strip():
        return ToolResult("bad_args", message="通知内容不能为空。")
    resp = api_send(user_id, text)                     # 真实发送
    res = ToolResult("ok", data={"message_id": resp["id"]},
                     message="已发送。", side_effect_done=True, cost=0.01)
    IDEMPOTENCY_STORE[idempotency_key] = res
    return res
def db_search(q, limit): raise NotImplementedError
def api_send(user_id, text): raise NotImplementedError