1. 从单点调用到体系化架构AI 模块设计的核心命题做过 AI 应用的人大概都有这个体会Demo 跑通只要一个下午但真要把它做成一个能持续迭代、能换模型、能接知识库、能跑复杂任务流的系统坑是一个接一个。我最早做 AI 集成的时候代码里到处是openai.ChatCompletion.create模型名硬编码API Key 散落在各个文件想换个模型得全局搜索替换想加个知识库检索得把整个调用链重写一遍。这种写法在原型阶段没问题但只要项目稍微长大一点维护成本就指数级上升。这篇要聊的就是怎么把 AI 模块从一堆散落的 API 调用变成一套有层次、可替换、可编排的架构。核心围绕三块多 Provider 切换、RAG 知识库、Agent 编排。这三块不是孤立的它们其实是一条链上的三个层次——Provider 层解决用哪个模型的问题RAG 层解决模型该知道什么的问题Agent 层解决模型该怎么干活的问题。把这三层拆清楚、接好口整个 AI 模块才算真正立起来。适合谁看如果你正在做 AI 应用或者准备把 AI 能力集成进现有系统又或者你已经踩过换模型改半天知识库检索不准Agent 跑着跑着就乱了这些坑那这篇内容应该能给你一些可直接参考的思路和代码结构。我不会只讲概念会把每一层的设计取舍、接口定义、关键参数、常见故障都摊开说。2. 多 Provider 切换别让模型绑定成为技术债2.1 为什么必须做 Provider 抽象先说一个真实的场景。我有个项目最早用的是某家模型的 API跑得好好的结果对方调整了计费策略成本直接翻了三倍。当时想换到另一家发现代码里模型调用散在十几个文件里每个地方都写死了modelxxx和对应的base_url改起来简直是噩梦。更麻烦的是不同 Provider 的请求格式、返回结构、错误码都不一样有的用messages数组有的要prompt字符串有的流式返回是 SSE有的是 WebSocket。这就是不做 Provider 抽象的代价。Provider 层的核心价值是把模型能力变成一种可替换的资源就像数据库连接池一样上层业务不关心底层用的是 MySQL 还是 PostgreSQL只关心我要一次对话补全或我要一次向量化。抽象的关键在于定义一套统一的内部接口把所有 Provider 的差异都收敛到适配器里。我通常会把能力拆成几类对话补全chat、文本向量化embedding、重排序rerank、以及可能的图像理解。每一类定义一个抽象基类具体 Provider 实现这个基类。from abc import ABC, abstractmethod from typing import List, Dict, AsyncIterator class BaseChatProvider(ABC): abstractmethod async def chat(self, messages: List[Dict], **kwargs) - str: ... abstractmethod async def chat_stream(self, messages: List[Dict], **kwargs) - AsyncIterator[str]: ... abstractmethod def count_tokens(self, text: str) - int: ...注意这里我把count_tokens也放进来了。很多人会忽略这一点但 token 计数是成本控制和上下文裁剪的基础不同 Provider 的分词方式不同必须由 Provider 自己实现。2.2 Provider 注册与运行时选择有了抽象基类接下来要解决怎么知道有哪些 Provider 可用以及运行时怎么选。我的做法是做一个Provider Registry用装饰器或配置驱动的方式注册。配置里声明每个 Provider 的类型、base_url、api_key的环境变量名、支持的模型列表、以及优先级。providers: - name: provider_a type: openai_compatible base_url: ${PROVIDER_A_BASE_URL} api_key_env: PROVIDER_A_KEY models: [gpt-4o, gpt-4o-mini] priority: 1 - name: provider_b type: anthropic_compatible base_url: ${PROVIDER_B_BASE_URL} api_key_env: PROVIDER_B_KEY models: [claude-3-5-sonnet] priority: 2这里有个关键设计base_url必须显式配置不能依赖 SDK 的默认值。我见过太多配置错误provider 缺少 base_url 配置的报错根源就是代码里假设了某个默认端点换环境就崩。把base_url作为必填项启动时校验能省掉大量运行时排查。运行时选择策略我一般分三层显式指定 模型映射 默认优先级。显式指定就是调用方直接传providerprovider_a模型映射是维护一张模型名到 Provider的表比如gpt-4o走 Aclaude-3-5-sonnet走 B默认优先级则是在都没指定的情况下按配置里的 priority 选第一个可用的。2.3 故障转移与降级策略Provider 不可能永远可用。网络抖动、限流、模型临时下线这些都得处理。我的经验是故障转移要分错误类型区别对待不能一报错就无脑重试或切换。错误类型典型表现处理策略网络超时连接超时、读超时同 Provider 重试 1-2 次再切换限流 429rate limit exceeded退避等待或切换备用 Provider认证失败 401invalid api key不重试直接标记该 Provider 不可用请求过大 413payload too large裁剪上下文后重试不切换模型不可用model is unavailable切换到同能力模型或备用 Provider配置错误 400缺少 base_url 等启动时就应该拦截运行时记录并跳过这里特别说一下413 payload too large。这个错误不是 Provider 的锅是你上下文塞太多了。正确做法是在 Provider 层之上做上下文预算管理根据模型的上下文窗口和当前 token 数动态裁剪。我一般会预留 20% 的余量给输出输入部分按系统提示 最近对话 历史摘要的优先级保留。还有一个坑流式请求的故障转移比非流式复杂得多。因为流已经开始返回了中途断了没法重放。我的做法是流式请求只在首字节返回前允许切换一旦开始输出就只做重试不切换并且把已输出的内容缓存下来重试时作为前缀续写。2.4 实操心得Provider 层的三个避坑点第一不要在业务代码里直接 import 具体 Provider 的 SDK。所有调用都走 Registry 拿到的抽象接口。这样换 Provider 时业务代码零改动。第二API Key 永远从环境变量或密钥管理服务读取配置文件里只写变量名。我见过把 Key 写进 YAML 提交到仓库的后果不用多说。第三给每个 Provider 加健康检查。启动时做一次轻量探测比如发一个极短的请求把不可用的 Provider 提前标记出来避免第一个真实请求就撞墙。3. RAG 知识库让模型答得准而不只是答得像3.1 RAG 的本质与常见误区RAG 这个词现在被用得很泛但它的本质其实很朴素在模型生成之前先从外部知识库里检索出相关内容拼进上下文让模型基于这些内容回答。它解决的是大模型的两个硬伤——知识过时和幻觉。但我见过太多 RAG 项目效果不好问题往往不在模型而在检索。常见的误区有这么几个只做向量检索不做关键词检索。向量检索擅长语义相似但对专有名词、编号、代码标识符这类精确匹配很弱。用户问错误码 E1024 怎么解决向量检索可能返回一堆语义相近但错误码不对的内容。切块太粗或太细。切太粗一块里混了好几个主题检索出来噪声大切太细上下文不完整模型拼不出完整答案。不做重排序。初步检索召回 top-50直接取 top-5 塞给模型中间没有精排质量参差不齐。忽略元数据过滤。知识库里有多个来源、多个版本的内容检索时不按来源或时间过滤容易召回过期信息。3.2 检索链路的分层设计我现在的 RAG 链路一般分四层查询理解 → 多路召回 → 融合重排 → 上下文组装。查询理解这层容易被跳过但很重要。用户的问题往往口语化、有指代、有隐含条件。我会做几件事查询改写把口语化问题改写成检索友好的形式、查询扩展生成几个同义或相关的子查询、以及意图识别判断这是事实查询、对比查询还是操作查询不同意图检索策略不同。多路召回是核心。我通常并行跑三路向量检索语义、BM25 关键词检索精确、以及基于元数据的结构化检索。三路各召回 top-20 到 top-50合并去重后进入下一层。融合重排用 RRFReciprocal Rank Fusion做初步融合再用一个 rerank 模型做精排。RRF 的好处是不需要调权重直接按排名倒数求和对多路召回很友好。def rrf_fusion(rankings: List[List[str]], k: int 60) - List[str]: scores {} for ranking in rankings: for rank, doc_id in enumerate(ranking): scores[doc_id] scores.get(doc_id, 0) 1.0 / (k rank 1) return sorted(scores, keyscores.get, reverseTrue)上下文组装要考虑 token 预算。我会按重排分数从高到低填充同时做去重和相邻块合并如果两个块来自同一文档且位置相邻合并成一个更大的块避免上下文割裂。3.3 切块策略没有银弹只有权衡切块是 RAG 里最需要根据数据特点调的部分。我试过几种策略各有适用场景切块策略适用场景优点缺点固定长度通用文本实现简单容易切断语义按段落结构化文档语义完整段落长度不均递归切分混合内容兼顾结构与长度参数需调语义切分高质量要求语义边界准计算成本高按标题层级技术文档保留结构依赖文档格式我现在的默认方案是递归切分 标题感知优先按标题切标题下内容超长再按段落切段落还超长再按句子切最后才硬切。块大小控制在 300-500 token重叠 50-80 token。重叠是为了避免边界信息丢失但重叠太多会导致检索结果冗余需要权衡。提示切块大小不是越小越好。块太小单块信息量不足模型拼不出答案块太大检索精度下降噪声增多。我的经验是先用 400 token 左右起步根据实际检索效果微调。3.4 向量化与索引选型向量化模型的选择直接影响检索质量。我的原则是检索用什么模型查询就用什么模型两者必须一致否则向量空间不对齐检索结果会莫名其妙地差。索引方面数据量小的时候几万条以内用内存索引如 FAISS 的 flat 索引就够了召回率最高。数据量大了再考虑 IVF、HNSW 这类近似索引。HNSW 在召回率和速度之间平衡得比较好是我比较常用的选择。这里有个容易忽略的点向量维度和距离度量要匹配。余弦相似度适合归一化后的向量内积适合未归一化的欧氏距离对量纲敏感。选错了度量方式检索效果会打折扣。3.5 检索命中率的排查思路RAG 效果不好时第一步永远是把检索结果打出来看。我一般会记录每次查询的召回内容、分数、以及最终进入上下文的部分。如果检索结果里根本没有正确答案那是召回问题如果召回里有但没进上下文那是重排或组装问题如果都进了但模型还是答错那才是生成问题。排查召回问题我会按这个顺序查查询改写是否失真、向量模型是否匹配、切块是否合理、索引是否更新、元数据过滤是否过严。这几个点里切块和索引更新是最常出问题的。文档更新了但索引没重建检索出来的还是旧内容这种低级错误我踩过不止一次。4. Agent 编排从单次问答到多步任务执行4.1 Agent 到底解决什么问题普通的 LLM 调用是一问一答模型只能基于已有知识或给定上下文回答。但很多真实任务不是一次问答能搞定的——比如帮我查一下这个季度的销售数据分析趋势然后生成一份报告这需要多步查数据、分析、写报告中间还可能要根据结果调整下一步。Agent 的核心就是让模型能够自主决定下一步做什么调用工具观察结果再决定下一步。它把 LLM 从文本生成器变成了任务执行器。但 Agent 也是最容易失控的地方。我见过太多 Agent 跑着跑着就陷入循环、或者调用了一堆无关工具、或者在中途丢失了目标。所以 Agent 编排的重点不是让它能跑而是让它可控地跑。4.2 编排模式从 ReAct 到工作流Agent 编排模式大致分两类自主决策型和工作流型。自主决策型以 ReActReasoning Acting为代表模型自己决定调用哪个工具、什么时候结束。灵活但不可控适合探索性任务。工作流型则是预先定义好步骤和分支模型只在特定节点做决策。可控但灵活性差适合流程明确的任务。我现在的做法是混合整体用工作流框住关键节点用自主决策。比如一个报告生成 Agent整体流程是取数 → 分析 → 撰写 → 校验这是固定的但分析这一步内部模型可以自主决定用哪些分析工具、要不要多轮探索。class AgentWorkflow: def __init__(self, steps: List[Step]): self.steps steps async def run(self, task: str, context: dict): state {task: task, context: context, history: []} for step in self.steps: result await step.execute(state) state[history].append({step: step.name, result: result}) if step.is_terminal(state): break return state每个 Step 可以是一个 LLM 调用、一个工具调用、或者一个子 Agent。这样既保证了整体流程可控又给了局部灵活性。4.3 工具设计与调用约束Agent 的能力边界由工具决定。工具设计有几个原则工具描述要精确。模型是根据描述来决定调不调、怎么调的。描述模糊模型就会乱调。我一般会写清楚这个工具做什么、什么时候用、参数含义、返回什么、有什么限制。参数要强类型校验。模型生成的参数经常有格式问题比如该传数组传了字符串、该传数字传了带引号的字符串。在工具入口做校验和纠错比让模型自己改要可靠。工具数量要克制。我见过一个 Agent 挂了三十多个工具结果模型选择困难经常调错。我的经验是单次任务暴露的工具不超过 10 个多了就分组或分层。要有超时和重试上限。工具调用可能卡住必须有超时。重试也要有上限否则模型可能反复调同一个失败的工具。4.4 状态管理与上下文控制Agent 多步执行会产生大量中间状态如果不加控制上下文会迅速膨胀最后撞上 token 上限。我的做法是分层管理状态短期状态当前步骤的输入输出完整保留中期状态最近几步的摘要压缩保留长期状态任务目标和关键结论始终保留。每步执行前根据当前 token 预算决定加载哪些状态。还有一个技巧是显式记录任务进度。让 Agent 在每步后更新一个已完成/待完成清单这样即使上下文被裁剪任务目标也不会丢。这个清单本身也占不了多少 token但能显著降低 Agent 跑偏的概率。4.5 Agent 失控的典型场景与对策失控场景表现对策死循环反复调同一工具设最大步数检测重复调用目标漂移做着做着忘了原始任务每步注入任务目标定期校验工具滥用调一堆无关工具限制工具数量加调用理由要求上下文爆炸token 超限报错分层状态管理定期压缩错误累积一步错步步错关键节点加校验允许回滚我踩过最深的一个坑是错误累积。Agent 第一步取数取错了后面分析、撰写全建立在错误数据上最后产出一份看起来很专业但完全错误的报告。后来我在关键节点加了校验步骤比如取数后先做一次数据合理性检查不通过就中断而不是继续。5. 三层如何协同接口设计与数据流5.1 层间接口的边界三层拆开了但真正难的是怎么接。我的原则是上层不感知下层实现下层不假设上层意图。Provider 层对外只暴露给定消息返回结果的能力不关心这消息是用户直接问的还是 RAG 拼的。RAG 层对外只暴露给定查询返回相关上下文的能力不关心这查询是用户问的还是 Agent 生成的。Agent 层则组合前两者但通过接口调用不直接依赖具体实现。这样设计的好处是每层可以独立测试、独立替换。我换 Provider 不影响 RAG换 RAG 策略不影响 Agent。5.2 一次完整请求的数据流以一个基于知识库的问答 Agent为例一次请求的数据流大致是Agent 接收用户任务判断需要检索知识库Agent 调用 RAG 层的检索接口传入查询RAG 层做查询理解、多路召回、重排、组装返回上下文Agent 把上下文和任务一起交给 Provider 层Provider 层选择可用 Provider发起调用返回结果Agent 判断任务是否完成未完成则继续循环这个流程里每一层的输入输出都是明确定义的数据结构层与层之间通过接口通信。任何一层出问题都能快速定位。5.3 可观测性别等出问题才想起来三层架构如果没有可观测性排查问题会很痛苦。我一般会在每层加日志和指标Provider 层记录每次调用的 Provider、模型、耗时、token 数、错误码RAG 层记录查询、召回数量、重排分数、最终上下文Agent 层记录每步的输入输出、工具调用、状态变化。这些数据不仅能排查问题还能指导优化。比如发现某个 Provider 的 P99 延迟特别高就可以调整优先级发现某类查询召回率低就可以针对性优化切块或查询改写。6. 常见问题速查与避坑清单6.1 Provider 层常见报错报错信息根因解决缺少 base_url 配置配置未显式指定端点启动时校验必填项model is unavailable模型下线或名称错误维护模型映射表加降级413 payload too large上下文超限上下文预算管理动态裁剪401 invalid api keyKey 错误或过期检查环境变量加健康检查429 rate limit触发限流退避重试切换备用6.2 RAG 效果优化清单检索结果先打出来看定位是召回还是生成问题向量模型和查询模型必须一致混合检索向量 关键词优于单一检索重排序能显著提升 top-k 质量切块大小按数据特点调400 token 起步文档更新后必须重建索引元数据过滤能排除过期和无关内容6.3 Agent 稳定性清单设最大步数上限防止死循环每步注入任务目标防止漂移工具数量克制描述精确工具调用加超时和重试上限分层状态管理控制上下文膨胀关键节点加校验允许中断和回滚7. 我在实际项目中的几点体会这套三层架构我前后迭代了好几版最大的体会是架构的价值不在于一开始就设计得多完美而在于让后续的修改成本足够低。我第一版 Provider 层只抽象了 chat 接口后来要加 embedding 和 rerank因为接口设计得还算干净扩展起来没伤筋动骨。RAG 层最早只有向量检索后来加关键词和重排也是因为召回接口是统一的加一路不影响其他。另一个体会是别过度设计。我见过有人一上来就搞微服务、消息队列、分布式向量库结果数据量才几千条纯属给自己找麻烦。架构要匹配当前规模留好扩展点就行等真到了那个量级再演进。最后分享一个小技巧给每层都写一个 mock 实现。Provider 层有个 mock provider 返回固定文本RAG 层有个 mock 检索返回固定片段Agent 层就能在没有真实依赖的情况下跑通全流程。这在开发和测试阶段能省大量时间尤其是当外部服务不稳定的时候。这套东西后续还能往几个方向扩展比如给 RAG 加图谱检索做多跳推理给 Agent 加多智能体协作做任务分解给 Provider 层加本地模型做混合部署。但那是下一步的事了当前这套先把基础打牢后面加什么都不会太费劲。