1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年“AI工程”这个词被炒得火热招聘JD上动不动就要求“熟悉大模型应用开发”“有AI工程化落地经验”。但真到了动手阶段很多人第一反应就是pip install openai然后写个几十行的调用脚本觉得这就是AI工程了。我刚开始也这么想直到在一个真实项目里被狠狠教育了一顿——线上并发一上来接口超时、Token超限、上下文丢失、成本失控各种问题像约好了一样同时爆发。那时候我才意识到AI工程和“调个API”之间隔着一整套工程化的思维和方法论。ai-engineering-from-scratch这个标题核心讲的其实就是一件事不依赖现成的高级框架从最基础的原理和组件出发一步步搭建起一套能扛住真实业务压力的AI应用系统。它适合那些已经会写Python、懂一点后端开发但在AI工程化方面还处于“只会调接口”阶段的开发者也适合那些想真正理解大模型应用底层运转逻辑、不想永远当“调包侠”的技术人。这篇文章我会把自己从零搭建AI工程能力的完整路径拆开来讲包括整体设计思路、核心模块的实现细节、实操过程中踩过的坑以及那些文档里不会写的经验技巧。读完你至少能明白一个能上生产的AI应用到底需要哪些东西以及每一块该怎么落地。2. 整体架构设计先想清楚要解决什么问题2.1 从“能跑”到“能扛”AI工程的核心矛盾很多人对AI应用的认知停留在“输入问题→模型返回答案”这个层面。这在Demo阶段没问题但一旦放到真实业务场景里立刻会遇到几个绕不开的矛盾。第一个矛盾是响应速度与模型能力之间的取舍。大模型推理本身就有延迟参数量越大延迟越高。用户可不会管你背后跑的是多大的模型他们只关心“我点了按钮之后多久能看到结果”。我实测下来一个7B参数的模型在消费级显卡上单次推理大概需要1到3秒如果加上前后处理逻辑很容易突破5秒。这个延迟在聊天场景里勉强能接受但在搜索推荐、实时客服这类场景里就是灾难。第二个矛盾是成本与并发量之间的平衡。按Token计费的模式下每一次调用都是真金白银。我见过一个团队做内部知识库问答没做任何缓存和限流结果一个月账单跑到了五位数。后来加了语义缓存和请求合并成本直接砍掉六成以上。第三个矛盾是输出稳定性与业务要求之间的差距。大模型的输出是概率性的同样的输入可能得到不同的输出。但业务系统往往要求确定性——比如订单查询必须返回准确数据不能“大概”“可能”。这就需要在模型之外加一层“护栏”用规则、检索、校验等手段把输出约束在可控范围内。ai-engineering-from-scratch要解决的就是这三个矛盾。它不是教你训练一个更强的模型而是教你用工程手段把模型的能力包装成可靠的服务。2.2 分层架构每一层只做一件事我最终采用的架构分成四层从下往上依次是模型接入层、能力封装层、业务编排层、接口服务层。这个分层不是拍脑袋想的而是根据实际排查问题的经验倒推出来的。模型接入层负责和底层推理服务打交道包括模型加载、推理请求发送、结果解析、异常重试。这一层的核心目标是屏蔽不同模型之间的差异让上层不需要关心底层跑的是哪个模型、部署在哪里。我试过同时接入本地部署的模型和云端API如果没有这一层抽象上层代码里会到处是if-else。能力封装层把模型的基础能力包装成一个个独立的“技能”比如文本摘要、意图识别、实体抽取、问答生成。每个技能有自己的输入输出格式、提示词模板、后处理逻辑。这一层的关键是可复用——同一个摘要能力既可以用在新闻聚合场景也可以用在会议纪要场景只是提示词和参数不同。业务编排层负责把多个技能组合起来完成一个完整的业务流程。比如一个智能客服场景可能需要先做意图识别再根据意图路由到不同的处理逻辑最后生成回复。这一层要处理的是流程控制、条件分支、异常降级。接口服务层对外暴露HTTP或WebSocket接口处理鉴权、限流、日志、监控。这一层看起来最“传统”但恰恰是很多AI项目最容易忽视的地方。我见过太多项目把模型调用直接写在Flask路由函数里结果连个像样的错误码都没有。注意分层不是目的隔离变化才是。每一层的变化频率不同模型接入层可能因为换模型而变业务编排层可能因为需求调整而变。分层是为了让某一层的变化不会波及到其他层。2.3 技术选型为什么我选了这些组件在具体技术选型上我遵循一个原则能用标准库就不用第三方库能用轻量级就不用重型框架。这不是排斥框架而是从零搭建的过程中每引入一个依赖都会增加理解成本和排查难度。模型推理方面我选择了transformers加accelerate的组合而不是直接用更高层的推理框架。原因很简单我需要清楚地知道每一次推理请求到底经过了哪些步骤显存是怎么分配的batch是怎么组装的。transformers的代码可读性足够好遇到问题能直接翻源码定位。服务框架用的是FastAPI主要看中它的异步支持和自动文档生成。AI应用的接口往往需要处理较长的推理时间同步框架很容易把线程池打满。FastAPI的async特性配合uvicorn的多worker模式实测下来在同等硬件条件下能支撑的并发数比Flask高出一大截。缓存层用了Redis但用法和传统Web应用不太一样。除了常规的键值缓存我还用Redis的向量相似度搜索功能做语义缓存——把用户的问题向量化之后在Redis里查找相似的历史问题如果相似度超过阈值就直接返回缓存答案。这个优化在问答类场景里效果非常明显命中率能做到30%到40%。任务队列用的是Celery加Redis作为broker。有些AI任务耗时较长比如批量文档处理、模型微调不适合在HTTP请求周期内完成。用Celery把这些任务异步化之后接口响应时间从几十秒降到了几百毫秒。3. 核心模块实现从模型加载到服务暴露3.1 模型加载与推理优化别让显存成为瓶颈模型加载是AI工程的第一步也是最容易出问题的一步。我刚开始的时候直接from_pretrained加载了一个13B的模型结果显存直接爆了。后来才搞明白模型加载涉及几个关键参数每一个都会影响显存占用和推理速度。torch_dtype决定了模型权重的精度。默认是float32一个13B模型大概需要52GB显存。改成float16直接减半到26GB改成int8再减半到13GB。但精度降低会带来输出质量下降需要根据业务场景权衡。我的经验是对话类场景用float16基本无损分类和抽取类任务用int8也能接受。device_map控制模型各层分布在哪些设备上。如果有多张显卡可以设置成auto让accelerate自动分配。但自动分配不一定最优我遇到过自动分配导致跨卡通信频繁、推理速度反而变慢的情况。这时候需要手动指定每一层的设备把计算密集的层放在同一张卡上。max_memory参数可以限制每张卡的最大显存使用量防止模型加载时把显存占满导致其他服务无法运行。这个参数在多服务共享GPU的环境下特别重要。from transformers import AutoModelForCausalLM, AutoTokenizer import torch model_name your-model-path tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.float16, device_mapauto, max_memory{0: 20GB, 1: 20GB}, low_cpu_mem_usageTrue )推理阶段还有一个容易被忽视的优化点KV Cache。大模型在生成每一个Token时都需要重新计算之前所有Token的注意力。如果不做缓存生成长度为N的文本需要O(N²)的计算量。开启KV Cache之后之前计算过的Key和Value会被缓存下来后续Token只需要计算新的部分计算量降到O(N)。transformers默认开启KV Cache但在某些自定义推理逻辑里可能会被关掉需要特别注意。批处理是另一个提升吞吐量的关键手段。单条推理和批量推理的显存占用差距远小于吞吐量差距。我实测下来batch size从1增加到8吞吐量能提升5倍左右而显存只增加了不到2倍。但batch size也不是越大越好太大会导致单次推理延迟增加影响用户体验。需要根据业务对延迟和吞吐的要求找到平衡点。3.2 提示词工程化把“玄学”变成可管理的资产提示词是AI应用里最“玄学”的部分但工程化的核心就是把玄学变成可管理、可测试、可迭代的资产。我的做法是把提示词从代码里抽出来用独立的模板文件管理每个模板有版本号、适用场景、测试用例。模板文件用YAML格式结构大概是这样summarize: version: 1.2 description: 通用文本摘要 template: | 请对以下文本进行摘要要求 1. 保留核心信息 2. 不超过{max_length}字 3. 用{language}输出 文本内容 {text} variables: - name: max_length type: int default: 200 - name: language type: str default: 中文 test_cases: - input: text: 很长的一段文本... max_length: 100 expected_keywords: [关键词1, 关键词2]这样做的好处是产品经理也能参与提示词的调整不需要改代码。每次修改都有版本记录出问题可以快速回滚。测试用例可以在CI流程里自动跑确保修改提示词不会破坏已有功能。提示词设计本身也有一些经验性的技巧。角色设定要具体不要只说“你是一个助手”而是说“你是一个有十年经验的儿科医生擅长用通俗语言解释专业问题”。输出格式要明确如果需要JSON输出就在提示词里给出JSON schema并加上“只输出JSON不要有其他内容”的约束。少样本示例要精选两到三个高质量示例比十个普通示例效果好得多。还有一个容易被忽视的点提示词的Token长度。很多模型有上下文窗口限制比如4096个Token。如果提示词本身就用掉了3000个Token留给模型生成的空间就很少了。我习惯在模板里加一个Token计数逻辑超过阈值就自动截断或压缩。3.3 输出解析与校验让模型输出变得“可用”模型输出的是自然语言但业务系统需要的是结构化数据。这中间的转换就是输出解析要做的事。最简单的做法是让模型输出JSON然后用json.loads解析。但实际跑下来模型输出的JSON经常有问题多了个逗号、少了引号、用了单引号、嵌套了markdown代码块标记。我的解决方案是三层解析策略。第一层直接尝试json.loads能解析成功最好。第二层用正则表达式提取JSON部分去掉markdown标记和多余文字。第三层用更宽松的解析器比如json5或者自己写一个容错解析逻辑。如果三层都失败就触发重试机制把解析错误信息拼回提示词里让模型重新生成。import json import re def parse_model_output(text): # 第一层直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 第二层提取JSON块 json_match re.search(r\{.*\}, text, re.DOTALL) if json_match: try: return json.loads(json_match.group()) except json.JSONDecodeError: pass # 第三层宽松解析 try: import json5 return json5.loads(text) except: pass # 全部失败返回None触发重试 return None除了格式校验内容校验也很重要。比如模型生成了一个日期格式是“2024年13月45日”格式上没问题但语义上错误。这时候需要用规则引擎做二次校验把明显不合理的值过滤掉。对于关键字段还可以用另一个模型做交叉验证或者用检索增强的方式从知识库里核对。提示输出解析的容错逻辑一定要有日志记录。每次解析失败都是一次改进提示词的机会。我习惯把失败案例收集起来每周review一次针对性地优化提示词模板。3.4 缓存与限流成本控制的两把利器AI应用的成本大头在模型推理而推理成本又和调用次数直接相关。减少不必要的调用就是最直接的成本优化手段。缓存和限流是两个最有效的方法。缓存分两种精确缓存和语义缓存。精确缓存就是传统的键值缓存把输入文本的哈希作为键输出作为值。实现简单但命中率有限因为用户换个说法就匹配不上了。语义缓存是把输入文本向量化在向量数据库里查找相似的历史请求。相似度超过阈值就返回缓存结果。我用的阈值是0.92实测下来准确率能到95%以上同时命中率比精确缓存高出三到四倍。语义缓存的实现关键是向量化模型的选择。用大模型做向量化成本太高我选了一个轻量级的句子向量模型推理速度快向量维度适中。Redis从7.0版本开始支持向量相似度搜索不需要额外部署向量数据库对中小规模应用来说足够用了。import redis import numpy as np from sentence_transformers import SentenceTransformer r redis.Redis() encoder SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def semantic_cache_get(query, threshold0.92): query_vec encoder.encode(query).astype(np.float32).tobytes() # Redis向量搜索逻辑 results r.ft(idx).search( redis.commands.search.query.Query(*[KNN 1 vec $vec AS score]) .sort_by(score) .return_fields(response, score) .dialect(2), {vec: query_vec} ) if results.docs: score float(results.docs[0].score) if score threshold: return results.docs[0].response return None限流则是从另一个维度控制成本。我用的策略是分层限流免费用户每分钟5次付费用户每分钟50次内部服务每分钟500次。限流算法用的是滑动窗口比固定窗口更平滑不会出现窗口切换时的流量突刺。Redis的ZSET结构天然适合实现滑动窗口限流每次请求把时间戳作为score插入然后统计窗口内的请求数。限流触发时不能简单返回429错误那样用户体验很差。我的做法是返回一个排队提示把请求放入延迟队列等窗口滑动后再处理。对于实时性要求不高的场景这个方案用户几乎无感知。4. 实操全流程从零到一搭建一个问答服务4.1 环境准备与依赖安装先说一下我的硬件环境一台带RTX 3090的服务器24GB显存32GB内存。这个配置跑7B到13B的模型没问题再大就需要量化或者多卡了。操作系统是Ubuntu 22.04Python版本3.10。依赖安装有几个坑需要提前说。torch的版本要和CUDA版本匹配我用的CUDA 11.8对应的torch是2.0以上。transformers的版本不要太新也不要太旧太新可能有API变动太旧可能不支持某些模型。我锁定在4.36版本实测比较稳定。pip install torch2.1.0 --index-url https://download.pytorch.org/whl/cu118 pip install transformers4.36.0 accelerate0.25.0 pip install fastapi uvicorn redis celery sentence-transformersaccelerate的版本要和transformers匹配否则device_mapauto可能不生效。sentence-transformers用来做向量化它依赖的torch版本可能和主环境冲突建议单独建一个虚拟环境跑向量化服务。4.2 模型服务封装与启动模型服务我封装成了一个独立的类初始化时加载模型对外暴露generate方法。这个类被FastAPI的多个路由共享避免重复加载模型。class ModelService: def __init__(self, model_path): self.tokenizer AutoTokenizer.from_pretrained(model_path) self.model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypetorch.float16, device_mapauto ) self.model.eval() torch.no_grad() def generate(self, prompt, max_new_tokens512, temperature0.7): inputs self.tokenizer(prompt, return_tensorspt).to(self.model.device) outputs self.model.generate( **inputs, max_new_tokensmax_new_tokens, temperaturetemperature, do_sampleTrue, top_p0.9, repetition_penalty1.1 ) response self.tokenizer.decode(outputs[0], skip_special_tokensTrue) # 去掉输入部分只保留生成内容 return response[len(prompt):]temperature控制输出的随机性值越低输出越确定。问答场景我一般设0.3到0.7创意写作可以设0.8到1.0。top_p是核采样参数0.9表示只从累积概率前90%的Token里采样能有效避免生成奇怪的内容。repetition_penalty大于1可以抑制重复生成但设太高会导致语句不通顺1.1到1.2是比较安全的范围。启动服务用uvicornworker数量根据GPU数量来定。单卡的话建议1个worker因为多个worker会争抢显存。多卡可以用--workers参数指定每个worker绑定一张卡。uvicorn main:app --host 0.0.0.0 --port 8000 --workers 14.3 完整问答流程的代码实现一个完整的问答请求会经过接收请求→语义缓存查询→提示词组装→模型推理→输出解析→缓存写入→返回响应。我把这些步骤串成一个Pipeline每一步都有独立的异常处理和日志记录。app.post(/qa) async def qa_endpoint(request: QARequest): # 1. 语义缓存查询 cached semantic_cache_get(request.question) if cached: return {answer: cached, source: cache} # 2. 提示词组装 prompt build_prompt( questionrequest.question, contextrequest.context, max_lengthrequest.max_length ) # 3. 模型推理 try: raw_output model_service.generate(prompt) except Exception as e: logger.error(fModel inference failed: {e}) return {error: 服务暂时不可用请稍后重试} # 4. 输出解析 parsed parse_model_output(raw_output) if parsed is None: # 重试一次 raw_output model_service.generate(prompt \n请确保输出为合法JSON格式。) parsed parse_model_output(raw_output) # 5. 缓存写入 if parsed: semantic_cache_set(request.question, parsed) return {answer: parsed, source: model}这个流程看起来简单但每一步都有细节。比如缓存写入要设置过期时间问答类内容一般设24小时。缓存键要用归一化后的问题去掉标点和空格差异。模型推理的超时时间要设置合理太长会拖垮整个服务太短会频繁触发降级。4.4 性能压测与调优记录服务搭好之后一定要做压测不然你不知道它能扛多少并发。我用locust做压测模拟了10、50、100三个并发级别。10并发的时候平均响应时间1.2秒P99是2.5秒完全没问题。50并发的时候平均响应时间涨到了3.8秒P99到了8秒开始有请求超时。100并发的时候服务直接卡死大量请求堆积。排查下来发现两个瓶颈一是模型推理本身是串行的同一时刻只能处理一个请求二是FastAPI的默认线程池太小请求排队等待。第一个问题的解决方案是请求队列加批处理。把所有请求放入队列攒够一批或者等待超过50毫秒就触发一次批量推理。这样虽然单次推理时间变长了但吞吐量大幅提升。实测50并发下批处理方案的平均响应时间降到了2.1秒P99降到了4秒。第二个问题的解决方案是调整uvicorn的limit_concurrency参数并给模型推理加信号量控制。同时把一些非核心逻辑比如日志写入、缓存更新改成异步执行不阻塞主流程。调优之后同样的硬件能稳定支撑80到100并发对于内部工具和中小型应用来说足够了。如果还要更高就需要考虑模型量化、多卡并行或者换更小的模型。5. 常见问题与排查技巧实录5.1 模型输出不稳定怎么办这是被问得最多的问题。同样的输入有时候输出很好有时候完全跑偏。原因通常有三个温度参数太高、提示词不够明确、模型本身能力不足。排查顺序是这样的先把temperature设成0看输出是否稳定。如果稳定了说明是随机性问题适当降低温度或者用do_sampleFalse。如果还不稳定检查提示词里有没有歧义。比如“总结一下”就不如“用三句话总结每句话不超过20字”明确。如果提示词没问题那就是模型能力不够考虑换更大的模型或者用少样本示例引导。还有一个隐藏原因是上下文污染。如果多轮对话的历史没有正确清理之前的对话内容会影响当前轮次的输出。我的做法是每轮对话只保留最近3轮历史并且用明确的分隔符把历史对话和当前问题隔开。5.2 显存溢出OOM的排查思路OOM是AI工程里最常见的错误之一。排查的时候按这个顺序来先看模型加载时的显存占用再看推理时的峰值显存最后看是否有内存泄漏。模型加载OOM通常是精度太高或者max_memory没设置。把torch_dtype改成float16或者int8设置合理的max_memory限制。推理OOM通常是输入太长或者batch size太大。检查输入Token数是否超过了模型的最大长度检查batch size是否超过了显存承受能力。可以用torch.cuda.max_memory_allocated()打印峰值显存找到瓶颈。内存泄漏比较隐蔽表现为服务运行一段时间后OOM。常见原因是缓存没有清理、中间变量没有释放、日志对象持有引用。用gc.collect()和torch.cuda.empty_cache()定期清理能缓解大部分问题。5.3 接口超时的降级策略AI接口的超时时间设置很讲究。设太短正常请求也会超时设太长异常请求会拖垮服务。我的经验值是同步接口超时时间设为P99响应时间的1.5倍。比如P99是4秒超时就设6秒。超时之后的降级策略分三级。第一级是返回缓存结果如果有的话。第二级是返回一个简化的回答比如“当前请求较多请稍后重试”。第三级是直接返回错误码让客户端决定怎么处理。降级策略要提前配置好不能等出问题了再临时加。我习惯在服务启动时就注册好降级回调超时触发时自动执行。5.4 常见问题速查表问题现象可能原因排查方法解决方案输出乱码或重复温度过高、重复惩罚过低检查temperature和repetition_penalty降低温度到0.3-0.7提高重复惩罚到1.1-1.2显存溢出精度过高、batch过大、输入过长打印峰值显存检查输入长度降低精度、减小batch、截断输入接口超时并发过高、模型推理慢查看P99延迟和队列长度加缓存、批处理、限流降级输出格式错误提示词不明确、模型能力不足检查提示词模板和测试用例明确输出格式要求加少样本示例缓存命中率低阈值过高、向量模型不匹配统计相似度分布调整阈值换更适合的向量模型服务启动慢模型加载耗时、依赖冲突查看启动日志时间戳预加载模型锁定依赖版本注意这张表是我在实际运维中总结的但每个项目的具体情况不同。遇到问题时先看日志再看监控指标最后才动手改代码。盲目调参往往会让问题更复杂。5.5 几个只有踩过坑才知道的经验第一个经验日志要记录完整的提示词和输出。很多人为了省磁盘空间只记录摘要结果出问题的时候根本不知道模型收到了什么、生成了什么。我现在的做法是记录完整内容但设置自动清理策略保留最近7天。第二个经验模型版本要锁定。from_pretrained如果不指定revision每次拉取的可能不是同一个版本。模型更新后输出风格可能变化导致线上服务行为不一致。我习惯把模型文件下载到本地用本地路径加载彻底避免版本漂移。第三个经验不要迷信大模型。很多任务用小模型加好的提示词效果不比大模型差但成本和延迟低得多。我做过一个意图分类任务7B模型加少样本示例的准确率是92%13B模型是93%但13B的推理时间是7B的两倍多。这种场景下7B显然是更优选择。第四个经验监控要覆盖业务指标。除了CPU、内存、显存这些系统指标还要监控缓存命中率、平均响应时间、错误率、Token消耗量这些业务指标。我遇到过一次线上问题系统指标全部正常但用户反馈回答质量下降。后来查监控发现是缓存命中率突然从35%掉到了5%原因是向量模型更新后向量空间变了导致相似度计算全部失效。6. 工程化之外的思考AI工程能力的真正壁垒6.1 从“能用”到“好用”的距离把模型跑起来不难难的是让它稳定、高效、低成本地跑下去。这中间的差距就是工程能力。我见过太多团队花大力气调模型、刷榜单但上线之后被最基础的并发问题、缓存问题、监控问题搞得焦头烂额。AI工程的壁垒不在模型本身而在对业务场景的理解、对系统边界的把握、对异常情况的预案。一个经验丰富的AI工程师拿到需求之后第一反应不是“用什么模型”而是“这个场景的QPS是多少”“延迟要求是多少”“成本预算是多少”“数据敏感度如何”。这些问题的答案决定了技术方案的选择。6.2 持续迭代的机制建设AI应用上线不是终点而是起点。用户的提问方式会变业务需求会变模型也会更新。如果没有一套持续迭代的机制服务很快就会跟不上变化。我的做法是建立三个闭环。数据闭环收集用户反馈和bad case定期分析找出共性问题。提示词闭环每次修改提示词都跑一遍回归测试确保不破坏已有功能。模型闭环定期评估新模型在业务数据上的表现有提升就灰度上线没提升就继续观察。这三个闭环不需要很复杂的工具用Excel加脚本就能跑起来。关键是坚持每周花一个小时做review比攒到季度末集中处理有效得多。6.3 给刚入门的开发者的建议如果你刚开始接触AI工程我的建议是先不要碰框架。用最基础的transformers加FastAPI手动实现一遍完整的流程。你会遇到各种问题但每解决一个问题你对AI工程的理解就深一层。等你能徒手搭建一个稳定运行的问答服务之后再去了解LangChain、LlamaIndex这些框架。这时候你会发现框架帮你封装的东西你都已经理解了用起来会更有判断力遇到问题也知道从哪里排查。还有一点很重要不要只盯着模型。AI工程是一个系统工程模型只是其中一环。缓存、队列、监控、降级、限流这些传统后端工程的东西同样重要。我见过模型能力很强但工程做得一塌糊涂的项目也见过模型一般但工程做得很扎实的项目。后者的用户体验往往更好。最后分享一个我自己的习惯每做一个新项目我都会画一张架构图标出每一个可能出问题的点然后针对每个点写一个预案。这张图会随着项目迭代不断更新。时间长了你会发现很多问题是相通的解决了一个其他项目也能复用。这种积累才是AI工程能力真正的护城河。