LLM 调用的枯燥周边：FastAPI 生产级实战笔记

今年我学到的大部分东西都来自构建那些有趣的部分：检索、prompt、agent 循环。这些东西思考起来很有意思。

然后我在一个项目前加了一个 FastAPI 端点，拿给朋友看，结果大约四分钟就被他玩坏了。不是恶意的——他只是问了一个很长很奇怪的问题，请求就一直在那里转圈，直到他关掉了标签页。

那一刻我才意识到，模型可能只占实际工作的三分之一。剩下的都是包裹在它周围的东西，这些东西很无聊，也没人写教程，因为写起来真的没意思。

这就是我最终折腾出来的方案。我还是个学生，所以把它当作笔记而不是过来人的建议。有些地方我可能理解错了。

默认情况下调用没有超时，这是个问题

这是第一个坑到我的。

我以为某处会有个合理的默认值。确实有，但取决于你的客户端，往往要么非常长，要么实际上不存在。所以当提供商变慢时，你的请求不会失败，只是等着。而在等待的过程中，它占着一个什么也做不了的 worker。

response = await litellm.acompletion(
    model="deepseek/deepseek-chat",
    messages=messages,
    timeout=30,
)

三十秒一开始我觉得挺激进的，然后我实际测量了 p95，才发现没有任何合法请求会超过十五秒。如果一个调用到了三十秒，说明已经出问题了，再等下去也没有帮助。

有件事困惑了我一阵：如果你的请求在一个有自己的超时机制的东西后面（nginx、云负载均衡器、API 网关），而你的超时比它的长，你会得到最糟糕的版本。用户收到 504，而你的调用还在继续运行、继续烧钱，等一个永远没人会看的答案。你的超时应该是最短的那个。

重试很容易做对一半

所有人都告诉你用指数退避来重试。这部分没问题，大多数库都帮你做好了。

我没考虑到的是，不同的错误意味着不同的东西。

429 表示慢下来，你太快了，此时退避是正确的。500 表示提供商出了问题，重试是合理的。400 表示你的请求格式有问题，重试一千次也会得到完全相同的错误，同时你还要花钱去买这个教训。

response = await litellm.acompletion(
    model="deepseek/deepseek-chat",
    messages=messages,
    timeout=30,
    num_retries=2,
)

两次重试，不是五次。我曾经有一个 bug，重试循环和验证失败互相喂养，同一个请求在我注意到之前已经发出去了十几二十次。没什么大事发生，因为我的测试语料库很小，每次就几分钱，但同样的 bug 形态放在真实工作负载上，就是你向别人解释账单的的方式。

另外，记录那些耗尽重试次数后放弃的请求也值得。在日志里只记录暴露给用户的错误很容易，而那些安静地在三次尝试后失败的请求，恰恰是你最想知道的。

两端都需要 token 上限

我知道 max_tokens。所有人都知道 max_tokens。它限制了返回的内容。

我花了更长时间才好好思考输入端，在 RAG 场景下那才是真正的风险所在，因为你把检索到的 chunks 塞进 prompt，而你并不完全控制这些 chunks 包含什么。大多数文档都正常。其中一个巨大无比。它走了和其他所有内容一样的代码路径，成本大约是典型请求的十倍，而我之所以发现它，只是因为我在看每个调用成本的相关原因。

所以现在我在发送前计算 token 数，而不是之后：

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")

def cap_context(chunks, budget=6000):
    kept, used = [], 0
    for chunk in chunks:
        cost = len(enc.encode(chunk))
        if used + cost > budget:
            break
        kept.append(chunk)
        used += cost
    return kept

很粗糙，而且它按位置丢弃 chunks 而不是按相关性，这不太理想。但一个你真正有的粗糙上限，比一个你计划要加的优雅上限有用得多。

兜底比我预想的更重要

提供商会宕机。不经常，但确实会发生，而发生时你什么都做不了只能干等，这是演示时才发现的糟糕体验。

LiteLLM 让这变得真的很简单，这是我使用它的主要原因：

response = await litellm.acompletion(
    model="deepseek/deepseek-chat",
    messages=messages,
    timeout=30,
    num_retries=2,
    fallbacks=["gemini/gemini-2.0-flash"],
)

兜底方案不一定要一样好。这是我一开始误解的地方。差别在于"答案稍微差一点"和"根本没有答案"，用户对前者的包容度要高得多。

但真的要测试一下。我配置了一个兜底方案好一阵，如果真的触发了本来会失败，因为模型名称错了，从来没有走过那条代码路径。我是通过故意塞一个垃圾的主要模型名称进去看看会发生什么来发现这个问题的，花了两分钟，我应该一开始就做这个测试。

验证告诉你什么东西坏了，而不是告诉你该怎么办

Pydantic 很棒。你定义你想要的形状，当模型返回别的东西时，你会得到一个清晰的错误。

from pydantic import BaseModel

class Answer(BaseModel):
    text: str
    confidence: float
    sources: list[str]

困扰我的是，捕获错误只是半个决策。你仍然需要选择接下来怎么办，而我有一段时间没有做这个选择，这意味着我的选择默认就是"给用户扔一个 500"。

我最终考虑过的选项：

重试一次，把验证错误反馈到 prompt 里。对于小的 schema 错误，这招出奇地有效。这也意味着又花一次调用费和多等几秒，所以不是免费的。

回退到更简单的方案。如果结构化版本一直失败，就拿纯文本、丢掉结构，而不是丢掉整个响应。

好好地失败，带一个真实的错误信息。有时候这是对的。但"抱歉，出错了"比一个堆栈跟踪好得多，也比静静地返回一个会在三层之后引发问题的空对象好得多。

这些选项没有哪个普遍正确。关键是有意识地选一个。

消费上限，因为仪表盘是事后才告诉你

如果让我重新开始，这是我第一个要加的东西。

提供商的仪表盘很好。但它们也是回顾性的。它们在你花完钱之后才告诉你花了多少，如果什么东西循环跑了一整夜，你到早上才会发现。

所以我在进程里放了一个计数器：

class SpendGuard:
    def __init__(self, ceiling_usd):
        self.ceiling = ceiling_usd
        self.spent = 0.0

    def record(self, response):
        cost = response._hidden_params.get("response_cost", 0)
        self.spent += cost
        if self.spent > self.ceiling:
            raise RuntimeError(
                f"Spend ceiling hit: ${self.spent:.2f} of ${self.ceiling:.2f}"
            )

就二十行代码，而且很 naive。进程重启时它会重置，跨多个 worker 也不会有帮助，除非你把计数器放到 Redis 这样的共享地方。但它把"无界"变成了"有界"，而这才是真正重要的部分。我整个论文实验跑在大约四英镑的算力上，知道一个 bug 不可能把那变成四百英镑，让我迭代时能自由很多。

如果你在托管提供商上，也要在他们的控制台里设置一个硬性的账单上限。双重保险。

如果重来，我会怎么做

说实话，我会先把这些都写了，再去写那些有趣的部分。

这里的每一项我都是在被坑到之后才加的，这意味着每一个都是一次小惊吓而不是一个主动的决策。代码不多。就一个超时、一个重试上限、两个 token 上限、一个兜底、一个验证分支、一个计数器。开始时放进去可能一个小时都不用，而不用一个个地吃一堑长一智。

模型是所有人都谈论的部分。而围绕它的那些东西，才是决定这个东西能不能在接触到真实用户后存活下来的部分。

如果你比我走得更远，而我这里有什么说错了，我真的想知道。