Function Calling 实战：JSON Schema 设计、并行调用与错误处理

from pydantic import BaseModel
from openai import OpenAI

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

# 严格模式：100% schema 符合
response = client.responses.parse(
    model="gpt-4o-2024-08-06",
    input="Alice 和 Bob 周五要去科技展。",
    text_format=CalendarEvent,
)
event = response.output_parsed  # 一定是 CalendarEvent 类型

// ❌ 烂：模型瞎猜
{
  "name": "search_orders",
  "parameters": {
    "properties": {
      "q": {"type": "string"}
    }
  }
}

// ✅ 好：description 明确
{
  "name": "search_orders",
  "description": "搜索订单。支持按订单号、客户姓名、下单时间筛选。返回最近的 20 条。",
  "parameters": {
    "properties": {
      "q": {
        "type": "string",
        "description": "搜索关键词，可以是订单号（ORD-XXXX 格式）、客户姓名或手机号后四位"
      },
      "status": {
        "type": "string",
        "enum": ["pending", "paid", "shipped", "completed"],
        "description": "订单状态过滤。'pending'=待付款, 'paid'=已付款, 'shipped'=已发货, 'completed'=已完成"
      }
    },
    "required": ["q"]
  }
}

// ❌ string：模型可能返回 "Pending" / "PENDING" / "pending" / "待付款"
"status": {"type": "string"}

// ✅ enum：模型只能从给定值里选
"status": {"type": "string", "enum": ["pending", "paid", "shipped", "completed"]}

{
  "properties": {
    "phone": {
      "type": ["string", "null"],
      "description": "客户手机号，没有就传 null"
    }
  },
  "required": ["phone"]
}

{
  "properties": {
    "phone": {"type": "string", "description": "可选"}
  },
  // 严格模式下漏掉 required 会报错
}

// ❌ 4 层嵌套：模型经常填错
{
  "user": {
    "profile": {
      "address": {
        "city": {"type": "string"}
      }
    }
  }
}

// ✅ 拍平
{
  "user_name": {"type": "string"},
  "user_city": {"type": "string"},
  "user_country": {"type": "string"}
}

# Anthropic tool search tool (2025)
tools = [
    {
        "name": "search_tools",
        "description": "当你不确定该用哪个工具时，先调用这个搜索工具"
    },
    {
        "name": "tool_search_result_2025",
        # 模型调用 search_tools 后返回的工具描述
    }
]

# OpenAI
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[...],
    tools=[get_product_tool, compare_tool]
)

# response.choices[0].message.tool_calls 可能是多个
for tool_call in response.choices[0].message.tool_calls:
    if tool_call.function.name == "get_product":
        product_id = json.loads(tool_call.function.arguments)["product_id"]
        # 并行执行
        ...

# ✅ 正确：一次返回所有结果
messages.append({
    "role": "user",
    "content": [
        {"type": "tool_result", "tool_use_id": "id_1", "content": json.dumps(result_A)},
        {"type": "tool_result", "tool_use_id": "id_2", "content": json.dumps(result_B)},
    ]
})

# ❌ 错误：分开发两次，模型会学成串行调用
messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": "id_1", ...}]})
# 再发一次
messages.append({"role": "user", "content": [{"type": "tool_result", "tool_use_id": "id_2", ...}]})

# ✅ tool_result 在前，文本在后
content = [
    {"type": "tool_result", "tool_use_id": "id_1", "content": "..."},
    {"type": "tool_result", "tool_use_id": "id_2", "content": "..."},
    {"type": "text", "text": "结果如上"}
]

# 模型生成的"程序"
result = []
for product_id in ["A", "B", "C"]:
    data = get_product(product_id=product_id)  # 沙箱内工具调用
    if data["price"] > 100:
        result.append(data)
return summarize(result)

import random
import time

def call_with_retry(fn, max_retries=5):
    for attempt in range(max_retries):
        try:
            return fn()
        except (RateLimitError, APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise
            
            # 指数退避 + 抖动
            base_delay = min(2 ** attempt, 32)  # 1, 2, 4, 8, 16, 32...
            jitter = random.uniform(0, base_delay * 0.5)
            delay = base_delay + jitter
            
            print(f"Attempt {attempt+1} failed: {e}. Retry in {delay:.2f}s")
            time.sleep(delay)

# OpenAI SDK 内置重试
from openai import OpenAI
client = OpenAI(
    max_retries=5,           # 最多重试 5 次
    timeout=30.0,            # 单次超时 30 秒
)

def execute_with_error_feedback(tool_call):
    try:
        result = execute_tool(tool_call)
        return {"tool_call_id": tool_call.id, "content": json.dumps(result)}
    except ValidationError as e:
        # Schema 校验失败：告诉模型具体错在哪
        return {
            "tool_call_id": tool_call.id,
            "content": json.dumps({
                "error": "validation_failed",
                "message": str(e),
                "fix_suggestion": "请检查参数类型，必填字段是否齐全"
            }),
            "is_error": True  # Anthropic 用这个标记
        }
    except BusinessError as e:
        # 业务错误：让模型决定怎么应对
        return {
            "tool_call_id": tool_call.id,
            "content": json.dumps({
                "error": "business_rule_violation",
                "message": str(e)
            }),
            "is_error": True
        }

class CircuitBreaker:
    def __init__(self, failure_threshold=0.5, min_calls=10, reset_timeout=60):
        self.failure_rate = 0.0
        self.call_count = 0
        self.failure_count = 0
        self.state = "closed"  # closed / open / half-open
        self.opened_at = None
        self.failure_threshold = failure_threshold
        self.min_calls = min_calls
        self.reset_timeout = reset_timeout
    
    def call(self, fn, *args):
        # 熔断状态：直接拒绝
        if self.state == "open":
            if time.time() - self.opened_at > self.reset_timeout:
                self.state = "half-open"
            else:
                raise CircuitOpenError("Circuit breaker is open")
        
        try:
            result = fn(*args)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.call_count += 1
        if self.state == "half-open":
            self.state = "closed"
            self.failure_count = 0
            self.call_count = 0
    
    def on_failure(self):
        self.failure_count += 1
        self.call_count += 1
        
        if self.call_count >= self.min_calls:
            rate = self.failure_count / self.call_count
            if rate > self.failure_threshold:
                self.state = "open"
                self.opened_at = time.time()
                print(f"⚠️ Circuit breaker opened. Failure rate: {rate:.1%}")

# ❌ 粗：一个超级工具
def manage_order(action, order_id, **kwargs):
    if action == "create":
        ...
    elif action == "update":
        ...
    elif action == "cancel":
        ...

# ✅ 细：拆成三个工具
def create_order(customer_id, items, ...): ...
def update_order(order_id, changes, ...): ...
def cancel_order(order_id, reason, ...): ...

# ❌ 返回自然语言描述
def get_order(order_id):
    return f"订单 ORD-123 状态为已发货，预计 3 天后送达"

# ✅ 返回结构化 JSON，让模型自己组织语言
def get_order(order_id):
    return {
        "order_id": "ORD-123",
        "status": "shipped",
        "estimated_arrival": "2024-12-04",
        "carrier": "顺丰"
    }

def search_articles(query, limit=10):
    results = db.search(query, limit=limit)
    
    # 截断每个结果
    truncated = [
        {**r, "content": r["content"][:500] + "..." if len(r["content"]) > 500 else r["content"]}
        for r in results
    ]
    
    return {
        "count": len(results),
        "results": truncated,
        "next_page_token": "..." if len(results) == limit else None
    }

import structlog

logger = structlog.get_logger()

def log_tool_call(tool_name, args, result, duration_ms, error=None):
    logger.info(
        "tool_call",
        tool=tool_name,
        args_preview=str(args)[:200],
        result_size=len(str(result)),
        duration_ms=duration_ms,
        error=str(error) if error else None,
        timestamp=datetime.utcnow().isoformat()
    )

metrics = {
    "tool_call_count": 0,            # 总调用次数
    "tool_call_errors": 0,           # 失败次数（分网络 vs 参数错误）
    "tool_retry_count": 0,           # 模型自纠重试次数
    "tool_choice_distribution": {},  # 工具选择分布（看有没有冷门工具）
    "avg_tool_duration_ms": 0,       # 平均耗时
}