☰
从零构建AI工程体系:数据管道、推理链与评估监控实战
2026/10/1 11:55:31 网站建设 项目流程

1. 从零搭建AI工程体系,为什么我劝你别一上来就啃论文

“ai-engineering-from-scratch”这个标题,我第一次看到的时候,心里咯噔了一下。不是因为它有多高深,而是因为它太诚实了——它承认了一件事:现在市面上绝大多数所谓的AI教程,其实都是从“调包”开始的。你打开任何一个平台,搜“AI入门”,出来的全是“三行代码调用大模型API”“五分钟搭建你的第一个RAG”。这些东西有没有用?有用,但它们跳过了最要命的部分:当模型效果不对的时候,你根本不知道从哪里下手排查。

我自己在这个坑里蹲了差不多两年。最开始做AI应用,就是典型的“API工程师”——把提示词拼一拼,调一调温度参数,效果不好就换模型,再不好就加几个示例。直到有一次,线上一个分类任务准确率突然掉了十几个点,我花了整整三天才定位到问题:不是模型的问题,也不是提示词的问题,而是数据预处理阶段一个分词器的截断策略在特定长度的输入上出了问题。那一刻我才意识到,不懂底层工程细节,你连bug都找不到。

所以这篇内容,我想认真聊聊“从零构建AI工程能力”这件事。它不是教你从零训练一个GPT,那是研究机构干的事;它讲的是作为一个应用侧的AI工程师,你需要具备哪些底层的、可迁移的工程能力,才能在模型之上构建出稳定、可维护、可迭代的系统。适合谁看?如果你已经会用API调模型,但总觉得心里没底,遇到问题只能靠“换模型”“改提示词”来碰运气,那这篇就是写给你的。如果你还在犹豫要不要入行AI应用开发,也可以看看,了解一下这个方向真正需要的能力栈是什么样的。

2. 整体思路拆解:为什么“从零”不等于“从底层造轮子”

2.1 先搞清楚“AI工程”到底在工程什么

很多人对AI工程有个误解,觉得就是要懂Transformer架构、会推导反向传播、能手写注意力机制。这些东西重要吗?重要,但它们是算法工程师和研究员的门槛,不是AI应用工程师的日常。我面过不少候选人,能把多头注意力的公式写得清清楚楚,但问他“如果线上推理延迟突然从200ms涨到2s,你会怎么排查”,就卡住了。

AI工程的核心,我自己的理解是三个词:数据流、推理链、反馈环。数据流是从原始输入到模型可消费格式的完整管道,包括清洗、分块、向量化、缓存;推理链是模型调用前后的所有处理逻辑,包括提示词组装、上下文管理、输出解析、后处理;反馈环是线上效果的监控、评估和迭代机制。这三件事,没有一件需要你从头训练模型,但每一件都需要你对工程细节有足够的掌控力。

所以“from scratch”的正确打开方式,不是从零实现一个Transformer,而是从零搭建一套你能完全理解和控制的AI应用工程体系。这个体系里,模型只是一个可替换的组件,真正值钱的是围绕它的那些“脏活累活”。

2.2 为什么我选择“先跑通再优化”的路线

市面上很多教程喜欢一上来就讲架构设计、讲微服务拆分、讲向量数据库选型。我试过跟着走,结果就是搭了一个看起来很专业的架子,但核心的AI能力根本没跑通,调了半天连一个能用的问答都出不来。这种“先设计后实现”的思路,在传统软件开发里可能没问题,但在AI应用里特别容易翻车,因为AI组件的行为不确定性太高了,你根本没法在纸面上把接口设计对。

我后来总结出来的路线是:先用最笨的办法跑通一个端到端的demo,然后逐个环节替换和优化。比如做一个文档问答,第一步就是用一个Python脚本,把文档读进来、切一切、拼到提示词里、调API、打印结果。这个版本可能只有50行代码,没有任何工程美感,但它能跑通,你能看到输入和输出的完整链路。然后你再逐步替换:把硬编码的提示词换成模板管理,把内存里的文档换成向量检索,把同步调用换成异步批处理。每一步替换你都知道为什么换、换了之后什么变了。

这个思路的好处是,你始终有一个可工作的基线,任何优化都是在这个基线上做增量,而不是在空气里盖楼。坏处是,前期看起来进展很慢,别人已经在那讲“企业级RAG架构”了,你还在写单文件脚本。但相信我,等到线上出问题的时候,你的排查速度会比那些“架构师”快十倍。

2.3 工具选型的核心原则:可替换性大于性能

在从零搭建的过程中,你会面临大量的工具选型:向量数据库用哪个、编排框架用LangChain还是自己写、推理用vLLM还是直接调API。我的建议是,在早期阶段,可替换性比性能重要得多。

举个例子,向量检索这块,Milvus、Qdrant、Chroma、FAISS我都用过。如果单纯看性能,Milvus在千万级向量上的表现确实好,但它的部署复杂度也高。在项目早期,你的数据量可能就几千条,用Chroma或者FAISS完全够用,而且换起来特别容易——接口就那几个,add、search、delete。等你真的到了千万级,再迁移到Milvus,迁移成本也就是写一个数据导出导入的脚本。

反过来,如果你一上来就上Milvus集群,光是运维就够你喝一壶的,而且你的代码会和Milvus的SDK深度绑定,后面想换个轻量方案都难。我见过太多项目,死在了“过度工程”上——花了两周搭基础设施,结果核心的AI效果根本没时间调。

所以我的原则是:任何组件,只要能用最笨的办法替代,就先别引入重型依赖。向量检索先用numpy算余弦相似度,编排先用函数调用,缓存先用字典。等这些笨办法真的成为瓶颈了,再换专业工具,那时候你才知道自己真正需要什么。

3. 核心细节解析:从零构建AI工程能力的四个关键环节

3.1 数据管道:AI应用里最容易被低估的“脏活”

数据管道是AI工程里最没有存在感、但出问题最多的地方。我统计过自己经手的项目,线上效果问题里,大概有六成最终定位到数据环节,而不是模型本身。数据管道具体包括什么?从原始文档的解析、清洗、分块,到向量化、索引、更新,每一步都有坑。

先说文档解析。你以为PDF读出来就是文本?太天真了。我遇到过扫描版PDF读出来全是乱码的,遇到过表格跨页导致内容错位的,遇到过页眉页脚混进正文的。这些问题的解决方案没有银弹,只能针对你的数据源做定制。我的做法是,先拿一批样本数据跑一遍解析,人工检查解析结果,把常见问题归类,然后写针对性的清洗规则。比如页眉页脚,通常有固定的位置特征或者重复模式,用正则或者位置信息就能过滤掉。

再说分块。分块策略直接决定了检索质量,但很多人就是简单按固定长度切。我试过按500字符切,结果把一句话切成两半,检索出来的片段语义不完整。后来改成按语义边界切,比如按段落、按标题层级,效果好很多。但语义分块也有问题,就是块的长度不均匀,有的块特别长,有的特别短。我的折中方案是:先按语义边界切,然后对超长的块做二次切分,对过短的块做合并。具体参数上,我一般把目标块长度设在300到500个token之间,重叠部分设50到100个token,这个范围在大多数检索场景下表现比较稳。

还有一个容易被忽略的点是元数据。每个块除了文本内容,还应该带上来源、位置、时间等元信息。这些元数据在检索时可以用于过滤,在展示时可以用于引用。我见过一些项目,检索出来的片段没法追溯到原文位置,用户想验证都验证不了,体验很差。

3.2 提示词工程:从“写作文”到“写代码”

提示词工程这个词被用烂了,很多人觉得就是“把话说清楚”。但在我眼里,提示词工程的核心不是写作能力,而是结构化思维能力。一个好的提示词,应该像一段代码一样,有明确的输入、输出、边界条件和异常处理。

我自己的提示词模板通常包含这几个部分:角色定义、任务描述、输入格式说明、输出格式说明、约束条件、示例。角色定义不是让你写“你是一个资深专家”这种废话,而是明确模型应该以什么视角和知识范围来回答。任务描述要具体到可执行,比如“从以下文本中提取所有日期和对应的事件”就比“分析这段文本”好得多。输出格式说明特别重要,如果你要程序化解析输出,最好用JSON格式,并在提示词里给出schema。

但提示词工程最大的坑在于:你以为你写清楚了,模型的理解和你的预期可能完全不一样。我踩过最惨的一次,是让模型做情感分类,输出“正面”或“负面”。结果模型有时候输出“正面情感”,有时候输出“positive”,有时候还带个句号。后来我学乖了,输出格式用JSON schema约束,并且在提示词里明确说“只输出JSON,不要输出任何其他内容”。即使这样,偶尔还是会有模型“不听话”的时候,所以后处理里必须加一层解析和兜底逻辑。

还有一个经验是:提示词要版本化管理。不要直接在代码里写死提示词字符串,而是把提示词抽出来放到配置文件或者数据库里,每次修改都记录版本和对应的效果指标。这样当效果波动的时候,你可以快速回滚到上一个版本,也可以对比不同版本的效果差异。我现在的做法是,提示词用YAML文件管理,每个版本带一个注释说明改了什么、为什么改,配合A/B测试框架,能比较科学地迭代。

3.3 推理链设计:模型调用前后的所有“隐形工作”

推理链这个词听起来很玄,其实说白了就是:从用户输入到最终输出,中间除了模型调用本身,还有哪些处理步骤。这些步骤往往是决定系统稳定性的关键,但也是最容易被忽略的。

我拿一个典型的问答场景来拆解。用户输入一个问题,系统需要:1)判断问题类型(是事实查询、观点咨询还是闲聊);2)如果是事实查询,决定是否需要检索外部知识;3)如果需要检索,把问题转成查询向量,检索相关文档;4)把检索结果和问题组装成提示词;5)调用模型;6)解析模型输出;7)如果输出格式不对,重试或者兜底;8)返回结果。

这八步里,每一步都有失败的可能。问题分类可能分错,检索可能召回不相关的内容,模型可能输出格式错误,等等。我的做法是,每一步都加日志和监控,记录输入、输出、耗时、是否成功。这样当最终结果不对的时候,我可以沿着链路一步步排查,看是哪一步出了问题。

还有一个重要的设计决策是:同步还是异步。如果推理链比较长,同步调用会让用户等很久。我的经验是,超过3秒的操作就应该考虑异步化。比如文档索引、批量推理这些,完全可以放到后台任务队列里,前端只返回一个任务ID,用户过一会儿再来查结果。这样用户体验好很多,系统压力也小。

3.4 评估与监控:没有度量就没有改进

AI应用和传统软件最大的区别在于:传统软件的bug是确定性的,输入A一定得到错误B;AI应用的“bug”是概率性的,同样的输入可能这次对下次错。所以你不能靠“测试用例全过”来判断系统是否正常,必须建立一套持续的评估和监控机制。

评估这块,我分两层:离线评估和在线监控。离线评估是在开发阶段,用一批标注好的测试集来跑,看准确率、召回率、F1这些指标。测试集的构建很关键,要覆盖各种边界情况,不能只挑简单的样本。我一般会从线上真实数据里采样,然后人工标注,保证测试集和实际分布一致。

在线监控是在生产环境,实时跟踪关键指标。除了常规的延迟、错误率,AI应用还需要监控一些特殊指标:比如输出长度分布(突然变短可能意味着模型在“偷懒”)、输出格式错误率、检索命中率、用户反馈(点赞点踩)。这些指标能帮你提前发现效果退化,而不是等用户投诉了才知道。

我踩过的一个坑是:只监控了技术指标(延迟、错误率),没监控业务指标(回答质量)。结果模型供应商悄悄更新了模型版本,技术指标一切正常,但回答质量明显下降,过了两周才发现。后来我加了一个“黄金测试集”的定时任务,每天跑一遍,对比历史结果,一旦发现显著差异就告警。这个做法虽然简单,但特别有效。

4. 实操过程:从零搭建一个可用的AI问答系统

4.1 环境准备与最小依赖安装

我不喜欢一上来就装一堆框架,所以这个实操里,核心依赖只有几个:Python 3.10以上、一个HTTP客户端(requests或者httpx)、numpy(做向量计算)、以及一个模型API的SDK。向量数据库先用numpy实现,不引入额外依赖。

pip install httpx numpy python-dotenv

python-dotenv是用来管理API密钥的,不要把密钥硬编码在代码里,这是个基本的安全习惯。我一般会在项目根目录建一个.env文件,里面写API_KEY=xxx,然后.gitignore里加上.env,避免误提交。

提示:如果你用的是某个云厂商的模型服务,SDK的安装方式可能不同,但核心思路是一样的——保持依赖最小化,能自己写的就自己写,实在写不了的再引库。

4.2 第一步:跑通最笨的端到端流程

先别管什么架构,写一个最简单的脚本,把“读文档-切分-检索-拼提示词-调模型-输出”这条链路跑通。我把它叫做v0_baseline.py。

import os import numpy as np import httpx from dotenv import load_dotenv load_dotenv() # 1. 读文档 with open("knowledge.txt", "r", encoding="utf-8") as f: raw_text = f.read() # 2. 简单分块:按段落切 chunks = [p.strip() for p in raw_text.split("\n\n") if p.strip()] print(f"共切出 {len(chunks)} 个块") # 3. 向量化(这里用随机向量模拟,实际应该调embedding接口) def fake_embed(text): np.random.seed(hash(text) % 2**32) return np.random.randn(128) chunk_vectors = np.array([fake_embed(c) for c in chunks]) # 4. 检索 def search(query, top_k=3): q_vec = fake_embed(query) scores = chunk_vectors @ q_vec / ( np.linalg.norm(chunk_vectors, axis=1) * np.linalg.norm(q_vec) ) top_idx = np.argsort(scores)[-top_k:][::-1] return [chunks[i] for i in top_idx] # 5. 拼提示词并调用模型 def ask(query): contexts = search(query) prompt = f"""基于以下资料回答问题。如果资料中没有相关信息,就说不知道。 资料: {chr(10).join(contexts)} 问题:{query} 回答:""" # 实际调用模型API # response = httpx.post(...) # return response.json()["answer"] return prompt # 先打印提示词看看 if __name__ == "__main__": print(ask("你的测试问题"))

这个脚本很粗糙,embedding是假的,模型调用是注释掉的,但它把整个链路的结构展示出来了。你可以先跑一遍,看看分块结果、检索结果、拼出来的提示词长什么样。这一步的目的是建立直觉:你知道数据是怎么流动的,每个环节的输出是什么样子。

4.3 第二步:替换真实组件,逐个验证

基线跑通之后,开始逐个替换。先替换embedding,用真实的embedding接口。这时候你会遇到第一个工程问题:批量调用还是逐条调用。逐条调用简单,但慢;批量调用快,但需要处理批次大小和错误重试。我的做法是写一个带重试的批量调用函数,批次大小根据接口限制来定,一般32或者64。

def embed_batch(texts, batch_size=32, max_retries=3): all_vectors = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] for attempt in range(max_retries): try: # vectors = call_embedding_api(batch) vectors = [fake_embed(t) for t in batch] # 占位 all_vectors.extend(vectors) break except Exception as e: if attempt == max_retries - 1: raise print(f"批次 {i} 第 {attempt+1} 次失败,重试...") return np.array(all_vectors)

替换完embedding,再替换模型调用。这时候你会遇到第二个工程问题:超时和重试。模型接口有时候会慢,有时候会返回错误。我的经验是,超时设30秒,重试2次,重试之间加指数退避。如果重试都失败了,返回一个兜底回答,而不是让整个请求挂掉。

4.4 第三步:加上缓存和异步,提升响应速度

到这一步,系统能用了,但每次请求都要重新算embedding、重新检索、重新调模型,响应时间可能好几秒。优化的第一步是加缓存。缓存分两层:embedding缓存和回答缓存。

Embedding缓存很简单,用一个字典存text -> vector的映射,同样的文本不用重复算。回答缓存稍微复杂一点,因为同样的query可能因为上下文不同而需要不同回答。我的做法是,对query做归一化(去空格、转小写),然后哈希作为key,缓存回答。但要注意设置过期时间,因为知识库可能会更新。

from functools import lru_cache import hashlib @lru_cache(maxsize=10000) def cached_embed(text): return tuple(fake_embed(text)) answer_cache = {} def ask_with_cache(query): key = hashlib.md5(query.strip().lower().encode()).hexdigest() if key in answer_cache: return answer_cache[key] result = ask(query) answer_cache[key] = result return result

异步化是另一个大杀器。如果你的模型调用是IO密集型的,用异步可以大幅提升吞吐量。Python里用asyncio和httpx.AsyncClient就能实现。我实测下来,同样的硬件,异步版本的QPS能比同步版本高3到5倍。

4.5 第四步:加上评估脚本,让效果可度量

没有评估,你就不知道优化有没有效果。我一般会建一个eval.py,里面放一批测试问题和标准答案,每次改动后跑一遍,看准确率变化。

test_cases = [ {"question": "问题1", "expected": "答案1"}, {"question": "问题2", "expected": "答案2"}, ] def evaluate(): correct = 0 for case in test_cases: result = ask(case["question"]) if case["expected"] in result: correct += 1 print(f"准确率:{correct}/{len(test_cases)} = {correct/len(test_cases):.2%}") if __name__ == "__main__": evaluate()

这个评估很粗糙,但比没有好。随着项目发展,你可以把评估做得更精细,比如用模型来打分、加人工评估、做A/B测试。关键是养成“改完就跑评估”的习惯,不要凭感觉判断效果好坏。

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

5.1 检索结果不相关,怎么一步步定位

检索不相关是最常见的问题,但原因可能有很多。我的排查顺序是:先看query的embedding是否正常,再看检索到的块的embedding和query的相似度分数,最后看块的内容本身是否包含答案。

如果query的embedding明显异常(比如和任何块的相似度都极低),可能是embedding接口出了问题,或者query里有特殊字符导致编码异常。如果相似度分数正常但内容不相关,可能是分块策略有问题——块太大导致语义稀释,或者块太小导致信息不完整。如果块的内容本身就不包含答案,那就是知识库覆盖问题,需要补充数据。

我遇到过一个典型案例:用户问“如何重置密码”,检索出来的全是“密码强度要求”相关的块。排查发现,是因为“重置”这个词在embedding空间里和“设置”很接近,而知识库里“设置密码”的内容比较多。解决方案是在检索时加上关键词过滤,或者对query做改写,把“重置密码”改写成“忘记密码怎么办”。

5.2 模型输出格式不稳定,怎么加兜底

模型输出格式不稳定是另一个高频问题。你要求输出JSON,它有时候输出JSON,有时候输出带markdown代码块的JSON,有时候还加一句“好的,以下是JSON”。我的兜底策略分三层:第一层,提示词里明确约束,给出schema和示例;第二层,后处理里用正则提取JSON部分;第三层,如果提取失败,重试一次,重试时在提示词里强调“只输出JSON”。

import json import re def parse_json_output(text): # 第一层:直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 第二层:提取代码块里的JSON match = re.search(r'```(?:json)?\s*(.*?)```', text, re.DOTALL) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: pass # 第三层:提取第一个花括号对 match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass return None

如果三层都失败了,就返回一个默认值或者错误提示,不要让整个请求崩掉。

5.3 响应时间忽快忽慢,怎么排查

响应时间波动大,通常有几个原因:模型接口本身不稳定、检索数据量大导致计算慢、缓存命中率低、并发请求互相争抢资源。我的排查方法是加详细的耗时日志,把每个环节的耗时都打出来。

import time def ask_with_timing(query): t0 = time.time() contexts = search(query) t1 = time.time() prompt = build_prompt(query, contexts) t2 = time.time() result = call_model(prompt) t3 = time.time() print(f"检索:{t1-t0:.2f}s,拼提示词:{t2-t1:.2f}s,模型调用:{t3-t2:.2f}s") return result

这样一眼就能看出瓶颈在哪。如果是模型调用慢,考虑换更快的模型或者加缓存;如果是检索慢,考虑加索引或者减少检索范围;如果是拼提示词慢,那可能是提示词太长了,需要精简。

5.4 常见问题速查表

问题现象可能原因排查方法解决方案
检索结果不相关分块策略不当、embedding质量差、query歧义检查相似度分数、人工查看检索块调整分块大小、换embedding模型、query改写
模型输出格式错误提示词约束不够、模型能力不足打印原始输出、检查提示词加schema约束、后处理提取、重试机制
响应时间波动大模型接口不稳定、缓存命中率低、并发争抢加耗时日志、看缓存命中率加缓存、异步化、限流
效果逐渐退化数据分布变化、模型版本更新定期跑评估、对比历史结果更新测试集、锁定模型版本、重新调优
内存占用持续增长缓存无上限、向量数据未释放监控内存、检查缓存大小加缓存淘汰策略、分批处理数据

注意:这张表里的解决方案都是“方向性”的,具体参数需要根据你的实际情况调整。比如缓存淘汰策略,LRU还是LFU,缓存多大,都要看你的数据访问模式。

5.5 几个我踩过的坑和对应的经验

第一个坑:过度依赖框架。我早期用LangChain,它的抽象层很厚,出问题的时候很难定位到底是框架的bug还是我的代码问题。后来我改成自己写核心逻辑,只在非核心环节用框架,排查效率高了很多。我的建议是,核心链路自己写,辅助功能可以用框架。

第二个坑:忽略数据更新。知识库不是一成不变的,文档会更新、会新增。我一开始没做增量索引,每次更新都要全量重建,耗时很长。后来改成增量更新,只对新文档做embedding和索引,旧文档不动。但增量更新也有问题,就是删除的文档需要单独处理,不然会一直留在索引里。我的做法是给每个块加一个source_id和updated_at,定期清理过期的块。

第三个坑:没有做输入长度限制。用户可能输入超长文本,导致embedding接口报错或者模型上下文超限。我的做法是在入口处做长度检查,超过限制就截断或者提示用户。截断的时候要注意,不要截断在句子中间,最好按句子边界截。

第四个坑:日志里打印了敏感信息。AI应用的输入输出可能包含用户隐私,日志里不要直接打印原始文本。我的做法是,日志里只记录文本的哈希值和长度,需要排查的时候再用哈希值去查原始数据。

6. 从能用到好用,下一步可以做什么

走到这一步,你已经有了一个能跑通的AI问答系统,有基本的检索、推理、缓存、评估。但“能用”和“好用”之间还有距离。如果继续往下走,我建议从这几个方向入手。

检索优化。现在的检索是单路向量检索,可以加上关键词检索做混合检索,提升召回率。还可以加一个重排序模型,对检索结果做精排,把最相关的排到前面。重排序模型不需要太大,一个小型的交叉编码器就能有明显提升。

提示词优化。现在的提示词是静态的,可以根据query类型动态选择不同的提示词模板。比如事实查询用一套模板,观点咨询用另一套。还可以加few-shot示例,把历史上的好回答作为示例放进提示词,引导模型输出更符合预期的结果。

评估体系完善。现在的评估是简单的字符串匹配,可以升级成模型打分或者人工评估。还可以加在线A/B测试,对比不同策略的效果。评估数据要持续积累,把线上真实case加进测试集,让评估越来越贴近实际。

成本优化。模型调用是主要成本,可以通过缓存、批处理、模型降级来降低成本。比如简单问题用便宜的小模型,复杂问题才用大模型。还可以对提示词做压缩,去掉冗余信息,减少token消耗。

可观测性建设。现在的日志是打印到控制台,可以接入专业的监控系统,做实时告警和可视化。关键指标包括:请求量、延迟分布、错误率、缓存命中率、检索命中率、模型输出长度分布。这些指标能帮你提前发现问题,而不是等用户投诉。

我个人在实际操作中的体会是,AI工程这件事,前期慢就是快。把数据管道、推理链、评估监控这些基础打牢,后面迭代速度会越来越快。反过来,如果基础不牢,每次优化都像在沼泽里跑步,越跑越累。所以如果你刚开始,不要急着上高级功能,先把最笨的端到端流程跑通,然后逐个环节打磨,每一步都留下可度量的记录。这样走下来,你不仅有一个能用的系统,更重要的是,你有了一个自己能完全理解和控制的系统。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询