一、接口通了，和真的用起来了，是两件事

去年下半年我们做了一件当时看起来非常正确的事：把散落在各个小组手里的十几个调用大模型的脚本收拢成一个内部服务，对外只留一个 HTTP 入口，鉴权、限流、日志、模型路由都放在这一层。上线那天大家还挺高兴，六条业务线陆陆续续接了进来，客服摘要、工单分类、文档问答、销售话术生成、代码评审辅助、内部知识检索，覆盖面看着很漂亮。

三个月后我们做季度回顾，做了一张很朴素的表：每条业务线这一周被调用了多少次、有多少个不同的人在用。结果有点难看。六条线里真正在跑的是两条，一条每天几万次调用，一条几千次；剩下四条，接口通了，Demo 演示过了，然后就停在那儿了，有的一周只有几十次调用，看用户 ID 全是当初做联调那两个人。

更让人不舒服的是，这个结论我们是靠人工翻日志得出来的，而不是系统主动告诉我们的。也就是说，如果那个季度回顾没人提，这件事可以再糊三个月。

后面的内容是我们那次复盘留下来的四块短板，以及我们后来补的东西。每一块我都尽量把当时怎么想的写出来，因为回过头看，判断错的地方往往不是技术能力，是懒得去拿真实数据。

这几块短板补完，我们才算真正把企业级 AI 工作提效这件事做扎实——不是接口通了就结束，而是几条线都稳定在用、并且用得怎么样随时看得见。现在针对ToB端的企业提效已经成为各个企业、工作团队都比较关注和急于去改变和升级的关键一环。说到用AI提效，最重要的就是获取到又便宜又足量的token资源，jiekou.vip 近期上线了企业资源包，也是面向这类多团队接入场景的，成本仅需按量使用的75%。

二、短板一：容量规模全靠拍脑袋，手上一条真实数据都没有

现象。 做规划的时候，我们需要估一个数：这套服务上线后大概会被调用多少次、单次请求会有多大。当时的做法是开会，让每条业务线的负责人报一个数。报上来的数字非常整齐：客服说"每天大概两万单，一单一次调用"，文档问答说"每天几千次吧"。我们把这些数加起来，乘 1.2，写进了容量规划文档。

我们当时怎么想的。 觉得业务方最了解自己的量，他们报的数应该最准。而且这件事看着不难，不至于为了一个估算专门去跑一轮实测。

实际原因。 业务方报的是业务量，不是请求量，这两个中间隔着一层我们自己写的逻辑。客服摘要那条线，一个会话最终会被切成 3 到 5 段分别调用，再加一次汇总，实际请求量是业务量的 4 倍多。反过来，文档问答那条线报了"每天几千次"，实际因为入口藏得深，一天只有一两百次真实调用。两个方向的偏差都不小，加总之后看着"差不多对"，其实是两个错误互相抵消。

还有一个更隐蔽的问题：我们只讨论了"平均多大"，没人问过"最大有多大"。而单次请求的字符数在真实数据里分布得非常偏，客服会话里有大量三五句话就结束的，也有客户贴了一整页报错日志的长会话，后者的耗时是前者的十几倍。用平均值做规划，等于假装那条长尾不存在。

后来怎么改的。 规矩改成一句话：任何容量数字，必须有一份小样本实测垫底。从每条业务线的真实流水里随机抽 200 条，用生产同样的 prompt 模板跑一遍，把输入字符数、输出字符数、单次耗时全部记下来，然后看 P50 和 P95，不看平均值。

"""
用真实业务样本实测单次调用的字符数与耗时分布。
运行前:
  export LLM_API_BASE=https://<your-endpoint>/v1
  export LLM_API_KEY=<your-key>
  export LLM_MODEL=<model-name>
  export SAMPLE_FILE=samples.jsonl   # 每行 {"prompt": "..."}
"""
import json
import math
import os
import time

import requests

API_BASE = os.environ["LLM_API_BASE"].rstrip("/")
API_KEY = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "qwen2.5-72b-instruct")
TIMEOUT = float(os.environ.get("LLM_TIMEOUT", "60"))

SESSION = requests.Session()
SESSION.headers.update({
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
})


def call_once(prompt: str) -> dict:
    payload = {
        "model": MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.2,
    }
    t0 = time.perf_counter()
    ok, text, err = True, "", None
    try:
        resp = SESSION.post(f"{API_BASE}/chat/completions",
                            json=payload, timeout=TIMEOUT)
        resp.raise_for_status()
        text = resp.json()["choices"][0]["message"]["content"]
    except Exception as exc:                     # 实测阶段先全部收下来再分类
        ok, err = False, f"{type(exc).__name__}: {exc}"
    return {
        "ok": ok,
        "latency": round(time.perf_counter() - t0, 3),
        "in_chars": len(prompt),
        "out_chars": len(text),
        "error": err,
    }


def percentile(values, p):
    """最近秩法。样本量只有几百条时，它比线性插值更好向别人解释。"""
    if not values:
        return 0.0
    xs = sorted(values)
    k = max(1, math.ceil(p / 100 * len(xs)))
    return xs[k - 1]


def load_samples(path):
    with open(path, encoding="utf-8") as f:
        return [json.loads(line)["prompt"] for line in f if line.strip()]


def main():
    samples = load_samples(os.environ.get("SAMPLE_FILE", "samples.jsonl"))
    records = [call_once(p) for p in samples]
    good = [r for r in records if r["ok"]]
    if not good:
        raise SystemExit("全部失败，先检查 API_BASE / API_KEY / 模型名")

    lat = [r["latency"] for r in good]
    inp = [r["in_chars"] for r in good]
    out = [r["out_chars"] for r in good]

    print(f"样本数         : {len(records)}")
    print(f"成功 / 失败率  : {len(good)} / {1 - len(good) / len(records):.1%}")
    print(f"输入字符 P50   : {percentile(inp, 50)}")
    print(f"输入字符 P95   : {percentile(inp, 95)}")
    print(f"输出字符 P50   : {percentile(out, 50)}")
    print(f"输出字符 P95   : {percentile(out, 95)}")
    print(f"耗时 P50 / P95 : {percentile(lat, 50):.2f}s / {percentile(lat, 95):.2f}s")
    print(f"耗时最慢一条   : {max(lat):.2f}s")


if __name__ == "__main__":
    main()

我们跑客服摘要那批样本，输出长这样：

样本数         : 200
成功 / 失败率  : 197 / 1.5%
输入字符 P50   : 1180
输入字符 P95   : 6420
输出字符 P50   : 214
输出字符 P95   : 386
耗时 P50 / P95 : 2.10s / 6.40s
耗时最慢一条   : 18.72s

P95 的输入字符数是 P50 的五倍多，最慢那条 18 秒。这两个数字是会上任何人拍脑袋都拍不出来的，而它们直接决定了后面的超时阈值和并发配置该怎么设。顺便说一句，那 1.5% 的失败率也是意外收获——我们原本以为是 0。

三、短板二：按最理想的情况排并发，一点波动余量都没留

现象。 服务上线后的第二周，客服那条线在一个周一上午集中出现超时，前端页面转圈，业务同学在群里问"是不是挂了"。看监控，服务本身没挂，是排队排上去了，队列深度一路涨到几百。

我们当时怎么想的。 规划的时候我们算过一版并发：日调用次数除以 86400 得到平均每秒请求量，再按单次耗时 2 秒去推在途请求数。数字很小，看着非常安全，所以并发上限设得很保守。

实际原因。 两个错误叠在一起。第一，把一天的量摊到 86400 秒，等于假设凌晨三点和上午十点的请求量一样多——真实情况是八成以上的调用集中在四个小时里，峰值大概是全天均值的三倍。第二，用 P50 耗时 2 秒去算在途请求数，而排队恰恰是被那些 6 秒、18 秒的长请求撑起来的，它们占着并发槽位不放。

后来怎么改的。 把并发估算写成一段代码，输入是上一步实测出来的 P95 耗时和业务线的日调用次数，输出是建议并发数，中间的假设全部显式写成参数，谁想调就改参数，不要在会上争论。用的就是最朴素的 Little's Law：在途请求数约等于到达率乘停留时间，然后乘一个余量。

"""
从实测数据推并发配置。所有假设都是显式参数，方便按业务线各自调整。
"""
import math

P95_LATENCY = 6.4          # 秒，来自上一步实测，不要用 P50
SUCCESS_TARGET = 0.995     # 目标成功率
RETRY_FACTOR = 1 / SUCCESS_TARGET   # 重试会额外放大请求量


def peak_qps_from_daily(daily_calls, busy_hours=4.0, peak_ratio=3.0):
    """日调用次数 -> 峰值每秒请求量。
    busy_hours: 请求主要集中在几个小时内
    peak_ratio: 峰值相对忙时均值的倍数
    """
    busy_avg_qps = daily_calls / (busy_hours * 3600)
    return busy_avg_qps * peak_ratio


def concurrency_for(peak_qps, p95_latency=P95_LATENCY, headroom=1.3):
    """在途请求数 ≈ 到达率 × 停留时间，再留一段波动余量。"""
    inflight = peak_qps * p95_latency * RETRY_FACTOR
    return math.ceil(inflight * headroom)


ROWS = [
    ("客服摘要", 42000),
    ("工单分类", 8600),
    ("文档问答", 3100),
]

print(f"{'业务线':<10}{'日调用':>8}{'峰值QPS':>10}{'建议并发':>10}")
total = 0
for name, daily in ROWS:
    qps = peak_qps_from_daily(daily)
    conc = concurrency_for(qps)
    total += conc
    print(f"{name:<10}{daily:>8}{qps:>10.2f}{conc:>10}")
print(f"{'合计':<10}{'':>8}{'':>10}{total:>10}")

输出：

业务线         日调用     峰值QPS      建议并发
客服摘要        42000      8.75        74
工单分类         8600      1.79        15
文档问答         3100      0.65         6
合计                                   95

95 和我们最初拍的那个数差了将近一个数量级。这里有个前置条件要提醒一下：这个数字算出来只是你自己这一侧的期望值，能不能真的跑到，取决于你接的平台在并发上限、超时策略和可用模型列表上给到什么，接入前最好照着自己的 P95 并发把这几项参数核一遍；jiekou.vip 近期上线了面向企业的资源包，参数口径也在文档里列了出来，核对的时候一并对一下就行。真正要避免的是反过来——先把并发数写进配置，再去看平台那边支不支持。

另外，headroom 这个参数我们后来调过两次。1.3 是个起点，不是真理；如果你的业务有明显的日内尖峰（比如每天早上九点整全员打开工作台），这个值要更大。

四、短板三：上线之后就没人再回头看它了

现象。 就是开头说的那件事：六条线接了四条实际没在用，而我们是靠人工翻日志才发现的。更尴尬的是，其中两条线的负责人早就换人了，新来的同事根本不知道这条链路存在。

我们当时怎么想的。 觉得"用得好不好"是业务方自己的事，我们做平台的把可用性守住就行。这个想法现在看是偷懒——我们既然把入口统一了，就是全公司唯一能看到全景的人，这个信息只有我们能产出。

实际原因。 缺一个定期主动推送的东西。日志一直都在写，字段也够，但没人订阅。观测这件事，如果需要人主动去查才能看到，等价于没有。

后来怎么改的。 加了一个每周一早上跑的聚合脚本，从调用日志里按业务线算调用次数、使用人数、失败率、P95 耗时和人均调用次数，打印成一张表发到群里。人均调用次数这一列是后来加的，它比总调用次数更能暴露问题：如果一条线一周有几百次调用但只有两个用户，那大概率还是联调状态。

"""
从调用日志聚合出一张周报表。日志为 JSON Lines，每行至少包含:
  {"ts": "2026-08-04T10:12:33", "biz": "客服摘要", "user_id": "u_1024",
   "ok": true, "latency": 2.31}
运行:
  export CALL_LOG=calls.jsonl
  export REPORT_DAYS=7
"""
import json
import math
import os
from collections import defaultdict
from datetime import datetime, timedelta

LOG_PATH = os.environ.get("CALL_LOG", "calls.jsonl")
DAYS = int(os.environ.get("REPORT_DAYS", "7"))


def percentile(xs, p):
    if not xs:
        return 0.0
    ys = sorted(xs)
    return ys[max(1, math.ceil(p / 100 * len(ys))) - 1]


def iter_records(path, since):
    with open(path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
                ts = datetime.fromisoformat(rec["ts"])
            except (json.JSONDecodeError, KeyError, ValueError):
                continue                      # 脏行跳过，不要让周报因为一行日志挂掉
            if ts >= since:
                yield rec


def main():
    since = datetime.now() - timedelta(days=DAYS)
    stats = defaultdict(lambda: {"calls": 0, "fail": 0, "users": set(), "lat": []})

    for rec in iter_records(LOG_PATH, since):
        s = stats[rec.get("biz", "unknown")]
        s["calls"] += 1
        s["users"].add(rec.get("user_id", "-"))
        s["lat"].append(float(rec.get("latency", 0.0)))
        if not rec.get("ok", False):
            s["fail"] += 1

    print(f"近 {DAYS} 天调用情况（截至 {datetime.now():%Y-%m-%d %H:%M}）")
    header = f"{'业务线':<10}{'调用次数':>9}{'使用人数':>9}{'人均':>7}{'失败率':>8}{'P95耗时':>9}"
    print(header)
    print("-" * 54)

    for biz, s in sorted(stats.items(), key=lambda kv: -kv[1]["calls"]):
        users = len(s["users"]) or 1
        per_user = s["calls"] / users
        fail_rate = s["fail"] / s["calls"] if s["calls"] else 0.0
        flag = "  <- 仍像联调" if per_user < 5 or len(s["users"]) <= 2 else ""
        print(f"{biz:<10}{s['calls']:>9}{len(s['users']):>9}"
              f"{per_user:>7.1f}{fail_rate:>8.1%}"
              f"{percentile(s['lat'], 95):>9.2f}{flag}")


if __name__ == "__main__":
    main()

第一次跑出来的表大概是这个样子，看完基本不用再讨论了：

近 7 天调用情况（截至 2026-08-10 09:02）
业务线        调用次数   使用人数    人均    失败率    P95耗时
------------------------------------------------------
客服摘要        286431      1204   237.9     1.2%     6.41
工单分类         59120       318   185.9     0.9%     3.05
文档问答          1974        86    23.0     2.4%     8.77
代码评审           142         3    47.3     0.7%     5.10  <- 仍像联调
销售话术            67         2    33.5     4.5%    11.20  <- 仍像联调
知识检索            18         2     9.0     0.0%     4.02  <- 仍像联调

这张表后来改变了我们的做事顺序。原本每个季度都在讨论"再接几条线进来"，看完表之后变成了先去问那三条静默的线：是入口太深没人找得到，还是效果不达标不敢用，还是负责人换了没人接手。三个原因我们都遇到了，处理方式完全不同，但如果没有这张表，我们只会继续往外扩。

五、短板四：出问题的时候没有一个可观测的落点

现象。 业务同学在群里说"AI 这块又不好使了"，我们的第一反应是问：哪条线、什么时候、大概什么请求、报什么错。对方通常答不上来，只能截个页面转圈的图。然后我们去翻日志，靠时间范围模糊比对，运气好十分钟能定位，运气不好就成了悬案。

我们当时怎么想的。 服务的访问日志是有的，网关也有指标，觉得排查信息足够。

实际原因。 日志是有的，但它是按请求写的，不是按一次业务操作写的。一次业务操作可能触发五次上游调用，其中一次失败重试了两轮，日志里散成七八行，彼此之间没有共同的标识可以串起来。业务同学手上更是什么都没有，页面上只有一句"服务异常"，连个可以报给我们的编号都没有。

后来怎么改的。 在统一入口那层做了一个客户端封装，做三件事：给每次业务操作生成一个 trace 标识，失败时把这个标识带回给前端展示；重试用指数退避加抖动，只对可重试的状态码重试；主模型连续失败之后降级到备用模型，降级这件事本身写进日志。这段代码同时也是上面那张周报表的数据来源，两块工作在这里合上了。

"""
带 trace 标识、重试与降级的统一客户端。
运行前:
  export LLM_API_BASE=https://<your-endpoint>/v1
  export LLM_API_KEY=<your-key>
  export LLM_MODEL_PRIMARY=<primary-model>
  export LLM_MODEL_FALLBACK=<smaller-model>
  export CALL_LOG=calls.jsonl
"""
import json
import os
import random
import threading
import time
import uuid
from datetime import datetime

import requests

API_BASE = os.environ["LLM_API_BASE"].rstrip("/")
API_KEY = os.environ["LLM_API_KEY"]
PRIMARY = os.environ.get("LLM_MODEL_PRIMARY", "qwen2.5-72b-instruct")
FALLBACK = os.environ.get("LLM_MODEL_FALLBACK", "qwen2.5-7b-instruct")
LOG_PATH = os.environ.get("CALL_LOG", "calls.jsonl")
TIMEOUT = float(os.environ.get("LLM_TIMEOUT", "30"))

RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}

_SESSION = requests.Session()
_SESSION.headers.update({
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
})
_LOG_LOCK = threading.Lock()


def _write_log(record):
    record["ts"] = datetime.now().isoformat(timespec="seconds")
    line = json.dumps(record, ensure_ascii=False)
    with _LOG_LOCK, open(LOG_PATH, "a", encoding="utf-8") as f:
        f.write(line + "\n")


class UpstreamError(RuntimeError):
    """带 trace 标识的失败，前端可以把 trace 直接显示给用户。"""

    def __init__(self, trace, detail):
        super().__init__(f"[trace={trace}] {detail}")
        self.trace = trace


def chat(prompt, biz, user_id, max_retry=3):
    trace = uuid.uuid4().hex[:12]
    started = time.perf_counter()
    attempt = 0
    last_err = "unknown"

    for model in (PRIMARY, FALLBACK):
        for i in range(max_retry):
            attempt += 1
            t0 = time.perf_counter()
            try:
                resp = _SESSION.post(
                    f"{API_BASE}/chat/completions",
                    json={
                        "model": model,
                        "messages": [{"role": "user", "content": prompt}],
                        "temperature": 0.2,
                    },
                    timeout=TIMEOUT,
                    headers={"X-Trace-Id": trace},
                )
                if resp.status_code in RETRYABLE_STATUS:
                    raise requests.HTTPError(f"HTTP {resp.status_code}")
                resp.raise_for_status()
                text = resp.json()["choices"][0]["message"]["content"]
                _write_log({
                    "trace": trace, "biz": biz, "user_id": user_id,
                    "model": model, "ok": True, "attempt": attempt,
                    "degraded": model != PRIMARY,
                    "in_chars": len(prompt), "out_chars": len(text),
                    "latency": round(time.perf_counter() - t0, 3),
                })
                return text
            except Exception as exc:
                last_err = f"{type(exc).__name__}: {exc}"
                if i < max_retry - 1:
                    # 指数退避 + 抖动，避免一批请求同时重试把上游再压一次
                    time.sleep(min(8.0, 2 ** i) * (0.5 + random.random()))

    _write_log({
        "trace": trace, "biz": biz, "user_id": user_id,
        "model": None, "ok": False, "attempt": attempt, "degraded": True,
        "in_chars": len(prompt), "out_chars": 0,
        "latency": round(time.perf_counter() - started, 3),
        "error": last_err,
    })
    raise UpstreamError(trace, f"{attempt} 次尝试均失败，最后一次: {last_err}")


if __name__ == "__main__":
    try:
        answer = chat("用两句话概括这段客服会话的核心诉求：客户反馈登录后无法进入工作台。",
                      biz="客服摘要", user_id="u_1024")
        print("结果:", answer[:80])
    except UpstreamError as err:
        print("失败，把这个标识发给我们即可:", err.trace)

上线之后群里的对话形态变了。以前是"AI 又不好使了"，现在是"trace=9f3c1a2b7d04 这条失败了"，我们拿标识直接 grep 日志，能一眼看到它试了几次、有没有降级、每次失败在哪一步。定位时间从十几分钟压到一两分钟。降级那个字段也带来了额外的好处：我们发现有一段时间降级比例悄悄涨到 4%，顺着查出来是主模型在某个时段的超时变多了，这个问题在加字段之前完全是隐形的。

六、回头看

这四块短板其实是同一个毛病的四种表现：我们习惯了在会议室里用共识代替测量。报一个日调用次数比抽 200 条样本跑一遍容易，用平均耗时算并发比看 P95 分布容易，假设"接了就在用"比每周产出一张表容易，相信"日志够用"比重新设计一次埋点容易。每一次省下来的都是几个小时，而每一次的代价都是几周之后的一次返工。

如果只留一条经验，我会选这个：任何要写进配置文件的数字，都应该有一段能重新跑出它的代码，而不只是一份文档里的结论。数字会过时，业务量会变，模型会换，但那段代码可以隔一个季度再跑一遍，然后你就知道该不该改配置了。我们现在把上面这几段脚本放进了定时任务，第一段每个月跑一次刷新分布，第三段每周一早上出表。它们都不复杂，加起来不到四百行，但那次季度回顾之后，我们没有再靠拍脑袋定过容量。