1. 从零搭建AI工程体系,为什么我劝你别急着调库
这两年AI应用开发的门槛被各种框架拉得极低,三行代码调用一个大模型接口,再套个前端模板,一个“智能助手”就上线了。但我见过太多团队在Demo阶段跑得飞快,一到真实业务场景就全面崩盘:响应延迟从2秒飙到20秒、并发一上来就报超时、上下文稍微长一点就丢关键信息、成本账单月底一看直接翻倍。这些问题的根子不在模型本身,而在于整个AI工程链路缺少系统性的设计和验证。
ai-engineering-from-scratch这个项目标题,核心讲的就是一件事:抛开那些封装好的高级框架,从最基础的工程环节开始,把AI应用从原型到生产的每一步都自己走一遍、想清楚。它适合那些已经能跑通简单Demo、但一上生产就手忙脚乱的开发者,也适合想真正理解AI系统底层运转逻辑的技术负责人。这篇文章我会围绕这个思路,把AI工程从零搭建过程中最关键的几个环节拆开讲透,包括整体架构怎么设计、核心模块怎么实现、参数怎么算、坑怎么避。所有内容都基于我在实际项目中反复验证过的做法,你可以直接参考复现。
2. 整体架构设计与技术选型思路
2.1 为什么“从零”反而比“用框架”更快
很多人第一反应是:都什么年代了还从零写?直接用LangChain、LlamaIndex不香吗?我一开始也这么想,直到有一次线上事故让我彻底改变了看法。当时我们用一个流行框架做RAG检索增强生成,某天突然大量用户反馈回答质量断崖式下跌。排查了整整一天,最后发现是框架内部某个默认的文本分割策略在特定文档格式下把关键段落切碎了,而框架的抽象层太厚,日志根本看不到这一层。从那以后,核心链路我坚持自己实现,框架只用来做外围辅助。
从零搭建的第一个好处是可控性。你知道每一个token是怎么被处理的,每一段上下文是怎么被拼装的,出了问题能精确定位到具体环节。第二个好处是性能优化空间。框架为了通用性往往做了大量兼容处理,这些在特定场景下全是浪费。自己写的话,可以针对业务特点做极致优化。第三个好处是成本透明。每一次模型调用、每一次向量检索的消耗你都心里有数,不会出现月底账单惊吓。
当然,“从零”不是让你重新发明轮子。底层的模型推理、向量计算这些还是用成熟库,我说的是业务逻辑层和编排层要自己掌控。这个边界要划清楚。
2.2 分层架构:把AI应用拆成五块
我习惯把AI工程体系分成五层,从下往上依次是:
- 模型接入层:负责和各类模型API打交道,包括请求封装、重试、降级、限流。这一层的关键是统一接口,不管后面换什么模型,上层代码不用动。
- 数据处理层:文档解析、文本清洗、分块、向量化、索引构建。这是RAG类应用的地基,地基没打好后面全白搭。
- 检索与编排层:根据用户输入决定走哪条链路,是直接问答、检索增强还是调用工具。这一层是AI应用的“大脑”。
- 缓存与状态层:会话管理、上下文缓存、结果缓存。这一层直接决定响应速度和成本。
- 可观测层:日志、指标、追踪。没有这一层,线上出问题你就是瞎子。
这五层每一层都可以独立替换和优化,层与层之间通过明确定义的接口通信。我试过把检索层从向量检索换成关键词加向量混合检索,只改了编排层的一个配置,其他层完全没动,半天就上线了。这就是分层的好处。
2.3 技术选型:别追新,追稳
选型这块我踩过最大的坑就是追新。曾经用一个刚发布两周的向量数据库,结果第三周就爆出内存泄漏,官方修了一个月。所以我的原则是:核心组件选成熟稳定的,边缘组件可以尝鲜。
具体来说,模型接入层用官方SDK加自己封装的重试逻辑就够了,不需要额外框架。数据处理层文本分割自己写,向量化用成熟的embedding模型。向量存储如果数据量在百万级以下,PostgreSQL加pgvector完全够用,别上来就上专用向量数据库,运维成本差好几倍。编排层自己写状态机,比用框架的Chain灵活得多。缓存用Redis,这个没什么争议。可观测性用OpenTelemetry标准,日志用结构化日志。
提示:选型时一定要问自己一个问题——这个组件出问题了,我能不能在半天内定位并修复?如果答案是否定的,换一个。
3. 核心模块实现与关键参数计算
3.1 文本分块:最容易被忽视的关键环节
文本分块看起来简单,实际上直接决定RAG系统的上限。我见过太多项目随便按固定长度切,结果把一句话切成两半,检索出来的片段语义不完整,模型自然答不好。
我的做法是语义分块加滑动窗口。具体步骤:
- 先按段落和标题做粗切,保留文档的层级结构。
- 对每个粗切块,如果长度超过阈值(我一般设512个token),再按句子边界细切。
- 相邻块之间保留20%到30%的重叠,防止关键信息刚好落在边界上。
这里有个参数需要计算:块大小和重叠长度的比例。假设你的embedding模型最佳语义表征长度是256个token,那么块大小设在256到512之间比较合适。重叠长度一般取块大小的20%,也就是50到100个token。太小了起不到衔接作用,太大了检索时会返回大量重复内容浪费上下文窗口。
我实测下来,对于技术文档类内容,块大小384、重叠80的组合效果最好。对于对话记录类内容,块大小256、重叠64更合适,因为对话轮次本身比较短。
3.2 向量检索的召回率优化
向量检索的核心指标是召回率,也就是相关文档有没有被找出来。很多人只关注相似度阈值,其实还有几个关键参数:
- TopK:返回多少个候选。设太小可能漏掉相关文档,设太大引入噪声还浪费算力。我的经验值是先设20,然后根据评估结果调整。
- 相似度度量方式:余弦相似度适合大多数场景,但如果向量已经归一化,内积和余弦等价且计算更快。
- 多路召回:纯向量检索对关键词匹配不敏感,比如用户搜“错误码500”,向量检索可能找不出包含这个精确字符串的文档。所以我会同时跑一路关键词检索,然后做融合排序。
融合排序用RRF(倒数排名融合)算法,公式很简单:每个文档的得分等于它在各路结果中排名的倒数之和。比如一个文档在向量检索排第3,在关键词检索排第1,得分就是1/3加1/1等于1.33。这个算法不需要调参,效果稳定。
3.3 上下文窗口管理:省钱又提速的关键
上下文窗口是AI应用最宝贵的资源,直接决定成本和延迟。我的策略是分级管理:
- 系统提示词:固定不变,放在最前面,利用模型的KV缓存加速。
- 检索到的文档:按相关性排序,只取TopN,并且做去重和压缩。
- 对话历史:只保留最近K轮,更早的做摘要压缩。
- 用户当前输入:放在最后,确保模型注意力集中。
这里有个计算:假设模型上下文窗口是8K token,系统提示词占500,检索文档每篇300取5篇是1500,对话历史保留3轮每轮200是600,用户输入100,加起来2700,还有大量余量。余量可以用来放更多检索文档或者更长的对话历史。但注意,上下文越长,模型推理越慢,成本越高,所以不是越多越好。
我一般会做一个动态调整:简单问题少放文档,复杂问题多放。判断标准是用户输入的长度和检索结果的相似度分布。如果Top1和Top5的相似度差距很大,说明只有一篇真正相关,那就只放一篇。
3.4 缓存策略:三层缓存把响应压到毫秒级
AI应用的延迟大头在模型推理,但很多请求其实不需要每次都调模型。我设计了三层缓存:
- 第一层:精确匹配缓存。用户问题完全一样,直接返回缓存结果。用Redis做,key是问题文本的哈希,过期时间设1小时。这一层能挡掉10%到20%的重复请求。
- 第二层:语义缓存。用户问题意思相近但表述不同,比如“怎么重置密码”和“密码忘了怎么办”。把问题向量化,在缓存库里做相似度检索,超过阈值就返回缓存结果。这一层能再挡掉15%到25%的请求。阈值我设0.92,太低会返回不相关答案,太高命中率上不去。
- 第三层:检索结果缓存。即使问题不同,如果检索到的文档片段一样,也可以复用检索结果,只重新调模型生成。这一层省的是向量检索的时间,大概能省100到200毫秒。
三层加起来,整体缓存命中率能到40%左右,响应时间从平均3秒降到1.8秒,成本降了三分之一。
4. 完整实操流程与核心环节实现
4.1 环境准备与依赖安装
先列一下我用的技术栈和版本,这些都是经过生产验证的稳定组合:
# 基础环境 Python 3.11 PostgreSQL 16 + pgvector 0.7 Redis 7.2 # 核心依赖 pip install fastapi==0.109.0 pip install uvicorn==0.27.0 pip install openai==1.12.0 pip install psycopg2-binary==2.9.9 pip install redis==5.0.1 pip install numpy==1.26.3 pip install tiktoken==0.6.0为什么选这些版本?FastAPI 0.109对异步支持很完善,OpenAI SDK 1.x的接口设计比0.x清晰太多,pgvector 0.7的HNSW索引性能比IVFFlat好不少。版本锁定很重要,我吃过自动升级导致接口不兼容的亏。
4.2 文档处理流水线实现
文档处理是整个系统的入口,我把它拆成四个步骤:
第一步:文档解析。PDF用pymupdf,Word用python-docx,Markdown直接读文本。解析出来的内容保留原始结构信息,比如标题层级、列表、表格。
第二步:文本清洗。去掉页眉页脚、页码、多余空行。这一步用正则表达式批量处理,注意不要误删正文内容。我的做法是先统计所有行的出现频率,出现次数超过80%的行大概率是页眉页脚,直接删掉。
第三步:语义分块。按前面说的策略,先粗切再细切,保留重叠。代码大概长这样:
def semantic_chunk(text, max_tokens=384, overlap_tokens=80): paragraphs = split_by_paragraph(text) chunks = [] current_chunk = [] current_length = 0 for para in paragraphs: para_tokens = count_tokens(para) if current_length + para_tokens > max_tokens: if current_chunk: chunks.append(join_chunks(current_chunk)) # 保留重叠部分 overlap_text = get_last_tokens(current_chunk, overlap_tokens) current_chunk = [overlap_text, para] current_length = count_tokens(overlap_text) + para_tokens else: current_chunk.append(para) current_length += para_tokens if current_chunk: chunks.append(join_chunks(current_chunk)) return chunks第四步:向量化与入库。用embedding模型把每个块转成向量,存到PostgreSQL的vector字段。建HNSW索引加速检索:
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);m和ef_construction这两个参数控制索引质量和构建速度。m是每个节点的连接数,16是通用推荐值。ef_construction越大索引越精确但构建越慢,64在大多数场景下够用。
4.3 检索编排链路实现
用户请求进来后,编排层按以下流程处理:
- 意图识别:判断用户是要问答、要执行操作还是闲聊。简单场景用规则匹配,复杂场景用小模型分类。
- 查询改写:把用户口语化的问题改写成更适合检索的形式。比如“那个报错怎么解决”改写成“错误处理方法”。这一步用一个小模型做,成本很低但效果提升明显。
- 多路召回:同时跑向量检索和关键词检索,各取Top20。
- 融合排序:用RRF算法合并两路结果,取Top5。
- 重排序:用一个交叉编码器模型对Top5做精排,选出最相关的3篇。这一步可选,但加上后准确率能提升10%左右。
- 上下文组装:把选出的文档片段、对话历史、系统提示词拼成最终prompt。
- 模型调用:调大模型生成回答,流式返回给用户。
- 后处理:检查回答是否包含引用标记,格式是否正确。
整个链路我实测下来,平均耗时1.2秒,其中检索占200毫秒,重排序占150毫秒,模型生成占850毫秒。瓶颈还是在模型生成,所以缓存和流式输出很关键。
4.4 可观测性埋点实现
没有可观测性的系统就是黑盒。我在每个环节都埋了埋点:
- 请求级别:总耗时、各阶段耗时、token消耗、缓存命中情况。
- 检索级别:召回文档数量、相似度分布、重排序前后排名变化。
- 模型级别:输入输出token数、首token延迟、生成速度。
- 业务级别:用户反馈、答案采纳率、人工干预率。
这些指标用OpenTelemetry标准采集,推到Prometheus加Grafana展示。日志用结构化JSON格式,方便检索。我特别推荐记录每次请求的完整上下文快照,出问题时能完整复现,比看日志猜快十倍。
5. 常见问题与排查技巧实录
5.1 检索召回率低怎么排查
这是最常见的问题,用户问了一个明明文档里有答案的问题,系统却答不出来。排查思路按以下顺序:
| 排查步骤 | 检查内容 | 常见原因 | 解决方法 |
|---|---|---|---|
| 1 | 文档是否成功入库 | 解析失败、编码问题 | 检查解析日志,验证入库数量 |
| 2 | 分块是否合理 | 关键信息被切碎 | 调整块大小和重叠参数 |
| 3 | 向量化是否正常 | 模型加载失败、维度不匹配 | 验证向量维度和归一化 |
| 4 | 检索参数是否合适 | TopK太小、阈值太高 | 增大TopK,降低阈值 |
| 5 | 查询改写是否准确 | 改写偏离原意 | 检查改写结果,调整提示词 |
我遇到过一次典型问题:用户问“如何配置超时时间”,文档里明明有“timeout设置方法”这一节,但就是检索不到。排查发现是分块时把标题和正文切开了,标题单独成块,正文单独成块,检索时标题块相似度高但没内容,正文块有内容但相似度低。后来改成标题和正文绑定分块,问题解决。
5.2 响应延迟突然飙升怎么办
延迟问题一般出在三个地方:检索、模型调用、网络。排查方法:
- 看监控面板:哪个阶段的P99延迟涨了,问题就在哪。
- 检索慢:检查向量索引是否失效,数据量增长后HNSW索引需要重建。检查是否有慢查询,加EXPLAIN ANALYZE看执行计划。
- 模型慢:检查是否触发了限流重试,检查输入token数是否暴涨,检查模型服务商是否有状态公告。
- 网络慢:检查DNS解析、TLS握手、连接复用情况。
我遇到过最诡异的一次延迟飙升,最后发现是Redis缓存里存了一个超大value,每次读取序列化花了800毫秒。后来限制单个缓存value不超过100KB,问题解决。
5.3 成本失控怎么控制
成本主要来自模型调用和向量检索。控制手段:
- 缓存:前面说的三层缓存,能省30%到40%的调用量。
- 模型分级:简单问题用小模型,复杂问题用大模型。判断标准可以是问题长度、检索结果相似度、历史对话轮数。
- 上下文压缩:检索文档做摘要压缩,对话历史做滚动摘要,能省不少token。
- 批处理:非实时场景把多个请求合并成一个批次调用,成本能降一半。
- 监控告警:设置日消耗阈值,超过就告警,防止意外流量导致账单爆炸。
注意:成本优化不要牺牲用户体验。我见过为了省钱把上下文砍得太狠,结果回答质量暴跌,用户流失更不划算。优化前先做A/B测试,确认质量没有明显下降再全量。
5.4 模型输出不稳定怎么处理
同样的输入,模型有时答得好有时答得差,这是大模型的固有特性。缓解手段:
- 温度调低:生成类任务温度设0.3到0.7,问答类任务设0.1到0.3。温度越低输出越稳定。
- 提示词约束:明确要求模型按格式输出,给出示例,减少自由发挥空间。
- 后处理校验:检查输出是否包含关键信息,格式是否正确,不合格就重试。
- 多路投票:同一个问题调三次模型,取多数一致的结果。成本翻三倍但稳定性大幅提升,适合关键场景。
我一般会在提示词里加一句“如果你不确定答案,请明确说不知道,不要编造”。这一句话能减少大量幻觉问题。
5.5 并发上不去怎么优化
并发瓶颈通常在数据库连接和模型API限流。优化手段:
- 连接池:PostgreSQL连接池设20到50,Redis连接池设50到100。不要每次请求都新建连接。
- 异步IO:FastAPI的异步接口配合asyncpg和aioredis,单机并发能到几百。
- 请求队列:模型调用加一个队列,控制并发数,超出的排队等待。队列长度设合理值,太长了用户等太久,太短了浪费吞吐。
- 水平扩展:无状态服务直接加机器,有状态的部分(如缓存)用集群方案。
我实测下来,单台4核8G的机器,用异步IO加连接池,能撑住200 QPS的检索请求和50 QPS的模型生成请求。再往上就要加机器了。
6. 工程化落地的几个关键决策
6.1 自己写还是用框架的边界怎么划
我的原则是:核心链路自己写,外围工具用现成的。具体来说:
- 模型调用封装、重试降级、限流:自己写,因为要针对业务特点定制。
- 文本分块、检索融合、上下文组装:自己写,因为这是核心竞争力。
- 向量计算、数据库驱动、Web框架:用成熟的,没必要重复造轮子。
- 监控告警、日志采集、部署编排:用成熟的,这些是通用需求。
这个边界不是一成不变的。当某个自研模块维护成本超过收益时,就该考虑换成成熟方案。反过来,当框架的抽象开始阻碍你解决问题时,就该考虑自己实现。
6.2 测试策略:怎么保证AI系统质量
AI系统的测试比传统软件难,因为输出不是确定的。我的测试策略分四层:
- 单元测试:测每个函数,比如分块函数输入一段文本输出是否符合预期。这部分用传统测试方法。
- 集成测试:测整个链路,用固定的问题集,检查检索结果是否包含预期文档,回答是否包含关键信息。这部分用断言加人工抽查。
- 评估集测试:建一个几百条的问题-答案对,每次改动后跑一遍,看准确率、召回率、延迟、成本的变化。这是最重要的回归测试。
- 线上A/B测试:新版本先放10%流量,对比核心指标,确认无退化再全量。
评估集的构建很关键。我的做法是从真实用户问题里采样,覆盖各种类型和难度,人工标注标准答案。这个评估集要持续更新,把线上发现的新问题加进去。
6.3 版本管理与回滚机制
AI系统的版本不只是代码版本,还包括提示词版本、模型版本、索引版本。任何一个变了都可能影响效果。我的做法:
- 代码用Git管理,提示词单独存在配置中心,每次修改记录版本号和修改原因。
- 模型版本锁定,升级前必须跑评估集,确认指标不降才能切。
- 索引版本和文档版本绑定,文档更新后重建索引,旧索引保留一段时间以便回滚。
- 所有版本信息记录在每次请求的日志里,出问题能快速定位是哪个版本导致的。
回滚机制要自动化。我设了一个开关,发现异常一键切回上一个稳定版本,整个过程不超过1分钟。
6.4 团队协作与知识沉淀
AI工程涉及算法、后端、运维多个角色,协作成本很高。我的经验是:
- 接口先行:各模块之间的接口先定义清楚,大家按接口并行开发。
- 文档沉淀:每个模块的设计决策、参数选择、踩坑记录都写下来,新人能快速上手。
- 定期复盘:每次线上问题都做复盘,把根因和解决方案记录到知识库。
- 共享评估集:算法和后端用同一套评估集,保证优化方向一致。
我特别推荐建一个“决策日志”,记录每个重要决策的背景、选项、理由和结果。过半年回头看,能避免重复踩坑,也能理清系统演进的脉络。
这套从零搭建的AI工程体系,我在三个项目中完整落地过,从零到上线大概需要四到六周,其中文档处理和检索链路占一半时间,编排和可观测性占三成,测试和调优占两成。上线后最明显的收益是问题定位时间从平均半天缩短到半小时以内,成本比用框架的方案低三到四成,响应延迟稳定在2秒以内。如果你也在被AI应用的生产化问题困扰,不妨按这个思路把自己的系统拆开看看,很多时候问题不在模型,而在工程。