尧图网页设计 
1. 从零开始做AI工程我到底在搭什么拿到“ai-engineering-from-scratch”这个题目很多人第一反应是“又要写一篇AI入门教程”。但如果你真在工业界待过几年就会明白这里的“from scratch”远不是跑通一个Jupyter Notebook那么简单。它意味着你要从一台裸机、一份Python环境、一个空目录开始把数据管线、模型服务、评估体系、监控告警、迭代流程一层层垒起来最终形成一个能稳定产出业务价值的AI系统。这篇文章不是什么大而全的手册更像是我把自己过去从零搭建AI工程项目踩过的坑、验证过的方案、推翻过的设计按一条可复现的主线串起来。内容适合两类人一类是刚接手AI工程化项目、需要对整体技术栈建立清晰认知的同学另一类是已经跑通过一些模型实验、但发现“离线能跑”和“线上稳定”之间隔着巨大鸿沟的工程师。读完之后你会对AI工程化需要哪些核心组件、每个组件解决什么问题、组件之间怎么衔接有一个系统性的判断而不是被碎片化的教程带着走。我最终落地的这套方案技术栈是Python 3.11 FastAPI PostgreSQL Redis Docker Compose模型层用ONNX Runtime做推理向量检索用pgvector任务队列用Celery监控用Prometheus Grafana。下面我按实际搭建顺序把每个环节的设计思路和实操细节逐一拆开讲。2. 工程化之前先把需求翻译成技术约束2.1 业务目标与技术方案的映射逻辑任何AI项目的第一件事都不是选框架而是把业务语言翻译成技术约束。比如“我们要做一个智能客服助手”这个描述里至少藏着三个关键约束响应时延要求多高直接影响是否能用流式输出、是否要上GPU推理、知识更新频率多快决定检索方案是静态索引还是实时向量化、错误容忍度多大决定是否需要人工审核兜底、是否需要置信度阈值。我习惯用一张表格把这类需求显式写出来避免后续设计跑偏业务需求技术约束设计选择回答准确率不低于90%需要评估集与回归测试搭建离线评估流水线每次模型更新自动跑基准首字响应小于500ms推理链路总耗时预算ONNX Runtime CPU量化模型避免GPU依赖知识每周更新一次知识库同步机制向量化任务入Celery队列错峰执行单日请求量峰值2万并发与限流策略FastAPI Gunicorn多workerRedis滑动窗口限流系统故障可定位日志与指标可观测结构化日志 Prometheus指标 告警规则这个环节很多新手会跳过直接开始写代码。但我在实际项目中吃过亏有一版智能问答系统所有评测指标都达标结果业务方反馈“回答太慢了”一查才发现我们把大量场景设计成了同步阻塞调用接口平均耗时1.8秒完全超出了客服坐席边聊天边等待的心理阈值。需求翻译这一步省掉的每一分钟都会在后续返工中加倍偿还。2.2 离线实验与线上系统的边界划定从零搭建时最容易犯的错误是把Notebook里的实验代码直接搬进生产服务。离线实验和线上系统至少有三条边界必须划定清楚数据边界——线下可以用全量历史数据训练线上只能看到截止当前时刻的数据资源边界——线下可以不计成本地调参跑实验线上必须考虑推理成本和时延评估边界——线下可以用准确率、F1这类离线指标衡量线上更关心的是用户留存、转化率、工单解决率这类业务指标。我的做法是维护两份独立的代码一份是experiments/目录下的研究代码自由度极高另一份是services/目录下的工程代码必须经过代码审查、测试覆盖、构建产物化。两者之间唯一的桥梁是模型产物和评估报告实验代码产出的模型经过评估后以版本化方式交给工程侧部署。这样既保留了研究阶段的灵活性又保证了生产环境的稳定性。3. 技术选型每一项选择背后的真实理由3.1 为什么是FastAPI而不是Flask或Django如果你去搜“Python Web框架选型”会看到大量对比文章。但AI工程场景下我选FastAPI的核心原因其实就两条一是原生异步支持LLM服务普遍是IO密集型等待模型推理结果、等待数据库查询async/await能把单机并发能力提升一个量级二是自动生成OpenAPI文档在前后端联调、接口对接时能省掉大量沟通成本。打个不那么准确的比方Flask像是手动挡汽车结构简单、什么都能自己控制但每个环节都要自己操作Django像是带了一整套生活用品的房车沉重但啥都有FastAPI像是自动挡的现代轿车该有的都有、日常开最顺手。对于AI服务这种“接口不多但并发要求高、迭代频繁”的场景FastAPI正好踩在最舒服的位置上。3.2 向量检索选了pgvector而不是独立向量数据库这个决定我犹豫了很久。刚开始我倾向于Milvus或Weaviate这类专用向量数据库因为它们功能全、性能强。后来之所以定pgvector核心原因是运维复杂度的控制。一个从零开始的AI项目如果同时要维护PostgreSQL、Redis、向量库、模型服务、任务队列任何一个小组件出问题都够折腾半天。用pgvector可以直接复用PostgreSQL的备份、恢复、权限体系少维护一个组件数据一致性也更好保证。它的性能到底够不够用我实测下来在单机PostgreSQL上pgvector对100万条768维向量的ANN检索使用IVFFlat索引单次查询延迟在10-30毫秒之间对绝大多数AI应用场景完全够用。只有当数据量到千万级以上、查询QPS非常高的时候才需要认真考虑独立向量数据库。对于从零起步的项目先用pgvector把业务跑通等量级上来再迁移是性价比最高的路径。3.3 模型部署为什么选了ONNX Runtime模型推理这块我见过太多团队一上来就搞TensorRT、搞vLLM结果发现工程复杂度远超预期。我自己一开始用的也是PyTorch直接加载模型做推理但很快遇到两个问题一是Python进程内存占用高多worker部署时显存和内存都吃紧二是模型部署环境需要完整安装PyTorch依赖镜像体积动辄几个GB。ONNX Runtime解决了这两个痛点模型从PyTorch导出为ONNX格式后推理时不再依赖PyTorch运行时镜像可以缩到几百MB同时ONNX Runtime对CPU推理做了大量优化配合int8量化在CPU上跑BERT类模型的延迟能做到原来的三分之一左右。当然ONNX导出过程有时候会遇到算子兼容问题这个后面我会专门讲几个典型坑。4. 环境准备从裸机到可复现的开发环境4.1 Python版本与依赖管理的坑从零搭建第一个要确定的就是Python版本。我推荐Python 3.11原因很简单它是目前兼容性、性能、生态三方平衡最好的版本。3.12虽然更新但部分深度学习库的预编译wheel还跟进得不完美3.10以下则逐渐进入维护末期没必要新项目踩旧版本。依赖管理方面我强烈建议直接上Poetry或uv而不是裸用requirements.txt。裸用requirements.txt最常见的灾难是开发环境装的是numpy1.24.3测试环境被某次pip install悄悄升级到了1.26然后模型推理结果发生了微妙变化——这类问题排查起来极其痛苦。Poetry通过poetry.lock锁定所有传递依赖的精确版本配合poetry install可以在任何机器上复现出一模一样的环境。初始化命令很简单# 安装poetry推荐pipx方式避免污染全局环境 pipx install poetry # 初始化项目 poetry new ai-engineering-from-scratch cd ai-engineering-from-scratch # 添加核心依赖 poetry add fastapi uvicorn[standard] sqlalchemy asyncpg redis celery onnxruntime poetry add --group dev pytest pytest-asyncio ruff mypy4.2 Docker Compose搭建基础设施环境开发环境的可复现性除了Python依赖还包括基础设施。我从来不在本机直接安装PostgreSQL和Redis而是全部容器化。项目根目录下的docker-compose.yml我贴一个精简版本version: 3.8 services: postgres: image: pgvector/pgvector:pg16 environment: POSTGRES_USER: ai_app POSTGRES_PASSWORD: ai_app_password POSTGRES_DB: ai_platform ports: - 5432:5432 volumes: - pg_data:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U ai_app] interval: 5s timeout: 3s retries: 10 redis: image: redis:7-alpine ports: - 6379:6379 healthcheck: test: [CMD, redis-cli, ping] interval: 5s timeout: 3s retries: 10 volumes: pg_data:这里有个细节容易被忽略PostgreSQL镜像特意选了pgvector/pgvector:pg16而不是官方的postgres:16因为pgvector扩展需要预装到数据库镜像里否则后面CREATE EXTENSION vector会报错。这种“看似不起眼但影响全局”的选择就是工程经验和纯教程的区别。起环境只需要一条命令docker-compose up -d在CI/CD中这些healthcheck配置还能帮我们实现“等服务真正就绪才跑测试”避免出现测试一启动就连不上数据库的随机失败。5. 核心管线实现数据、模型、服务的串联5.1 数据接入与治理AI工程的隐形地基很多AI项目死在第一步——数据根本没法用。我在这个项目里定义了一套标准化的数据治理流程按照这个流程走能规避80%的数据坑。原始数据落库。所有采集到的原始数据先原样存入raw_data表不做任何清洗保证可回溯。这张表永远只做插入不做更新和删除。标准化处理。从原始数据里提取统一schema的字段比如文本去重、编码统一为UTF-8、时间格式统一为ISO 8601。这一步用Python Pandas写定时任务处理。特征与标注管理。对于监督学习部分标注数据要单独管理每次标注版本都要记录标注人、标注时间、标注规范版本。我用了一套极简的标注管理方式——每批数据一个标注规范文件Markdown格式 一条数据库记录谁标了哪些数据一目了然。有一个经验我要特别强调数据质量问题的排查成本远远高于模型问题。模型效果不对你还能调参重训但如果训练数据里混入了重复样本、错标样本、时域泄漏样本你的模型会“很稳定地犯错”而且极难定位。宁可花70%的精力在数据治理上也不要把这个债留给后续所有环节。5.2 模型服务化从PyTorch到ONNX Runtime的转换训练好的PyTorch模型要变成线上服务第一步是导出为ONNX格式。这个环节有很多细节坑我都逐一踩过先记录正确的操作路径import torch import onnxruntime as ort # 以HuggingFace的BERT模型为例 from transformers import BertModel, BertTokenizer # 1. 加载训练好的模型权重 model BertModel.from_pretrained(your_finetuned_model) model.eval() # 2. 用dummy input导出ONNX tokenizer BertTokenizer.from_pretrained(your_finetuned_model) dummy_input tokenizer(这是一个测试输入, return_tensorspt) torch.onnx.export( model, tuple(dummy_input.values()), model.onnx, input_names[input_ids, attention_mask, token_type_ids], output_names[last_hidden_state], dynamic_axes{ input_ids: {0: batch_size, 1: seq_len}, attention_mask: {0: batch_size, 1: seq_len}, last_hidden_state: {0: batch_size, 1: seq_len} }, opset_version17 )这里有个关键决策dynamic_axes。我建议所有维度都声明为动态虽然会带来少量性能损失ONNX Runtime需要动态分配内存但换来的是服务端不用处理输入长度分组代码大幅简化。只有在单条输入长度非常固定的场景比如固定224x224图像分类才考虑用静态shape换取极致性能。模型导出后用onnxruntime-gpu还是onnxruntime取决于你的部署环境。我在初期只做CPU推理配合int8量化效果已经非常好。量化代码大致如下import onnxruntime as ort from onnxruntime.quantization import quantize_dynamic, QuantType # int8动态量化最简单、最稳定的量化方式 quantized_model_path model_int8.onnx quantize_dynamic( model.onnx, quantized_model_path, weight_typeQuantType.QUInt8 ) # 验证量化前后的一致性 import numpy as np sess ort.InferenceSession(model.onnx) sess_q ort.InferenceSession(quantized_model_path) test_input {input_ids: np.array([[1, 2, 3]]), attention_mask: np.array([[1, 1, 1]])} output sess.run(None, test_input) output_q sess_q.run(None, test_input) print(Max abs diff:, np.max(np.abs(output[0] - output_q[0])))实测BERT base模型量化前后最大输出差异在0.01量级完全不影响下游任务的判别结果但推理速度提升约3倍、内存占用降低约60%。这种性价比极高的优化在从零搭建阶段应该优先做。5.3 在线推理服务FastAPI的最佳实践推理API是AI系统的门面用户感知到的延迟、稳定性都由它决定。我分享一个经过生产验证的FastAPI推理服务骨架import asyncio import numpy as np from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from contextlib import asynccontextmanager import onnxruntime as ort class InferenceRequest(BaseModel): text: str Field(..., min_length1, max_length512) top_k: int Field(5, ge1, le20) class InferenceResponse(BaseModel): result: dict latency_ms: float asynccontextmanager async def lifespan(app: FastAPI): # 全局只加载一次模型避免每个请求重复加载 app.state.session ort.InferenceSession( model_int8.onnx, providers[CPUExecutionProvider] ) yield # 关闭时清理资源 app FastAPI(lifespanlifespan) app.post(/v1/inference, response_modelInferenceResponse) async def inference(req: InferenceRequest): import time start time.perf_counter() try: inputs preprocess(req.text) outputs app.state.session.run(None, inputs) result postprocess(outputs, req.top_k) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.perf_counter() - start) * 1000 return InferenceResponse(resultresult, latency_mslatency) def preprocess(text: str): # 假设tokenizer在服务启动时也初始化了 tokens app.state.tokenizer(text, return_tensorsnp) return { input_ids: tokens[input_ids], attention_mask: tokens[attention_mask], token_type_ids: tokens[token_type_ids] }这个骨架里三个关键细节值得特别留意。第一是lifespan机制。模型初始化是重操作如果放在请求处理函数里第一个请求会额外增加几秒到几十秒的加载时间线上监控会直接告警超时。用lifespan在服务启动时加载所有worker进程共享一份模型句柄后续请求零加载开销。第二是同步推理与异步接口的共存。ONNX Runtime的run是同步阻塞调用我把它放在async函数里直接调用看起来像是“阻塞了事件循环”。这块我研究过在CPython里ONNX Runtime的run会释放GIL所以同步调用并不会明显阻塞其他异步任务实测在8核机器上、4个Gunicorn worker能稳定扛住每秒200次以上的推理请求。如果你用的是GPU版建议用线程池调度避免阻塞。第三是超时控制。AI服务最大的风险是一个慢请求拖垮整个进程。FastAPI的默认行为是请求无限期等待我建议在uvicorn启动参数里加上超时配置或者在服务层用asyncio.wait_for包一层try: result await asyncio.wait_for( asyncio.to_thread(run_inference, req.text, req.top_k), timeout3.0 ) except asyncio.TimeoutError: raise HTTPException(status_code504, detailInference timeout)这样保证任何情况下单个请求最多占用3秒不会出现连接被拖死的情况。5.4 RAG管线的实现检索增强生成的工程核心如果你做的是LLM相关的应用那RAG检索增强生成管线基本是标配。我实现的精简但完整的RAG链路如下离线索引构建。先对海量知识文档做切分我这里用的是递归字符切分器按500字符一块、80字符重叠from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap80, separators[\n\n, \n, 。, ., , ] ) chunks splitter.split_text(raw_text)切分参数的选择有讲究。chunk_size太小检索到的上下文不完整太大命中片段包含太多无关信息且浪费LLM上下文窗口。500-800字符对于大多数技术文档类知识库是比较稳的经验值chunk_overlap设为10%-20%可以避免关键信息被切分截断。向量化我建议用固定维度的嵌入模型比如text-embedding-ada-002或bge-large-zh生成的向量直接插入pgvectorCREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS document_chunks ( id BIGSERIAL PRIMARY KEY, chunk_text TEXT NOT NULL, embedding vector(1024), metadata JSONB DEFAULT {}::jsonb, created_at TIMESTAMPTZ DEFAULT now() ); CREATE INDEX idx_document_chunks_embedding ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);IVFFlat索引的lists参数需要根据数据量设置经验法则是lists ≈ 5% * 数据量但建议在100到1000之间搜寻调优。如果数据量小小于1万条甚至可以不建索引暴力扫描更快因为ANN索引本身有构建开销和召回率损失。在线检索服务。用户查询进来先向量化查询文本然后在pgvector里做最近邻搜索from sqlalchemy import text query_embedding generate_embedding(query_text) sql text( SELECT chunk_text, metadata, 1 - (embedding :query_vec) AS similarity FROM document_chunks ORDER BY embedding :query_vec LIMIT :top_k ) results db.execute(sql, { query_vec: query_embedding, top_k: 5 }).fetchall() context \n\n.join([r.chunk_text for r in results if r.similarity 0.5])是pgvector的余弦距离算子1 - distance得到的就是余弦相似度。这里加了一个相似度大于0.5的阈值过滤防止检索完全无关的内容被强行塞进LLM的上下文——这点我在真实项目里反复验证过没有阈值过滤时LLM会被无关上下文带偏一本正经地胡说八道。最后把检索到的上下文和用户问题拼装成prompt调LLM生成回答。这部分的工程化重点不在prompt模板本身而在于整个链路的监控检索召回率、生成时延、上下文占用token数都要有埋点统计。没有这些数据后续做优化只能靠猜。6. 发布流程与部署架构6.1 模型版本管理像管理代码一样管理模型从零搭建AI项目时模型版本管理是最容易被忽略、后期最痛苦的问题。我用过最简单的方案是——在训练脚本里把模型保存路径加上日期和git commit号model_save_path fmodels/model_{datetime.now():%Y%m%d_%H%M%S}_{git_commit_short}.onnx这样每次产出的模型文件名自带版本和时间信息不会出现“最后跑出来的模型不知道是哪个版本”的问题。更进一步我建议在数据库里建一张模型版本记录表CREATE TABLE model_versions ( id BIGSERIAL PRIMARY KEY, model_name VARCHAR(100) NOT NULL, version VARCHAR(50) NOT NULL, git_commit VARCHAR(50), metrics JSONB, artifact_path VARCHAR(500), status VARCHAR(20) DEFAULT staging, -- staging/production/retired created_at TIMESTAMPTZ DEFAULT now(), UNIQUE(model_name, version) );部署时API服务通过环境变量或配置中心指定要加载哪个版本的模型而不是硬编码模型路径。这样做的价值在于a/b测试时两个服务实例可以分别加载不同版本模型出问题时一秒钟就能回滚到上一个稳定版本。6.2 服务部署Docker化与滚动发布AI服务部署我全程用Docker保证了“本地能跑”和“线上能跑”的一致性。一个精简的Dockerfile示例FROM python:3.11-slim as builder ENV POETRY_VERSION1.7.1 RUN pip install poetry${POETRY_VERSION} WORKDIR /app COPY pyproject.toml poetry.lock ./ RUN poetry config virtualenvs.create false \ poetry install --no-root --only main FROM python:3.11-slim COPY --frombuilder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY --frombuilder /usr/local/bin /usr/local/bin WORKDIR /app COPY . . EXPOSE 8000 CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]这里用了多阶段构建第一层安装依赖第二层只拷贝site-packages镜像体积能小很多。注意--workers 4的选择逻辑——我先用nproc看CPU核数再用压测工具我用的locust测出4个worker在目标并发下的CPU和响应时间表现最后定下来。部署到服务器上我用Docker Compose编排应用服务和基础设施。新增一个服务条目api: build: . ports: - 8000:8000 environment: DATABASE_URL: postgresql://ai_app:ai_app_passwordpostgres:5432/ai_platform REDIS_URL: redis://redis:6379/0 MODEL_VERSION: prod_20240521_v2 depends_on: postgres: condition: service_healthy redis: condition: service_healthy restart: unless-stoppedMODEL_VERSION这个环境变量就是上文说的模型版本管理在部署层的落地。更新模型时只改这个变量、重启服务就能完成模型切换如果新模型有问题改回旧版本号再重启回滚瞬间完成。滚动发布我直接用最朴素的方案先起一个新版本容器跑健康检查通过后用Nginx把流量切过去再停掉旧容器。这套流程配合Github Actions或者Jenkins十几行配置就能实现不需要引入K8s这种重型武器——从零起步的项目复杂度要一点一点加。6.3 任务队列异步处理慢任务的基石AI系统里总有异步场景知识库更新、文档向量化、批量预测、消息通知。这些任务如果在Web进程里同步执行会直接阻塞请求响应。我的选择是Celery Redis。Celery配置非常简洁from celery import Celery celery_app Celery( ai_tasks, brokerredis://localhost:6379/1, backendredis://localhost:6379/2 ) celery_app.conf.update( task_serializerjson, result_serializerjson, accept_content[json], timezoneAsia/Shanghai, enable_utcTrue, worker_max_tasks_per_child200, # 防止任务内累积内存泄漏 task_time_limit300, # 单个任务最长5分钟 ) celery_app.task def embed_document_batch(doc_ids: list[int]): # 批量向量化逻辑 ...那个worker_max_tasks_per_child200是经验之谈。AI任务经常涉及加载模型、处理大文本内存碎片化很快。限制每个worker子进程处理200个任务后重启能有效避免内存持续膨胀导致的OOM。类似的task_time_limit防止个别卡死任务占着worker不放。调用方式也很简单embed_document_batch.delay(doc_ids[1, 2, 3])异步任务丢进队列后立即返回Web请求不会被阻塞。Celery worker单独以容器方式运行可以独立扩缩容——如果发现向量化任务积压多开几个worker容器就能缓解跟API服务的扩容互不影响。7. 可观测性当AI系统出问题时怎么快速定位7.1 日志、指标、追踪三件套AI系统出故障时最大的痛苦在于“不知道问题出在哪一段”——是数据不对模型输出异常还是依赖服务超时要回答这个问题必须在系统建设初期就搭好可观测体系具体就是我常说的日志、指标、追踪三件套。日志我全部用结构化格式JSON每行日志里带上timestamp、level、service、request_id、message和自定义字段。这样在ELK里可以直接按request_id把一条链路的所有日志串起来。一条标准日志示例{timestamp:2024-05-21T10:30:12.345Z,level:INFO,service:api,request_id:a3f9c2,event:inference_completed,model_version:prod_20240521_v2,latency_ms:45,status:success}指标用Prometheus收集。我曝光四个核心指标请求量QPS、时延分布P50/P95/P99、错误率、模型推理时延。FastAPI接入Prometheus客户端后几行代码就能完成埋点from prometheus_client import Counter, Histogram REQUESTS Counter(http_requests_total, Total HTTP requests, [method, path, status]) LATENCY Histogram(http_request_duration_seconds, HTTP request latency, [method, path]) app.middleware(http) async def metrics_middleware(request, call_next): start time.perf_counter() response await call_next(request) duration time.perf_counter() - start REQUESTS.labels(request.method, request.path, response.status_code).inc() LATENCY.labels(request.method, request.path).observe(duration) return response追踪主要看外部依赖的调用链。AI链路里尤其要盯的是向量检索耗时、LLM调用耗时、下游服务耗时。用OpenTelemetry可以自动埋点但小团队我建议先手动打点每个环节都记录开始时间、结束时间、耗时汇总成一个trace_id下的结构化日志。等系统复杂到需要跨服务追踪时再上完整的OpenTelemetry体系。7.2 监控告警什么样的告警才不会打扰到你告警配置的核心哲学是宁可漏报不要误报。与其频繁被打断去处理“无关紧要的告警”不如把告警阈值调到真正会出问题的那一刻。我目前保留的告警规则少得可怜但每一条都意义明确告警项触发条件重要程度P95时延超过阈值P95响应时间 2s持续3分钟高请求错误率上升5XX错误率 1%持续5分钟高模型变化监控模型版本变更却无对应部署记录中队列积压Celery任务队列长度 5000持续10分钟中内存健康容器内存使用率 85%持续15分钟中Grafana里配置告警规则方式就是“在仪表板的对应Panel上配置Threshold”。有一个技巧告警消息里一定要带上链接到仪表板和最近日志查询入口否则值班同学收到告警还要到处找系统入口耽误时间。7.3 一套实用的AI系统健康度打分模型我基于实际运维经验总结了一个简洁的系统健康度评估模型可以自动化给AI系统打分健康分 0.3 * 接口可用性分 0.3 * 响应速度分 0.2 * 业务效果分 0.2 * 数据新鲜度分接口可用性分(1 - 5XX错误率) * 100。低于95分告警。响应速度分基于P95时延映射P95 800ms给100分每增加200ms扣10分。业务效果分线下定期评估集推理结果对比基准版本的指标变化好于基准加分否则扣分。数据新鲜度分知识库最后更新时间距当前时间超过预设周期如7天则扣分。这套打分模型我会每天定时跑一次生成一条“今日系统健康度”日报发到团队群里。它不会直接告警但能让所有人对系统运行状态有个全局感知并且能在业务指标变差之前提前发现数据停留、模型退化等苗头。8. 典型问题实录我从这些坑里爬出来的8.1 ONNX导出时的动态轴报错第一次导出ONNX时我遇到了RuntimeError: Failed to export an ONNX attribute axes...这类报错。原因是模型内部有固定维度的操作比如位置编码的arange张量导出时需要指定动态轴但某些算子不支持。当时最有效的排查手段是用onnxruntime的onnxruntime.transformers优化器打印模型输入输出信息以及检查每个算子的版本是否老化。解决方案比想象中简单把opset_version升到14以上或者对模型做简化处理用onnxsim工具移除训练相关的动态控制流。多数算子兼容问题都能靠这两个手段解决。8.2 pgvector索引失效导致检索全表扫描有段时间检索延迟从20ms暴增到2秒查了PostgreSQL的执行计划才发现IVFFlat索引没有被用到。原因是数据量变化后索引的lists参数不再合适。pgvector的IVFFlat索引需要一次抽样训练建索引时会重新聚类但数据量增长后原来指定的lists参数对应的聚类效果变差。解决方法是重新建索引并调整参数DROP INDEX idx_document_chunks_embedding; CREATE INDEX idx_document_chunks_embedding ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists 200);另外检查是否用了参数化查询。如果你传入的向量不是常量而是参数有的数据库版本有query plan cache问题会退化成顺序扫描。当时最后定位到的原因很简单数据量从30万涨到了120万lists100严重不够用重建为200后性能恢复。这事的教训是索引参数不是建完就完事的要随着数据量增长定期评估。8.3 LLM接口超时但HTTP状态码还是200这是最隐蔽的一个坑。我们的调用方曾经反馈“接口很慢但不报错”排查后发现LLM生成回答耗时超过我设置的上游超时时间但底层HTTP客户端吞掉了超时异常返回了空内容被上层误判为“成功但空回复”。追查代码时发现当时用了requests库的默认行为超时设置只对连接生效对读取响应不生效。修复方式很直接requests.post(url, jsondata, timeout(3.05, 30))timeout元组的第一个值是连接超时第二个值是读取超时。这个参数强烈建议所有调用外部API的代码里都显式设置否则线上一定会出现“等待三分钟才崩溃”的慢请求。8.4 多进程环境下Redis连接数爆了Celery和API都连Redis默认连接池参数不调优时高并发下会出现Cannot assign requested address报错。原因就是底层TCP连接数达到系统限制。解决方式两个一是给Redis客户端配置合适的连接池上限二是服务容器里设置ulimit -n提高文件描述符上限。我最终的配置是redis_client redis.Redis( hostlocalhost, port6379, max_connections200, socket_connect_timeout3, socket_timeout3 )这块的经验是任何系统加了并发压测运行几分钟后看系统日志出现的连接类报错基本都能归因于连接池或系统文件描述符限制。8.5 模型精度线上与离线不一致遇到过最诡异的问题同一个模型线下评测F10.92线上抽样显示F10.76。反复验证后发现问题出在数据预处理不一致——离线代码里做了一遍文本清洗线上推理服务的预处理函数没有同步这个逻辑。这类问题的根因在于“代码漂移”。这次之后我在代码库加了一道硬性检查训练和推理必须共用同一个preprocess模块并且模型产物里除了权重还要包含一份preprocess_config.json记录清洗规则、分词参数、归一化参数。加载模型时服务端读取这份配置并初始化预处理这样线上线下就永远一致了。9. 从小项目到持续演进工程化没有终点如果你是从零起步搭AI系统上面这几个章节走完已经能形成一个可以稳定运行的完整闭环。不过在我的经验里这还只是工程化的第一阶段。真正让AI工程长期健康迭代的还有几个“软性”却关键的习惯。实验记录规范化。我要求每个团队成员每次做实验都必须记录数据版本、代码commit、模型参数、测评结果、结论。哪怕只是调了个学习率也要留痕。开始觉得繁琐但当一个月后你发现模型效果下降、需要回溯是哪次改动导致的整洁的实验记录能救你命。模型效果回归测试。每次模型上线前必须跑一遍固定的评测集。这个评测集要有覆盖面、要长期维护特别要包含历次踩坑的边界场景。AI模型不像传统软件有明确的“功能正确与否”回归测试就是AI系统的“自动化测试”。定期体检AI系统的健康度。上面说的健康度打分每天跑、每周看趋势、每月复盘。我建议选定一个固定时间比如每周一上午把上周的健康分趋势和业务指标放在一起看这个习惯能让你比业务用户更早发现系统的微妙退化。做AI工程最关键的觉悟是模型能力决定上限工程能力决定下限。很聪明的一个模型部署不好、监控缺失、无法迭代最终也发挥不出价值相反一个中规中矩的模型只要工程链路扎实、迭代顺畅用户感受到的稳定性和可用性会非常好。所以不用急着追最前沿的模型架构先把工程地基打牢。地基之上模型的每一次进步都能稳稳地变成用户价值。 从零搭建AI工程：FastAPI+ONNX Runtime+pgvector实践指南
1. 从零开始做AI工程我到底在搭什么拿到“ai-engineering-from-scratch”这个题目很多人第一反应是“又要写一篇AI入门教程”。但如果你真在工业界待过几年就会明白这里的“from scratch”远不是跑通一个Jupyter Notebook那么简单。它意味着你要从一台裸机、一份Python环境、一个空目录开始把数据管线、模型服务、评估体系、监控告警、迭代流程一层层垒起来最终形成一个能稳定产出业务价值的AI系统。这篇文章不是什么大而全的手册更像是我把自己过去从零搭建AI工程项目踩过的坑、验证过的方案、推翻过的设计按一条可复现的主线串起来。内容适合两类人一类是刚接手AI工程化项目、需要对整体技术栈建立清晰认知的同学另一类是已经跑通过一些模型实验、但发现“离线能跑”和“线上稳定”之间隔着巨大鸿沟的工程师。读完之后你会对AI工程化需要哪些核心组件、每个组件解决什么问题、组件之间怎么衔接有一个系统性的判断而不是被碎片化的教程带着走。我最终落地的这套方案技术栈是Python 3.11 FastAPI PostgreSQL Redis Docker Compose模型层用ONNX Runtime做推理向量检索用pgvector任务队列用Celery监控用Prometheus Grafana。下面我按实际搭建顺序把每个环节的设计思路和实操细节逐一拆开讲。2. 工程化之前先把需求翻译成技术约束2.1 业务目标与技术方案的映射逻辑任何AI项目的第一件事都不是选框架而是把业务语言翻译成技术约束。比如“我们要做一个智能客服助手”这个描述里至少藏着三个关键约束响应时延要求多高直接影响是否能用流式输出、是否要上GPU推理、知识更新频率多快决定检索方案是静态索引还是实时向量化、错误容忍度多大决定是否需要人工审核兜底、是否需要置信度阈值。我习惯用一张表格把这类需求显式写出来避免后续设计跑偏业务需求技术约束设计选择回答准确率不低于90%需要评估集与回归测试搭建离线评估流水线每次模型更新自动跑基准首字响应小于500ms推理链路总耗时预算ONNX Runtime CPU量化模型避免GPU依赖知识每周更新一次知识库同步机制向量化任务入Celery队列错峰执行单日请求量峰值2万并发与限流策略FastAPI Gunicorn多workerRedis滑动窗口限流系统故障可定位日志与指标可观测结构化日志 Prometheus指标 告警规则这个环节很多新手会跳过直接开始写代码。但我在实际项目中吃过亏有一版智能问答系统所有评测指标都达标结果业务方反馈“回答太慢了”一查才发现我们把大量场景设计成了同步阻塞调用接口平均耗时1.8秒完全超出了客服坐席边聊天边等待的心理阈值。需求翻译这一步省掉的每一分钟都会在后续返工中加倍偿还。2.2 离线实验与线上系统的边界划定从零搭建时最容易犯的错误是把Notebook里的实验代码直接搬进生产服务。离线实验和线上系统至少有三条边界必须划定清楚数据边界——线下可以用全量历史数据训练线上只能看到截止当前时刻的数据资源边界——线下可以不计成本地调参跑实验线上必须考虑推理成本和时延评估边界——线下可以用准确率、F1这类离线指标衡量线上更关心的是用户留存、转化率、工单解决率这类业务指标。我的做法是维护两份独立的代码一份是experiments/目录下的研究代码自由度极高另一份是services/目录下的工程代码必须经过代码审查、测试覆盖、构建产物化。两者之间唯一的桥梁是模型产物和评估报告实验代码产出的模型经过评估后以版本化方式交给工程侧部署。这样既保留了研究阶段的灵活性又保证了生产环境的稳定性。3. 技术选型每一项选择背后的真实理由3.1 为什么是FastAPI而不是Flask或Django如果你去搜“Python Web框架选型”会看到大量对比文章。但AI工程场景下我选FastAPI的核心原因其实就两条一是原生异步支持LLM服务普遍是IO密集型等待模型推理结果、等待数据库查询async/await能把单机并发能力提升一个量级二是自动生成OpenAPI文档在前后端联调、接口对接时能省掉大量沟通成本。打个不那么准确的比方Flask像是手动挡汽车结构简单、什么都能自己控制但每个环节都要自己操作Django像是带了一整套生活用品的房车沉重但啥都有FastAPI像是自动挡的现代轿车该有的都有、日常开最顺手。对于AI服务这种“接口不多但并发要求高、迭代频繁”的场景FastAPI正好踩在最舒服的位置上。3.2 向量检索选了pgvector而不是独立向量数据库这个决定我犹豫了很久。刚开始我倾向于Milvus或Weaviate这类专用向量数据库因为它们功能全、性能强。后来之所以定pgvector核心原因是运维复杂度的控制。一个从零开始的AI项目如果同时要维护PostgreSQL、Redis、向量库、模型服务、任务队列任何一个小组件出问题都够折腾半天。用pgvector可以直接复用PostgreSQL的备份、恢复、权限体系少维护一个组件数据一致性也更好保证。它的性能到底够不够用我实测下来在单机PostgreSQL上pgvector对100万条768维向量的ANN检索使用IVFFlat索引单次查询延迟在10-30毫秒之间对绝大多数AI应用场景完全够用。只有当数据量到千万级以上、查询QPS非常高的时候才需要认真考虑独立向量数据库。对于从零起步的项目先用pgvector把业务跑通等量级上来再迁移是性价比最高的路径。3.3 模型部署为什么选了ONNX Runtime模型推理这块我见过太多团队一上来就搞TensorRT、搞vLLM结果发现工程复杂度远超预期。我自己一开始用的也是PyTorch直接加载模型做推理但很快遇到两个问题一是Python进程内存占用高多worker部署时显存和内存都吃紧二是模型部署环境需要完整安装PyTorch依赖镜像体积动辄几个GB。ONNX Runtime解决了这两个痛点模型从PyTorch导出为ONNX格式后推理时不再依赖PyTorch运行时镜像可以缩到几百MB同时ONNX Runtime对CPU推理做了大量优化配合int8量化在CPU上跑BERT类模型的延迟能做到原来的三分之一左右。当然ONNX导出过程有时候会遇到算子兼容问题这个后面我会专门讲几个典型坑。4. 环境准备从裸机到可复现的开发环境4.1 Python版本与依赖管理的坑从零搭建第一个要确定的就是Python版本。我推荐Python 3.11原因很简单它是目前兼容性、性能、生态三方平衡最好的版本。3.12虽然更新但部分深度学习库的预编译wheel还跟进得不完美3.10以下则逐渐进入维护末期没必要新项目踩旧版本。依赖管理方面我强烈建议直接上Poetry或uv而不是裸用requirements.txt。裸用requirements.txt最常见的灾难是开发环境装的是numpy1.24.3测试环境被某次pip install悄悄升级到了1.26然后模型推理结果发生了微妙变化——这类问题排查起来极其痛苦。Poetry通过poetry.lock锁定所有传递依赖的精确版本配合poetry install可以在任何机器上复现出一模一样的环境。初始化命令很简单# 安装poetry推荐pipx方式避免污染全局环境 pipx install poetry # 初始化项目 poetry new ai-engineering-from-scratch cd ai-engineering-from-scratch # 添加核心依赖 poetry add fastapi uvicorn[standard] sqlalchemy asyncpg redis celery onnxruntime poetry add --group dev pytest pytest-asyncio ruff mypy4.2 Docker Compose搭建基础设施环境开发环境的可复现性除了Python依赖还包括基础设施。我从来不在本机直接安装PostgreSQL和Redis而是全部容器化。项目根目录下的docker-compose.yml我贴一个精简版本version: 3.8 services: postgres: image: pgvector/pgvector:pg16 environment: POSTGRES_USER: ai_app POSTGRES_PASSWORD: ai_app_password POSTGRES_DB: ai_platform ports: - 5432:5432 volumes: - pg_data:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U ai_app] interval: 5s timeout: 3s retries: 10 redis: image: redis:7-alpine ports: - 6379:6379 healthcheck: test: [CMD, redis-cli, ping] interval: 5s timeout: 3s retries: 10 volumes: pg_data:这里有个细节容易被忽略PostgreSQL镜像特意选了pgvector/pgvector:pg16而不是官方的postgres:16因为pgvector扩展需要预装到数据库镜像里否则后面CREATE EXTENSION vector会报错。这种“看似不起眼但影响全局”的选择就是工程经验和纯教程的区别。起环境只需要一条命令docker-compose up -d在CI/CD中这些healthcheck配置还能帮我们实现“等服务真正就绪才跑测试”避免出现测试一启动就连不上数据库的随机失败。5. 核心管线实现数据、模型、服务的串联5.1 数据接入与治理AI工程的隐形地基很多AI项目死在第一步——数据根本没法用。我在这个项目里定义了一套标准化的数据治理流程按照这个流程走能规避80%的数据坑。原始数据落库。所有采集到的原始数据先原样存入raw_data表不做任何清洗保证可回溯。这张表永远只做插入不做更新和删除。标准化处理。从原始数据里提取统一schema的字段比如文本去重、编码统一为UTF-8、时间格式统一为ISO 8601。这一步用Python Pandas写定时任务处理。特征与标注管理。对于监督学习部分标注数据要单独管理每次标注版本都要记录标注人、标注时间、标注规范版本。我用了一套极简的标注管理方式——每批数据一个标注规范文件Markdown格式 一条数据库记录谁标了哪些数据一目了然。有一个经验我要特别强调数据质量问题的排查成本远远高于模型问题。模型效果不对你还能调参重训但如果训练数据里混入了重复样本、错标样本、时域泄漏样本你的模型会“很稳定地犯错”而且极难定位。宁可花70%的精力在数据治理上也不要把这个债留给后续所有环节。5.2 模型服务化从PyTorch到ONNX Runtime的转换训练好的PyTorch模型要变成线上服务第一步是导出为ONNX格式。这个环节有很多细节坑我都逐一踩过先记录正确的操作路径import torch import onnxruntime as ort # 以HuggingFace的BERT模型为例 from transformers import BertModel, BertTokenizer # 1. 加载训练好的模型权重 model BertModel.from_pretrained(your_finetuned_model) model.eval() # 2. 用dummy input导出ONNX tokenizer BertTokenizer.from_pretrained(your_finetuned_model) dummy_input tokenizer(这是一个测试输入, return_tensorspt) torch.onnx.export( model, tuple(dummy_input.values()), model.onnx, input_names[input_ids, attention_mask, token_type_ids], output_names[last_hidden_state], dynamic_axes{ input_ids: {0: batch_size, 1: seq_len}, attention_mask: {0: batch_size, 1: seq_len}, last_hidden_state: {0: batch_size, 1: seq_len} }, opset_version17 )这里有个关键决策dynamic_axes。我建议所有维度都声明为动态虽然会带来少量性能损失ONNX Runtime需要动态分配内存但换来的是服务端不用处理输入长度分组代码大幅简化。只有在单条输入长度非常固定的场景比如固定224x224图像分类才考虑用静态shape换取极致性能。模型导出后用onnxruntime-gpu还是onnxruntime取决于你的部署环境。我在初期只做CPU推理配合int8量化效果已经非常好。量化代码大致如下import onnxruntime as ort from onnxruntime.quantization import quantize_dynamic, QuantType # int8动态量化最简单、最稳定的量化方式 quantized_model_path model_int8.onnx quantize_dynamic( model.onnx, quantized_model_path, weight_typeQuantType.QUInt8 ) # 验证量化前后的一致性 import numpy as np sess ort.InferenceSession(model.onnx) sess_q ort.InferenceSession(quantized_model_path) test_input {input_ids: np.array([[1, 2, 3]]), attention_mask: np.array([[1, 1, 1]])} output sess.run(None, test_input) output_q sess_q.run(None, test_input) print(Max abs diff:, np.max(np.abs(output[0] - output_q[0])))实测BERT base模型量化前后最大输出差异在0.01量级完全不影响下游任务的判别结果但推理速度提升约3倍、内存占用降低约60%。这种性价比极高的优化在从零搭建阶段应该优先做。5.3 在线推理服务FastAPI的最佳实践推理API是AI系统的门面用户感知到的延迟、稳定性都由它决定。我分享一个经过生产验证的FastAPI推理服务骨架import asyncio import numpy as np from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from contextlib import asynccontextmanager import onnxruntime as ort class InferenceRequest(BaseModel): text: str Field(..., min_length1, max_length512) top_k: int Field(5, ge1, le20) class InferenceResponse(BaseModel): result: dict latency_ms: float asynccontextmanager async def lifespan(app: FastAPI): # 全局只加载一次模型避免每个请求重复加载 app.state.session ort.InferenceSession( model_int8.onnx, providers[CPUExecutionProvider] ) yield # 关闭时清理资源 app FastAPI(lifespanlifespan) app.post(/v1/inference, response_modelInferenceResponse) async def inference(req: InferenceRequest): import time start time.perf_counter() try: inputs preprocess(req.text) outputs app.state.session.run(None, inputs) result postprocess(outputs, req.top_k) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.perf_counter() - start) * 1000 return InferenceResponse(resultresult, latency_mslatency) def preprocess(text: str): # 假设tokenizer在服务启动时也初始化了 tokens app.state.tokenizer(text, return_tensorsnp) return { input_ids: tokens[input_ids], attention_mask: tokens[attention_mask], token_type_ids: tokens[token_type_ids] }这个骨架里三个关键细节值得特别留意。第一是lifespan机制。模型初始化是重操作如果放在请求处理函数里第一个请求会额外增加几秒到几十秒的加载时间线上监控会直接告警超时。用lifespan在服务启动时加载所有worker进程共享一份模型句柄后续请求零加载开销。第二是同步推理与异步接口的共存。ONNX Runtime的run是同步阻塞调用我把它放在async函数里直接调用看起来像是“阻塞了事件循环”。这块我研究过在CPython里ONNX Runtime的run会释放GIL所以同步调用并不会明显阻塞其他异步任务实测在8核机器上、4个Gunicorn worker能稳定扛住每秒200次以上的推理请求。如果你用的是GPU版建议用线程池调度避免阻塞。第三是超时控制。AI服务最大的风险是一个慢请求拖垮整个进程。FastAPI的默认行为是请求无限期等待我建议在uvicorn启动参数里加上超时配置或者在服务层用asyncio.wait_for包一层try: result await asyncio.wait_for( asyncio.to_thread(run_inference, req.text, req.top_k), timeout3.0 ) except asyncio.TimeoutError: raise HTTPException(status_code504, detailInference timeout)这样保证任何情况下单个请求最多占用3秒不会出现连接被拖死的情况。5.4 RAG管线的实现检索增强生成的工程核心如果你做的是LLM相关的应用那RAG检索增强生成管线基本是标配。我实现的精简但完整的RAG链路如下离线索引构建。先对海量知识文档做切分我这里用的是递归字符切分器按500字符一块、80字符重叠from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap80, separators[\n\n, \n, 。, ., , ] ) chunks splitter.split_text(raw_text)切分参数的选择有讲究。chunk_size太小检索到的上下文不完整太大命中片段包含太多无关信息且浪费LLM上下文窗口。500-800字符对于大多数技术文档类知识库是比较稳的经验值chunk_overlap设为10%-20%可以避免关键信息被切分截断。向量化我建议用固定维度的嵌入模型比如text-embedding-ada-002或bge-large-zh生成的向量直接插入pgvectorCREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS document_chunks ( id BIGSERIAL PRIMARY KEY, chunk_text TEXT NOT NULL, embedding vector(1024), metadata JSONB DEFAULT {}::jsonb, created_at TIMESTAMPTZ DEFAULT now() ); CREATE INDEX idx_document_chunks_embedding ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists 100);IVFFlat索引的lists参数需要根据数据量设置经验法则是lists ≈ 5% * 数据量但建议在100到1000之间搜寻调优。如果数据量小小于1万条甚至可以不建索引暴力扫描更快因为ANN索引本身有构建开销和召回率损失。在线检索服务。用户查询进来先向量化查询文本然后在pgvector里做最近邻搜索from sqlalchemy import text query_embedding generate_embedding(query_text) sql text( SELECT chunk_text, metadata, 1 - (embedding :query_vec) AS similarity FROM document_chunks ORDER BY embedding :query_vec LIMIT :top_k ) results db.execute(sql, { query_vec: query_embedding, top_k: 5 }).fetchall() context \n\n.join([r.chunk_text for r in results if r.similarity 0.5])是pgvector的余弦距离算子1 - distance得到的就是余弦相似度。这里加了一个相似度大于0.5的阈值过滤防止检索完全无关的内容被强行塞进LLM的上下文——这点我在真实项目里反复验证过没有阈值过滤时LLM会被无关上下文带偏一本正经地胡说八道。最后把检索到的上下文和用户问题拼装成prompt调LLM生成回答。这部分的工程化重点不在prompt模板本身而在于整个链路的监控检索召回率、生成时延、上下文占用token数都要有埋点统计。没有这些数据后续做优化只能靠猜。6. 发布流程与部署架构6.1 模型版本管理像管理代码一样管理模型从零搭建AI项目时模型版本管理是最容易被忽略、后期最痛苦的问题。我用过最简单的方案是——在训练脚本里把模型保存路径加上日期和git commit号model_save_path fmodels/model_{datetime.now():%Y%m%d_%H%M%S}_{git_commit_short}.onnx这样每次产出的模型文件名自带版本和时间信息不会出现“最后跑出来的模型不知道是哪个版本”的问题。更进一步我建议在数据库里建一张模型版本记录表CREATE TABLE model_versions ( id BIGSERIAL PRIMARY KEY, model_name VARCHAR(100) NOT NULL, version VARCHAR(50) NOT NULL, git_commit VARCHAR(50), metrics JSONB, artifact_path VARCHAR(500), status VARCHAR(20) DEFAULT staging, -- staging/production/retired created_at TIMESTAMPTZ DEFAULT now(), UNIQUE(model_name, version) );部署时API服务通过环境变量或配置中心指定要加载哪个版本的模型而不是硬编码模型路径。这样做的价值在于a/b测试时两个服务实例可以分别加载不同版本模型出问题时一秒钟就能回滚到上一个稳定版本。6.2 服务部署Docker化与滚动发布AI服务部署我全程用Docker保证了“本地能跑”和“线上能跑”的一致性。一个精简的Dockerfile示例FROM python:3.11-slim as builder ENV POETRY_VERSION1.7.1 RUN pip install poetry${POETRY_VERSION} WORKDIR /app COPY pyproject.toml poetry.lock ./ RUN poetry config virtualenvs.create false \ poetry install --no-root --only main FROM python:3.11-slim COPY --frombuilder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY --frombuilder /usr/local/bin /usr/local/bin WORKDIR /app COPY . . EXPOSE 8000 CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]这里用了多阶段构建第一层安装依赖第二层只拷贝site-packages镜像体积能小很多。注意--workers 4的选择逻辑——我先用nproc看CPU核数再用压测工具我用的locust测出4个worker在目标并发下的CPU和响应时间表现最后定下来。部署到服务器上我用Docker Compose编排应用服务和基础设施。新增一个服务条目api: build: . ports: - 8000:8000 environment: DATABASE_URL: postgresql://ai_app:ai_app_passwordpostgres:5432/ai_platform REDIS_URL: redis://redis:6379/0 MODEL_VERSION: prod_20240521_v2 depends_on: postgres: condition: service_healthy redis: condition: service_healthy restart: unless-stoppedMODEL_VERSION这个环境变量就是上文说的模型版本管理在部署层的落地。更新模型时只改这个变量、重启服务就能完成模型切换如果新模型有问题改回旧版本号再重启回滚瞬间完成。滚动发布我直接用最朴素的方案先起一个新版本容器跑健康检查通过后用Nginx把流量切过去再停掉旧容器。这套流程配合Github Actions或者Jenkins十几行配置就能实现不需要引入K8s这种重型武器——从零起步的项目复杂度要一点一点加。6.3 任务队列异步处理慢任务的基石AI系统里总有异步场景知识库更新、文档向量化、批量预测、消息通知。这些任务如果在Web进程里同步执行会直接阻塞请求响应。我的选择是Celery Redis。Celery配置非常简洁from celery import Celery celery_app Celery( ai_tasks, brokerredis://localhost:6379/1, backendredis://localhost:6379/2 ) celery_app.conf.update( task_serializerjson, result_serializerjson, accept_content[json], timezoneAsia/Shanghai, enable_utcTrue, worker_max_tasks_per_child200, # 防止任务内累积内存泄漏 task_time_limit300, # 单个任务最长5分钟 ) celery_app.task def embed_document_batch(doc_ids: list[int]): # 批量向量化逻辑 ...那个worker_max_tasks_per_child200是经验之谈。AI任务经常涉及加载模型、处理大文本内存碎片化很快。限制每个worker子进程处理200个任务后重启能有效避免内存持续膨胀导致的OOM。类似的task_time_limit防止个别卡死任务占着worker不放。调用方式也很简单embed_document_batch.delay(doc_ids[1, 2, 3])异步任务丢进队列后立即返回Web请求不会被阻塞。Celery worker单独以容器方式运行可以独立扩缩容——如果发现向量化任务积压多开几个worker容器就能缓解跟API服务的扩容互不影响。7. 可观测性当AI系统出问题时怎么快速定位7.1 日志、指标、追踪三件套AI系统出故障时最大的痛苦在于“不知道问题出在哪一段”——是数据不对模型输出异常还是依赖服务超时要回答这个问题必须在系统建设初期就搭好可观测体系具体就是我常说的日志、指标、追踪三件套。日志我全部用结构化格式JSON每行日志里带上timestamp、level、service、request_id、message和自定义字段。这样在ELK里可以直接按request_id把一条链路的所有日志串起来。一条标准日志示例{timestamp:2024-05-21T10:30:12.345Z,level:INFO,service:api,request_id:a3f9c2,event:inference_completed,model_version:prod_20240521_v2,latency_ms:45,status:success}指标用Prometheus收集。我曝光四个核心指标请求量QPS、时延分布P50/P95/P99、错误率、模型推理时延。FastAPI接入Prometheus客户端后几行代码就能完成埋点from prometheus_client import Counter, Histogram REQUESTS Counter(http_requests_total, Total HTTP requests, [method, path, status]) LATENCY Histogram(http_request_duration_seconds, HTTP request latency, [method, path]) app.middleware(http) async def metrics_middleware(request, call_next): start time.perf_counter() response await call_next(request) duration time.perf_counter() - start REQUESTS.labels(request.method, request.path, response.status_code).inc() LATENCY.labels(request.method, request.path).observe(duration) return response追踪主要看外部依赖的调用链。AI链路里尤其要盯的是向量检索耗时、LLM调用耗时、下游服务耗时。用OpenTelemetry可以自动埋点但小团队我建议先手动打点每个环节都记录开始时间、结束时间、耗时汇总成一个trace_id下的结构化日志。等系统复杂到需要跨服务追踪时再上完整的OpenTelemetry体系。7.2 监控告警什么样的告警才不会打扰到你告警配置的核心哲学是宁可漏报不要误报。与其频繁被打断去处理“无关紧要的告警”不如把告警阈值调到真正会出问题的那一刻。我目前保留的告警规则少得可怜但每一条都意义明确告警项触发条件重要程度P95时延超过阈值P95响应时间 2s持续3分钟高请求错误率上升5XX错误率 1%持续5分钟高模型变化监控模型版本变更却无对应部署记录中队列积压Celery任务队列长度 5000持续10分钟中内存健康容器内存使用率 85%持续15分钟中Grafana里配置告警规则方式就是“在仪表板的对应Panel上配置Threshold”。有一个技巧告警消息里一定要带上链接到仪表板和最近日志查询入口否则值班同学收到告警还要到处找系统入口耽误时间。7.3 一套实用的AI系统健康度打分模型我基于实际运维经验总结了一个简洁的系统健康度评估模型可以自动化给AI系统打分健康分 0.3 * 接口可用性分 0.3 * 响应速度分 0.2 * 业务效果分 0.2 * 数据新鲜度分接口可用性分(1 - 5XX错误率) * 100。低于95分告警。响应速度分基于P95时延映射P95 800ms给100分每增加200ms扣10分。业务效果分线下定期评估集推理结果对比基准版本的指标变化好于基准加分否则扣分。数据新鲜度分知识库最后更新时间距当前时间超过预设周期如7天则扣分。这套打分模型我会每天定时跑一次生成一条“今日系统健康度”日报发到团队群里。它不会直接告警但能让所有人对系统运行状态有个全局感知并且能在业务指标变差之前提前发现数据停留、模型退化等苗头。8. 典型问题实录我从这些坑里爬出来的8.1 ONNX导出时的动态轴报错第一次导出ONNX时我遇到了RuntimeError: Failed to export an ONNX attribute axes...这类报错。原因是模型内部有固定维度的操作比如位置编码的arange张量导出时需要指定动态轴但某些算子不支持。当时最有效的排查手段是用onnxruntime的onnxruntime.transformers优化器打印模型输入输出信息以及检查每个算子的版本是否老化。解决方案比想象中简单把opset_version升到14以上或者对模型做简化处理用onnxsim工具移除训练相关的动态控制流。多数算子兼容问题都能靠这两个手段解决。8.2 pgvector索引失效导致检索全表扫描有段时间检索延迟从20ms暴增到2秒查了PostgreSQL的执行计划才发现IVFFlat索引没有被用到。原因是数据量变化后索引的lists参数不再合适。pgvector的IVFFlat索引需要一次抽样训练建索引时会重新聚类但数据量增长后原来指定的lists参数对应的聚类效果变差。解决方法是重新建索引并调整参数DROP INDEX idx_document_chunks_embedding; CREATE INDEX idx_document_chunks_embedding ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists 200);另外检查是否用了参数化查询。如果你传入的向量不是常量而是参数有的数据库版本有query plan cache问题会退化成顺序扫描。当时最后定位到的原因很简单数据量从30万涨到了120万lists100严重不够用重建为200后性能恢复。这事的教训是索引参数不是建完就完事的要随着数据量增长定期评估。8.3 LLM接口超时但HTTP状态码还是200这是最隐蔽的一个坑。我们的调用方曾经反馈“接口很慢但不报错”排查后发现LLM生成回答耗时超过我设置的上游超时时间但底层HTTP客户端吞掉了超时异常返回了空内容被上层误判为“成功但空回复”。追查代码时发现当时用了requests库的默认行为超时设置只对连接生效对读取响应不生效。修复方式很直接requests.post(url, jsondata, timeout(3.05, 30))timeout元组的第一个值是连接超时第二个值是读取超时。这个参数强烈建议所有调用外部API的代码里都显式设置否则线上一定会出现“等待三分钟才崩溃”的慢请求。8.4 多进程环境下Redis连接数爆了Celery和API都连Redis默认连接池参数不调优时高并发下会出现Cannot assign requested address报错。原因就是底层TCP连接数达到系统限制。解决方式两个一是给Redis客户端配置合适的连接池上限二是服务容器里设置ulimit -n提高文件描述符上限。我最终的配置是redis_client redis.Redis( hostlocalhost, port6379, max_connections200, socket_connect_timeout3, socket_timeout3 )这块的经验是任何系统加了并发压测运行几分钟后看系统日志出现的连接类报错基本都能归因于连接池或系统文件描述符限制。8.5 模型精度线上与离线不一致遇到过最诡异的问题同一个模型线下评测F10.92线上抽样显示F10.76。反复验证后发现问题出在数据预处理不一致——离线代码里做了一遍文本清洗线上推理服务的预处理函数没有同步这个逻辑。这类问题的根因在于“代码漂移”。这次之后我在代码库加了一道硬性检查训练和推理必须共用同一个preprocess模块并且模型产物里除了权重还要包含一份preprocess_config.json记录清洗规则、分词参数、归一化参数。加载模型时服务端读取这份配置并初始化预处理这样线上线下就永远一致了。9. 从小项目到持续演进工程化没有终点如果你是从零起步搭AI系统上面这几个章节走完已经能形成一个可以稳定运行的完整闭环。不过在我的经验里这还只是工程化的第一阶段。真正让AI工程长期健康迭代的还有几个“软性”却关键的习惯。实验记录规范化。我要求每个团队成员每次做实验都必须记录数据版本、代码commit、模型参数、测评结果、结论。哪怕只是调了个学习率也要留痕。开始觉得繁琐但当一个月后你发现模型效果下降、需要回溯是哪次改动导致的整洁的实验记录能救你命。模型效果回归测试。每次模型上线前必须跑一遍固定的评测集。这个评测集要有覆盖面、要长期维护特别要包含历次踩坑的边界场景。AI模型不像传统软件有明确的“功能正确与否”回归测试就是AI系统的“自动化测试”。定期体检AI系统的健康度。上面说的健康度打分每天跑、每周看趋势、每月复盘。我建议选定一个固定时间比如每周一上午把上周的健康分趋势和业务指标放在一起看这个习惯能让你比业务用户更早发现系统的微妙退化。做AI工程最关键的觉悟是模型能力决定上限工程能力决定下限。很聪明的一个模型部署不好、监控缺失、无法迭代最终也发挥不出价值相反一个中规中矩的模型只要工程链路扎实、迭代顺畅用户感受到的稳定性和可用性会非常好。所以不用急着追最前沿的模型架构先把工程地基打牢。地基之上模型的每一次进步都能稳稳地变成用户价值。