企业官网建设流程全解析

当一批脚本在几秒内把大量格式异常的文件推给上游接口时，网关看到的不是正常业务流量，而是典型的探测流量：请求数密集、payload 小而畸形、失败后立刻重试，最终把并发压力全部转嫁到限流层。这类场景最近被反复讨论，某智能体批量向 Hugging Face 账户投递格式异常文件的公开报道，本质上就是"自动化脚本 + 高并发 + 无退避重试"的放大版。如果你也在维护类似的探测、巡检或回放任务，真正要解决的不是"要不要并发"，而是"并发上去之后 Token 怎么记账、限流怎么接住"。TaoToken（https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency ）在这一层的价值很直接：它把请求统一收口到 OpenAI 兼容的 Base URL，让并发控制、重试退避和用量统计都能落在你自己的代码里，而不是散落在十几个不同的供应商 SDK 中。下面这篇内容不做行业评论，只讲可复制的接入与排障：并发配置片段怎么写、限流日志长什么样、用量对照表怎么回填，以及 Claude Code、Codex、CC Switch 三件套如何改到同一条链路上。

1. 探测式并发的三个特征，决定了它一定会先撞限流

先把问题定义清楚。探测类任务的流量特征和普通聊天应用完全不同：

第一，请求密度高但单请求极短。一个文件探测脚本可能在 10 秒内发出几百个请求，每个请求的 prompt 只有几十个 token，输出也只要几个 token。单请求成本低，但请求数（RPM）会瞬间打满。

第二，失败重试是同步放大。很多脚本在except里直接continue或者无条件重试，一旦上游返回 429，重试线程和原始线程同时存在，实际并发是名义并发的 2 到 3 倍。

第三，payload 形态异常。格式异常的文件、超长文件名、非 UTF-8 内容，这些在网关侧看起来高度可疑，容易触发风控而不是单纯的配额限流。风控限流通常不给Retry-After，直接掐连接，这比 429 更难排查。

所以并发控制必须在客户端做两层：一层是并发闸门（同时最多几个在飞），另一层是速率闸门（每分钟最多几个请求、每分钟最多消耗多少 token）。只做第一层，遇到慢响应仍然会堆积；只做第二层，遇到突发仍然会瞬间超速。

在动手改脚本之前，先确认你的 Key 和 Base URL 是同一套。从 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency 的控制台拿到 Key 后，把请求地址统一指向https://taotoken.net/api，后续所有并发参数、重试策略、日志字段都在这个统一入口上生效，不需要为不同模型改不同的 SDK 初始化逻辑。

2. 最小可跑通：Key、Base URL 与一次请求验证

在改并发之前，先用一条命令确认链路是通的。不要跳过这一步，很多"限流"其实是 Key 写错或者路径拼错导致的 401/404，被误判成 429。

# 环境变量先固定下来，脚本和 CLI 工具共用同一套 export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL="你控制台里可用的模型 ID" # 非流式最小请求，用来确认鉴权和路径 curl -sS "$TAOTOKEN_BASE_URL/v1/chat/completions" \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "'"$TAOTOKEN_MODEL"'", "messages": [{"role": "user", "content": "只回复 ok"}], "max_tokens": 8, "stream": false }'

返回体里会带usage字段，包含prompt_tokens、completion_tokens、total_tokens。这三个数字是你后面做用量对照的唯一依据，务必在脚本里把它们记下来，而不是只看请求是否成功。

流式请求单独测一次，因为流式场景下限流表现不一样：连接建立成功后才开始计费，中途被限流时你可能已经拿到部分输出。可以用：

curl -N -sS "$TAOTOKEN_BASE_URL/v1/chat/completions" \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "'"$TAOTOKEN_MODEL"'", "messages": [{"role": "user", "content": "数到三"}], "max_tokens": 32, "stream": true }'

如果这一步报 401，检查 Key 是否带上了多余的引号或换行；报 404，检查 Base URL 后面有没有多写/v1之外的路径。确认无误后再进入并发改造。

3. 并发配置片段：信号量 + 速率闸门 + 退避重试

下面是可直接粘贴运行的异步脚本骨架。它做了四件事：限制同时在飞的请求数、限制单位时间内的请求数、按Retry-After退避、把每次请求的 token 用量打成结构化日志。

import asyncio import json import os import random import time from collections import deque import httpx BASE_URL = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") API_KEY = os.environ["TAOTOKEN_API_KEY"] MODEL = os.environ.get("TAOTOKEN_MODEL", "你的模型 ID") # ---- 并发与限流参数，按实测调整 ---- MAX_CONCURRENCY = 8 # 同时在飞的请求数上限 MAX_RPM = 120 # 每分钟请求数上限，先设保守值 MAX_RETRIES = 5 # 429/5xx 最大重试次数 BACKOFF_BASE = 1.5 # 退避基数（秒） BACKOFF_CAP = 30.0 # 退避上限（秒） REQUEST_TIMEOUT = 60.0 # 单请求超时（秒） _sem = asyncio.Semaphore(MAX_CONCURRENCY) _call_times = deque() # 记录请求时间戳，用于 RPM 闸门 def log_event(event: str, **fields) -> None: """统一的结构化日志，方便后面用 jq 聚合。""" record = {"ts": round(time.time(), 3), "event": event, **fields} print(json.dumps(record, ensure_ascii=False), flush=True) async def rpm_gate() -> None: """滚动窗口限速：保证最近 60 秒内的请求数不超过 MAX_RPM。""" while True: now = time.monotonic() while _call_times and now - _call_times[0] > 60: _call_times.popleft() if len(_call_times) < MAX_RPM: _call_times.append(now) return sleep_for = 60 - (now - _call_times[0]) + 0.05 await asyncio.sleep(max(sleep_for, 0.05)) def parse_retry_after(resp: httpx.Response) -> float: """优先读 Retry-After，没有就按退避基数指数增长并加抖动。""" raw = resp.headers.get("retry-after") if raw: try: return min(float(raw), BACKOFF_CAP) except ValueError: pass return 0.0 async def call_model(client: httpx.AsyncClient, trace_id: str, prompt: str) -> dict: payload = { "model": MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": 64, "stream": False, } headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } for attempt in range(1, MAX_RETRIES + 1): await rpm_gate() async with _sem: started = time.perf_counter() try: resp = await client.post( "/v1/chat/completions", json=payload, headers=headers, timeout=REQUEST_TIMEOUT, ) latency_ms = round((time.perf_counter() - started) * 1000, 1) if resp.status_code == 200: body = resp.json() usage = body.get("usage", {}) or {} log_event( "request_done", trace_id=trace_id, attempt=attempt, status=200, latency_ms=latency_ms, prompt_tokens=usage.get("prompt_tokens"), completion_tokens=usage.get("completion_tokens"), total_tokens=usage.get("total_tokens"), concurrency=MAX_CONCURRENCY, ) return body if resp.status_code in (429, 500, 502, 503, 504): wait = parse_retry_after(resp) if wait <= 0: wait = min(BACKOFF_BASE ** attempt, BACKOFF_CAP) wait += random.uniform(0, 0.4) # 抖动，避免重试风暴同频 log_event( "retry_scheduled", trace_id=trace_id, attempt=attempt, status=resp.status_code, wait_s=round(wait, 2), concurrency=MAX_CONCURRENCY, ) await asyncio.sleep(wait) continue # 4xx（非 429）通常是请求本身有问题，重试无意义 log_event( "request_failed", trace_id=trace_id, attempt=attempt, status=resp.status_code, body=resp.text[:300], ) return {"error": resp.status_code} except (httpx.TimeoutException, httpx.TransportError) as exc: wait = min(BACKOFF_BASE ** attempt, BACKOFF_CAP) + random.uniform(0, 0.4) log_event( "transport_error", trace_id=trace_id, attempt=attempt, error=type(exc).__name__, wait_s=round(wait, 2), ) await asyncio.sleep(wait) log_event("give_up", trace_id=trace_id, attempts=MAX_RETRIES) return {"error": "exhausted"} async def main() -> None: prompts = [f"探测样本 {i}，请只回复 ok" for i in range(200)] limits = httpx.Limits(max_connections=MAX_CONCURRENCY * 2, max_keepalive_connections=MAX_CONCURRENCY) async with httpx.AsyncClient(base_url=BASE_URL, limits=limits) as client: tasks = [ call_model(client, trace_id=f"probe-{i:04d}", prompt=p) for i, p in enumerate(prompts) ] await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(main())

几个参数的意义必须搞清楚，否则调参会变成玄学：

• MAX_CONCURRENCY控制"同时在飞"，直接决定峰值压力。先设 4 到 8，观察日志里的 429 数量再往上抬。
• MAX_RPM控制"每分钟请求数"，和并发数不是一回事。低并发 + 高 RPM 依然会触发请求数限流。
• MAX_RETRIES配合退避，本质上是在"补给速度"和"压力放大"之间取平衡。重试次数设太高，遇到持续限流时会把并发放大数倍。
• 抖动的存在是为了防止所有协程在同一毫秒重试，形成新的尖峰。

另外要注意httpx.Limits里的连接池上限要和信号量匹配。如果连接池只有 8 个连接而信号量是 16，多出来的协程会在连接池排队，你的"限流日志"里会看到大量latency_ms异常高的成功请求，容易被误判成上游变慢。

4. 限流日志：字段设计、采集与判定

并发改造做完之后，真正的产出物是日志。没有日志，你无法回答"到底是 RPM 超了还是 TPM 超了""重试放大了几倍"这两个问题。

建议的日志字段固定为下面这一组，不要随意加自由文本：

{"ts":1730000000.123,"event":"request_done","trace_id":"probe-0007","attempt":1,"status":200,"latency_ms":812.4,"prompt_tokens":128,"completion_tokens":64,"total_tokens":192,"concurrency":8} {"ts":1730000001.451,"event":"retry_scheduled","trace_id":"probe-0011","attempt":2,"status":429,"wait_s":3.12,"concurrency":8} {"ts":1730000002.907,"event":"retry_scheduled","trace_id":"probe-0019","attempt":1,"status":503,"wait_s":1.94,"concurrency":8} {"ts":1730000004.220,"event":"request_failed","trace_id":"probe-0023","attempt":1,"status":400,"body":"invalid request payload"}

判定逻辑要写死：

• status=429且带retry-after：属于配额类限流，降并发或降 RPM 后通常能恢复。
• status=429但不带retry-after：更可能是风控类拦截，此时降并发没用，要检查 payload 形态和请求特征是否过于异常。
• status=503/504：上游过载，退避重试有效，但要在日志里和 429 分开统计。
• status=400：请求本身有问题，重试是浪费配额，直接修 payload。

跑一轮之后，把日志导出成本地文件，用命令聚合。这些命令都在你本地执行，不涉及任何远端数据库：

# 按事件类型统计数量 jq -r '.event' run.log | sort | uniq -c | sort -rn # 统计每个并发档位下的总 token 和请求数 jq -r 'select(.event=="request_done") | [.concurrency, .total_tokens] | @tsv' run.log \ | awk '{sum[$1]+=$2; cnt[$1]++} END {for (c in sum) printf "并发=%s 请求=%d 总tokens=%d 均值=%.1f\n", c, cnt[c], sum[c], sum[c]/cnt[c]}' # 统计 429 占比 jq -r 'select(.event=="retry_scheduled" and .status==429) | .trace_id' run.log | wc -l # 提取所有退避等待时长，判断是否出现了长尾 jq -r 'select(.event=="retry_scheduled") | .wait_s' run.log | sort -n | tail -5

如果你在脚本里同时记录了latency_ms，还可以算出 P95：

jq -r 'select(.event=="request_done") | .latency_ms' run.log \ | sort -n \ | awk '{a[NR]=$1} END {printf "P50=%.0fms P95=%.0fms 样本=%d\n", a[int(NR*0.5)], a[int(NR*0.95)], NR}'

有了这些聚合结果，"限流日志"就不是一堆散乱的行，而是一份能直接支撑调参的证据。

5. 用量对照：把并发档位跑成一张可回填的表

用量对照的关键不是去追某个"最优并发数"，而是找到你自己任务形态下的拐点：从哪一档开始，429 开始明显上升，而吞吐不再增长。

测试方法要固定三件事：同一份 payload 集合、同一模型、同一时间段长度。然后逐档提升MAX_CONCURRENCY（例如 4、8、16、32），每档跑 5 到 10 分钟，把日志分别落盘。

下表是字段模板，数值请用你自己的日志回填，不要照抄任何外部数字：

几个读表要点：

看有效吞吐而不是看名义并发。如果并发从 16 提到 32，成功请求数没变但 429 翻倍，说明已经进入"重试吃掉配额"的阶段，此时应该降回 16，而不是继续加机器。

看 token 均值而不是总量。如果某一档的总tokens/成功请求数明显变大，说明输出长度在膨胀（可能是重试请求被计费、或者模型返回变长），这时候成本上升和并发无关。

看 P95 而不是平均值。平均值掩盖长尾。限流一旦发生，P95 会先动，平均值可能还很好看。

区分 prompt 与 completion。prompt tokens 主要来自你的输入（探测脚本里的文件内容、元数据），completion tokens 来自模型输出。如果 prompt 占比极高，优化方向是压缩输入而不是降并发。

把这四档跑完，你就能给出一个明确结论：在这个任务形态下，安全并发档位是多少，对应的每分钟 token 预算是多少。这个结论比任何"经验值"都可靠。

6. 客户端侧三件套：Claude Code、Codex 与 CC Switch 的配置

除了自己写脚本，很多人的探测任务是通过 Claude Code 或 Codex 这类 CLI 触发的。如果你希望这些工具也走同一条入口，需要分别改配置，注意两边的环境变量体系完全不同，不要混用。

Claude Code：改~/.claude/settings.json，使用ANTHROPIC_*前缀。

{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "你的模型 ID", "ANTHROPIC_SMALL_FAST_MODEL": "你的小模型 ID", "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1" } }

写完保存，重开终端让配置生效。如果仍然走旧链路，检查是否有 shell 里 export 的ANTHROPIC_BASE_URL覆盖了文件配置——环境变量优先级通常高于配置文件。

Codex：改~/.codex/config.toml，用model_providers段，不要用ANTHROPIC_*。

model = "你的模型 ID" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"

对应的环境变量是TAOTOKEN_API_KEY，值填YOUR_API_KEY。Codex 读取的是env_key指定的那个变量名，把ANTHROPIC_AUTH_TOKEN塞进来是不会生效的，这是最容易踩的坑。

CC Switch 三件套：保持三处一致，只切不混。

三件套指的是：Claude Code 配置文件、Codex 配置文件、以及 CC Switch 自身维护的供应商条目。切换时最容易出问题的是三处没对齐——Claude Code 已经从旧供应商切到新入口，Codex 那侧还留着旧配置，结果同一个终端窗口里两个工具打到两个不同的后端，日志对不上账。

一个稳妥的做法是，在切换完成后立刻做一次一致性校验：

# 本地校验：确认 CLI 实际使用的入口与 Key 是否指向同一套 grep -R "base_url\|BASE_URL" ~/.claude/settings.json ~/.codex/config.toml 2>/dev/null echo "env: ${TAOTOKEN_API_KEY:0:6}... ${ANTHROPIC_BASE_URL:-未设置}"

只要输出的入口都是https://taotoken.net/api，且 Key 前缀一致，就说明三件套已经对齐。

关于工具链的完整配置说明，可以参考 Claude Code 文档：https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency ，里面有环境变量和配置文件的对应关系，比反复试错省时间。

7. 常见报错与排查路径对照

把排障路径固化下来，能省掉大量来回猜测。

排查顺序建议固定为：先确认鉴权与路径（401/404），再确认限流类型（429 有无 Retry-After），最后才动并发参数。反过来做的话，你会在一个本来就配错的链路上反复调参，越调越乱。

8. 把并发控制变成可复用的工程习惯

回到最初那个场景：格式异常的文件被批量并发投递，暴露出来的从来不是"模型能力"问题，而是工程侧的三个缺口——没有并发闸门、没有退避策略、没有用量记账。补上这三个缺口之后，同一份探测脚本的稳定性会有明显变化。

落地顺序建议是这样：

1. 先把 Key 和 Base URL 统一到https://taotoken.net/api，用一次 curl 确认链路通。
2. 把脚本里的重试逻辑改成"读 Retry-After + 指数退避 + 抖动"，去掉无条件重试。
3. 加上结构化日志，保证每次请求都能落到trace_id、status、usage三个字段。
4. 跑 4/8/16/32 四档并发，回填上面的对照表，找到自己的拐点。
5. 把 CLI 工具（Claude Code、Codex、CC Switch）也对齐到同一条链路，避免多套配置互相污染。

做完这五步，你手里就有了一份可复现的并发配置、一份可聚合的限流日志和一张有数据的用量对照表。以后再遇到类似的批量任务，直接复用这套骨架即可。

如果你还没开始，建议先从最小验证走起：在模型对话页发一条请求确认 Key 可用，再看 Coding Plan 了解配额与并发档位的匹配关系，接着在控制台创建独立的 API Key 用于脚本（不要和交互式工具共用一个 Key，方便分开记账），最后按 Claude Code 文档把 CLI 侧配置补齐。

• 模型对话体验：https://taotoken.net/models/detail/chat?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency
• Coding Plan 说明：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency
• 创建 API Key：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency
• Claude Code 配置文档：https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency

官方入口与最新说明以 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=hf_probe_concurrency 为准，Base URL 固定为https://taotoken.net/api，Key 统一用YOUR_API_KEY占位替换即可。