这是团队AI客服项目上线90天后的复盘记录。没有完美的架构只有在踩坑中不断迭代的工程。本文记录了模型选型、中转站接入、成本优化、稳定性保障四个维度的真实决策过程和教训。一、项目背景我们的产品是一个面向C端用户的智能客服系统日均请求量约5万次。核心需求准确理解用户问题家电使用咨询、售后流程、订单查询多轮对话能力能结合上下文回答响应延迟控制在2秒以内月度API成本控制在5000元以内技术栈Python FastAPI PostgreSQL Redis。最初版本直连OpenAI API上线两周后暴露了一系列问题开始了为期一个月的架构重构。二、坑1直连海外API的网络稳定性问题上线第一周用户反馈客服偶尔不回话。查日志发现是OpenAI API请求超时——国内服务器直连api.openai.com的延迟在800ms-3000ms之间波动高峰期超时率超过5%。决策过程当时有三个方案方案优点缺点海外服务器部署网络稳定运维成本高数据出境合规问题自建代理服务器可控性强需要维护代理节点仍有网络波动接入中转站开箱即用多通道冗余增加一层依赖最终选择了接入中转站原因是不需要自建基础设施中转站自带多通道负载均衡比单点代理更稳定。接入过程我们对比了几个主流中转站后选择了多接入方案主备模式主通道魔芋AImoyu.info—— 国内中转站延迟稳定在200-400ms新用户有免费额度可以测试备用通道OpenRouteropenrouter.ai—— 国际中转站模型覆盖最全作为降级方案降级通道硅基流动siliconflow.cn—— 国内中转站开源模型为主价格便宜主备切换逻辑很简单——主通道超时或报错时自动切换到备用通道pythonimport asyncio from openai import AsyncOpenAI class MultiProviderClient: def __init__(self): # 多中转站配置主备模式 self.providers [ { name: 魔芋AI, client: AsyncOpenAI( api_keykey-1, base_urlhttps://api.moyu.info/v1 ), priority: 1 }, { name: OpenRouter, client: AsyncOpenAI( api_keykey-2, base_urlhttps://openrouter.ai/api/v1 ), priority: 2 }, { name: 硅基流动, client: AsyncOpenAI( api_keykey-3, base_urlhttps://api.siliconflow.cn/v1 ), priority: 3 } ] async def chat(self, model, messages, **kwargs): 带自动降级的聊天调用 for provider in self.providers: try: response await asyncio.wait_for( provider[client].chat.completions.create( modelmodel, messagesmessages, **kwargs ), timeout10 ) return response except Exception as e: print(f[{provider[name]}] 失败: {e}) continue raise Exception(所有通道均失败)效果接入中转站后API超时率从5%降到0.3%以下。主通道魔芋AI承担了95%的流量备用通道只在主通道偶发故障时触发。三、坑2模型选型与成本失控问题最初全量使用GPT-4o月度API成本达到了12000元远超5000元预算。分析请求分布后发现60%的请求是简单的使用咨询怎么开机保修多久25%是中等复杂度的售后流程咨询15%是复杂的多轮对话或投诉处理用GPT-4o回答怎么开机属于杀鸡用牛刀。决策模型分级路由按请求复杂度分配不同模型pythondef select_model(messages, user_input): 根据请求复杂度选择模型 # 简单问题用便宜模型 simple_keywords [怎么开机, 保修, 说明书, 电话, 地址] if any(kw in user_input for kw in simple_keywords): return gpt-4o-mini # $0.15/1M tokens # 代码/技术问题用擅长代码的模型 if any(kw in user_input for kw in [代码, 报错, 故障码]): return claude-3.5-sonnet # $3/1M tokens # 复杂多轮对话用强模型 if len(messages) 6: # 超过3轮对话 return gpt-4o # $2.5/1M tokens # 默认用中等模型 return gpt-4o-mini效果模型分级后成本分布变成GPT-4o-mini承担60%请求成本占比15%Claude 3.5 Sonnet承担10%请求成本占比30%GPT-4o承担30%请求成本占比55%月度成本从12000元降到4800元降幅60%且用户满意度没有下降简单问题用mini回答质量足够。四、坑3流式响应的中断问题问题用户反馈客服说到一半就断了。排查发现是流式响应SSE中途断开原因有三Nginx缓冲Nginx默认缓冲SSE流导致数据积压后一次性发送客户端以为超时断开了中转站超时部分中转站对SSE连接有30秒超时限制长回复会被截断客户端断线重连移动端网络切换时连接断开但没有续传机制解决方案Nginx配置关闭SSE缓冲nginxlocation /chat/stream { proxy_pass http://backend; proxy_buffering off; # 关键关闭缓冲 proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding on; proxy_read_timeout 300s; # 延长超时 }断线续传客户端保存已接收内容重连时从断点继续pythonasync def stream_with_resume(client, model, messages, collected): 带续传的流式请求 if collected: messages messages [ {role: assistant, content: collected}, {role: user, content: 请继续} ] stream await client.chat.completions.create( modelmodel, messagesmessages, streamTrue ) try: async for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: content chunk.choices[0].delta.content collected content yield content except Exception as e: # 断流返回已收集的内容 print(f流中断: {e}) return collected return collected五、坑4Token计费对不上账问题财务对账时发现我们自己统计的Token消耗和中转站账单对不上差异约8%。排查Tokenizer差异我们用tiktoken统计GPT-4o的Token但Claude用的是Anthropic自己的Tokenizer统计结果有差异流式Token统计遗漏流式响应默认不返回usage我们之前的统计是按字符数估算的偏差大缓存TokenAnthropic的Prompt Cache缓存读取只收10%费用但我们按全价统计了解决方案开启stream_options{include_usage: True}获取准确Token数pythonstream await client.chat.completions.create( modelmodel, messagesmessages, streamTrue, stream_options{include_usage: True} # 让API返回准确usage ) total_input 0 total_output 0 async for chunk in stream: if chunk.usage: total_input chunk.usage.prompt_tokens total_output chunk.usage.completion_tokens统计口径以中转站返回的usage为准不再自己估算。对账差异降到1%以内。六、坑5高并发下的限流问题促销活动期间并发请求量从50 QPS飙升到300 QPS开始频繁出现429限流。解决方案客户端令牌桶限流pythonimport time import asyncio class TokenBucket: def __init__(self, rate, capacity): self.rate rate # 每秒生成令牌数 self.capacity capacity # 桶容量 self.tokens capacity self.last_update time.time() self.lock asyncio.Lock() async def acquire(self): async with self.lock: now time.time() elapsed now - self.last_update self.tokens min(self.capacity, self.tokens elapsed * self.rate) self.last_update now if self.tokens 1: self.tokens - 1 return True return False # 限流到50 QPS匹配中转站的速率限制 bucket TokenBucket(rate50, capacity100) async def rate_limited_chat(client, **kwargs): while not await bucket.acquire(): await asyncio.sleep(0.01) return await client.chat.completions.create(**kwargs)请求队列化超过限流能力的请求进入队列排队而非直接报错。七、坑6System Prompt的隐藏成本问题每个请求都带了一段800 Token的System Prompt产品知识库服务规范按5万次/天计算光System Prompt的输入Token成本就占了总成本的40%。解决方案Prompt CacheAnthropic的Prompt Cache可以缓存重复的System Prompt缓存命中时只收10%费用pythonmessages [ { role: system, content: [ { type: text, text: 【产品知识库和服务规范约800 tokens】..., cache_control: {type: ephemeral} # 启用缓存 } ] }, {role: user, content: user_input} ]效果System Prompt的计费Token从800降到80缓存命中单这一项就省了35%的成本。八、坑7监控盲区问题上线初期没有做好监控出问题全靠用户投诉发现。有一次中转站故障了2小时我们才知道。解决方案建立了四层监控请求级监控每个API调用的延迟、状态码、Token数聚合监控每分钟的QPS、错误率、平均延迟、P99延迟成本监控按小时统计Token消耗和成本超预算告警通道健康监控定时探测各中转站的可用性故障自动告警pythonimport time from collections import defaultdict class APIMonitor: def __init__(self): self.stats defaultdict(lambda: { count: 0, errors: 0, latency_sum: 0, tokens_in: 0, tokens_out: 0 }) def record(self, provider, model, latency, status, tokens_in0, tokens_out0): s self.stats[f{provider}/{model}] s[count] 1 if status 400: s[errors] 1 s[latency_sum] latency s[tokens_in] tokens_in s[tokens_out] tokens_out def health_check(self): 健康检查返回各通道的成功率 report {} for key, s in self.stats.items(): if s[count] 0: report[key] { success_rate: 1 - s[errors] / s[count], avg_latency: s[latency_sum] / s[count], total_tokens: s[tokens_in] s[tokens_out] } return report九、90天数据回顾指标上线初期90天后API超时率5%0.3%平均响应延迟2.8s1.2s月度API成本12000元4800元用户满意度72%89%故障发现时间靠投诉2h监控告警30s内十、总结90天踩了7个坑每个坑背后都是一个工程决策的迭代。核心经验不要直连海外API用中转站解决网络问题且要配多通道主备模型分级路由是成本优化的关键简单问题不要用贵模型流式响应要做断线续传移动端网络不稳定是常态Token统计要以API返回的usage为准自己估算必然有偏差监控比优化更重要没有可观测性的系统就是在裸奔如果刚开始做AI应用建议先接一个中转站跑起来魔芋AI、OpenRouter、硅基流动都可以在实战中踩坑和迭代比纸上谈兵有用得多。