1. 从零搭建AI工程体系为什么值得认真做一遍这两年AI应用层的热闹程度不用我多说各种模型、框架、工具层出不穷但真正落到工程落地的时候很多人会发现一个尴尬的现实demo跑得飞快上线之后处处是坑。模型调用超时、上下文管理混乱、评测没有标准、成本失控、版本迭代之后效果回退——这些问题几乎每一个做过AI应用的人都会遇到。ai-engineering-from-scratch这个项目标题核心指向的就是一件事不依赖现成的高级封装从最基础的环节开始把AI工程化的完整链路自己搭一遍。它解决的不是“怎么调一个API”这种问题而是“怎么让AI能力稳定、可观测、可迭代地跑在生产环境里”这个更本质的问题。适合看这篇内容的人有三类一是刚转做AI应用开发、想建立完整工程认知的工程师二是已经在做AI产品、但总觉得系统“不稳”的开发者三是想理解AI工程全貌、方便做技术决策的技术负责人。我会按照从整体设计到具体实操的顺序把这条链路拆开讲清楚包括每一步为什么这么做、参数怎么定、坑在哪里。需要先说明一点下面涉及的具体工具选型和参数是基于当前主流工程实践的合理补充不同团队可以根据自己的技术栈调整但底层的设计逻辑是通用的。2. 整体架构设计与核心思路拆解2.1 为什么强调“from scratch”很多人第一反应是现在框架这么多LangChain、LlamaIndex、各种Agent框架为什么还要从零做这个问题我认真想过也踩过坑。早期我直接用高层框架搭了一个问答系统两周上线看起来很顺。但后来要改一个检索策略、要加一个自定义的评测指标、要排查一次响应变慢的原因我发现框架的抽象层反而成了障碍——我不知道它内部到底做了什么日志也看不透。from scratch的价值不在于“重复造轮子”而在于建立对每一层的掌控力。当你自己写过一次上下文拼接、自己实现过一次重试和降级、自己搭过一次评测流水线你再用框架的时候就知道该在哪里信任它、在哪里绕过它。这是工程判断力的来源。从架构上看一个完整的AI工程体系大致分五层我把它整理成下面这张表方便对照理解层级职责关键产出接入层统一模型调用、鉴权、限流稳定的模型客户端编排层提示词管理、上下文组装、流程控制可复用的调用链路数据层文档处理、向量化、检索可检索的知识库评测层效果度量、回归测试可量化的质量指标观测层日志、追踪、成本统计可排查的运行数据这五层不是必须一次性全做完但心里要有这张图。我见过太多项目只做了接入层和编排层就上线结果出了问题完全靠猜这就是没有观测层和评测层的代价。2.2 分层设计的取舍逻辑为什么要把接入层单独抽出来因为模型是会换的。今天用这个模型明天可能因为成本或效果换成另一个如果调用逻辑散落在业务代码各处换一次模型就是一场灾难。把模型调用收敛到一个客户端里对外只暴露统一接口换模型的时候只改一个地方。编排层为什么要独立因为提示词和流程是会频繁调整的。把提示词硬编码在业务逻辑里改一句话就要重新发版效率极低。把提示词模板、上下文组装规则、多步流程抽出来可以做到不改代码就调整行为。数据层和评测层是最容易被忽略的两层。数据层决定了AI能不能用上你自己的知识评测层决定了你能不能判断“这次改动到底变好了还是变差了”。没有评测层的AI项目本质上是在盲改。提示不要一上来就追求五层全齐。我的建议是先做接入层和编排层让系统跑起来然后立刻补观测层因为观测层是排查问题的基础。数据层和评测层可以随着需求逐步完善。2.3 技术选型的基本原则选型这件事我的原则是核心链路自己掌控边缘能力用成熟库。比如模型调用、上下文管理、评测逻辑这些核心部分自己写因为需要深度定制而向量计算、文本分块、日志采集这些通用能力用成熟库没必要重复实现。具体到语言Python是当前AI工程的主流选择生态最全。但如果你要做高并发的在线服务可以考虑把关键路径用更高效的语言实现或者用异步框架把IO密集的部分处理好。这个后面实操部分会展开。3. 核心模块的细节解析与实操要点3.1 模型接入层把不稳定的外部调用变稳定模型调用最大的特点就是不稳定。网络会抖、服务会限流、响应会超时。如果业务代码直接调用一次抖动就可能让整个请求失败。所以接入层的核心任务就是把这些不确定性封装掉。我通常会把接入层设计成三个部分客户端、重试策略、降级方案。客户端负责统一封装不同模型的调用差异。不同模型的接口格式、参数名、返回结构都不一样客户端要把它们归一化成统一的输入输出。比如统一用messages数组作为输入统一返回content和usage字段。重试策略要区分错误类型。网络超时、限流这类可重试错误用指数退避重试参数错误、内容违规这类不可重试错误直接返回重试也没用。指数退避的公式是delay base * (2 ** attempt)base一般取0.5到1秒加上随机抖动避免同时重试打爆服务。import time import random def call_with_retry(client, request, max_attempts3, base_delay0.8): for attempt in range(max_attempts): try: return client.call(request) except RetryableError as e: if attempt max_attempts - 1: raise delay base_delay * (2 ** attempt) random.uniform(0, 0.3) time.sleep(delay) except NonRetryableError: raise降级方案是最后一道防线。当主模型连续失败时可以切换到备用模型或者返回一个缓存的兜底结果。降级不是失败而是保证系统整体可用。注意重试次数不要设太多3次基本够了。设太多会导致请求堆积反而拖垮系统。另外重试要有总超时控制避免单个请求占用资源过久。3.2 上下文与提示词管理让行为可调可控提示词管理这块我踩过最大的坑就是把提示词写死在代码里。后来改成模板化管理每个提示词是一个独立的模板文件带版本号可以单独更新和回滚。模板里用占位符表示变量比如{context}、{question}、{history}。组装的时候把实际内容填进去。这样做的好处是调整提示词不需要改代码测试不同版本也方便。上下文组装是另一个关键点。模型的上下文窗口是有限的不可能把所有历史都塞进去。我的做法是分层处理系统提示词永远保留最近几轮对话完整保留更早的对话做摘要压缩检索到的知识按相关度排序后截断。这里有个经验值可以参考假设模型上下文窗口是8K token系统提示词占500检索知识预留3000那么对话历史大概能分到4000左右。按每轮对话平均200 token算大概能保留20轮。超出部分就做摘要。def build_context(system_prompt, history, retrieved_docs, max_tokens8000): budget max_tokens - count_tokens(system_prompt) # 优先保留最近对话 recent [] for turn in reversed(history): if count_tokens(recent) count_tokens(turn) budget * 0.5: break recent.insert(0, turn) # 剩余预算给检索内容 remaining budget - count_tokens(recent) docs truncate_by_relevance(retrieved_docs, remaining) return assemble(system_prompt, docs, recent)提示token计数不要用估算不同模型的分词方式差异很大。用对应模型的分词器来精确计算否则很容易超出窗口导致报错。3.3 数据层文档处理与检索的实操细节数据层要做的事是把非结构化的文档变成可检索的知识。这条链路包括文档解析、文本分块、向量化、存储、检索。文档解析要根据格式选工具。PDF用专门的解析库注意处理扫描件和表格Markdown和纯文本直接读网页内容要先去噪再提取正文。这一步的质量直接决定后续效果解析错了后面全错。文本分块是最有讲究的一步。块太大检索出来的内容冗余浪费上下文块太小语义不完整检索不准。我的经验值是每块300到500个token块之间保留10%到20%的重叠。重叠是为了避免关键信息正好被切在边界上。分块策略也要看内容类型。技术文档按标题层级切代码按函数切对话记录按轮次切。一刀切的固定长度分块效果通常不如按语义边界切。向量化就是把文本块转成向量存起来。检索的时候把用户问题也转成向量算相似度取最相关的几个块。相似度一般用余弦相似度值在0到1之间越高越相关。def retrieve(query, index, top_k5, threshold0.7): query_vec embed(query) results index.search(query_vec, top_k) # 过滤低相关结果 return [r for r in results if r.score threshold]注意相似度阈值不要设太低否则会召回一堆不相关的内容反而干扰模型。我一般从0.7开始调根据实际效果微调。另外top_k也不是越多越好3到5个通常够用。3.4 评测层没有度量就没有改进评测层是区分“玩具项目”和“工程项目”的分水岭。没有评测你根本不知道改动是变好还是变坏。评测分两种离线评测和在线评测。离线评测用固定的测试集每次改动跑一遍看指标变化。在线评测看真实用户的反馈数据。离线评测集要自己攒。从真实问题里挑有代表性的人工标注期望答案。规模不用很大几十到几百条就能反映问题。关键是覆盖面要广包含各种类型的查询。评测指标要看任务类型。问答类任务看答案准确率和相关性生成类任务看流畅度和信息完整度检索类任务看召回率和准确率。有条件的话用模型来辅助打分但要注意模型打分本身也有偏差最好和人工打分做校准。def evaluate(pipeline, test_set): results [] for case in test_set: output pipeline.run(case[input]) score judge(output, case[expected]) results.append(score) return { accuracy: sum(results) / len(results), count: len(results) }提示评测集要定期更新避免过拟合。如果每次改动都在同一批数据上刷分很快这个分数就失去意义了。我一般每季度补充一批新样本。4. 完整实操流程与关键环节实现4.1 环境准备与依赖管理动手之前先把环境理清楚。Python版本建议3.10以上很多新库对低版本支持不好。依赖管理用虚拟环境别装到全局去否则不同项目之间会打架。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install -r requirements.txt依赖要锁定版本。requirements.txt里写死版本号避免某天自动升级导致行为变化。生产环境更推荐用pip-compile生成带哈希的锁定文件。配置文件和环境变量分开管理。密钥、接口地址这类敏感信息放环境变量不要提交到代码仓库。用.env文件本地管理生产环境用配置中心。4.2 搭建最小可用链路第一步先搭一条最小链路接收问题、检索知识、组装上下文、调用模型、返回答案。这条链路跑通了再逐步加东西。class MinimalPipeline: def __init__(self, client, index, prompt_template): self.client client self.index index self.template prompt_template def run(self, question): docs self.index.retrieve(question, top_k3) context \n.join(d.text for d in docs) prompt self.template.format(contextcontext, questionquestion) response self.client.call(prompt) return response.content这条链路虽然简单但已经包含了核心要素。跑通之后你会立刻发现一些问题检索不准、上下文太长、响应太慢。这些问题就是接下来要优化的方向。4.3 加入观测与日志链路跑通后第一件事是加观测。没有观测后面所有优化都是盲猜。要记录的信息包括请求ID、输入问题、检索到的文档ID和分数、组装的提示词、模型返回、耗时、token消耗。这些信息串起来就能还原一次请求的完整过程。import logging import time def traced_run(pipeline, question): trace_id generate_id() start time.time() logger.info(f[{trace_id}] input: {question}) try: result pipeline.run(question) logger.info(f[{trace_id}] output: {result[:200]}) return result finally: elapsed time.time() - start logger.info(f[{trace_id}] elapsed: {elapsed:.2f}s)日志要结构化方便后续查询和统计。用JSON格式输出字段固定这样可以直接导入日志系统做分析。注意日志里不要记录完整的敏感内容。用户输入和模型输出可能包含隐私信息记录前要做脱敏处理或者只记录摘要和哈希值。4.4 参数调优的实操记录参数调优是最需要耐心的一步。我拿检索的top_k举个例子记录一下实际调优过程。初始设置top_k10发现上下文太长模型经常忽略中间的内容而且响应慢。降到top_k3发现有些问题检索不到足够信息答案不完整。最后定在top_k5配合0.7的相似度阈值效果比较平衡。温度参数也调过。问答类任务温度设0.1到0.3保证答案稳定创意类任务设0.7到0.9增加多样性。这个没有标准答案要看具体任务。参数初始值调整后调整原因top_k105减少冗余提升响应速度相似度阈值无0.7过滤低相关结果温度0.70.2问答任务需要稳定输出最大token无限制2000控制成本和延迟调参的时候一次只改一个改完跑评测看指标变化。同时改多个参数你根本不知道是哪个起了作用。4.5 上线前的检查清单上线前我会过一遍这个清单避免低级错误密钥是否从环境变量读取没有硬编码是否有超时控制避免请求无限等待是否有重试和降级外部服务抖动时能兜住日志是否完整出问题能排查是否有成本监控避免账单失控评测是否跑过指标没有明显回退是否有回滚方案出问题能快速恢复这份清单看起来简单但每一条背后都是踩过的坑。我见过因为没设超时导致线程池被占满的也见过因为没监控成本一个月账单翻十倍的。5. 常见问题与排查技巧实录5.1 检索不准的排查思路检索不准是最常见的问题表现是模型答非所问或者答案不完整。排查要按链路一步步来。先看分块是否合理。如果块切得太碎语义不完整检索自然不准。把几个检索结果打出来看看如果都是半句话那就是分块问题。再看向量化是否合适。不同模型对语义的捕捉能力不同有些模型对中文支持不好。可以拿几个典型问题测试看检索结果是否相关。最后看相似度阈值是否合适。阈值太高会漏掉相关内容太低会引入噪声。调这个参数要配合评测集看准确率和召回率的平衡。提示检索问题很多时候根源在数据质量。文档解析错了、分块切乱了、有大量重复内容这些都会影响检索。先把数据层做干净再调检索参数。5.2 响应慢的定位方法响应慢要分段计时定位瓶颈在哪。是检索慢、模型调用慢、还是组装慢。def profile_run(pipeline, question): t1 time.time() docs pipeline.index.retrieve(question) t2 time.time() prompt pipeline.build_prompt(docs, question) t3 time.time() result pipeline.client.call(prompt) t4 time.time() print(fretrieve: {t2-t1:.2f}s, build: {t3-t2:.2f}s, call: {t4-t3:.2f}s)大部分情况下瓶颈在模型调用。这时候可以考虑换更快的模型、减少上下文长度、开启流式输出让用户先看到部分结果。检索慢的话检查索引是否建好、是否用了合适的索引结构。向量检索的数据量大时暴力搜索会很慢需要用近似最近邻算法。5.3 成本失控的预防成本失控通常有几个原因上下文太长、调用次数太多、没有缓存。上下文长度直接决定token消耗。我见过把整个知识库塞进上下文的做法一次调用几万token成本高得离谱。正确做法是只放检索到的相关内容。调用次数要控制。有些流程会多次调用模型比如先判断意图再回答这种要评估是否必要。能一次调用解决的不要拆成多次。缓存能省很多钱。相同或相似的问题直接返回缓存结果。缓存key可以用问题的哈希或者问题的向量做相似匹配。问题类型常见原因解决方向检索不准分块不合理、阈值不当调整分块策略和阈值响应慢模型调用慢、上下文长换模型、缩短上下文、流式输出成本高上下文长、调用多、无缓存精简上下文、合并调用、加缓存效果回退提示词改动、模型升级跑评测、对比版本、及时回滚5.4 版本迭代后的效果回退效果回退是最让人头疼的问题因为往往不明显等发现的时候已经影响用户了。预防的关键是每次改动都跑评测。提示词改了、模型换了、检索参数调了都要跑一遍评测集对比指标。指标下降超过阈值就告警。另外要保留版本历史。提示词、配置、模型版本都要记录出问题能快速定位是哪个改动导致的也能快速回滚。我个人的习惯是任何改动都先在小流量上验证确认没问题再全量。这样即使出问题影响范围也可控。6. 我在实际项目中的几点体会做AI工程这几年最大的体会是模型能力只是起点工程能力才决定上限。同一个模型工程做得好的团队和做得差的团队最终产品体验差距巨大。另一个体会是不要追求一步到位。先把最小链路跑通加上观测然后根据实际问题逐步优化。我见过太多团队一开始就想搭一个完美的架构结果几个月过去还没上线。快速迭代、小步快跑在AI工程里同样适用。最后分享一个实用的小技巧建一个“问题案例库”把线上遇到的bad case都记下来标注原因和解决方式。这个库既是评测集的来源也是团队的经验沉淀。时间长了你会发现大部分问题都是重复的有了这个库排查效率会高很多。这套从零搭建的思路后续还可以往几个方向扩展加入多模型路由根据问题类型自动选模型加入A/B测试框架对比不同策略的效果加入自动化的提示词优化用评测反馈来迭代提示词。这些都是建立在基础链路扎实的前提上的基础打好了往上加东西就是水到渠成的事。