1. 从零搭建AI工程体系，到底在搭什么

第一次看到 "ai-engineering-from-scratch" 这个标题，我脑子里蹦出来的不是某个具体框架，而是一整条链路。很多人把 AI 工程理解成“调个 API、写个 prompt”，真到项目落地才发现，模型只是中间一小环，前面有数据管道，后面有服务治理，中间还夹着评测、监控、成本控制。所谓 from scratch，不是让你从零手写一个 Transformer，而是从零把这条链路搭起来，让一个 AI 功能真正能跑在线上、扛住流量、算得清账。

这个项目适合谁？如果你已经会用 Python，调过几次大模型接口，但一到“怎么把它做成一个稳定服务”就卡壳，那这篇就是写给你的。如果你是从后端或数据方向转过来，想补齐 AI 工程这块拼图，同样适用。我会按我实际搭过几套系统的顺序，把每个环节为什么这么做、参数怎么定、坑在哪，一条条讲清楚。全文围绕的核心关键词就是ai-engineering-from-scratch，也就是从零构建 AI 工程能力这件事本身。

先说结论性的判断：AI 工程和传统后端工程最大的区别，在于不确定性被放大了。传统接口输入输出基本确定，AI 接口同样的输入可能给出不同输出，延迟波动大，成本还跟 token 数挂钩。所以从零搭建时，架构设计的重心要从“功能实现”转向“不确定性管理”。这句话听着虚，后面每一节我都会落到具体做法上。

2. 整体架构设计与技术选型思路

2.1 为什么先画数据流再选框架

我见过太多人一上来就纠结用 LangChain 还是自己写，用 FastAPI 还是 Flask。这个顺序是反的。正确的做法是先画数据流：用户请求进来，经过哪些处理，调用哪些模型，结果怎么返回，中间哪些环节要落库。把这张图画清楚，框架选型是水到渠成的事。

以我最近搭的一个文档问答服务为例，数据流是这样的：请求进来先做鉴权和限流，然后对 query 做预处理（清洗、改写），接着走向量检索拿到候选片段，拼装 prompt 后调用大模型，最后做后处理和引用标注，返回结果并异步写日志。这条链路里，检索和模型调用是两个耗时大头，其余都是轻量操作。画完这张图，你就知道哪些环节需要异步、哪些需要缓存、哪些需要重试。

提示：数据流图不用画得多漂亮，用纸笔或者任意白板工具，把“输入—处理—输出”三段标清楚就行。关键是标出每个环节的耗时量级和失败可能性。

2.2 分层设计：把易变的和稳定的隔开

AI 工程里变化最快的是什么？模型。今天用这家，明天可能换那家；今天这个版本，下周可能升级。所以架构上一定要把模型调用层单独抽出来，用一个统一的接口封装。上层业务只依赖这个接口，不直接依赖任何具体厂商的 SDK。

我的做法是定义一个LLMClient抽象，里面就几个方法：chat()、embed()、stream_chat()。具体实现可以是不同厂商的适配器。这样换模型时，只改适配器，业务代码一行不动。这个思路不新鲜，就是依赖倒置，但在 AI 场景里特别值钱，因为模型迭代太快了。

同样要隔离的还有向量存储层。检索方案从 FAISS 换到 Milvus 再换到 pgvector，业务层不应该感知。定义一个VectorStore接口，add()、search()、delete()三个方法起步，够用了。

2.3 技术栈选型的取舍逻辑

下面这张表是我实际用下来，针对中小规模 AI 服务的选型建议，附带选择理由。

选 pgvector 而不是专用向量库，是我踩过坑之后的决定。专用库性能确实好，但多一套运维成本，小团队扛不住。数据量没到千万级，pgvector 配合合适的索引完全够用，而且能和业务数据放一起做联合查询，省了很多同步逻辑。

3. 核心模块拆解与实操要点

3.1 请求预处理：别小看这一步

很多人直接把用户输入丢给模型，结果就是输出质量忽高忽低。预处理至少要做三件事：清洗（去掉多余空白、特殊字符）、长度控制（超长输入截断或分段）、意图识别（判断这是问答、闲聊还是指令）。

长度控制有个具体计算。假设模型上下文窗口是 8k token，你要留出 2k 给输出，那输入最多 6k。中文大致 1 个字约 1.5 个 token，英文 1 个词约 1.3 个 token。所以中文输入控制在 4000 字以内比较稳。这个数字不是拍脑袋，是实测出来的经验值，留足余量避免截断。

意图识别可以用一个轻量分类模型，也可以先用规则。我的建议是初期用规则加关键词，跑一段时间收集真实数据后再训分类器。上来就训模型，样本都不够，纯属浪费。

3.2 检索增强：召回质量决定上限

RAG 这套东西，召回不行，后面 prompt 写得再花也没用。检索环节的关键参数是chunk size和top-k。

chunk size 我一般设 300 到 500 字。太小，语义不完整；太大，噪声多还浪费 token。具体怎么定？看你的文档类型。技术文档段落短，300 字够；法律合同句子长，500 字更合适。切分时按语义边界切，别硬按字数切，否则一句话被劈成两半，检索出来是残的。

top-k 设多少？常见是 3 到 5。设 1 容易漏，设 10 噪声大还费 token。我的做法是先召回 20 个，再用一个轻量重排模型（比如 cross-encoder）精排取前 3。这样召回率和精度都能兼顾。重排模型不用太大，几千万参数的小模型就够，延迟增加几十毫秒，换来质量明显提升，值。

注意：chunk 之间要保留一定的重叠（overlap），一般设 chunk size 的 10% 到 20%。这样跨 chunk 的语义不会被切断。我吃过亏，一个关键结论正好卡在两个 chunk 边界，检索时两边都只拿到半句，模型直接答错。

3.3 模型调用层：重试、降级、超时一个都不能少

模型调用是最不稳定的环节。网络抖动、服务限流、偶发超时，都是家常便饭。所以这一层必须做三件事。

超时设置：连接超时 5 秒，读取超时根据任务定。短问答 30 秒够，长文生成可能要 120 秒。超时时间设太短，正常请求被误杀；设太长，故障时请求堆积。我的经验是设成 P99 延迟的 1.5 倍。

重试策略：只对可重试的错误重试，比如 429（限流）、5xx（服务端错误）。重试次数 2 到 3 次，用指数退避，间隔 1 秒、2 秒、4 秒。千万别对 400（参数错误）重试，重试多少次都一样，纯浪费。

降级方案：主模型挂了怎么办？准备一个备用模型，或者降级到规则回复。降级不是丢人，是保证服务可用。我一般配两级：主模型失败重试后仍失败，切备用模型；备用也失败，返回兜底话术并记录告警。

import time from typing import Optional def call_with_retry(client, prompt: str, max_retries: int = 3) -> Optional[str]: for attempt in range(max_retries): try: return client.chat(prompt, timeout=30) except RateLimitError: wait = 2 ** attempt time.sleep(wait) except ServerError: time.sleep(2 ** attempt) except BadRequestError: # 参数错误，重试无意义 return None return None

这段代码看着简单，但把“哪些错误该重试”这个判断做对了，能省掉大量无效等待。

3.4 输出后处理：让结果可用

模型输出不能直接返回给用户。至少要做格式校验和敏感内容过滤。如果要求输出 JSON，就用 Pydantic 校验，不合法就触发一次修复重试。修复重试的 prompt 要带上错误信息，比如“你上次输出缺少 field 字段，请补全”。

引用标注也是后处理的一部分。RAG 场景里，把模型引用的片段编号映射回原文，用户点一下能跳转。这个体验提升很大，但实现不难，就是在拼 prompt 时给每个片段编号，输出后解析编号即可。

4. 完整实操流程：从空目录到可运行服务

4.1 项目骨架搭建

先建目录结构。我的习惯是这样：

ai-service/ app/ api/ # 路由层 core/ # 配置、日志、异常 services/ # 业务逻辑 clients/ # 模型、向量库适配器 models/ # 数据模型 tests/ scripts/ # 数据导入、迁移脚本 requirements.txt

这个结构的好处是职责清晰。clients放所有外部依赖的封装，services放纯业务逻辑，api只做参数校验和响应组装。测试时，mock 掉clients就能测services，不用真调模型。

配置管理用 Pydantic Settings，从环境变量读。密钥、数据库地址这些绝不写进代码。我见过有人把 API key 硬编码提交到仓库，第二天就被刷爆了额度，这种低级错误千万别犯。

4.2 数据导入与索引构建

假设你有一批文档要入库。流程是：读取文档 → 切分 chunk → 生成 embedding → 写入向量库。

切分我用的是递归字符切分，优先按段落切，段落太长再按句子切。生成 embedding 时批量调用，一次传 100 条，比一条条调快几十倍。写入时用事务，要么全成功要么全回滚，避免索引半残。

def build_index(docs, client, store, batch_size=100): chunks = [] for doc in docs: chunks.extend(split_text(doc, chunk_size=400, overlap=80)) for i in range(0, len(chunks), batch_size): batch = chunks[i:i + batch_size] embeddings = client.embed(batch) store.add(batch, embeddings)

这段代码里overlap=80就是前面说的 20% 重叠。批量大小 100 是实测下来延迟和吞吐的平衡点，再大容易超时。

4.3 服务启动与压测

服务写完后，先本地跑通，再用工具压测。压测重点看三个指标：QPS、P95 延迟、错误率。我一般用 locust 或 wrk，模拟 50 并发持续 5 分钟。

第一次压测大概率会发现问题。我遇到最多的是连接池不够，模型调用排队。解决办法是把 HTTP 客户端的连接池调大，或者加并发 worker。另一个常见问题是内存泄漏，跑久了 OOM，多半是缓存没设上限，加个 LRU 策略就好。

压测时把模型调用 mock 掉，先测框架本身能扛多少。框架没问题了，再接入真实模型测端到端。这样能快速定位瓶颈在框架还是在模型。

4.4 上线前的检查清单

上线前我会过一遍这个清单，缺一不可：

• 所有密钥从环境变量读取，代码里搜不到明文
• 超时、重试、降级都配好了
• 日志里不打印用户敏感信息
• 有基本的监控指标（请求量、延迟、错误率、token 消耗）
• 数据库连接池大小合理
• 有回滚方案，出问题能快速切回旧版本

这份清单看着琐碎，但每一条背后都是真实事故。比如日志打印敏感信息，一旦日志被泄露就是大问题；没有监控指标，出故障时两眼一抹黑，只能靠猜。

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

5.1 输出不稳定，同样问题答案不一样

这是 AI 服务的常态，不是 bug。但如果差异大到影响业务，就要处理。手段有三个：降低 temperature（设 0 到 0.3）、固定随机种子（部分厂商支持）、在 prompt 里加约束（比如“只输出 JSON，不要解释”）。

temperature 设 0 也不是完全确定，因为底层并行计算有浮点误差。但对绝大多数业务场景，0 已经足够稳定了。创意类任务才需要调高，问答类任务一律调低。

5.2 延迟忽高忽低

先分段计时，看时间花在哪。我一般加三个计时点：预处理耗时、检索耗时、模型耗时。如果模型耗时波动大，那是厂商侧的问题，你能做的是加超时和降级。如果检索耗时波动大，多半是索引没建好，检查向量索引类型和参数。

pgvector 的索引，数据量小的时候用 IVFFlat，大了用 HNSW。HNSW 查询快但建索引慢、占内存。参数m和ef_construction影响精度和速度，默认值一般够用，调优要谨慎，别为了快牺牲召回。

5.3 成本失控

token 消耗是隐形杀手。我见过一个服务，因为 prompt 里塞了太多无关上下文，每月成本是预期的五倍。控制成本的手段：精简 prompt（去掉冗余说明）、缓存常见问题（相同问题直接返回缓存）、限制输出长度（设 max_tokens）。

缓存这块要小心，相同问题不同用户可能因为权限不同答案不同，缓存 key 要带上用户权限标识。我一般用 query 的哈希加权限等级做 key，简单有效。

5.4 常见问题速查表

这张表是我从多次故障里总结的，基本覆盖了八成常见问题。遇到新问题先往这几个方向靠，能省不少排查时间。

6. 工程化进阶：让系统能长期维护

6.1 评测体系：没有评测就没有优化

AI 服务最怕的是“感觉变差了”但说不清哪里差。所以要建评测集。做法是收集一批真实问题，人工标注标准答案，每次改动后跑一遍，看准确率变化。

评测集不用大，100 到 200 条就够，但要覆盖主要场景。我一般分三类：常见问题、边界问题、对抗问题（故意刁难的）。每次模型升级或 prompt 调整，跑一遍评测，用数据说话，别靠感觉。

评测指标除了准确率，还要看引用准确率（RAG 场景）和拒答率（该拒答的有没有拒）。拒答很重要，模型不懂装懂比直接说不知道危害大得多。

6.2 监控与告警：把问题扼杀在萌芽

监控指标分四类：流量（QPS、并发数）、延迟（P50、P95、P99）、错误（错误率、错误类型分布）、成本（token 消耗、调用次数）。这四类指标用 Prometheus 采集，Grafana 展示。

告警阈值怎么定？错误率超过 5% 告警，P95 延迟超过基线 2 倍告警，成本日环比增长超过 50% 告警。阈值别设太敏感，否则天天告警就麻木了。我吃过亏，一开始阈值设太严，半夜被叫醒好几次，后来发现都是正常波动。

6.3 版本管理与灰度发布

模型和 prompt 都要版本化。prompt 改动影响很大，必须能回滚。我的做法是把 prompt 存在配置里，带版本号，每次改动记录变更原因。发布时先灰度 10% 流量，观察指标正常再全量。

灰度期间重点看错误率和用户反馈。AI 服务的质量问题有时指标看不出来，得靠用户反馈。所以灰度期要留足时间，别急着全量。

7. 我踩过的几个真实坑

第一个坑是过度依赖单一模型。早期我所有功能都调一家模型，结果对方一次大规模故障，我整个服务瘫了半天。后来加了备用模型和降级逻辑，再没出现过全站不可用。

第二个坑是prompt 硬编码在代码里。改一次 prompt 要发一次版，效率极低。后来抽到配置文件，改完热加载，效率提升明显。这个改动不大，但收益很高，强烈建议一开始就这么做。

第三个坑是忽略 token 计费细节。有些厂商输入和输出计费不同，有些对缓存命中打折。不了解这些，成本估算会差很多。我现在的做法是每次调用都记录 token 数，按厂商计费规则算成本，月底对账，心里有数。

第四个坑是测试环境用真实模型。测试时频繁调用，成本高还慢。后来测试环境统一用 mock，只在集成测试时调真实模型，速度和成本都降下来了。

8. 后续可以怎么扩展

这套骨架搭起来后，扩展方向很多。想做多模态，就在预处理和模型层加图像、音频的处理分支。想做 Agent，就在服务层加工具调用和规划逻辑。想做私有化部署，就把模型适配层换成自托管模型的接口，其余不动。

我个人觉得，from scratch 搭一遍最大的价值，不是学会了某个框架，而是把整条链路的每个环节都摸了一遍。知道哪里会出问题，知道每个参数为什么这么设，这种手感是看多少教程都换不来的。后面再用什么高级框架，你都能一眼看出它在哪个环节做了封装、可能引入什么新问题。

最后分享一个小技巧：搭完之后，故意把某个环节弄挂，看系统怎么反应。比如把模型接口地址改错，看降级有没有生效；把向量库停掉，看错误处理对不对。这种“故障演练”比正常测试更能暴露问题，我每次上线前都会做一轮。