一、接口通了和真的用起来了是两件事去年下半年我们做了一件当时看起来非常正确的事把散落在各个小组手里的十几个调用大模型的脚本收拢成一个内部服务对外只留一个 HTTP 入口鉴权、限流、日志、模型路由都放在这一层。上线那天大家还挺高兴六条业务线陆陆续续接了进来客服摘要、工单分类、文档问答、销售话术生成、代码评审辅助、内部知识检索覆盖面看着很漂亮。三个月后我们做季度回顾做了一张很朴素的表每条业务线这一周被调用了多少次、有多少个不同的人在用。结果有点难看。六条线里真正在跑的是两条一条每天几万次调用一条几千次剩下四条接口通了Demo 演示过了然后就停在那儿了有的一周只有几十次调用看用户 ID 全是当初做联调那两个人。更让人不舒服的是这个结论我们是靠人工翻日志得出来的而不是系统主动告诉我们的。也就是说如果那个季度回顾没人提这件事可以再糊三个月。后面的内容是我们那次复盘留下来的四块短板以及我们后来补的东西。每一块我都尽量把当时怎么想的写出来因为回过头看判断错的地方往往不是技术能力是懒得去拿真实数据。这几块短板补完我们才算真正把企业级 AI 工作提效这件事做扎实——不是接口通了就结束而是几条线都稳定在用、并且用得怎么样随时看得见。现在针对ToB端的企业提效已经成为各个企业、工作团队都比较关注和急于去改变和升级的关键一环。说到用AI提效最重要的就是获取到又便宜又足量的token资源jiekou.vip 近期上线了企业资源包也是面向这类多团队接入场景的成本仅需按量使用的75%。二、短板一容量规模全靠拍脑袋手上一条真实数据都没有现象。做规划的时候我们需要估一个数这套服务上线后大概会被调用多少次、单次请求会有多大。当时的做法是开会让每条业务线的负责人报一个数。报上来的数字非常整齐客服说每天大概两万单一单一次调用文档问答说每天几千次吧。我们把这些数加起来乘 1.2写进了容量规划文档。我们当时怎么想的。觉得业务方最了解自己的量他们报的数应该最准。而且这件事看着不难不至于为了一个估算专门去跑一轮实测。实际原因。业务方报的是业务量不是请求量这两个中间隔着一层我们自己写的逻辑。客服摘要那条线一个会话最终会被切成 3 到 5 段分别调用再加一次汇总实际请求量是业务量的 4 倍多。反过来文档问答那条线报了每天几千次实际因为入口藏得深一天只有一两百次真实调用。两个方向的偏差都不小加总之后看着差不多对其实是两个错误互相抵消。还有一个更隐蔽的问题我们只讨论了平均多大没人问过最大有多大。而单次请求的字符数在真实数据里分布得非常偏客服会话里有大量三五句话就结束的也有客户贴了一整页报错日志的长会话后者的耗时是前者的十几倍。用平均值做规划等于假装那条长尾不存在。后来怎么改的。规矩改成一句话任何容量数字必须有一份小样本实测垫底。从每条业务线的真实流水里随机抽 200 条用生产同样的 prompt 模板跑一遍把输入字符数、输出字符数、单次耗时全部记下来然后看 P50 和 P95不看平均值。 用真实业务样本实测单次调用的字符数与耗时分布。 运行前: export LLM_API_BASEhttps://your-endpoint/v1 export LLM_API_KEYyour-key export LLM_MODELmodel-name export SAMPLE_FILEsamples.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: fBearer {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, jsonpayload, timeoutTIMEOUT) 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, encodingutf-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.72sP95 的输入字符数是 P50 的五倍多最慢那条 18 秒。这两个数字是会上任何人拍脑袋都拍不出来的而它们直接决定了后面的超时阈值和并发配置该怎么设。顺便说一句那 1.5% 的失败率也是意外收获——我们原本以为是 0。三、短板二按最理想的情况排并发一点波动余量都没留现象。服务上线后的第二周客服那条线在一个周一上午集中出现超时前端页面转圈业务同学在群里问是不是挂了。看监控服务本身没挂是排队排上去了队列深度一路涨到几百。我们当时怎么想的。规划的时候我们算过一版并发日调用次数除以 86400 得到平均每秒请求量再按单次耗时 2 秒去推在途请求数。数字很小看着非常安全所以并发上限设得很保守。实际原因。两个错误叠在一起。第一把一天的量摊到 86400 秒等于假设凌晨三点和上午十点的请求量一样多——真实情况是八成以上的调用集中在四个小时里峰值大概是全天均值的三倍。第二用 P50 耗时 2 秒去算在途请求数而排队恰恰是被那些 6 秒、18 秒的长请求撑起来的它们占着并发槽位不放。后来怎么改的。把并发估算写成一段代码输入是上一步实测出来的 P95 耗时和业务线的日调用次数输出是建议并发数中间的假设全部显式写成参数谁想调就改参数不要在会上争论。用的就是最朴素的 Littles 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_hours4.0, peak_ratio3.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_latencyP95_LATENCY, headroom1.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 合计 9595 和我们最初拍的那个数差了将近一个数量级。这里有个前置条件要提醒一下这个数字算出来只是你自己这一侧的期望值能不能真的跑到取决于你接的平台在并发上限、超时策略和可用模型列表上给到什么接入前最好照着自己的 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_LOGcalls.jsonl export REPORT_DAYS7 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, encodingutf-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(daysDAYS) 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(), keylambda 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_BASEhttps://your-endpoint/v1 export LLM_API_KEYyour-key export LLM_MODEL_PRIMARYprimary-model export LLM_MODEL_FALLBACKsmaller-model export CALL_LOGcalls.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: fBearer {API_KEY}, Content-Type: application/json, }) _LOG_LOCK threading.Lock() def _write_log(record): record[ts] datetime.now().isoformat(timespecseconds) line json.dumps(record, ensure_asciiFalse) with _LOG_LOCK, open(LOG_PATH, a, encodingutf-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_retry3): 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, }, timeoutTIMEOUT, headers{X-Trace-Id: trace}, ) if resp.status_code in RETRYABLE_STATUS: raise requests.HTTPError(fHTTP {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_idu_1024) print(结果:, answer[:80]) except UpstreamError as err: print(失败把这个标识发给我们即可:, err.trace)上线之后群里的对话形态变了。以前是AI 又不好使了现在是trace9f3c1a2b7d04 这条失败了我们拿标识直接 grep 日志能一眼看到它试了几次、有没有降级、每次失败在哪一步。定位时间从十几分钟压到一两分钟。降级那个字段也带来了额外的好处我们发现有一段时间降级比例悄悄涨到 4%顺着查出来是主模型在某个时段的超时变多了这个问题在加字段之前完全是隐形的。六、回头看这四块短板其实是同一个毛病的四种表现我们习惯了在会议室里用共识代替测量。报一个日调用次数比抽 200 条样本跑一遍容易用平均耗时算并发比看 P95 分布容易假设接了就在用比每周产出一张表容易相信日志够用比重新设计一次埋点容易。每一次省下来的都是几个小时而每一次的代价都是几周之后的一次返工。如果只留一条经验我会选这个任何要写进配置文件的数字都应该有一段能重新跑出它的代码而不只是一份文档里的结论。数字会过时业务量会变模型会换但那段代码可以隔一个季度再跑一遍然后你就知道该不该改配置了。我们现在把上面这几段脚本放进了定时任务第一段每个月跑一次刷新分布第三段每周一早上出表。它们都不复杂加起来不到四百行但那次季度回顾之后我们没有再靠拍脑袋定过容量。