接入LLM API一年的3条血泪教训：token计量、多模型抽象与生产防护

过去一年，我一直在基于 LLM API 开发功能——GPT、Claude、Gemini，还有通过 Ollama 调用的几个开源模型。这一路走来，消耗了不少额度，两次把生产环境搞挂，也付出了不少学费。

以下是三个我希望当初就有人告诉我的教训。

1. Token 计数不是可选项——它就是全部工作

早期，我把 token 限制当作"有则更好"的东西——随便数一下字符数，然后听天由命。这招一直用得很好，直到有个用户把一个 40KB 的日志文件粘贴到聊天窗口，我的后端默默截断了他们的上下文，给出的答案看起来信心满满，实际上完全错误。

我现在怎么做：每个请求在进入 LLM 之前，都要先过一遍 token 计数器。

import tiktoken

def count_tokens(text: str, model: str = "gpt-4") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

对于多模型设置，我写了一个小的抽象层来把模型名映射到对应的 tokenizer。claude-* 用 Anthropic 的 claude-tokenizer，Gemini 用 google-generativeai，而 Ollama 模型则回退到基于字符数的保守估算。

真正的洞见：你需要两个限制——一个是整个上下文窗口的硬截断，另一个是更宽松的"预留输出"预算，好让模型有空间真正回复。我留了 25% 的上下文窗口给输出 token。如果你的输入 + 预留输出 > 上下文限制，就裁剪输入——而不是反过来。

这一个改动消除了大约 90% 的"为什么 AI 给出的答案不完整？"这类 bug 报告。

2. 流式输出是基本配置，但结构化输出才是难点

每个人都知道应该流式返回 LLM 响应——用户不想盯着一个加载 spinner 等上 15 秒。这个很简单。

难点在于如何处理这个流。如果你从后端调用 LLM，需要向前端返回 JSON，你就不能直接流原始 token。你需要在后端侧实现结构化输出，而且要快。

我的方案：

import asyncio
import json

async def stream_and_parse(prompt: str, schema: dict):
    """Stream LLM response, parse JSON at the end."""
    buffer = []

    async for chunk in llm_stream(prompt):
        buffer.append(chunk)
        # Send raw text to frontend via WebSocket
        await ws.send(chunk)

    full_text = "".join(buffer)
    # Parse the structured result from completed text
    return json.loads(full_text)

但这里有个坑：不是所有 LLM 的 JSON 输出能力都一样。Claude 做得很好。Gemini 2.0 Flash 很快，但偶尔会漏掉闭合花括号。GPT-4o 处理得不错，但有时会把 JSON 包在 markdown 代码块里，需要你手动去掉。

我的经验法则：始终用你的 schema 验证解析后的 JSON。如果解析失败，用一个 system prompt 重试："Return ONLY valid JSON, no markdown fences, no explanations." 第二次重试能修复 95% 的失败。

对于生产环境，我也会记录每一次 JSON 解析失败时的原始响应——一个静默的解析错误会在三个调用者之后变成"NoneType has no attribute"错误，没有什么比这更让人崩溃的了。

3. 降级不是功能——它是架构

我的第一次 LLM 集成是这样的：

user → my API → OpenAI → response

当 OpenAI 发生故障时（半年内发生了两次），我的整个功能就死了。用户收到超时错误，却没有得到任何解释。

user → my API → primary provider (Claude)
              → fallback #1 (GPT-4o) — different provider
              → fallback #2 (Gemini Flash) — cheap, always available

但仅仅链式配置降级是不够的。你需要断路器：

import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=3, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure_time = 0

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()

    def is_open(self):
        if self.failures >= self.threshold:
            if time.time() - self.last_failure_time < self.reset_timeout:
                return True  # Circuit is open — skip this provider
            self.failures = 0  # Reset timeout passed, try again
        return False

如果 Claude 连续返回三个 5xx 错误，断路器就会打开 60 秒。请求会直接跳过它去到 GPT-4o，而不是在那里等待超时。当断路器重置时，它会用单个请求试水，然后才完全重新打开。

教训：LLM API 是外部依赖。要像对待数据库连接或第三方支付服务一样对待它们。它们一定会出问题，而你的用户不应该察觉到。

基于 LLM 构建应用并不难——不是因为模型复杂，而是因为围绕它们的基础设施需要具备弹性。Token 管理、结构化解析和提供商降级，不是上线后才补的可选功能。它们是你第一天就需要具备的东西。

这三个模式已经在生产环境运行了几个月。它们算不上多聪明，也算不上多新颖。只是久经考验而已。

你自己有什么 LLM API 的踩坑故事吗？很想知道，欢迎在评论区分享。