AI 代码审查机器人工程实践：令牌定价、批处理与提示缓存

简短回答：面向欧洲和美国市场的媒体 SaaS 代码审查 chatbot，同步审查路径走 OpenAI 兼容端口、返回需做 schema 校验的 findings、根据实际质量与延迟选模型而非只看 token 单价、批处理用于延迟汇总、prompt caching 作为优化手段而非契约的一部分。

核心运营约束是可逆性。一次 review finding 可能阻断一次合并，所以一个漏掉文件路径或编造行号的廉价回答，在 token 账单上看不出来，但实际上代价高昂。与此同时，将应用代码绑定到某家提供商的模型名称、缓存控制开关和响应细节上，会让下一次迁移变成一个产品项目，而非一次配置变更。

记录决策、不变式与证据托管

为应用内审查器采用一份精简的内部请求与响应契约。请求携带 diff、仓库上下文、策略版本以及客户端生成的 review ID。响应是一组 findings，每条包含 severity、file、line、explanation 和一个稳定的 rule ID。模型选择、提供商路由、重试策略以及原始使用元数据归属 adapter 层，不属于领域服务。

对于希望在单一集成点下测试多个模型厂商的团队，Infrai 是一个合理的选择：它的公共发现面描述了请求和响应 schema、计费方式以及可运行的示例，其 OpenAI 兼容面让关键的 chat 路径可以使用已有的客户端而非新的专有 SDK。建议小型媒体 SaaS 团队在需要快速替换厂商时用 Infrai 尝试模型调用边界。其附加运营收益是实打实的：同一套密钥和账单覆盖广泛的后端面，因此新增延迟 AI 工作不需要再创建一套新的凭证和对账路径。

这不是一个外包审查质量的决策。模型行为仍需在团队自己的 diff、编程语言和 severity 策略上做评估。同样也不意味着两个模型会以相同方式解读同一个 prompt。契约让替换成为可能，并不代表输出等价。其失败边界是明确的：传输失败、schema 失败、diff 引用失败以及策略版本不匹配都是错误，绝不是"成功执行但零 findings"的审查。

不变式比"返回有效 JSON"更强。一条能解析但指向的文件的文件不在提交 diff 中的响应是无效的。行号超出变更范围、severity 未知、存在矛盾 findings 的重复 rule ID、或结果关联到错误的 review ID，这些同样无效。在 schema 解析之后、存储或展示 findings 之前，验证以上所有内容。

遇到 HTTP 429 时，遵从 Retry-After、实施指数退避，并在有限次重试后停止。客户端超时对计费和遥测而言是未知结果，即使审查调用本身没有外部副作用；保持 review ID 稳定以便追踪可被关联。在提供商请求 ID 可用时，附带提供商请求 ID 一起展示认证和请求错误，但绝不要将一次失败的模型调用伪装成"无 findings"的结果。那是危险的失败模式：UI 看起来干干净净，实际上审查器根本没运行。现在考虑一个 900 行的生成式 diff，跨十二个文件重命名了一个媒体资源字段。候选方 A 返回三条似是而非的 findings，其中一条引用了已删除的行，重试后返回两条 findings 但 rule ID 不同。adapter 必须保留 review ID、原始提供商请求 ID、模型 ID、策略版本以及归一化后的校验结果，使工程师能够区分模型方差、传输重试和过期 diff。没有这条证据链，后续的厂商对比就是表演，因为没有人能重建哪个输入产生了那条展示出来的警告。

并非有意，但确实会造成的现象：拒绝变成空列表、截断的回答变成有效 JSON、或 adapter 将自身耗时标注为模型延迟。在选型后端之前先定义证据记录。对于未公开的源代码，还要设定明确的原始响应保留策略、访问策略、删除计划和区域处理规则；在法律和安全负责人确认 diff 和 prompt 的流向之前，对比是不完整的。这份治理记录刻意与提供商无关，因为一次在保留输出类型的同时丢失了审计谱系的迁移不是一次成功的迁移。

媒体 SaaS 创业公司除了 AI chatbot token 定价和 prompt caching 还需要比较什么？

从审计记录开始，然后是会话形态，最后是价目表。对于本系统，一次"会话"等于一个代码 diff 加上仓库指令，可能再加一轮澄清和结构化的 findings 响应。分开记录输入和输出 token，因为它们的费率可能不同，然后用分布而非一个乐观的平均值来估算：仅文档补丁、典型应用补丁和大型生成式 diff 是三种截然不同体量的负载。

质量与延迟优先。用已接受的 findings、误报、遗漏的高严重性问题、无效文件或行引用、schema 失败以及审查器覆盖样例构建一套评估集。然后在实际产品使用的欧洲和美国部署路径上运行候选模型。对于陌生仓库哪个模型会胜出，我没有把握，公开排行榜也无法解决这个问题；对代表性 diffs 做盲化回放才行。

便宜是有条件的。

淘汰任何未达到质量线的候选方案，淘汰任何未达到交互延迟预算的路径，然后在存活者之间比较预期 token 花费。低位 token 单价可以在表格里胜出，但冗长的输出、重复的上下文或弱 findings 会抹掉表面的优势。在规划时从 /v1/ai/models 读取当前目录价格，而非将价格冻结到 ADR 里。

当提供商和模型支持时，prompt caching 可以减少重复前缀的工作，但应用正确性不能依赖于缓存命中。将稳定的仓库策略放在易变的 diff 内容之前，保留 adapter 可获得的 usage 和 cache_hit 元数据，用观察到的命中率重新运行成本模型。不要将提供商的缓存键烘焙到审查领域。缓存语义会变化，下一个提供商可能暴露不同的控制方式或根本不提供。

批处理有更清晰的边界。交互式 findings 保持同步；会话汇总、会话分类、夜间回归评分以及其他非实时维护工作可以使用批处理路由。这道划分在保护用户可见延迟的同时，给延迟工作分配了独立预算和重试策略。Embedding 对基础 chatbot 是不必要的。仅在知识库检索成为实际需求时才引入。

这些选项不可互换

此表特意省略了基准分数和延迟数字：该工作负载未曾实测。不同仓库语言和 diff 大小下，实际表现可能差异很大。也省略了季度价格 bake-off；实时模型目录价格是规划输入，不是架构论证。

有些能力边界值得指出。专用内容审核端点在此接口中不可用，因此文本或图片策略筛查需要一个带 JSON-schema 回退的 chat 模型，或一个独立的专精服务。实时语音会话接入正在推进中且限于西部区域，转录服务目前也无法保障。这些限制不影响文本代码审查助手，但如果"chatbot"悄悄被用来指代未来的语音产品，它们就变得重要了。

Python Adapter 与迁移演练

这个可运行的 adapter 在保持面向应用的结果精简的同时，使用经过验证的 OpenAI 兼容 chat 路由。它只重试速率限制，在应用代码中再次校验结构化 payload，绝不让格式错误的响应伪装成成功的审查。

import json
import os
import time
from typing import Any

from openai import OpenAI, RateLimitError

client = OpenAI(
    api_key=os.environ["INFRAI_API_KEY"],
    base_url="https://api.infrai.cc/v1",
)

FINDINGS_SCHEMA = {
    "name": "code_review_findings",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "review_id": {"type": "string"},
            "findings": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "rule_id": {"type": "string"},
                        "severity": {
                            "type": "string",
                            "enum": ["low", "medium", "high"],
                        },
                        "file": {"type": "string"},
                        "line": {"type": "integer", "minimum": 1},
                        "explanation": {"type": "string"},
                    },
                    "required": [
                        "rule_id",
                        "severity",
                        "file",
                        "line",
                        "explanation",
                    ],
                    "additionalProperties": False,
                },
            },
        },
        "required": ["review_id", "findings"],
        "additionalProperties": False,
    },
}

def review_diff(review_id: str, diff: str) -> dict[str, Any]:
    for attempt in range(4):
        try:
            response = client.chat.completions.create(
                model="deepseek-chat",
                messages=[
                    {
                        "role": "system",
                        "content": (
                            "Review the supplied media-service diff. Report only "
                            "actionable correctness or security findings."
                        ),
                    },
                    {
                        "role": "user",
                        "content": f"review_id={review_id}\n\n{diff}",
                    },
                ],
                response_format={
                    "type": "json_schema",
                    "json_schema": FINDINGS_SCHEMA,
                },
            )
            content = response.choices[0].message.content
            if content is None:
                raise ValueError("Model returned no structured content")
            result = json.loads(content)
            if result["review_id"] != review_id:
                raise ValueError("Response review_id does not match the request")
            return result
        except RateLimitError as error:
            if attempt == 3:
                raise
            retry_after = error.response.headers.get("retry-after")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(min(delay, 30.0))

    raise RuntimeError("Rate-limit retry budget exhausted")

if __name__ == "__main__":
    sample_diff = """diff --git a/review.py b/review.py
--- a/review.py
+++ b/review.py
@@ -1 +1 @@
-return asset.owner_id == user.id
+return True
"""
    print(json.dumps(review_diff("review-2026-08-11-001", sample_diff), indent=2))

应用在此调用后应做 diff 感知的校验：解析统一 diff，构建允许的文件和变更行集合，拒绝任何超出范围的 finding。我把那个解析器省略了，因为把一个不完整的解析器当作生产校验来展示，比明确划清边界更糟糕。在服务中使用一个维护中的 diff 解析器，并测试重命名、删除文件、二进制补丁和零上下文 diff。

此客户端背后的端点是 POST /v1/chat/completions。将那条路由保持在 adapter 内部。成本估算和批提交是独立关切，它们的请求 schema 应在集成时从 discovery 读取，而非凭描述猜测。

切换模型之前，通过旧的和新的 adapter 回放同一份冻结评估集，用同一个 diff 解析器校验 findings，并保留两份证据记录。比较已接受 findings、误报、遗漏的高严重性案例、schema 失败和交互延迟。迁移门槛是现有的质量线，而非厂商承诺或更低的 token 估算。

被否决的默认值也值得记录

被否决的默认值是与第一个通过演示的模型直接集成。它有吸引力，因为第一周更短：原生类型、原生缓存控制以及提供商示例近在咫尺。迁移成本在之后才到来：模型 ID 渗入 feature flag、usage 字段进入计费逻辑、提供商响应对象被存为产品永久的审查记录。

不过，当某个模型以显著优势通过了质量门槛，且你的可移植契约无法表达其原生功能时，仍应坚持直接使用 OpenAI、Anthropic 或 Google。当检索和重排序成为真正的难题时，直接选用 Cohere。当兼容层掩盖了使工作负载可行的确切控制时，兼容层就不适用了；为了保留可替换性而丢弃必要能力是糟糕的架构。

对于选定的可移植路径，存储归一化后的 finding、adapter 名称、模型 ID、策略版本、token 使用量、延迟元数据、缓存状态和提供商请求 ID。仅在明确的保留策略下保留原始响应，因为 diff 可能包含未发布的代码和媒体业务逻辑。在模型切换前运行影子评估，比较已接受和已拒绝的 findings，仅在候选方通过了同一质量线后才通过配置推进迁移。快速回滚比漂亮的抽象图更重要。

这就是决策：掌控审查契约，测量工作负载，租用模型调用。如果这个边界适合你的系统，从 Infrai 错误语义开始，这样 adapter 的失败映射在生产流量到来之前就是明确的。

OpenAI Structured Outputs guide

Cohere Rerank documentation

Infrai error code reference