1. 从零搭建AI工程体系为什么我劝你别一上来就啃论文ai-engineering-from-scratch这个标题乍一看像是又一份从入门到精通的教程合集但真正做过AI项目落地的人会明白它指向的其实是一个更硬核的问题当你手里只有一台普通开发机、一个模糊的业务需求、以及一堆散落在各处的开源工具时怎么把能跑通demo变成能扛住线上流量。我自己第一次接触AI工程是在做一个商品标题生成的小工具当时觉得调个API就完事了结果上线第二天就遇到超时、并发打满、生成结果不稳定三连击。那次之后我才意识到AI工程和传统后端工程最大的区别在于它的不确定性来自模型本身而不是代码逻辑。代码你可以靠单元测试覆盖但模型输出的质量、延迟、成本全都是概率问题。所以这篇内容我想聊的不是怎么训练一个大模型而是怎么围绕模型搭一套能用的工程体系。适合谁看如果你是会写Python、懂基本后端概念但一碰到模型部署、推理优化、效果评估就发懵的开发者那这篇就是写给你的。我会从整体设计思路讲起一路拆到具体的目录结构、参数配置、排查技巧尽量做到你照着抄就能跑起来。2. 整体架构怎么设计先想清楚边界再谈技术选型2.1 为什么从零不等于从模型开始很多人对AI工程的误解是核心是模型。但实际项目里模型往往是最不需要你操心的那一环——你可以用开源权重可以调托管接口真正吃掉你80%时间的是模型之外的东西数据怎么进、结果怎么出、失败了怎么重试、成本怎么控。我习惯把AI工程体系拆成四层从下往上分别是基础设施层算力、存储、网络。这一层决定了你的天花板比如有没有GPU、带宽够不够、磁盘IO快不快。模型服务层模型加载、推理、批处理、缓存。这一层是AI工程的核心差异点。应用编排层Prompt管理、多模型路由、结果后处理、业务逻辑。可观测层日志、指标、追踪、评估。这一层最容易被忽略但出事时全靠它。from-scratch的关键在于你要有能力把这四层都握在自己手里而不是被某个平台绑死。我见过太多团队一开始图省事全用托管服务等到要降本、要换模型、要做私有化时发现迁移成本高得离谱。2.2 技术选型的三个取舍原则选型这件事没有标准答案但有三条原则我踩坑踩出来的第一优先选你能调试的而不是性能最好的。比如推理框架vLLM吞吐是高但如果你连它的调度逻辑都看不懂出问题时只能干瞪眼。新手阶段我更推荐先用Transformers跑通再逐步换。第二把可替换当成硬指标。模型、向量库、推理框架全都要能一键切换。做法很简单所有外部依赖都包一层自己的接口业务代码只调你的接口。这样换模型时改一个适配器就行。第三成本要能算清楚。每次推理消耗多少token、多少显存、多少电费这些数字必须能实时看到。我见过一个团队上线三个月才发现账单超预算十倍就是因为从来没做过成本埋点。下面这张表是我总结的常见选型对照供你参考环节轻量方案进阶方案适用场景推理框架TransformersvLLM / TGI前者调试友好后者高并发向量检索FAISSMilvus / Qdrant数据量小于百万用FAISS足够服务框架FastAPIRay Serve单模型用FastAPI多模型编排用Ray缓存内存字典Redis单机用字典分布式必须Redis监控打印日志Prometheus Grafana上线前必须补齐2.3 目录结构一个能长大的项目长什么样我见过太多AI项目最后变成一坨脚本根源就是一开始没规划目录。下面这个结构是我迭代了四五个项目后固定下来的你可以直接拿去用ai-engineering/ ├── configs/ # 所有配置按环境分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── src/ │ ├── models/ # 模型加载与推理封装 │ ├── services/ # 业务逻辑 │ ├── adapters/ # 外部依赖适配层 │ ├── pipelines/ # 编排流程 │ └── utils/ # 通用工具 ├── data/ │ ├── raw/ # 原始数据 │ ├── processed/ # 处理后数据 │ └── cache/ # 缓存 ├── tests/ │ ├── unit/ │ └── integration/ ├── scripts/ # 运维脚本 └── notebooks/ # 实验记录关键点在于adapters这一层。所有对模型、数据库、第三方API的调用都放这里业务层永远不直接import外部库。这样做的代价是多写一层代码收益是换任何组件时业务层零改动。我实测下来这个习惯至少帮我省了三次大规模重构。3. 核心细节拆解模型服务层到底该怎么写3.1 模型加载别小看这几行代码模型加载看起来简单但坑特别多。最常见的错误是每次请求都重新加载模型显存直接爆炸。正确做法是全局加载一次常驻内存。# src/models/loader.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer class ModelLoader: _instance None def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def load(self, model_path, devicecuda, dtypetorch.float16): self.tokenizer AutoTokenizer.from_pretrained(model_path) self.model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypedtype, device_mapdevice, ) self.model.eval() return self这里用单例模式保证模型只加载一次。torch_dtype选float16是权衡float32精度高但显存翻倍int8省显存但可能掉点。我一般先用float16跑通显存不够再考虑量化。注意device_map参数在单卡和多卡下行为不同。单卡直接写cuda多卡写auto让框架自动分配。写错了不会报错但会静默走CPU速度慢十倍。3.2 推理封装批处理是性能的分水岭单条推理和批量推理的性能差距实测能到5到10倍。原因是GPU的并行能力在单条请求下完全浪费了。所以推理层必须支持动态批处理。# src/models/inference.py import time from typing import List class InferenceEngine: def __init__(self, loader, max_batch_size8, max_wait_ms50): self.loader loader self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue [] def generate(self, prompts: List[str], max_new_tokens256): inputs self.loader.tokenizer( prompts, return_tensorspt, paddingTrue, truncationTrue ).to(self.loader.model.device) with torch.no_grad(): outputs self.loader.model.generate( **inputs, max_new_tokensmax_new_tokens, do_sampleTrue, temperature0.7, top_p0.9, ) return self.loader.tokenizer.batch_decode( outputs, skip_special_tokensTrue )max_batch_size和max_wait_ms这两个参数需要根据你的场景调。延迟敏感的场景把max_wait_ms调到10以内吞吐优先的场景可以放到100。我一般从batch_size8, wait50ms起步然后压测调优。3.3 缓存策略省下的都是真金白银AI推理的成本大头在算力而很多请求其实是重复的。缓存做得好成本能降一半以上。缓存分三层精确缓存请求完全一致时直接返回。用请求的hash做key存Redis。语义缓存请求语义相似时返回近似结果。用向量检索找最相似的缓存项相似度超过阈值就复用。前缀缓存多个请求共享相同前缀时复用KV Cache。这个需要推理框架支持vLLM有内置。精确缓存实现最简单收益也最直接import hashlib import json import redis class ExactCache: def __init__(self, redis_client, ttl3600): self.redis redis_client self.ttl ttl def _key(self, prompt, params): raw json.dumps({p: prompt, params: params}, sort_keysTrue) return ai:cache: hashlib.sha256(raw.encode()).hexdigest() def get(self, prompt, params): return self.redis.get(self._key(prompt, params)) def set(self, prompt, params, result): self.redis.setex(self._key(prompt, params), self.ttl, result)提示缓存key一定要包含所有影响输出的参数比如temperature、max_tokens。我踩过一次坑只用了prompt做key结果改了temperature后返回的还是旧结果排查了半天。3.4 参数调优temperature和top_p到底怎么设这两个参数是新手最容易懵的。用生活化的类比模型生成每个词时会给所有候选词打分。temperature控制敢不敢选冷门词top_p控制从多少候选里选。temperature0永远选最高分的词输出稳定但死板。适合分类、抽取类任务。temperature0.7适度随机输出自然。适合对话、创作。temperature1.5非常随机容易胡说。除非做头脑风暴否则别用。top_p一般设0.9到0.95意思是只从累计概率前90%的词里选。设太低会限制表达设太高等于没限制。我的经验配置任务类型temperaturetop_pmax_tokens信息抽取0.10.9512对话问答0.70.91024创意写作0.90.952048代码生成0.20.9520484. 实操全流程从零跑通一个AI服务4.1 环境准备与依赖安装先把基础环境搭起来。我推荐用conda隔离环境避免依赖冲突conda create -n ai-eng python3.10 -y conda activate ai-eng pip install torch transformers fastapi uvicorn redis pyyaml版本选择上Python 3.10是目前兼容性最好的3.11有些库还没跟上。PyTorch装之前先确认CUDA版本用nvidia-smi看驱动支持的CUDA版本然后去官网找对应命令。装错了不会报错但会用CPU跑速度差几十倍。4.2 配置文件设计把魔法数字赶出代码所有可变参数都放配置文件代码里只留逻辑。这是我从无数次改个参数要重新部署的痛苦中总结的# configs/base.yaml model: path: your-model-path device: cuda dtype: float16 max_batch_size: 8 max_wait_ms: 50 inference: default_max_new_tokens: 512 default_temperature: 0.7 default_top_p: 0.9 cache: enabled: true redis_host: localhost redis_port: 6379 ttl: 3600 server: host: 0.0.0.0 port: 8000 workers: 1加载配置的代码import yaml def load_config(envdev): with open(fconfigs/base.yaml) as f: base yaml.safe_load(f) with open(fconfigs/{env}.yaml) as f: override yaml.safe_load(f) return deep_merge(base, override)deep_merge是个递归合并字典的小工具让环境配置只写差异部分避免重复。4.3 服务接口实现FastAPI三件套服务层用FastAPI核心就三个接口健康检查、单条推理、批量推理。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List app FastAPI() engine None cache None class GenerateRequest(BaseModel): prompt: str max_new_tokens: int 512 temperature: float 0.7 top_p: float 0.9 class BatchRequest(BaseModel): prompts: List[str] max_new_tokens: int 512 app.get(/health) def health(): return {status: ok, model_loaded: engine is not None} app.post(/generate) def generate(req: GenerateRequest): params req.dict() cached cache.get(req.prompt, params) if cached: return {result: cached, cached: True} result engine.generate([req.prompt], req.max_new_tokens)[0] cache.set(req.prompt, params, result) return {result: result, cached: False} app.post(/batch_generate) def batch_generate(req: BatchRequest): if len(req.prompts) 32: raise HTTPException(400, batch too large) results engine.generate(req.prompts, req.max_new_tokens) return {results: results}启动命令uvicorn src.main:app --host 0.0.0.0 --port 8000 --workers 1注意workers千万别设大于1。因为模型是单例加载在进程内存里的多worker会导致每个进程都加载一份模型显存直接爆。要提并发就靠批处理不是靠多进程。4.4 压测与调优用数据说话服务跑起来后必须压测。我用的是locust写个简单的压测脚本from locust import HttpUser, task, between class AIUser(HttpUser): wait_time between(0.1, 0.5) task def generate(self): self.client.post(/generate, json{ prompt: 写一段产品介绍, max_new_tokens: 128 })跑起来后重点看三个指标QPS、P99延迟、显存占用。我实测的一组数据供参考单卡24G7B模型float16批大小QPSP99延迟显存占用13.2380ms15G49.8520ms17G814.5780ms19G1616.21400msOOM可以看到批大小到8之后QPS增长放缓延迟却涨得快。所以8是个甜点值。这个数字因模型和硬件而异你必须自己压一遍。5. 常见问题与排查技巧实录5.1 显存不够用怎么办这是最高频的问题。排查顺序是先看是不是多进程加载了模型再看批大小是不是太大最后考虑量化。# 查看显存占用 nvidia-smi # 查看具体进程 nvidia-smi --query-compute-appspid,used_memory --formatcsv如果确认是模型本身太大有几个降显存的手段按推荐顺序换float16从float32换过来直接省一半。开梯度检查点训练时用推理用不上。int8量化省一半显存精度掉1到2个点。int4量化省四分之三精度掉得明显慎用。CPU offload把部分层放CPU速度慢但能跑。5.2 输出结果不稳定怎么破同一个prompt两次输出不一样这是采样导致的正常。但如果差异大到影响业务就要处理把temperature调到0.1以下接近确定性输出。加seed参数固定随机种子。在Prompt里加约束比如只输出JSON不要解释。后处理做格式校验不合格就重试。我一般会写个重试装饰器def retry_on_invalid(max_retries3): def decorator(func): def wrapper(*args, **kwargs): for i in range(max_retries): result func(*args, **kwargs) if validate(result): return result raise ValueError(max retries exceeded) return wrapper return decorator5.3 排查速查表现象可能原因排查方法解决服务启动慢模型加载耗时看启动日志时间戳预热或异步加载首次请求慢冷启动对比首次和后续延迟启动时跑一次空推理显存OOM批太大或多进程nvidia-smi降批大小或改单worker输出乱码tokenizer不匹配检查模型和tokenizer路径用同一路径加载缓存不生效key设计问题打印key对比确保参数全进key并发上不去单worker阻塞看CPU利用率上批处理或异步5.4 几个我踩过的坑坑一tokenizer和模型版本不一致。有次我从两个地方分别下载模型和tokenizer结果词表对不上输出全是乱码。后来养成习惯永远从同一个路径加载。坑二忘了设eval模式。训练完直接推理忘了model.eval()dropout还在生效输出随机性大增。这个bug找了我一整天。坑三缓存没设过期。早期缓存永久有效结果模型更新后旧结果还在返回。现在所有缓存都带TTL最长一小时。坑四日志打太多。一开始把每个请求的完整prompt和输出都打日志磁盘一周就满了。现在只打摘要和指标详细内容按需开启。6. 可观测性上线前必须补齐的一环6.1 三个必须埋的指标AI服务的监控和普通服务不一样除了QPS、延迟、错误率还要关注Token吞吐每秒处理多少token这是成本的核心指标。缓存命中率命中率低于30%说明缓存策略有问题。输出长度分布突然变长可能是模型跑偏了。埋点代码import time from prometheus_client import Counter, Histogram REQUEST_COUNT Counter(ai_requests_total, Total requests, [status]) LATENCY Histogram(ai_latency_seconds, Request latency) TOKENS Counter(ai_tokens_total, Total tokens generated) def track(func): def wrapper(*args, **kwargs): start time.time() try: result func(*args, **kwargs) REQUEST_COUNT.labels(statussuccess).inc() TOKENS.inc(len(result.split())) return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise finally: LATENCY.observe(time.time() - start) return wrapper6.2 日志怎么打才有用日志不是越多越好关键是要能回答这次请求为什么慢/为什么错。我的日志格式固定包含请求ID、prompt摘要、参数、耗时、token数、缓存状态。import logging import uuid logger logging.getLogger(ai) def log_request(prompt, params, latency, tokens, cached): logger.info({ req_id: str(uuid.uuid4())[:8], prompt_preview: prompt[:50], params: params, latency_ms: round(latency * 1000), tokens: tokens, cached: cached, })用JSON格式打日志方便后续用工具解析。别用print也别用字符串拼接结构化日志能省你大量排查时间。6.3 效果评估怎么知道模型变好了还是变坏了这是AI工程最容易被忽略的部分。模型更新后怎么判断效果我的做法是维护一个小型评估集每次更新都跑一遍。评估集不用大100到200条就够但要覆盖典型场景和边界情况。每条包含输入和期望输出用规则或人工打分。关键指标准确率输出符合期望的比例。一致性同一输入多次输出的一致程度。格式合规率输出符合格式要求的比例。def evaluate(engine, eval_set): results {correct: 0, total: 0, format_ok: 0} for item in eval_set: output engine.generate([item[input]])[0] results[total] 1 if matches(output, item[expected]): results[correct] 1 if is_valid_format(output): results[format_ok] 1 return { accuracy: results[correct] / results[total], format_rate: results[format_ok] / results[total], }这套评估跑一次大概几分钟但能帮你避免感觉变好了其实变差了的误判。我吃过一次亏换了个新模型觉得输出更流畅结果评估集一跑准确率掉了8个点赶紧回滚。7. 成本控制AI工程绕不开的现实问题7.1 算清楚每一分钱花在哪AI服务的成本主要三块算力、存储、带宽。算力是大头尤其推理。我习惯给每个请求算一个成本分成本分 输入token数 * 单价 输出token数 * 单价 固定开销固定开销包括模型常驻显存的成本按小时摊到每个请求。这样每个请求的成本一目了然也方便做预算。7.2 降本的四个手段按性价比排序缓存命中率每提高10%成本降约8%。投入最小收益最大。批处理吞吐提升直接摊薄固定成本。量化int8量化能省一半显存等于同样硬件跑两倍模型。模型降级简单任务用小模型复杂任务才用大模型。这个需要做任务分类。我做过一个实验同样一批请求加上缓存和批处理后成本降了约60%。所以别急着换更便宜的硬件先把软件层优化做透。7.3 预算告警成本控制的关键是早知道。我设了三档告警日预算的50%、80%、100%。到80%就该查是不是有异常流量到100%自动降级到小模型。def check_budget(daily_cost, budget): ratio daily_cost / budget if ratio 1.0: return critical elif ratio 0.8: return warning elif ratio 0.5: return notice return ok这套机制帮我抓到过一次爬虫刷接口一天跑了平时一周的量告警及时止损。8. 写在最后一些个人体会做AI工程这两年我最大的感受是别把模型当黑盒但也别试图完全理解它。你需要知道它的能力边界、成本特性、失败模式但不需要懂每一层的数学推导。工程的价值在于把不确定性管起来让业务能稳定地用上AI能力。from-scratch的意义不在于什么都自己造而在于你清楚每一层在做什么、为什么这么选、出问题去哪查。这套体系我迭代了好几版从最初的几十行脚本到现在能扛住日均百万请求中间踩的坑比写的代码多得多。如果你刚开始搭我的建议是先跑通最小闭环一个模型、一个接口、一个缓存然后逐步加监控、加评估、加成本控制。别一上来就追求架构完美能跑起来、能观测、能回滚比什么都重要。最后分享一个小技巧每次改动前先跑一遍评估集记录基线。改动后再跑一遍对比数据。这个习惯能帮你把感觉变成事实少走很多弯路。 从零搭建AI工程体系：模型服务、推理优化与成本控制实战
1. 从零搭建AI工程体系为什么我劝你别一上来就啃论文ai-engineering-from-scratch这个标题乍一看像是又一份从入门到精通的教程合集但真正做过AI项目落地的人会明白它指向的其实是一个更硬核的问题当你手里只有一台普通开发机、一个模糊的业务需求、以及一堆散落在各处的开源工具时怎么把能跑通demo变成能扛住线上流量。我自己第一次接触AI工程是在做一个商品标题生成的小工具当时觉得调个API就完事了结果上线第二天就遇到超时、并发打满、生成结果不稳定三连击。那次之后我才意识到AI工程和传统后端工程最大的区别在于它的不确定性来自模型本身而不是代码逻辑。代码你可以靠单元测试覆盖但模型输出的质量、延迟、成本全都是概率问题。所以这篇内容我想聊的不是怎么训练一个大模型而是怎么围绕模型搭一套能用的工程体系。适合谁看如果你是会写Python、懂基本后端概念但一碰到模型部署、推理优化、效果评估就发懵的开发者那这篇就是写给你的。我会从整体设计思路讲起一路拆到具体的目录结构、参数配置、排查技巧尽量做到你照着抄就能跑起来。2. 整体架构怎么设计先想清楚边界再谈技术选型2.1 为什么从零不等于从模型开始很多人对AI工程的误解是核心是模型。但实际项目里模型往往是最不需要你操心的那一环——你可以用开源权重可以调托管接口真正吃掉你80%时间的是模型之外的东西数据怎么进、结果怎么出、失败了怎么重试、成本怎么控。我习惯把AI工程体系拆成四层从下往上分别是基础设施层算力、存储、网络。这一层决定了你的天花板比如有没有GPU、带宽够不够、磁盘IO快不快。模型服务层模型加载、推理、批处理、缓存。这一层是AI工程的核心差异点。应用编排层Prompt管理、多模型路由、结果后处理、业务逻辑。可观测层日志、指标、追踪、评估。这一层最容易被忽略但出事时全靠它。from-scratch的关键在于你要有能力把这四层都握在自己手里而不是被某个平台绑死。我见过太多团队一开始图省事全用托管服务等到要降本、要换模型、要做私有化时发现迁移成本高得离谱。2.2 技术选型的三个取舍原则选型这件事没有标准答案但有三条原则我踩坑踩出来的第一优先选你能调试的而不是性能最好的。比如推理框架vLLM吞吐是高但如果你连它的调度逻辑都看不懂出问题时只能干瞪眼。新手阶段我更推荐先用Transformers跑通再逐步换。第二把可替换当成硬指标。模型、向量库、推理框架全都要能一键切换。做法很简单所有外部依赖都包一层自己的接口业务代码只调你的接口。这样换模型时改一个适配器就行。第三成本要能算清楚。每次推理消耗多少token、多少显存、多少电费这些数字必须能实时看到。我见过一个团队上线三个月才发现账单超预算十倍就是因为从来没做过成本埋点。下面这张表是我总结的常见选型对照供你参考环节轻量方案进阶方案适用场景推理框架TransformersvLLM / TGI前者调试友好后者高并发向量检索FAISSMilvus / Qdrant数据量小于百万用FAISS足够服务框架FastAPIRay Serve单模型用FastAPI多模型编排用Ray缓存内存字典Redis单机用字典分布式必须Redis监控打印日志Prometheus Grafana上线前必须补齐2.3 目录结构一个能长大的项目长什么样我见过太多AI项目最后变成一坨脚本根源就是一开始没规划目录。下面这个结构是我迭代了四五个项目后固定下来的你可以直接拿去用ai-engineering/ ├── configs/ # 所有配置按环境分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── src/ │ ├── models/ # 模型加载与推理封装 │ ├── services/ # 业务逻辑 │ ├── adapters/ # 外部依赖适配层 │ ├── pipelines/ # 编排流程 │ └── utils/ # 通用工具 ├── data/ │ ├── raw/ # 原始数据 │ ├── processed/ # 处理后数据 │ └── cache/ # 缓存 ├── tests/ │ ├── unit/ │ └── integration/ ├── scripts/ # 运维脚本 └── notebooks/ # 实验记录关键点在于adapters这一层。所有对模型、数据库、第三方API的调用都放这里业务层永远不直接import外部库。这样做的代价是多写一层代码收益是换任何组件时业务层零改动。我实测下来这个习惯至少帮我省了三次大规模重构。3. 核心细节拆解模型服务层到底该怎么写3.1 模型加载别小看这几行代码模型加载看起来简单但坑特别多。最常见的错误是每次请求都重新加载模型显存直接爆炸。正确做法是全局加载一次常驻内存。# src/models/loader.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer class ModelLoader: _instance None def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def load(self, model_path, devicecuda, dtypetorch.float16): self.tokenizer AutoTokenizer.from_pretrained(model_path) self.model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypedtype, device_mapdevice, ) self.model.eval() return self这里用单例模式保证模型只加载一次。torch_dtype选float16是权衡float32精度高但显存翻倍int8省显存但可能掉点。我一般先用float16跑通显存不够再考虑量化。注意device_map参数在单卡和多卡下行为不同。单卡直接写cuda多卡写auto让框架自动分配。写错了不会报错但会静默走CPU速度慢十倍。3.2 推理封装批处理是性能的分水岭单条推理和批量推理的性能差距实测能到5到10倍。原因是GPU的并行能力在单条请求下完全浪费了。所以推理层必须支持动态批处理。# src/models/inference.py import time from typing import List class InferenceEngine: def __init__(self, loader, max_batch_size8, max_wait_ms50): self.loader loader self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue [] def generate(self, prompts: List[str], max_new_tokens256): inputs self.loader.tokenizer( prompts, return_tensorspt, paddingTrue, truncationTrue ).to(self.loader.model.device) with torch.no_grad(): outputs self.loader.model.generate( **inputs, max_new_tokensmax_new_tokens, do_sampleTrue, temperature0.7, top_p0.9, ) return self.loader.tokenizer.batch_decode( outputs, skip_special_tokensTrue )max_batch_size和max_wait_ms这两个参数需要根据你的场景调。延迟敏感的场景把max_wait_ms调到10以内吞吐优先的场景可以放到100。我一般从batch_size8, wait50ms起步然后压测调优。3.3 缓存策略省下的都是真金白银AI推理的成本大头在算力而很多请求其实是重复的。缓存做得好成本能降一半以上。缓存分三层精确缓存请求完全一致时直接返回。用请求的hash做key存Redis。语义缓存请求语义相似时返回近似结果。用向量检索找最相似的缓存项相似度超过阈值就复用。前缀缓存多个请求共享相同前缀时复用KV Cache。这个需要推理框架支持vLLM有内置。精确缓存实现最简单收益也最直接import hashlib import json import redis class ExactCache: def __init__(self, redis_client, ttl3600): self.redis redis_client self.ttl ttl def _key(self, prompt, params): raw json.dumps({p: prompt, params: params}, sort_keysTrue) return ai:cache: hashlib.sha256(raw.encode()).hexdigest() def get(self, prompt, params): return self.redis.get(self._key(prompt, params)) def set(self, prompt, params, result): self.redis.setex(self._key(prompt, params), self.ttl, result)提示缓存key一定要包含所有影响输出的参数比如temperature、max_tokens。我踩过一次坑只用了prompt做key结果改了temperature后返回的还是旧结果排查了半天。3.4 参数调优temperature和top_p到底怎么设这两个参数是新手最容易懵的。用生活化的类比模型生成每个词时会给所有候选词打分。temperature控制敢不敢选冷门词top_p控制从多少候选里选。temperature0永远选最高分的词输出稳定但死板。适合分类、抽取类任务。temperature0.7适度随机输出自然。适合对话、创作。temperature1.5非常随机容易胡说。除非做头脑风暴否则别用。top_p一般设0.9到0.95意思是只从累计概率前90%的词里选。设太低会限制表达设太高等于没限制。我的经验配置任务类型temperaturetop_pmax_tokens信息抽取0.10.9512对话问答0.70.91024创意写作0.90.952048代码生成0.20.9520484. 实操全流程从零跑通一个AI服务4.1 环境准备与依赖安装先把基础环境搭起来。我推荐用conda隔离环境避免依赖冲突conda create -n ai-eng python3.10 -y conda activate ai-eng pip install torch transformers fastapi uvicorn redis pyyaml版本选择上Python 3.10是目前兼容性最好的3.11有些库还没跟上。PyTorch装之前先确认CUDA版本用nvidia-smi看驱动支持的CUDA版本然后去官网找对应命令。装错了不会报错但会用CPU跑速度差几十倍。4.2 配置文件设计把魔法数字赶出代码所有可变参数都放配置文件代码里只留逻辑。这是我从无数次改个参数要重新部署的痛苦中总结的# configs/base.yaml model: path: your-model-path device: cuda dtype: float16 max_batch_size: 8 max_wait_ms: 50 inference: default_max_new_tokens: 512 default_temperature: 0.7 default_top_p: 0.9 cache: enabled: true redis_host: localhost redis_port: 6379 ttl: 3600 server: host: 0.0.0.0 port: 8000 workers: 1加载配置的代码import yaml def load_config(envdev): with open(fconfigs/base.yaml) as f: base yaml.safe_load(f) with open(fconfigs/{env}.yaml) as f: override yaml.safe_load(f) return deep_merge(base, override)deep_merge是个递归合并字典的小工具让环境配置只写差异部分避免重复。4.3 服务接口实现FastAPI三件套服务层用FastAPI核心就三个接口健康检查、单条推理、批量推理。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List app FastAPI() engine None cache None class GenerateRequest(BaseModel): prompt: str max_new_tokens: int 512 temperature: float 0.7 top_p: float 0.9 class BatchRequest(BaseModel): prompts: List[str] max_new_tokens: int 512 app.get(/health) def health(): return {status: ok, model_loaded: engine is not None} app.post(/generate) def generate(req: GenerateRequest): params req.dict() cached cache.get(req.prompt, params) if cached: return {result: cached, cached: True} result engine.generate([req.prompt], req.max_new_tokens)[0] cache.set(req.prompt, params, result) return {result: result, cached: False} app.post(/batch_generate) def batch_generate(req: BatchRequest): if len(req.prompts) 32: raise HTTPException(400, batch too large) results engine.generate(req.prompts, req.max_new_tokens) return {results: results}启动命令uvicorn src.main:app --host 0.0.0.0 --port 8000 --workers 1注意workers千万别设大于1。因为模型是单例加载在进程内存里的多worker会导致每个进程都加载一份模型显存直接爆。要提并发就靠批处理不是靠多进程。4.4 压测与调优用数据说话服务跑起来后必须压测。我用的是locust写个简单的压测脚本from locust import HttpUser, task, between class AIUser(HttpUser): wait_time between(0.1, 0.5) task def generate(self): self.client.post(/generate, json{ prompt: 写一段产品介绍, max_new_tokens: 128 })跑起来后重点看三个指标QPS、P99延迟、显存占用。我实测的一组数据供参考单卡24G7B模型float16批大小QPSP99延迟显存占用13.2380ms15G49.8520ms17G814.5780ms19G1616.21400msOOM可以看到批大小到8之后QPS增长放缓延迟却涨得快。所以8是个甜点值。这个数字因模型和硬件而异你必须自己压一遍。5. 常见问题与排查技巧实录5.1 显存不够用怎么办这是最高频的问题。排查顺序是先看是不是多进程加载了模型再看批大小是不是太大最后考虑量化。# 查看显存占用 nvidia-smi # 查看具体进程 nvidia-smi --query-compute-appspid,used_memory --formatcsv如果确认是模型本身太大有几个降显存的手段按推荐顺序换float16从float32换过来直接省一半。开梯度检查点训练时用推理用不上。int8量化省一半显存精度掉1到2个点。int4量化省四分之三精度掉得明显慎用。CPU offload把部分层放CPU速度慢但能跑。5.2 输出结果不稳定怎么破同一个prompt两次输出不一样这是采样导致的正常。但如果差异大到影响业务就要处理把temperature调到0.1以下接近确定性输出。加seed参数固定随机种子。在Prompt里加约束比如只输出JSON不要解释。后处理做格式校验不合格就重试。我一般会写个重试装饰器def retry_on_invalid(max_retries3): def decorator(func): def wrapper(*args, **kwargs): for i in range(max_retries): result func(*args, **kwargs) if validate(result): return result raise ValueError(max retries exceeded) return wrapper return decorator5.3 排查速查表现象可能原因排查方法解决服务启动慢模型加载耗时看启动日志时间戳预热或异步加载首次请求慢冷启动对比首次和后续延迟启动时跑一次空推理显存OOM批太大或多进程nvidia-smi降批大小或改单worker输出乱码tokenizer不匹配检查模型和tokenizer路径用同一路径加载缓存不生效key设计问题打印key对比确保参数全进key并发上不去单worker阻塞看CPU利用率上批处理或异步5.4 几个我踩过的坑坑一tokenizer和模型版本不一致。有次我从两个地方分别下载模型和tokenizer结果词表对不上输出全是乱码。后来养成习惯永远从同一个路径加载。坑二忘了设eval模式。训练完直接推理忘了model.eval()dropout还在生效输出随机性大增。这个bug找了我一整天。坑三缓存没设过期。早期缓存永久有效结果模型更新后旧结果还在返回。现在所有缓存都带TTL最长一小时。坑四日志打太多。一开始把每个请求的完整prompt和输出都打日志磁盘一周就满了。现在只打摘要和指标详细内容按需开启。6. 可观测性上线前必须补齐的一环6.1 三个必须埋的指标AI服务的监控和普通服务不一样除了QPS、延迟、错误率还要关注Token吞吐每秒处理多少token这是成本的核心指标。缓存命中率命中率低于30%说明缓存策略有问题。输出长度分布突然变长可能是模型跑偏了。埋点代码import time from prometheus_client import Counter, Histogram REQUEST_COUNT Counter(ai_requests_total, Total requests, [status]) LATENCY Histogram(ai_latency_seconds, Request latency) TOKENS Counter(ai_tokens_total, Total tokens generated) def track(func): def wrapper(*args, **kwargs): start time.time() try: result func(*args, **kwargs) REQUEST_COUNT.labels(statussuccess).inc() TOKENS.inc(len(result.split())) return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise finally: LATENCY.observe(time.time() - start) return wrapper6.2 日志怎么打才有用日志不是越多越好关键是要能回答这次请求为什么慢/为什么错。我的日志格式固定包含请求ID、prompt摘要、参数、耗时、token数、缓存状态。import logging import uuid logger logging.getLogger(ai) def log_request(prompt, params, latency, tokens, cached): logger.info({ req_id: str(uuid.uuid4())[:8], prompt_preview: prompt[:50], params: params, latency_ms: round(latency * 1000), tokens: tokens, cached: cached, })用JSON格式打日志方便后续用工具解析。别用print也别用字符串拼接结构化日志能省你大量排查时间。6.3 效果评估怎么知道模型变好了还是变坏了这是AI工程最容易被忽略的部分。模型更新后怎么判断效果我的做法是维护一个小型评估集每次更新都跑一遍。评估集不用大100到200条就够但要覆盖典型场景和边界情况。每条包含输入和期望输出用规则或人工打分。关键指标准确率输出符合期望的比例。一致性同一输入多次输出的一致程度。格式合规率输出符合格式要求的比例。def evaluate(engine, eval_set): results {correct: 0, total: 0, format_ok: 0} for item in eval_set: output engine.generate([item[input]])[0] results[total] 1 if matches(output, item[expected]): results[correct] 1 if is_valid_format(output): results[format_ok] 1 return { accuracy: results[correct] / results[total], format_rate: results[format_ok] / results[total], }这套评估跑一次大概几分钟但能帮你避免感觉变好了其实变差了的误判。我吃过一次亏换了个新模型觉得输出更流畅结果评估集一跑准确率掉了8个点赶紧回滚。7. 成本控制AI工程绕不开的现实问题7.1 算清楚每一分钱花在哪AI服务的成本主要三块算力、存储、带宽。算力是大头尤其推理。我习惯给每个请求算一个成本分成本分 输入token数 * 单价 输出token数 * 单价 固定开销固定开销包括模型常驻显存的成本按小时摊到每个请求。这样每个请求的成本一目了然也方便做预算。7.2 降本的四个手段按性价比排序缓存命中率每提高10%成本降约8%。投入最小收益最大。批处理吞吐提升直接摊薄固定成本。量化int8量化能省一半显存等于同样硬件跑两倍模型。模型降级简单任务用小模型复杂任务才用大模型。这个需要做任务分类。我做过一个实验同样一批请求加上缓存和批处理后成本降了约60%。所以别急着换更便宜的硬件先把软件层优化做透。7.3 预算告警成本控制的关键是早知道。我设了三档告警日预算的50%、80%、100%。到80%就该查是不是有异常流量到100%自动降级到小模型。def check_budget(daily_cost, budget): ratio daily_cost / budget if ratio 1.0: return critical elif ratio 0.8: return warning elif ratio 0.5: return notice return ok这套机制帮我抓到过一次爬虫刷接口一天跑了平时一周的量告警及时止损。8. 写在最后一些个人体会做AI工程这两年我最大的感受是别把模型当黑盒但也别试图完全理解它。你需要知道它的能力边界、成本特性、失败模式但不需要懂每一层的数学推导。工程的价值在于把不确定性管起来让业务能稳定地用上AI能力。from-scratch的意义不在于什么都自己造而在于你清楚每一层在做什么、为什么这么选、出问题去哪查。这套体系我迭代了好几版从最初的几十行脚本到现在能扛住日均百万请求中间踩的坑比写的代码多得多。如果你刚开始搭我的建议是先跑通最小闭环一个模型、一个接口、一个缓存然后逐步加监控、加评估、加成本控制。别一上来就追求架构完美能跑起来、能观测、能回滚比什么都重要。最后分享一个小技巧每次改动前先跑一遍评估集记录基线。改动后再跑一遍对比数据。这个习惯能帮你把感觉变成事实少走很多弯路。