1. 从零搭建AI工程体系为什么我劝你别急着调包这两年AI应用开发的门槛被各种框架拉得极低三行代码调用一个大模型接口再套个前端模板一个“智能助手”就上线了。但我见过太多团队在Demo阶段跑得飞快一进入真实业务场景就全面崩盘响应延迟从2秒飙到20秒、并发一上来就报错、上下文管理混乱导致答非所问、成本失控到月底账单不敢看。这些问题的根源几乎都指向同一个事实——跳过工程底座直接堆功能。ai-engineering-from-scratch这个项目标题核心讲的不是“怎么调用模型”而是从零构建一套能扛住真实流量的AI工程体系。它涉及的核心领域包括模型服务化部署、推理性能优化、上下文与记忆管理、请求编排与降级策略、可观测性建设、成本控制。适合谁看如果你已经能跑通基础的模型调用但一遇到并发、延迟、成本、稳定性问题就手足无措那这篇内容就是为你准备的。我会把每个环节的“为什么这么设计”讲透把参数计算过程摊开把踩过的坑标出来让你能直接抄作业。2. 整体架构设计与技术选型思路2.1 为什么不能“一个接口打天下”新手最常见的做法是写一个HTTP接口收到请求后直接同步调用模型API拿到结果返回。这个模式在单人测试时没问题但一旦并发超过个位数问题就集中爆发。模型推理本身是计算密集型任务单次调用耗时可能在几百毫秒到几秒不等同步阻塞意味着你的服务线程会被长时间占用连接池迅速耗尽后续请求全部排队超时。正确的思路是把“接收请求”和“执行推理”解耦。我采用的方案是接入层只负责鉴权、限流、参数校验和任务入队真正的推理任务交给独立的Worker池异步执行结果通过轮询或回调返回。这样做的好处是接入层可以轻松扛住高并发Worker池可以根据GPU/CPU资源灵活扩缩容两边互不拖累。具体选型上消息队列我用的是Redis Stream而非Kafka——在中小规模场景下Redis Stream的部署运维成本低得多而且延迟表现更好。Worker池用Python的asyncio配合信号量控制并发数避免一次性把太多请求压给模型服务。这个组合实测在单机8核16G的配置下能稳定支撑每秒50到80个推理请求取决于模型大小和生成长度。2.2 模型服务化的三种路径对比模型怎么“跑起来”直接决定了后续所有工程决策。我整理了三種常见路径的对比路径适用场景优点缺点直接调用云端API快速验证、低频调用零运维、按量付费延迟不可控、数据出境风险、成本随量线性增长本地部署开源模型数据敏感、高频调用数据不出域、边际成本低需要GPU资源、运维复杂、模型能力有上限混合路由多场景并存兼顾成本与效果路由逻辑复杂、需要统一抽象层我的建议是从混合路由起步。简单任务如意图分类、文本清洗走本地小模型复杂任务如长文生成、逻辑推理走云端大模型。这样既能控制成本又不会牺牲核心体验。关键在于抽象出一个统一的ModelProvider接口上层业务不感知底层用的是哪个模型切换时只改配置不改代码。2.3 上下文管理的核心设计多轮对话是AI应用的标配但上下文管理是最容易埋雷的地方。我见过最离谱的案例是把整段对话历史无脑拼接到prompt里结果token数爆炸不仅成本飙升模型还会因为上下文过长而“遗忘”关键信息。我的做法是分层管理上下文第一层是“系统指令”固定不变定义模型角色和输出格式第二层是“摘要记忆”把超过N轮的历史对话压缩成一段摘要由模型自己生成第三层是“近期原文”保留最近3到5轮的完整对话。这样既保留了关键信息又把token数控制在合理范围。摘要的触发阈值我设的是累计token超过2000压缩后控制在500token以内实测效果和成本平衡得比较好。3. 核心细节解析与实操要点3.1 推理性能优化的四个抓手推理延迟是用户体验的生命线。我把优化手段分成四个层面按投入产出比排序第一批处理Batching。这是提升吞吐量最有效的手段。把多个请求打包成一个batch送给模型GPU利用率能从30%拉到80%以上。但要注意batch size不是越大越好太大会导致单个请求的等待时间变长。我的经验值是在线场景batch size控制在8到16离线场景可以放到32甚至64。实现上可以用一个定时器每50毫秒或凑够batch size就触发一次推理。第二量化Quantization。把模型权重从FP16降到INT8甚至INT4显存占用能减少一半以上推理速度提升30%到50%。代价是精度会有轻微下降但在大多数业务场景下感知不明显。我用的是GPTQ量化方案4bit量化后模型效果保留约97%显存从14G降到6G性价比极高。第三KV Cache复用。多轮对话中系统指令和摘要记忆这部分前缀是固定的可以把它们的KV Cache缓存起来后续轮次直接复用避免重复计算。这个优化在长对话场景下能减少40%以上的计算量。第四投机采样Speculative Decoding。用一个小模型先“草拟”多个token再用大模型一次性验证能显著加速生成过程。实测在代码生成场景下能提速2倍左右但需要额外部署一个小模型适合对延迟极度敏感的场景。3.2 请求编排与降级策略真实业务中模型服务不可能100%可用。云端API会限流、会超时本地模型会OOM、会崩溃。如果没有降级策略一次抖动就是一次线上事故。我的编排逻辑是这样的每个请求进来后先走超时控制——设置一个总超时时间比如10秒超过就返回兜底话术。然后是重试策略——对于网络类错误重试2次每次间隔指数退避对于模型类错误如内容审核不通过不重试直接走降级。最后是降级链路——主模型不可用时自动切换到备用模型备用模型也不可用时返回预设的静态回复保证服务不中断。这里有个关键细节降级要有感知。每次降级都要打点上报运维人员能第一时间知道主链路出了问题。我见过有的系统降级了半个月都没人发现用户体验一直在打折。3.3 可观测性建设的最小闭环没有可观测性的AI系统就是黑盒。我建议至少采集以下指标请求维度QPS、P50/P95/P99延迟、错误率、降级率模型维度首token延迟、生成速度token/s、输入输出token数成本维度每请求成本、每日累计成本、按业务线拆分质量维度用户点赞/点踩率、人工抽检评分这些指标用Prometheus采集Grafana做看板再配一套告警规则。比如P99延迟超过5秒持续3分钟就告警错误率超过5%就告警。别小看这套东西它能帮你在用户投诉之前就发现问题。注意token计数一定要在服务端做不能依赖模型API返回的usage字段因为不同厂商的统计口径不一致而且有些流式响应根本不返回usage。4. 实操过程与核心环节实现4.1 环境准备与依赖安装我以Python技术栈为例走一遍完整的搭建流程。基础环境是Ubuntu 22.04Python 3.10CUDA 12.1如果用GPU的话。# 创建虚拟环境 python -m venv ai-eng source ai-eng/bin/activate # 核心依赖 pip install fastapi uvicorn redis asyncpg pip install transformers accelerate bitsandbytes pip install prometheus-client这里解释一下选型理由FastAPI自带异步支持和自动文档适合做接入层Redis既做消息队列又做缓存一物两用bitsandbytes用于量化加载模型prometheus-client用于指标暴露。版本上建议锁定避免自动升级引入不兼容。4.2 接入层与Worker池的代码骨架接入层的核心逻辑是校验请求、生成任务ID、入队、返回任务ID。Worker池从队列消费任务执行推理写回结果。# 接入层 from fastapi import FastAPI import redis.asyncio as redis import uuid app FastAPI() r redis.Redis(hostlocalhost, port6379) app.post(/v1/infer) async def infer(payload: dict): task_id str(uuid.uuid4()) await r.xadd(infer_stream, {task_id: task_id, payload: str(payload)}) return {task_id: task_id, status: queued}Worker池这边我用asyncio.Semaphore控制并发数避免一次性拉太多任务把显存撑爆。# Worker池 import asyncio sem asyncio.Semaphore(4) # 根据显存调整 async def worker(): while True: messages await r.xreadgroup(workers, worker-1, {infer_stream: }, count1, block1000) if not messages: continue async with sem: await process_task(messages)并发数怎么定我的计算方式是显存总量除以单次推理峰值显存再留20%余量。比如24G显存单次推理峰值4G那并发数就是24除以4再乘0.8约等于4。这个值不是固定的要根据实际压测结果微调。4.3 上下文压缩的具体实现上下文压缩是控制成本的关键。我的实现逻辑是每次对话结束后检查累计token数超过阈值就触发压缩。def compress_context(history: list, threshold: int 2000): total_tokens sum(count_tokens(m[content]) for m in history) if total_tokens threshold: return history # 保留最近3轮其余压缩成摘要 recent history[-6:] old history[:-6] summary call_model(f请将以下对话压缩成一段摘要{old}) return [{role: system, content: f历史摘要{summary}}] recent这里有个坑压缩本身也要调用模型会产生额外成本。所以阈值不能设太低否则压缩频率太高反而更贵。我的经验是阈值设在2000到3000token之间压缩后控制在500token以内这样压缩带来的成本远小于节省的上下文成本。4.4 压测与参数调优实录系统搭好后一定要压测。我用的是locust模拟50个并发用户持续请求。第一轮压测结果很惨P99延迟12秒错误率8%。排查后发现两个问题一是Worker并发数设太高设了8导致显存频繁OOM触发重试二是Redis Stream的消费者组没有及时ack消息堆积。调整方案并发数降到4增加OOM自动降级逻辑消费者处理完立即ack。第二轮压测P99降到3.2秒错误率0.5%。第三轮把batch size从1调到8P99进一步降到2.1秒吞吐量翻了3倍。这个调优过程说明一个道理AI工程的性能瓶颈往往不在模型本身而在工程链路的配置。多花时间在压测和调参上比盲目换更大的模型划算得多。5. 常见问题与排查技巧实录5.1 典型问题速查表现象可能原因排查方向解决方案延迟突然飙升显存不足触发swap查看GPU显存和swap使用降低并发数或启用量化错误率上升云端API限流查看API返回码增加重试降级链路成本异常增长上下文未压缩统计平均输入token数启用摘要压缩回答质量下降量化精度损失对比量化前后输出调整量化位数或换方案服务无响应消息队列堆积查看队列长度扩容Worker或限流5.2 三个我踩过的坑第一个坑忽略冷启动。模型第一次加载要几十秒如果服务刚启动就接流量大量请求会超时。我的解决办法是加一个预热接口服务启动后自动跑几条测试请求等模型完全加载后再接入负载均衡。第二个坑日志打太多。为了排查问题我把每次请求的完整prompt和response都打进日志结果磁盘一天就满了而且日志写入本身拖慢了主流程。后来改成只记录token数、延迟、错误码这些结构化字段完整内容只在采样时记录。第三个坑降级话术太生硬。早期降级直接返回“服务繁忙请稍后重试”用户体感很差。后来改成“当前咨询人数较多我先为您记录问题稍后回复”配合异步通知用户满意度明显提升。降级不可怕可怕的是让用户感知到降级。5.3 成本控制的独家技巧成本控制的核心是让每一分钱都花在刀刃上。我的做法是对请求做分级简单请求走小模型复杂请求走大模型。分级逻辑可以用一个轻量分类器实现准确率90%以上就够了分错的代价远小于全量走大模型的成本。另外缓存高频问题的答案。很多用户问的问题是重复的把高频问题的答案缓存起来命中缓存直接返回能省下大量推理成本。缓存key用问题的语义哈希而非字面哈希这样换个说法也能命中。最后再分享一个实操心得定期做成本审计。每周拉一次账单按业务线、按模型、按请求类型拆分找出成本大头。我就是在一次审计中发现某个内部测试接口被外部扫描器疯狂调用一天烧掉了几百块。加个鉴权就解决了。这套从零搭建的AI工程体系我前后迭代了三个版本踩了无数坑才稳定下来。它不是什么高深的技术核心就是把工程领域成熟的模式搬到AI场景里异步解耦、批处理、降级、可观测性、成本控制。这些词听起来不性感但它们是让AI应用从Demo走向生产的关键。你不需要一次全做完可以先从异步解耦和可观测性入手这两块投入产出比最高。等业务量上来了再逐步补全其他环节。
1. 从零搭建AI工程体系为什么我劝你别急着调包这两年AI应用开发的门槛被各种框架拉得极低三行代码调用一个大模型接口再套个前端模板一个“智能助手”就上线了。但我见过太多团队在Demo阶段跑得飞快一进入真实业务场景就全面崩盘响应延迟从2秒飙到20秒、并发一上来就报错、上下文管理混乱导致答非所问、成本失控到月底账单不敢看。这些问题的根源几乎都指向同一个事实——跳过工程底座直接堆功能。ai-engineering-from-scratch这个项目标题核心讲的不是“怎么调用模型”而是从零构建一套能扛住真实流量的AI工程体系。它涉及的核心领域包括模型服务化部署、推理性能优化、上下文与记忆管理、请求编排与降级策略、可观测性建设、成本控制。适合谁看如果你已经能跑通基础的模型调用但一遇到并发、延迟、成本、稳定性问题就手足无措那这篇内容就是为你准备的。我会把每个环节的“为什么这么设计”讲透把参数计算过程摊开把踩过的坑标出来让你能直接抄作业。2. 整体架构设计与技术选型思路2.1 为什么不能“一个接口打天下”新手最常见的做法是写一个HTTP接口收到请求后直接同步调用模型API拿到结果返回。这个模式在单人测试时没问题但一旦并发超过个位数问题就集中爆发。模型推理本身是计算密集型任务单次调用耗时可能在几百毫秒到几秒不等同步阻塞意味着你的服务线程会被长时间占用连接池迅速耗尽后续请求全部排队超时。正确的思路是把“接收请求”和“执行推理”解耦。我采用的方案是接入层只负责鉴权、限流、参数校验和任务入队真正的推理任务交给独立的Worker池异步执行结果通过轮询或回调返回。这样做的好处是接入层可以轻松扛住高并发Worker池可以根据GPU/CPU资源灵活扩缩容两边互不拖累。具体选型上消息队列我用的是Redis Stream而非Kafka——在中小规模场景下Redis Stream的部署运维成本低得多而且延迟表现更好。Worker池用Python的asyncio配合信号量控制并发数避免一次性把太多请求压给模型服务。这个组合实测在单机8核16G的配置下能稳定支撑每秒50到80个推理请求取决于模型大小和生成长度。2.2 模型服务化的三种路径对比模型怎么“跑起来”直接决定了后续所有工程决策。我整理了三種常见路径的对比路径适用场景优点缺点直接调用云端API快速验证、低频调用零运维、按量付费延迟不可控、数据出境风险、成本随量线性增长本地部署开源模型数据敏感、高频调用数据不出域、边际成本低需要GPU资源、运维复杂、模型能力有上限混合路由多场景并存兼顾成本与效果路由逻辑复杂、需要统一抽象层我的建议是从混合路由起步。简单任务如意图分类、文本清洗走本地小模型复杂任务如长文生成、逻辑推理走云端大模型。这样既能控制成本又不会牺牲核心体验。关键在于抽象出一个统一的ModelProvider接口上层业务不感知底层用的是哪个模型切换时只改配置不改代码。2.3 上下文管理的核心设计多轮对话是AI应用的标配但上下文管理是最容易埋雷的地方。我见过最离谱的案例是把整段对话历史无脑拼接到prompt里结果token数爆炸不仅成本飙升模型还会因为上下文过长而“遗忘”关键信息。我的做法是分层管理上下文第一层是“系统指令”固定不变定义模型角色和输出格式第二层是“摘要记忆”把超过N轮的历史对话压缩成一段摘要由模型自己生成第三层是“近期原文”保留最近3到5轮的完整对话。这样既保留了关键信息又把token数控制在合理范围。摘要的触发阈值我设的是累计token超过2000压缩后控制在500token以内实测效果和成本平衡得比较好。3. 核心细节解析与实操要点3.1 推理性能优化的四个抓手推理延迟是用户体验的生命线。我把优化手段分成四个层面按投入产出比排序第一批处理Batching。这是提升吞吐量最有效的手段。把多个请求打包成一个batch送给模型GPU利用率能从30%拉到80%以上。但要注意batch size不是越大越好太大会导致单个请求的等待时间变长。我的经验值是在线场景batch size控制在8到16离线场景可以放到32甚至64。实现上可以用一个定时器每50毫秒或凑够batch size就触发一次推理。第二量化Quantization。把模型权重从FP16降到INT8甚至INT4显存占用能减少一半以上推理速度提升30%到50%。代价是精度会有轻微下降但在大多数业务场景下感知不明显。我用的是GPTQ量化方案4bit量化后模型效果保留约97%显存从14G降到6G性价比极高。第三KV Cache复用。多轮对话中系统指令和摘要记忆这部分前缀是固定的可以把它们的KV Cache缓存起来后续轮次直接复用避免重复计算。这个优化在长对话场景下能减少40%以上的计算量。第四投机采样Speculative Decoding。用一个小模型先“草拟”多个token再用大模型一次性验证能显著加速生成过程。实测在代码生成场景下能提速2倍左右但需要额外部署一个小模型适合对延迟极度敏感的场景。3.2 请求编排与降级策略真实业务中模型服务不可能100%可用。云端API会限流、会超时本地模型会OOM、会崩溃。如果没有降级策略一次抖动就是一次线上事故。我的编排逻辑是这样的每个请求进来后先走超时控制——设置一个总超时时间比如10秒超过就返回兜底话术。然后是重试策略——对于网络类错误重试2次每次间隔指数退避对于模型类错误如内容审核不通过不重试直接走降级。最后是降级链路——主模型不可用时自动切换到备用模型备用模型也不可用时返回预设的静态回复保证服务不中断。这里有个关键细节降级要有感知。每次降级都要打点上报运维人员能第一时间知道主链路出了问题。我见过有的系统降级了半个月都没人发现用户体验一直在打折。3.3 可观测性建设的最小闭环没有可观测性的AI系统就是黑盒。我建议至少采集以下指标请求维度QPS、P50/P95/P99延迟、错误率、降级率模型维度首token延迟、生成速度token/s、输入输出token数成本维度每请求成本、每日累计成本、按业务线拆分质量维度用户点赞/点踩率、人工抽检评分这些指标用Prometheus采集Grafana做看板再配一套告警规则。比如P99延迟超过5秒持续3分钟就告警错误率超过5%就告警。别小看这套东西它能帮你在用户投诉之前就发现问题。注意token计数一定要在服务端做不能依赖模型API返回的usage字段因为不同厂商的统计口径不一致而且有些流式响应根本不返回usage。4. 实操过程与核心环节实现4.1 环境准备与依赖安装我以Python技术栈为例走一遍完整的搭建流程。基础环境是Ubuntu 22.04Python 3.10CUDA 12.1如果用GPU的话。# 创建虚拟环境 python -m venv ai-eng source ai-eng/bin/activate # 核心依赖 pip install fastapi uvicorn redis asyncpg pip install transformers accelerate bitsandbytes pip install prometheus-client这里解释一下选型理由FastAPI自带异步支持和自动文档适合做接入层Redis既做消息队列又做缓存一物两用bitsandbytes用于量化加载模型prometheus-client用于指标暴露。版本上建议锁定避免自动升级引入不兼容。4.2 接入层与Worker池的代码骨架接入层的核心逻辑是校验请求、生成任务ID、入队、返回任务ID。Worker池从队列消费任务执行推理写回结果。# 接入层 from fastapi import FastAPI import redis.asyncio as redis import uuid app FastAPI() r redis.Redis(hostlocalhost, port6379) app.post(/v1/infer) async def infer(payload: dict): task_id str(uuid.uuid4()) await r.xadd(infer_stream, {task_id: task_id, payload: str(payload)}) return {task_id: task_id, status: queued}Worker池这边我用asyncio.Semaphore控制并发数避免一次性拉太多任务把显存撑爆。# Worker池 import asyncio sem asyncio.Semaphore(4) # 根据显存调整 async def worker(): while True: messages await r.xreadgroup(workers, worker-1, {infer_stream: }, count1, block1000) if not messages: continue async with sem: await process_task(messages)并发数怎么定我的计算方式是显存总量除以单次推理峰值显存再留20%余量。比如24G显存单次推理峰值4G那并发数就是24除以4再乘0.8约等于4。这个值不是固定的要根据实际压测结果微调。4.3 上下文压缩的具体实现上下文压缩是控制成本的关键。我的实现逻辑是每次对话结束后检查累计token数超过阈值就触发压缩。def compress_context(history: list, threshold: int 2000): total_tokens sum(count_tokens(m[content]) for m in history) if total_tokens threshold: return history # 保留最近3轮其余压缩成摘要 recent history[-6:] old history[:-6] summary call_model(f请将以下对话压缩成一段摘要{old}) return [{role: system, content: f历史摘要{summary}}] recent这里有个坑压缩本身也要调用模型会产生额外成本。所以阈值不能设太低否则压缩频率太高反而更贵。我的经验是阈值设在2000到3000token之间压缩后控制在500token以内这样压缩带来的成本远小于节省的上下文成本。4.4 压测与参数调优实录系统搭好后一定要压测。我用的是locust模拟50个并发用户持续请求。第一轮压测结果很惨P99延迟12秒错误率8%。排查后发现两个问题一是Worker并发数设太高设了8导致显存频繁OOM触发重试二是Redis Stream的消费者组没有及时ack消息堆积。调整方案并发数降到4增加OOM自动降级逻辑消费者处理完立即ack。第二轮压测P99降到3.2秒错误率0.5%。第三轮把batch size从1调到8P99进一步降到2.1秒吞吐量翻了3倍。这个调优过程说明一个道理AI工程的性能瓶颈往往不在模型本身而在工程链路的配置。多花时间在压测和调参上比盲目换更大的模型划算得多。5. 常见问题与排查技巧实录5.1 典型问题速查表现象可能原因排查方向解决方案延迟突然飙升显存不足触发swap查看GPU显存和swap使用降低并发数或启用量化错误率上升云端API限流查看API返回码增加重试降级链路成本异常增长上下文未压缩统计平均输入token数启用摘要压缩回答质量下降量化精度损失对比量化前后输出调整量化位数或换方案服务无响应消息队列堆积查看队列长度扩容Worker或限流5.2 三个我踩过的坑第一个坑忽略冷启动。模型第一次加载要几十秒如果服务刚启动就接流量大量请求会超时。我的解决办法是加一个预热接口服务启动后自动跑几条测试请求等模型完全加载后再接入负载均衡。第二个坑日志打太多。为了排查问题我把每次请求的完整prompt和response都打进日志结果磁盘一天就满了而且日志写入本身拖慢了主流程。后来改成只记录token数、延迟、错误码这些结构化字段完整内容只在采样时记录。第三个坑降级话术太生硬。早期降级直接返回“服务繁忙请稍后重试”用户体感很差。后来改成“当前咨询人数较多我先为您记录问题稍后回复”配合异步通知用户满意度明显提升。降级不可怕可怕的是让用户感知到降级。5.3 成本控制的独家技巧成本控制的核心是让每一分钱都花在刀刃上。我的做法是对请求做分级简单请求走小模型复杂请求走大模型。分级逻辑可以用一个轻量分类器实现准确率90%以上就够了分错的代价远小于全量走大模型的成本。另外缓存高频问题的答案。很多用户问的问题是重复的把高频问题的答案缓存起来命中缓存直接返回能省下大量推理成本。缓存key用问题的语义哈希而非字面哈希这样换个说法也能命中。最后再分享一个实操心得定期做成本审计。每周拉一次账单按业务线、按模型、按请求类型拆分找出成本大头。我就是在一次审计中发现某个内部测试接口被外部扫描器疯狂调用一天烧掉了几百块。加个鉴权就解决了。这套从零搭建的AI工程体系我前后迭代了三个版本踩了无数坑才稳定下来。它不是什么高深的技术核心就是把工程领域成熟的模式搬到AI场景里异步解耦、批处理、降级、可观测性、成本控制。这些词听起来不性感但它们是让AI应用从Demo走向生产的关键。你不需要一次全做完可以先从异步解耦和可观测性入手这两块投入产出比最高。等业务量上来了再逐步补全其他环节。
