1. 从零搭建AI工程体系，为什么我劝你别一上来就啃框架

这两年AI应用开发的门槛肉眼可见地降低了，随便拉个程序员过来，告诉他“调个API把大模型接进业务里”，半天就能跑通一个Demo。但真正把AI能力做成一个能上线、能扛量、能持续迭代的工程系统，跟写个Demo完全是两码事。我见过太多团队，模型选型讨论了两周，Prompt调了几十版，结果卡在“怎么把这一堆脚本变成一个服务”上，最后项目不了了之。

ai-engineering-from-scratch这个标题，说的就是从零开始构建AI工程能力这件事。它不是教你训一个大模型，也不是教你调某个具体框架的API，而是把AI应用从“能跑”推进到“能交付”的整套工程方法论。核心解决的是这么几个问题：模型怎么接、数据怎么流、服务怎么部署、效果怎么评估、线上出问题怎么排查。适合谁看？我认为有三类人最该认真读：一是刚转行做AI应用的后端或全栈工程师，二是带团队做AI产品但缺乏工程规范的技术负责人，三是自己想做AI产品但被工程细节卡住的独立开发者。

我自己的经历比较典型。最早做AI应用的时候，我也是“脚本流”——一个Python文件里塞满模型调用、数据处理、结果输出，本地跑得好好的，一上服务器就各种超时、内存溢出、并发崩溃。后来踩了足够多的坑，才慢慢把这一套东西拆成清晰的模块：接入层、编排层、缓存层、评估层、监控层。这篇文章我就把这套从零搭建的思路完整拆一遍，包括每一步为什么这么做、参数怎么定、坑在哪里。你不需要有很深的AI背景，但最好有一点后端开发经验，这样理解起来会更顺。

2. 整体架构设计：先把数据流画清楚，再动手写代码

2.1 为什么“先画数据流”比“先选框架”重要十倍

很多人做AI工程的第一步是打开文档看LangChain怎么用、LlamaIndex怎么配，这其实是本末倒置。框架是工具，数据流才是骨架。你连请求从哪进来、经过哪些处理、在哪一步调用模型、结果怎么返回都没想清楚，选什么框架都是白搭。

我习惯的做法是拿一张白纸，把整个链路画成一条线：用户输入 → 预处理 → 上下文组装 → 模型调用 → 后处理 → 结果返回。每个环节标注三件事：输入什么格式、输出什么格式、可能出什么错。这一步花半小时，能省掉后面至少两天的返工。

举个具体例子。假设你要做一个智能客服问答系统，数据流大概是这样：用户问题进来，先做意图识别（判断是咨询、投诉还是闲聊），然后根据意图走不同的处理分支，咨询类走知识库检索增强，投诉类走工单系统对接，闲聊类直接走通用模型。每个分支的模型调用参数、超时设置、降级策略都不一样。如果你不提前画清楚，写到一半发现“哎这个分支怎么处理”，就得回头重构。

提示：数据流图不用画得多漂亮，用纸笔或者白板工具都行，关键是每个环节的输入输出和异常路径都要标出来。我一般会额外标一个“最坏情况”——比如模型超时了怎么办、检索没结果怎么办，这些边界条件才是工程化的核心。

2.2 分层设计：接入层、编排层、能力层、数据层

数据流画清楚之后，就可以做分层了。我推荐四层结构，这个结构在我做过的多个AI项目里都验证过，扩展性和可维护性都不错。

接入层负责跟外界打交道，包括HTTP接口、WebSocket、消息队列消费等。这一层只做三件事：参数校验、限流、请求转发。不要在这一层写任何业务逻辑，更不要直接调模型。我见过有人把模型调用写在Controller里，后来想换个模型供应商，改了几十个文件，痛苦得要命。

编排层是整个系统的核心，负责串联各个能力模块。比如一个请求进来，编排层决定先调检索、再调模型、最后调后处理。这一层用代码写也行，用工作流引擎也行，关键是要把流程和具体实现解耦。我早期用硬编码的if-else写编排，后来流程一复杂就变成意大利面条，改成配置驱动的工作流之后清爽多了。

能力层是真正干活的地方，包括模型调用、向量检索、文本处理、外部API对接等。每个能力封装成一个独立的模块，对外暴露统一的接口。比如模型调用模块，不管底层是OpenAI还是本地部署的模型，对外都是generate(prompt, params)这样一个方法。这样上层编排层不需要关心底层用的是什么模型。

数据层负责持久化和缓存。对话历史、检索索引、评估结果、日志，这些都要有明确的存储方案。我一般会用PostgreSQL存结构化数据，Redis做缓存和会话状态，向量数据用专门的向量数据库。不要小看这一层，AI应用的数据量增长往往比预期快得多，早期不设计好，后期迁移成本极高。

2.3 技术选型：别追新，选你团队最熟悉的

技术选型这块我踩过的坑最多。早期总想用最新最酷的框架，结果团队没人熟悉，出了问题查文档都查不到。后来我定了一个原则：核心链路用最成熟的技术，边缘功能可以尝鲜。

具体来说，Web框架用FastAPI或者Flask就行，别纠结性能，AI应用的瓶颈在模型调用不在Web层。任务队列用Celery或者RQ，处理异步任务足够。向量数据库选型看数据量，百万级以下用FAISS或者Chroma就够，千万级以上再考虑Milvus或者Qdrant。模型调用层建议自己封装一层，不要直接依赖某个SDK，这样换供应商的时候只改一个文件。

有个细节值得说：模型调用的超时和重试策略。我见过太多人用默认配置，结果模型响应慢的时候整个服务被拖死。我的经验是，同步接口超时设15秒，异步任务超时设60秒，重试最多两次，且第二次重试要换一个模型实例或者降级到更小的模型。这些参数不是拍脑袋定的，是根据实际压测结果调的——15秒覆盖了95%的正常请求，超过这个时间的请求大概率是模型侧有问题，继续等只会浪费资源。

3. 核心模块拆解：每个环节的实操要点和避坑指南

3.1 模型接入层：统一接口是王道

模型接入看起来简单，不就是调个API吗？但实际做起来，光是处理不同供应商的返回格式就够你喝一壶的。OpenAI的返回结构、Claude的返回结构、国内各家模型的返回结构都不一样，如果你在业务代码里直接处理这些差异，后面换模型就是噩梦。

我的做法是定义一个统一的ModelResponse结构，包含content、usage、finish_reason、raw四个字段。所有模型适配器负责把各家返回转成这个结构。这样业务层只认ModelResponse，完全不关心底层是谁。

from dataclasses import dataclass from typing import Optional, Dict, Any @dataclass class ModelResponse: content: str usage: Dict[str, int] finish_reason: str raw: Optional[Dict[str, Any]] = None class ModelAdapter: def generate(self, prompt: str, **kwargs) -> ModelResponse: raise NotImplementedError class OpenAIAdapter(ModelAdapter): def generate(self, prompt: str, **kwargs) -> ModelResponse: # 调用OpenAI API并转换返回格式 resp = self.client.chat.completions.create( model=kwargs.get("model", "gpt-4"), messages=[{"role": "user", "content": prompt}], temperature=kwargs.get("temperature", 0.7), max_tokens=kwargs.get("max_tokens", 2048) ) return ModelResponse( content=resp.choices[0].message.content, usage={"prompt_tokens": resp.usage.prompt_tokens, "completion_tokens": resp.usage.completion_tokens}, finish_reason=resp.choices[0].finish_reason, raw=resp.model_dump() )

参数配置这块，temperature和max_tokens是最常调的两个。我的经验值：做事实性问答temperature设0.1到0.3，做创意生成设0.7到0.9，做代码生成设0.2左右。max_tokens不要设太大，够用就行，设大了不仅浪费钱，还可能让模型输出一堆废话。一般问答场景512到1024足够，长文生成再往上加。

注意：模型接入层一定要做熔断。当某个模型连续失败超过阈值（比如10秒内失败5次），自动切换到备用模型，同时发告警。这个机制在模型供应商出问题的时候能救你一命。我试过一次某供应商突发故障，因为提前配了熔断和备用模型，业务几乎没受影响。

3.2 上下文管理：别把整个对话历史都塞进去

上下文管理是AI工程里最容易被低估的环节。很多人图省事，把整个对话历史拼成prompt直接发给模型，结果token消耗飞快，响应越来越慢，最后超出模型上下文限制直接报错。

正确的做法是分层管理上下文。我把上下文分成三层：系统提示词、近期对话、相关记忆。系统提示词固定不变，定义模型的人设和任务边界。近期对话保留最近N轮，N根据场景定，一般5到10轮。相关记忆是从历史对话里检索出来的关键信息，比如用户之前提过的偏好、已经确认过的事实。

具体实现上，我会维护一个滑动窗口，当对话轮数超过阈值时，把最早的几轮对话做摘要压缩，压缩后的摘要作为“长期记忆”存起来。下次组装上下文时，系统提示词 + 长期记忆摘要 + 近期对话，这样既保留了关键信息，又控制了token量。

class ContextManager: def __init__(self, max_recent_turns=8, max_tokens=3000): self.max_recent_turns = max_recent_turns self.max_tokens = max_tokens self.history = [] self.summary = "" def add_turn(self, role: str, content: str): self.history.append({"role": role, "content": content}) if len(self.history) > self.max_recent_turns * 2: self._compress() def _compress(self): # 把最早的几轮对话做摘要 old_turns = self.history[:4] self.summary = self._summarize(old_turns) self.history = self.history[4:] def build_prompt(self, system_prompt: str, user_input: str) -> str: parts = [system_prompt] if self.summary: parts.append(f"历史对话摘要：{self.summary}") for turn in self.history: parts.append(f"{turn['role']}: {turn['content']}") parts.append(f"user: {user_input}") return "\n".join(parts)

这里有个关键参数：摘要触发的阈值。我一般设成对话轮数超过8轮或者总token超过3000就触发压缩。这个值不是固定的，要根据你的模型上下文窗口大小和业务对历史信息的依赖程度来调。如果业务需要记住很多细节，阈值可以放宽；如果只是简单问答，可以收紧。

3.3 检索增强：向量检索不是银弹，混合检索才靠谱

检索增强生成（RAG）现在几乎是AI应用的标配，但很多人做出来的效果并不好。问题往往出在检索环节——纯向量检索在语义相似度上表现不错，但对关键词匹配、精确查询、数字和专有名词的处理经常翻车。

我的经验是混合检索：向量检索 + 关键词检索，两路结果合并后重排序。向量检索负责语义匹配，关键词检索（比如BM25）负责精确匹配。两路各取Top 20，合并去重后用重排序模型（比如bge-reranker）精排，取Top 5送给模型。

分块策略也很关键。我试过固定长度分块、按段落分块、按语义分块，最后发现按段落分块 + 重叠窗口效果最稳。具体来说，每个块300到500字，相邻块之间重叠50到100字，这样不会因为分块边界把关键信息切断。块太小信息不完整，块太大检索精度下降，300到500字是我实测下来比较平衡的范围。

提示：检索这块一定要做评估。准备一批测试问题，人工标注正确答案在哪个文档块里，然后跑检索看召回率。召回率低于80%就说明检索环节有问题，这时候调模型参数没用，得先修检索。我见过太多人检索召回率只有50%就急着调Prompt，纯属浪费时间。

3.4 输出后处理：模型说的不一定对，得验一遍

模型输出不能直接返回给用户，这是铁律。后处理至少要做三件事：格式校验、事实校验、安全过滤。

格式校验最简单，如果你要求模型输出JSON，就用JSON解析器验一遍，解析失败就重试或者降级。事实校验复杂一些，对于关键信息（比如数字、日期、人名），可以用规则或者小模型做二次验证。安全过滤就是敏感词和敏感内容的检测，这个必须有，而且要在返回给用户之前做。

我一般会加一个置信度评估环节。让模型在输出的时候附带一个置信度分数，或者用另一个模型对输出做打分。置信度低于阈值的请求，要么走人工审核，要么返回一个保守的回复。这个机制在客服、医疗、金融等场景特别重要。

def post_process(raw_output: str, context: dict) -> dict: result = {"content": raw_output, "confidence": 1.0, "flags": []} # 格式校验 try: parsed = json.loads(raw_output) result["parsed"] = parsed except json.JSONDecodeError: result["flags"].append("format_error") result["confidence"] *= 0.5 # 事实校验（示例：检查数字是否在合理范围） numbers = extract_numbers(raw_output) for num in numbers: if not is_reasonable(num, context): result["flags"].append(f"unreasonable_number:{num}") result["confidence"] *= 0.7 # 安全过滤 if contains_sensitive(raw_output): result["flags"].append("sensitive_content") result["confidence"] = 0.0 return result

4. 部署与运维：上线才是真正的开始

4.1 部署方案：容器化是底线，别在裸机上跑

AI应用的部署跟普通Web应用有个很大的区别：资源需求波动大。模型调用是IO密集型，但预处理和后处理可能是CPU密集型，向量检索又吃内存。如果所有模块混在一起部署，资源争抢会很严重。

我的做法是按模块拆分部署。Web层和编排层打包成一个容器，模型调用层单独一个容器，向量检索单独一个容器。每个容器独立扩缩容，互不影响。用Docker Compose做本地开发，Kubernetes做生产部署。如果团队规模小，用Docker Swarm也够，别为了K8s而K8s。

资源限制一定要设。我见过没设内存限制的容器把宿主机搞挂的。经验值：Web层限制512MB内存，模型调用层限制1GB，向量检索层根据数据量定，一般2GB起步。CPU限制用--cpus参数，Web层1核，模型调用层2核，向量检索层2核。这些值要根据实际压测调整，但一定要有上限。

4.2 监控告警：没有监控的AI系统就是盲盒

AI系统的监控比普通系统复杂，因为除了常规的QPS、延迟、错误率，还要监控模型质量指标。我一般会监控这几类：

质量指标这块特别重要。我一般会每天抽样一批线上请求，人工评估或者用评估模型打分，跟踪质量变化趋势。如果发现质量下降，要能快速定位是模型侧的问题、检索侧的问题还是数据侧的问题。

注意：告警一定要分级。P0告警（服务不可用）直接打电话，P1告警（质量下降）发消息通知，P2告警（成本超支）发邮件。不分级的后果就是告警疲劳，最后没人看告警。

4.3 成本控制：token就是钱，省着点花

AI应用的成本大头在模型调用。我见过一个月烧掉几万块token费的团队，一问才知道，很多请求根本不需要调大模型。成本控制有几个立竿见影的手段：

缓存是最有效的。相同或相似的请求直接返回缓存结果，能省掉大量重复调用。我一般用两级缓存：精确匹配用Redis，语义相似用向量缓存。语义缓存的阈值设0.95以上，低于这个值说明问题差异较大，不适合复用。

模型分级也很关键。简单问题用小模型，复杂问题才用大模型。我一般会先用规则或者小模型做意图分类，简单问答走小模型，复杂推理走大模型。实测下来能省40%到60%的成本。

Prompt压缩是另一个手段。把系统提示词精简到最核心的几句话，去掉冗余的示例和说明。我试过把一个2000token的系统提示词压到500token，效果几乎没变化，但每次调用省了75%的输入token。

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

5.1 模型响应慢，怎么快速定位

模型响应慢是最常见的问题，排查思路要按链路走。先看是网络问题还是模型问题：在服务器上直接curl模型API，如果curl快但应用慢，说明是应用层的问题；如果curl也慢，那就是网络或者模型侧的问题。

应用层的问题通常是这几个：连接池不够、序列化耗时、上下文太长。连接池我一般设20到50，根据并发量调。序列化用orjson比标准json快3到5倍。上下文长度要定期检查，超过模型窗口的80%就要考虑压缩。

模型侧的问题一般是负载高或者限流。这时候要么切换备用模型，要么加队列缓冲。我一般会配一个降级策略：主模型超时超过3秒，自动切到备用模型，同时记录日志。

5.2 输出质量不稳定，怎么系统性排查

输出质量不稳定，原因可能出在输入、模型、后处理任何一个环节。我的排查顺序是：先固定输入看输出是否稳定，如果输入固定输出还飘，那就是模型参数的问题（temperature太高）；如果输入固定输出稳定，那就是输入本身的问题（上下文组装有随机性）。

上下文组装这块有个隐蔽的坑：字典遍历顺序。Python 3.7之前字典是无序的，如果上下文里用了字典且依赖顺序，每次组装的prompt可能不一样。虽然现在字典有序了，但如果你用了set或者从数据库查出来的数据没排序，还是会有随机性。我一般会在组装上下文之前对所有数据进行显式排序。

5.3 并发上来就崩，怎么压测和调优

并发崩溃通常是资源瓶颈。先用压测工具（比如locust或者wrk）找到瓶颈点：是CPU打满了、内存不够了、还是连接数超了。找到瓶颈之后针对性优化。

CPU瓶颈就加机器或者优化代码，内存瓶颈就查泄漏或者加内存，连接数瓶颈就调连接池参数。我一般会做阶梯压测：10并发、50并发、100并发、200并发，每个级别跑5分钟，观察各项指标的变化。找到拐点之后，把生产环境的资源按拐点的1.5倍配置。

提示：压测一定要用真实数据。我见过用假数据压测没问题，一上真实数据就崩的，因为真实数据的分布和长度跟假数据完全不一样。压测数据从线上日志里采样，脱敏后使用。

5.4 常见问题速查表

6. 我踩过的几个大坑和最后的经验分享

第一个大坑是过早优化。早期我总想着把架构设计得完美，结果花了两周搭架子，真正跑通业务只用了两天。后来我学乖了，先用最简方案跑通闭环，再根据实际瓶颈逐步优化。AI工程这个领域变化太快，过度设计往往意味着白费功夫。

第二个坑是忽视数据质量。有段时间输出质量怎么调都上不去，最后发现是检索库里的文档本身就有大量错误和过时信息。模型再强，喂进去垃圾也只能产出垃圾。现在我都会在数据入库前加一道清洗和校验流程，虽然麻烦，但值得。

第三个坑是不做评估。早期改Prompt全凭感觉，改完觉得“好像好了一点”就上线，结果经常是这边好了那边坏了。后来我建了一个评估集，每次改动都跑一遍评估，用数据说话。评估集不用很大，100到200条高质量标注就够，但一定要覆盖主要场景和边界情况。

最后分享一个实用技巧：给每个请求打上trace_id，从入口到模型调用到后处理，全链路日志都带上这个ID。出问题的时候，拿trace_id一搜，整个链路的耗时、输入输出、中间状态全都能看到。这个习惯帮我省了无数排查时间，强烈建议你从第一天就加上。