这是团队AI客服项目上线90天后的复盘记录。没有完美的架构，只有在踩坑中不断迭代的工程。本文记录了模型选型、中转站接入、成本优化、稳定性保障四个维度的真实决策过程和教训。

一、项目背景

我们的产品是一个面向C端用户的智能客服系统，日均请求量约5万次。核心需求：

• 准确理解用户问题（家电使用咨询、售后流程、订单查询）
• 多轮对话能力，能结合上下文回答
• 响应延迟控制在2秒以内
• 月度API成本控制在5000元以内

技术栈：Python + FastAPI + PostgreSQL + Redis。最初版本直连OpenAI API，上线两周后暴露了一系列问题，开始了为期一个月的架构重构。

二、坑1：直连海外API的网络稳定性

问题

上线第一周，用户反馈"客服偶尔不回话"。查日志发现是OpenAI API请求超时——国内服务器直连api.openai.com的延迟在800ms-3000ms之间波动，高峰期超时率超过5%。

决策过程

当时有三个方案：

方案	优点	缺点
海外服务器部署	网络稳定	运维成本高，数据出境合规问题
自建代理服务器	可控性强	需要维护代理节点，仍有网络波动
接入中转站	开箱即用，多通道冗余	增加一层依赖

最终选择了接入中转站，原因是：不需要自建基础设施，中转站自带多通道负载均衡，比单点代理更稳定。

接入过程

我们对比了几个主流中转站后，选择了多接入方案（主备模式）：

• 主通道：魔芋AI（moyu.info）—— 国内中转站，延迟稳定在200-400ms，新用户有免费额度可以测试
• 备用通道：OpenRouter（openrouter.ai）—— 国际中转站，模型覆盖最全，作为降级方案
• 降级通道：硅基流动（siliconflow.cn）—— 国内中转站，开源模型为主，价格便宜

主备切换逻辑很简单——主通道超时或报错时自动切换到备用通道：

python

import asyncio
from openai import AsyncOpenAI

class MultiProviderClient:
    def __init__(self):
        # 多中转站配置，主备模式
        self.providers = [
            {
                "name": "魔芋AI",
                "client": AsyncOpenAI(
                    api_key="key-1",
                    base_url="https://api.moyu.info/v1"
                ),
                "priority": 1
            },
            {
                "name": "OpenRouter",
                "client": AsyncOpenAI(
                    api_key="key-2",
                    base_url="https://openrouter.ai/api/v1"
                ),
                "priority": 2
            },
            {
                "name": "硅基流动",
                "client": AsyncOpenAI(
                    api_key="key-3",
                    base_url="https://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(
                        model=model, messages=messages, **kwargs
                    ),
                    timeout=10
                )
                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回答"怎么开机"，属于杀鸡用牛刀。

决策：模型分级路由

按请求复杂度分配不同模型：

python

def 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）中途断开，原因有三：

1. Nginx缓冲：Nginx默认缓冲SSE流，导致数据积压后一次性发送，客户端以为超时断开了
2. 中转站超时：部分中转站对SSE连接有30秒超时限制，长回复会被截断
3. 客户端断线重连：移动端网络切换时连接断开，但没有续传机制

解决方案

Nginx配置（关闭SSE缓冲）：

nginx

location /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;       # 延长超时
}

断线续传（客户端保存已接收内容，重连时从断点继续）：

python

async 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(
        model=model, messages=messages, stream=True
    )
    
    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

五、坑4：Token计费对不上账

问题

财务对账时发现，我们自己统计的Token消耗和中转站账单对不上，差异约8%。

排查

1. Tokenizer差异：我们用tiktoken统计GPT-4o的Token，但Claude用的是Anthropic自己的Tokenizer，统计结果有差异
2. 流式Token统计遗漏：流式响应默认不返回usage，我们之前的统计是按字符数估算的，偏差大
3. 缓存Token：Anthropic的Prompt Cache，缓存读取只收10%费用，但我们按全价统计了

解决方案

开启stream_options={"include_usage": True}获取准确Token数：

python

stream = await client.chat.completions.create(
    model=model,
    messages=messages,
    stream=True,
    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限流。

解决方案

客户端令牌桶限流：

python

import 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(rate=50, capacity=100)

async def rate_limited_chat(client, **kwargs):
    while not await bucket.acquire():
        await asyncio.sleep(0.01)
    return await client.chat.completions.create(**kwargs)

请求队列化：超过限流能力的请求进入队列排队，而非直接报错。

七、坑6：System Prompt的隐藏成本

问题

每个请求都带了一段800 Token的System Prompt（产品知识库+服务规范），按5万次/天计算，光System Prompt的输入Token成本就占了总成本的40%。

解决方案：Prompt Cache

Anthropic的Prompt Cache可以缓存重复的System Prompt，缓存命中时只收10%费用：

python

messages = [
    {
        "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小时我们才知道。

解决方案

建立了四层监控：

1. 请求级监控：每个API调用的延迟、状态码、Token数
2. 聚合监控：每分钟的QPS、错误率、平均延迟、P99延迟
3. 成本监控：按小时统计Token消耗和成本，超预算告警
4. 通道健康监控：定时探测各中转站的可用性，故障自动告警

python

import 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_in=0, tokens_out=0):
        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.8s	1.2s
月度API成本	12000元	4800元
用户满意度	72%	89%
故障发现时间	靠投诉（2h+）	监控告警（30s内）

十、总结

90天踩了7个坑，每个坑背后都是一个工程决策的迭代。核心经验：

1. 不要直连海外API，用中转站解决网络问题，且要配多通道主备
2. 模型分级路由是成本优化的关键，简单问题不要用贵模型
3. 流式响应要做断线续传，移动端网络不稳定是常态
4. Token统计要以API返回的usage为准，自己估算必然有偏差
5. 监控比优化更重要，没有可观测性的系统就是在裸奔

如果刚开始做AI应用，建议先接一个中转站跑起来（魔芋AI、OpenRouter、硅基流动都可以），在实战中踩坑和迭代，比纸上谈兵有用得多。