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多worker,Redis滑动窗口限流 |
| 系统故障可定位 | 日志与指标可观测 | 结构化日志 + 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最常见的灾难是:开发环境装的是numpy==1.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_tensors="pt") 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_version=17 )这里有个关键决策: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_type=QuantType.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_length=1, max_length=512) top_k: int = Field(5, ge=1, le=20) 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(lifespan=lifespan) @app.post("/v1/inference", response_model=InferenceResponse) 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_code=500, detail=str(e)) latency = (time.perf_counter() - start) * 1000 return InferenceResponse(result=result, latency_ms=latency) def preprocess(text: str): # 假设tokenizer在服务启动时也初始化了 tokens = app.state.tokenizer(text, return_tensors="np") 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), timeout=3.0 ) except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="Inference timeout")这样保证任何情况下单个请求最多占用3秒,不会出现连接被拖死的情况。
5.4 RAG管线的实现:检索增强生成的工程核心
如果你做的是LLM相关的应用,那RAG(检索增强生成)管线基本是标配。我实现的精简但完整的RAG链路如下:
离线索引构建。先对海量知识文档做切分,我这里用的是递归字符切分器,按500字符一块、80字符重叠:
from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, 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),生成的向量直接插入pgvector:
CREATE 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 = f"models/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_VERSION=1.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 --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY --from=builder /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_password@postgres: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", broker="redis://localhost:6379/1", backend="redis://localhost:6379/2" ) celery_app.conf.update( task_serializer="json", result_serializer="json", accept_content=["json"], timezone="Asia/Shanghai", enable_utc=True, worker_max_tasks_per_child=200, # 防止任务内累积内存泄漏 task_time_limit=300, # 单个任务最长5分钟 ) @celery_app.task def embed_document_batch(doc_ids: list[int]): # 批量向量化逻辑 ...那个worker_max_tasks_per_child=200是经验之谈。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万,lists=100严重不够用,重建为200后性能恢复。
这事的教训是:索引参数不是建完就完事的,要随着数据量增长定期评估。
8.3 LLM接口超时但HTTP状态码还是200
这是最隐蔽的一个坑。我们的调用方曾经反馈“接口很慢但不报错”,排查后发现:LLM生成回答耗时超过我设置的上游超时时间,但底层HTTP客户端吞掉了超时异常,返回了空内容,被上层误判为“成功但空回复”。追查代码时发现,当时用了requests库的默认行为:超时设置只对连接生效,对读取响应不生效。
修复方式很直接:
requests.post(url, json=data, timeout=(3.05, 30))timeout元组的第一个值是连接超时,第二个值是读取超时。这个参数强烈建议所有调用外部API的代码里都显式设置,否则线上一定会出现“等待三分钟才崩溃”的慢请求。
8.4 多进程环境下Redis连接数爆了
Celery和API都连Redis,默认连接池参数不调优时,高并发下会出现Cannot assign requested address报错。原因就是底层TCP连接数达到系统限制。解决方式两个:一是给Redis客户端配置合适的连接池上限,二是服务容器里设置ulimit -n提高文件描述符上限。
我最终的配置是:
redis_client = redis.Redis( host="localhost", port=6379, max_connections=200, socket_connect_timeout=3, socket_timeout=3 )这块的经验是,任何系统加了并发压测,运行几分钟后看系统日志出现的连接类报错,基本都能归因于连接池或系统文件描述符限制。
8.5 模型精度线上与离线不一致
遇到过最诡异的问题:同一个模型,线下评测F1=0.92,线上抽样显示F1=0.76。反复验证后发现问题出在数据预处理不一致——离线代码里做了一遍文本清洗,线上推理服务的预处理函数没有同步这个逻辑。这类问题的根因在于“代码漂移”。
这次之后,我在代码库加了一道硬性检查:训练和推理必须共用同一个preprocess模块,并且模型产物里除了权重,还要包含一份preprocess_config.json,记录清洗规则、分词参数、归一化参数。加载模型时,服务端读取这份配置并初始化预处理,这样线上线下就永远一致了。
9. 从小项目到持续演进:工程化没有终点
如果你是从零起步搭AI系统,上面这几个章节走完,已经能形成一个可以稳定运行的完整闭环。不过在我的经验里,这还只是工程化的第一阶段。真正让AI工程长期健康迭代的,还有几个“软性”却关键的习惯。
实验记录规范化。我要求每个团队成员每次做实验,都必须记录:数据版本、代码commit、模型参数、测评结果、结论。哪怕只是调了个学习率,也要留痕。开始觉得繁琐,但当一个月后你发现模型效果下降、需要回溯是哪次改动导致的,整洁的实验记录能救你命。
模型效果回归测试。每次模型上线前,必须跑一遍固定的评测集。这个评测集要有覆盖面、要长期维护,特别要包含历次踩坑的边界场景。AI模型不像传统软件有明确的“功能正确与否”,回归测试就是AI系统的“自动化测试”。
定期体检AI系统的健康度。上面说的健康度打分,每天跑、每周看趋势、每月复盘。我建议选定一个固定时间,比如每周一上午,把上周的健康分趋势和业务指标放在一起看,这个习惯能让你比业务用户更早发现系统的微妙退化。
做AI工程,最关键的觉悟是:模型能力决定上限,工程能力决定下限。很聪明的一个模型,部署不好、监控缺失、无法迭代,最终也发挥不出价值;相反,一个中规中矩的模型,只要工程链路扎实、迭代顺畅,用户感受到的稳定性和可用性会非常好。所以不用急着追最前沿的模型架构,先把工程地基打牢。地基之上,模型的每一次进步都能稳稳地变成用户价值。