1. 为什么我要从零手搓一套AI工程流水线第一次听到“ai-engineering-from-scratch”这个说法是在一个做推荐系统的老哥群里。有人甩了张截图说现在招AI工程师JD上写的全是“熟悉LangChain、会调OpenAI API、了解RAG”结果进来的人连embedding维度怎么选、向量检索的召回率怎么算都说不清楚。底下有人回了一句“都是调包侠真让他们从零搭一套全得露馅。”这话我记了很久。我自己带过几个项目从最早的规则引擎到后来的深度学习模型再到这两年的大模型应用踩过的坑比吃过的盐还多。我的体会是AI工程不是把模型跑起来就完事了而是要让模型在真实业务里稳定、可控、可迭代地干活。而“from scratch”的意思不是让你从零实现Transformer而是让你理解每一层抽象下面到底发生了什么这样出了问题你才知道往哪儿查。这篇文章适合谁看如果你是刚转行做AI应用的开发者或者已经会用现成框架但总觉得心里没底再或者你是技术负责人想给团队搭一套靠谱的AI工程底座那这篇内容应该能帮到你。我会把整个从零搭建AI工程体系的过程拆开讲清楚每个环节为什么这么设计、参数怎么算、坑在哪里。全文基于我自己的实操经验不是教科书搬运。2. 整体架构设计与技术选型思路2.1 先想清楚AI工程到底要解决什么问题很多人一上来就纠结用PyTorch还是TensorFlow用LangChain还是LlamaIndex。我的建议是先把问题定义清楚。AI工程的核心任务说白了就三件事数据怎么进来、模型怎么跑、结果怎么出去。但每一件事往下拆都是一堆细节。数据进来你要考虑数据源是结构化的还是非结构化的是批量导入还是实时流式需不需要做清洗和标注。模型跑起来你要考虑是本地推理还是调API是单模型还是多模型编排推理延迟和成本怎么平衡。结果出去你要考虑输出格式怎么约束怎么评估质量怎么收集反馈做迭代。我见过太多项目一开始就上最火的框架结果业务逻辑和框架耦合太深换个模型要改半个月。所以我的设计原则是分层解耦每一层都可以独立替换。具体来说我把它分成四层数据接入层、模型服务层、编排逻辑层、应用接口层。层与层之间通过明确定义的接口通信这样你换向量数据库也好换大模型也好都不会牵一发动全身。2.2 技术选型的取舍逻辑选型这块我踩过最大的坑就是“追新”。有次为了用某个刚出的向量数据库结果社区文档不全出了bug只能自己啃源码项目延期了两周。后来我定了个规矩核心组件必须满足三个条件——社区活跃、文档完整、有生产环境案例。具体到各个模块我的选型思路是这样的。数据接入层如果是文本数据我一般用Python的pandas做批处理配合FastAPI做流式接口。向量化这块早期我用过Sentence-BERT后来发现对于中文场景BGE系列模型效果更稳而且有不同尺寸可选小到base大到large可以根据延迟要求灵活切换。向量数据库我对比过Milvus、Qdrant和Chroma最后选了Qdrant原因是它的过滤查询性能好而且Rust写的内存占用可控。模型服务层如果是调API我会封装一层统一的LLM Client把重试、限流、日志都做在里面。如果是本地部署vLLM是目前比较省心的选择PagedAttention对显存利用率提升明显。编排逻辑层我没有用LangChain而是自己写了一套轻量的Chain抽象原因是LangChain的抽象层级太多调试的时候堆栈能打印两屏自己写反而清晰。应用接口层就是FastAPI加Pydantic做请求校验和响应序列化。提示选型时不要只看benchmark数字要看你的实际场景。比如向量检索如果数据量在百万级以下Chroma完全够用没必要上Milvus集群。2.3 目录结构怎么组织才不乱项目一开始的目录结构如果没设计好后面会越来越乱。我习惯按功能模块划分而不是按文件类型划分。比如ai-engineering-from-scratch/ ├── configs/ # 配置文件 ├── data_pipeline/ # 数据接入与清洗 ├── embedding/ # 向量化模块 ├── vector_store/ # 向量存储与检索 ├── llm/ # 大模型调用与本地推理 ├── orchestration/ # 编排逻辑 ├── evaluation/ # 评估模块 ├── api/ # 对外接口 └── tests/ # 测试每个模块内部再分core.py、schemas.py、utils.py。这样你找代码的时候先定位模块再定位文件不会在一堆utils里迷路。配置文件我统一用YAML因为支持注释而且嵌套结构比JSON好读。环境变量用.env管理敏感信息绝不进代码库。3. 核心模块拆解与实操要点3.1 数据接入别小看清洗这一步数据接入看起来简单实际上是最容易埋雷的地方。我做过一个客服问答项目原始数据是从工单系统导出的CSV里面混了HTML标签、特殊字符、还有重复记录。如果直接拿去做向量化检索出来的结果全是乱码。我的处理流程是这样的先用pandas读进来做一轮基础清洗——去HTML标签用BeautifulSoup去特殊字符用正则去重按内容哈希。然后做分块分块策略很关键。对于问答对我按问答边界切对于长文档我按语义段落切块大小控制在256到512个token之间。为什么是这个范围因为太小了语义不完整太大了检索精度下降。实测下来512token的块在召回率和生成质量之间平衡得比较好。分块之后要做元数据标注。每条数据至少要有来源、时间、类别这三个字段。来源用于追溯时间用于时效性排序类别用于过滤。这些元数据在检索时可以当过滤条件用比如用户问“最近的退货政策”你就可以按时间过滤。注意清洗规则一定要写成可配置的不要硬编码。不同数据源的脏数据模式不一样硬编码意味着每来一个新源就要改代码。3.2 向量化模型选型和维度计算向量化就是把文本变成一串数字让语义相近的文本在向量空间里距离也相近。模型选型上我试过OpenAI的text-embedding-ada-002效果确实好但成本和数据隐私是问题。后来转向开源模型BGE-large-zh在中文检索任务上表现很稳维度是1024。如果对延迟敏感可以用BGE-small-zh维度512速度快一倍多效果下降大概5%到8%。维度怎么选不是越高越好。维度高意味着存储和计算成本高而且到一定程度后边际收益递减。我的经验是对于百万级以下的数据量512到768维足够千万级以上可以考虑1024维。另外要注意不同模型的向量空间不兼容你不能用A模型建库用B模型查询那样距离计算完全没意义。向量化的时候要批量处理不要一条一条来。我一般设batch_size为64或128具体看显存。CPU上跑的话batch_size小一点32就行。还有个小技巧向量化之前把文本做一次归一化去掉多余空格和换行这样能减少噪声。from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-large-zh-v1.5) texts [你的文本1, 你的文本2] embeddings model.encode(texts, batch_size64, normalize_embeddingsTrue)normalize_embeddingsTrue这个参数很重要它把向量归一化到单位长度这样余弦相似度就可以直接用点积算快很多。3.3 向量检索HNSW参数怎么调向量数据库的核心是索引。Qdrant默认用HNSW这个索引有几个关键参数m、ef_construct、ef_search。m是每个节点的连接数越大索引越精确但内存占用越高一般设16到32。ef_construct是建索引时的候选集大小越大建索引越慢但质量越高一般设100到200。ef_search是查询时的候选集大小越大召回率越高但延迟越大一般设50到100。我实测过一组数据10万条向量768维m16ef_construct100ef_search50的情况下召回率大概92%查询延迟在5毫秒左右。把ef_search提到100召回率到96%延迟到8毫秒。所以如果你的场景对召回率要求极高可以适当调大ef_search但要注意延迟预算。检索的时候还有个技巧混合检索。纯向量检索对关键词匹配不敏感比如用户搜“iPhone 15 Pro Max”向量检索可能返回一堆手机相关的内容但不一定是这个型号。这时候可以加一层关键词过滤或者用BM25做粗排再用向量做精排。Qdrant支持payload过滤你可以把关键词作为payload字段存进去查询时加过滤条件。3.4 大模型调用重试、限流、降级一个都不能少调大模型API看起来就是发个HTTP请求但生产环境里要考虑的事情很多。首先是重试网络抖动、服务端限流都会导致请求失败我一般设3次重试指数退避初始间隔1秒。其次是限流不同API的QPS限制不一样要在客户端做令牌桶限流避免触发服务端封禁。最后是降级如果主模型不可用要有备用模型顶上哪怕效果差一点也比服务挂掉强。我封装了一个LLM Client核心逻辑是这样的import time import requests from tenacity import retry, stop_after_attempt, wait_exponential class LLMClient: def __init__(self, api_key, base_url, model, fallback_modelNone): self.api_key api_key self.base_url base_url self.model model self.fallback_model fallback_model self.rate_limiter TokenBucket(rate10, capacity20) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) def _call(self, model, messages, **kwargs): self.rate_limiter.acquire() resp requests.post( f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, json{model: model, messages: messages, **kwargs}, timeout30 ) resp.raise_for_status() return resp.json() def chat(self, messages, **kwargs): try: return self._call(self.model, messages, **kwargs) except Exception as e: if self.fallback_model: return self._call(self.fallback_model, messages, **kwargs) raise e这里TokenBucket是自己实现的令牌桶tenacity库做重试。注意timeout一定要设不设的话请求可能挂住几分钟把线程池占满。提示API Key绝对不要硬编码在代码里也不要在日志里打印。我见过有人把Key打在日志里结果日志被传到公共平台第二天就被刷爆了。3.5 编排逻辑自己写Chain比用框架更可控编排逻辑就是把检索、模型调用、后处理串起来。LangChain确实方便但它的抽象层太厚出问题的时候你很难定位是检索的问题还是模型的问题。我自己写了一套轻量的Chain核心是一个Pipeline类每个步骤是一个Step步骤之间通过Context传递数据。class Context: def __init__(self): self.data {} class Step: def run(self, ctx: Context) - Context: raise NotImplementedError class RetrieveStep(Step): def __init__(self, retriever): self.retriever retriever def run(self, ctx): query ctx.data[query] docs self.retriever.search(query, top_k5) ctx.data[docs] docs return ctx class GenerateStep(Step): def __init__(self, llm_client): self.llm_client llm_client def run(self, ctx): prompt build_prompt(ctx.data[query], ctx.data[docs]) resp self.llm_client.chat([{role: user, content: prompt}]) ctx.data[answer] resp[choices][0][message][content] return ctx class Pipeline: def __init__(self, steps): self.steps steps def run(self, query): ctx Context() ctx.data[query] query for step in self.steps: ctx step.run(ctx) return ctx.data这样每一步都可以单独测试日志也清晰。如果某个步骤出问题你一眼就能看出来。而且你可以很方便地在步骤之间加缓存、加监控、加A/B测试。4. 完整实操流程从零搭一个问答系统4.1 环境准备与依赖安装我假设你用的是Linux或者macOSPython版本3.10以上。先建虚拟环境这是好习惯避免污染系统Python。python -m venv venv source venv/bin/activate pip install fastapi uvicorn pydantic sentence-transformers qdrant-client openai tenacity pandas beautifulsoup4 pyyaml python-dotenv如果你要用本地推理再加vllm或者transformers。注意vllm对CUDA版本有要求装之前先看官方文档。4.2 数据准备与向量入库假设你有一批文档放在data/raw/目录下格式是txt。先写个脚本做清洗和分块import os import re from bs4 import BeautifulSoup def clean_text(text): text BeautifulSoup(text, html.parser).get_text() text re.sub(r\s, , text) return text.strip() def chunk_text(text, chunk_size512, overlap64): words text.split() chunks [] for i in range(0, len(words), chunk_size - overlap): chunk .join(words[i:i chunk_size]) if chunk: chunks.append(chunk) return chunks def load_documents(dir_path): docs [] for fname in os.listdir(dir_path): if fname.endswith(.txt): with open(os.path.join(dir_path, fname), r, encodingutf-8) as f: text clean_text(f.read()) for idx, chunk in enumerate(chunk_text(text)): docs.append({ text: chunk, source: fname, chunk_id: idx }) return docs然后向量化并写入Qdrantfrom qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from sentence_transformers import SentenceTransformer client QdrantClient(path./qdrant_data) model SentenceTransformer(BAAI/bge-large-zh-v1.5) client.recreate_collection( collection_namedocs, vectors_configVectorParams(size1024, distanceDistance.COSINE) ) docs load_documents(data/raw) texts [d[text] for d in docs] embeddings model.encode(texts, batch_size64, normalize_embeddingsTrue) points [ PointStruct(idi, vectorembeddings[i].tolist(), payloaddocs[i]) for i in range(len(docs)) ] client.upsert(collection_namedocs, pointspoints)这里recreate_collection会删掉已有集合生产环境慎用。size1024对应BGE-large的维度如果你换模型这个值要改。4.3 检索与生成串联检索的时候查询也要用同一个模型向量化def search(query, top_k5): query_vec model.encode([query], normalize_embeddingsTrue)[0] results client.search( collection_namedocs, query_vectorquery_vec.tolist(), limittop_k ) return [r.payload[text] for r in results]生成的时候把检索到的文档拼进promptdef build_prompt(query, docs): context \n\n.join(docs) return f基于以下资料回答问题。如果资料中没有相关信息就说不知道。 资料 {context} 问题{query} 回答然后调LLM Client生成回答。整个流程跑通之后你可以用FastAPI包一层from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): query: str app.post(/chat) def chat(req: QueryRequest): docs search(req.query) prompt build_prompt(req.query, docs) resp llm_client.chat([{role: user, content: prompt}]) return {answer: resp[choices][0][message][content]}启动命令uvicorn api.main:app --host 0.0.0.0 --port 8000。4.4 评估与迭代系统跑起来只是第一步关键是评估。我一般从三个维度看检索召回率、生成准确率、响应延迟。检索召回率可以用标注好的问答对来测看正确文档有没有出现在top_k里。生成准确率可以人工抽检或者用另一个LLM做裁判。响应延迟用日志记录P99控制在2秒以内算合格。如果召回率低先调ef_search再考虑换更大的embedding模型。如果生成质量差先看prompt有没有问题再看检索到的文档是不是相关。如果延迟高先看是哪一步慢检索慢就调索引参数生成慢就换小模型或者加缓存。注意评估集一定要和训练集分开不然你调出来的参数只是过拟合了评估集上线就露馅。5. 常见问题与排查技巧实录5.1 检索结果不相关怎么办这是最常见的问题。排查思路是先看查询向量和文档向量的相似度分数如果分数普遍很低说明模型不适合你的领域考虑微调或者换模型。如果分数高但结果不相关说明分块有问题可能一个块里混了多个主题。这时候要调整分块策略按语义切而不是按固定长度切。还有个容易被忽略的点查询改写。用户的问题往往很短比如“退货”直接检索可能效果不好。可以用LLM把查询扩展成“退货政策是什么怎么申请退货”再检索。实测下来查询改写能把召回率提升10%到15%。5.2 大模型输出格式不稳定如果你要求模型输出JSON它有时候会多输出一段解释文字。解决办法是在prompt里明确约束并且给示例。如果还是不稳定可以用response_format参数OpenAI支持或者在生成后做一次解析解析失败就重试。我一般会在prompt最后加一句“只输出JSON不要输出任何其他内容。”然后给一个示例。这样大部分情况下都能稳定。如果对格式要求极高可以用function calling或者tool use让模型按schema输出。5.3 延迟太高怎么优化延迟优化是个系统工程。首先看检索如果ef_search设得太大调小一点。其次看生成如果用的是大模型换小模型或者加缓存。缓存策略我一般用两级内存缓存LRU存高频查询Redis存全量查询。相同查询直接返回缓存结果延迟从秒级降到毫秒级。还有个技巧是流式输出。用户不需要等完整回答生成完可以边生成边显示。这样首字延迟能控制在500毫秒以内体验好很多。FastAPI支持StreamingResponse配合LLM的流式接口就能实现。5.4 常见问题速查表问题现象可能原因排查方法解决方案检索结果不相关模型不匹配/分块不合理看相似度分数换模型/调整分块生成内容胡编检索文档不相关检查检索结果加查询改写/调检索参数响应延迟高检索慢/生成慢分步计时调索引参数/换小模型/加缓存输出格式错乱prompt约束不够看原始输出加格式约束/用function callingAPI调用失败限流/网络问题看错误码加重试/限流/降级内存占用高向量维度大/索引参数大看监控降维度/调小m和ef_construct5.5 几个我踩过的坑第一个坑是向量维度不一致。有次我建库用的是768维模型查询用的是1024维模型结果检索出来的东西完全随机。排查了半天才发现是维度问题。所以换模型的时候一定要重新建库。第二个坑是Qdrant的payload索引。如果你经常按某个字段过滤一定要给那个字段建索引不然查询会全表扫描慢得离谱。建索引的命令是client.create_payload_index(collection_namedocs, field_namesource, field_schemakeyword)。第三个坑是LLM的token限制。检索回来的文档如果太长加上prompt可能超过模型的上下文窗口。我一般会限制检索文档的总token数比如不超过2000token超了就截断或者只取top_3。6. 后续扩展方向与个人体会这套框架搭好之后扩展性其实很强。你可以加多路召回比如向量检索加BM25加知识图谱然后做融合排序。也可以加Agent能力让模型自己决定调哪个工具。还可以加反馈闭环把用户的点赞点踩收集起来定期微调模型。我个人在实际操作中的体会是AI工程最难的不是模型而是工程。模型效果差一点用户可能感知不明显但工程不稳定用户分分钟流失。所以我的建议是先把工程底座搭稳再考虑模型优化。底座稳了换模型就是改个配置的事。最后分享一个小技巧日志一定要打全。每次请求的query、检索到的文档ID、生成的回答、耗时全部记下来。出问题的时候这些日志就是你的救命稻草。我见过太多人出了问题只能靠猜就是因为日志没打全。