1. 为什么我要用 Python 微服务压测 Kimi K3 的 APIKimi K3 的长文本能力已经被聊得很多了20 万字上下文、跨段落信息检索、技术报告摘要这些确实是它的看家本领。但如果你打算把它接进生产系统真正决定能不能上线的往往不是“它能不能读懂长文”而是“它在高并发下会不会超时”“错误码返回得规不规范”“重试之后会不会重复计费”。这些问题光靠对话窗口里聊几句是测不出来的。我这次做的事情很具体写一个 Python 微服务把 Kimi K3 的 API 包在里面然后用压测脚本打它观察并发调用、超时重试、错误码处理这三件事的真实表现。压测载体选 Python 是因为它写起来快、改起来也快微服务框架用 FastAPI异步 HTTP 客户端用 httpx压测脚本用 asyncio 直接怼并发。整套东西不需要很重的依赖本地就能跑起来。适合谁看如果你正在评估 Kimi K3 能不能进你的技术栈或者你已经决定要用但不确定接入层该怎么写这篇文章里的配置骨架和压测脚本可以直接复制去改。如果你只是想了解 Kimi K3 的 API 工程表现不看代码看结论也有参考价值。整篇的节奏是先讲清楚问题场景再给可复制的配置然后跑验证最后把踩过的坑列出来。2. TaoToken 前置统一 Key 与 config.toml 骨架在写压测代码之前先把 Key 管理和配置骨架定下来。我试过把 Key 硬编码在代码里后来换环境的时候改得想哭所以这次直接用 config.toml 把配置抽出来Key 走环境变量注入。TaoToken 在这里的角色是统一接入层。你可以在它的控制台里创建一个 Key然后这个 Key 既能调 Kimi K3也能调其他模型切换模型只需要改 config.toml 里的 model 字段不用换 Key、不用换 base_url。对于压测场景来说这很实用因为你可以用同一套压测脚本对比不同模型在相同并发下的表现。先看 config.toml 的骨架[app] name kimi-k3-stress host 0.0.0.0 port 8000 log_level info [llm] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model kimi-k3 timeout_seconds 30 max_retries 3 retry_backoff_base 2 [stress] concurrency 20 total_requests 200 prompt 请用一句话解释什么是微服务架构中的服务熔断。 max_tokens 128 temperature 0.2几个关键点说明一下。base_url填https://taotoken.net/api这是 API 入口不要加多余的路径。api_key_env指定从哪个环境变量读 Key这样配置文件可以进版本库Key 不会泄露。timeout_seconds设 30 秒因为 Kimi K3 在长文本场景下响应时间会比短 prompt 长但压测用的是短 prompt30 秒足够覆盖绝大多数情况。max_retries设 3配合retry_backoff_base 2做指数退避。环境变量这样设置export TAOTOKEN_API_KEY你的KeyKey 的获取路径是 TaoToken 控制台的 API Keys 页面创建之后复制出来即可。如果你还没创建可以去 API Keys 管理页 建一个。接入文档在 这里里面有完整的请求格式和错误码说明压测之前建议先扫一眼。3. 可复制配置FastAPI 微服务 异步压测脚本3.1 微服务端封装 Kimi K3 调用微服务的职责很简单接收一个 prompt转发给 Kimi K3返回结果和耗时。但为了压测能观察到重试和错误码需要在调用层把 httpx 的超时、重试、异常处理都写清楚。# service.py import os import time import tomllib import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel with open(config.toml, rb) as f: config tomllib.load(f) app FastAPI(titleKimi K3 Stress Target) class PromptRequest(BaseModel): prompt: str max_tokens: int 128 temperature: float 0.2 class PromptResponse(BaseModel): content: str latency_ms: float retries: int status: str def get_api_key() - str: key os.environ.get(config[llm][api_key_env]) if not key: raise RuntimeError(f环境变量 {config[llm][api_key_env]} 未设置) return key async def call_kimi(prompt: str, max_tokens: int, temperature: float): url f{config[llm][base_url]}/v1/chat/completions headers { Authorization: fBearer {get_api_key()}, Content-Type: application/json, } payload { model: config[llm][model], messages: [{role: user, content: prompt}], max_tokens: max_tokens, temperature: temperature, } timeout config[llm][timeout_seconds] max_retries config[llm][max_retries] backoff_base config[llm][retry_backoff_base] last_error None for attempt in range(max_retries): try: async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post(url, headersheaders, jsonpayload) if resp.status_code 200: data resp.json() content data[choices][0][message][content] return content, attempt elif resp.status_code 429: last_error f429 限流: {resp.text[:200]} elif resp.status_code 401: raise HTTPException(status_code401, detailKey 无效) elif 500 resp.status_code 600: last_error f{resp.status_code} 服务端错误 else: last_error f{resp.status_code}: {resp.text[:200]} except httpx.TimeoutException: last_error 请求超时 except httpx.ConnectError as e: last_error f连接失败: {e} if attempt max_retries - 1: wait backoff_base ** attempt await asyncio.sleep(wait) raise HTTPException(status_code502, detailf重试耗尽: {last_error}) app.post(/generate, response_modelPromptResponse) async def generate(req: PromptRequest): start time.perf_counter() content, retries await call_kimi(req.prompt, req.max_tokens, req.temperature) latency (time.perf_counter() - start) * 1000 return PromptResponse( contentcontent, latency_msround(latency, 2), retriesretries, statusok, )这段代码里有几个设计决策值得说。第一重试逻辑放在服务端而不是压测脚本里因为生产环境的重试通常是在接入层做的压测要模拟真实链路。第二429 和 5xx 走重试401 直接抛出不重试因为 Key 无效重试多少次都没用。第三每次重试都新建httpx.AsyncClient这在压测场景下会有额外开销但能避免连接池状态污染生产环境可以改成复用客户端。3.2 压测脚本asyncio 并发打点压测脚本用 asyncio 的 Semaphore 控制并发数每个请求记录耗时和重试次数最后汇总 P50、P95、P99 和错误分布。# stress.py import asyncio import time import tomllib import httpx import statistics with open(config.toml, rb) as f: config tomllib.load(f) STRESS config[stress] TARGET fhttp://127.0.0.1:{config[app][port]}/generate async def one_request(sem, client, idx, results): async with sem: payload { prompt: STRESS[prompt], max_tokens: STRESS[max_tokens], temperature: STRESS[temperature], } start time.perf_counter() try: resp await client.post(TARGET, jsonpayload, timeout60) latency (time.perf_counter() - start) * 1000 if resp.status_code 200: data resp.json() results.append({ idx: idx, status: ok, latency: latency, retries: data.get(retries, 0), }) else: results.append({ idx: idx, status: fhttp_{resp.status_code}, latency: latency, retries: -1, }) except Exception as e: latency (time.perf_counter() - start) * 1000 results.append({ idx: idx, status: fexception_{type(e).__name__}, latency: latency, retries: -1, }) async def main(): sem asyncio.Semaphore(STRESS[concurrency]) results [] async with httpx.AsyncClient() as client: tasks [ one_request(sem, client, i, results) for i in range(STRESS[total_requests]) ] start time.perf_counter() await asyncio.gather(*tasks) total_time time.perf_counter() - start ok [r for r in results if r[status] ok] fail [r for r in results if r[status] ! ok] latencies sorted([r[latency] for r in ok]) print(f总请求: {len(results)}) print(f成功: {len(ok)} 失败: {len(fail)}) print(f总耗时: {total_time:.2f}s) print(fQPS: {len(results) / total_time:.2f}) if latencies: print(fP50: {statistics.median(latencies):.0f}ms) print(fP95: {latencies[int(len(latencies) * 0.95)]:.0f}ms) print(fP99: {latencies[int(len(latencies) * 0.99)]:.0f}ms) retry_total sum(r[retries] for r in ok if r[retries] 0) print(f发生重试的请求数: {retry_total}) if fail: from collections import Counter print(失败分布:, Counter(r[status] for r in fail)) if __name__ __main__: asyncio.run(main())压测参数在 config.toml 里改concurrency 20表示同时 20 个请求在飞total_requests 200表示总共打 200 个。这两个数可以根据你的实际场景调整但建议先从 10 并发、100 请求开始观察服务端日志和耗时分布再逐步加压。4. 三步验证起服务、跑压测、核对日志4.1 第一步本地起服务安装依赖pip install fastapi uvicorn httpx pydantic启动微服务uvicorn service:app --host 0.0.0.0 --port 8000 --log-level info启动之后先手动打一发确认链路通curl -X POST http://127.0.0.1:8000/generate \ -H Content-Type: application/json \ -d {prompt:用一句话解释什么是幂等性。,max_tokens:64,temperature:0.2}正常返回应该类似{ content: 幂等性是指同一个操作执行一次和执行多次对系统状态产生的影响相同。, latency_ms: 1240.5, retries: 0, status: ok }如果这里就报 401检查环境变量TAOTOKEN_API_KEY是否设置正确。如果报连接错误检查base_url是否写成了https://taotoken.net/api不要多加/v1因为代码里已经拼了/v1/chat/completions。4.2 第二步跑压测脚本python stress.py一次典型的输出总请求: 200 成功: 198 失败: 2 总耗时: 18.43s QPS: 10.85 P50: 1420ms P95: 2870ms P99: 4120ms 发生重试的请求数: 6 失败分布: Counter({http_502: 2})这组数据说明几件事。P50 在 1.4 秒左右对于短 prompt 的 Kimi K3 调用来说是正常范围。P95 跳到 2.8 秒说明有部分请求遇到了排队或重试。P99 到 4.1 秒大概率是重试叠加导致的。6 个请求发生了重试最终 2 个失败返回 502说明重试 3 次之后仍然没成功可能是遇到了持续限流或服务端抖动。4.3 第三步核对日志与耗时服务端日志里能看到每次重试的间隔和错误码。如果你在call_kimi里加了日志输出会看到类似[重试] 429 限流2秒后重试 (第1次) [重试] 429 限流4秒后重试 (第2次) [成功] 第3次尝试成功这里的关键观察点是429 出现之后指数退避是否生效以及重试之后是否成功。如果大量请求都在重试 429说明并发数超过了当前 Key 的速率限制需要降低concurrency或者申请更高的配额。另一个要核对的是耗时分布。如果 P99 远高于 P95说明有长尾请求可能是网络抖动或服务端排队。如果 P50 和 P95 差距不大说明服务端响应比较稳定。5. 本篇常见错排查5.1 401 无效的 API Key最常见的原因是环境变量没设置或者设置在了错误的 shell 会话里。export只在当前会话有效换终端就没了。建议写进.bashrc或.zshrc或者用.env文件配合python-dotenv加载。另一个原因是 Key 复制的时候带了空格检查一下首尾字符。5.2 429 限流压测场景下 429 几乎必然出现关键是看重试之后能不能恢复。如果重试 3 次仍然 429说明并发数确实超了。处理方式有三种降低concurrency、增大retry_backoff_base、或者在 TaoToken 控制台查看当前 Key 的速率限制并申请调整。不要无脑加大重试次数因为限流窗口没过去的话重试只是浪费配额。5.3 超时设置不合理timeout_seconds 30对短 prompt 够用但如果你压测的是长文本场景30 秒可能不够。Kimi K3 处理长文本时首 token 延迟会明显增加建议长文本压测把超时设到 60 秒以上。另外注意 httpx 的 timeout 是总超时不是首字节超时如果你需要更细粒度的控制可以拆成connect、read、write分别设置。5.4 JSON 解析失败如果 Kimi K3 返回的内容不是合法 JSONresp.json()会抛异常。在压测脚本里我用了resp.json()直接解析如果服务端返回了非 JSON 的错误页会走到 exception 分支。更稳妥的做法是先用resp.text拿到原始内容再尝试json.loads失败时记录原始文本便于排查。5.5 连接池耗尽压测脚本里每次请求都新建httpx.AsyncClient在 20 并发下问题不大但如果并发上到 100 以上可能会遇到连接池耗尽或文件描述符不够的问题。生产环境建议复用客户端用httpx.AsyncClient的上下文管理器在应用启动时创建、关闭时释放。压测脚本里也可以改成全局复用一个 client减少握手开销。5.6 重试导致重复计费这是最需要注意的一点。如果你的重试逻辑没有做幂等处理同一个请求重试 3 次可能会产生 3 次计费。Kimi K3 的 API 本身不提供幂等键所以重试的代价需要你自己评估。在压测场景下重试是为了测稳定性但在生产环境建议对重试次数和重试条件做严格限制比如只对 5xx 重试不对 429 重试或者对 429 重试但设置更长的退避时间。6. 接入建议与下一步压测跑完之后你应该对 Kimi K3 在你当前并发下的表现有了一个量化认知。如果 P95 和 P99 在可接受范围内错误率低于 1%那就可以考虑接入生产。如果 429 频繁出现先去 API Keys 页面 看看当前 Key 的配额或者调整压测参数找到稳定的并发上限。对于长期跑编码任务或 Agent 工作流的场景单次调用的稳定性比峰值 QPS 更重要可以考虑用 Coding Plan 来管理调用配额和模型切换。如果你只是想先手动验证一下 Kimi K3 在具体 prompt 上的表现可以直接在 模型对话 里试几轮确认输出质量符合预期之后再写压测代码。接入文档里有完整的错误码列表和请求示例压测之前过一遍能省不少排查时间。整套代码你可以在本地跑通之后把base_url和model换成其他模型对比同一套压测脚本下的表现差异这也是统一 Key 接入层的一个实际好处。
1. 为什么我要用 Python 微服务压测 Kimi K3 的 APIKimi K3 的长文本能力已经被聊得很多了20 万字上下文、跨段落信息检索、技术报告摘要这些确实是它的看家本领。但如果你打算把它接进生产系统真正决定能不能上线的往往不是“它能不能读懂长文”而是“它在高并发下会不会超时”“错误码返回得规不规范”“重试之后会不会重复计费”。这些问题光靠对话窗口里聊几句是测不出来的。我这次做的事情很具体写一个 Python 微服务把 Kimi K3 的 API 包在里面然后用压测脚本打它观察并发调用、超时重试、错误码处理这三件事的真实表现。压测载体选 Python 是因为它写起来快、改起来也快微服务框架用 FastAPI异步 HTTP 客户端用 httpx压测脚本用 asyncio 直接怼并发。整套东西不需要很重的依赖本地就能跑起来。适合谁看如果你正在评估 Kimi K3 能不能进你的技术栈或者你已经决定要用但不确定接入层该怎么写这篇文章里的配置骨架和压测脚本可以直接复制去改。如果你只是想了解 Kimi K3 的 API 工程表现不看代码看结论也有参考价值。整篇的节奏是先讲清楚问题场景再给可复制的配置然后跑验证最后把踩过的坑列出来。2. TaoToken 前置统一 Key 与 config.toml 骨架在写压测代码之前先把 Key 管理和配置骨架定下来。我试过把 Key 硬编码在代码里后来换环境的时候改得想哭所以这次直接用 config.toml 把配置抽出来Key 走环境变量注入。TaoToken 在这里的角色是统一接入层。你可以在它的控制台里创建一个 Key然后这个 Key 既能调 Kimi K3也能调其他模型切换模型只需要改 config.toml 里的 model 字段不用换 Key、不用换 base_url。对于压测场景来说这很实用因为你可以用同一套压测脚本对比不同模型在相同并发下的表现。先看 config.toml 的骨架[app] name kimi-k3-stress host 0.0.0.0 port 8000 log_level info [llm] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model kimi-k3 timeout_seconds 30 max_retries 3 retry_backoff_base 2 [stress] concurrency 20 total_requests 200 prompt 请用一句话解释什么是微服务架构中的服务熔断。 max_tokens 128 temperature 0.2几个关键点说明一下。base_url填https://taotoken.net/api这是 API 入口不要加多余的路径。api_key_env指定从哪个环境变量读 Key这样配置文件可以进版本库Key 不会泄露。timeout_seconds设 30 秒因为 Kimi K3 在长文本场景下响应时间会比短 prompt 长但压测用的是短 prompt30 秒足够覆盖绝大多数情况。max_retries设 3配合retry_backoff_base 2做指数退避。环境变量这样设置export TAOTOKEN_API_KEY你的KeyKey 的获取路径是 TaoToken 控制台的 API Keys 页面创建之后复制出来即可。如果你还没创建可以去 API Keys 管理页 建一个。接入文档在 这里里面有完整的请求格式和错误码说明压测之前建议先扫一眼。3. 可复制配置FastAPI 微服务 异步压测脚本3.1 微服务端封装 Kimi K3 调用微服务的职责很简单接收一个 prompt转发给 Kimi K3返回结果和耗时。但为了压测能观察到重试和错误码需要在调用层把 httpx 的超时、重试、异常处理都写清楚。# service.py import os import time import tomllib import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel with open(config.toml, rb) as f: config tomllib.load(f) app FastAPI(titleKimi K3 Stress Target) class PromptRequest(BaseModel): prompt: str max_tokens: int 128 temperature: float 0.2 class PromptResponse(BaseModel): content: str latency_ms: float retries: int status: str def get_api_key() - str: key os.environ.get(config[llm][api_key_env]) if not key: raise RuntimeError(f环境变量 {config[llm][api_key_env]} 未设置) return key async def call_kimi(prompt: str, max_tokens: int, temperature: float): url f{config[llm][base_url]}/v1/chat/completions headers { Authorization: fBearer {get_api_key()}, Content-Type: application/json, } payload { model: config[llm][model], messages: [{role: user, content: prompt}], max_tokens: max_tokens, temperature: temperature, } timeout config[llm][timeout_seconds] max_retries config[llm][max_retries] backoff_base config[llm][retry_backoff_base] last_error None for attempt in range(max_retries): try: async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post(url, headersheaders, jsonpayload) if resp.status_code 200: data resp.json() content data[choices][0][message][content] return content, attempt elif resp.status_code 429: last_error f429 限流: {resp.text[:200]} elif resp.status_code 401: raise HTTPException(status_code401, detailKey 无效) elif 500 resp.status_code 600: last_error f{resp.status_code} 服务端错误 else: last_error f{resp.status_code}: {resp.text[:200]} except httpx.TimeoutException: last_error 请求超时 except httpx.ConnectError as e: last_error f连接失败: {e} if attempt max_retries - 1: wait backoff_base ** attempt await asyncio.sleep(wait) raise HTTPException(status_code502, detailf重试耗尽: {last_error}) app.post(/generate, response_modelPromptResponse) async def generate(req: PromptRequest): start time.perf_counter() content, retries await call_kimi(req.prompt, req.max_tokens, req.temperature) latency (time.perf_counter() - start) * 1000 return PromptResponse( contentcontent, latency_msround(latency, 2), retriesretries, statusok, )这段代码里有几个设计决策值得说。第一重试逻辑放在服务端而不是压测脚本里因为生产环境的重试通常是在接入层做的压测要模拟真实链路。第二429 和 5xx 走重试401 直接抛出不重试因为 Key 无效重试多少次都没用。第三每次重试都新建httpx.AsyncClient这在压测场景下会有额外开销但能避免连接池状态污染生产环境可以改成复用客户端。3.2 压测脚本asyncio 并发打点压测脚本用 asyncio 的 Semaphore 控制并发数每个请求记录耗时和重试次数最后汇总 P50、P95、P99 和错误分布。# stress.py import asyncio import time import tomllib import httpx import statistics with open(config.toml, rb) as f: config tomllib.load(f) STRESS config[stress] TARGET fhttp://127.0.0.1:{config[app][port]}/generate async def one_request(sem, client, idx, results): async with sem: payload { prompt: STRESS[prompt], max_tokens: STRESS[max_tokens], temperature: STRESS[temperature], } start time.perf_counter() try: resp await client.post(TARGET, jsonpayload, timeout60) latency (time.perf_counter() - start) * 1000 if resp.status_code 200: data resp.json() results.append({ idx: idx, status: ok, latency: latency, retries: data.get(retries, 0), }) else: results.append({ idx: idx, status: fhttp_{resp.status_code}, latency: latency, retries: -1, }) except Exception as e: latency (time.perf_counter() - start) * 1000 results.append({ idx: idx, status: fexception_{type(e).__name__}, latency: latency, retries: -1, }) async def main(): sem asyncio.Semaphore(STRESS[concurrency]) results [] async with httpx.AsyncClient() as client: tasks [ one_request(sem, client, i, results) for i in range(STRESS[total_requests]) ] start time.perf_counter() await asyncio.gather(*tasks) total_time time.perf_counter() - start ok [r for r in results if r[status] ok] fail [r for r in results if r[status] ! ok] latencies sorted([r[latency] for r in ok]) print(f总请求: {len(results)}) print(f成功: {len(ok)} 失败: {len(fail)}) print(f总耗时: {total_time:.2f}s) print(fQPS: {len(results) / total_time:.2f}) if latencies: print(fP50: {statistics.median(latencies):.0f}ms) print(fP95: {latencies[int(len(latencies) * 0.95)]:.0f}ms) print(fP99: {latencies[int(len(latencies) * 0.99)]:.0f}ms) retry_total sum(r[retries] for r in ok if r[retries] 0) print(f发生重试的请求数: {retry_total}) if fail: from collections import Counter print(失败分布:, Counter(r[status] for r in fail)) if __name__ __main__: asyncio.run(main())压测参数在 config.toml 里改concurrency 20表示同时 20 个请求在飞total_requests 200表示总共打 200 个。这两个数可以根据你的实际场景调整但建议先从 10 并发、100 请求开始观察服务端日志和耗时分布再逐步加压。4. 三步验证起服务、跑压测、核对日志4.1 第一步本地起服务安装依赖pip install fastapi uvicorn httpx pydantic启动微服务uvicorn service:app --host 0.0.0.0 --port 8000 --log-level info启动之后先手动打一发确认链路通curl -X POST http://127.0.0.1:8000/generate \ -H Content-Type: application/json \ -d {prompt:用一句话解释什么是幂等性。,max_tokens:64,temperature:0.2}正常返回应该类似{ content: 幂等性是指同一个操作执行一次和执行多次对系统状态产生的影响相同。, latency_ms: 1240.5, retries: 0, status: ok }如果这里就报 401检查环境变量TAOTOKEN_API_KEY是否设置正确。如果报连接错误检查base_url是否写成了https://taotoken.net/api不要多加/v1因为代码里已经拼了/v1/chat/completions。4.2 第二步跑压测脚本python stress.py一次典型的输出总请求: 200 成功: 198 失败: 2 总耗时: 18.43s QPS: 10.85 P50: 1420ms P95: 2870ms P99: 4120ms 发生重试的请求数: 6 失败分布: Counter({http_502: 2})这组数据说明几件事。P50 在 1.4 秒左右对于短 prompt 的 Kimi K3 调用来说是正常范围。P95 跳到 2.8 秒说明有部分请求遇到了排队或重试。P99 到 4.1 秒大概率是重试叠加导致的。6 个请求发生了重试最终 2 个失败返回 502说明重试 3 次之后仍然没成功可能是遇到了持续限流或服务端抖动。4.3 第三步核对日志与耗时服务端日志里能看到每次重试的间隔和错误码。如果你在call_kimi里加了日志输出会看到类似[重试] 429 限流2秒后重试 (第1次) [重试] 429 限流4秒后重试 (第2次) [成功] 第3次尝试成功这里的关键观察点是429 出现之后指数退避是否生效以及重试之后是否成功。如果大量请求都在重试 429说明并发数超过了当前 Key 的速率限制需要降低concurrency或者申请更高的配额。另一个要核对的是耗时分布。如果 P99 远高于 P95说明有长尾请求可能是网络抖动或服务端排队。如果 P50 和 P95 差距不大说明服务端响应比较稳定。5. 本篇常见错排查5.1 401 无效的 API Key最常见的原因是环境变量没设置或者设置在了错误的 shell 会话里。export只在当前会话有效换终端就没了。建议写进.bashrc或.zshrc或者用.env文件配合python-dotenv加载。另一个原因是 Key 复制的时候带了空格检查一下首尾字符。5.2 429 限流压测场景下 429 几乎必然出现关键是看重试之后能不能恢复。如果重试 3 次仍然 429说明并发数确实超了。处理方式有三种降低concurrency、增大retry_backoff_base、或者在 TaoToken 控制台查看当前 Key 的速率限制并申请调整。不要无脑加大重试次数因为限流窗口没过去的话重试只是浪费配额。5.3 超时设置不合理timeout_seconds 30对短 prompt 够用但如果你压测的是长文本场景30 秒可能不够。Kimi K3 处理长文本时首 token 延迟会明显增加建议长文本压测把超时设到 60 秒以上。另外注意 httpx 的 timeout 是总超时不是首字节超时如果你需要更细粒度的控制可以拆成connect、read、write分别设置。5.4 JSON 解析失败如果 Kimi K3 返回的内容不是合法 JSONresp.json()会抛异常。在压测脚本里我用了resp.json()直接解析如果服务端返回了非 JSON 的错误页会走到 exception 分支。更稳妥的做法是先用resp.text拿到原始内容再尝试json.loads失败时记录原始文本便于排查。5.5 连接池耗尽压测脚本里每次请求都新建httpx.AsyncClient在 20 并发下问题不大但如果并发上到 100 以上可能会遇到连接池耗尽或文件描述符不够的问题。生产环境建议复用客户端用httpx.AsyncClient的上下文管理器在应用启动时创建、关闭时释放。压测脚本里也可以改成全局复用一个 client减少握手开销。5.6 重试导致重复计费这是最需要注意的一点。如果你的重试逻辑没有做幂等处理同一个请求重试 3 次可能会产生 3 次计费。Kimi K3 的 API 本身不提供幂等键所以重试的代价需要你自己评估。在压测场景下重试是为了测稳定性但在生产环境建议对重试次数和重试条件做严格限制比如只对 5xx 重试不对 429 重试或者对 429 重试但设置更长的退避时间。6. 接入建议与下一步压测跑完之后你应该对 Kimi K3 在你当前并发下的表现有了一个量化认知。如果 P95 和 P99 在可接受范围内错误率低于 1%那就可以考虑接入生产。如果 429 频繁出现先去 API Keys 页面 看看当前 Key 的配额或者调整压测参数找到稳定的并发上限。对于长期跑编码任务或 Agent 工作流的场景单次调用的稳定性比峰值 QPS 更重要可以考虑用 Coding Plan 来管理调用配额和模型切换。如果你只是想先手动验证一下 Kimi K3 在具体 prompt 上的表现可以直接在 模型对话 里试几轮确认输出质量符合预期之后再写压测代码。接入文档里有完整的错误码列表和请求示例压测之前过一遍能省不少排查时间。整套代码你可以在本地跑通之后把base_url和model换成其他模型对比同一套压测脚本下的表现差异这也是统一 Key 接入层的一个实际好处。
