实战题解：LLM API 封装器、流式重试与 Agent 工作流实操

在 AI 浪潮下，包括 EliseAI、Decagon、Scale AI、Anthropic、Perplexity 在内的众多企业，在技术电面中经常要求候选人直接编写一个健壮的 LLM API 客户端与工作流调度引擎。

这类题目考察候选人能否处理真实世界中的 API 限制：429 限流、Token 消耗速率控制（Token Bucket Rate Limiter）、指数退避重试（Exponential Backoff with Jitter）、以及结构化 JSON 校验与工具分发（Tool Dispatcher）。

---

核心架构设计

sequenceDiagram
    participant User as 用户请求
    participant RateLimiter as 令牌桶限流器
    participant Wrapper as LLM API Wrapper
    participant MockAPI as 底层 LLM Provider
    participant Tool as 本地工具 Tool

    User->>RateLimiter: 请求发送 Prompt
    RateLimiter->>Wrapper: 获取 Token 许可
    loop 重试机制 (最多 3 次)
        Wrapper->>MockAPI: 发起调用
        alt 返回 429 / 500
            MockAPI-->>Wrapper: 报错
            Wrapper->>Wrapper: 指数退避等待 (Backoff + Jitter)
        else 返回 200 + ToolCall
            MockAPI-->>Wrapper: 解析 JSON / Tool 调用
        end
    end
    Wrapper->>Tool: 执行本地函数
    Tool-->>Wrapper: 返回函数执行结果
    Wrapper-->>User: 最终合成回复

---

工业级标准实现（Python）

import time
import random
import json
from typing import Callable, Dict, Any, Optional

class TokenBucketRateLimiter:
    """令牌桶限流器：控制每秒请求速率 (RPS)"""
    def __init__(self, capacity: float, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate  # 每秒补充多少令牌
        self.tokens = capacity
        self.last_refill = time.time()

    def acquire(self, tokens: float = 1.0) -> bool:
        now = time.time()
        # 补充令牌
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

class LLMClientWrapper:
    def __init__(self, api_caller: Callable, rate_limiter: Optional[TokenBucketRateLimiter] = None):
        self.api_caller = api_caller
        self.rate_limiter = rate_limiter or TokenBucketRateLimiter(capacity=5, refill_rate=2)
        self.tools: Dict[str, Callable] = {}

    def register_tool(self, name: str, func: Callable):
        """注册本地可执行工具"""
        self.tools[name] = func

    def call_with_retry(self, prompt: str, max_retries: int = 3, base_delay: float = 1.0) -> Dict[str, Any]:
        """带指数退避与限流控制的调用"""
        for attempt in range(max_retries):
            # 1. 尝试获取速率令牌
            while not self.rate_limiter.acquire(1.0):
                time.sleep(0.1)

            try:
                # 2. 发起底层调用
                response = self.api_caller(prompt)
                
                # 3. 校验并处理 Tool Call
                if response.get("type") == "tool_call":
                    tool_name = response["tool_name"]
                    args = response.get("arguments", {})
                    if tool_name in self.tools:
                        tool_result = self.tools[tool_name](**args)
                        # 将工具结果回填给二次请求
                        return self.call_with_retry(f"{prompt} [Tool Output: {json.dumps(tool_result)}]")
                    else:
                        raise ValueError(f"Unknown tool: {tool_name}")

                return response

            except Exception as e:
                # 429 限流或 5xx 错误触发退避
                if attempt == max_retries - 1:
                    raise e
                # 指数退避 + 随机抖动 Jitter
                sleep_time = (base_delay * (2 ** attempt)) + random.uniform(0.1, 0.5)
                time.sleep(sleep_time)
        
        raise RuntimeError("Max retries exceeded")

---

核心考察细节

1. Jitter（随机抖动）的必要性：
   • 在分布式客户端中，若所有挂掉的请求都在同一时间点（如严格的 1s、2s、4s）重试，会瞬间形成惊群效应（Thundering Herd）再次压垮服务器。加入随机 Jitter 可以将重试流量在时间线上平滑打散。
2. 结构化输出校验（JSON Schema Validation）：
   • 在实际面试中，通常要求对模型输出的非标准 JSON 做正则修复（如多余的 markdown 反引号、尾部逗号等），展现候选人的工程实战严谨度。

---

相关实战资源

• 2026 北美 15 家热门 AI Startup 面试真题合集
• In-Memory File System 实战全攻略
• AI Startup 面试全景准备指南
• SDE 1 对 1 定制面试辅导