ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

AI应用架构核心:多Provider切换、RAG知识库与Agent编排实战

AI应用架构核心:多Provider切换、RAG知识库与Agent编排实战 这个系列写到第四篇终于可以聊一个稍微大一点的话题。前面几篇把模型调用、prompt 工程、数据预处理这些基础工作梳理完之后接下来要面对的是一个面向生产环境的 AI 模块到底该怎么组织代码结构。今天这篇聚焦三块核心内容——多 Provider 切换、RAG 知识库与 Agent 编排它们分别回答三个问题怎么让应用不被某个模型厂商绑死怎么让模型能回答训练数据之外的知识怎么让模型从一个对话接口变成能自己拆解任务、调用工具的执行系统。如果你正在搭建 AI 应用想把零散的大模型调用整理成可维护的模块或者已经在做 RAG 和 Agent 但总觉得差一口气这篇内容应该对你有用。文章不会只讲概念重点放在工程化的落地方案和我在实际调试中踩过的坑。1. 整体架构设计为什么这三件事必须放在一起做1.1 多 Provider 切换解决的是绑定问题先聊第一件多 Provider 切换。很多人一开始做 AI 应用都是直接调某一家厂商的 SDK把调用代码写死在业务逻辑里。最开始的 demo 阶段完全没问题但一旦要上生产、要接多个客户、要控制成本问题就出来了——不同厂商的价格、速率、能力差异很大有一天你想换一家更便宜的或者想用开源的本地模型兜底发现代码里到处散落着对原厂商数据结构、异常类型、API 路径的引用改起来牵一发动全身。我见过最典型的案例是项目里直接用厂商 A 的 Python SDK所有函数都返回厂商 A 的 Message 对象等到厂商 A 的某个模型下架或者价格调整整个团队的迭代节奏直接被拖着走。所以架构层面的多 Provider 切换本质上是把模型供应当成一个可替换的组件而不是把某个厂商的 SDK 直接拿进来到处用。这个思路跟后端开发里数据库访问层很像——你不可能因为要从 MySQL 换成 PostgreSQL 就去改所有业务代码一定是在中间加一层适配。1.2 RAG 解决的是知识问题第二件RAG 知识库。业界做 RAG 有一个朴素的原因——大模型训练数据有截止日期也没有企业内部的私有知识。你要么不断微调要么把知识放到外置的向量库里让模型在回答前先检索再生成。RAG 好在不用重新训练知识更新只需要重新切分、索引、做向量化当天就能生效成本远低于微调。这也是为什么这两年 RAG 几乎成了知识型 AI 应用的标配。但 RAG 不是简单地装个向量库就行它涉及的决策点非常多文档怎么切、粒度多大、用什么 embedding 模型、向量库选哪种、检索回来怎么过滤、怎么排序、要不要做重排、要不要让模型根据检索结果决定回答还是拒答。任何一环没做好出来的效果就是答非所问。后面第三章我会把这些点展开讲。这里先记住一个判断标准RAG 的价值不是用上了向量库而是模型回答时真的用上了你给它的知识并且用对了。1.3 Agent 编排解决的是执行问题第三件Agent 编排。聊 Agent 之前要分清楚模型是大脑但不等于Agent。一个普通的聊天窗口你问一句它答一句没有工具调用、没有多步推理、没有记忆管理那它只是一个聊天接口。Agent 的核心在于编排——给模型配上工具、设定目标、规划步骤、循环执行直到达成最终结果。为什么编排这件事要单独作为一个架构层因为 Agent 的执行过程不是一次调用就结束的它可能是模型决定调用搜索工具 → 拿到结果 → 继续分析 → 再调用代码解释器 → 返回最终答案这个循环里每一步都有失败的可能。没有编排层管理状态和重试写出来的代码会非常脆弱。我把 Provider、RAG、Agent 编排放在一篇里讲是因为它们三个正好是一个 AI 模块的底座、补给和引擎Provider 提供统一的模型访问RAG 提供动态知识Agent 编排负责把所有能力串联成可执行的流程。三者分开设计、组合使用模块才谈得上可维护。2. 多 Provider 切换的工程化落地2.1 统一接口层先把 SDK 挡住做多 Provider 切换第一步是在业务代码和厂商 SDK 之间加一个统一接口层。这个接口不需要很复杂两个核心抽象就够了一个是模型调用的统一入口一个是响应结果的统一结构。拿代码举例我会在项目里定义一个 chat_provider.pyfrom abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import AsyncIterator, Optional dataclass class ChatMessage: role: str # system / user / assistant / tool content: str name: Optional[str] None tool_calls: Optional[list] None dataclass class ChatResponse: message: ChatMessage provider: str model: str usage: dict field(default_factorydict) raw: object None class BaseProvider(ABC): abstractmethod async def chat( self, messages: list[ChatMessage], tools: Optional[list] None, temperature: float 0.3, **kwargs ) - ChatResponse: 统一聊天/补全入口所有厂商必须实现 raise NotImplementedError abstractmethod async def chat_stream( self, messages: list[ChatMessage], tools: Optional[list] None, temperature: float 0.3, **kwargs ) - AsyncIterator[ChatResponse]: 流式版本用于生成类场景 raise NotImplementedError这个抽象有两个关键设计第一所有厂商返回的数据最后都收敛成 ChatResponse业务代码只认这一个结构第二每个具体 Provider 类必须同时实现普通调用和流式调用避免某些场景只能用某一家厂商。ChatResponse 里还留了 raw 字段实在需要拿厂商特有的元数据比如某些厂商返回的 reasoning 内容时可以从 raw 里取而不是把厂商的数据结构透传出去。这个设计非常朴素但实际用下来它能挡住 90% 的厂商绑定问题。后面所有业务代码、Agent 编排、RAG 生成都只依赖这个统一结构厂商怎么实现我不关心。2.2 配置驱动把 base_url、api_key、model 全部外置有了统一接口层下一步是让每个 Provider 的细节从代码里剥离出来。最常见的方式是配置文件。我不会把配置直接写在业务模块里而是在项目的 config 目录下维护一个 providers.yamlproviders: openai-compatible: type: openai_compatible base_url: ${OPENAI_BASE_URL} api_key: ${OPENAI_API_KEY} default_model: gpt-4o-mini deepseek: type: openai_compatible base_url: https://api.deepseek.com api_key: ${DEEPSEEK_API_KEY} default_model: deepseek-chat local-ollama: type: openai_compatible base_url: http://localhost:11434/v1 api_key: not-needed default_model: qwen2.5:7b为什么要统一成 openai_compatible 这种格式因为目前绝大多数厂商都提供了 OpenAI 兼容的 /chat/completions 接口统一协议后只需要在配置层换 base_url、api_key、model代码基本不用改。我在实际项目中会再加一层配置校验启动时检查必填字段def load_providers(config_path: str) - dict[str, ProviderConfig]: raw yaml.safe_load(open(config_path)) providers {} for name, cfg in raw[providers].items(): if not cfg.get(base_url): raise ConfigError(fprovider [{name}] 缺少 base_url 配置) if cfg.get(type) ! openai_compatible: raise ConfigError(f暂不支持的 provider type: {cfg[type]}) providers[name] ProviderConfig(**cfg) return providers这块看起来平平无奇但事实上大部分线上问题都是配置引起的。很多框架的报错信息很直白比如provider 缺少 base_url 配置no api key for provider route本质就是配置没做校验、没做默认值兜底。后面第五章我会把这些真实报错列成速查表。2.3 切换策略优先级、fallback 与动态路由配置层准备好之后切换策略就有几种玩法。最简单的场景是手动切换在配置里设一个 active_provider 字段切换时改文件或改环境变量重启生效。这个方案适合工具类应用但不适合生产环境——生产环境更常见的是 fallback 和路由。fallback 的逻辑是主 Provider 超时或报错时自动降级到备用 Provider。async def chat_with_fallback( messages: list[ChatMessage], tools: Optional[list] None, **kwargs ) - ChatResponse: candidates [ (openai-compatible, {base_url: ..., api_key: ...}), (deepseek, {base_url: https://api.deepseek.com, api_key: ...}), (local-ollama, {base_url: http://localhost:11434/v1, api_key: not-needed}), ] last_error None for name, cfg in candidates: try: provider get_provider(name, cfg) return await provider.chat(messages, toolstools, **kwargs) except (TimeoutError, RateLimitError, ServiceUnavailableError) as e: last_error e continue raise RuntimeError(f全部 provider 均失败: {last_error})动态路由则更复杂一点比如按用户来源、按成本预算、按模型能力矩阵做分流。这个在早期可以不用做先把手动配置 自动 fallback跑通后面规模大了再引入规则引擎。我自己的经验是先保证任何一个 Provider 挂了系统还能服务比追求复杂的智能路由更重要。很多团队一上来就搞一个花哨的 Router 模块结果配置混乱、排障困难其实 fallback 才是性价比最高的那一步。2.4 从报错看配置管理的关键细节我在调研和实测过程中其实在配置管理上栽过好几次跟头。举几个很有代表性的真实报错你在做多 Provider 切换时大概率也会遇到报错信息里出现provider: default; model: gpt-6-astra; cause: 配置错误: codex provider 缺少 base_url 配置——base_url 缺失。这类报错典型原因是新加了一个 Provider但配置模板没有同步。no api key for provider route deepseek-official——API key 没配或命名不匹配。很多框架的 key 读取路径非常严格你配置里写的字段名和它期待的不一致就会报这个。400 配置错误: claude provider 缺少 base_url 配置——Claude 在统一网关下同样需要配置 base_url很多人以为官方 SDK 不用配实际上只要走了中转或统一网关层就必须显式声明。这些报错背后的共同教训是配置系统必须缺了就报错、错了就明说宁可启动时 fail fast不要运行到一半才发现某个 key 为空。所以我把配置校验放在进程启动阶段并打印每个 Provider 的检查状态方便一眼看出哪个没配好。这个习惯救过我很多次尤其是环境变量在 CI/CD 里被覆盖漏掉的情况。3. RAG 知识库从原型到可用的距离3.1 先形成链路认知索引、检索、重排、生成RAG 的流程可以简化为四步文档切分、向量化索引、检索召回、生成回答。看起来不复杂但在工程上每一步都有无数细节。一个标准的检索链路大概是文档加载txt、pdf、md、html、docx 都可能文本切分chunk切成大小合适的片段用 embedding 模型把每个 chunk 转成向量写入向量库用户提问时把问题转成向量在库里做相似度检索取 top-k如果需要对 top-k 结果做重排rerank把重排后的文本片段注入 prompt让模型基于这些上下文生成回答。很多人只关注第 4 步但实际影响效果最大的是第 2 步和第 5 步chunk 切多碎以及检索结果怎么排序。切得太碎语义不完整切得太大检索精度下降、token 浪费。合理的做法是先用滑动窗口切分每段 300-800 字中文可以按 500 字左右重叠 50-100 字然后在小规模验证集上反复调。我习惯把这个链路拆成独立的模块每一层都能单独跑脚本验证否则出了问题根本不知道是切分的问题、向量库的问题还是 prompt 的问题。3.2 切分与 embedding 选型的实战经验chunk 大小没有绝对标准取决于你的文档类型。合同、论文这种逻辑块清晰的可以按标题、段落结构切网页、百科这种杂乱文本更适合固定窗口 重叠。我在一个知识库项目里做过对比固定 500 字、重叠 80 字比按段切在命中率上高约 8%原因是很多段落本身超过 1000 字直接按段切会引入大量无关内容。另一个容易被忽略的细节是chunk 之间要保留结构化信息比如文档标题、章节号。把标题 正文拼在一起做向量化比只对正文做向量化要好很多这一点在处理长文档时特别明显。embedding 选型上如果你的场景主要是中文建议直接用中文优化过的模型。常见选择是 OpenAI 的 text-embedding-3-small、BGE 系列、M3E 系列本地可以跑 BGE-M3。需要注意一点检索用的 embedding 模型和生成模型没有关系不用跟主模型同厂商。我之前就见过有人误以为用了某家大模型就必须用它的 embedding其实完全没必要。向量库选型生产环境常用的是 Milvus、Qdrant、Weaviate个人和小团队建议先用 Chroma、LanceDB 这种嵌入式库零运维单机就能跑。rag 个人免费版本地知识库这类需求用 Chroma 本地模型基本就够。向量库本身不是瓶颈检索效果才是。3.3 用 hit rate 量化检索质量RAG 效果的评估不能靠感觉。业界最基础的指标就是 hit rate命中率也叫 recallk。它的定义很朴素对测试集中的每个问题看 top-k 检索结果里是否包含应该命中的那一段。比如你有 100 个测试问题每个问题对应的标准文档片段是已知的如果 top-5 里有 80 个问题命中了标准片段hit rate5 就是 0.8。hit rate 怎么用它是最初级的过滤器能快速判断 chunk 大小、embedding 模型、向量库参数是否合理。但注意hit rate 高不代表最终回答好——检索到了未必用得好。所以我一般会配合 MRR平均倒数排名看命中位置的靠前程度配合人工抽检验证最终生成质量。做评估集的时候不需要一下子做几千条先做 30-50 条典型问题跑一轮看 hit rate 变化再手动看几条 bad case迭代效率很高。我见过不少团队RAG 上线后没人维护效果越来越差却不做评估。实际上只要把评估脚本沉淀下来每次调整搜索参数、更换 embedding 模型之后跑一遍对比心里特别有底。3.4 轻量落地方案知识库的工程组合聊点实际可抄的内容。个人知识库或小团队知识库我推荐一套低成本组合文档处理unstructured 或 langchain 的文档加载器负责解析各类格式切分langchain 的 RecursiveCharacterTextSplitter或者直接自己写embedding本地跑 BGE-M3也可以用云端 API存储与检索Chroma本地持久化支持类似 SQLite 的轻量体验检索增强如果条件允许加一个 rerank 模型比如 BGE-Reranker。这套组合在 8GB 内存的机器上就能跑起来适合做rag 个人免费版、本地知识库这类场景。生产环境的话把 Chroma 换成 Milvus加上并行索引其他逻辑基本可以复用。这里我想多说一句 rerank基础向量检索的 top-k 结果里通常有噪声直接全部塞给模型会浪费 token、干扰回答用一个小的 rerank 模型把 top-20 重排成 top-5效果提升非常明显而且成本很低。3.5 进阶方向GraphRAG、Ontology RAG 与 Agentic RAGRAG 这两年演进很快GraphRAG、Ontology RAG、Agentic RAG 都是它在不同方向上的延伸。GraphRAG 的核心是给知识建图——把实体和关系抽取出来形成知识图谱再在图上做检索。优点是擅长多跳问题比如A 公司的核心产品是 XX 的供应商是 YY 最近发生了什么事缺点是构建成本高运行也慢。如果只是拿来做个问答机器人GraphRAG 的收益不一定配得上它的复杂度。Ontology RAG 更进一步引入本体约束给知识图谱加上语义模式适合对概念层级、逻辑关系有严格要求的垂直领域比如医疗、法律。这个词里的 ontology 本质上是更严格的 schema能做但不是所有场景都需要。Agentic RAG 则是把检索本身变成 Agent 的行为——不是每次都先检索而是由大模型决定这个问题需要不需要检索、检索多少次、要不要细粒度追问。它更灵活但也更复杂大概率会成为未来知识系统的主流形态。这三个方向不用一上来就全做。我的建议是先做好基础向量 RAG跑出 hit rate 和实际效果再根据业务复杂度逐渐升级。最近热词里频繁出现的agentic rag也印证了这个趋势但工程上仍然要遵循先简单后复杂的原则。4. Agent 编排模型如何变成执行者4.1 先分清 Workflow 和 AgentAgent 编排这个概念很多人容易和 Workflow 混淆。我在实际被问过很多次workflow 编排和 agent 编排到底什么关系harness 和 agent 有什么区别Dify 编排的应用能不能直接当 API 用。这里先给一个简单定义Workflow 是固定的流程步骤是预先写好的模型只负责填某个步骤里的空。比如一个客服工单流程查客户信息 → 查订单状态 → 生成回复每一步都是固定代码模型只做最后一步。好处是可控、可预测、成本低坏处是不灵活遇到流程之外的场景就崩。Agent 是动态的流程模型自己决定下一步调用什么工具、执行什么动作直到任务完成为止。比如你给 Agent 一个目标帮我对比这三家云厂商的价格它会自己决定调用搜索、筛选、总结中间可能经历很多轮。好处是灵活、能处理开放任务坏处是结果不可控、token 消耗大、容易陷入死循环。所以架构设计上正确的做法不是二选一而是按任务复杂度分层确定性高的任务走 Workflow开放性的任务走 Agent。这也是为什么编排层需要单独设计——它要同时支持两种执行模式。这个区分在做需求拆解时非常重要它直接决定了你要写多少代码、要买多少算力。4.2 编排引擎选型自研还是用框架目前主流的编排引擎我做一个对比方案定位适用场景学习成本可控性LangGraph图状态机需要精细控制每个节点的 Agent 应用较高高Dify低代码编排快速搭建、业务流程清晰低中AgentScope多智能体研究/生产分布式多智能体、消息传递中高Flowise低代码 RAG/Agent原型验证低中自研循环简单 while 循环单 Agent、工具调用场景低最高我个人的经验是还在原型阶段可以用 Dify 或者 Flowise 快速验证一旦要上生产、要精细控制状态和错误处理LangGraph 和自研循环更靠谱。之前有一些团队跟我讨论Dify 编排的应用能不能当作 Continue 的 API 使用这个场景其实是把低代码编排暴露成服务 API技术上可行但要额外处理鉴权、并发、模型上下文等细节反而比直接写代码更费劲。这里也给一个选型建议如果你只是做内部工具Agent 的调用频率低、流程不复杂自研 200 行循环足够如果你要做多租户、高并发、需要可观测性和人工介入审批那就老老实实用成熟的编排框架别自己造轮子。4.3 一个最小可用的编排实现自研一个最小 Agent 编排器其实不复杂。核心就是一个循环模型 → 判断是否要调用工具 → 调用工具 → 把结果喂给模型 → 继续直到模型不再请求工具。async def run_agent( agent: Agent, user_input: str ) - str: messages [ChatMessage(roleuser, contentuser_input)] max_iterations 10 for i in range(max_iterations): response await agent.provider.chat( messages, toolsagent.tools, ) # 模型没有请求工具说明回答完成 if not response.message.tool_calls: return response.message.content # 模型请求了工具依次执行 for tool_call in response.message.tool_calls: tool_result await agent.execute_tool(tool_call) messages.append(ChatMessage( roletool, nametool_call.name, contenttool_result, )) # 把模型的工具请求也加入上下文 messages.append(response.message) raise AgentTimeoutError(达到最大迭代次数Agent 未能完成)这段代码虽然短但已经包含了 Agent 编排最核心的机制循环、工具执行、上下文累积、终止条件。实际生产时还要补充超时控制、异常重试、上下文窗口溢出处理做摘要压缩、工具并发限制、审计日志等。但骨架就是这么简单。我特别想强调把模型的工具请求也加入上下文这一步。很多人第一次写 Agent 循环会漏掉它结果模型在下一轮里不记得自己刚才调过什么工具表现为连续调用同一个工具、逻辑断裂。把 response.message 追加进 messages才能让模型看到自己已经做过哪些决策。4.4 多智能体协作中的状态与容错如果要从单 Agent 升级到多 Agent复杂度会明显上升。多智能体的协作模式主要有三种主从式一个主控 Agent 负责任务拆解把子任务分发给多个子 Agent汇总结果。流水线式一个 Agent 的输出作为下一个 Agent 的输入适合流程明确的任务。竞争式多个 Agent 各给各的方案由另一个裁判 Agent 选择最优。多智能体编排最怕的是状态不同步。子智能体执行完之后它产出的中间状态比如回忆、工具结果、置信度要能被下一个智能体读取。所以我会在架构里设计一个共享状态区可以是一份结构化 JSON也可以是向量记忆区每个智能体的输入输出都从状态区读写避免直接传参导致隐式依赖。容错方面单 Agent 的失败模式已经很复杂了多 Agent 会把失败放大。常见问题包括agent execution terminated due to error——子 Agent 执行失败导致整条链终止。我的处理策略是给每个子任务加独立的 retry 和降级结果机制不要让一个子任务的失败拖垮整个编排。比如子 Agent 调用搜索失败时可以降级为直接返回空结果并标注置信度低让主控 Agent 自己判断下一步。这个设计不是偷懒而是承认 Agent 本身就有不确定性编排器的职责是让失败可控、可降级、可观测。5. 常见问题排查与避坑实录5.1 Provider 层排查表把前面几章里出现过的实际报错整理成速查表方便你直接对号入座报错现象原因修复方式缺少 base_url 配置配置文件中没有 base_url 或字段名不匹配在配置模板中补全 base_url检查字段命名是否与框架约定一致no api key for provider route配置里没有 API key或 key 名称对不上框架期望检查环境变量/配置文件中的 key 字段考虑加统一配置校验missing session id网关类 Provider 缺少会话标识检查网关配置确认请求头或参数包含 session id413 payload too large上传的文档/请求体超过服务端限制压缩请求体、减少单次输入长度必要时扩服务端限制model is unavailable配置的模型名不存在或当前不可用核对厂商模型列表替换为有效模型名上游请求失败上游服务故障或账号额度限制检查账号额度、服务状态配置 fallback看到这些报错时先别慌着改代码一定要先想一层这个报错是配置问题还是服务问题。配置问题我们自己能修服务问题只能等或换。排查顺序建议是先看环境变量 → 再看配置文件 → 再看框架版本 → 最后看服务状态。5.2 RAG 效果问题定位RAG 效果差先别急着换向量库。我调试 RAG 的顺序是先看 hit rate再人工看 5 条 bad case最后才调整 chunk 或 embedding。比较常见的坑embedding 不一致索引时用了 A 模型检索时用了 B 模型语义空间对不上召回必挂。这个坑太常见了一定要确保同一个知识库从始至终用同一个 embedding 模型。换模型的时候全量重新索引不要增量混合。检索阈值失灵有的项目给相似度设了固定阈值比如低于 0.7 就不返回。但不同 embedding 模型的分数分布差异巨大0.7 在某个模型里可能是高分在另一个模型里是低分。建议基于历史分数分布动态判断不要写死。chunk 太小导致上下文不足检索命中了但生成回答空洞大概率是 chunk 太小或没有做上下文扩展把命中 chunk 的前后片段一起带入 prompt。也可以让模型在发现上下文不足时主动发起新一轮检索这就是上面说的 Agentic RAG 思路。5.3 Agent 执行异常复盘Agent 层我看到最多的三个问题第一agent execution terminated due to error——执行链断在某个工具调用。我会在工具调用处加统一的异常捕获把异常转成一句话放入上下文而不是直接让进程崩溃。实际上很多模型看到tool execution failed: xxx后会自己换一种方式完成给模型一个补救机会往往能救回来。这里的关键是异常消息要带上足够的上下文比如工具名、参数、错误类型否则模型也无从判断怎么补救。第二上下文长度爆炸。Agent 循环到第 5、6 轮时很容易超窗口。解决方法是状态压缩不是把每一轮的消息都带上而是定期把历史对话摘要成一个 system 消息再配合只保留最后 N 轮完整消息的策略。我目前的做法是每 3 轮做一次压缩效果还不错。第三死循环。模型反复调用同一个工具且结果不变。这种要设置 max_iterations 上限按实际任务设日常任务 8-15 轮足矣同时可以做简单的重复检测比如检测到同一工具在最近 3 轮返回了完全相同的签名直接打断。还有一点容易被忽略给工具加副作用保护。比如发送邮件这种不可逆操作在 Agent 架构里不能让它被自动调用必须走人工确认。这一条是上线前必须检查的别等到事故了才后悔。写在最后个人体会这个系列写到第四篇我自己的感触是 AI 模块架构设计最难的不是某个算法选型而是把 Provider、RAG、Agent 这三层之间的边界划清楚。划清楚了每一层可以独立演进划不清楚任何一个厂商升级、任何一个知识库数据量翻倍都可能引发连锁故障。以我个人的经验来看刚开始做架构时不要追求把所有高级特性一步到位先把多 Provider 能配通 RAG 能跑通 Agent 能执行完一次任务这条主线打通再谈优化。最后分享一个小技巧把每类踩过的坑都留一条复现记录在 repository 的 docs 目录里下次队友或者未来的你再遇到直接查表就能少折腾半天。你踩过的坑越多这个模块就会越稳固。
返回列表