用标准库构建探针：诊断免费 Token 的延迟和截断问题

免费令牌并不等同于可用端点。

我的上一个自动化脚本通过了身份验证，却仍然错过了截止时间。问题不在于成本，而在于延迟、截断的输出，以及缺失的 finish_reason 检查。

学习问题：我能否在将免费模型端点接入流水线之前，预测它是否会在我的超时和令牌预算内运行？

这是一个探针，带有一种内置的失败模式。它只使用了 Python 标准库。

无需第三方包

一个 OpenAI 兼容的 chat/completions 端点，或者下面的小型模拟服务器

我正在测试的契约

大多数模型端点接受这种格式：

{
  "model": "mock-llm",
  "messages": [{"role": "user", "content": "..."}],
  "max_tokens": 64,
  "temperature": 0
}

它们返回一个 choices 数组和一个 usage 对象。我关心四个字段：

elapsed_s 来自我自己的计时器

finish_reason: "length" 表示模型达到了上限，它并非自然结束。

第一步：一个用于可复现环境的模拟服务器

首先，创建 mock_llm.py。它模拟一个基础延迟加上每个令牌的小延迟。这使得探针能够以可预测的方式失败，而无需真实的密钥。

from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import time

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length)
        try:
            request = json.loads(body)
        except json.JSONDecodeError:
            self.send_response(400)
            self.end_headers()
            return

        cap = int(request.get("max_tokens", 1))
        time.sleep(0.2 + 0.01 * cap)

        payload = {
            "id": "mock-1",
            "object": "chat.completion",
            "created": int(time.time()),
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": "ok"},
                "finish_reason": "length",
            }],
            "usage": {
                "prompt_tokens": 12,
                "completion_tokens": cap,
                "total_tokens": 12 + cap,
            },
        }
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(payload).encode())

    def log_message(self, fmt, *args):
        print("mock:", fmt % args)

if __name__ == "__main__":
    print("mock listening on :8010")
    HTTPServer(("127.0.0.1", 8010), Handler).serve_forever()

在一个终端中运行它：

python mock_llm.py

现在创建 budget_probe.py。它用三个不同的令牌上限发送相同的提示词，并记录响应。

import json
import os
import time
import urllib.error
import urllib.request

BASE = os.environ.get("LLM_BASE", "http://127.0.0.1:8010/v1")
KEY = os.environ.get("LLM_KEY", "demo")
MODEL = os.environ.get("LLM_MODEL", "mock-llm")

def call(max_tokens):
    payload = {
        "model": MODEL,
        "messages": [
            {"role": "user", "content": "Return one short JSON object only."}
        ],
        "max_tokens": max_tokens,
        "temperature": 0,
    }
    data = json.dumps(payload).encode()
    url = f"{BASE.rstrip('/')}/chat/completions"
    request = urllib.request.Request(
        url,
        data=data,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {KEY}",
        },
        method="POST",
    )

    started = time.perf_counter()
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            raw = response.read()
    except urllib.error.URLError as exc:
        return {"error": str(exc), "elapsed_s": round(time.perf_counter() - started, 3)}

    elapsed = time.perf_counter() - started
    try:
        parsed = json.loads(raw)
        return {
            "elapsed_s": round(elapsed, 3),
            "prompt_tokens": parsed["usage"]["prompt_tokens"],
            "completion_tokens": parsed["usage"]["completion_tokens"],
            "total_tokens": parsed["usage"]["total_tokens"],
            "finish_reason": parsed["choices"][0]["finish_reason"],
        }
    except (KeyError, IndexError, json.JSONDecodeError) as exc:
        return {"error": f"bad payload: {exc}", "elapsed_s": round(elapsed, 3)}

for cap in [1, 64, 256]:
    print(cap, call(cap))

在第二个终端中运行它：

export LLM_BASE=http://127.0.0.1:8010/v1
export LLM_KEY=demo
python budget_probe.py

1 {'elapsed_s': 0.202, 'prompt_tokens': 12, 'completion_tokens': 1, 'total_tokens': 13, 'finish_reason': 'length'}
64 {'elapsed_s': 0.844, 'prompt_tokens': 12, 'completion_tokens': 64, 'total_tokens': 76, 'finish_reason': 'length'}
256 {'elapsed_s': 2.803, 'prompt_tokens': 12, 'completion_tokens': 256, 'total_tokens': 268, 'finish_reason': 'length'}

这些数字是小规模设定的，出于刻意。它们展示了我需要看到的关系：更高的令牌上限意味着更大的延迟。

第三步：一种错误输入

将 budget_probe.py 中的循环改为：

for cap in [1, 64, 1000]:
    print(cap, call(cap))

再次运行。模拟服务器会休眠 0.2 + 0.01 * 1000 = 10.2 秒，但探针有 5 秒的超时限制。第三次调用失败。

1000 {'error': '...timed out...', 'elapsed_s': 5.001}

这就是我想要在自动化真实任务之前发现的失败模式。免费令牌配额不会拯救这个请求。

经过这些之后我应该理解什么

身份验证成功不等于端点可用。

免费令牌降低成本，但不减少等待时间。

大的 max_tokens 上限可以把一个有用的调用变成超时。

finish_reason: "length" 意味着输出被切断了，而不是正常完成。

我需要测量延迟和令牌使用量，而不只是读取状态码。

容易犯的错误

忘记设置超时。默认的网络等待可能让一个糟糕的端点隐藏很长时间。

忽略 finish_reason。被截断的完成响应可能看起来仍然有用。

比较原始令牌价格却忽略延迟和重试。

假设免费层级的行为与付费生产端点相同。

免费端点在这个工作流中适合什么场景

披露：本文是 MonkeyCode 产品推广的一部分。

同样的探针可以指向任何 OpenAI 兼容的端点。MonkeyCode 被其运营者描述为一个开源项目，提供免费模型访问和免费服务器选项。运营者还宣称目前有 3000 万令牌的配额。我将其视为运营者提供的数据，而非永久保证。限制和模型名称会变化，所以在依赖它们之前请验证当前的项目页面。

针对免费端点的实用测试序列应该是：

将 LLM_BASE 和 LLM_MODEL 指向当前文档中的端点。

用 1、64 和 256 个令牌运行探针。

记录 elapsed_s 和 finish_reason。

将响应中的使用量与你自己请求的成本进行比较。

用一个更大的上限值打破它，以找到超时墙。

这告诉我免费层级是一个可用的实验室资源，还是仅仅是一个轻量级的试用版。

谁不应该使用这种方法

只关心模型质量、且已有已知 SLO 的付费端点的人。

需要正常运行时间、延迟保证或安全评估的生产系统。

依赖特定模型身份的工作流。这个探针测试的是 HTTP 契约，而不是模型行为。

后续步骤

添加重试逻辑，测量 20 次运行中的 p50 和 p95 延迟。然后改变提示词长度，观察 prompt_tokens 和 elapsed 时间如何变化。如果你测试了 MonkeyCode 的免费端点，分享你在 256 个令牌时看到的 finish_reason。一个受控的数字比另一张聊天窗口的截图更有用。