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 服务的选型建议附带选择理由。环节推荐方案备选选择理由Web 框架FastAPIFlask原生异步、自动生成文档、Pydantic 校验省心模型调用统一适配层直连 SDK隔离厂商变化方便做重试和降级向量库pgvectorFAISS / Milvus数据量百万级以内复用现有 PG 最省运维任务队列Celery / RQ自己写线程池长任务异步化避免阻塞请求缓存Redis内存字典跨进程共享支持过期策略监控Prometheus Grafana日志分析指标化延迟和成本都能量化选 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, timeout30) 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_size100): chunks [] for doc in docs: chunks.extend(split_text(doc, chunk_size400, overlap80)) for i in range(0, len(chunks), batch_size): batch chunks[i:i batch_size] embeddings client.embed(batch) store.add(batch, embeddings)这段代码里overlap80就是前面说的 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 常见问题速查表现象可能原因排查方向解决手段输出乱码编码问题检查请求和响应编码统一 UTF-8检索不到相关内容chunk 切分不当检查切分边界调整 chunk size 和 overlap模型答非所问prompt 不清晰检查 prompt 模板加约束和示例服务偶发 502上游超时看上游延迟指标加超时和重试内存持续增长缓存无上限看内存曲线加 LRU 和过期策略并发上不去连接池太小看连接等待时间调大连接池这张表是我从多次故障里总结的基本覆盖了八成常见问题。遇到新问题先往这几个方向靠能省不少排查时间。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 搭一遍最大的价值不是学会了某个框架而是把整条链路的每个环节都摸了一遍。知道哪里会出问题知道每个参数为什么这么设这种手感是看多少教程都换不来的。后面再用什么高级框架你都能一眼看出它在哪个环节做了封装、可能引入什么新问题。最后分享一个小技巧搭完之后故意把某个环节弄挂看系统怎么反应。比如把模型接口地址改错看降级有没有生效把向量库停掉看错误处理对不对。这种“故障演练”比正常测试更能暴露问题我每次上线前都会做一轮。
