ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

从零搭建AI工程能力:模型抽象、异步服务化与成本控制实战

从零搭建AI工程能力:模型抽象、异步服务化与成本控制实战 1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年“AI工程”这个词被说得太多了多到有点变味。招聘JD上写着“熟悉AI工程化落地”点进去一看要求会调三个API、会写Prompt、会用某个开源框架搭个Demo。说实话这不叫AI工程这叫“AI体验”。真正做过从零到一项目的人心里都清楚把一个模型从Jupyter Notebook里跑通到它能稳定地服务真实用户中间隔着的不是一行pip install而是一整套工程体系。ai-engineering-from-scratch这个标题我第一次看到的时候就觉得它戳中了痛点。市面上讲AI的教程分两类一类是纯算法向的推导反向传播、讲Transformer架构看完你懂原理但不会落地另一类是纯工具向的教你调LangChain、搭RAG跑起来很爽但一出问题就抓瞎因为你不知道底下发生了什么。而“from scratch”这个短语在工程语境里意味着一种态度——不依赖黑盒从最基础的组件开始一层一层把系统搭起来每一层都清楚它为什么存在、解决什么问题、边界在哪里。这篇内容适合谁看如果你已经会用Python能看懂基本的HTTP请求对AI模型有概念性的了解但每次想做一个完整的AI应用时总觉得东拼西凑、心里没底那这篇就是写给你的。我会按照一个真实项目的推进节奏把AI工程从零搭建的完整路径拆开讲从环境与项目骨架、数据管道、模型接入与抽象、服务化与并发、可观测性与评测一直到成本控制和迭代策略。每一块我都会说清楚“为什么这么设计”而不只是“怎么做”。我自己的背景是做了七八年后端和基础设施最近三年主要在做AI相关的工程落地。踩过的坑包括但不限于把模型调用写死在业务代码里导致换模型要改几十个文件、没有做超时和重试导致线上雪崩、评测集和训练数据泄漏导致指标虚高、Token成本失控一个月烧掉五位数。这些坑本质上都不是算法问题而是工程问题。所以这篇内容的核心不是教你某个框架的用法而是帮你建立一套能扛住真实流量的AI工程思维。2. 项目骨架与环境搭建先把地基打对2.1 为什么目录结构比框架选型更重要很多人做AI项目第一步是纠结用LangChain还是LlamaIndex用FastAPI还是Flask。我的经验是这些选择在项目初期的影响远没有你想的那么大真正影响后期维护成本的是目录结构和模块边界。一个典型的AI应用至少包含这几块数据接入层、模型调用层、业务逻辑层、服务接口层、评测与监控层。如果一开始不把这些边界划清楚代码写到三千行以后就是一团浆糊。我推荐的项目骨架大概长这样你可以直接抄project/ ├── configs/ # 配置文件按环境分 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── src/ │ ├── data/ # 数据加载、清洗、切分 │ ├── models/ # 模型客户端封装与抽象 │ ├── pipelines/ # 业务编排逻辑 │ ├── services/ # 对外服务接口 │ └── utils/ # 通用工具 ├── tests/ │ ├── unit/ │ └── integration/ ├── eval/ # 评测集与评测脚本 ├── scripts/ # 运维与数据处理脚本 └── pyproject.toml这个结构的关键在于models和pipelines的分离。models层只负责“怎么跟模型说话”pipelines层负责“什么时候跟模型说什么”。这样当你从某个云端模型切换到本地部署模型时只需要改models层的一个实现类业务代码一行不动。我见过太多项目把模型调用直接写在业务函数里结果换模型的时候要全局搜索替换改完还得重新测一遍所有流程。2.2 依赖管理与环境隔离的实操细节Python的依赖管理是个老生常谈的问题但在AI项目里尤其要命因为AI相关的库版本冲突特别严重。我的建议是坚决用pyproject.toml配合uv或者poetry不要用裸的requirements.txt。原因很简单AI项目经常需要区分训练依赖和推理依赖requirements.txt很难表达这种分组。用uv的话初始化大概是这样uv init ai-engineering-from-scratch cd ai-engineering-from-scratch uv add fastapi uvicorn pydantic httpx uv add --dev pytest ruff mypy这里有个细节值得说httpx而不是requests。因为AI应用大量涉及异步调用httpx原生支持async而requests是同步的。如果你后期要做并发调用多个模型或者多个工具用requests会把自己逼到用线程池的尴尬境地。这个选择在项目第一天做成本几乎为零但后期改起来很痛苦。环境隔离方面我强烈建议用容器化开发。不是说要你一开始就搞K8s而是用一个简单的Dockerfile把运行环境固定下来。AI项目最怕的就是“我本地能跑”这句话因为CUDA版本、Python版本、各种底层库的版本差异太大了。一个最小化的DockerfileFROM python:3.11-slim WORKDIR /app COPY pyproject.toml uv.lock ./ RUN pip install uv uv sync --frozen COPY src/ ./src/ CMD [uvicorn, src.services.main:app, --host, 0.0.0.0, --port, 8000]注意不要用python:latest一定要锁定具体版本。我踩过一次坑某次构建时基础镜像从3.11升到了3.12结果某个依赖库还没适配整个服务起不来排查了两个小时才发现是基础镜像变了。2.3 配置管理别把密钥写进代码配置管理是新手最容易忽视的环节。我见过有人把API Key直接写在代码里然后推到公开仓库也见过把所有配置硬编码在函数参数里改一个超时时间要重新部署。正确的做法是分层配置敏感信息走环境变量非敏感的业务配置走配置文件。用pydantic-settings可以很优雅地做到这一点from pydantic_settings import BaseSettings class Settings(BaseSettings): model_api_key: str model_base_url: str https://api.example.com request_timeout: int 30 max_retries: int 3 class Config: env_file .env env_prefix APP_这样本地开发用.env文件线上用环境变量注入代码完全不用改。而且pydantic会在启动时做类型校验配置写错了直接启动失败比运行到一半才报错要好得多。3. 模型接入层把不确定性关进笼子3.1 为什么必须做模型抽象模型调用是AI工程里最不可控的一环。网络会抖、服务会限流、模型会抽风返回奇怪的东西。如果你把模型调用散落在业务代码各处那你就等着半夜被报警叫醒吧。我的做法是做一个统一的模型客户端抽象把所有的不确定性都封装在这一层。抽象的核心是一个基类定义清楚输入输出契约from abc import ABC, abstractmethod from pydantic import BaseModel class ModelRequest(BaseModel): prompt: str temperature: float 0.7 max_tokens: int 1024 class ModelResponse(BaseModel): text: str input_tokens: int output_tokens: int latency_ms: int class BaseModelClient(ABC): abstractmethod async def generate(self, request: ModelRequest) - ModelResponse: ...这个抽象看起来简单但它带来的好处是巨大的。首先业务层只依赖BaseModelClient不关心背后是哪个厂商的模型。其次ModelResponse里强制带上Token数和延迟这两个数据是后期成本核算和性能优化的基础如果一开始不采集后面想补就得改所有调用点。3.2 超时、重试与降级的正确姿势模型调用的超时设置是个技术活。设太短正常的长文本生成会被误杀设太长一个卡住的请求会拖垮整个服务。我的经验值是普通对话类请求30秒长文本生成类请求120秒流式输出的话首Token超时单独设10秒。重试策略要区分错误类型。网络超时、5xx错误可以重试但4xx错误比如参数错误、鉴权失败重试没有意义只会浪费配额。而且重试必须带指数退避否则限流的时候你越重试越糟import asyncio from tenacity import retry, stop_after_attempt, wait_exponential retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retrylambda e: isinstance(e, (TimeoutError, ServerError)) ) async def call_model_with_retry(client, request): return await client.generate(request)降级策略是很多人忽略的。当模型服务完全不可用时你的应用不应该直接报错白屏而应该有一个兜底方案。最简单的降级是返回一个预设的友好提示好一点的降级是切换到备用模型再复杂一点的是走缓存或者规则引擎。降级方案不需要很完美但必须有这是线上服务的基本素养。3.3 流式输出的工程实现流式输出是AI应用体验的关键但它的工程复杂度比非流式高一个量级。核心难点在于流式响应一旦开始就不能再改HTTP状态码了所以错误处理必须在流开始之前完成。我的做法是先用一个轻量的预检请求确认模型可用再开始流式传输。在FastAPI里实现流式输出大概是这样from fastapi.responses import StreamingResponse async def stream_generator(client, request): try: async for chunk in client.stream(request): yield fdata: {chunk}\n\n except Exception as e: yield fdata: [ERROR] {str(e)}\n\n finally: yield data: [DONE]\n\n app.post(/chat/stream) async def chat_stream(request: ModelRequest): return StreamingResponse( stream_generator(client, request), media_typetext/event-stream )提示流式接口一定要设置X-Accel-Buffering: no响应头否则经过Nginx等反向代理时会被缓冲用户看到的还是一次性输出流式就白做了。这个坑我踩过排查了半天以为是代码问题结果是代理配置。4. 数据管道与上下文管理AI应用的隐形战场4.1 数据清洗比模型选择更影响效果很多人以为AI应用的效果主要取决于模型能力实际上在垂直场景里数据质量的影响往往更大。我做过一个客服问答的项目同样的模型只是把知识库里的文档做了清洗和结构化准确率从62%提到了81%。清洗包括去重、去噪、分段、加元数据这几步。分段策略尤其关键。按固定字数切分是最省事的但效果最差因为会把一个完整的语义单元切断。我推荐按语义边界切分比如按段落、按标题层级。如果文档结构不规整可以用一个轻量模型做语义分段成本不高但效果提升明显。每段建议控制在200到500字之间太短信息不完整太长检索精度下降。元数据是另一个容易被忽略的点。每段文本应该带上来源、时间、类别等元信息这样检索的时候可以做过滤。比如用户问“最新的退款政策”你可以先按时间过滤掉过期文档再在有效文档里做语义检索准确率会高很多。4.2 上下文窗口的预算管理上下文窗口是有限资源怎么分配是有讲究的。一个典型的对话请求上下文里可能包含系统提示词、历史对话、检索到的知识、用户当前问题。如果无脑全塞进去一是成本高二是模型注意力会被稀释效果反而下降。我的做法是给每一部分设预算。系统提示词控制在500 Token以内历史对话只保留最近N轮或者做摘要压缩检索知识控制在2000 Token以内剩下的留给用户问题和模型输出。这个预算不是拍脑袋定的而是根据实际评测调出来的。你可以做一个简单的实验固定其他变量只改检索知识的Token预算看准确率怎么变化找到性价比最高的点。历史对话的压缩是个实用技巧。当对话轮次多了以后不要简单截断而是用一个便宜的模型把前面的对话总结成一段话。这样既保留了关键信息又大幅节省了Token。我实测下来十轮对话压缩成200字左右的摘要信息保留率在90%以上Token消耗降低70%。4.3 检索增强的工程细节检索增强生成RAG现在几乎是AI应用的标配但做好不容易。向量检索只是其中一环完整的流程包括查询改写、多路召回、重排序、上下文组装。查询改写解决的是用户问题口语化、信息不全的问题。比如用户问“那个退款的事”直接拿去做向量检索效果很差改写成“退款政策 退款流程 退款条件”就好很多。改写可以用规则也可以用模型我建议先用规则覆盖高频场景再用模型兜底。多路召回是指同时走向量检索和关键词检索然后合并结果。向量检索擅长语义匹配关键词检索擅长精确匹配两者互补。合并的时候用RRFReciprocal Rank Fusion算法简单有效不需要调参。重排序是提升精度的关键一步。召回阶段追求高召回率可能返回20条结果然后用一个交叉编码器模型对这20条做精排选出最相关的5条。这一步会增加延迟但对准确率的提升很明显。如果延迟敏感可以用轻量级的重排序模型或者只对Top 10做重排。5. 服务化与并发处理让系统扛得住真实流量5.1 异步架构的必要性AI应用是典型的IO密集型服务大部分时间在等模型返回。如果用同步框架一个请求占一个线程并发量稍微上来线程池就满了。所以异步是必须的FastAPI httpx的异步组合是我目前最推荐的方案。但异步不是银弹用不好反而会出问题。最常见的错误是在异步函数里调用同步的阻塞代码比如用requests发请求、用time.sleep等待。这会导致整个事件循环被阻塞异步的优势荡然无存。我的建议是开启asyncio的调试模式它会警告你哪些地方阻塞了事件循环import asyncio asyncio.get_event_loop().set_debug(True)另一个坑是数据库操作。大部分Python数据库驱动是同步的在异步框架里要用异步驱动比如asyncpg配PostgreSQLmotor配MongoDB。如果实在没有异步驱动就用run_in_executor把同步操作丢到线程池里别直接调用。5.2 并发控制与限流模型服务通常有并发限制你不控制的话会被限流甚至封禁。我推荐在客户端做信号量控制import asyncio class RateLimitedClient: def __init__(self, client, max_concurrent10): self.client client self.semaphore asyncio.Semaphore(max_concurrent) async def generate(self, request): async with self.semaphore: return await self.client.generate(request)这个信号量的大小要根据模型服务的实际承载能力来定。可以先从小开始比如5然后压测逐步往上加观察错误率和延迟的变化找到拐点。不要一上来就设100那样只会触发限流。除了并发控制还要做请求队列。当并发满了以后新请求不应该直接拒绝而是排队等待。队列要有长度限制和超时否则会堆积大量请求把内存撑爆。我一般设队列长度为并发数的2到3倍排队超时10秒。5.3 缓存策略省钱又提速AI应用的缓存分好几层。最外层是精确匹配缓存用户问了完全相同的问题直接返回缓存结果。这一层用Redis做Key是问题的哈希Value是回答。命中率取决于场景客服类场景命中率能到30%以上。第二层是语义缓存用户问的问题意思相同但表述不同也能命中。这层用向量相似度做把历史问题向量化存起来新问题来了先做相似度检索超过阈值就返回缓存答案。阈值要调太高命中率低太低会返回不相关的答案。我一般从0.95开始试。第三层是Prompt缓存有些模型服务支持对相同的Prompt前缀做缓存能降低延迟和成本。这个需要你的Prompt结构设计得合理把固定部分放前面变化部分放后面。注意缓存一定要设过期时间尤其是知识类问答。政策、价格这类信息变化快缓存太久会返回过期答案。我一般设1到24小时根据内容更新频率调整。6. 可观测性与评测没有度量就没有优化6.1 日志、指标、追踪三件套AI服务的可观测性比传统服务更重要因为它的输出是不确定的出了问题很难复现。日志要记录完整的请求和响应包括Prompt、模型输出、Token数、延迟。但要注意脱敏用户隐私信息不能进日志。指标方面除了常规的QPS、延迟、错误率AI服务还要关注几个特有指标Token消耗速率、缓存命中率、检索召回率、模型降级次数。这些指标能帮你快速定位问题。比如延迟突然升高你看一下是模型调用变慢了还是检索变慢了就能缩小排查范围。追踪在AI应用里尤其有用因为一个请求可能经过查询改写、检索、重排序、模型生成多个环节。用OpenTelemetry做分布式追踪每个环节打一个Span出问题的时候一眼就能看出是哪个环节的锅。6.2 评测体系的搭建评测是AI工程里最容易被跳过但最重要的一环。没有评测你所有的优化都是盲目的。评测集要覆盖三类场景高频常见问题、边界情况、对抗性输入。每类至少准备50到100条少了没有统计意义。评测指标要分层。第一层是检索质量用召回率和精确率衡量。第二层是生成质量可以用人工评分也可以用模型评分。模型评分成本低但不够准我建议关键版本用人工日常迭代用模型评分做粗筛。评测要自动化每次代码变更都跑一遍。我一般用GitHub Actions或者类似的CI工具把评测脚本挂上去指标下降超过阈值就阻断合并。这样能防止优化一个场景的同时劣化另一个场景。6.3 线上问题的排查思路线上AI服务出问题排查顺序很重要。我的经验是先看是不是模型服务的问题再看是不是数据的问题最后才怀疑自己的代码。因为模型服务和数据的问题占大多数。模型服务问题的典型表现是延迟升高、错误率上升、返回内容异常。这时候先看模型服务商的状态页再看自己的调用指标。如果是限流就降并发如果是服务故障就切备用模型。数据问题的典型表现是回答不相关、答非所问。这时候先看检索结果如果检索出来的内容就不对那是检索的问题如果检索对了但生成不对那是Prompt或者模型的问题。检索问题一般是索引没更新、分段不合理、查询改写失效这几种。代码问题的典型表现是特定请求必现错误、并发高了才出错。前者一般是逻辑bug后者一般是资源竞争或者连接池配置问题。这类问题用追踪工具最容易定位。7. 成本控制与持续迭代让项目活得久7.1 Token成本的精细化管理Token成本是AI应用的主要运营成本不控制的话很容易失控。控制手段分几个层面。第一层是减少不必要的调用比如能走缓存的走缓存能用规则解决的不用模型。第二层是优化Prompt去掉冗余的示例和说明能省不少Token。第三层是模型分级简单任务用便宜的小模型复杂任务才用大模型。我做过一个统计一个中等复杂度的客服应用经过上述优化后Token成本能降低60%以上。具体做法是先做一轮缓存命中率30%再做Prompt精简平均Token数降低25%最后做模型分级70%的请求走小模型。三项叠加成本降幅很可观。模型分级需要一个路由逻辑。最简单的路由是按请求类型比如FAQ类走小模型投诉类走大模型。复杂一点的路由可以用一个分类模型先判断请求难度再决定用哪个模型。路由逻辑本身也有成本要权衡。7.2 迭代节奏与版本管理AI应用的迭代和传统软件不太一样因为模型和Prompt的变更会带来效果的非线性变化。我的做法是把Prompt、模型配置、检索参数都纳入版本管理每次变更都记录变更内容和评测结果。这样出问题的时候可以快速回滚也能积累经验知道什么改动有效。迭代节奏上我建议小步快跑。每次只改一个变量改完跑评测有效就保留无效就回滚。不要一次改好几个地方那样即使效果变好了你也不知道是哪个改动起的作用。这个原则听起来简单但执行起来需要纪律尤其是在赶进度的时候。版本管理用Git就够了但要注意把Prompt和配置从代码里抽出来单独放一个目录。这样非技术人员也能参与Prompt的调整不用改代码。我见过一些团队把Prompt写在Python字符串里改一个标点都要走代码评审效率很低。7.3 我踩过的几个典型坑第一个坑是评测集泄漏。有一次我们做检索优化效果提升很明显上线后却发现线上效果没变。排查后发现评测集里的问题和知识库里的文档有重叠模型在评测时其实是在“背答案”。后来我们重新构建了评测集确保问题和文档没有直接的字面重叠评测结果才可信。第二个坑是忽略长尾请求。我们的超时设的是30秒大部分请求几百毫秒就返回了但偶尔有请求会跑到20多秒。这些长尾请求占总量不到1%但拖慢了整体P99延迟。后来我们做了超时分级普通请求15秒长文本请求单独走一个通道P99延迟降了一半。第三个坑是缓存穿透。有一次缓存服务挂了所有请求直接打到模型服务瞬间触发限流整个服务不可用。后来我们加了本地缓存作为二级兜底即使Redis挂了也能扛一阵。这个教训是任何外部依赖都要考虑它挂掉的情况。8. 写在最后一些零散但有用的经验做AI工程这几年我最大的体会是模型能力在快速进步但工程能力是慢变量需要一点一点积累。你今天踩的坑明天换个模型可能还在。所以与其追新模型不如把工程基础打扎实。关于工具选型我的建议是不要过早引入重框架。LangChain这类框架在快速原型阶段很好用但到了需要精细控制的阶段它的抽象反而会成为负担。我现在的做法是核心链路自己写只在边缘环节用框架保持对关键路径的完全掌控。关于团队协作Prompt和评测集应该像代码一样管理有版本、有评审、有测试。我见过太多团队Prompt散落在各个人的笔记里换个人接手就抓瞎。把Prompt纳入工程体系是AI团队走向规范化的第一步。最后分享一个我常用的调试技巧当你不知道模型为什么输出某个结果时把完整的Prompt打印出来逐段删减看删到哪一段输出发生变化。这个方法很笨但非常有效能帮你快速定位是哪个部分在起作用。我用这个方法发现过好几次Prompt里的隐藏问题比如某个示例和当前问题冲突导致模型被误导。AI工程这个领域变化很快但底层的工程原则是稳定的抽象、隔离、可观测、可回滚。把这些原则落实到你的项目里不管上层技术怎么变你都能从容应对。
返回列表