1. 从零搭建AI工程能力为什么我劝你别一上来就啃论文这两年“AI工程师”这个岗位被炒得火热招聘JD上动不动就是“熟悉Transformer、有LLM微调经验、掌握RAG架构”。很多人一看就慌了转头去啃《Attention Is All You Need》结果公式推导看了三遍连一个能跑起来的推理服务都没搭出来。我自己带过几个刚入行的同学也见过不少转岗的朋友最大的误区就是把“AI工程”等同于“AI研究”。这两件事的差别比“会开车”和“会造发动机”的差别还大。ai-engineering-from-scratch这个标题核心讲的其实是一件事如何从工程视角而不是学术视角一步步把AI能力落地成可运行、可维护、可扩展的系统。它解决的不是“模型为什么有效”的问题而是“模型怎么跑起来、怎么接业务、怎么扛住流量、怎么持续迭代”的问题。适合谁看适合那些已经会写Python、懂基本后端开发但面对AI项目不知道从哪下手的人也适合已经在做AI应用、但总觉得自己的系统“能跑但不敢上线”的工程师。我自己的经验是AI工程能力可以拆成四层环境与工具链、模型调用与推理、数据管道与检索、服务化与运维。这四层缺一层系统就是瘸的。下面我就按这个顺序把每一层里最容易踩坑的地方、最值得抄的配置、最容易被忽略的细节全部摊开讲一遍。你不需要先成为算法专家但你需要成为一个能把算法“用起来”的工程师。2. 环境与工具链别让配环境吃掉你三天时间2.1 为什么我坚持用uv而不是pip刚入门的人最容易在环境上翻车。我见过一个同学光装PyTorch就折腾了两天最后发现是CUDA版本和驱动对不上。这里我给一个非常明确的建议用uv做Python包管理用conda做CUDA环境隔离两者分工明确。uv是这两年崛起的包管理器速度比pip快一个数量级而且它自带虚拟环境管理。你不需要再单独装virtualenv也不需要记source activate那一套。安装uv只需要一行curl -LsSf https://astral.sh/uv/install.sh | sh装完之后创建一个新项目uv init ai-eng-demo cd ai-eng-demo uv venv --python 3.11 source .venv/bin/activate为什么强调Python 3.11因为3.12对部分AI库的兼容性还不稳定3.10又缺少一些新特性。3.11是目前生态最稳的版本实测下来torch、transformers、vllm都能正常跑。至于CUDA我建议用conda单独建一个环境因为conda能帮你把cudatoolkit、cudnn这些底层库一次性装好不用自己手动配LD_LIBRARY_PATH。命令如下conda create -n cuda-env python3.11 conda activate cuda-env conda install pytorch torchvision torchaudio pytorch-cuda12.1 -c pytorch -c nvidia注意conda环境和uv环境不要混用。我的做法是conda只管CUDA和PyTorch其他纯Python依赖全部用uv装。这样职责清晰出问题好排查。2.2 目录结构决定你后期改代码的痛苦程度很多人写AI项目所有代码堆在一个文件夹里train.py、inference.py、utils.py混在一起。等到要加一个检索功能发现import路径全乱了。我从实际项目里总结出一个最小可用的目录结构你可以直接抄ai-eng-demo/ ├── configs/ # 配置文件yaml格式 │ ├── model.yaml │ └── service.yaml ├── src/ │ ├── data/ # 数据加载、清洗、切分 │ ├── models/ # 模型定义、加载、推理封装 │ ├── retrieval/ # 检索相关向量库、embedding │ ├── service/ # API服务FastAPI路由 │ └── utils/ # 日志、监控、通用工具 ├── tests/ # 单元测试 ├── scripts/ # 一次性脚本数据预处理等 ├── pyproject.toml # uv管理的依赖 └── README.md这个结构的好处是每一层职责单一依赖方向清晰。service层可以importmodels和retrieval但反过来不行。这样你后期想把模型从本地换成API调用只需要改models层service层完全不用动。2.3 配置文件别硬编码用YAML加环境变量我见过太多项目把模型路径、API密钥、超时时间直接写在代码里。一旦要换环境就得改代码重新部署。正确做法是用YAML管结构用环境变量管敏感信息。比如configs/model.yamlmodel: name: bge-small-zh path: ${MODEL_PATH:-./models/bge-small-zh} device: cuda max_length: 512 batch_size: 32然后在代码里用os.environ读取配合pydantic做校验。这样本地开发时用默认路径线上部署时通过环境变量覆盖不用改一行代码。实操心得我习惯在configs里放一个local.yaml和prod.yaml通过ENVprod来切换。这样配置差异一目了然不会出现“本地能跑线上挂”的情况。3. 模型调用与推理从“能跑”到“跑得稳”的关键细节3.1 本地推理和API调用的选择逻辑很多人一上来就问“我该用本地模型还是调API”。这个问题没有标准答案但有一个判断框架看你的数据敏感度、调用频率、延迟要求和预算。如果数据不能出内网那必须本地部署。如果调用频率低、延迟要求不严、预算充足那调API更省事。我自己的做法是开发阶段用API快速验证生产阶段根据数据合规要求决定是否本地化。本地推理目前最稳的方案是vllm它支持连续批处理吞吐量比HuggingFace的pipeline高好几倍。安装uv add vllm启动一个OpenAI兼容的服务python -m vllm.entrypoints.openai.api_server \ --model ./models/Qwen2.5-7B-Instruct \ --dtype auto \ --max-model-len 8192 \ --gpu-memory-utilization 0.9这里有几个参数值得解释。--dtype auto让vllm自动选择float16还是bfloat16省得你手动试。--max-model-len控制上下文长度设太大显存扛不住设太小长文本会被截断。--gpu-memory-utilization 0.9表示用90%的显存留10%给系统避免OOM。注意vllm启动时会预分配显存如果你同时跑多个模型一定要算好总显存。我试过在一张24G的卡上同时跑一个7B模型和一个embedding模型结果第二个直接OOM。后来改成embedding用CPU跑才稳住。3.2 推理封装的三个必备能力不管你用本地模型还是API推理层必须封装三个能力重试、超时、降级。这三个能力不做好线上就是定时炸弹。重试的逻辑是遇到网络抖动或服务暂时不可用自动重试2到3次每次间隔指数退避。超时的逻辑是设置一个合理的超时时间比如30秒超过就放弃。降级的逻辑是主模型不可用时自动切到备用模型或返回缓存结果。我用tenacity做重试代码大概长这样from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_model(prompt: str) - str: response client.chat.completions.create( modelqwen2.5-7b, messages[{role: user, content: prompt}], timeout30 ) return response.choices[0].message.content这段代码的意思是最多重试3次第一次等2秒第二次等4秒第三次等8秒。为什么用指数退避因为如果服务是过载导致的失败你立刻重试只会加重过载。等一会儿再试成功率更高。3.3 批处理与流式输出的取舍批处理能提高吞吐流式输出能降低首字延迟。这两个怎么选我的经验是离线任务用批处理在线对话用流式。批处理的实现很简单把多个请求攒成一个batch一次性送给模型。vllm的LLM.generate支持传入一个prompt列表返回一个结果列表。但要注意batch size不是越大越好。我实测下来batch size超过32之后吞吐提升就不明显了但显存占用线性增长。所以32是一个比较安全的默认值。流式输出用streamTrue然后逐块读取。这里有个坑流式输出时如果客户端断开连接服务端要能感知并停止生成。否则模型会一直跑下去浪费算力。FastAPI里可以用request.is_disconnected()来检测。from fastapi import Request from fastapi.responses import StreamingResponse app.post(/chat) async def chat(request: Request, body: ChatRequest): async def generate(): async for chunk in model.stream(body.prompt): if await request.is_disconnected(): break yield chunk return StreamingResponse(generate(), media_typetext/event-stream)实操心得流式输出一定要加心跳。如果模型生成很慢客户端可能以为连接断了。我习惯每5秒发一个空行作为心跳保持连接活跃。4. 数据管道与检索RAG系统的命脉在这里4.1 文档切分的颗粒度怎么定RAG系统里文档切分是最容易被忽视但影响最大的环节。切得太碎检索出来的片段缺乏上下文切得太粗检索精度下降。我的经验是中文文档按300到500字切分英文按200到300词切分重叠50字。为什么要有重叠因为一句话可能被切断重叠能保证语义完整。比如“AI工程的核心是落地”这句话如果正好在“核心”后面切断检索“AI工程落地”时就匹配不上。重叠50字能解决大部分这类问题。切分工具我推荐langchain-text-splitters它支持按字符、按token、按递归分割。递归分割最智能它会先按段落切段落太长再按句子切句子太长再按字符切。配置如下from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n\n, \n, 。, , , , , ] )注意separators的顺序中文标点要放在英文标点前面否则中文句子会被错误切分。4.2 向量库选型Chroma、Milvus还是Qdrant向量库的选择取决于数据量和部署复杂度。我列一个对比表向量库适用数据量部署复杂度特点Chroma10万条以下极低pip装完就能用适合原型验证Qdrant100万到1000万中等需要Docker过滤功能强性能好Milvus1000万以上高需要集群分布式适合大规模我自己的项目里原型阶段用Chroma上线后切到Qdrant。Chroma的API极其简单import chromadb client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(docs) collection.add(documentschunks, ids[str(i) for i in range(len(chunks))]) results collection.query(query_texts[AI工程], n_results5)但Chroma的过滤能力弱如果你需要按时间、按标签过滤就得换Qdrant。Qdrant的过滤语法更灵活而且支持payload索引查询速度快很多。注意向量库的维度必须和embedding模型一致。比如bge-small-zh是512维text-embedding-3-small是1536维。换模型时一定要重建索引否则查询结果全是乱的。4.3 检索策略向量检索不够还得加关键词纯向量检索有个问题对专有名词和数字不敏感。比如你搜“Qwen2.5”向量检索可能返回一堆“Qwen”相关的文档但精确匹配不到“Qwen2.5”。解决办法是混合检索向量检索加BM25关键词检索然后融合排序。融合排序最简单的方法是RRFReciprocal Rank Fusion公式是score sum(1 / (k rank_i))其中k通常取60rank_i是文档在第i路检索中的排名。RRF的好处是不需要调权重两路检索的分数直接融合。def rrf_fusion(vector_results, bm25_results, k60): scores {} for rank, doc_id in enumerate(vector_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) for rank, doc_id in enumerate(bm25_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)实测下来混合检索比纯向量检索的召回率高15%到20%尤其是对技术文档和产品手册这类专有名词多的场景。5. 服务化与运维让系统敢上线的最后一步5.1 FastAPI的异步陷阱FastAPI是AI服务最常用的框架但很多人用错了异步。最常见的错误是在async def路由里调用同步的阻塞函数比如requests.get或model.generate。这会导致整个事件循环被阻塞并发能力直接归零。正确做法是阻塞操作放到线程池里跑。用run_in_executor或者anyio.to_thread.run_syncimport anyio from fastapi import FastAPI app FastAPI() app.post(/infer) async def infer(body: InferRequest): result await anyio.to_thread.run_sync(model.generate, body.prompt) return {result: result}这样模型推理在线程池里跑事件循环不被阻塞其他请求还能正常处理。实操心得我见过一个项目QPS上不去排查了半天发现是日志写文件用了同步IO。改成异步日志后QPS直接翻倍。所以任何IO操作都要检查是不是阻塞的。5.2 监控指标别只看QPSAI服务的监控和普通Web服务不一样。除了QPS、延迟、错误率还要看token吞吐量、显存占用、队列长度。这几个指标能提前预警。token吞吐量突然下降可能是模型遇到了长文本生成变慢。显存占用持续上涨可能是内存泄漏需要重启。队列长度超过阈值说明服务过载要扩容或限流。我用prometheus-client暴露指标from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT Counter(ai_requests_total, Total requests, [endpoint, status]) LATENCY Histogram(ai_latency_seconds, Request latency, [endpoint]) GPU_MEMORY Gauge(ai_gpu_memory_bytes, GPU memory usage)然后在推理前后打点。Grafana面板上把这三个指标放在一起看基本能判断服务健康度。5.3 限流与降级保护自己不被流量打死AI服务的特点是单次请求消耗大一个长文本推理可能占用几秒的GPU时间。如果不限流几个并发请求就能把服务打满。限流我推荐用slowapi基于令牌桶算法from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.post(/chat) limiter.limit(10/minute) async def chat(request: Request, body: ChatRequest): ...这个配置表示每个IP每分钟最多10次请求。超过就返回429。降级策略是当GPU显存不足或队列过长时自动切到小模型或返回缓存。我习惯在服务里维护一个fallback_model主模型不可用时自动切换。切换逻辑要记录日志方便事后分析。注意限流阈值要根据实际压测结果来定。我一般先用locust压测找到服务能稳定支撑的最大QPS然后限流阈值设为这个值的80%留20%的余量。6. 常见问题与排查技巧实录6.1 模型加载慢、首次推理慢怎么办这是新手最常问的问题。模型加载慢是因为要从磁盘读权重首次推理慢是因为要初始化CUDA kernel。解决办法有两个预热和缓存。预热是在服务启动后立刻跑一次推理把CUDA kernel初始化好。这样第一个真实请求就不会慢。代码很简单app.on_event(startup) async def warmup(): model.generate(warmup)缓存是把模型权重放到内存文件系统比如/dev/shm读取速度比磁盘快很多。但要注意/dev/shm的大小限制默认是内存的一半。6.2 检索结果不相关怎么调RAG系统检索不准通常有三个原因切分粒度不对、embedding模型不匹配、检索策略单一。排查顺序是先看切分后的片段是否语义完整再看embedding模型是否适合中文最后看是否加了混合检索。我遇到过一个案例用户搜“如何退款”检索出来的全是“退款政策”的文档但用户想要的是“退款操作步骤”。原因是切分时把操作步骤和 policy 混在一起了。后来改成按标题切分每个标题下的内容单独成块问题就解决了。6.3 显存泄漏怎么排查显存泄漏的表现是服务跑一段时间后OOM。排查方法是在每次推理前后打印torch.cuda.memory_allocated()看是否持续增长。如果增长说明有张量没释放。常见原因是把中间结果存到了全局变量或者用了torch.no_grad()但没加del。解决办法是推理函数里用with torch.no_grad():结束后手动del中间变量并调用torch.cuda.empty_cache()。实操心得我习惯在服务里加一个定时任务每处理1000个请求就调一次empty_cache()。虽然会稍微降低性能但能有效防止显存碎片化导致的OOM。6.4 常见问题速查表问题现象可能原因排查方法解决方案服务启动慢模型加载耗时看启动日志时间戳预热、权重放内存盘首次推理慢CUDA kernel初始化对比首次和后续延迟启动时跑一次warmup检索不准切分粒度或embedding问题人工检查检索片段调整切分、换embedding、加混合检索显存持续增长张量未释放打印memory_allocatedno_grad、del、empty_cacheQPS上不去阻塞IO或限流过严看事件循环延迟异步化、调整限流阈值流式输出中断客户端断开未检测看服务端日志is_disconnected检测7. 我踩过的坑和给你的三条建议第一条建议不要追求一步到位。我见过太多人想一次性把RAG、微调、Agent全做完结果哪个都没做好。正确的做法是先跑通一个最小闭环一个模型、一个向量库、一个API能回答一个问题就行。然后再逐步加检索、加缓存、加监控。第二条建议日志要打全但别打敏感信息。AI服务的日志里经常包含用户输入和模型输出这些可能含隐私。我习惯在日志里只打请求ID、耗时、token数不打具体内容。需要调试时用请求ID去查专门的调试日志。第三条建议压测要趁早。很多人等到上线前才压测结果发现一堆问题来不及改。我的做法是服务能跑通后就立刻压测用locust模拟10个并发用户看延迟和错误率。如果10个并发就扛不住那说明架构有问题早发现早改。最后分享一个小技巧用nvidia-smi的--query-gpu参数做实时监控比看默认输出清晰得多。nvidia-smi --query-gpuutilization.gpu,memory.used,memory.total --formatcsv -l 1这个命令每秒刷新一次输出GPU利用率、已用显存、总显存。配合watch命令可以一直挂在终端里看。我调试推理性能时这个命令基本不离手。AI工程这个方向说到底就是把不确定性管起来。模型输出不确定那就加校验和降级流量不确定那就加限流和弹性数据不确定那就加检索和过滤。你不需要成为算法专家但你需要成为那个能让系统稳定跑起来的人。这条路我走了好几年踩过的坑比写过的代码还多但每填一个坑系统就稳一分。希望这些经验能帮你少走点弯路。
AI工程从零搭建：避开论文陷阱，掌握落地四层能力
1. 从零搭建AI工程能力为什么我劝你别一上来就啃论文这两年“AI工程师”这个岗位被炒得火热招聘JD上动不动就是“熟悉Transformer、有LLM微调经验、掌握RAG架构”。很多人一看就慌了转头去啃《Attention Is All You Need》结果公式推导看了三遍连一个能跑起来的推理服务都没搭出来。我自己带过几个刚入行的同学也见过不少转岗的朋友最大的误区就是把“AI工程”等同于“AI研究”。这两件事的差别比“会开车”和“会造发动机”的差别还大。ai-engineering-from-scratch这个标题核心讲的其实是一件事如何从工程视角而不是学术视角一步步把AI能力落地成可运行、可维护、可扩展的系统。它解决的不是“模型为什么有效”的问题而是“模型怎么跑起来、怎么接业务、怎么扛住流量、怎么持续迭代”的问题。适合谁看适合那些已经会写Python、懂基本后端开发但面对AI项目不知道从哪下手的人也适合已经在做AI应用、但总觉得自己的系统“能跑但不敢上线”的工程师。我自己的经验是AI工程能力可以拆成四层环境与工具链、模型调用与推理、数据管道与检索、服务化与运维。这四层缺一层系统就是瘸的。下面我就按这个顺序把每一层里最容易踩坑的地方、最值得抄的配置、最容易被忽略的细节全部摊开讲一遍。你不需要先成为算法专家但你需要成为一个能把算法“用起来”的工程师。2. 环境与工具链别让配环境吃掉你三天时间2.1 为什么我坚持用uv而不是pip刚入门的人最容易在环境上翻车。我见过一个同学光装PyTorch就折腾了两天最后发现是CUDA版本和驱动对不上。这里我给一个非常明确的建议用uv做Python包管理用conda做CUDA环境隔离两者分工明确。uv是这两年崛起的包管理器速度比pip快一个数量级而且它自带虚拟环境管理。你不需要再单独装virtualenv也不需要记source activate那一套。安装uv只需要一行curl -LsSf https://astral.sh/uv/install.sh | sh装完之后创建一个新项目uv init ai-eng-demo cd ai-eng-demo uv venv --python 3.11 source .venv/bin/activate为什么强调Python 3.11因为3.12对部分AI库的兼容性还不稳定3.10又缺少一些新特性。3.11是目前生态最稳的版本实测下来torch、transformers、vllm都能正常跑。至于CUDA我建议用conda单独建一个环境因为conda能帮你把cudatoolkit、cudnn这些底层库一次性装好不用自己手动配LD_LIBRARY_PATH。命令如下conda create -n cuda-env python3.11 conda activate cuda-env conda install pytorch torchvision torchaudio pytorch-cuda12.1 -c pytorch -c nvidia注意conda环境和uv环境不要混用。我的做法是conda只管CUDA和PyTorch其他纯Python依赖全部用uv装。这样职责清晰出问题好排查。2.2 目录结构决定你后期改代码的痛苦程度很多人写AI项目所有代码堆在一个文件夹里train.py、inference.py、utils.py混在一起。等到要加一个检索功能发现import路径全乱了。我从实际项目里总结出一个最小可用的目录结构你可以直接抄ai-eng-demo/ ├── configs/ # 配置文件yaml格式 │ ├── model.yaml │ └── service.yaml ├── src/ │ ├── data/ # 数据加载、清洗、切分 │ ├── models/ # 模型定义、加载、推理封装 │ ├── retrieval/ # 检索相关向量库、embedding │ ├── service/ # API服务FastAPI路由 │ └── utils/ # 日志、监控、通用工具 ├── tests/ # 单元测试 ├── scripts/ # 一次性脚本数据预处理等 ├── pyproject.toml # uv管理的依赖 └── README.md这个结构的好处是每一层职责单一依赖方向清晰。service层可以importmodels和retrieval但反过来不行。这样你后期想把模型从本地换成API调用只需要改models层service层完全不用动。2.3 配置文件别硬编码用YAML加环境变量我见过太多项目把模型路径、API密钥、超时时间直接写在代码里。一旦要换环境就得改代码重新部署。正确做法是用YAML管结构用环境变量管敏感信息。比如configs/model.yamlmodel: name: bge-small-zh path: ${MODEL_PATH:-./models/bge-small-zh} device: cuda max_length: 512 batch_size: 32然后在代码里用os.environ读取配合pydantic做校验。这样本地开发时用默认路径线上部署时通过环境变量覆盖不用改一行代码。实操心得我习惯在configs里放一个local.yaml和prod.yaml通过ENVprod来切换。这样配置差异一目了然不会出现“本地能跑线上挂”的情况。3. 模型调用与推理从“能跑”到“跑得稳”的关键细节3.1 本地推理和API调用的选择逻辑很多人一上来就问“我该用本地模型还是调API”。这个问题没有标准答案但有一个判断框架看你的数据敏感度、调用频率、延迟要求和预算。如果数据不能出内网那必须本地部署。如果调用频率低、延迟要求不严、预算充足那调API更省事。我自己的做法是开发阶段用API快速验证生产阶段根据数据合规要求决定是否本地化。本地推理目前最稳的方案是vllm它支持连续批处理吞吐量比HuggingFace的pipeline高好几倍。安装uv add vllm启动一个OpenAI兼容的服务python -m vllm.entrypoints.openai.api_server \ --model ./models/Qwen2.5-7B-Instruct \ --dtype auto \ --max-model-len 8192 \ --gpu-memory-utilization 0.9这里有几个参数值得解释。--dtype auto让vllm自动选择float16还是bfloat16省得你手动试。--max-model-len控制上下文长度设太大显存扛不住设太小长文本会被截断。--gpu-memory-utilization 0.9表示用90%的显存留10%给系统避免OOM。注意vllm启动时会预分配显存如果你同时跑多个模型一定要算好总显存。我试过在一张24G的卡上同时跑一个7B模型和一个embedding模型结果第二个直接OOM。后来改成embedding用CPU跑才稳住。3.2 推理封装的三个必备能力不管你用本地模型还是API推理层必须封装三个能力重试、超时、降级。这三个能力不做好线上就是定时炸弹。重试的逻辑是遇到网络抖动或服务暂时不可用自动重试2到3次每次间隔指数退避。超时的逻辑是设置一个合理的超时时间比如30秒超过就放弃。降级的逻辑是主模型不可用时自动切到备用模型或返回缓存结果。我用tenacity做重试代码大概长这样from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_model(prompt: str) - str: response client.chat.completions.create( modelqwen2.5-7b, messages[{role: user, content: prompt}], timeout30 ) return response.choices[0].message.content这段代码的意思是最多重试3次第一次等2秒第二次等4秒第三次等8秒。为什么用指数退避因为如果服务是过载导致的失败你立刻重试只会加重过载。等一会儿再试成功率更高。3.3 批处理与流式输出的取舍批处理能提高吞吐流式输出能降低首字延迟。这两个怎么选我的经验是离线任务用批处理在线对话用流式。批处理的实现很简单把多个请求攒成一个batch一次性送给模型。vllm的LLM.generate支持传入一个prompt列表返回一个结果列表。但要注意batch size不是越大越好。我实测下来batch size超过32之后吞吐提升就不明显了但显存占用线性增长。所以32是一个比较安全的默认值。流式输出用streamTrue然后逐块读取。这里有个坑流式输出时如果客户端断开连接服务端要能感知并停止生成。否则模型会一直跑下去浪费算力。FastAPI里可以用request.is_disconnected()来检测。from fastapi import Request from fastapi.responses import StreamingResponse app.post(/chat) async def chat(request: Request, body: ChatRequest): async def generate(): async for chunk in model.stream(body.prompt): if await request.is_disconnected(): break yield chunk return StreamingResponse(generate(), media_typetext/event-stream)实操心得流式输出一定要加心跳。如果模型生成很慢客户端可能以为连接断了。我习惯每5秒发一个空行作为心跳保持连接活跃。4. 数据管道与检索RAG系统的命脉在这里4.1 文档切分的颗粒度怎么定RAG系统里文档切分是最容易被忽视但影响最大的环节。切得太碎检索出来的片段缺乏上下文切得太粗检索精度下降。我的经验是中文文档按300到500字切分英文按200到300词切分重叠50字。为什么要有重叠因为一句话可能被切断重叠能保证语义完整。比如“AI工程的核心是落地”这句话如果正好在“核心”后面切断检索“AI工程落地”时就匹配不上。重叠50字能解决大部分这类问题。切分工具我推荐langchain-text-splitters它支持按字符、按token、按递归分割。递归分割最智能它会先按段落切段落太长再按句子切句子太长再按字符切。配置如下from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n\n, \n, 。, , , , , ] )注意separators的顺序中文标点要放在英文标点前面否则中文句子会被错误切分。4.2 向量库选型Chroma、Milvus还是Qdrant向量库的选择取决于数据量和部署复杂度。我列一个对比表向量库适用数据量部署复杂度特点Chroma10万条以下极低pip装完就能用适合原型验证Qdrant100万到1000万中等需要Docker过滤功能强性能好Milvus1000万以上高需要集群分布式适合大规模我自己的项目里原型阶段用Chroma上线后切到Qdrant。Chroma的API极其简单import chromadb client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(docs) collection.add(documentschunks, ids[str(i) for i in range(len(chunks))]) results collection.query(query_texts[AI工程], n_results5)但Chroma的过滤能力弱如果你需要按时间、按标签过滤就得换Qdrant。Qdrant的过滤语法更灵活而且支持payload索引查询速度快很多。注意向量库的维度必须和embedding模型一致。比如bge-small-zh是512维text-embedding-3-small是1536维。换模型时一定要重建索引否则查询结果全是乱的。4.3 检索策略向量检索不够还得加关键词纯向量检索有个问题对专有名词和数字不敏感。比如你搜“Qwen2.5”向量检索可能返回一堆“Qwen”相关的文档但精确匹配不到“Qwen2.5”。解决办法是混合检索向量检索加BM25关键词检索然后融合排序。融合排序最简单的方法是RRFReciprocal Rank Fusion公式是score sum(1 / (k rank_i))其中k通常取60rank_i是文档在第i路检索中的排名。RRF的好处是不需要调权重两路检索的分数直接融合。def rrf_fusion(vector_results, bm25_results, k60): scores {} for rank, doc_id in enumerate(vector_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) for rank, doc_id in enumerate(bm25_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)实测下来混合检索比纯向量检索的召回率高15%到20%尤其是对技术文档和产品手册这类专有名词多的场景。5. 服务化与运维让系统敢上线的最后一步5.1 FastAPI的异步陷阱FastAPI是AI服务最常用的框架但很多人用错了异步。最常见的错误是在async def路由里调用同步的阻塞函数比如requests.get或model.generate。这会导致整个事件循环被阻塞并发能力直接归零。正确做法是阻塞操作放到线程池里跑。用run_in_executor或者anyio.to_thread.run_syncimport anyio from fastapi import FastAPI app FastAPI() app.post(/infer) async def infer(body: InferRequest): result await anyio.to_thread.run_sync(model.generate, body.prompt) return {result: result}这样模型推理在线程池里跑事件循环不被阻塞其他请求还能正常处理。实操心得我见过一个项目QPS上不去排查了半天发现是日志写文件用了同步IO。改成异步日志后QPS直接翻倍。所以任何IO操作都要检查是不是阻塞的。5.2 监控指标别只看QPSAI服务的监控和普通Web服务不一样。除了QPS、延迟、错误率还要看token吞吐量、显存占用、队列长度。这几个指标能提前预警。token吞吐量突然下降可能是模型遇到了长文本生成变慢。显存占用持续上涨可能是内存泄漏需要重启。队列长度超过阈值说明服务过载要扩容或限流。我用prometheus-client暴露指标from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT Counter(ai_requests_total, Total requests, [endpoint, status]) LATENCY Histogram(ai_latency_seconds, Request latency, [endpoint]) GPU_MEMORY Gauge(ai_gpu_memory_bytes, GPU memory usage)然后在推理前后打点。Grafana面板上把这三个指标放在一起看基本能判断服务健康度。5.3 限流与降级保护自己不被流量打死AI服务的特点是单次请求消耗大一个长文本推理可能占用几秒的GPU时间。如果不限流几个并发请求就能把服务打满。限流我推荐用slowapi基于令牌桶算法from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.post(/chat) limiter.limit(10/minute) async def chat(request: Request, body: ChatRequest): ...这个配置表示每个IP每分钟最多10次请求。超过就返回429。降级策略是当GPU显存不足或队列过长时自动切到小模型或返回缓存。我习惯在服务里维护一个fallback_model主模型不可用时自动切换。切换逻辑要记录日志方便事后分析。注意限流阈值要根据实际压测结果来定。我一般先用locust压测找到服务能稳定支撑的最大QPS然后限流阈值设为这个值的80%留20%的余量。6. 常见问题与排查技巧实录6.1 模型加载慢、首次推理慢怎么办这是新手最常问的问题。模型加载慢是因为要从磁盘读权重首次推理慢是因为要初始化CUDA kernel。解决办法有两个预热和缓存。预热是在服务启动后立刻跑一次推理把CUDA kernel初始化好。这样第一个真实请求就不会慢。代码很简单app.on_event(startup) async def warmup(): model.generate(warmup)缓存是把模型权重放到内存文件系统比如/dev/shm读取速度比磁盘快很多。但要注意/dev/shm的大小限制默认是内存的一半。6.2 检索结果不相关怎么调RAG系统检索不准通常有三个原因切分粒度不对、embedding模型不匹配、检索策略单一。排查顺序是先看切分后的片段是否语义完整再看embedding模型是否适合中文最后看是否加了混合检索。我遇到过一个案例用户搜“如何退款”检索出来的全是“退款政策”的文档但用户想要的是“退款操作步骤”。原因是切分时把操作步骤和 policy 混在一起了。后来改成按标题切分每个标题下的内容单独成块问题就解决了。6.3 显存泄漏怎么排查显存泄漏的表现是服务跑一段时间后OOM。排查方法是在每次推理前后打印torch.cuda.memory_allocated()看是否持续增长。如果增长说明有张量没释放。常见原因是把中间结果存到了全局变量或者用了torch.no_grad()但没加del。解决办法是推理函数里用with torch.no_grad():结束后手动del中间变量并调用torch.cuda.empty_cache()。实操心得我习惯在服务里加一个定时任务每处理1000个请求就调一次empty_cache()。虽然会稍微降低性能但能有效防止显存碎片化导致的OOM。6.4 常见问题速查表问题现象可能原因排查方法解决方案服务启动慢模型加载耗时看启动日志时间戳预热、权重放内存盘首次推理慢CUDA kernel初始化对比首次和后续延迟启动时跑一次warmup检索不准切分粒度或embedding问题人工检查检索片段调整切分、换embedding、加混合检索显存持续增长张量未释放打印memory_allocatedno_grad、del、empty_cacheQPS上不去阻塞IO或限流过严看事件循环延迟异步化、调整限流阈值流式输出中断客户端断开未检测看服务端日志is_disconnected检测7. 我踩过的坑和给你的三条建议第一条建议不要追求一步到位。我见过太多人想一次性把RAG、微调、Agent全做完结果哪个都没做好。正确的做法是先跑通一个最小闭环一个模型、一个向量库、一个API能回答一个问题就行。然后再逐步加检索、加缓存、加监控。第二条建议日志要打全但别打敏感信息。AI服务的日志里经常包含用户输入和模型输出这些可能含隐私。我习惯在日志里只打请求ID、耗时、token数不打具体内容。需要调试时用请求ID去查专门的调试日志。第三条建议压测要趁早。很多人等到上线前才压测结果发现一堆问题来不及改。我的做法是服务能跑通后就立刻压测用locust模拟10个并发用户看延迟和错误率。如果10个并发就扛不住那说明架构有问题早发现早改。最后分享一个小技巧用nvidia-smi的--query-gpu参数做实时监控比看默认输出清晰得多。nvidia-smi --query-gpuutilization.gpu,memory.used,memory.total --formatcsv -l 1这个命令每秒刷新一次输出GPU利用率、已用显存、总显存。配合watch命令可以一直挂在终端里看。我调试推理性能时这个命令基本不离手。AI工程这个方向说到底就是把不确定性管起来。模型输出不确定那就加校验和降级流量不确定那就加限流和弹性数据不确定那就加检索和过滤。你不需要成为算法专家但你需要成为那个能让系统稳定跑起来的人。这条路我走了好几年踩过的坑比写过的代码还多但每填一个坑系统就稳一分。希望这些经验能帮你少走点弯路。
