从零搭建AI工程能力：后端老兵的踩坑与重构实录

1. 从零搭建AI工程能力一个后端老兵的踩坑与重构实录ai-engineering-from-scratch这个标题第一次看到的时候我以为是又一个教你调包的教程。点进去翻了翻发现它想做的事情比调包大得多——它试图回答一个很具体的问题一个没有AI背景的工程师怎么从最底层开始把AI工程这套东西真正搭起来、跑通、并且能持续迭代。我自己是后端出身做了七八年业务系统两年前因为项目需要被拉去做AI相关的工程落地。刚开始那段时间非常痛苦因为网上大部分资料要么是论文式的理论推导要么是三行代码调用API的玩具示例中间那层——怎么把模型能力变成稳定可靠的工程系统——几乎没人系统讲。所以看到这个标题的时候我是有共鸣的。这篇东西我想按自己的理解把从零构建AI工程能力这件事拆开讲。不是复述某个教程而是把我自己走过一遍、踩过坑、后来重构过的路径整理出来。适合谁看有编程基础、想往AI工程方向转的开发者已经在做AI应用但感觉系统很脆、想补工程底子的同学以及带团队做AI项目、需要一套可落地方法论的技术负责人。我会尽量说人话把每个选择背后的为什么讲清楚而不是只给结论。2. 整体设计思路为什么不能一上来就写模型代码2.1 先搞清楚AI工程到底在工程什么很多人对AI工程的想象是训练模型、调参、部署。但真正做过项目的人知道模型只是其中一小块。一个能上线的AI系统大致包含这几层数据层数据的采集、清洗、标注、版本管理模型层选型、微调、评估、迭代服务层推理服务的封装、并发、限流、降级应用层Prompt编排、上下文管理、结果后处理运维层监控、日志、成本控制、效果回归from scratch的价值就在于它逼你把这五层都过一遍而不是只盯着模型层。我见过太多团队模型效果调得不错但一上线就崩——因为服务层没做限流应用层没做兜底运维层没有效果监控。这些坑只有从零搭一遍才会真正理解。2.2 为什么建议自底向上而不是自顶向下市面上主流的入门路径是自顶向下先跑通一个Demo再逐步深入。这个路径上手快但有个致命问题——你会对底层形成错误的直觉。比如你调一个API觉得很简单就会以为部署模型也很简单你用现成的向量库觉得检索很轻松就会低估数据清洗的工作量。我自己的经验是如果目标是建立真正的工程能力自底向上更扎实。具体来说顺序应该是先理解一个最小可用的推理流程哪怕是用最笨的方式跑通再理解数据怎么进来、怎么出去然后才是服务化、并发、监控最后才是模型微调、效果优化这个顺序的好处是每一步你都知道自己在解决什么问题而不是被框架牵着走。坏处是前期慢容易劝退。所以我的建议是用自顶向下的方式建立兴趣用自底向上的方式建立能力两者结合。2.3 技术选型的核心原则可替换性优先从零搭AI工程最容易犯的错是过早绑定某个框架。今天用LangChain明天想换LlamaIndex发现代码全耦合在一起迁移成本极高。我的原则是每一层都留好接口让上层不依赖下层的具体实现。比如层级抽象接口可选实现模型调用LLMClient.generate()本地模型、云端API、不同厂商向量检索VectorStore.search()FAISS、Milvus、pgvector数据存储DataStore.read/write()文件、数据库、对象存储评估Evaluator.evaluate()规则、模型打分、人工这样做的直接好处是当你想换模型、换向量库的时候只需要改一个适配器而不是重写整个系统。这个原则听起来简单但真正落地需要克制——克制住直接用框架高级API的冲动。3. 核心细节解析从零搭建的四个关键环节3.1 环境与依赖别小看这一步从零开始第一件事是环境。这里有个反直觉的点AI工程的环境比普通后端复杂得多因为涉及GPU、CUDA、各种Python包版本冲突。我的建议是分三层管理系统层用容器隔离别在宿主机上直接装CUDA驱动Python层用conda或uv管理虚拟环境锁定版本项目层用requirements.txt或pyproject.toml明确依赖一个具体的坑torch和transformers的版本兼容性。我遇到过升级transformers后torch报错的情况排查了半天。后来养成习惯每次升级核心包之前先在一个独立环境里验证。提示如果你的机器没有GPU不要硬上。很多AI工程能力数据处理、服务封装、评估流程在CPU上完全可以练模型推理用云端API代替即可。等流程跑通了再考虑本地GPU。3.2 数据管道最脏最累但最重要数据管道是AI工程里最不性感、但决定成败的部分。我统计过自己做过的项目花在数据上的时间大概占60%模型相关只占20%剩下20%是服务和运维。一个最小可用的数据管道应该包含采集从各种来源拿到原始数据清洗去重、去噪、格式统一切分把长文本切成合适粒度的块向量化把文本转成向量存储存进向量库和元数据库这里重点说切分因为它最容易被忽视。切分粒度直接决定检索效果。切太大检索到的内容包含太多无关信息切太小语义不完整。我的经验参数中文文本300-500字一块重叠50-100字英文文本200-400词一块重叠50词代码按函数或类切分不要按行重叠的作用是防止关键信息正好被切在边界上。这个参数没有标准答案必须根据你的数据特点实测调整。3.3 模型调用层封装比调用重要调用模型本身很简单难的是封装成一个稳定、可观测、可降级的服务。我见过的最常见问题代码里到处散落着模型调用没有统一入口。结果想换模型、想加日志、想加重试都要改几十个地方。正确的做法是做一个LLMClient抽象class LLMClient: def __init__(self, provider, model, **kwargs): self.provider provider self.model model self.config kwargs def generate(self, prompt, **options): # 统一处理重试、超时、日志、计费 for attempt in range(self.config.get(max_retries, 3)): try: response self._call(prompt, **options) self._log(prompt, response) return response except Exception as e: if attempt self.config[max_retries] - 1: raise time.sleep(2 ** attempt) # 指数退避这个封装里重试和日志是必须的。模型调用失败是常态没有重试机制的系统非常脆。日志则是后续排查问题和优化效果的基础。3.4 评估体系没有评估就没有迭代这是最容易被跳过、但最重要的一环。很多人做完一个AI功能靠感觉判断好不好。这是不可持续的。一个最小评估体系包含测试集一批有标准答案的输入输出对评估指标准确率、召回率、相关性打分等回归测试每次改动后跑一遍确保没退化对于生成类任务评估比较难。我的做法是分层评估规则层格式是否正确、是否包含关键信息模型层用另一个模型给结果打分人工层抽样人工检查三层结合既保证效率又保证质量。这里的关键是把评估流程自动化否则没人会坚持跑。4. 实操过程一个最小AI问答系统的完整搭建4.1 需求定义与边界划定假设我们要做一个基于文档的问答系统。先明确边界输入用户问题 一批文档输出基于文档内容的回答约束回答必须来自文档不能编造这个约束很重要它决定了整个架构。如果允许编造那直接调模型就行要求基于文档就必须做检索增强。4.2 数据准备与向量化实操第一步把文档处理成可检索的形式。import hashlib from typing import List def chunk_text(text: str, chunk_size: int 400, overlap: int 80) - List[str]: 按字符切分文本带重叠 chunks [] start 0 while start len(text): end start chunk_size chunk text[start:end] if chunk.strip(): chunks.append(chunk) start end - overlap return chunks def process_documents(docs: List[dict]) - List[dict]: 处理文档返回带元数据的块 processed [] for doc in docs: chunks chunk_text(doc[content]) for i, chunk in enumerate(chunks): processed.append({ id: hashlib.md5(f{doc[id]}_{i}.encode()).hexdigest(), doc_id: doc[id], chunk_index: i, content: chunk, metadata: doc.get(metadata, {}) }) return processed这里用hashlib生成稳定ID方便后续去重和更新。元数据保留原始文档信息检索时可以过滤。向量化部分如果不想依赖外部服务可以用本地的sentence-transformersfrom sentence_transformers import SentenceTransformer model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def embed_chunks(chunks: List[dict]) - List[dict]: texts [c[content] for c in chunks] embeddings model.encode(texts, batch_size32, show_progress_barTrue) for chunk, emb in zip(chunks, embeddings): chunk[embedding] emb.tolist() return chunksbatch_size设32是个经验值太大容易爆内存太小速度慢。show_progress_bar在处理大量数据时很有用能看到进度。4.3 检索与生成的核心逻辑检索部分先用最简单的余弦相似度import numpy as np def cosine_similarity(a, b): return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) def retrieve(query: str, chunks: List[dict], top_k: int 5): query_emb model.encode([query])[0] scores [] for chunk in chunks: score cosine_similarity(query_emb, chunk[embedding]) scores.append((score, chunk)) scores.sort(keylambda x: x[0], reverseTrue) return [chunk for _, chunk in scores[:top_k]]生成部分把检索到的内容拼进Promptdef build_prompt(query: str, contexts: List[dict]) - str: context_text \n\n.join([ f[文档{c[doc_id]} 片段{c[chunk_index]}]\n{c[content]} for c in contexts ]) return f基于以下文档内容回答问题。如果文档中没有相关信息请明确说明文档中未找到相关信息不要编造。 文档内容 {context_text} 问题{query} 回答这个Prompt设计有两个关键点一是明确要求基于文档二是要求无法回答时明确说明。后者能大幅降低幻觉。4.4 服务化与接口封装把上面的流程封装成一个服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): question: str top_k: int 5 class QueryResponse(BaseModel): answer: str sources: List[dict] app.post(/query, response_modelQueryResponse) async def query(req: QueryRequest): if not req.question.strip(): raise HTTPException(status_code400, detail问题不能为空) contexts retrieve(req.question, chunks, req.top_k) prompt build_prompt(req.question, contexts) answer llm_client.generate(prompt) return QueryResponse( answeranswer, sources[{doc_id: c[doc_id], chunk_index: c[chunk_index]} for c in contexts] )用FastAPI是因为它自带文档、类型校验和异步支持。sources字段返回引用来源方便用户核查也方便排查问题。4.5 效果验证与迭代搭完之后准备一批测试问题跑一遍看效果。我通常会记录问题检索到的内容生成的回答是否准确问题类型Q1相关正确是事实查询Q2部分相关部分正确否多跳推理Q3不相关编造否超出范围根据这个表能快速定位问题是检索不准还是生成有问题还是数据本身缺失。这个表是迭代的核心工具比任何指标都直观。5. 常见问题与排查技巧实录5.1 检索不准的三种典型情况情况一检索到的内容完全不相关。通常是向量模型不适合你的领域。比如用通用模型处理法律、医疗文本效果会差。解决办法是换领域模型或者用领域数据微调向量模型。情况二相关内容排不到前面。可能是切分粒度不对或者相似度计算方式不适合。可以试试混合检索——向量检索加关键词检索两者结果融合。情况三多个文档内容冲突。这是数据问题不是检索问题。需要在数据层做去重和冲突检测或者在Prompt里要求模型指出冲突。5.2 生成质量不稳定的排查思路生成质量波动大通常从这几个方向查Prompt是否稳定有没有随机性元素比如时间戳、随机ID上下文长度是不是超过了模型的有效窗口导致信息丢失温度参数temperature设太高会导致输出发散问答类任务建议0.1-0.3检索质量检索到的内容质量直接决定生成质量先确保检索没问题我遇到过一个案例生成结果时好时坏最后发现是检索返回的块顺序不稳定导致Prompt里上下文顺序变化影响了模型判断。解决办法是对检索结果按相关度稳定排序。5.3 性能与成本的平衡AI系统的成本主要在模型调用。几个实用的优化手段缓存相同问题直接返回缓存结果命中率高的场景能省一大半批处理多个请求合并成一次调用减少往返开销小模型优先简单问题用小模型复杂问题才用大模型检索优化减少检索返回的块数降低输入长度这里有个权衡块数减少会降低召回可能影响效果。我的做法是先保证效果再逐步优化成本而不是一上来就抠成本。5.4 常见问题速查表现象可能原因排查方向回答编造检索无结果但未提示检查Prompt兜底逻辑回答不完整上下文被截断检查模型窗口和输入长度响应慢模型调用或检索慢分别计时定位瓶颈结果不稳定温度高或检索顺序不稳固定参数稳定排序成本高调用频繁或输入过长加缓存优化检索6. 从能跑到好用工程化的几个进阶方向6.1 可观测性建设系统跑起来之后最重要的是看得见。我通常会加这几类监控调用量每天/每小时请求数看趋势延迟分布P50、P95、P99看长尾错误率失败请求占比按错误类型分类成本token消耗按天统计效果指标抽样评估的准确率这些数据不需要很复杂一个简单的日志加定时统计就够了。关键是坚持看很多问题在爆发前都有征兆。6.2 持续迭代的机制AI系统和传统系统最大的区别是它的效果会随数据和用户行为变化而漂移。所以必须有持续迭代机制收集badcase用户反馈、低分结果定期整理归因分析是数据问题、检索问题还是生成问题针对性优化补数据、调检索、改Prompt回归验证确保优化没引入新问题这个循环跑起来系统才会越用越好。我见过很多团队做完就放着结果几个月后效果明显下降就是因为没有迭代机制。6.3 团队协作与知识沉淀如果是团队做AI工程还有一件事很重要把经验沉淀成文档和工具。比如Prompt模板库常用场景的Prompt整理成模板评估数据集积累测试用例新人可以直接用踩坑记录每个坑怎么踩的、怎么解决的这些看起来是软的东西但决定了团队能不能规模化。一个人踩过的坑全团队不用再踩这就是效率。7. 我个人的一些体会从零搭AI工程这件事最大的收获不是学会了某个框架或工具而是建立了一套判断力——知道什么问题该用什么方案知道哪些坑是必然要踩的知道系统的瓶颈通常在哪里。如果让我给刚开始的人一个建议我会说别追求一步到位先跑通最小闭环。一个能跑通的简单系统比一个设计完美但跑不起来的系统有价值得多。跑通之后再一层层优化每一步都有反馈这样学得最快。另外别被AI这个词吓到。剥开外壳它还是软件工程——数据、接口、服务、监控、迭代这些基本功一样都不能少。AI工程的特殊性在于它多了一层不确定性而工程的价值恰恰在于用确定的手段去管理不确定性。想明白这一点很多设计选择就顺理成章了。最后分享一个我常用的检查清单每次做完一个AI功能我会过一遍数据管道是否可重复运行模型调用是否有重试和降级是否有评估集和回归测试是否有监控和日志是否有成本控制手段是否有迭代机制这六条都打勾这个功能才算真正工程化了。缺哪条哪条就是下一个要补的坑。