1. 从零搭建AI工程体系为什么我劝你别一上来就啃框架这两年AI工程这个词被聊得太多多到有点泛滥。打开任何一个技术社区满屏都是“大模型应用开发”“RAG实战”“Agent落地”好像不聊两句向量数据库和提示词工程都不好意思说自己是搞技术的。但我观察到一个很有意思的现象真正能把一个AI应用从demo推到生产环境、并且稳定跑上三个月的人少之又少。大部分人卡在的不是模型效果而是工程化这一环——数据怎么流转、服务怎么拆分、成本怎么控制、效果怎么评估、线上出问题怎么排查。这些事框架文档里不会写培训班也不教。ai-engineering-from-scratch这个标题我第一次看到就觉得它戳中了痛点。它讲的不是“怎么调某个库的API”而是从零开始把AI工程当成一门正经的工程学科来对待。说白了就是教你从最底层的数据处理、模型封装、服务编排、监控评估这些环节一层一层把AI系统搭起来而不是站在别人的脚手架上刷漆。这篇文章我想聊的就是这套从零构建的思路到底包含哪些东西每一步为什么这么设计以及我在实际落地中踩过的那些坑。适合谁看如果你已经会写Python、懂一点机器学习基础但一到“把模型变成能用的服务”就发怵那这篇就是写给你的。如果你是想转行做AI工程的后端或数据开发同样能从中找到可复用的路径。我先把核心观点摆出来AI工程和传统软件工程最大的区别不在于用了多复杂的模型而在于不确定性。传统后端接口输入确定、输出确定、逻辑确定AI系统里模型输出是概率性的数据分布会漂移效果会随时间和场景变化。所以从零搭建AI工程本质上是在搭建一套管理不确定性的工程体系。理解了这一点后面所有的设计取舍就都说得通了。2. 整体架构设计先想清楚数据怎么流再想模型怎么放2.1 为什么我坚持“数据管道先行”而不是“模型先行”很多人做AI项目第一反应是先把模型跑通找个开源权重加载进来写个推理脚本看到输出就兴奋了。然后呢然后发现真实数据脏得没法看格式五花八门字段缺失、编码混乱、长度参差。这时候再回头补数据清洗整个代码结构已经烂了只能推倒重来。我吃过这个亏所以现在做任何AI项目第一步永远是把数据管道画出来。数据管道要回答几个问题数据从哪来数据库、文件、消息队列、第三方接口经过哪些处理清洗、去重、分块、向量化、标注存到哪对象存储、向量库、关系库怎么被消费训练、推理、评估这条链路里每个环节的输入输出格式必须提前定义清楚最好用schema约束住。我习惯用Pydantic或者dataclass把每个阶段的数据结构写死这样上下游对接的时候不会出现“我以为你传的是list结果你传的是dict”这种低级错误。提示数据管道的设计文档哪怕只有一页纸也要在写代码前画出来。我见过太多项目因为跳过这一步后期改一个字段要动五个模块。2.2 分层架构把“会变的”和“不变的”隔离开AI系统里什么最容易变模型会换、提示词会调、检索策略会改、业务规则会更新。什么相对稳定数据存储、服务框架、监控体系。所以架构分层的核心原则就是把易变的部分收敛到独立的层用清晰的接口和稳定的层通信。我通常分成四层。最底层是数据层负责原始数据的存储和基础处理这层尽量用成熟组件别自己造轮子。往上是能力层包括模型推理、向量检索、工具调用这些原子能力每个能力封装成独立模块对外暴露统一接口。再往上是编排层负责把多个能力组合成业务流程比如“先检索再生成再校验”这种链路这层是变化最频繁的所以要用配置驱动而不是硬编码。最上面是接入层处理请求路由、鉴权、限流、日志这层和传统后端没区别。这么分的好处是换模型只动能力层改流程只动编排层互不影响。我见过把模型调用直接写在业务逻辑里的代码换个模型要改几十处那维护成本简直灾难。2.3 技术选型的取舍逻辑别追新追稳AI领域新技术层出不穷今天流行这个向量库明天流行那个编排框架。我的建议是核心链路用最稳的边缘实验用最新的。什么叫核心链路就是数据存储、服务框架、监控这些一旦出问题整个系统就挂掉的部分。这些地方我倾向于用经过大规模验证的组件比如PostgreSQL存元数据、Redis做缓存、FastAPI做服务、Prometheus做监控。这些不一定最时髦但文档全、社区大、出问题能搜到答案。边缘实验部分比如新的检索算法、新的提示词技巧可以大胆试但一定要隔离在独立模块里试错了直接下线不影响主流程。我一般会用一个feature flag机制来控制这些实验的开关线上出问题一键回滚。3. 核心模块拆解每个环节到底在解决什么问题3.1 数据预处理脏数据才是常态干净数据是意外AI工程里数据预处理的工作量能占到整个项目的百分之六十以上但很多人低估了它。我接手过一个文本分类项目原始数据是从多个渠道爬来的有的带HTML标签有的编码是GBK有的字段里混了换行符和制表符还有大量重复和近似重复。如果直接喂给模型效果差得离谱你还以为是模型不行。预处理的核心步骤我总结成一套固定流程解码归一化、结构解析、内容清洗、去重、分块、格式化。解码归一化解决编码问题统一转UTF-8。结构解析把非结构化文本里的结构信息抽出来比如从HTML里提取正文。内容清洗去掉无关字符、统一标点、处理特殊符号。去重分精确去重和近似去重精确去重用哈希近似去重用MinHash或者SimHash。分块是针对长文本的要按语义边界切不能硬按字数切否则会把一句话切断。最后格式化成下游需要的结构。这里有个细节很多人忽略分块策略直接影响检索效果。我试过固定长度分块、按段落分块、按语义相似度分块实测下来对于技术文档类内容按标题层级加段落的分块方式效果最好因为能保留上下文结构。分块大小也不是越小越好太小会丢失上下文太大检索精度下降一般中文内容控制在三百到五百字比较合适具体要看业务。3.2 模型封装让模型变成可替换的零件模型封装的目标是解耦。业务代码不应该知道用的是哪个模型、什么版本、什么参数。我通常定义一个统一的推理接口输入是标准化的请求对象输出是标准化的响应对象。具体实现可以是本地模型、远程API、或者多个模型的组合业务层不关心。封装的时候有几个关键点。第一是超时和重试模型推理可能很慢必须设超时超时后要有降级策略比如返回缓存结果或者默认值。第二是批处理单条推理效率低要支持批量但批量大小要可配置因为不同模型对batch size的敏感度不同。第三是资源隔离模型推理吃GPU或内存要和主服务隔离开避免把服务拖垮。我一般用独立的推理服务通过HTTP或gRPC通信虽然多了一次网络开销但稳定性和可扩展性好太多。注意模型版本管理一定要做。每次模型更新都要记录版本号、更新日期、效果指标线上出问题能快速定位是哪个版本引入的。我见过因为没做版本管理模型效果下降后查了一周才找到原因的案例。3.3 检索增强不是所有问题都需要微调检索增强生成这套东西现在很火但很多人用错了地方。我的经验是知识频繁更新、事实性要求高、领域知识密集的场景用检索增强风格迁移、格式转换、固定模式的任务用微调或者提示词。检索增强的优势是知识可更新、可溯源、成本低劣势是延迟高、依赖检索质量。检索增强的核心是检索质量。检索质量取决于三件事分块质量、向量模型、检索策略。分块前面说了向量模型要选和业务语言匹配的中文场景用中文优化的模型别直接用英文模型。检索策略上纯向量检索对语义相似但字面不重叠的内容效果好但对精确匹配的关键词效果差所以我一般用混合检索向量检索加关键词检索两路结果融合排序。融合算法用RRF倒数排名融合比较简单有效不需要调参。重排序是提升检索精度的关键一步。粗排召回一批候选再用交叉编码器精排精度能提升不少。但重排序模型推理慢所以候选数量要控制一般召回二十到五十条重排后取前五到十条。这个数量要实测调太少漏召回太多影响延迟。3.4 评估体系没有评估就没有优化AI系统最怕的就是“感觉效果还行”。感觉是靠不住的必须有量化评估。评估体系分两层离线评估和在线评估。离线评估用标注数据集算准确率、召回率、F1这些指标。在线评估用真实流量看点击率、转化率、用户反馈这些业务指标。离线评估的关键是测试集的质量。测试集要覆盖真实场景的分布不能只挑简单的样本。我一般会从真实数据里分层采样保证各类别都有足够样本。还要留一部分“困难样本”专门测模型的边界能力。评估指标也要选对分类任务看F1生成任务看BLEU、ROUGE这些还不够最好加上人工评估或者用更强的模型做裁判。在线评估要做A/B测试。新版本先小流量跑对比核心指标显著优于旧版本再全量。A/B测试要注意样本量和显著性别跑了一天就下结论至少跑一周覆盖完整周期。我踩过的坑是有一次新模型离线指标很好上线后业务指标反而降了后来发现是离线测试集和真实分布有偏差。所以离线在线必须结合看。4. 实操落地从零搭一个可用的AI服务4.1 环境准备与依赖管理动手之前先把环境理清楚。我强烈建议用容器化Docker加docker-compose起步别在裸机上装一堆依赖。Python版本选3.10或3.11太新太旧都容易踩坑。依赖管理用poetry或者pip-tools把直接依赖和间接依赖都锁死避免“在我机器上能跑”的问题。目录结构我习惯这么组织data/放数据处理脚本和配置models/放模型封装和推理服务pipelines/放编排逻辑services/放对外接口evaluation/放评估脚本configs/放各种配置文件。每个目录下再按功能分子目录。这样结构清晰新人接手能快速定位。# 典型的项目初始化 mkdir ai-engineering-project cd ai-engineering-project python -m venv .venv source .venv/bin/activate pip install poetry poetry init poetry add fastapi uvicorn pydantic numpy pandas4.2 数据管道的代码实现数据管道我用阶段式设计每个阶段是一个独立的处理函数输入输出都是标准化的数据结构。这样每个阶段可以单独测试、单独替换。下面是一个简化的文本预处理管道示例。from dataclasses import dataclass from typing import List import re import hashlib dataclass class Document: doc_id: str content: str metadata: dict def normalize_encoding(text: str) - str: # 统一编码和换行符 text text.encode(utf-8, errorsignore).decode(utf-8) text text.replace(\r\n, \n).replace(\r, \n) return text def clean_content(text: str) - str: # 去掉HTML标签和多余空白 text re.sub(r[^], , text) text re.sub(r\s, , text) return text.strip() def deduplicate(docs: List[Document]) - List[Document]: seen set() result [] for doc in docs: fingerprint hashlib.md5(doc.content.encode()).hexdigest() if fingerprint not in seen: seen.add(fingerprint) result.append(doc) return result def chunk_document(doc: Document, chunk_size: int 400, overlap: int 50) - List[Document]: # 按语义边界分块这里简化为按句子聚合 sentences re.split(r(?[。.!?]), doc.content) chunks [] current for sent in sentences: if len(current) len(sent) chunk_size and current: chunks.append(Document( doc_idf{doc.doc_id}_chunk_{len(chunks)}, contentcurrent.strip(), metadata{**doc.metadata, chunk_index: len(chunks)} )) current current[-overlap:] if overlap else current sent if current.strip(): chunks.append(Document( doc_idf{doc.doc_id}_chunk_{len(chunks)}, contentcurrent.strip(), metadata{**doc.metadata, chunk_index: len(chunks)} )) return chunks这段代码里overlap参数很关键它保证相邻块之间有重叠内容避免关键信息正好被切断。实测下来重叠五十字左右对中文文本比较合适。4.3 推理服务的封装与部署推理服务我用FastAPI封装核心是异步处理和批处理。模型推理是IO密集或GPU密集的同步处理会阻塞所以用async。批处理通过一个队列实现请求进来先入队攒够一批或者超时了再一起推理。from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() class InferenceRequest(BaseModel): text: str max_length: int 512 class InferenceResponse(BaseModel): result: str latency_ms: float class BatchProcessor: def __init__(self, batch_size8, timeout0.1): self.batch_size batch_size self.timeout timeout self.queue asyncio.Queue() self.results {} async def process(self): while True: batch [] try: while len(batch) self.batch_size: item await asyncio.wait_for(self.queue.get(), timeoutself.timeout) batch.append(item) except asyncio.TimeoutError: pass if batch: await self._infer_batch(batch) async def _infer_batch(self, batch): # 实际推理逻辑这里用占位 texts [item[1] for item in batch] # outputs model.predict(texts) outputs [fprocessed: {t} for t in texts] for (future, _), output in zip(batch, outputs): future.set_result(output) processor BatchProcessor() app.on_event(startup) async def startup(): asyncio.create_task(processor.process()) app.post(/infer, response_modelInferenceResponse) async def infer(req: InferenceRequest): import time start time.time() future asyncio.get_event_loop().create_future() await processor.queue.put((future, req.text)) result await future return InferenceResponse( resultresult, latency_ms(time.time() - start) * 1000 )这个批处理模式能把吞吐量提升好几倍尤其是GPU推理场景。但要注意batch size不是越大越好太大显存扛不住太小浪费算力。我一般从8开始试逐步加到16、32看延迟和吞吐的平衡点。4.4 监控与日志线上出问题能快速定位监控我分三个维度系统指标、业务指标、模型指标。系统指标看CPU、内存、GPU利用率、请求延迟、错误率用Prometheus加Grafana。业务指标看请求量、成功率、用户反馈从日志里聚合。模型指标看输出分布、置信度分布、异常输出比例这个最容易被忽略但最重要。日志要结构化每条日志带request_id、用户标识、模型版本、输入输出摘要、耗时。这样出问题能按request_id串起整条链路。我习惯用structlog或者loguru输出JSON格式方便ELK或者Loki采集。提示模型输出一定要采样记录尤其是低置信度的输出。这些样本是后续优化的金矿能帮你发现模型的系统性偏差。5. 踩坑实录那些文档里不会写的问题5.1 常见问题速查表问题现象可能原因排查方向解决方案检索结果不相关分块过大或过小检查分块大小和边界调整分块策略增加重叠模型输出不稳定温度参数过高检查生成参数降低temperature固定随机种子服务延迟高批处理等待超时查看队列积压调整batch size和timeout内存持续增长缓存未清理检查缓存和对象引用加LRU淘汰定期重启效果逐渐下降数据分布漂移对比新旧数据分布定期重训加漂移检测并发上不去同步阻塞检查IO和锁改异步加连接池5.2 三个我踩过的深坑第一个坑是向量库的维度陷阱。我一开始用某个向量模型维度是768后来换了个模型维度变成1024结果向量库里的旧数据全废了因为维度不匹配没法检索。教训是向量库的schema设计要预留维度变更的空间最好把向量模型版本和维度一起存进元数据换模型时能识别出哪些数据需要重新向量化。第二个坑是提示词的版本管理。提示词改了之后效果变差但没记录改之前是什么只能凭记忆恢复。后来我强制要求所有提示词都存文件用git管理每次改动写清楚原因和预期效果。提示词也是代码必须版本化。第三个坑是评估集的污染。有一次模型效果突然变好查了半天发现是评估集里的样本不小心混进了训练数据。这种数据泄漏很难发现但影响致命。现在我都会在训练前做一次训练集和评估集的交叉检查用哈希或者相似度比对确保没有重叠。5.3 性能优化的几个实用技巧推理性能优化我总结了几条实测有效的量化能把模型体积和推理时间降一半精度损失通常可接受缓存对重复查询效果显著尤其是高频问题预计算把能提前算的向量、特征都算好别在请求时现算降级策略在高峰期返回简化结果保证服务不挂。成本控制也是AI工程的重要一环。API调用按token计费长文本成本高所以要控制输入长度能摘要的先摘要。自建推理看GPU利用率利用率低就合并服务利用率高就加机器。我一般会算一个单位请求成本定期review发现异常及时优化。6. 从能跑到好用中间隔着多少工程细节把AI系统跑起来不难难的是让它稳定、可控、可优化。我现在的习惯是任何AI项目上线前都要过一遍检查清单数据管道有没有监控模型版本有没有记录评估集有没有更新降级策略有没有测试日志能不能定位问题成本有没有算过这些问题看起来琐碎但每一个都可能在关键时刻救你一命。ai-engineering-from-scratch这个思路的价值不在于教你某个具体工具怎么用而在于帮你建立一套完整的工程思维。工具会过时框架会迭代但数据怎么流、服务怎么拆、效果怎么评、问题怎么查这些底层能力是通用的。我见过太多人追着新框架跑结果连最基本的日志和监控都没做好线上出问题两眼一抹黑。与其这样不如沉下心来从数据管道开始一层一层把地基打牢。最后分享一个我自己的习惯每做一个新项目我都会写一份“运维手册”记录这个系统的架构图、关键配置、常见问题、应急流程。这份手册不追求好看只追求实用新人接手能照着操作自己隔几个月回来看也能快速回忆起来。这个习惯帮我省了无数时间也让我对每个系统的理解更深了一层。AI工程这条路很长但每一步扎实的积累最后都会变成你的底气。