企业内部的 AI 调用入口通常是某个团队先搭起来的后来陆续接进来更多团队。接入方式往往很朴素大家共用同一份配置、同一把凭证各自requests.post一下就完事。现在企业级的AI提效已经是大势所趋不过想使用AI来提升效率最重要的是token资源jiekou.vip 近期上线了企业资源包做企业内部多团队接入的话可顺手了解一下。回到主题上这种写法在团队数量增加后会暴露同一个问题日志里只有POST /chat/completions 200没有任何字段能区分请求来自谁。一旦某周失败率异常开会问一圈每个团队都说自己没改东西——而且他们说的都是真的因为确实没人能证明。这篇是完整实战教程四步解决这个问题给调用加身份标记、按团队隔离凭证、加本地配额防止互相影响、聚合日志出使用情况表。代码可直接运行。阅读前提会 Python用过任意大模型的 HTTP 接口。环境准备全文共用这套依赖和环境变量凭证从环境读取不写进源码pip install requests2.32.3# Windows PowerShell $env:AI_API_BASEhttps://api.example.com/v1 $env:AI_API_KEY_TICKETsk-team-ticket-xxxx $env:AI_API_KEY_KBsk-team-kb-xxxx $env:AI_MODELyour-model-name这套方案有两个平台侧的前提条件接入层要支持给每个团队签发独立凭证并且能按凭证分别查询用量明细。缺了任一项第二步的凭证隔离就做不下去。接入前先在平台文档里核对。国内常见平台如 jiekou.vip提供多凭证管理和按凭证维度的用量查询本文代码按这种形式组织。步骤一给每次调用加上身份标记这是整套改造的地基也是最容易被跳过的一步。核心改动只有一处调用时必须声明自己是谁。# client.py import os import time import uuid import logging import requests API_BASE os.environ[AI_API_BASE].rstrip(/) MODEL os.environ.get(AI_MODEL, your-model-name) log logging.getLogger(ai) RETRYABLE_STATUS {429, 500, 502, 503, 504} # 团队 - 环境变量名。凭证不落代码这里只登记映射关系 TEAM_KEY_ENV { ticket: AI_API_KEY_TICKET, kb: AI_API_KEY_KB, } class AIClient: def __init__(self, timeout30, max_retry3): self.timeout timeout self.max_retry max_retry self.session requests.Session() self.session.trust_env False def _key_for(self, team): env_name TEAM_KEY_ENV.get(team) if not env_name: raise ValueError(f未登记的团队: {team}) key os.environ.get(env_name) if not key: raise RuntimeError(f缺少环境变量 {env_name}) return key def chat(self, prompt, team, task, user, max_tokens1024): team/task/user 三个参数必填故意不给默认值。 trace_id uuid.uuid4().hex[:12] key self._key_for(team) last_err None for attempt in range(self.max_retry): started time.monotonic() try: resp self.session.post( f{API_BASE}/chat/completions, headers{Authorization: fBearer {key}}, json{ model: MODEL, max_tokens: max_tokens, messages: [{role: user, content: prompt}], }, timeoutself.timeout, ) elapsed int((time.monotonic() - started) * 1000) if resp.status_code 429: log.warning( throttled trace%s team%s task%s user%s attempt%d, trace_id, team, task, user, attempt, ) last_err http_429 time.sleep(min(2 ** attempt, 8)) continue if resp.status_code in RETRYABLE_STATUS: last_err fhttp_{resp.status_code} time.sleep(min(2 ** attempt, 8)) continue resp.raise_for_status() text resp.json()[choices][0][message][content] log.info( ok trace%s team%s task%s user%s ms%d chars_in%d chars_out%d retry%d, trace_id, team, task, user, elapsed, len(prompt), len(text), attempt, ) return text except requests.Timeout: last_err timeout time.sleep(min(2 ** attempt, 8)) except requests.RequestException as exc: last_err type(exc).__name__ time.sleep(min(2 ** attempt, 8)) log.error(fail trace%s team%s task%s user%s err%s, trace_id, team, task, user, last_err) raise RuntimeError(fcall failed: {last_err} (trace{trace_id}))两个细节是踩坑之后才定下来的team/task/user不给默认值。第一版给了teamunknown结果一半调用点懒得传日志里 40% 的请求归属是unknown改造等于白做。去掉默认值后漏传直接报错一天就全补齐了。429单独记一行throttled。它和 500 一样会触发重试但含义完全不同一个是自己请求太密一个是上游故障。混在一起记排查时会把限流误判成服务不稳。步骤二按团队隔离凭证共用一把凭证最大的问题不是安全而是无法归属。上游只能按凭证维度统计用量五个团队共用一把统计表上永远只有一行。拆凭证时按团队-环境命名管理后台一眼能看出归属ticket-prod / ticket-staging kb-prod / kb-staging report-prod / report-staging拆完之后有个额外收益TEAM_KEY_ENV里登记的团队天然成了白名单新团队接入必须先登记顺手解决了「到底有多少团队在用」这个问题。还有一个改造前没预料到的好处。某个团队的凭证曾被误提交到内部仓库因为凭证独立处理方式就是单独轮换它其他团队完全不受影响。共用一把的话这种事只能全员改配置。步骤三加本地配额防止一个团队拖垮全局拆完凭证仍有风险某团队写了死循环重试把上游并发额度吃满其他团队一起报错。上游侧配额是第一道防线客户端也该有一层自我约束。一个进程内令牌桶就够# quota.py import threading import time class TokenBucket: 按团队限制每秒请求数。超出直接拒绝不排队。 def __init__(self, rate_per_sec, burstNone): self.rate float(rate_per_sec) self.capacity float(burst if burst is not None else rate_per_sec) self._tokens self.capacity self._last time.monotonic() self._lock threading.Lock() def try_acquire(self): with self._lock: now time.monotonic() self._tokens min( self.capacity, self._tokens (now - self._last) * self.rate, ) self._last now if self._tokens 1.0: self._tokens - 1.0 return True return False # 数字来自各团队自报的峰值 QPS 上浮一倍 BUCKETS { ticket: TokenBucket(20, burst40), kb: TokenBucket(8, burst16), report: TokenBucket(2, burst4), } class QuotaExceeded(RuntimeError): pass def guard(team): bucket BUCKETS.get(team) if bucket is None: raise ValueError(f未配置配额的团队: {team}) if not bucket.try_acquire(): raise QuotaExceeded(fteam{team} 超出本地配额)接进调用路径发请求前先拦一道from quota import guard, QuotaExceeded def chat_guarded(client, prompt, team, task, user, **kw): try: guard(team) except QuotaExceeded: log.warning(quota_reject team%s task%s user%s, team, task, user) raise return client.chat(prompt, teamteam, tasktask, useruser, **kw)两点经验超限就拒绝不要排队。第一版做成阻塞等待某团队突发流量时全部请求堵在队列里调用方看到的是「所有请求都变慢了」比直接报错更难排查。改成立即拒绝并记quota_reject后问题在日志里一目了然。配额数字要写明来历。上面那句「自报峰值上浮一倍」的注释很关键。半年后有人问为什么kb是 8没这行注释答案就只能是「不知道一直是 8」。步骤四聚合日志定期回看前面三步记的字段到这一步才产生价值# aggregate.py import re import sys from collections import defaultdict PAT re.compile( r(?Plvlok|fail|quota_reject|throttled) trace(?Ptrace\w)?\s* rteam(?Pteam\S) task(?Ptask\S) user(?Puser\S) ) MS re.compile(rms(\d)) def main(path): agg defaultdict( lambda: {n: 0, fail: 0, rej: 0, thr: 0, ms: [], users: set()} ) for line in open(path, encodingutf-8): m PAT.search(line) if not m: continue rec agg[(m.group(team), m.group(task))] lvl m.group(lvl) rec[users].add(m.group(user)) if lvl quota_reject: rec[rej] 1 continue if lvl throttled: rec[thr] 1 continue rec[n] 1 if lvl fail: rec[fail] 1 continue hit MS.search(line) if hit: rec[ms].append(int(hit.group(1))) print(f{团队:10}{任务:16}{请求:7}{失败率:8} f{中位ms:8}{人数:6}{限流:6}{超配额:7}) for (team, task), r in sorted(agg.items(), keylambda kv: -kv[1][n]): med sorted(r[ms])[len(r[ms]) // 2] if r[ms] else 0 rate (r[fail] / r[n] * 100) if r[n] else 0.0 print(f{team:10}{task:16}{r[n]:7}{rate:7.1f}% f{med:8}{len(r[users]):6}{r[thr]:6}{r[rej]:7}) idle [k for k, r in agg.items() if r[n] 0] if idle: print(\n本周无成功请求:, , .join(f{t}/{k} for t, k in idle)) if __name__ __main__: main(sys.argv[1] if len(sys.argv) 1 else ai.log)第一周的实测输出团队 任务 请求 失败率 中位ms 人数 限流 超配额 ticket classify 21400 0.3% 94 46 0 0 ticket shorten_title 14120 0.4% 121 46 0 0 kb kb_answer 11960 3.8% 2910 83 742 118 report weekly_draft 2240 0.6% 3580 24 0 0 本周无成功请求: crm/entity_extractkb那一行的 742 次限流和 118 次超配额直接解答了当周失败率异常的原因该团队上线了一个批量重建索引的任务把请求打成了脉冲。之前查不出来不是因为难查而是日志里根本没有team字段。最后一行「本周无成功请求」只有三行代码却是最常被用到的信息crm那个入口几个月前接入过之后没人再用据此可以安全下掉。小结完整顺序给调用加身份 → 按团队拆凭证 → 加配额防止互相影响 → 聚合表定期回看。第一步最简单也最关键。team/task/user三个字段只是在调用点多传几个参数但没有它们后面三步全都无从判断。上一篇写了按任务给模型分级的路由层两篇可以配合使用分级决定一次请求走哪个模型身份标记决定这次请求算在谁头上。