构建统一AI入口：千问/ChatGPT/豆包/文心一言集成架构设计

2次阅读

共计 1899 个字符，预计需要花费 5 分钟才能阅读完成。

企业同时接入多个 AI 服务时（如千问、ChatGPT、豆包、文心一言），通常会面临以下典型问题：

接口碎片化 ：每个厂商的 API 协议不同（JSON 结构、鉴权方式、错误码体系）
会话割裂 ：对话历史无法跨模型共享，导致上下文连贯性被破坏
维护成本高 ：需要为每个 AI 服务单独开发适配层，代码重复率超过 60%
流量分配不灵活 ：无法根据业务场景动态调整模型调用权重

（注：此为示意图，实际部署需考虑高可用方案）

协议适配层
转换各厂商 API 到统一内部协议
实现签名生成、错误码映射等脏活
路由决策引擎
基于成本、时延、准确率等维度打分
支持 AB 测试分流和故障自动切换
结果标准化模块
统一输出格式：{code, data, msg} 结构
敏感信息过滤和内容安全审查

方案	适用场景	QPS 基准测试	开发复杂度
REST	简单查询类请求	2k	★★☆
gRPC	流式对话 / 大数据量传输	15k	★★★
WebSocket	长会话保活	8k	★★☆

建议组合使用 ：普通问答用 REST，语音 / 长文本用 gRPC 流式传输

class Router:
    def __init__(self):
        # 模型权重配置（可热更新）self.weights = {
            'qianwen': 0.4, 
            'chatgpt': 0.3,
            'wenxin': 0.3
        }
        self.circuit_breaker = {}  # 熔断状态记录

    async def select_model(self, query):
        # 排除已熔断的模型
        available = [m for m in self.weights 
                    if not self.circuit_breaker.get(m, False)]

        # 按权重随机选择（可替换为更智能的算法）chosen = random.choices(
            available, 
            weights=[self.weights[m] for m in available]
        )[0]

        # 记录选择结果用于后期调优
        log_analytics(query, chosen)  
        return chosen

// 全局连接池（支持并发安全访问）type ModelPool struct {
    sync.Mutex
    pools map[string]*Pool  // 各模型独立连接池
}

// 获取连接（带超时控制）func (p *ModelPool) GetConn(model string) (*Conn, error) {p.Lock()
    defer p.Unlock()

    pool, exists := p.pools[model]
    if !exists {return nil, fmt.Errorf("invalid model")
    }

    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    conn, err := pool.Get(ctx)
    if err != nil {metrics.RecordError(model) // 监控打点
        return nil, err
    }
    return conn.(*Conn), nil
}

方案	平均延迟	99 分位延迟	单节点 QPS
直连各厂商 API	320ms	890ms	1,200
统一入口	280ms	650ms	3,800

测试环境：4 核 8G 云主机，混合负载（70% 短文本 +30% 长对话）

鉴权方案 ：
使用 JWT 携带租户 ID 和权限声明
每次请求验证签名 + 有效期 + 黑名单

限流策略 ：

limit_req_zone $binary_remote_addr zone=ai_api:10m rate=100r/s;
location /v1/chat {
    limit_req zone=ai_api burst=50;
    proxy_pass http://ai_gateway;
}

典型场景 ：
– 用户切换模型时历史对话未迁移
– 异步响应时关联 ID 匹配失败

解决方案 ：
1. 设计全局 Session 服务统一存储对话历史
2. 使用唯一 trace_id 串联整个调用链

推荐采用双超时机制：
1. 客户端设置 5 秒短超时（快速失败）
2. 服务端后台继续处理，结果存入缓存
3. 客户端可轮询或等待服务端推送

当前架构已解决基础接入问题，但更高级的场景可能需要：
– 知识融合 ：跨模型结果投票 / 加权聚合
– 意图识别 ：先判断问题类型再路由到最擅长该领域的模型
– 成本优化 ：根据查询复杂度动态选择性价比最优模型

欢迎在评论区分享你的模型调度策略设计经验！

正文完