ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

从零搭建AI工程能力:环境、部署、服务化与迭代闭环实战

从零搭建AI工程能力:环境、部署、服务化与迭代闭环实战 1. 从零搭建AI工程能力到底在搭什么很多人第一次看到“ai-engineering-from-scratch”这个标题脑子里浮现的画面是打开一个代码编辑器从零开始手写一个Transformer。我一开始也这么以为后来真正动手做了几轮之后才发现这个方向的核心根本不是“手搓模型”而是把AI能力从实验室状态变成可交付、可维护、可迭代的工程系统。这两件事之间的距离比大多数人想象的要大得多。举个具体的例子。你在本地用几行代码调通了一个开源模型输入一句话它能给你返回一段像模像样的回答这时候你会觉得“AI不过如此”。但当你需要让它每天处理上万条请求、需要控制响应延迟在可接受范围内、需要在模型输出不稳定时自动兜底、需要追踪每一次调用的成本和效果——你会发现之前那几行代码连整个系统的百分之五都不到。剩下的百分之九十五就是AI工程要解决的问题。这个方向适合什么人我的判断是三类人最值得投入第一类是有一定编程基础但没接触过AI系统搭建的后端或全栈开发者你们缺的不是编码能力而是对AI组件特性的理解第二类是做数据分析或算法出身、想把自己的模型真正推到生产环境的人你们缺的是工程化的思维和工具链第三类是对AI应用有想法但不知道技术边界在哪的产品或创业者你们需要知道哪些事现在能做、哪些事做了会翻车。我自己的路径是从第二类往第一类靠。早期我只会写训练脚本模型跑出指标就觉得任务完成了。后来被现实反复教育推理服务的吞吐量上不去、模型版本更新后线上效果回退、用户输入超出预期分布时系统直接崩溃。这些问题没有一个是调参能解决的全部要靠工程手段。所以这篇内容我会围绕“从零搭建AI工程能力”这个主题把我在实际项目中踩过的坑、总结的方法、验证过的工具链尽可能完整地摊开来讲。2. 环境底座别急着写模型代码先把这层打牢2.1 为什么环境管理是AI工程的第一道分水岭我见过太多项目死在环境问题上。不是模型不行是跑不起来。Python版本冲突、CUDA驱动不匹配、依赖包版本打架、不同项目之间互相污染——这些问题在纯软件开发里也有但在AI工程里被放大了好几倍因为AI技术栈的依赖链条特别长从底层驱动到深度学习框架到上层应用库任何一环出问题都会导致整个系统不可用。我的做法是从项目第一天就强制使用隔离环境。具体来说Python层面用venv或者conda创建独立环境这是最基本的。但仅仅这样还不够因为AI项目经常需要特定版本的CUDA和cuDNN而这些是系统级的。我的经验是如果团队规模不大、机器数量有限直接用Docker把整个运行环境打包包括Python版本、CUDA版本、所有依赖包一次性解决。如果团队有成熟的Kubernetes集群那就把环境做成镜像通过容器编排来管理。FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04 RUN apt-get update apt-get install -y python3.10 python3-pip RUN pip install torch2.1.0 transformers4.35.0 fastapi0.104.0 uvicorn0.24.0 WORKDIR /app COPY . /app CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]这个Dockerfile看起来简单但里面有几个关键决策。基础镜像选nvidia/cuda而不是普通的ubuntu是因为AI推理需要GPU支持CUDA运行时必须预装。PyTorch版本锁定在2.1.0而不是用最新版是因为新版本可能引入不兼容的API变更生产环境要的是稳定而不是尝鲜。FastAPI和Uvicorn的组合是我实测下来在AI服务场景里最顺手的异步支持好、启动快、和Python生态融合自然。注意不要在生产环境的Dockerfile里用pip install torch不带版本号。我吃过这个亏某次自动构建拉到了最新版结果和现有的模型权重不兼容线上服务直接挂了两个小时。2.2 硬件选型不是越贵越好而是匹配需求AI工程绕不开硬件。但我的观点很明确大多数场景下你不需要最顶级的GPU。我见过创业团队一上来就买A100结果模型参数量才几亿利用率不到百分之十纯属浪费。选硬件的逻辑应该反过来先确定你的模型规模和并发量再倒推需要什么卡。一个粗略的估算方法是模型推理时显存占用大约是参数量的1.2到1.5倍FP16精度下再加上输入输出的缓存。比如一个7B参数的模型FP16推理大概需要14到16GB显存那RTX 4090的24GB就够用没必要上A100。场景推荐显存典型卡型说明小模型推理1B参数8-12GBRTX 3060/4060成本低适合验证和轻量服务中等模型推理1B-13B16-24GBRTX 4090/RTX 6000 Ada性价比最高的区间大模型推理13B-70B40-80GBA100/A800需要多卡或量化训练/微调80GBA100/H100训练比推理吃资源得多这张表是我根据实际项目经验整理的不是理论值。实际选型时还要考虑功耗、散热、机箱空间这些物理限制。我有个朋友在办公室塞了四张3090结果夏天室温直接飙到三十五度最后不得不加装工业风扇。2.3 依赖管理的隐藏陷阱AI项目的依赖管理有个特殊难点不同组件对同一依赖的版本要求可能冲突。比如你的推理服务需要transformers4.30但某个微调工具只支持transformers4.28。这种时候硬装是装不上的。我的处理策略是分层管理。核心推理链路用一套固定的依赖版本单独放在一个环境里。训练和微调工具用另一套环境通过API或者文件交换数据和模型权重。这样虽然多了一层交互成本但避免了依赖地狱。如果非要在同一个环境里解决那就用pip-tools或者poetry做依赖解析把冲突显式暴露出来而不是等到运行时才发现。3. 模型接入从调用API到自己部署的决策逻辑3.1 什么时候该用外部API什么时候该自己部署这是AI工程里最关键的决策之一没有标准答案但有清晰的判断框架。我的经验是看三个维度数据敏感性、调用频率、定制化需求。数据敏感度低、调用频率不高、不需要微调的场景直接用外部API是最优解。省去了硬件采购、环境维护、模型更新的全部麻烦按量付费起步成本极低。我早期做原型验证时全部走这条路一个下午就能跑通完整流程。但一旦涉及敏感数据或者每天调用量超过一定阈值自己部署就开始划算了。这个阈值我粗略估算在每天五千到一万次调用左右具体取决于API的定价和你自己的硬件成本。自己部署的另一个好处是完全的控制权你可以决定什么时候更新模型、用什么量化策略、怎么处理异常输入这些在外部API上都是黑盒。# 外部API调用的典型封装 import httpx class LLMClient: def __init__(self, api_key: str, base_url: str): self.client httpx.Client( base_urlbase_url, headers{Authorization: fBearer {api_key}}, timeout30.0 ) def generate(self, prompt: str, max_tokens: int 512) - str: response self.client.post(/v1/chat/completions, json{ model: your-model-name, messages: [{role: user, content: prompt}], max_tokens: max_tokens, temperature: 0.7 }) response.raise_for_status() return response.json()[choices][0][message][content]这个封装看起来简单但有几个细节值得注意。timeout必须设置否则遇到网络问题会一直挂起。raise_for_status()要加不然错误响应会被当成正常结果处理。temperature参数要根据场景调整需要确定性输出时设成0需要创意时设成0.7到1.0。3.2 自己部署时的推理框架选择决定自己部署之后下一个问题是选什么推理框架。我实测过的主流方案有几种各有适用场景。原生PyTorch是最直接的方式适合快速验证和模型结构特殊的场景。缺点是性能没有优化显存占用大并发能力弱。我一般只在调试阶段用。vLLM是我目前最推荐的方案尤其适合中小规模部署。它的PagedAttention机制对显存管理做了深度优化吞吐量比原生PyTorch高好几倍。安装也简单基本是pip install vllm然后几行代码就能起服务。from vllm import LLM, SamplingParams llm LLM(modelyour-model-path, tensor_parallel_size1) sampling_params SamplingParams(temperature0.7, max_tokens512) outputs llm.generate([你的输入文本], sampling_params) print(outputs[0].outputs[0].text)**TGIText Generation Inference**是另一个成熟方案HuggingFace出品和transformers生态结合紧密。它的优势在于对HuggingFace模型格式的支持最完善如果你用的模型来自HuggingFace HubTGI基本是开箱即用。TensorRT-LLM是NVIDIA的官方方案性能最强但配置最复杂需要把模型转换成TensorRT引擎对模型结构有要求。我一般只在追求极致性能且团队有足够工程能力时才推荐。提示选择推理框架时不要只看benchmark上的吞吐量数字。实际项目中部署的便捷性、社区活跃度、出问题时的排查难度这些因素往往比峰值性能更重要。3.3 模型量化的取舍量化是降低显存占用和提升推理速度的常用手段但很多人对它的理解停留在“精度换速度”这个层面。实际上量化的策略选择很讲究。FP16是最安全的量化方式精度损失几乎可以忽略显存占用是FP32的一半。如果你的显存够用优先选FP16。INT8量化能把显存再降一半精度损失在大多数任务上不明显但在需要精细推理的任务上比如数学计算、代码生成可能会有可感知的下降。INT4量化进一步压缩但精度损失开始变得明显。我一般只在显存极度受限、且任务对精度要求不高时才用。量化的实现方式也有区别。训练后量化PTQ简单快速不需要重新训练。量化感知训练QAT效果好但需要训练资源。我大多数场景下用PTQ就够了用bitsandbytes或者GPTQ工具链都能做。4. 服务化把模型变成别人能用的东西4.1 API设计不只是包一层HTTP把模型包成HTTP接口这件事看起来简单但设计好坏直接影响后续的维护成本。我见过太多项目在这层偷懒最后接口混乱到没人敢改。我的API设计原则是输入输出结构化、错误处理显式化、版本管理前置化。输入不要只接受一个裸字符串而是用结构化的JSON把可能变化的参数都显式列出来。输出也不要只返回一个字符串把token用量、模型版本、耗时这些元信息带上后续排查问题时会感谢自己。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import time app FastAPI() class GenerateRequest(BaseModel): prompt: str max_tokens: int 512 temperature: float 0.7 model_version: str v1 class GenerateResponse(BaseModel): text: str model_version: str latency_ms: float token_usage: int app.post(/generate, response_modelGenerateResponse) async def generate(req: GenerateRequest): start time.time() try: result await model_pool.generate(req) except Exception as e: raise HTTPException(status_code500, detailstr(e)) return GenerateResponse( textresult.text, model_versionreq.model_version, latency_ms(time.time() - start) * 1000, token_usageresult.token_count )这个结构里model_version字段是关键。它让你可以在不破坏现有接口的情况下切换模型版本也让你在出问题时能快速定位是哪个版本的表现异常。4.2 并发处理AI服务的特殊挑战AI服务的并发处理和普通Web服务有本质区别。普通Web服务的请求处理时间通常在毫秒级而AI推理动辄几百毫秒到几秒。这意味着同样的并发量下AI服务需要更多的资源也更容易出现请求堆积。我的处理策略是异步队列超时三层防护。异步框架用FastAPI或者Tornado让请求处理不阻塞主线程。队列用Redis或者内存队列做缓冲防止瞬时高峰打垮服务。超时设置要合理太短会导致正常请求被误杀太长会让异常请求拖垮整个系统。import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers4) async def generate_with_timeout(prompt: str, timeout: float 10.0): loop asyncio.get_event_loop() try: result await asyncio.wait_for( loop.run_in_executor(executor, model.generate, prompt), timeouttimeout ) return result except asyncio.TimeoutError: return {error: generation timeout, fallback: 请稍后重试}这段代码的核心思路是把同步的模型推理放到线程池里执行然后用asyncio.wait_for控制超时。max_workers的设置要根据GPU数量和显存来定不是越大越好。我一般从GPU数量乘以2开始试观察GPU利用率和请求延迟再调整。4.3 监控与日志上线只是开始服务上线之后真正的挑战才开始。你需要知道服务现在是否健康、响应时间是否正常、有没有异常请求、模型输出质量有没有下降。这些都需要监控和日志来支撑。我必配的监控指标包括请求量、响应延迟分布、错误率、GPU利用率、显存占用、队列长度。这些指标用Prometheus采集Grafana展示基本是行业标准组合。日志方面我建议把每次请求的输入、输出、耗时、模型版本都记录下来。但要注意隐私和存储成本输入输出可能需要脱敏日志保留时间要根据合规要求设定。我一般保留最近七天的详细日志更早的只保留聚合统计。注意不要等到出问题才想起来加监控。我有个项目上线三个月没做监控某天用户反馈“回答变差了”结果因为没有历史数据完全无法定位是模型问题、数据问题还是代码问题。5. 迭代闭环让系统越用越好5.1 效果评估不能只靠感觉AI系统最麻烦的地方在于它的输出是自然语言没有简单的对错判断。你不能像传统软件那样写个断言来判断输出是否正确。所以效果评估必须专门设计。我的做法是自动指标人工抽检用户反馈三层结合。自动指标用BLEU、ROUGE这类文本相似度算法快速筛出明显异常的case。人工抽检定期做由标注人员对输出质量打分。用户反馈通过界面上的点赞点踩按钮收集这是最真实的信号。from rouge_score import rouge_scorer scorer rouge_scorer.RougeScorer([rouge1, rougeL], use_stemmerTrue) def evaluate_output(prediction: str, reference: str) - dict: scores scorer.score(reference, prediction) return { rouge1: scores[rouge1].fmeasure, rougeL: scores[rougeL].fmeasure }自动指标只能作为参考不能作为唯一标准。我见过ROUGE分数很高但实际读起来完全不通顺的输出也见过分数一般但用户很满意的回答。所以自动指标用来做粗筛和趋势监控真正的质量判断还是要靠人工和用户反馈。5.2 数据回流把使用过程变成训练素材AI系统有一个传统软件没有的优势每一次使用都在产生新的数据。这些数据如果利用好就是持续优化的燃料。我的做法是建立一个数据回流管道。用户每次请求的输入、模型的输出、用户的反馈点赞点踩、修改后的文本全部收集起来定期做清洗和标注然后用于微调或者构建评测集。这个管道的关键是闭环速度。从数据产生到用于优化周期越短越好。我一般设定为每周一次小循环每月一次大循环。小循环用新数据做增量微调大循环重新评估整体效果并决定是否更新主模型。5.3 版本管理与回滚模型版本管理是AI工程里容易被忽视但极其重要的一环。每次模型更新都可能带来效果波动没有好的版本管理出了问题连回滚都做不到。我的做法是模型文件、配置、代码三者版本绑定。每次发布新版本把模型权重、推理配置、服务代码打包成一个版本号记录在案。线上服务通过版本号来加载对应的资源。回滚时只需要切换版本号不需要重新构建。# 版本目录结构示例 models/ v1.0.0/ weights/ config.json service.py v1.1.0/ weights/ config.json service.py这个结构看起来简单但实际执行时需要配合CI/CD流程。每次代码合并到主分支自动构建新版本镜像并打标签。部署时指定标签回滚时切回旧标签。整个过程自动化减少人为失误。6. 那些没人告诉你但一定会踩的坑6.1 显存泄漏最隐蔽的杀手显存泄漏是AI服务最头疼的问题之一。表现是服务运行一段时间后显存逐渐占满最终OOM崩溃。重启能暂时解决但过一段时间又会复现。根本原因通常是PyTorch的计算图没有正确释放。在推理场景下如果忘记用torch.no_grad()包裹前向计算PyTorch会保留计算图用于反向传播显存就不会释放。# 错误写法显存会持续增长 def generate(prompt): output model(prompt) return output # 正确写法显存稳定 def generate(prompt): with torch.no_grad(): output model(prompt) return output除了no_grad还要注意缓存清理。HuggingFace的transformers库会缓存一些中间结果长时间运行需要定期清理。我一般在每次请求结束后调用torch.cuda.empty_cache()虽然会带来一点性能开销但能有效防止显存碎片化。6.2 输入长度超限最常见的线上事故用户输入的长度是不可控的。你永远不知道用户会粘贴多长的文本进来。如果模型的最大输入长度是2048个token而用户输入了5000个token直接调用会报错或者产生截断。我的处理策略是前置校验优雅降级。在API入口处先做token计数超过限制的请求直接返回明确的错误信息告诉用户输入过长以及最大允许长度。如果业务上不能拒绝那就做截断或者分段处理但要明确告知用户结果可能不完整。from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(your-model) def validate_input(text: str, max_tokens: int 2048) - tuple[bool, str]: tokens tokenizer.encode(text) if len(tokens) max_tokens: return False, f输入长度{len(tokens)}超过限制{max_tokens}请缩短后重试 return True, 6.3 模型更新导致的效果回退新模型不一定比旧模型好。这是我在实际项目中被反复教育的一点。新模型可能在整体指标上更好但在某些特定场景下表现更差。如果没有完善的评测体系这种回退很难被发现。我的做法是每次模型更新前必须跑完整的回归测试。测试集要覆盖所有已知的重要场景包括边界case和之前出过问题的case。只有新模型在所有测试集上都不低于旧模型才允许上线。如果某些场景有下降需要评估影响范围再决定是否接受。6.4 并发下的输出串扰这个问题在多用户并发时特别容易出现。表现是A用户的请求收到了B用户的回答或者同一个请求收到了混合了两个回答的内容。根本原因通常是全局状态被并发修改。避免这个问题的原则是推理函数必须是纯函数不依赖任何全局可变状态。每次请求的上下文、缓存、中间变量都应该是局部的。如果必须用全局资源比如模型本身确保它是只读的。# 危险使用全局可变状态 current_user None def generate(prompt): global current_user current_user get_user() return model.generate(prompt, usercurrent_user) # 安全所有状态通过参数传递 def generate(prompt, user_context): return model.generate(prompt, useruser_context)7. 工具链与效率让工程能力可复制7.1 实验管理别再用Excel记结果了AI工程离不开实验。不同的模型、不同的参数、不同的数据配比组合起来可能有几十上百次实验。如果用Excel或者记事本记录很快就会乱套。我推荐用MLflow或者Weights Biases做实验管理。每次实验自动记录参数、指标、模型文件、代码版本可以随时对比不同实验的结果。MLflow可以本地部署数据完全自己掌控。WB是云端服务界面更友好但需要联网。import mlflow mlflow.set_experiment(my-ai-project) with mlflow.start_run(): mlflow.log_param(model_name, your-model) mlflow.log_param(temperature, 0.7) mlflow.log_metric(rouge_l, 0.45) mlflow.log_artifact(outputs/result.json)这几行代码就能把一次实验的完整信息记录下来。后续要复现或者对比直接查MLflow界面就行比翻聊天记录和Excel表格高效太多。7.2 配置管理把魔法数字从代码里赶出去AI项目里充满了各种参数模型路径、推理超时、最大token数、温度值、并发数。如果这些硬编码在代码里每次调整都要改代码、重新部署效率极低。我的做法是所有可调参数都放到配置文件里代码只负责读取。配置文件用YAML或者JSON格式支持环境变量覆盖方便在不同环境开发、测试、生产之间切换。# config.yaml model: path: /models/v1.0.0 max_tokens: 2048 temperature: 0.7 service: host: 0.0.0.0 port: 8000 timeout: 30 max_workers: 4 monitoring: enabled: true log_level: INFOimport yaml def load_config(path: str config.yaml) - dict: with open(path, r) as f: config yaml.safe_load(f) return config这个做法看起来简单但收益很大。调整参数不需要改代码不同环境用不同配置文件出问题时可以快速对比配置差异。7.3 自动化测试AI系统也需要很多人觉得AI系统没法做自动化测试因为输出不确定。但实际上大部分工程层面的问题是可以用自动化测试覆盖的。我必写的测试包括API接口的输入输出格式测试、超长输入的处理测试、并发请求的隔离测试、模型加载和卸载的测试、配置加载的测试。这些测试不关心模型输出内容是否正确只关心系统行为是否符合预期。import pytest from fastapi.testclient import TestClient from main import app client TestClient(app) def test_generate_basic(): response client.post(/generate, json{prompt: 你好}) assert response.status_code 200 assert text in response.json() def test_generate_too_long(): long_prompt 测试 * 5000 response client.post(/generate, json{prompt: long_prompt}) assert response.status_code 400 assert 超过限制 in response.json()[detail]这些测试跑起来很快但能在部署前拦住大部分低级错误。我一般把它们挂在CI流程里每次代码提交自动运行。8. 从能跑到好用我的个人经验总结做了这么多轮AI工程项目我最大的体会是工程能力比模型能力更稀缺。找一个会调模型的人不难找一个能把模型变成稳定服务的人很难。这个差距就是“ai-engineering-from-scratch”这个方向的价值所在。如果让我给刚入门的同行一个建议我会说不要一上来就追求最先进的模型和最复杂的架构。先用最简单的方案把整个链路跑通从输入到推理到输出到监控全部走一遍。跑通之后再针对瓶颈逐个优化。我见过太多人卡在模型选型上纠结几周结果连一个能用的服务都没搭起来。另一个经验是把每次踩坑都记录下来。我有个习惯每次解决一个线上问题就在文档里写清楚现象、原因、解决方案、预防措施。积累下来就是一份非常宝贵的内部知识库。新同事入职时看这份文档能避免重复踩坑。最后说一个容易被忽视的点成本意识。AI服务的成本不只是GPU还包括存储、网络、人力。我见过模型效果很好但成本高到无法商业化的项目。从第一天起就要关注每次调用的成本优化推理效率、合理设置缓存、选择合适的硬件这些工程决策直接影响项目的可持续性。这个方向还在快速演进新的工具和方案层出不穷。但底层的工程原则是相对稳定的隔离环境、结构化接口、完善监控、持续迭代。把这些基础打牢上层用什么新工具都能快速上手。
返回列表