从零搭建AI工程体系：RAG检索增强与模型调用实战指南

1. 从零搭建AI工程体系，为什么我劝你别一上来就啃框架

这两年AI应用开发的门槛肉眼可见地降低了，随便拉个前端都能用现成的API搭出一个能跑通的对话Demo。但真到了要把这套东西塞进生产环境、扛住真实流量、控制住成本、还得让模型输出稳定可控的时候，你会发现光会调接口根本不够用。ai-engineering-from-scratch这个项目标题，说的就是从零开始把AI工程这套东西一层一层搭起来，而不是站在LangChain或者各种封装框架的肩膀上做个调包侠。

我自己带过几个从零到一的AI项目，踩过的坑基本都集中在同一个地方：前期图快，直接上高层框架，结果出了问题根本不知道是哪一层崩的。检索召回不准、上下文拼接超长、模型输出格式飘忽、并发一上来延迟直接爆炸，这些问题在Demo阶段全被掩盖了，上线之后集中爆发。所以这个项目的核心价值，是帮你建立一套完整的AI工程心智模型，知道一个AI应用从输入到输出中间到底经过了哪些环节，每个环节的边界在哪，出问题该往哪个方向排查。

这篇文章适合三类人看。第一类是后端或者全栈工程师，想转AI应用方向但不想只做API搬运工；第二类是做了一段时间AI Demo，发现往生产推的时候处处碰壁的开发者；第三类是想系统理解AI工程全貌的技术负责人，需要知道一个AI系统到底由哪些模块组成、各模块的成熟度如何。我会按照从底层到上层的顺序，把数据准备、检索增强、提示工程、模型调用、输出解析、评估监控这几个核心环节拆开讲，每个环节都给出可落地的方案和我在实际项目里验证过的参数。

需要提前说明的是，AI工程这个领域变化极快，今天的最佳实践可能半年后就被推翻。所以我更侧重讲清楚每个环节的设计逻辑和取舍依据，而不是给你一份死记硬背的配置清单。理解了为什么这么设计，你才能在工具迭代的时候快速迁移。

2. 整体架构设计与技术选型思路

2.1 为什么选择从零手搓而不是直接用框架

市面上主流的AI应用框架，比如LangChain、LlamaIndex，确实能让你在半小时内跑通一个RAG Demo。但我在实际项目里的经验是，这类框架的抽象层太厚，一旦你需要做精细化的控制，比如自定义检索策略、调整上下文压缩逻辑、实现特定的重排序算法，就会发现自己一直在跟框架的设计理念做斗争。

从零搭建的好处是每一层都透明。你知道用户的问题进来之后，先经过什么处理，再经过什么处理，每一步的输入输出长什么样。出了问题，你可以精确地定位到是检索层召回率不够，还是提示词模板拼接有问题，还是模型本身对某类输入不敏感。这种可观测性在生产环境里是刚需，框架帮你省下的那点开发时间，在排查线上问题的时候会加倍还回去。

当然，从零搭建不意味着所有东西都自己写。向量数据库、嵌入模型、推理服务这些基础设施该用现成的就用现成的，没必要重复造轮子。我的原则是：业务逻辑层自己控制，基础设施层用成熟方案。这样既保证了灵活性，又不会在底层浪费太多时间。

2.2 核心模块拆解与数据流向

一个完整的AI工程系统，从用户输入到最终输出，大致会经过这么几个阶段。用户的问题首先进入输入预处理层，做意图识别、查询改写、敏感内容过滤。然后进入检索层，如果是知识密集型任务，会去向量库或者关键词索引里召回相关文档。召回的结果经过重排序和压缩，筛选出最相关的片段。接着进入提示组装层，把系统指令、检索到的上下文、用户问题、历史对话按特定模板拼成最终的提示词。再往下是模型调用层，负责请求分发、重试、降级、流式输出。模型返回的原始文本进入输出解析层，做结构化提取、格式校验、后处理。最后是评估与监控层，记录每次请求的完整链路数据，用于后续的效果分析和迭代优化。

这个数据流向看起来线性，但实际实现的时候会有很多分支。比如有些查询不需要检索，直接走模型知识回答；有些查询需要多轮检索，先召回再根据中间结果二次召回；有些场景需要并行调用多个模型然后做结果融合。这些分支逻辑就是AI工程真正复杂的地方，也是从零搭建时你需要自己设计清楚的部分。

2.3 技术栈选型的取舍逻辑

在具体技术选型上，我倾向于把选择分成三类来看。第一类是必须自己控制的，包括提示词模板管理、检索策略、输出解析逻辑、评估指标定义。这些东西直接决定了系统的效果上限，必须掌握在自己手里。第二类是可以用开源方案的，包括向量数据库、嵌入模型、重排序模型、推理服务框架。这些组件相对标准化，社区方案已经比较成熟。第三类是建议用云服务的，包括底层算力、模型API、监控告警基础设施。这些自建成本太高，除非有特殊合规要求，否则没必要自己搞。

具体到向量数据库，我在中小规模场景下更倾向用轻量级方案，比如基于FAISS或者Chroma做本地索引，部署简单、调试方便。数据量上到千万级再考虑Milvus或者Qdrant这类分布式方案。嵌入模型的选择要看语言和领域，通用场景下多语言模型基本够用，垂直领域建议做微调或者至少做一轮领域适配的评估。重排序模型是提升检索精度的关键，很多团队会忽略这一步，但实测下来加上重排序之后，Top-3的命中率能有明显提升。

3. 核心模块的细节实现与实操要点

3.1 数据准备与索引构建的关键细节

AI工程的上限很大程度上由数据质量决定。我见过太多项目，模型和框架都选得不错，但知识库里的文档乱七八糟，切分粒度不合理，元数据缺失，导致检索效果怎么调都上不去。数据准备这一步，核心要解决三个问题：文档怎么切、元数据怎么打、索引怎么建。

文档切分不是简单地按固定字数切。我的经验是，切分粒度要跟内容的语义结构对齐。技术文档按章节切，FAQ按问答对切，长文章按段落切，代码按函数或类切。切分的时候要保留一定的重叠，通常设置10%到20%的重叠比例，避免关键信息刚好被切断。重叠太多会浪费存储和检索开销，太少会丢上下文，这个比例需要根据实际内容做几轮测试来定。

元数据的设计往往被低估。除了文档来源、创建时间这些基础字段，我建议至少加上内容类型、所属主题、置信度这几个维度。内容类型用于区分是事实性知识还是观点性内容，检索的时候可以给不同类型设置不同权重。所属主题用于做检索时的范围过滤，比如用户问的是产品功能，就只在产品文档范围内检索。置信度用于标记内容的可靠性，低置信度的内容在拼接上下文时可以做降权处理。

索引构建阶段，我通常会同时建向量索引和关键词索引两套。向量索引负责语义召回，关键词索引负责精确匹配。用户查询里如果包含特定的产品名、错误码、专有名词，关键词索引的召回效果往往比向量索引更准。两套索引的结果做融合，用RRF或者加权求和的方式合并排序，实测下来比单走向量索引的召回率高出一截。

注意：索引构建完成后一定要做一轮召回测试，准备一批典型查询，人工标注哪些文档是真正相关的，然后计算召回率和精确率。没有这步验证，后面所有调优都是盲调。

3.2 检索增强生成的参数调优实战

RAG这套东西看起来简单，但参数特别多，每个参数都会影响最终效果。我按重要性排个序，讲一下实际调优时的思路。

Top-K的取值是最先要定的。K太小，召回不全，模型没有足够信息回答；K太大，上下文里塞了一堆无关内容，反而干扰模型判断，还浪费token。我的经验是，先用一个中等值比如10做基线，然后观察不同K值下最终答案的质量。通常K在5到15之间会有一个比较明显的拐点，超过这个点之后效果提升就很小了。如果用了重排序，可以先把召回K设大一些比如50，重排序之后再取Top-5或者Top-8送给模型。

相似度阈值是第二个关键参数。低于某个相似度的文档，宁可不要也不要硬塞进去。这个阈值跟嵌入模型的特性有关，不同模型的分值分布不一样，不能直接套用。我的做法是拿一批标注数据，画出相关文档和不相关文档的相似度分布，找两条分布曲线的交叉点附近作为阈值。实际用的时候可以稍微放宽一点，避免漏掉边缘相关的文档。

上下文压缩是提升效果的重要手段。召回的长文档里往往只有一两句话是真正相关的，把整篇文档塞给模型既浪费token又引入噪声。压缩的做法有两种，一种是抽取式，用一个小模型或者规则从文档里挑出最相关的句子；另一种是生成式，让模型对文档做摘要，只保留跟问题相关的信息。抽取式速度快、成本低，适合对延迟敏感的场景；生成式效果好但会增加一次模型调用，适合对质量要求高的场景。

提示词模板的设计直接决定模型能不能用好检索到的上下文。我见过很多模板就是把文档一股脑塞进去然后问模型问题，这样模型很容易被无关内容带偏。好的模板会明确告诉模型：下面给你一些参考资料，请基于这些资料回答问题，如果资料里没有相关信息就明确说不知道，不要自己编。同时会给模型一些格式上的约束，比如要求分点回答、要求引用来源、要求控制回答长度。这些约束看起来是小事，但对输出稳定性的影响很大。

3.3 模型调用层的稳定性设计

模型调用看起来就是发个HTTP请求，但在生产环境里要考虑的事情很多。超时和重试是最基础的。模型推理的延迟波动很大，同样的请求可能这次200毫秒返回，下次要3秒。超时时间设置太短会导致大量请求失败，太长会让用户等太久。我的经验是设置一个分级超时，比如首token超时设5秒，整体超时设30秒，超过就触发重试或者降级。

降级策略是保证可用性的关键。当主模型不可用或者响应太慢的时候，要能自动切换到备用模型。备用模型可以是更小更快的版本，也可以是不同厂商的同类模型。降级的时候要注意提示词模板可能需要微调，因为不同模型对指令的遵循程度不一样。我通常会给每个模型维护一套适配过的模板，切换的时候连带模板一起切。

流式输出对用户体验的影响很大。用户不需要等整个回答生成完才能看到内容，首token返回之后就可以开始展示。实现流式输出要注意几个细节：一是要处理好流式返回的拼接和解析，特别是当输出需要做结构化提取的时候；二是要设置合理的缓冲策略，避免网络抖动导致输出断断续续；三是要考虑流式中断的情况，用户可能看到一半就不想看了，这时候要及时取消后端的生成请求，释放资源。

并发控制是另一个容易被忽略的点。模型API通常有速率限制，并发太高会被限流。我的做法是在调用层做一个令牌桶或者漏桶的限流器，控制单位时间内的请求数。同时要做好请求排队，超过并发上限的请求进入队列等待，而不是直接拒绝。队列的长度和等待超时也要设置合理，避免用户等太久。

3.4 输出解析与结构化提取的实操方案

模型返回的原始文本往往不能直接使用，需要做解析和结构化。最简单的场景是要求模型返回JSON，但实际用下来你会发现模型经常会在JSON外面包一层解释性文字，或者JSON格式有细微错误。我的处理流程是：先尝试直接解析，失败的话用正则提取JSON部分再解析，再失败的话用一个小模型做格式修复，最后还不行就返回兜底结果并记录日志。

对于需要提取特定字段的场景，我倾向于用函数调用或者结构化输出的能力，而不是靠提示词约束。现在主流模型基本都支持指定输出格式，让模型按照给定的schema返回，解析成功率比纯提示词约束高很多。如果模型不支持，那就退回到提示词约束加后处理校验的方案。

输出校验这步不能省。即使模型返回了格式正确的内容，也要做业务逻辑上的校验。比如要求返回日期，就要检查是不是合法日期；要求返回枚举值，就要检查是不是在允许范围内；要求返回数值，就要检查是不是在合理区间。校验不通过的输出，要么触发重试，要么走降级逻辑，绝对不能直接透传给用户。

提示：建议在开发阶段把每次模型调用的原始输入输出都完整记录下来，包括提示词、模型返回、解析结果、校验结果。这些数据在排查问题和做效果分析的时候非常有用，等到线上出问题再想加日志就来不及了。

4. 完整实操流程与关键环节实现

4.1 环境搭建与依赖管理

从零搭建的第一步是把环境弄干净。我强烈建议用虚拟环境或者容器来隔离依赖，AI工程涉及的东西太多，不同项目之间的依赖冲突很常见。Python环境下用venv或者conda都行，关键是每个项目独立。依赖管理用requirements.txt或者poetry，把版本号锁死，避免因为某个库自动升级导致行为变化。

核心依赖大概分这么几类：模型调用客户端，比如openai的官方库或者兼容接口的第三方库；向量计算库，比如numpy、faiss-cpu或者faiss-gpu；文本处理库，比如用于分词的jieba或者用于解析的beautifulsoup；Web框架，比如fastapi或者flask，用于对外提供服务；监控和日志库，比如prometheus客户端和结构化日志库。这些库的版本要仔细选，特别是faiss和numpy之间有版本兼容性要求，装错了会直接报错。

配置管理我习惯用环境变量加配置文件的方式。敏感信息比如API密钥只放环境变量，不写进代码和配置文件。业务配置比如模型名称、超时时间、检索参数放配置文件，方便不同环境切换。配置文件用YAML或者TOML格式，比JSON可读性好，支持注释。

4.2 索引构建的完整代码实现

索引构建的流程可以拆成四步：加载文档、切分文本、生成嵌入、写入索引。加载文档这步要根据数据源来定，如果是本地文件就用文件读取，如果是数据库就写查询，如果是网页就做抓取和解析。我通常会把加载逻辑抽象成一个接口，不同数据源实现不同的加载器，这样扩展起来方便。

文本切分我一般用递归切分的方式，先按大结构切，比如章节，再按小结构切，比如段落，最后按长度切。切分的时候维护一个元数据字典，记录每个片段的来源、位置、类型等信息。切分完的片段列表要做一个去重，完全相同的片段只保留一份，避免检索的时候重复召回。

嵌入生成这步要注意批处理。一条一条调嵌入接口太慢，通常一次批处理几十到几百条。批大小要根据接口的限制和内存情况来定，太大容易超时或者OOM，太小效率低。生成嵌入的时候要做好错误处理和重试，网络抖动导致的失败要能自动恢复。嵌入向量建议做归一化，这样后面算余弦相似度的时候可以直接用点积，速度快一些。

写入索引的时候，如果用的是FAISS，要注意索引类型的选择。小规模数据用Flat索引就行，精度最高。数据量大了考虑IVF或者HNSW，用一定的精度损失换速度。索引文件要定期持久化，避免进程重启后需要重新构建。如果数据会更新，还要考虑增量索引的方案，全量重建的成本太高。

import numpy as np import faiss from typing import List, Dict class VectorIndex: def __init__(self, dim: int): self.dim = dim self.index = faiss.IndexFlatIP(dim) # 内积索引，配合归一化向量使用 self.metadata: List[Dict] = [] def add(self, vectors: np.ndarray, metas: List[Dict]): # 向量归一化 faiss.normalize_L2(vectors) self.index.add(vectors) self.metadata.extend(metas) def search(self, query_vector: np.ndarray, top_k: int = 10): faiss.normalize_L2(query_vector) scores, indices = self.index.search(query_vector, top_k) results = [] for score, idx in zip(scores[0], indices[0]): if idx == -1: continue results.append({ "score": float(score), "metadata": self.metadata[idx] }) return results

上面这段代码是一个最简版的向量索引封装，实际项目里还要加上持久化、增量更新、多索引管理等逻辑。但核心思路就是这样：归一化向量、建索引、查询时归一化查询向量、返回相似度和元数据。

4.3 检索与重排序的串联实现

检索环节我通常做成两阶段：粗排召回和精排重排序。粗排用向量索引加关键词索引并行召回，各取Top-50，然后合并去重。合并的时候用RRF算法，它对不同来源的分数尺度不敏感，比较适合这种多路召回的场景。RRF的核心思想是根据每个文档在各路召回中的排名来计算综合得分，排名越靠前得分越高，公式是 score = sum(1 / (k + rank))，k一般取60。

精排阶段用重排序模型对粗排结果重新打分。重排序模型通常是交叉编码器，把查询和文档拼在一起输入模型，输出相关性分数。这种方式的精度比向量相似度高很多，但计算量大，所以只适合对少量候选做精排。重排序之后取Top-5或者Top-8作为最终上下文。

上下文组装的时候要注意顺序和格式。我的经验是把最相关的放最前面和最后面，中间放次相关的，因为模型对开头和结尾的内容注意力更集中。每个片段前面加上来源标记，方便模型引用，也方便后续做溯源。片段之间用分隔符隔开，避免模型把不同片段的内容混在一起。

def rrf_fusion(result_lists: List[List[Dict]], k: int = 60) -> List[Dict]: """RRF融合多路召回结果""" score_map = {} doc_map = {} for results in result_lists: for rank, item in enumerate(results): doc_id = item["metadata"]["doc_id"] if doc_id not in score_map: score_map[doc_id] = 0.0 doc_map[doc_id] = item score_map[doc_id] += 1.0 / (k + rank + 1) fused = [] for doc_id, score in sorted(score_map.items(), key=lambda x: -x[1]): item = doc_map[doc_id].copy() item["rrf_score"] = score fused.append(item) return fused

4.4 提示组装与模型调用的完整链路

提示组装这步看起来简单，但细节很多。系统指令要写清楚角色定位、任务目标、输出格式要求、约束条件。用户问题要原样保留，不要做过度改写，避免丢失信息。检索到的上下文要标注来源，并且明确告诉模型这些是参考资料，不是绝对正确的。历史对话要控制长度，太长的历史做摘要或者只保留最近几轮。

模型调用我封装成一个统一的客户端，对外暴露一个generate方法，内部处理重试、降级、流式、日志等逻辑。调用的时候传入提示词、模型名称、温度、最大token数等参数。温度这个参数要根据任务类型来定，事实性问答用低温度比如0.1到0.3，创意生成用高温度比如0.7到0.9。最大token数要设置合理，太小会导致回答被截断，太大浪费资源。

流式输出的处理要特别注意。如果最终输出需要做结构化解析，流式返回的片段要缓存起来，等全部返回完再解析。如果只是展示给用户看，可以边收边展示，但要处理好markdown格式的渲染，避免因为片段不完整导致格式错乱。

class LLMClient: def __init__(self, primary_model: str, fallback_model: str, timeout: int = 30): self.primary_model = primary_model self.fallback_model = fallback_model self.timeout = timeout def generate(self, prompt: str, temperature: float = 0.3, max_tokens: int = 1024): try: return self._call(self.primary_model, prompt, temperature, max_tokens) except Exception as e: # 记录主模型失败日志 print(f"Primary model failed: {e}, falling back") return self._call(self.fallback_model, prompt, temperature, max_tokens) def _call(self, model: str, prompt: str, temperature: float, max_tokens: int): # 实际调用逻辑，包含重试 for attempt in range(3): try: # 这里替换成实际的模型调用 response = call_model_api(model, prompt, temperature, max_tokens, self.timeout) return response except TimeoutError: if attempt == 2: raise continue

4.5 评估体系的搭建与指标定义

没有评估就没有优化。AI工程的评估分两个层面：组件级评估和端到端评估。组件级评估针对检索、重排序、解析这些单独模块，端到端评估看最终回答的质量。

检索评估的核心指标是召回率和精确率。准备一批查询，人工标注每个查询对应的相关文档，然后看检索结果里有多少相关文档被召回了，召回的里面有多少是真正相关的。重排序评估看的是排序质量，用NDCG或者MRR这类指标。解析评估看的是结构化提取的准确率和格式合规率。

端到端评估更复杂一些，因为回答质量很难用单一指标衡量。我通常从几个维度来评：事实准确性，回答里的信息是否跟知识库一致；完整性，是否覆盖了问题的所有方面；相关性，是否回答了用户真正想问的；格式合规性，是否符合要求的输出格式。每个维度可以人工打分，也可以用模型来打分，但模型打分需要先做校准。

评估数据集的构建是个持续的过程。初期可以手工构造一批典型查询，覆盖主要场景。上线之后从真实用户请求里采样，把有问题的case补充进评估集。评估集要定期更新，避免过拟合到旧数据上。

5. 常见问题与排查技巧实录

5.1 检索效果差的排查路径

检索效果差是最常见的问题，排查的时候按这个顺序来。先看查询本身，用户的问题是不是太短或者太模糊，如果是的话考虑做查询改写或者扩展。再看嵌入模型，是不是模型跟你的领域不匹配，通用模型在垂直领域的效果可能很差，考虑换模型或者做微调。然后看切分粒度，片段是不是太大或者太小，太大导致噪声多，太小导致信息不完整。接着看索引类型，Flat索引精度最高，如果用了IVF或者HNSW，调一下nprobe或者efSearch参数。最后看相似度阈值，是不是设得太高导致相关文档被过滤掉了。

我遇到过一个典型案例，用户问的是产品的一个具体功能怎么用，但检索出来的全是产品介绍和营销文案。排查下来发现是切分粒度的问题，产品文档被按固定长度切了，功能介绍和操作步骤被切到了不同片段里，检索的时候只召回了介绍部分。改成按章节切分之后问题就解决了。

5.2 模型输出不稳定的应对策略

模型输出不稳定表现为同样的问题有时候回答得好，有时候答非所问，有时候格式不对。这个问题要从几个方面入手。温度参数先调低，事实性任务温度设0.1到0.2，基本能消除大部分随机性。提示词要写得更明确，把要求一条一条列清楚，避免模糊表述。输出格式用结构化输出能力来约束，比纯提示词约束可靠得多。上下文要控制好，无关内容太多会干扰模型，该压缩就压缩，该过滤就过滤。

还有一个容易被忽略的点是模型版本。同一个模型的不同版本行为可能差异很大，升级模型版本之后一定要重新跑评估集，确认效果没有退化。我习惯在配置里把模型版本号写死，避免自动升级带来的意外。

5.3 性能瓶颈的定位与优化

性能问题通常出在几个地方。嵌入计算如果是实时做的，会显著增加延迟，能预计算的就预计算。向量检索在数据量大或者索引类型不合适的时候会变慢，考虑换索引类型或者加缓存。模型调用是最大的延迟来源，能流式就流式，能并行就并行，能用小模型就用小模型。网络传输在跨区域调用的时候影响很大，尽量让服务离模型端点近一些。

优化的时候要有数据支撑，不能凭感觉。在每个环节打点记录耗时，找出真正的瓶颈再动手。我见过团队花大力气优化检索，结果发现90%的时间花在模型调用上，优化方向完全错了。

5.4 成本控制的实操经验

AI应用的成本主要是token消耗。控制成本的核心是减少不必要的token。上下文压缩是最直接的手段，把长文档压缩成关键句，能省很多token。缓存是另一个手段，相同或者相似的问题直接返回缓存结果，不用重复调用模型。模型分级也很重要，简单任务用小模型，复杂任务才用大模型。

我通常会做一个token预算管理，给每个请求设置token上限，超过就触发压缩或者截断。同时监控每天的token消耗，设置告警阈值，避免因为异常流量导致成本失控。缓存要注意失效策略，知识库更新之后相关缓存要及时清除，避免返回过时信息。

注意：缓存相似度匹配的时候要小心，太宽松会导致返回不相关的缓存结果，太严格又起不到缓存效果。建议用向量相似度做缓存匹配，阈值设高一些比如0.95以上，只缓存几乎完全相同的问题。

6. 迭代方向与个人实操体会

这套从零搭建的AI工程体系，我在几个项目里跑下来，最大的体会是前期慢就是快。花时间把每一层的边界划清楚，把评估体系建起来，后面迭代的时候效率会高很多。相反，如果一开始图快直接上框架，后面每改一个地方都要跟框架的抽象做斗争，反而更慢。

后续的迭代方向，我觉得有几个值得深入。一是检索策略的自动化调优，现在参数还是靠人工试，可以做成基于评估集的自动搜索。二是提示词的版本管理，把提示词当成代码来管理，每次修改都跑评估，避免改坏。三是多模型路由，根据查询类型自动选择最合适的模型，在效果和成本之间找平衡。四是在线学习，把用户的反馈信号收集起来，持续优化检索和生成策略。

最后分享一个小技巧。我在每个项目里都会维护一个问题案例库，把线上遇到的bad case记录下来，标注问题类型和修复方案。这个库既是排查问题的参考，也是评估集的来源，还是团队知识沉淀的载体。坚持记下来，几个月之后你会发现这是项目里最有价值的资产之一。