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，跑得好好的，结果对方调整了计费策略，成本直接翻了三倍。当时想换到另一家，发现代码里模型调用散在十几个文件里，每个地方都写死了model="xxx"和对应的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作为必填项，启动时校验，能省掉大量运行时排查。
运行时选择策略我一般分三层：显式指定 > 模型映射 > 默认优先级。显式指定就是调用方直接传provider="provider_a"；模型映射是维护一张"模型名到 Provider"的表，比如gpt-4o走 A，claude-3-5-sonnet走 B；默认优先级则是在都没指定的情况下，按配置里的 priority 选第一个可用的。
2.3 故障转移与降级策略
Provider 不可能永远可用。网络抖动、限流、模型临时下线，这些都得处理。我的经验是故障转移要分错误类型区别对待，不能一报错就无脑重试或切换。
| 错误类型 | 典型表现 | 处理策略 |
|---|---|---|
| 网络超时 | 连接超时、读超时 | 同 Provider 重试 1-2 次，再切换 |
| 限流 429 | rate limit exceeded | 退避等待，或切换备用 Provider |
| 认证失败 401 | invalid api key | 不重试，直接标记该 Provider 不可用 |
| 请求过大 413 | payload 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，合并去重后进入下一层。
融合重排用 RRF（Reciprocal 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, key=scores.get, reverse=True)上下文组装要考虑 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 编排模式大致分两类：自主决策型和工作流型。
自主决策型以 ReAct（Reasoning + 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 key | Key 错误或过期 | 检查环境变量，加健康检查 |
| 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 层加本地模型做混合部署。但那是下一步的事了，当前这套先把基础打牢，后面加什么都不会太费劲。
