1. 从零搭建AI工程能力:为什么我劝你别一上来就调包
这两年AI应用层的工具链成熟得吓人,LangChain、LlamaIndex、各种Agent框架轮番上阵,好像谁都能在一下午拼出一个"智能问答机器人"。但我带过不少新人,也看过很多团队的项目,一个很普遍的现象是:能跑通Demo的人一大把,能扛住线上流量、能把成本压下来、能定位诡异Bug的人少得可怜。问题出在哪?出在大家把"调包"当成了"工程",把"能跑"当成了"能用"。
ai-engineering-from-scratch这个标题,我理解的核心不是"从零训练一个大模型"——那玩意儿不是个人能玩的,动辄几千万预算。它真正指向的是:从零构建一套AI工程能力体系,包括数据管道、推理服务、提示词管理、评测闭环、成本控制、可观测性这些真正决定项目生死的东西。说白了,模型是别人的,但工程是你自己的,而工程能力才是拉开差距的地方。
这篇文章适合三类人看:一是刚转行做AI应用、只会调API的开发者;二是带团队做AI产品、被线上问题折磨过的技术负责人;三是对AI工程感兴趣、想系统了解全貌的学生或爱好者。我会把整套思路拆开,讲清楚每一步为什么这么做、坑在哪里、怎么落地。全文基于我自己的实践和踩坑经验,不是教科书搬运。
2. 整体设计思路:先想清楚"工程"到底管什么
2.1 为什么"从零"反而比"用框架"更快
很多人有个误区,觉得从零就是重复造轮子,浪费时间。我的观点恰恰相反:在AI工程这个领域,从零搭一遍最小可用系统,是理解框架设计取舍的最快路径。你只有自己写过一次带重试、带超时、带降级的推理调用,才会明白为什么有些框架要引入"链"的概念;你只有自己处理过一次上下文超长被截断的问题,才会理解为什么要有记忆管理和摘要压缩。
框架是别人对通用问题的抽象,但你的业务有大量非通用约束。比如你的场景要求响应时间必须控制在800毫秒以内,那框架默认的"多轮反思"策略就直接废了。从零搭一遍,你会对每个环节的耗时、成本、失败率有肌肉记忆,后面用框架时才知道哪里该改、哪里该绕。
具体来说,我建议的最小系统包含五个模块:输入预处理、提示词组装、模型调用、输出解析、结果落库。这五个模块串起来就是一个完整的请求生命周期,任何AI应用本质上都是这个循环的变体。
2.2 技术选型:别被"最新"绑架
选型这块我踩过最大的坑就是追新。去年有个项目,我选了一个当时很火的向量数据库,结果它的Python SDK在并发场景下有内存泄漏,查了三天才定位到。后来我总结了一条原则:核心链路上的组件,优先选成熟稳定、社区活跃、有商业支持的;边缘实验性的部分,随便折腾。
具体到几个关键选型:
| 组件类型 | 保守选择 | 激进选择 | 我的建议 |
|---|---|---|---|
| 模型调用 | 官方SDK | 第三方聚合网关 | 官方SDK,可控性最强 |
| 向量存储 | PostgreSQL + pgvector | 专用向量数据库 | 数据量小于千万级用pgvector |
| 任务队列 | Redis + RQ | Kafka | 中小项目Redis足够 |
| 可观测 | OpenTelemetry + 自建 | 商业APM | 先自建,规模上来再换 |
这张表不是说激进选择不好,而是说在你还没搞清楚自己的瓶颈在哪之前,保守选择能让你少花时间在排查基础设施问题上。等你的QPS上来了、数据量上来了,再针对性替换,那时候你也有足够的监控数据支撑决策。
2.3 成本意识要贯穿设计始终
AI工程和传统后端工程最大的区别之一,就是每一次请求都是真金白银。传统接口调一百万次可能就几毛钱电费,AI接口调一百万次可能是几万块。所以成本控制不是优化项,是设计约束。
我在设计阶段会强制问三个问题:这个请求必须调用大模型吗?能不能用小模型?能不能缓存?这三个问题能砍掉至少一半的无效调用。举个例子,用户问"你们几点上班",这种固定问答完全可以用规则匹配或者小模型分类,没必要走大模型。我见过一个客服系统,80%的请求都是这类简单问题,全走大模型,一个月烧掉六位数,后来加了意图分类前置,成本直接降到五分之一。
3. 核心模块拆解:每个环节的实操要点
3.1 输入预处理:脏数据是第一杀手
线上环境和实验室环境最大的差别就是输入质量。实验室里你测试的都是"请帮我总结这篇文章",线上用户可能发来一坨乱码、一个超长URL、或者干脆是空的。输入预处理的核心任务就三件事:清洗、校验、截断。
清洗包括去除多余空白、统一编码、过滤特殊字符。这里有个细节:不要过度清洗,有些特殊字符是有意义的,比如代码场景里的缩进和符号。我的做法是维护一个白名单,只保留业务需要的字符集。
校验主要是长度和类型。长度校验要注意,不同模型的上下文窗口不一样,而且计费是按token算的,不是按字符。中文大概1个token对应1.5到2个汉字,英文大概1个token对应4个字符。我一般会在预处理阶段就估算token数,超过阈值直接走截断或拒绝,避免把超长请求发给模型再被拒,浪费一次网络往返。
截断策略也有讲究。简单粗暴地从尾部截断会丢失关键信息,我常用的是"首尾保留+中间摘要":保留开头200token和结尾200token,中间部分如果超长就用小模型压缩成摘要。这个策略在长文档问答场景下效果很好,实测比单纯截断的准确率高不少。
注意:预处理阶段一定要记录原始输入和清洗后输入的哈希值,方便后续排查"为什么同样的输入结果不一样"这类问题。
3.2 提示词组装:模板化是唯一出路
提示词散落在代码各处是维护灾难。我见过一个项目,提示词硬编码在十几个文件里,改一个措辞要全局搜索替换,还经常漏改。正确做法是把提示词当配置管理,模板化、版本化、可回滚。
我的做法是用简单的模板引擎,比如Python的Jinja2,把提示词拆成"系统角色+任务描述+上下文+用户输入+输出格式"几个部分。每个部分独立维护,组装时按需拼接。这样做的好处是,调整输出格式不用动角色设定,换角色不用重写任务描述。
版本管理这块,我建议每次提示词变更都记录:变更人、变更时间、变更原因、变更前后的效果对比。听起来很繁琐,但当你发现效果突然下降、需要回滚时,这份记录能救命。我一般用数据库表存提示词版本,配合一个简单的管理界面,非技术人员也能改。
还有一个容易被忽略的点:提示词里的变量要做转义。用户输入如果包含模板语法字符,直接拼接会导致模板渲染错误甚至注入。我吃过这个亏,用户输入里带了个花括号,整个提示词渲染崩了,请求直接500。
3.3 模型调用:重试、超时、降级一个都不能少
模型调用是整条链路最不稳定的一环。网络抖动、服务限流、模型过载,各种问题都会导致调用失败。没有重试和降级的调用代码,都是玩具代码。
重试策略我一般用指数退避,初始间隔1秒,最大重试3次,每次间隔翻倍。但要注意,不是所有错误都值得重试。像参数错误、内容违规这类,重试多少次都一样,直接失败返回。只有超时、限流、5xx这类才重试。判断逻辑要写清楚,不然会浪费大量无效重试。
超时设置要分层。连接超时短一点,比如3秒;读取超时长一点,因为大模型生成确实慢,我一般设30到60秒,具体看业务对延迟的容忍度。这里有个坑:超时时间要小于上游网关的超时时间,不然你的服务还在等模型返回,网关已经把连接断了,用户看到的是网关的错误页,你的日志里却什么都没有。
降级策略是保命用的。当主模型不可用时,切到备用模型;当所有模型都不可用时,返回兜底话术。兜底话术不要写"服务暂时不可用"这种冷冰冰的,我一般会写"当前咨询人数较多,请稍后再试,或联系人工客服",用户体验好很多。
def call_model_with_fallback(prompt, max_retries=3): for attempt in range(max_retries): try: return primary_model.call(prompt, timeout=30) except RateLimitError: time.sleep(2 ** attempt) except TimeoutError: continue except InvalidRequestError: raise # 参数错误不重试 try: return backup_model.call(prompt, timeout=30) except Exception: return {"text": "当前咨询人数较多,请稍后再试"}3.4 输出解析:别信模型会乖乖听话
你让模型输出JSON,它可能给你输出带markdown代码块的JSON,可能给你输出解释性文字加JSON,甚至可能输出一个语法错误的JSON。输出解析必须做容错,不能假设模型100%遵守格式。
我的解析流程是:先尝试直接解析,失败则用正则提取JSON片段,再失败则调用一次修复提示词让模型自己修,最后还失败就记录原始输出并返回错误。这个流程听起来复杂,但实际跑下来,90%以上的请求第一次就能解析成功,剩下的大部分在正则提取阶段就解决了。
对于结构化输出,我现在更倾向于用模型提供的结构化输出功能,比如有些模型支持JSON mode或者function calling。这些功能底层做了约束,比纯提示词可靠得多。但要注意,结构化输出会增加延迟,因为模型生成时要做额外的约束检查,对延迟敏感的场景要权衡。
解析后的数据还要做业务校验。比如模型返回了一个日期字段,你要校验它是不是合法日期;返回了一个金额,你要校验它是不是正数。这些校验能拦住很多模型幻觉导致的问题。
4. 实操全流程:从零搭一个可用的问答服务
4.1 环境准备与依赖安装
我假设你用Python,这是AI工程最主流的语言。环境管理我强烈建议用uv或者poetry,别用裸pip,依赖冲突会让你怀疑人生。下面是我常用的依赖清单:
# 用uv创建虚拟环境 uv venv source .venv/bin/activate # 核心依赖 uv pip install fastapi uvicorn httpx pydantic uv pip install jinja2 redis sqlalchemy psycopg2-binary uv pip install pgvector openai tiktoken这里解释几个关键依赖的选择理由。httpx而不是requests,因为httpx原生支持异步,在并发调用模型时性能好很多。pydantic用来做数据校验和序列化,配合FastAPI能自动生成API文档。tiktoken用来精确计算token数,比估算准确得多,成本控制全靠它。
数据库我选PostgreSQL加pgvector扩展。pgvector让Postgres具备了向量检索能力,对于中小规模应用完全够用,而且省去了维护两套存储的麻烦。数据量特别大再考虑专用向量库。
4.2 数据库表结构设计
表结构设计要提前想清楚,后面改起来很痛苦。我一般会建这几张核心表:
-- 会话表 CREATE TABLE conversations ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id VARCHAR(64) NOT NULL, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- 消息表 CREATE TABLE messages ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), conversation_id UUID REFERENCES conversations(id), role VARCHAR(16) NOT NULL, content TEXT NOT NULL, token_count INT, model_name VARCHAR(64), latency_ms INT, created_at TIMESTAMP DEFAULT NOW() ); -- 提示词版本表 CREATE TABLE prompt_versions ( id SERIAL PRIMARY KEY, name VARCHAR(64) NOT NULL, version INT NOT NULL, template TEXT NOT NULL, is_active BOOLEAN DEFAULT FALSE, created_at TIMESTAMP DEFAULT NOW() );消息表里我特意加了token_count、model_name、latency_ms三个字段。这三个字段是后续做成本分析和性能优化的基础。没有这些数据,你根本不知道钱花在哪了、慢在哪了。我见过太多项目上线后才发现没记录这些,想优化都无从下手。
4.3 核心请求处理流程
整个请求处理流程我拆成七步,每一步都有明确的输入输出和错误处理:
- 接收请求:FastAPI接收HTTP请求,Pydantic做参数校验,不合法直接返回400。
- 加载会话:根据会话ID从数据库加载历史消息,没有则创建新会话。
- 输入预处理:清洗用户输入,计算token数,超长则截断或摘要。
- 组装提示词:从数据库加载当前激活的提示词模板,填充变量。
- 调用模型:带重试和降级的模型调用,记录耗时和token消耗。
- 解析输出:容错解析模型输出,业务校验。
- 落库返回:保存消息记录,更新会话时间,返回响应。
这个流程里,第3步和第6步是最容易出问题的。第3步的token计算如果不准,要么浪费钱要么被模型拒绝;第6步的解析如果不健壮,线上会频繁报错。我建议这两步都写单元测试,覆盖各种边界情况。
4.4 参数计算:token预算怎么定
token预算的计算是成本控制的核心。假设你的模型上下文窗口是8K token,你要给输出留多少?我的经验是输出预留至少1K token,系统提示词控制在500 token以内,剩下的才是历史消息和用户输入的空间。
具体算一下:8K窗口 - 1K输出 - 0.5K系统提示 = 6.5K给历史消息和用户输入。如果用户输入平均200 token,历史消息平均每轮300 token,那最多能保留20轮历史。超过20轮就要做摘要压缩,把早期对话压缩成一段摘要。
这个计算不是拍脑袋,要基于实际数据调整。上线后统计一下用户输入的平均token数和历史消息的平均长度,再反推合理的窗口分配。我一般会留20%的余量,因为token计算本身有误差,而且不同语言的token密度不一样。
提示:中文的token密度比英文高,同样长度的文本,中文消耗的token更多。做预算时如果用户以中文为主,要按更保守的估算。
5. 常见问题与排查技巧实录
5.1 模型返回内容被截断怎么办
这是最常见的问题之一。表现是模型输出到一半突然停了,没有结束标记。原因通常是达到了max_tokens限制。排查步骤:先看请求里的max_tokens设置是多少,再看实际输出的token数是不是接近这个值。
解决方法有两个:一是调大max_tokens,但要注意不能超过模型上下文窗口减去输入token数;二是优化提示词,让模型输出更简洁。我一般会先调大限制,观察一段时间,如果经常触顶,再考虑优化提示词或者分段生成。
分段生成是个进阶技巧:让模型先输出大纲,再逐段展开。这样每段的输出都不会太长,避免触顶。但代价是调用次数增加,成本和延迟都上去了,要权衡。
5.2 相同输入结果不一致怎么排查
大模型有随机性,相同输入结果不一致是正常的。但如果差异特别大,就要排查了。首先确认temperature参数,如果大于0,结果本来就会波动。做需要确定性的场景,把temperature设为0。
即使temperature为0,有些模型因为底层并行计算的原因,结果也可能有微小差异。如果业务要求完全一致,那只能加缓存:把输入哈希作为key,输出作为value,相同输入直接返回缓存结果。缓存还能省钱,一举两得。
排查这类问题时,一定要记录完整的请求参数,包括模型版本、temperature、top_p等。我遇到过模型版本悄悄升级导致结果变化的情况,没有记录版本号的话,根本查不出来。
5.3 成本突然飙升的排查思路
成本飙升一般有三个原因:调用量增加、单次token增加、模型切换。排查顺序是:先看调用量曲线,再看平均token数曲线,最后看模型分布。
调用量增加可能是被刷了,要检查是否有异常IP或用户。单次token增加可能是提示词变长了,或者用户输入变长了。模型切换可能是降级逻辑被触发,备用模型比主模型贵。
我一般会做一个成本看板,按天、按用户、按接口维度展示成本。这样异常一出现就能定位到具体维度。没有看板的项目,等月底账单出来才发现,那就晚了。
| 问题现象 | 可能原因 | 排查方法 | 解决措施 |
|---|---|---|---|
| 成本突增 | 调用量增加 | 看调用量曲线 | 限流、封禁异常用户 |
| 成本突增 | 单次token增加 | 看平均token曲线 | 优化提示词、截断输入 |
| 成本突增 | 模型切换 | 看模型分布 | 检查降级逻辑 |
| 响应变慢 | 模型过载 | 看模型延迟指标 | 切换模型、增加超时 |
| 解析失败 | 输出格式变化 | 看原始输出日志 | 更新解析逻辑、加容错 |
5.4 几个我踩过的坑
第一个坑:没做幂等。用户网络不好重复提交,导致同一个问题被处理两次,扣了两次费。后来加了请求ID去重,同一个ID的请求只处理一次。
第二个坑:日志里记了敏感信息。用户输入里可能有手机号、身份证号,直接记日志有合规风险。后来加了脱敏处理,敏感字段用掩码替换。
第三个坑:没做限流。上线第一天被爬虫刷了,一晚上烧掉半个月预算。后来加了基于用户和IP的双重限流,才稳住。
第四个坑:提示词里的示例过时了。模型升级后,原来的few-shot示例反而误导了模型,效果下降。后来把示例也纳入版本管理,模型升级时同步检查示例。
6. 评测闭环:没有评测的AI工程都是盲人摸象
6.1 为什么评测比开发还重要
AI应用有个特点:你改一个提示词,可能修好了A场景,却弄坏了B场景。没有评测,你根本不知道这次改动是净收益还是净损失。我见过团队靠感觉调提示词,调了两个月,效果还不如最初版本。
评测的核心是建立一套可重复、可量化、可对比的测试集。测试集要覆盖典型场景、边界场景、对抗场景。典型场景就是正常用户会问的问题,边界场景是超长、超短、多语言的输入,对抗场景是故意诱导模型出错的输入。
测试集的规模不用很大,初期50到100条就够,但每条都要有明确的预期输出或评分标准。评分可以用人工标注,也可以用模型自动评分,但自动评分本身也要校准,不能全信。
6.2 评测指标怎么定
不同场景指标不一样。问答场景看准确率和召回率,生成场景看流畅度和相关性,分类场景看F1值。我一般会定一个主指标加几个辅助指标,主指标决定是否上线,辅助指标用来诊断问题。
主指标我推荐用人工评分,虽然贵但准。自动指标比如BLEU、ROUGE这些,和人类判断的相关性其实不高,只能做参考。我一般用自动指标做日常监控,用人工评分做版本决策。
评测频率上,每次提示词变更、模型切换、代码改动都要跑一遍全量评测。日常可以跑抽样评测,比如每天抽100条线上请求做自动评分,监控效果趋势。
6.3 评测结果怎么用
评测结果不是用来打分的,是用来做决策的。我一般会看三个东西:整体指标变化、分场景指标变化、失败案例。整体指标涨了但某个场景跌了,说明改动有偏科,要针对性优化。失败案例是最有价值的,能直接告诉你模型在哪里不行。
我习惯把失败案例分类归档,比如"格式错误"、"事实错误"、"拒答"、"答非所问"。每类问题的解决思路不一样,格式错误改提示词,事实错误加检索,拒答调安全策略,答非所问改任务描述。分类之后,优化方向就清晰了。
7. 可观测性:线上问题定位的命根子
7.1 必须记录的日志字段
AI应用的日志比传统应用复杂,因为多了模型相关的维度。我必记的字段包括:请求ID、用户ID、会话ID、模型名称、模型版本、输入token数、输出token数、总token数、延迟、重试次数、是否降级、解析是否成功、错误类型。
这些字段看起来多,但每一个在排查问题时都可能用到。比如"是否降级"能告诉你备用模型的使用比例,"解析是否成功"能告诉你输出格式的稳定性。没有这些字段,排查问题就像在黑箱里摸。
日志的存储我建议用结构化日志,JSON格式,方便后续查询和分析。别用纯文本日志,grep起来太痛苦。如果量特别大,可以考虑采样,但错误日志必须全量记录。
7.2 关键监控指标
监控指标分四类:流量、延迟、错误、成本。流量看QPS和并发数,延迟看P50、P95、P99,错误看错误率和错误类型分布,成本看每小时消耗和单次平均成本。
延迟这块要特别注意P99,因为大模型的延迟分布是长尾的,P50可能只有1秒,P99可能到10秒。如果只看平均值,会低估用户的真实体验。我一般会设P99的告警阈值,超过就排查。
成本监控要设日预算和小时预算,超了告警。我见过项目因为没设预算告警,一个月超支好几倍才发现。成本告警阈值可以设得宽松一点,比如预算的80%告警,给自己留处理时间。
7.3 链路追踪怎么做
AI应用的链路比传统应用长,一次请求可能涉及预处理、检索、模型调用、后处理多个环节。没有链路追踪,你根本不知道时间花在哪了。我一般用OpenTelemetry做埋点,每个环节一个span,记录开始时间、结束时间、关键参数。
链路追踪的价值在排查慢请求时特别明显。一个请求总耗时5秒,追踪一看,预处理0.1秒,检索0.5秒,模型调用4秒,后处理0.4秒,那瓶颈就在模型调用。是模型本身慢,还是网络慢,再看模型调用的子span就知道了。
8. 写在最后:一些个人体会
做AI工程这两年,我最大的感受是:这行变化太快,但底层的东西变化很慢。模型半年换一代,但重试、降级、缓存、评测、监控这些工程手段,和十年前做后端时没什么本质区别。所以别焦虑追不上新模型,把工程基本功打扎实,换什么模型你都能快速接上。
另一个体会是:别追求一步到位。我见过太多项目想一开始就搭一套完美的架构,结果三个月没上线,需求早变了。正确的做法是先搭最小可用版本,跑通闭环,然后根据实际瓶颈逐步优化。先让它跑起来,再让它跑得稳,最后让它跑得省。
最后分享一个小技巧:给你的AI应用建一个"事故本",每次线上出问题,记录现象、原因、解决过程、后续改进。这个本子比任何文档都有价值,因为它是你真实踩过的坑。我现在的团队,新人入职第一件事就是读事故本,比读代码快多了。
这套东西后续还可以扩展的方向很多,比如多模型路由、A/B测试框架、自动化评测流水线。但那是下一步的事,先把基础闭环跑通,比什么都重要。