1. 从零搭建AI工程体系为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地在降低随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过不少新人也面试过不少号称“做过大模型应用”的候选人发现一个很普遍的问题大家会用LangChain会写Prompt但一旦线上出了幻觉、延迟飙升、成本失控就完全不知道从哪下手。这就是典型的“只会用工具不懂工程”。ai-engineering-from-scratch这个标题我理解它想表达的核心诉求是抛开那些封装好的高级框架从最底层的工程视角把AI应用从原型到上线的完整链路自己走一遍。它不是一个具体的开源项目名而是一类学习路径和实践方法的统称。适合谁看我认为有三类人一是刚转行做AI应用开发、想补齐工程底子的程序员二是已经在做AI产品、但系统总是“能用但不好用”的开发者三是对AI工程感兴趣、想了解真实生产环境长什么样的技术管理者。这篇文章我会按照我自己实际搭建一套AI工程体系的经验来展开从整体架构设计、核心模块拆解、实操落地步骤到踩过的坑和排查技巧尽量把每个决策背后的“为什么”讲清楚。你不需要有很深的机器学习背景但至少要会写Python、了解基本的HTTP服务概念。读完你至少能自己搭出一套可控、可观测、可迭代的AI应用骨架而不是永远停留在“调包侠”阶段。2. 整体架构设计为什么我不建议一上来就上微服务2.1 从单体到模块化而不是从单体到微服务很多人一提到“工程化”第一反应就是拆微服务、上K8s、搞消息队列。我见过一个三人小团队为了做一个内部知识库问答硬是拆了六个服务结果光是调试链路就耗掉了大半精力。我的建议很明确AI应用的起点应该是一个模块化单体而不是微服务。原因很简单AI应用的核心不确定性在于模型输出而不是服务拆分。你早期最需要快速迭代的是Prompt、检索策略、后处理逻辑这些东西放在一个代码库里改起来最快。等到某个模块真的成为性能瓶颈比如向量检索的QPS扛不住了再把它单独拆出去也不迟。我通常会把一个AI应用分成这么几层接入层负责请求路由和鉴权编排层负责串联整个处理流程能力层包含模型调用、检索、工具调用等具体能力数据层管理向量库、缓存和日志。每一层在代码里是独立的包或模块但部署在一起。这样既保证了边界清晰又避免了分布式系统的复杂度。2.2 编排层是整个系统的心脏如果只能让我强调一个模块那一定是编排层。它决定了用户的一次请求进来之后到底经过哪些步骤、每步用什么参数、失败了怎么降级。我见过太多项目把编排逻辑散落在各个API处理函数里改一个流程要动五六个文件最后没人敢改。我的做法是定义一个显式的Pipeline结构。每个步骤是一个独立的处理单元有明确的输入输出类型。Pipeline本身只负责按顺序执行、处理异常、记录耗时。这样做的好处是你可以单独测试每个步骤也可以很方便地调整顺序或替换某个步骤。比如最开始你用“检索→拼接Prompt→调用模型”后来想加一个“查询改写”步骤只需要在检索前面插一个单元就行其他代码完全不用动。提示编排层不要引入太重的框架。我试过用一些工作流引擎来做这件事结果发现配置成本比手写还高。对于大多数AI应用一个几百行的Pipeline类足够了。2.3 可观测性必须从第一天就设计进去这是我最想强调的一点。AI应用和传统应用最大的区别在于它的输出是不确定的。传统接口返回500你知道是bug但AI应用返回了一段看似合理实则错误的内容你如果不记录中间过程根本无从排查。所以我在设计阶段就会把可观测性作为一等公民。每次请求都要记录原始输入、改写后的查询、检索到的文档片段及相似度分数、最终拼接的Prompt、模型返回的原始内容、后处理后的结果、每步耗时、token消耗。这些数据不一定都要实时展示但必须落盘方便事后分析。我一般会用一个结构化的日志格式比如JSON Lines每行一条完整记录后面无论是做离线评估还是问题复现都很方便。3. 核心模块拆解与实操要点3.1 模型调用层别把SDK直接散落在业务代码里新手最常见的做法是在业务函数里直接import openai然后调chat.completions.create。这样做在Demo阶段没问题但一旦你要换模型、加重试、做降级就会非常痛苦。我的做法是封装一个统一的模型客户端接口业务代码只依赖这个接口。这个接口至少要考虑几件事。第一是重试策略模型服务偶尔超时或返回限流是常态我一般用指数退避重试两到三次但要注意不是所有错误都值得重试比如参数错误重试多少次都没用。第二是超时控制一定要设置合理的超时时间我见过因为没设超时导致请求堆积把服务拖垮的案例。第三是降级方案当主模型不可用时是切换到备用模型还是返回缓存结果还是直接给用户一个兜底话术这些都要提前想好。class ModelClient: def __init__(self, primary, fallbackNone, max_retries2, timeout30): self.primary primary self.fallback fallback self.max_retries max_retries self.timeout timeout def complete(self, messages, **kwargs): for attempt in range(self.max_retries 1): try: return self._call(self.primary, messages, **kwargs) except RetryableError: if attempt self.max_retries and self.fallback: return self._call(self.fallback, messages, **kwargs) time.sleep(2 ** attempt)这段代码看起来简单但把重试、降级、超时都收拢在一个地方后面维护起来会轻松很多。3.2 检索模块向量检索不是银弹做RAG检索增强生成的人很容易陷入一个误区觉得只要把文档切块、向量化、存进向量库检索效果就应该好。实际做下来你会发现纯向量检索在很多场景下表现并不理想尤其是当用户查询包含具体关键词或专有名词时。我的经验是采用混合检索策略向量检索负责语义匹配关键词检索比如BM25负责精确匹配然后对两路结果做融合排序。融合算法我常用RRFReciprocal Rank Fusion它不需要调权重对两路结果的分数尺度不敏感实现也简单。def rrf_fusion(vector_results, keyword_results, k60): scores {} for rank, doc in enumerate(vector_results): scores[doc.id] scores.get(doc.id, 0) 1 / (k rank 1) for rank, doc in enumerate(keyword_results): scores[doc.id] scores.get(doc.id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)另外文档切块策略也非常关键。我试过固定长度切块、按段落切块、按语义切块最后发现对于技术文档按标题层级切块加上适当重叠效果最稳。重叠长度我一般设成块大小的10%到20%避免关键信息刚好被切断。3.3 缓存层省下的都是真金白银AI应用的成本大头在模型调用而很多请求其实是重复或高度相似的。加一层缓存能显著降低成本、提升响应速度。但缓存的设计有几个坑要注意。首先是缓存键的设计。如果你直接用用户原始输入做键那“今天天气怎么样”和“今天天气如何”会命中两次模型调用。我的做法是先对输入做归一化比如去掉标点、统一大小写再结合检索到的文档ID一起做键。这样既保证了语义相同的问题能命中又避免了不同上下文下的错误复用。其次是缓存失效策略。如果你的知识库更新了旧缓存可能返回过时信息。我一般会给缓存设置一个合理的TTL同时在知识库更新时主动清除相关缓存。对于时效性要求高的场景比如新闻问答缓存时间要设得很短甚至不设缓存。注意不要缓存包含用户隐私信息的请求结果除非你做了脱敏处理。这是合规底线。3.4 后处理模块别让模型输出直接见用户模型返回的内容直接展示给用户是有风险的可能包含格式错误、敏感信息、或者不符合产品调性的表达。后处理模块就是最后一道关卡。我通常会在后处理里做几件事格式校验比如要求返回JSON就解析一下解析失败就触发重试或降级敏感词过滤根据业务场景配置词表引用标注如果是RAG场景把模型引用的文档片段标注出来方便用户核实长度控制超长内容做截断或摘要。这里有个实操心得后处理的规则不要写得太死。我见过一个项目因为敏感词表过于严格把“苹果手机”里的“苹果”都过滤了导致正常回答被误伤。规则要可配置、可灰度上线前一定要用真实数据跑一遍。4. 完整实操流程从零搭一套可运行的AI工程骨架4.1 环境准备与依赖管理我假设你用Python做开发。第一步不是急着写代码而是把依赖管理好。我强烈建议用poetry或uv这类工具而不是直接pip install。原因很简单AI项目的依赖又多又容易冲突没有锁文件的话换台机器就跑不起来。# 以uv为例 uv init ai-engineering-demo cd ai-engineering-demo uv add fastapi uvicorn httpx pydantic uv add sentence-transformers rank-bm25这里我刻意没有引入LangChain这类框架因为我们的目标是从零理解工程链路。等你把底层跑通了再用框架提效也不迟。4.2 定义数据结构和接口契约在写任何逻辑之前先把核心数据结构定下来。我一般用Pydantic定义请求和响应模型这样既能做校验又能自动生成文档。from pydantic import BaseModel from typing import List, Optional class QueryRequest(BaseModel): query: str top_k: int 5 use_cache: bool True class RetrievedDoc(BaseModel): doc_id: str content: str score: float source: str class QueryResponse(BaseModel): answer: str references: List[RetrievedDoc] latency_ms: int cached: bool把契约定清楚的好处是前后端可以并行开发测试也可以基于这些模型写。4.3 搭建检索索引我用一个简单的例子演示。假设你有一批Markdown格式的技术文档先做切块。import re from pathlib import Path def split_by_heading(text, max_chunk500, overlap50): sections re.split(r\n(?## ), text) chunks [] for sec in sections: if len(sec) max_chunk: chunks.append(sec) else: for i in range(0, len(sec), max_chunk - overlap): chunks.append(sec[i:i max_chunk]) return chunks然后分别建向量索引和关键词索引。向量索引用sentence-transformers生成embedding关键词索引用rank_bm25。这两步都比较直接我就不展开代码了。关键是要把文档ID和原始内容对应存好后面融合排序和引用标注都要用。4.4 实现编排Pipeline这是核心部分。我把整个流程拆成几个步骤每个步骤是一个函数Pipeline负责串联。class Pipeline: def __init__(self, steps): self.steps steps def run(self, context): for step in self.steps: context step(context) return context每个步骤接收一个context字典返回修改后的context。比如查询改写步骤、检索步骤、Prompt拼接步骤、模型调用步骤、后处理步骤。这样做的好处是每个步骤可以单独测试也可以很方便地调整顺序。4.5 接入可观测性我在每个步骤里都会记录耗时和关键中间结果统一写到一个trace对象里。请求结束后把trace序列化成JSON写日志。import time import json def traced_step(name, func): def wrapper(context): start time.time() result func(context) context[trace].append({ step: name, latency_ms: int((time.time() - start) * 1000), output_summary: summarize(result) }) return result return wrappersummarize函数根据步骤类型提取关键信息比如检索步骤记录命中的文档ID和分数模型调用步骤记录token消耗。这些数据积累下来后面做优化就有依据了。4.6 启动服务与压测用FastAPI把Pipeline包成一个HTTP接口然后用uvicorn启动。上线前一定要做压测我一般用locust或wrk。压测时重点关注几个指标P99延迟、错误率、模型调用的并发限制。很多模型服务对并发有硬限制压测能帮你提前发现瓶颈。提示压测时不要用真实模型调用成本太高。可以先用一个mock客户端返回固定内容把链路压通再小流量接真实模型。5. 常见问题与排查技巧实录5.1 模型输出不稳定同样的问题答案差异很大这是最常见的问题。排查思路分几步。先看是不是温度参数设太高了如果是事实性问答温度设0到0.3比较合适。如果温度已经很低还是不稳定那可能是Prompt本身有歧义或者检索到的上下文每次不一样。这时候要把每次请求的完整Prompt和检索结果打出来对比找到差异点。我遇到过一个案例用户问“这个功能怎么用”检索模块有时召回的是A功能的文档有时是B功能的文档因为查询太模糊了。解决办法是在检索前加一个查询改写步骤结合对话历史把问题具体化。5.2 响应延迟忽高忽低延迟问题要分段排查。先看是检索慢还是模型调用慢。如果检索慢可能是向量库索引没建好或者检索的top_k设太大了。如果模型调用慢要看是网络问题还是模型服务本身负载高。我一般会在trace里记录每步耗时一眼就能看出瓶颈在哪。还有一个容易被忽略的点是冷启动。如果你的服务一段时间没请求第一次请求可能会触发模型加载或连接建立延迟特别高。解决办法是加一个定时预热任务或者保持最小连接数。5.3 成本超出预期成本失控通常有几个原因。一是没有缓存重复请求反复调模型。二是Prompt太长检索返回的文档没有做截断或压缩。三是用了过大的模型处理简单任务。我的做法是给不同类型的请求分配不同的模型简单分类任务用小模型复杂生成用大模型。同时监控每天的token消耗设置告警阈值。5.4 常见问题速查表问题现象可能原因排查方法解决方向答案与问题无关检索召回错误检查检索结果和相似度分数优化切块策略加混合检索答案包含过时信息缓存未失效或知识库未更新检查缓存TTL和索引更新时间更新索引清除相关缓存响应时间超过10秒模型调用超时或检索慢查看trace各步耗时设超时优化检索换更快的模型同一问题答案不一致温度高或上下文不同对比多次请求的Prompt降低温度固定检索结果服务偶尔报错模型服务限流或网络抖动查看错误日志和重试记录加重试和降级设并发限制5.5 几个我踩过的坑第一个坑是过度依赖向量检索。早期我做一个法律问答用户问“劳动合同法第四十七条”向量检索完全找不到正确文档因为向量模型对具体条款编号不敏感。后来加了关键词检索才解决。第二个坑是忽略Prompt长度限制。有一次检索返回了太多文档拼接后的Prompt超出了模型上下文窗口模型直接报错。后来加了token计数和截断逻辑优先保留高分文档。第三个坑是日志记录不完整。早期只记录了最终答案出了问题完全没法复现。后来强制要求记录完整中间过程排查效率提升了很多。6. 迭代与扩展这套骨架还能怎么用6.1 加入评估体系系统跑起来之后下一步就是持续优化。优化的前提是能衡量效果。我一般会建一个小规模的评估集包含问题和标准答案每次改动后跑一遍看准确率、召回率、延迟、成本的变化。评估集不用很大几十到几百条就够但一定要覆盖典型场景和边界情况。评估指标我通常关注这几个答案准确率人工或模型打分、检索命中率正确文档是否在召回结果里、平均延迟、单次请求成本。这些指标定期记录形成趋势图就能看出改动是正向还是负向。6.2 支持多轮对话单轮问答跑通后多轮对话是自然的扩展方向。核心是要维护对话历史并在检索和生成时都考虑上下文。我的做法是把最近几轮对话做摘要和当前问题一起做查询改写这样检索时能带上上下文信息。生成时把对话历史按token预算截断后拼进Prompt。这里要注意的是对话历史不是越长越好。太长的历史会稀释当前问题的权重还会增加成本。我一般只保留最近三到五轮更早的做摘要。6.3 工具调用与Agent化当你的应用需要查实时数据、执行操作时就需要引入工具调用。我的建议是从最简单的单工具开始比如只加一个“查天气”的工具把调用链路跑通。然后再逐步增加工具数量同时做好工具选择的准确性评估。工具调用的一个关键点是错误处理。工具可能超时、返回格式错误、或者参数不对。每个工具调用都要有超时和重试失败时要让模型知道并给出合理的兜底回复而不是直接报错给用户。6.4 部署与扩缩容最后说部署。小规模用单机加uvicorn多worker就够了。规模上来之后检索和模型调用可以拆成独立服务各自扩缩容。模型调用层可以加一个队列做削峰填谷避免突发流量打垮服务。但记住我开头说的不要过早拆微服务等真的遇到瓶颈再拆。我个人在实际操作中的体会是AI工程最难的不是写代码而是建立一套能快速发现问题、定位问题、验证修复的机制。这套骨架的价值就在于它让你在系统还小的时候就把可观测性和可迭代性设计进去后面无论加什么功能都不会失控。