☰
AI应用生产级架构实战:多Provider路由、RAG检索优化与Agent编排
2026/9/28 8:53:36 网站建设 项目流程

1. 从单点调用到多 Provider 路由:为什么不能把模型写死

做过 AI 应用的人大概都经历过这个阶段:第一版代码里,模型名是硬编码的,API Key 是写死在配置文件里的,调用逻辑就是一次 HTTP 请求加一层 try-catch。跑 Demo 没问题,一旦要上线、要换模型、要做灰度,整个调用层就得推倒重来。

我见过太多项目卡在这一步。业务侧说“DeepSeek 便宜,先切过去试试”,工程侧打开代码一看,模型名散落在十几个文件里,base_url 和 api_key 的读取逻辑各写各的,切一次要改半天还得回归测试。这不是技术难题,是架构债。

多 Provider 切换要解决的核心问题其实就三个:统一调用契约、运行时动态路由、故障自动降级。听起来像微服务那套东西,本质上确实就是——把大模型当成一个不稳定的外部依赖来对待。

1.1 统一调用契约:把不同厂商的差异吃掉

不同厂商的 API 长得不一样。有的用messages数组,有的用prompt字符串;有的返回choices[0].message.content,有的返回output.text;流式返回的 SSE 格式也各有各的写法。如果业务代码直接对接这些差异,那每接一家就是一次重构。

正确的做法是在 Provider 层做一层适配,对外暴露统一的接口。我通常定义这样一个抽象:

class BaseProvider: def chat(self, messages: list, **kwargs) -> ChatResponse: raise NotImplementedError def stream_chat(self, messages: list, **kwargs): raise NotImplementedError def embed(self, texts: list[str]) -> list[list[float]]: raise NotImplementedError

每个具体 Provider 继承这个基类,把自家 API 的请求格式、响应解析、错误码映射全部封装在内部。业务层只认chat和stream_chat,不关心背后是哪个模型。

这里有个容易忽略的细节:错误码的归一化。不同厂商对“限流”的返回码不一样,有的是 429,有的是自定义的 error code。如果不做归一化,上层做重试和降级时就得写一堆 if-else。我的做法是定义一组内部错误类型,比如RateLimitError、AuthError、ContextLengthError、ProviderUnavailableError,每个 Provider 负责把自家错误映射过来。

1.2 运行时路由:配置驱动而不是代码驱动

Provider 的选择不应该写死在代码里。我习惯用一份配置来描述路由规则:

providers: deepseek: type: openai_compatible base_url: https://api.deepseek.com/v1 api_key: ${DEEPSEEK_API_KEY} models: [deepseek-chat, deepseek-reasoner] qwen: type: openai_compatible base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 api_key: ${QWEN_API_KEY} models: [qwen-max, qwen-plus] routing: default: deepseek/deepseek-chat rules: - match: {task: "reasoning"} target: deepseek/deepseek-reasoner - match: {task: "long_context"} target: qwen/qwen-max fallback: [qwen/qwen-plus, deepseek/deepseek-chat]

这份配置的价值在于:换模型不用改代码,改配置重启即可;灰度新模型只需要调整 routing 规则;某个 Provider 挂了,fallback 链自动接管。

注意:配置里的 api_key 一定要走环境变量注入,不要明文写在 YAML 里。我见过有人把 key 提交到 Git 仓库,第二天就收到了异常调用账单。

1.3 降级策略:什么时候该切,什么时候不该切

降级不是无脑重试。有些错误重试有用(网络抖动、临时限流),有些错误重试一万次也没用(认证失败、参数错误、上下文超长)。我的经验是分三类处理:

错误类型处理策略是否切换 Provider
网络超时、连接失败同 Provider 重试 2 次重试失败后切换
429 限流指数退避重试连续 3 次后切换
401/403 认证失败不重试,直接报错切换(可能是 key 失效)
上下文超长不重试切换到大上下文模型
参数格式错误不重试不切换,这是代码 bug

这套策略落地后,线上因为单个 Provider 抖动导致的服务不可用基本消失了。实测下来,加了 fallback 链之后,端到端成功率从 97% 左右提到了 99.5% 以上。

2. RAG 知识库:从“能检索”到“检索得准”

RAG 这个词现在被用烂了,好像只要把文档切块、向量化、存进向量库就叫 RAG。但真正跑过生产环境的人都知道,RAG 的难点从来不是搭起来,而是检索命中率。用户问一个问题,你召回了 5 个片段,其中 3 个是无关的,模型基于这些噪声生成答案,结果就是胡说八道。

2.1 切块策略:固定长度是最偷懒也最坑的做法

很多人上手 RAG 的第一版代码就是RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50),然后发现检索效果一塌糊涂。问题出在哪?固定长度切块会把一个完整的语义单元拦腰截断。比如一段讲“配置 base_url”的说明,前半段在 chunk 3,后半段在 chunk 4,用户搜“base_url 怎么配”,可能只召回 chunk 3,信息不完整。

我的做法是按文档结构切,而不是按字符数切。Markdown 按标题层级切,代码文档按函数/类切,PDF 按段落和表格边界切。切完之后再做一层“语义合并”:如果相邻两个 chunk 的向量相似度超过阈值,说明它们讲的是同一件事,合并成一个。

def semantic_merge(chunks, embed_fn, threshold=0.85): merged = [chunks[0]] for chunk in chunks[1:]: sim = cosine_sim(embed_fn(merged[-1]), embed_fn(chunk)) if sim > threshold: merged[-1] += "\n" + chunk else: merged.append(chunk) return merged

这个逻辑不复杂,但效果提升很明显。我在一个技术文档库上测过,语义合并后检索命中率从 62% 提到了 81%。

2.2 混合检索:向量不是万能的

纯向量检索有个致命问题:它对精确匹配不敏感。用户搜“error code 4001”,向量检索可能返回一堆讲“错误处理”的片段,但就是找不到那个具体的 4001。因为向量空间里,“4001”和“4002”几乎没区别。

解决办法是混合检索——向量检索 + 关键词检索(BM25),然后做融合排序。我通常用 RRF(Reciprocal Rank Fusion)来合并两路结果:

def rrf_fusion(vector_results, bm25_results, k=60): scores = {} for rank, doc in enumerate(vector_results): scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank) for rank, doc in enumerate(bm25_results): scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + rank) return sorted(scores.items(), key=lambda x: -x[1])

RRF 的好处是不需要调权重,两路结果直接按排名融合,工程上很省心。实测在技术问答场景下,混合检索比纯向量检索的 Top-5 命中率高出 20 个百分点以上。

2.3 重排序:最后一道精度闸门

召回阶段追求的是“不漏”,重排序阶段追求的是“精准”。我一般先用混合检索召回 Top-20,再用一个 Cross-Encoder 重排序模型精排出 Top-5 送给大模型。

重排序模型的选择上,如果追求效果可以用 bge-reranker 系列,如果追求速度可以用轻量级的。这里有个经验:重排序的收益在召回质量差的时候特别明显,但如果召回本身就很准,重排序的提升有限。所以不要一上来就堆重排序,先把切块和混合检索做好。

提示:重排序模型和嵌入模型最好来自同一家族,这样语义空间一致,效果更稳。混用不同厂商的模型有时候会有意想不到的偏差。

2.4 知识库更新的工程问题

RAG 不是建一次就完事的。文档会更新,旧内容会失效。如果每次更新都全量重建索引,成本高且没必要。我的做法是给每个 chunk 打上doc_id和version标签,更新时只重建变化的文档对应的 chunk,删除时按doc_id批量清理。

这里有个坑:向量库的删除操作往往比插入慢。如果文档更新频繁,建议用“软删除 + 定期压缩”的策略,而不是每次实时删除。

3. Agent 编排:让模型自己决定下一步做什么

Agent 和普通 Chain 的区别在于:Chain 的流程是人写死的,Agent 的流程是模型在运行时决定的。这个区别听起来很酷,但落地的时候会发现——模型经常做出愚蠢的决定。所以 Agent 编排的核心不是“让模型自由发挥”,而是“在可控的范围内给模型选择权”。

3.1 工具设计:粒度比数量重要

新手做 Agent 最容易犯的错是工具给太多。一口气注册二十个工具,模型在选工具的时候就开始犯迷糊,要么选错,要么反复调用同一个工具。我的经验是:单个 Agent 的工具数量控制在 5-8 个,超过这个数就应该拆分成多个 Agent。

工具的描述也很关键。不要写“查询数据库”这种模糊描述,要写清楚“根据用户 ID 查询订单状态,输入参数为 user_id(字符串),返回订单列表”。模型是靠描述来选工具的,描述越具体,选错的概率越低。

tools = [ { "name": "search_knowledge_base", "description": "在技术文档知识库中检索相关内容。适用于回答产品功能、配置方法、错误排查类问题。输入为自然语言查询语句。", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "检索查询语句"} }, "required": ["query"] } } ]

3.2 编排模式:ReAct 不是唯一选择

提到 Agent 编排,很多人第一反应就是 ReAct(Reasoning + Acting)。ReAct 确实通用,但它不是万能的。我实际用下来,不同场景适合不同的编排模式:

编排模式适用场景优点缺点
ReAct开放式任务,步骤不确定灵活容易绕圈,token 消耗大
Plan-and-Execute复杂任务,步骤可预规划全局视角好计划可能脱离实际
Workflow流程固定的业务稳定可控不灵活
Multi-Agent需要多角色协作分工明确通信开销大

我的建议是:能用 Workflow 就别用 Agent,能用单 Agent 就别用 Multi-Agent。每增加一层自主性,就增加一层不确定性。很多所谓的“Agent 需求”,拆开看其实就是几个固定的步骤,用 Workflow 编排反而更稳。

3.3 状态管理与上下文控制

Agent 执行过程中会产生大量中间状态:思考过程、工具调用记录、工具返回结果。如果不加控制,上下文会迅速膨胀,最后要么超长报错,要么模型被无关信息干扰。

我的做法是分层管理上下文:

  • 短期记忆:当前任务的完整对话历史,保留最近 N 轮
  • 工作记忆:当前步骤的工具调用结果,用完即弃
  • 长期记忆:跨会话的重要信息,存向量库,按需检索

工具返回的结果不要原封不动塞回上下文。比如搜索返回了 10 个片段,先做一次摘要压缩,只把最相关的 2-3 句放回去。这一步能省大量 token,也能提升模型判断的准确率。

3.4 失败处理:Agent 卡住了怎么办

Agent 最常见的失败模式是“死循环”——反复调用同一个工具,或者在一个步骤上反复纠结。我一般设三个保险:

  1. 最大步数限制:超过 N 步强制终止,返回当前最优结果
  2. 重复检测:如果连续两次工具调用参数完全相同,中断并提示模型换策略
  3. 超时控制:单次任务总时长超过阈值就终止

这些限制看起来简单,但能避免 90% 的线上事故。我踩过一次坑:一个 Agent 因为工具返回格式异常,陷入了无限重试,半小时烧掉了几百万 token。从那以后,所有 Agent 都必须带步数和超时限制。

4. 三者的协同:Provider、RAG、Agent 怎么串起来

单独看 Provider 切换、RAG、Agent 编排,每个都不算太难。但真正做项目的时候,难点在于把它们串成一个整体,还要保证稳定性和可观测性。

4.1 调用链路的分层设计

我习惯把整个系统分成四层:

  • 接入层:处理用户请求,做鉴权和限流
  • 编排层:Agent 逻辑,决定调用哪些工具、走什么流程
  • 能力层:RAG 检索、工具执行、Provider 调用
  • 基础设施层:向量库、缓存、日志、监控

分层的价值在于:每层可以独立替换和测试。比如换一个向量库,只动基础设施层;换一个模型 Provider,只动能力层的 Provider 适配器。编排层的逻辑不受影响。

4.2 可观测性:出了问题要能定位

AI 应用最头疼的就是“效果不好”,但你不知道是哪一步出了问题。是检索没召回?是模型理解错了?还是工具返回了脏数据?

我的做法是全链路打点,每次请求记录:

  • 用户原始输入
  • 检索召回的 chunk 列表及分数
  • 重排序后的结果
  • 送给模型的完整 prompt
  • 模型的原始输出
  • 工具调用记录及返回
  • 最终响应
  • 各阶段耗时

这些数据存下来,出问题的时候可以完整回放。我一般还会做一个“检索质量看板”,统计每天的召回命中率、重排序提升幅度、Agent 平均步数等指标。数据一掉,马上能发现。

4.3 成本控制:别让 Agent 变成烧钱机器

Agent 的 token 消耗是普通对话的几倍甚至几十倍,因为每一步都要带上完整上下文。如果不加控制,成本会失控。几个实用的省钱技巧:

  • 缓存:相同或相似的查询直接返回缓存结果,尤其是 RAG 检索结果
  • 模型分级:简单任务用小模型,复杂任务才用大模型
  • 上下文压缩:工具返回结果先摘要再入上下文
  • 提前终止:Agent 已经拿到足够信息就停止,不要为了“完整”而多跑几步

我实测过,加了缓存和上下文压缩之后,同样的任务 token 消耗降低了 60% 左右,效果基本没损失。

4.4 一个真实的踩坑记录

最后分享一个我踩过的坑。有一次上线新版本,Provider 配置里加了一个新的模型,routing 规则也改了。测试环境一切正常,上线后却发现部分请求返回空结果。

排查了半天,发现是 fallback 链的问题:主 Provider 返回了一个“模型不存在”的错误,但这个错误被归类成了“可重试”,于是系统不断重试,重试失败后切换到 fallback,但 fallback 的模型不支持当前请求的某个参数,又失败了。最终返回空。

根因是错误分类不够细。“模型不存在”应该是不可重试错误,直接切换 Provider,而不是先重试。这个坑让我意识到:错误分类的粒度直接决定了降级策略的有效性。后来我把错误类型从 5 种细化到了 12 种,类似的问题再没出现过。

这套架构跑了大半年,最大的体会是:AI 应用的不确定性来自模型,但稳定性来自工程。Provider 路由、RAG 检索、Agent 编排,本质上都是在用工程手段给模型的不确定性兜底。模型会犯错,但系统不能崩。把这三块做扎实,AI 应用才能从 Demo 走向生产。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询