Qoni 能承接多少并发，没有通用答案：同一个平台，输入长度、是否带工具调用、上游依赖的响应时间不同，能跑到的档位就不一样。上线前更稳的做法是先在自己的环境里做梯度并发压测，用失败率和耗时分位找出容量拐点，再把拐点之前的档位写进生产侧限流配置，而不是先上线再等报错。压测给出的不是精确容量，而是一个可以据此设阈值、再靠线上指标校准的起始区间。

建议用 1、2、4、8、16……逐档并发跑固定请求量，记录每档的 P50/P95 耗时、失败率和错误码分布，找出失败率开始偏离基线的那一档作为拐点。生产侧限流阈值先取拐点前一档，超限走有上限的排队或快速拒绝，并持续盯拒绝数、队列深度和 P95。压测前需确认压测机自身的 CPU、文件描述符和出网连接数不是瓶颈，否则拐点会偏早。

定义压测指标与观测点

指标的作用是让不同并发档的结果能横向比较。如果每档只记“感觉还行”，后面就没法判断拐点在哪。建议每跑完一档就输出一行固定字段的结果，用同一份字段名，方便直接贴进表格或追加到 CSV。

写一个可调并发的压测脚本骨架

先用单文件脚本，不引入具体 SDK，减少变量。下面这段是通用起点，换成你自己的 Qoni 调用地址、鉴权头和请求体即可运行。前置条件：一台能访问被测端点的机器，Python 3.8 以上，准备一份固定输入文件，同一份输入反复使用，避免把输入差异算成容量差异。

# bench.py 通用骨架：替换 `--url` / 鉴权头 / payload 后运行
import argparse, collections, json, time, urllib.error, urllib.request
from concurrent.futures import ThreadPoolExecutor

def call_once(url, payload, headers, timeout, retry):
    attempts = 0
    t0 = time.perf_counter()
    while True:
        attempts += 1
        req = urllib.request.Request(url, data=json.dumps(payload).encode('utf-8'),
                                     headers=headers, method='POST')
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r:
                r.read()
                code = str(r.status)
            break
        except urllib.error.HTTPError as e:
            e.read()
            code = str(e.code)
            if not (e.code >= 500 and attempts <= retry):
                break
        except Exception as e:
            code = 'EXC_' + type(e).__name__
            if attempts > retry:
                break
    return code, (time.perf_counter() - t0) * 1000, attempts

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('`--url`', required=True)           # 替换：Qoni 调用地址
    ap.add_argument('`--payload-file`', required=True)  # 替换：固定输入 JSON
    ap.add_argument('`--concurrency`', type=int, default=1)
    ap.add_argument('`--total`', type=int, default=300)
    ap.add_argument('`--timeout`', type=float, default=30)
    ap.add_argument('`--retry`', type=int, default=0)
    ap.add_argument('`--tag`', default='run')
    a = ap.parse_args()

    payload = json.load(open(a.payload_file, encoding='utf-8'))
    headers = {'Content-Type': 'application/json'}     # 替换：鉴权头

    with ThreadPoolExecutor(max_workers=a.concurrency) as pool:
        futs = [pool.submit(call_once, a.url, payload, headers, a.timeout, a.retry)
                for _ in range(a.total)]
        rows = [f.result() for f in futs]

    codes = collections.Counter(r[0] for r in rows)
    lat = sorted(r[1] for r in rows)
    def pct(p):
        return lat[min(len(lat) - 1, int(len(lat) * p))]
    ok = sum(v for k, v in codes.items() if k.startswith('2'))
    print(json.dumps({
        'tag': a.tag, 'concurrency': a.concurrency, 'total': a.total,
        'retry': a.retry,
        'p50_ms': round(pct(0.5), 1), 'p95_ms': round(pct(0.95), 1),
        'fail_rate': round(1 - ok / len(rows), 4),
        'codes': dict(codes),
        'attempts_avg': round(sum(r[2] for r in rows) / len(rows), 2),
    }, ensure_ascii=False))

if __name__ == '__main__':
    main()

• `--url` 与鉴权头：换成 Qoni 的实际调用地址和平台要求的认证字段。
• `--payload-file`：保存一份有代表性的真实输入；如果要覆盖多种输入形态，就每份输入单独跑一轮，不要混在同一轮里算失败率。
• `--total`：每档固定请求总数，各档样本量一致，失败率才可比。

按梯度提高并发并记录失败率拐点

从低并发逐档上调，一般按 1、2、4、8、16、32 这样的倍率，也可以按你预估的量级插档，每档跑固定请求总数并记录一行。低并发档是基线：失败率通常接近零，P95 比较平稳。随着并发上升，先变化的是耗时，P95 开始明显抬升，之后失败率才跟着上升。

concurrency,total,p50_ms,p95_ms,fail_rate,codes
1,300,...,...,0.000,{"200":300}
8,300,...,...,0.000,{"200":300}
16,300,...,...,0.003,{"200":299,"429":1}

判定拐点时有几点要提前处理：失败率的抬头线先定好，常见做法是设一条容忍线（例如 1%，按业务重要性调整），超过容忍线的那一档记为拐点档，前一阵列为当前可用档；同一档最好重复跑两到三次，结果波动大说明该档本身已不稳定，取值要更保守；同时确认压测机没到瓶颈，看它的 CPU、文件描述符上限和出网连接数，如果压测机先饱和，拐点位置会偏早。

对比开启与关闭重试时的结果差异

重试会把一部分失败转成后置的成功，让结果比实际乐观。所以在拐点附近，同一并发档跑两组，参数只改重试次数：一组 `--retry` 0，一组 `--retry` 2，其余完全相同。

• 失败率：开重试后通常更低，但这个数字不代表容量变大。
• 总耗时：开重试后 P95 和整体运行时间往往上升，因为失败请求被重复执行。
• 重复执行次数：看 attempts_avg，如果它明显大于 1，说明这一档已经在靠重试兜底。

判断方式：如果关重试时拐点出现在第 N 档，开重试时拐点被推到 N 档之后，生产侧就按关重试的结果设阈值，或者把重试次数明确计入容量预算，因为生产流量里的重试同样是真实成本。还要区分错误类型，429 和连接类错误通常可重试，参数错误一类的 4xx 重试没有意义。

依据拐点设定生产侧限流与排队策略

落成配置时核心是两个数：每个租户、API Key 或接口的在途并发上限，以及超限时的动作。阈值建议取拐点的前一档：假设 16 并发档失败率仍接近基线、32 并发档开始抬升，生产上限可以先设在 16 附近；业务对延迟敏感时再保守一档。QPS 上限可以由并发和单请求耗时换算，QPS 约等于并发除以平均耗时（秒），换算时用 P95 而不是 P50，避免低估。

# 生产侧限流与排队示意，字段名按你的网关实际定义替换
rate_limit:
  key: api_key            # 限流维度：租户 / API Key / 接口
  max_concurrency: 16     # 取压测拐点的前一档
  max_qps: 60             # 并发除以平均耗时(秒)，用 P95 换算更保守
  strategy: token_bucket
  burst: 8
queue:
  enabled: true
  max_depth: 200
  max_wait_ms: 2000       # 超过直接拒绝，避免请求堆积
overflow:
  action: reject_429
  retry_after_s: 2
  fallback: simplified_result   # 没有兜底内容时改为 reject

超限时不要无限排队：队列要有最大深度和最长等待时间，超过就快速失败并返回明确状态码，比如 429 加 Retry-After，让上游知道是暂时不可用而不是请求写错了。对 AI Agent 这类长耗时调用，还可以准备降级路径，返回简化结果或提示稍后重试，避免请求堆在队列里。

上线后需要持续盯的指标：被拒绝请求数、队列深度、排队等待时长、P95 与 P99 耗时、错误码分布，以及 Qoni 侧上游依赖的错误率。如果拒绝数长期为零而队列深度稳定偏高，说明阈值可能偏紧；如果 P95 在阈值附近持续抬升，说明实际容量低于压测时，需要下调上限。压测给出的是起点，最终阈值要靠这组线上指标反复校准。