这半年我在公司内部做 AI 接入的答疑从 CRM 工单分类到知识库问答再到风控辅助审核前后对接了七八条业务线。有意思的是不同团队的问题高度重合基本就是下面这 8 个反复被问而且大多数人踩的坑也一样。所以我干脆按问答的形式把它们整理出来。每个问题先给结论再说为什么能用代码说明的就直接贴代码。代码都是能跑的配置全部走环境变量别把密钥写进仓库。需要说明的是这 8 个问题都是围绕企业级 AI 工作提效展开的——目标不是把某个模型调通而是让几条业务线都能稳定用上、并且能持续看清用得怎么样。现在企业级的AI提效已经是大势所趋不过想使用AI来提升效率最重要的是token资源jiekou.vip 近期上线了企业资源包做企业内部多团队接入的话可顺手了解一下。Q1这次调用应该走同步还是丢进异步队列结论只看两件事——单次调用的实测耗时以及用户此刻是不是在屏幕前等结果。其他都是次要的。短分类、短抽取、单句改写这类任务实测下来大多在 400 到 1200 毫秒之间用户在等走同步没问题。批量清洗历史数据、给几万条工单打标、夜里跑摘要用户不在等必须走异步没有讨论空间。难的是中间地带1 到 3 秒那一档。我的判断依据是看这个接口所在的链路上还串了几个调用。如果 AI 调用是链路里唯一的慢操作同步可以接受如果前面已经有两次数据库查询加一次外部接口那这 2 秒会把整个接口的 P95 推到 5 秒以上前端体验会明显变差这时候宁可改成先返回一个占位结果AI 结果异步回填。同步方案的隐藏问题是它把模型侧的延迟抖动直接暴露给了 web 服务。模型平时 1 秒偶尔 8 秒这 8 秒里你的 worker 是被占住的并发数上不去表现出来就是整个服务连正常请求一起变慢。异步的问题则是工程量要多一个队列、一张任务表、一个结果通知通道还要有人维护。给个实用建议先按同步实现但把调用包在一个函数里函数签名只表达输入什么、拿到什么不要在签名上体现同步这件事。后面要改异步换实现就行调用方不用动。Q2超时到底设多少合适结论别拍脑袋写 30 秒用你自己业务的实测 P95 乘 1.5 到 2 倍。超时设 30 秒的坏处不是等太久是失败发现得太慢。上游真出问题的时候你的 worker 会被一批注定失败的请求占满 30 秒重试逻辑在这 30 秒里完全不起作用服务对外表现是整体卡死而不是局部失败。反过来超时设得比 P95 还短你会人为制造一堆本来能成功的失败失败率虚高还触发无意义的重试。正确做法是先拿真实的提示词跑一批采样看清耗时分布再定值。跑一次的事别跳过。import os import time import requests BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) PROMPT 把下面这句话归类为 投诉/咨询/表扬 之一只输出类别两个字门店空调坏了三天没人来修 def one_call(timeout60.0): t0 time.perf_counter() resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: MODEL, messages: [{role: user, content: PROMPT}], temperature: 0, }, timeouttimeout, ) resp.raise_for_status() return (time.perf_counter() - t0) * 1000 def percentile(values, p): if not values: return 0.0 values sorted(values) idx int(round((len(values) - 1) * p)) return values[max(0, min(len(values) - 1, idx))] if __name__ __main__: latencies, failed [], 0 for i in range(30): try: latencies.append(one_call()) except Exception as exc: failed 1 print(f[{i:02d}] 失败: {type(exc).__name__}) print(f\n样本 {len(latencies)} 条失败 {failed} 条) print(fP50 {percentile(latencies, 0.50):.0f} ms) print(fP95 {percentile(latencies, 0.95):.0f} ms) print(fP99 {percentile(latencies, 0.99):.0f} ms) print(f建议超时 {percentile(latencies, 0.95) * 1.8 / 1000:.1f} s)我们工单分类那条线跑出来是这样样本 30 条失败 0 条 P50 612 ms P95 1483 ms P99 2071 ms 建议超时 2.7 s于是超时定在 3 秒。这个值要跟着提示词变提示词从两句话涨到一页背景说明之后重新跑一次别用老值。Q3重试要怎么写才不会把自己搞雪崩结论三条缺一不可——指数退避、加随机抖动、只重试该重试的错误。少了任意一条重试就从提高成功率变成放大故障。最常见的写法是for i in range(3): try... except: continue没有间隔。这种写法在上游正常时看不出问题一旦上游抖动你的请求量瞬间变成平时的 3 倍本来能自愈的抖动被你推成了持续故障。加了固定间隔也不够因为你所有 worker 是同时收到失败的它们会在同一时刻一起重试形成整齐的脉冲这就是要加抖动的原因——把重试时间点打散。第三条最容易被忽略不是所有错误都该重试。429 和 5xx 该重试超时该重试。400 参数错误、401 鉴权失败重试一万次也是失败只会白白拉长响应时间。还有一类要特别小心非幂等的业务操作比如调用模型生成一段文案然后直接写入数据库并发通知这种整体不能重试只能重试其中读取模型结果这一步写库和通知要单独做幂等保护。import os import random import time import requests BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) RETRIABLE_STATUS {408, 429, 500, 502, 503, 504} MAX_ATTEMPTS 4 BASE_DELAY 0.5 MAX_DELAY 8.0 class NonRetriable(Exception): pass def chat(prompt, timeout3.0): last_err None for attempt in range(MAX_ATTEMPTS): try: resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: MODEL, messages: [{role: user, content: prompt}], temperature: 0, }, timeouttimeout, ) if resp.status_code in RETRIABLE_STATUS: raise requests.HTTPError(fstatus{resp.status_code}) if resp.status_code 400: raise NonRetriable(fstatus{resp.status_code} body{resp.text[:200]}) return resp.json()[choices][0][message][content].strip() except NonRetriable: raise except (requests.Timeout, requests.ConnectionError, requests.HTTPError) as exc: last_err exc if attempt MAX_ATTEMPTS - 1: break delay min(MAX_DELAY, BASE_DELAY * (2 ** attempt)) delay delay * (0.5 random.random() * 0.5) print(f第 {attempt 1} 次失败({exc}){delay:.2f}s 后重试) time.sleep(delay) raise RuntimeError(f重试 {MAX_ATTEMPTS} 次仍失败: {last_err}) if __name__ __main__: print(chat(用一句话说明什么是幂等))抖动那行是关键delay * (0.5 random.random() * 0.5)把等待时间压到理论值的 50% 到 100% 之间相同时刻失败的一批请求会散在一个区间里重试而不是撞在一起。还要提醒一句重试次数不是越多越好。我们线上定的是 4 次超过这个数意味着上游确实不可用了继续试只是在拖长调用方的等待。这时候正确的动作是快速失败并降级比如工单分类失败就落到待人工状态别让用户卡在转圈上。Q4几条业务线共用一个入口怎么知道到底是谁在调结论每条业务线一把独立的 Key加上结构化日志里带业务标识。两个都要有缺一个都查不清。只靠日志不够因为日志里的业务标识是应用自己写的写错了、漏写了、新同事复制老代码没改你查出来的归属就是错的。只靠 Key 也不够Key 只能告诉你哪个应用在调告诉不了你是这个应用里的哪个功能、哪个租户。所以是两层Key 划应用边界日志字段划功能和租户边界。顺带说一个接入前要确认的技术前提这套做法要求平台侧支持一个应用签发多把 Key并且能按 Key 维度回看调用记录否则你的独立 Key 只是个摆设拆了也看不出区别。这一点各家实现差别不小写代码之前先翻文档确认比如 jiekou.vip 近期上线了企业资源包多 Key 签发和按 Key 查调用记录在它的文档里能查到具体字段接入前对一下就行。确认清楚之后剩下的就是把埋点写规范。日志字段我建议至少包含这几个业务线、功能点、租户、模型、耗时、是否成功、重试次数、请求唯一 ID。输出成一行 JSON方便后面直接聚合。import json import os import time import uuid import logging logging.basicConfig(levellogging.INFO, format%(message)s) logger logging.getLogger(ai_call) # 每条业务线一把 Key从环境变量按业务线取 KEY_MAP { ticket: os.environ.get(AI_KEY_TICKET, ), kb_qa: os.environ.get(AI_KEY_KB_QA, ), risk: os.environ.get(AI_KEY_RISK, ), } def resolve_key(biz): key KEY_MAP.get(biz) if not key: raise RuntimeError(f业务线 {biz} 没有配置独立 Key拒绝走公共 Key) return key def traced_call(biz, feature, tenant, prompt): trace_id uuid.uuid4().hex[:12] key resolve_key(biz) t0 time.perf_counter() ok, retries, err True, 0, None try: # 这里换成 Q3 里的 chat()key 传 resolve_key(biz) 的结果 result f模拟返回 {prompt[:12]} except Exception as exc: ok, err, result False, f{type(exc).__name__}: {exc}, None logger.info(json.dumps({ trace_id: trace_id, biz: biz, feature: feature, tenant: tenant, model: os.environ.get(AI_MODEL, gpt-4o-mini), key_tail: key[-4:] if key else , elapsed_ms: round((time.perf_counter() - t0) * 1000, 1), ok: ok, retries: retries, err: err, chars_in: len(prompt), chars_out: len(result or ), }, ensure_asciiFalse)) return result if __name__ __main__: traced_call(ticket, auto_classify, tenant_8812, 门店空调坏了三天没人来修)输出是一行 JSON{trace_id: a3f19c4e77b1, biz: ticket, feature: auto_classify, tenant: tenant_8812, model: gpt-4o-mini, key_tail: 9c2f, elapsed_ms: 0.1, ok: true, retries: 0, chars_in: 14, chars_out: 19, err: null}注意key_tail只写后 4 位完整 Key 绝对不能进日志。这条我见过有团队踩日志系统的权限往往比生产环境宽松得多。另外那个resolve_key里的抛错是故意的。如果找不到业务线对应的 Key 就退回公共 Key半年后你会发现所有调用都挂在公共 Key 上谁也说不清是谁在调。宁可让新接入的业务线在第一次跑就报错逼着他去申请自己的 Key。Q5上线前怎么估我需要多少并发结论不要靠猜也不要靠我们大概有多少用户这种口径。拿小样本实测出耗时分布再用目标请求量和 P95 反推最小并发数然后往上留一倍余量。推导很简单一个 worker 在 P95 耗时下每秒能处理1 / P95个请求要支撑每秒 N 个请求就需要N × P95个并发位。这就是 Littles Law 的实用版本。用 P95 而不是 P50 是因为你要扛的是坏情况用平均值算出来的并发数在高峰期一定不够。举个我们真实做过的知识库问答那条线P95 实测 2.4 秒业务方给的目标是高峰期每秒 12 个请求算出来12 × 2.4 28.8取 29 个并发位留余量到 60。为什么留一倍一是提示词后面会变长耗时会涨二是上游抖动时 P95 会临时翻倍没有余量就直接排队积压。import math import os import time import requests from concurrent.futures import ThreadPoolExecutor BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) SAMPLES [ 把这条工单归类APP 打开就闪退, 把这条工单归类想问一下账号怎么改绑手机号, 把这条工单归类客服小李服务很好想表扬, 把这条工单归类门店空调坏了三天没人来修, ] TARGET_RPS 12.0 # 业务方给的高峰目标请求量 SAFETY 2.0 # 余量倍数 def timed(prompt): t0 time.perf_counter() try: resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{model: MODEL, messages: [{role: user, content: prompt}], temperature: 0}, timeout10, ) resp.raise_for_status() return (time.perf_counter() - t0), True except Exception: return (time.perf_counter() - t0), False def percentile(vals, p): vals sorted(vals) return vals[max(0, min(len(vals) - 1, int(round((len(vals) - 1) * p))))] if __name__ __main__: jobs SAMPLES * 10 with ThreadPoolExecutor(max_workers4) as pool: results list(pool.map(timed, jobs)) durs [d for d, ok in results if ok] fail_rate 1 - len(durs) / len(results) p50, p95 percentile(durs, 0.50), percentile(durs, 0.95) need math.ceil(TARGET_RPS * p95) print(f样本 {len(results)} 条失败率 {fail_rate:.1%}) print(fP50 {p50:.2f}s P95 {p95:.2f}s) print(f支撑 {TARGET_RPS:.0f} rps 的最小并发 {need}) print(f建议连接池 / 线程池上限 {math.ceil(need * SAFETY)})跑出来大致是样本 40 条失败率 0.0% P50 0.71s P95 2.38s 支撑 12 rps 的最小并发 29 建议连接池 / 线程池上限 58一个容易忽略的细节算出来的并发数要同时体现在三个地方——线程池上限、HTTP 连接池上限、以及你在平台侧配置的并发上限。只调线程池不调连接池requests 底层默认的连接池会成为瓶颈你会看到大量请求卡在等连接而不是等模型表现出来是耗时莫名变长但模型侧没异常。用requests.Session挂一个HTTPAdapter(pool_maxsize58)就行。Q6模型返回不稳定、格式不对怎么处理结论不要指望靠提示词把格式说清楚就万事大吉。必须是三层输出强约束、结果强校验、校验失败降级重问。这是我见过上线后最容易出事的一环。开发的时候试了十次都是规整的 JSON上线跑了两万次其中三百多次前面多了一句好的以下是分类结果你的json.loads直接抛异常整条链路挂掉。更隐蔽的情况是 JSON 合法但字段值不在你的枚举里模型自己发明了一个类别这种错不会抛异常会直接污染你的数据。具体做法能用response_format约束成 JSON 就一定要用解析之后必须按业务枚举做一次显式校验不能只判断能不能解析校验失败时不要直接报错而是把错误信息拼回去让模型改一次改一次还不对就落到人工兜底。这里注意重问的次数要卡死在 1 到 2 次不能无限循环否则遇到模型固执的边缘输入会一直转。import json import os import requests BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) ALLOWED {投诉, 咨询, 表扬} SYSTEM ( 你是工单分类器。只输出 JSON格式为 {category: 投诉|咨询|表扬, confidence: 0.0~1.0, reason: 不超过20字}。 category 必须严格是这三个词之一不要输出其他内容。 ) def post(messages): resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: MODEL, messages: messages, temperature: 0, response_format: {type: json_object}, }, timeout5, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def validate(raw): 返回 (对象, 错误说明)。错误说明非空表示校验失败。 try: obj json.loads(raw) except json.JSONDecodeError as exc: return None, f不是合法 JSON: {exc} if not isinstance(obj, dict): return None, 顶层必须是对象 cat obj.get(category) if cat not in ALLOWED: return None, fcategory 必须是 {/.join(sorted(ALLOWED))} 之一你给的是 {cat!r} conf obj.get(confidence) if not isinstance(conf, (int, float)) or not 0 conf 1: return None, fconfidence 必须是 0~1 的数字你给的是 {conf!r} return obj, def classify(text, max_fix1): messages [ {role: system, content: SYSTEM}, {role: user, content: text}, ] for attempt in range(max_fix 1): raw post(messages) obj, err validate(raw) if not err: return {ok: True, data: obj, fixed: attempt} print(f第 {attempt 1} 次输出不合规{err}) if attempt max_fix: break messages [ {role: assistant, content: raw}, {role: user, content: f输出不符合要求{err}。请只重新输出合规的 JSON。}, ] return {ok: False, data: None, fallback: manual_review} if __name__ __main__: for t in [门店空调坏了三天没人来修, 账号怎么改绑手机号, 客服小李态度特别好]: print(t, -, classify(t))输出门店空调坏了三天没人来修 - {ok: True, data: {category: 投诉, confidence: 0.95, reason: 报修长期未处理}, fixed: 0} 账号怎么改绑手机号 - {ok: True, data: {category: 咨询, confidence: 0.97, reason: 询问改绑操作流程}, fixed: 0} 客服小李态度特别好 - {ok: True, data: {category: 表扬, confidence: 0.96, reason: 肯定服务态度}, fixed: 0}fixed这个字段建议保留并打进日志。它反映了提示词的质量如果线上fixed 0的比例超过百分之几说明你的提示词或者枚举定义有问题该去改提示词而不是靠重问兜着。我们把这个比例当成一个常规看的指标改提示词之后它从 4.1% 掉到 0.3%效果很直观。confidence那个字段要说明一下模型自报的置信度并不可靠别拿它当概率用。但它作为一个粗糙的分流阈值是有用的——低于 0.6 的直接转人工能挡掉相当一部分模糊工单。Q7提示词改了、模型版本换了怎么知道有没有退化结论留一个小规模的回归集每次改动跑一遍比对结果差异。不需要多复杂30 到 50 条就足够发现大部分退化。很多团队在这一步是完全裸奔的改提示词靠手动试三五条觉得好像变好了就上线两周后业务方反馈某类工单全分错了回头查不知道是哪次改动引入的。问题在于人工抽查的样本永远偏向你熟悉的那几条而退化通常发生在边缘输入上。回归集怎么建我的经验是三类样本按比例混一半是典型样本保证主路径不出问题三成是历史上真出过错的样本这部分最有价值每次线上发现一个分错的 case 就补进去回归集会越来越贴合你的实际分布剩下两成放边界样本比如空输入、超长输入、掺杂表情和乱码的输入、语义上确实模棱两可的输入。判断标准要分开定别一刀切。有唯一正确答案的题目分类、抽取、判断直接比对期望值一次成率掉了就是退化这类可以完全自动化。开放生成类的题目摘要、改写没有唯一答案只能做弱校验长度是否在合理区间、是否包含必须出现的关键实体、是否没有出现禁止出现的表述。不要为了追求自动打分去引入一个模型给另一个模型打分的复杂机制那套东西本身也会漂维护它的精力比它省下的多。还要记住一件事回归集只能证明没有明显退化不能证明变好了。变好这件事得看线上真实数据看那些人工修正率、转人工比例的变化看一两周的趋势。回归集的作用是拦住明显的坏改动让你敢改这个价值已经足够大了。实操上建议把回归集放进版本库跟提示词放在一起改提示词的那个提交必须带上回归结果的对比。这样半年后回头查哪次改动导致的翻提交记录就能定位。Q8上线之后怎么持续知道用得好不好结论把 Q4 那份结构化日志按业务线聚合出几个固定指标做成周报自动发出来。不用建复杂的看板一份能定期看到的表格比一个没人打开的仪表盘管用得多。我们固定看六个调用次数、使用人数、失败率、P95 耗时、平均字符数、重试次数。这几个指标的组合能覆盖大部分要判断的事。举两个实际判断的例子调用次数涨但使用人数没涨说明是少数几个人在重复调往往是他们对结果不满意在反复试该去看这条业务线的输出质量失败率不高但 P95 明显变长通常是提示词被人悄悄加长了去 diff 一下提示词就知道。import json import os import statistics from collections import defaultdict LOG_PATH os.environ.get(AI_LOG_PATH, ai_call.log) def percentile(vals, p): vals sorted(vals) return vals[max(0, min(len(vals) - 1, int(round((len(vals) - 1) * p))))] def weekly_report(path): buckets defaultdict(lambda: { calls: 0, fails: 0, users: set(), latencies: [], chars: [], retries: 0, }) with open(path, encodingutf-8) as fh: for line in fh: line line.strip() if not line: continue try: rec json.loads(line) except json.JSONDecodeError: continue b buckets[rec.get(biz, unknown)] b[calls] 1 b[users].add(rec.get(tenant, -)) b[retries] rec.get(retries, 0) if rec.get(ok): b[latencies].append(rec.get(elapsed_ms, 0)) b[chars].append(rec.get(chars_in, 0) rec.get(chars_out, 0)) else: b[fails] 1 header f{业务线:10}{调用次数:8}{使用人数:8}{失败率:8}{P95(ms):10}{均字符:8}{重试:6} print(header) print(- * len(header)) for biz, b in sorted(buckets.items(), keylambda kv: -kv[1][calls]): lat percentile(b[latencies], 0.95) if b[latencies] else 0 chars statistics.mean(b[chars]) if b[chars] else 0 print(f{biz:10}{b[calls]:8}{len(b[users]):8} f{b[fails] / b[calls]:7.1%}{lat:10.0f}{chars:8.0f}{b[retries]:6}) if __name__ __main__: weekly_report(LOG_PATH)跑一周的日志出来是这样业务线 调用次数 使用人数 失败率 P95(ms) 均字符 重试 -------------------------------------------------------------------- ticket 18432 214 0.4% 1521 186 92 kb_qa 6207 88 1.1% 2386 742 143 risk 1140 12 0.2% 903 231 6这张表我们每周一早上自动发到群里。价值不在于某一周的绝对数字在于连续看几周之后你对正常长什么样有了感觉异常一眼就能看出来。上面这份里 kb_qa 的重试次数偏高143 次重试对应 6207 次调用比例超过 2%比另外两条线高一个数量级后来查出来是它的超时值还是按早期短提示词定的 1.5 秒提示词加长之后大量请求擦着超时线过触发了重试。改超时值之后就正常了——这就是 Q2 那句提示词变了要重新跑一遍采样的由来。指标别贪多。我一开始定了十几个三周之后发现没人看砍到六个反而每周都有人真的去读、去提问题。这 8 个问题里前 3 个决定你的接入会不会在高峰期崩第 4 个决定半年后你还能不能说清楚系统在干什么后 4 个决定这套东西能不能持续演进而不是上线即腐化。顺序不是随便排的也大致是我建议的落地顺序。如果只能挑一件事先做我会选 Q4 的结构化日志。它本身不解决任何问题但没有它后面所有问题你都只能靠猜。 把 AI 能力接进内部系统，工程上最常被问到的 8 个问题
这半年我在公司内部做 AI 接入的答疑从 CRM 工单分类到知识库问答再到风控辅助审核前后对接了七八条业务线。有意思的是不同团队的问题高度重合基本就是下面这 8 个反复被问而且大多数人踩的坑也一样。所以我干脆按问答的形式把它们整理出来。每个问题先给结论再说为什么能用代码说明的就直接贴代码。代码都是能跑的配置全部走环境变量别把密钥写进仓库。需要说明的是这 8 个问题都是围绕企业级 AI 工作提效展开的——目标不是把某个模型调通而是让几条业务线都能稳定用上、并且能持续看清用得怎么样。现在企业级的AI提效已经是大势所趋不过想使用AI来提升效率最重要的是token资源jiekou.vip 近期上线了企业资源包做企业内部多团队接入的话可顺手了解一下。Q1这次调用应该走同步还是丢进异步队列结论只看两件事——单次调用的实测耗时以及用户此刻是不是在屏幕前等结果。其他都是次要的。短分类、短抽取、单句改写这类任务实测下来大多在 400 到 1200 毫秒之间用户在等走同步没问题。批量清洗历史数据、给几万条工单打标、夜里跑摘要用户不在等必须走异步没有讨论空间。难的是中间地带1 到 3 秒那一档。我的判断依据是看这个接口所在的链路上还串了几个调用。如果 AI 调用是链路里唯一的慢操作同步可以接受如果前面已经有两次数据库查询加一次外部接口那这 2 秒会把整个接口的 P95 推到 5 秒以上前端体验会明显变差这时候宁可改成先返回一个占位结果AI 结果异步回填。同步方案的隐藏问题是它把模型侧的延迟抖动直接暴露给了 web 服务。模型平时 1 秒偶尔 8 秒这 8 秒里你的 worker 是被占住的并发数上不去表现出来就是整个服务连正常请求一起变慢。异步的问题则是工程量要多一个队列、一张任务表、一个结果通知通道还要有人维护。给个实用建议先按同步实现但把调用包在一个函数里函数签名只表达输入什么、拿到什么不要在签名上体现同步这件事。后面要改异步换实现就行调用方不用动。Q2超时到底设多少合适结论别拍脑袋写 30 秒用你自己业务的实测 P95 乘 1.5 到 2 倍。超时设 30 秒的坏处不是等太久是失败发现得太慢。上游真出问题的时候你的 worker 会被一批注定失败的请求占满 30 秒重试逻辑在这 30 秒里完全不起作用服务对外表现是整体卡死而不是局部失败。反过来超时设得比 P95 还短你会人为制造一堆本来能成功的失败失败率虚高还触发无意义的重试。正确做法是先拿真实的提示词跑一批采样看清耗时分布再定值。跑一次的事别跳过。import os import time import requests BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) PROMPT 把下面这句话归类为 投诉/咨询/表扬 之一只输出类别两个字门店空调坏了三天没人来修 def one_call(timeout60.0): t0 time.perf_counter() resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: MODEL, messages: [{role: user, content: PROMPT}], temperature: 0, }, timeouttimeout, ) resp.raise_for_status() return (time.perf_counter() - t0) * 1000 def percentile(values, p): if not values: return 0.0 values sorted(values) idx int(round((len(values) - 1) * p)) return values[max(0, min(len(values) - 1, idx))] if __name__ __main__: latencies, failed [], 0 for i in range(30): try: latencies.append(one_call()) except Exception as exc: failed 1 print(f[{i:02d}] 失败: {type(exc).__name__}) print(f\n样本 {len(latencies)} 条失败 {failed} 条) print(fP50 {percentile(latencies, 0.50):.0f} ms) print(fP95 {percentile(latencies, 0.95):.0f} ms) print(fP99 {percentile(latencies, 0.99):.0f} ms) print(f建议超时 {percentile(latencies, 0.95) * 1.8 / 1000:.1f} s)我们工单分类那条线跑出来是这样样本 30 条失败 0 条 P50 612 ms P95 1483 ms P99 2071 ms 建议超时 2.7 s于是超时定在 3 秒。这个值要跟着提示词变提示词从两句话涨到一页背景说明之后重新跑一次别用老值。Q3重试要怎么写才不会把自己搞雪崩结论三条缺一不可——指数退避、加随机抖动、只重试该重试的错误。少了任意一条重试就从提高成功率变成放大故障。最常见的写法是for i in range(3): try... except: continue没有间隔。这种写法在上游正常时看不出问题一旦上游抖动你的请求量瞬间变成平时的 3 倍本来能自愈的抖动被你推成了持续故障。加了固定间隔也不够因为你所有 worker 是同时收到失败的它们会在同一时刻一起重试形成整齐的脉冲这就是要加抖动的原因——把重试时间点打散。第三条最容易被忽略不是所有错误都该重试。429 和 5xx 该重试超时该重试。400 参数错误、401 鉴权失败重试一万次也是失败只会白白拉长响应时间。还有一类要特别小心非幂等的业务操作比如调用模型生成一段文案然后直接写入数据库并发通知这种整体不能重试只能重试其中读取模型结果这一步写库和通知要单独做幂等保护。import os import random import time import requests BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) RETRIABLE_STATUS {408, 429, 500, 502, 503, 504} MAX_ATTEMPTS 4 BASE_DELAY 0.5 MAX_DELAY 8.0 class NonRetriable(Exception): pass def chat(prompt, timeout3.0): last_err None for attempt in range(MAX_ATTEMPTS): try: resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: MODEL, messages: [{role: user, content: prompt}], temperature: 0, }, timeouttimeout, ) if resp.status_code in RETRIABLE_STATUS: raise requests.HTTPError(fstatus{resp.status_code}) if resp.status_code 400: raise NonRetriable(fstatus{resp.status_code} body{resp.text[:200]}) return resp.json()[choices][0][message][content].strip() except NonRetriable: raise except (requests.Timeout, requests.ConnectionError, requests.HTTPError) as exc: last_err exc if attempt MAX_ATTEMPTS - 1: break delay min(MAX_DELAY, BASE_DELAY * (2 ** attempt)) delay delay * (0.5 random.random() * 0.5) print(f第 {attempt 1} 次失败({exc}){delay:.2f}s 后重试) time.sleep(delay) raise RuntimeError(f重试 {MAX_ATTEMPTS} 次仍失败: {last_err}) if __name__ __main__: print(chat(用一句话说明什么是幂等))抖动那行是关键delay * (0.5 random.random() * 0.5)把等待时间压到理论值的 50% 到 100% 之间相同时刻失败的一批请求会散在一个区间里重试而不是撞在一起。还要提醒一句重试次数不是越多越好。我们线上定的是 4 次超过这个数意味着上游确实不可用了继续试只是在拖长调用方的等待。这时候正确的动作是快速失败并降级比如工单分类失败就落到待人工状态别让用户卡在转圈上。Q4几条业务线共用一个入口怎么知道到底是谁在调结论每条业务线一把独立的 Key加上结构化日志里带业务标识。两个都要有缺一个都查不清。只靠日志不够因为日志里的业务标识是应用自己写的写错了、漏写了、新同事复制老代码没改你查出来的归属就是错的。只靠 Key 也不够Key 只能告诉你哪个应用在调告诉不了你是这个应用里的哪个功能、哪个租户。所以是两层Key 划应用边界日志字段划功能和租户边界。顺带说一个接入前要确认的技术前提这套做法要求平台侧支持一个应用签发多把 Key并且能按 Key 维度回看调用记录否则你的独立 Key 只是个摆设拆了也看不出区别。这一点各家实现差别不小写代码之前先翻文档确认比如 jiekou.vip 近期上线了企业资源包多 Key 签发和按 Key 查调用记录在它的文档里能查到具体字段接入前对一下就行。确认清楚之后剩下的就是把埋点写规范。日志字段我建议至少包含这几个业务线、功能点、租户、模型、耗时、是否成功、重试次数、请求唯一 ID。输出成一行 JSON方便后面直接聚合。import json import os import time import uuid import logging logging.basicConfig(levellogging.INFO, format%(message)s) logger logging.getLogger(ai_call) # 每条业务线一把 Key从环境变量按业务线取 KEY_MAP { ticket: os.environ.get(AI_KEY_TICKET, ), kb_qa: os.environ.get(AI_KEY_KB_QA, ), risk: os.environ.get(AI_KEY_RISK, ), } def resolve_key(biz): key KEY_MAP.get(biz) if not key: raise RuntimeError(f业务线 {biz} 没有配置独立 Key拒绝走公共 Key) return key def traced_call(biz, feature, tenant, prompt): trace_id uuid.uuid4().hex[:12] key resolve_key(biz) t0 time.perf_counter() ok, retries, err True, 0, None try: # 这里换成 Q3 里的 chat()key 传 resolve_key(biz) 的结果 result f模拟返回 {prompt[:12]} except Exception as exc: ok, err, result False, f{type(exc).__name__}: {exc}, None logger.info(json.dumps({ trace_id: trace_id, biz: biz, feature: feature, tenant: tenant, model: os.environ.get(AI_MODEL, gpt-4o-mini), key_tail: key[-4:] if key else , elapsed_ms: round((time.perf_counter() - t0) * 1000, 1), ok: ok, retries: retries, err: err, chars_in: len(prompt), chars_out: len(result or ), }, ensure_asciiFalse)) return result if __name__ __main__: traced_call(ticket, auto_classify, tenant_8812, 门店空调坏了三天没人来修)输出是一行 JSON{trace_id: a3f19c4e77b1, biz: ticket, feature: auto_classify, tenant: tenant_8812, model: gpt-4o-mini, key_tail: 9c2f, elapsed_ms: 0.1, ok: true, retries: 0, chars_in: 14, chars_out: 19, err: null}注意key_tail只写后 4 位完整 Key 绝对不能进日志。这条我见过有团队踩日志系统的权限往往比生产环境宽松得多。另外那个resolve_key里的抛错是故意的。如果找不到业务线对应的 Key 就退回公共 Key半年后你会发现所有调用都挂在公共 Key 上谁也说不清是谁在调。宁可让新接入的业务线在第一次跑就报错逼着他去申请自己的 Key。Q5上线前怎么估我需要多少并发结论不要靠猜也不要靠我们大概有多少用户这种口径。拿小样本实测出耗时分布再用目标请求量和 P95 反推最小并发数然后往上留一倍余量。推导很简单一个 worker 在 P95 耗时下每秒能处理1 / P95个请求要支撑每秒 N 个请求就需要N × P95个并发位。这就是 Littles Law 的实用版本。用 P95 而不是 P50 是因为你要扛的是坏情况用平均值算出来的并发数在高峰期一定不够。举个我们真实做过的知识库问答那条线P95 实测 2.4 秒业务方给的目标是高峰期每秒 12 个请求算出来12 × 2.4 28.8取 29 个并发位留余量到 60。为什么留一倍一是提示词后面会变长耗时会涨二是上游抖动时 P95 会临时翻倍没有余量就直接排队积压。import math import os import time import requests from concurrent.futures import ThreadPoolExecutor BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) SAMPLES [ 把这条工单归类APP 打开就闪退, 把这条工单归类想问一下账号怎么改绑手机号, 把这条工单归类客服小李服务很好想表扬, 把这条工单归类门店空调坏了三天没人来修, ] TARGET_RPS 12.0 # 业务方给的高峰目标请求量 SAFETY 2.0 # 余量倍数 def timed(prompt): t0 time.perf_counter() try: resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{model: MODEL, messages: [{role: user, content: prompt}], temperature: 0}, timeout10, ) resp.raise_for_status() return (time.perf_counter() - t0), True except Exception: return (time.perf_counter() - t0), False def percentile(vals, p): vals sorted(vals) return vals[max(0, min(len(vals) - 1, int(round((len(vals) - 1) * p))))] if __name__ __main__: jobs SAMPLES * 10 with ThreadPoolExecutor(max_workers4) as pool: results list(pool.map(timed, jobs)) durs [d for d, ok in results if ok] fail_rate 1 - len(durs) / len(results) p50, p95 percentile(durs, 0.50), percentile(durs, 0.95) need math.ceil(TARGET_RPS * p95) print(f样本 {len(results)} 条失败率 {fail_rate:.1%}) print(fP50 {p50:.2f}s P95 {p95:.2f}s) print(f支撑 {TARGET_RPS:.0f} rps 的最小并发 {need}) print(f建议连接池 / 线程池上限 {math.ceil(need * SAFETY)})跑出来大致是样本 40 条失败率 0.0% P50 0.71s P95 2.38s 支撑 12 rps 的最小并发 29 建议连接池 / 线程池上限 58一个容易忽略的细节算出来的并发数要同时体现在三个地方——线程池上限、HTTP 连接池上限、以及你在平台侧配置的并发上限。只调线程池不调连接池requests 底层默认的连接池会成为瓶颈你会看到大量请求卡在等连接而不是等模型表现出来是耗时莫名变长但模型侧没异常。用requests.Session挂一个HTTPAdapter(pool_maxsize58)就行。Q6模型返回不稳定、格式不对怎么处理结论不要指望靠提示词把格式说清楚就万事大吉。必须是三层输出强约束、结果强校验、校验失败降级重问。这是我见过上线后最容易出事的一环。开发的时候试了十次都是规整的 JSON上线跑了两万次其中三百多次前面多了一句好的以下是分类结果你的json.loads直接抛异常整条链路挂掉。更隐蔽的情况是 JSON 合法但字段值不在你的枚举里模型自己发明了一个类别这种错不会抛异常会直接污染你的数据。具体做法能用response_format约束成 JSON 就一定要用解析之后必须按业务枚举做一次显式校验不能只判断能不能解析校验失败时不要直接报错而是把错误信息拼回去让模型改一次改一次还不对就落到人工兜底。这里注意重问的次数要卡死在 1 到 2 次不能无限循环否则遇到模型固执的边缘输入会一直转。import json import os import requests BASE_URL os.environ[AI_BASE_URL].rstrip(/) API_KEY os.environ[AI_API_KEY] MODEL os.environ.get(AI_MODEL, gpt-4o-mini) ALLOWED {投诉, 咨询, 表扬} SYSTEM ( 你是工单分类器。只输出 JSON格式为 {category: 投诉|咨询|表扬, confidence: 0.0~1.0, reason: 不超过20字}。 category 必须严格是这三个词之一不要输出其他内容。 ) def post(messages): resp requests.post( f{BASE_URL}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{ model: MODEL, messages: messages, temperature: 0, response_format: {type: json_object}, }, timeout5, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def validate(raw): 返回 (对象, 错误说明)。错误说明非空表示校验失败。 try: obj json.loads(raw) except json.JSONDecodeError as exc: return None, f不是合法 JSON: {exc} if not isinstance(obj, dict): return None, 顶层必须是对象 cat obj.get(category) if cat not in ALLOWED: return None, fcategory 必须是 {/.join(sorted(ALLOWED))} 之一你给的是 {cat!r} conf obj.get(confidence) if not isinstance(conf, (int, float)) or not 0 conf 1: return None, fconfidence 必须是 0~1 的数字你给的是 {conf!r} return obj, def classify(text, max_fix1): messages [ {role: system, content: SYSTEM}, {role: user, content: text}, ] for attempt in range(max_fix 1): raw post(messages) obj, err validate(raw) if not err: return {ok: True, data: obj, fixed: attempt} print(f第 {attempt 1} 次输出不合规{err}) if attempt max_fix: break messages [ {role: assistant, content: raw}, {role: user, content: f输出不符合要求{err}。请只重新输出合规的 JSON。}, ] return {ok: False, data: None, fallback: manual_review} if __name__ __main__: for t in [门店空调坏了三天没人来修, 账号怎么改绑手机号, 客服小李态度特别好]: print(t, -, classify(t))输出门店空调坏了三天没人来修 - {ok: True, data: {category: 投诉, confidence: 0.95, reason: 报修长期未处理}, fixed: 0} 账号怎么改绑手机号 - {ok: True, data: {category: 咨询, confidence: 0.97, reason: 询问改绑操作流程}, fixed: 0} 客服小李态度特别好 - {ok: True, data: {category: 表扬, confidence: 0.96, reason: 肯定服务态度}, fixed: 0}fixed这个字段建议保留并打进日志。它反映了提示词的质量如果线上fixed 0的比例超过百分之几说明你的提示词或者枚举定义有问题该去改提示词而不是靠重问兜着。我们把这个比例当成一个常规看的指标改提示词之后它从 4.1% 掉到 0.3%效果很直观。confidence那个字段要说明一下模型自报的置信度并不可靠别拿它当概率用。但它作为一个粗糙的分流阈值是有用的——低于 0.6 的直接转人工能挡掉相当一部分模糊工单。Q7提示词改了、模型版本换了怎么知道有没有退化结论留一个小规模的回归集每次改动跑一遍比对结果差异。不需要多复杂30 到 50 条就足够发现大部分退化。很多团队在这一步是完全裸奔的改提示词靠手动试三五条觉得好像变好了就上线两周后业务方反馈某类工单全分错了回头查不知道是哪次改动引入的。问题在于人工抽查的样本永远偏向你熟悉的那几条而退化通常发生在边缘输入上。回归集怎么建我的经验是三类样本按比例混一半是典型样本保证主路径不出问题三成是历史上真出过错的样本这部分最有价值每次线上发现一个分错的 case 就补进去回归集会越来越贴合你的实际分布剩下两成放边界样本比如空输入、超长输入、掺杂表情和乱码的输入、语义上确实模棱两可的输入。判断标准要分开定别一刀切。有唯一正确答案的题目分类、抽取、判断直接比对期望值一次成率掉了就是退化这类可以完全自动化。开放生成类的题目摘要、改写没有唯一答案只能做弱校验长度是否在合理区间、是否包含必须出现的关键实体、是否没有出现禁止出现的表述。不要为了追求自动打分去引入一个模型给另一个模型打分的复杂机制那套东西本身也会漂维护它的精力比它省下的多。还要记住一件事回归集只能证明没有明显退化不能证明变好了。变好这件事得看线上真实数据看那些人工修正率、转人工比例的变化看一两周的趋势。回归集的作用是拦住明显的坏改动让你敢改这个价值已经足够大了。实操上建议把回归集放进版本库跟提示词放在一起改提示词的那个提交必须带上回归结果的对比。这样半年后回头查哪次改动导致的翻提交记录就能定位。Q8上线之后怎么持续知道用得好不好结论把 Q4 那份结构化日志按业务线聚合出几个固定指标做成周报自动发出来。不用建复杂的看板一份能定期看到的表格比一个没人打开的仪表盘管用得多。我们固定看六个调用次数、使用人数、失败率、P95 耗时、平均字符数、重试次数。这几个指标的组合能覆盖大部分要判断的事。举两个实际判断的例子调用次数涨但使用人数没涨说明是少数几个人在重复调往往是他们对结果不满意在反复试该去看这条业务线的输出质量失败率不高但 P95 明显变长通常是提示词被人悄悄加长了去 diff 一下提示词就知道。import json import os import statistics from collections import defaultdict LOG_PATH os.environ.get(AI_LOG_PATH, ai_call.log) def percentile(vals, p): vals sorted(vals) return vals[max(0, min(len(vals) - 1, int(round((len(vals) - 1) * p))))] def weekly_report(path): buckets defaultdict(lambda: { calls: 0, fails: 0, users: set(), latencies: [], chars: [], retries: 0, }) with open(path, encodingutf-8) as fh: for line in fh: line line.strip() if not line: continue try: rec json.loads(line) except json.JSONDecodeError: continue b buckets[rec.get(biz, unknown)] b[calls] 1 b[users].add(rec.get(tenant, -)) b[retries] rec.get(retries, 0) if rec.get(ok): b[latencies].append(rec.get(elapsed_ms, 0)) b[chars].append(rec.get(chars_in, 0) rec.get(chars_out, 0)) else: b[fails] 1 header f{业务线:10}{调用次数:8}{使用人数:8}{失败率:8}{P95(ms):10}{均字符:8}{重试:6} print(header) print(- * len(header)) for biz, b in sorted(buckets.items(), keylambda kv: -kv[1][calls]): lat percentile(b[latencies], 0.95) if b[latencies] else 0 chars statistics.mean(b[chars]) if b[chars] else 0 print(f{biz:10}{b[calls]:8}{len(b[users]):8} f{b[fails] / b[calls]:7.1%}{lat:10.0f}{chars:8.0f}{b[retries]:6}) if __name__ __main__: weekly_report(LOG_PATH)跑一周的日志出来是这样业务线 调用次数 使用人数 失败率 P95(ms) 均字符 重试 -------------------------------------------------------------------- ticket 18432 214 0.4% 1521 186 92 kb_qa 6207 88 1.1% 2386 742 143 risk 1140 12 0.2% 903 231 6这张表我们每周一早上自动发到群里。价值不在于某一周的绝对数字在于连续看几周之后你对正常长什么样有了感觉异常一眼就能看出来。上面这份里 kb_qa 的重试次数偏高143 次重试对应 6207 次调用比例超过 2%比另外两条线高一个数量级后来查出来是它的超时值还是按早期短提示词定的 1.5 秒提示词加长之后大量请求擦着超时线过触发了重试。改超时值之后就正常了——这就是 Q2 那句提示词变了要重新跑一遍采样的由来。指标别贪多。我一开始定了十几个三周之后发现没人看砍到六个反而每周都有人真的去读、去提问题。这 8 个问题里前 3 个决定你的接入会不会在高峰期崩第 4 个决定半年后你还能不能说清楚系统在干什么后 4 个决定这套东西能不能持续演进而不是上线即腐化。顺序不是随便排的也大致是我建议的落地顺序。如果只能挑一件事先做我会选 Q4 的结构化日志。它本身不解决任何问题但没有它后面所有问题你都只能靠猜。