1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过不少新人也面试过不少号称“做过大模型应用”的候选人发现一个很普遍的问题大家会用LangChain会写Prompt但一旦线上出了幻觉、延迟飙升、成本失控就完全不知道从哪里下手。这就是典型的“只会用工具不懂工程”。ai-engineering-from-scratch这个标题我理解它想表达的核心诉求是抛开那些封装好的高级框架从最底层的原理和组件出发亲手搭建一套能支撑AI应用运行的工程体系。它解决的不是“怎么让模型说话”的问题而是“怎么让模型说得稳、说得快、说得便宜、说得可控”的问题。适合谁看如果你已经能跑通一个简单的AI Demo但想搞清楚背后到底发生了什么或者你正准备把一个原型推向生产环境那这篇内容就是为你准备的。我自己在过去一年里从零搭建过两套面向C端的AI问答系统踩过的坑包括但不限于向量检索召回率低得离谱、流式输出在弱网下频繁断连、上下文长度爆炸导致费用翻倍。这些问题的根因往往不在模型本身而在工程链路的每一个环节。下面我就按实际搭建的顺序把整套思路和关键细节拆开来讲。2. 整体架构设计先想清楚数据怎么流2.1 为什么我不建议直接上LangChain这类框架很多人一上来就用LangChain的Chain和Agent觉得省事。但我的经验是如果你连最基础的请求链路都没手写过一遍用框架只会让你在出问题时更加迷茫。框架帮你屏蔽了细节但也屏蔽了你对问题的感知。我建议的路径是先用原生SDK手写一遍完整的请求-处理-响应流程理解每个环节的输入输出然后再决定要不要引入框架来提效。从零搭建的架构核心就四层接入层、编排层、模型层、数据层。接入层负责处理用户请求的鉴权、限流、格式校验编排层是核心决定了一次请求要经过哪些步骤比如要不要查知识库、要不要调用外部工具、要不要做多轮改写模型层封装对各家模型API的调用做统一的错误处理和重试数据层包括向量库、缓存、日志存储。这四层之间的数据流必须是清晰的、可追踪的否则线上出了问题你连日志都看不懂。2.2 一次完整请求的生命周期拆解我拿一个典型的RAG问答场景来举例。用户输入一个问题系统首先在接入层做敏感词过滤和长度截断然后进入编排层。编排层第一步是意图识别判断这个问题是闲聊还是需要查知识库。如果是查知识库就把问题做Embedding去向量库检索Top-K个相关片段。这里有个关键决策检索回来的片段要不要做重排序我的做法是如果Top-K的相似度分数差距不大就加一个轻量级的重排序模型把最相关的排到前面。然后把这些片段和原始问题拼成一个Prompt送给模型层生成回答。生成过程中如果开了流式输出就要在接入层做SSE的封装同时把每个token的生成时间戳记下来方便后续分析延迟。整个链路里每一个环节都可能成为瓶颈。Embedding模型选得太大检索就慢重排序模型太复杂首字延迟就高Prompt拼得太长费用就上去了。所以架构设计阶段就要想清楚你的核心指标是什么是首字延迟、是回答准确率、还是单次调用成本不同的指标导向会导致完全不同的技术选型。2.3 关键选型对比向量库、Embedding模型、重排序策略向量库的选择上我实测过FAISS、Chroma、Qdrant和Milvus。如果是单机小规模数据百万级以下FAISS足够快但缺乏持久化和分布式能力Chroma上手最简单但生产环境下的稳定性和并发能力一般Qdrant和Milvus更适合生产但运维成本高。我的建议是原型阶段用Chroma快速验证一旦数据量超过50万条或者QPS超过10就考虑迁移到Qdrant。Embedding模型方面OpenAI的text-embedding-3-small性价比很高但如果你对数据隐私有要求就得用本地的BGE或M3E。这里有个坑不同Embedding模型产出的向量维度不同切换模型意味着整个向量库都要重建。所以一开始就要想清楚别等到数据量大了再换。重排序策略上我试过用Cohere的Rerank API效果确实好但每次检索都要多一次网络调用延迟增加200ms左右。后来我改用本地的BGE-Reranker-Base虽然效果略差一点但延迟可控而且没有额外费用。这个取舍取决于你的场景如果是内部知识库对延迟不敏感可以用API如果是C端产品本地模型更稳妥。3. 核心模块实操手写一个带检索增强的问答链路3.1 环境准备与依赖安装我习惯用Python 3.10以上的版本依赖管理用Poetry而不是pip因为Poetry能锁定依赖版本避免线上环境因为某个包自动升级导致行为不一致。核心依赖包括openai用于调用模型APIsentence-transformers用于本地Embeddingqdrant-client用于向量库操作fastapi和uvicorn用于提供HTTP服务。poetry init poetry add openai sentence-transformers qdrant-client fastapi uvicorn这里有个细节sentence-transformers会默认安装PyTorch如果你的服务器没有GPU记得装CPU版本否则会拉下来几个G的CUDA依赖浪费磁盘空间。我一般会先手动装torch的CPU版再装sentence-transformers。3.2 文档切分与向量化入库文档切分看起来简单但直接影响检索效果。我试过固定长度切分、按段落切分、按语义切分三种方式。固定长度切分最省事但容易把一句话切断按段落切分保留了语义完整性但段落长度差异太大按语义切分效果最好但需要额外的模型来判断句子边界成本高。我的折中方案是先按段落切分如果某个段落超过500个token再按句子切分确保每个片段在200到500个token之间。from sentence_transformers import SentenceTransformer from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, VectorParams, Distance model SentenceTransformer(BAAI/bge-small-zh-v1.5) client QdrantClient(path./qdrant_data) client.recreate_collection( collection_nameknowledge_base, vectors_configVectorParams(size512, distanceDistance.COSINE) ) def index_documents(docs): points [] for i, doc in enumerate(docs): vector model.encode(doc[text]).tolist() points.append(PointStruct( idi, vectorvector, payload{text: doc[text], source: doc[source]} )) client.upsert(collection_nameknowledge_base, pointspoints)注意bge-small-zh-v1.5的输出维度是512如果你换用其他模型这里的size参数必须同步修改否则入库会报错。另外Qdrant的recreate_collection会清空已有数据生产环境千万别这么干应该用create_collection并做好版本管理。3.3 检索与重排序的代码实现检索的时候我一般先取Top-20然后用重排序模型筛出Top-5。为什么是20和5因为向量检索的召回率在前20里通常能达到90%以上但排序不一定准重排序模型对少量候选的排序效果很好但候选太多会拖慢速度。20进5出是我实测下来延迟和效果的平衡点。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-base) def search(query, top_k20, final_k5): query_vector model.encode(query).tolist() hits client.search( collection_nameknowledge_base, query_vectorquery_vector, limittop_k ) pairs [[query, hit.payload[text]] for hit in hits] scores reranker.predict(pairs) ranked sorted(zip(hits, scores), keylambda x: x[1], reverseTrue) return [hit.payload[text] for hit, _ in ranked[:final_k]]这里有个性能陷阱CrossEncoder.predict是同步阻塞的如果并发请求多会拖垮整个服务。我的做法是把重排序放到一个单独的线程池里执行或者用ONNX Runtime加速推理。实测下来用ONNX能把重排序的延迟从300ms降到80ms左右。3.4 流式输出的工程实现流式输出是提升用户体验的关键但实现起来坑很多。首先模型API返回的是SSE格式的数据流你需要逐行解析提取出delta里的内容。其次如果你在中间做了内容审核或者格式化要确保这些操作不会阻塞流的传输。我的做法是用一个异步生成器来包装模型调用每收到一个chunk就yield出去同时在后台异步记录日志。import asyncio from openai import AsyncOpenAI client AsyncOpenAI() async def stream_chat(prompt): stream await client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], streamTrue ) async for chunk in stream: delta chunk.choices[0].delta.content if delta: yield delta注意AsyncOpenAI的流式调用必须在异步函数里使用如果你在同步的FastAPI路由里直接调用会报错。另外流式输出过程中如果客户端断开连接要确保后台的模型调用也被取消否则会浪费token。我一般会在生成器里捕获asyncio.CancelledError然后主动关闭流。4. 性能优化与成本控制那些文档里不会写的细节4.1 缓存策略什么该缓存什么不该缓存缓存是降低成本和延迟最直接的手段但不是什么都能缓存。我的经验是Embedding结果可以缓存因为同样的文本每次Embedding结果都一样检索结果可以缓存但要注意时效性如果知识库更新了缓存必须失效模型生成的回答不建议缓存因为同样的Prompt在不同温度参数下结果不同而且用户可能期望每次回答都有变化。我一般用Redis做两级缓存第一级是Embedding缓存key是文本的MD5value是向量过期时间设24小时第二级是检索缓存key是查询文本的MD5value是检索到的文档ID列表过期时间设1小时。这样既能省下重复Embedding的计算又能减少向量库的查询压力。4.2 上下文窗口管理怎么在有限token里塞进最多信息模型的上下文窗口是有限的而且越长越贵。我见过有人把检索回来的10个片段全部塞进Prompt结果光上下文就占了3000个token模型还没开始生成费用就已经上去了。我的做法是先对检索结果做去重和压缩把相似度低于阈值的片段直接丢掉然后对每个片段做摘要只保留和问题最相关的部分最后按相关性排序从高到低填充直到接近窗口上限的80%就停止。这里有个技巧在Prompt里明确告诉模型“以下内容按相关性从高到低排列请优先参考前面的内容”。这样即使后面的片段被截断模型也能抓住重点。另外如果对话是多轮的历史对话也要做压缩我一般只保留最近三轮的对话更早的用一句话摘要代替。4.3 错误处理与降级方案线上环境什么都会发生模型API超时、向量库连接断开、重排序模型OOM。如果没有降级方案一次故障就会导致整个服务不可用。我的做法是模型调用设置3秒超时超时后自动切换到备用模型比如从GPT-4o切到GPT-4o-mini向量库查询失败时直接跳过检索让模型基于自身知识回答同时在响应里标注“未使用知识库”重排序失败时直接按向量相似度排序不阻塞主流程。这些降级逻辑必须写在代码里而不是靠运维手动切换。我一般会用tenacity库来做重试但重试次数不要超过2次否则会放大延迟。另外所有的降级事件都要打点上报方便后续分析故障率。5. 常见问题与排查技巧实录5.1 检索结果不相关怎么一步步排查检索不相关是最常见的问题排查思路要按链路走。第一步检查Embedding模型是否适合你的语言和领域。我遇到过用英文模型处理中文法律文书效果惨不忍睹换成中文法律领域微调过的模型后召回率直接翻倍。第二步检查文档切分是否合理。如果切分粒度太粗一个片段里混了好几个主题检索时就会引入噪声。第三步检查相似度阈值是否设得太低。我一般会把相似度低于0.6的结果直接丢掉宁可少召回也不要引入不相关内容。如果以上都没问题那可能是重排序模型的问题。你可以把重排序前后的结果都打印出来对比看看重排序是不是把相关的结果排到了后面。如果是说明重排序模型不适合你的场景考虑换模型或者干脆不用。5.2 流式输出中断的几种原因和修复方法流式输出中断最常见的原因是反向代理的超时设置太短。Nginx默认的proxy_read_timeout是60秒如果模型生成一个长回答超过60秒连接就会被切断。解决办法是在Nginx配置里把这个值调到300秒以上。另一个原因是服务端的缓冲区设置。如果用了Gunicorn或Uvicorn要确保--timeout参数足够大并且关闭响应缓冲。还有一种情况是客户端的问题。有些浏览器或HTTP客户端对SSE的支持不完整收到不完整的数据包就断开了。我一般会在服务端加一个心跳机制每隔15秒发送一个空注释行保持连接活跃。同时在前端做自动重连一旦检测到连接断开就从最后一个成功的token继续请求。5.3 成本突然翻倍的排查清单成本翻倍通常不是单一原因我整理了一个排查清单按优先级排序排查项可能原因检查方法Prompt长度检索片段变多或变长打印每次请求的token数模型切换误用了更贵的模型检查环境变量和配置文件重试次数失败重试导致重复调用查看日志中的重试记录缓存失效缓存命中率下降监控Redis的命中率指标并发量流量突增对比QPS和成本曲线我遇到过最隐蔽的一次是缓存key的设计问题key里包含了时间戳导致每次请求都缓存不命中。后来把时间戳去掉只保留文本的MD5命中率立刻从10%升到85%。5.4 几个我踩过的坑和对应的解决方案第一个坑向量库的索引类型选错了。Qdrant默认用的是HNSW索引召回率高但内存占用大。如果你的数据量很大但内存有限可以考虑用IVF索引牺牲一点召回率换取内存节省。第二个坑Embedding模型没有做归一化。有些模型输出的向量没有归一化直接算余弦相似度会得到错误的结果。解决办法是在入库前手动做L2归一化。第三个坑异步代码里混用了同步的HTTP客户端导致事件循环阻塞。我一般会用httpx的异步客户端确保所有网络调用都是非阻塞的。6. 从原型到生产还需要补哪些工程能力6.1 可观测性日志、指标、追踪一个都不能少原型阶段可以靠print调试生产环境必须有一套完整的可观测性体系。日志要结构化每条日志包含请求ID、用户ID、耗时、token数、模型名称等字段方便后续聚合分析。指标要覆盖QPS、延迟分布、错误率、缓存命中率、成本等核心维度。追踪要能串联一次请求经过的所有服务我一般用OpenTelemetry做埋点把每个环节的耗时都记录下来。有了这些数据你才能回答一些关键问题首字延迟的P99是多少哪个环节最耗时成本主要花在哪个模型上没有这些数据优化就是盲人摸象。6.2 评测体系怎么量化回答质量AI应用的质量很难用单一指标衡量我一般会从三个维度做评测相关性、忠实度、流畅度。相关性用人工标注或者GPT-4打分判断回答是否切题忠实度检查回答是否基于检索到的内容有没有编造流畅度用困惑度或者人工评估。我建议每周跑一次评测集跟踪指标变化一旦发现下降就及时排查。评测集的建设也很重要。我一般会从线上日志里采样真实用户问题加上人工构造的边界case组成一个200到500条的评测集。每次模型或Prompt有改动都跑一遍评测集确保没有退化。6.3 安全与合规内容过滤和权限控制内容过滤是必须的但不要只依赖模型自身的安全能力。我一般会在接入层做一层关键词过滤在输出层再做一次模型审核。关键词过滤用AC自动机实现速度快能拦截大部分明显违规的内容模型审核用一个小模型或者API处理更隐蔽的违规。权限控制方面不同用户能访问的知识库范围不同检索的时候必须带上用户ID做过滤防止越权访问。6.4 持续迭代怎么根据线上反馈优化系统线上反馈是最宝贵的优化依据。我一般会做两件事一是收集用户的点赞点踩作为评测集的补充二是分析低分回答的共性看看是检索问题、Prompt问题还是模型问题。如果是检索问题就调整切分策略或换Embedding模型如果是Prompt问题就优化指令的表述如果是模型问题就考虑换模型或者做微调。迭代的节奏也很重要。不要一次性改太多东西每次只改一个变量然后观察指标变化。否则出了问题你都不知道是哪个改动导致的。我一般会保持两周一个迭代周期每次迭代只解决一个核心问题。最后分享一个我在实际搭建中体会最深的心得AI工程的核心不是模型而是数据流。模型只是数据流中的一个节点真正决定系统上限的是数据怎么进来、怎么处理、怎么出去。把这条链路理顺了换什么模型都不会太差链路不顺用再贵的模型也是白搭。