☰
AI项目工程化实战:从脚本到可维护系统的目录结构、配置分离与部署
2026/9/26 19:09:25 网站建设 项目流程

1. 从脚本到系统:AI 项目工程化到底在解决什么问题

写了四五十课的 Python,语法、爬虫、数据分析、可视化基本都摸过一遍了,这时候很多人会卡在同一个坎上:单个.py文件跑得挺欢,一旦要把一个 AI 能力做成能给别人用、能长期维护、能反复迭代的东西,就立刻乱成一锅粥。模型权重放哪、提示词写死在代码里、依赖版本对不上、换台机器就跑不起来、接口一改下游全崩——这些都不是算法问题,而是工程问题。

AI 项目工程化,说白了就是把“我本地能跑”变成“谁都能跑、天天都能跑、改了还能跑”。它跟传统软件工程有重叠,但多了几块特有的麻烦:模型和数据是有体积的、提示词和参数是需要版本管理的、推理结果是带随机性的、成本是按 token 或算力计费的。这一课我打算把这几块拆开讲透,从目录结构、依赖管理、配置分离,到模型封装、服务化、测试和部署,给一套可以直接抄的骨架。适合已经会写 Python、但项目一超过三个文件就头疼的人,也适合做 AI Agent、大模型应用、量化策略这类需要长期维护的开发者。

我踩过的最大一个坑,就是早期把所有逻辑塞进一个main.py,提示词直接写成字符串常量,模型路径写死成绝对路径。结果换台机器,路径全废;想调提示词,得翻几百行代码;想对比两个版本效果,只能靠手动改完再跑一遍,改回去还容易漏。工程化要解决的,就是这种“改一处、崩一片”的连锁反应。

2. 项目骨架设计:目录结构决定你能走多远

2.1 为什么目录结构是工程化的第一道门槛

很多人觉得目录结构是形式主义,能跑就行。但目录结构本质上是职责边界的物理体现。你把配置、数据、模型、业务逻辑、测试混在一起,代码耦合就是必然的,因为物理上它们都挨着,随手 import 一下就跨层调用了。等到项目膨胀到几千行,你会发现改一个提示词模板,居然要动到数据加载模块,这就是边界没划清的代价。

一个能扛住长期迭代的 AI 项目,我一般会按下面这个骨架来搭。它不是唯一解,但每一层都有明确理由:

ai-project/ ├── configs/ # 配置层:环境、模型、提示词 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── src/ │ ├── __init__.py │ ├── data/ # 数据加载与预处理 │ ├── models/ # 模型封装与推理 │ ├── prompts/ # 提示词模板集中管理 │ ├── services/ # 业务编排层 │ └── utils/ # 通用工具 ├── tests/ # 测试 ├── scripts/ # 一次性脚本、数据迁移 ├── notebooks/ # 探索性分析,不进生产 ├── requirements.txt ├── pyproject.toml └── README.md

关键点在于configs和src的分离。配置是“会变的东西”,代码是“相对稳定的东西”,把易变的抽出去,代码就不用频繁改。notebooks单独放,是因为探索性代码天生脏乱,绝不能让它污染生产路径——我见过太多项目,最后生产代码里 import 了一个 notebook 里的函数,那个 notebook 还带着一堆调试输出。

2.2 配置分离:把提示词和参数从代码里赶出去

AI 项目跟普通项目最大的区别,就是提示词也是代码资产。它需要版本管理、需要 A/B 对比、需要按环境切换。把提示词写死在 Python 字符串里,等于放弃了这一切。我的做法是统一放prompts/目录,用模板文件管理:

prompts/ ├── summarize_v1.txt ├── summarize_v2.txt └── classify.txt

然后在配置里指定用哪个版本。这样调提示词就是改文件,不用碰代码,还能用 git 直接 diff 两个版本的差异。参数同理,温度、最大长度、重试次数这些,全部进 YAML:

# configs/base.yaml model: name: "your-model-name" temperature: 0.7 max_tokens: 1024 timeout: 30 retry: 3 prompt: summarize: "summarize_v2" classify: "classify" paths: data_dir: "./data" output_dir: "./outputs"

加载配置我推荐用pydantic配合pyyaml,因为 pydantic 能在加载时做类型校验,配置写错了立刻报错,而不是跑到一半才崩。这一点在 AI 项目里特别重要,因为很多错误(比如温度写成字符串)在运行时才暴露,排查成本极高。

注意:配置文件里绝对不要放密钥。密钥走环境变量,配置里只放“从哪个环境变量读”的引用。这是安全底线,也是团队协作的基本规矩。

2.3 依赖管理:为什么 requirements.txt 不够用

pip install -r requirements.txt是入门做法,但它有个致命问题:不锁定传递依赖。你写torch>=2.0,今天装的是 2.0.1,明天可能是 2.1.0,行为可能就变了。AI 项目对版本极其敏感,一个 CUDA 版本对不上,整个推理就废了。

我的建议是分两层:requirements.in写你直接依赖的顶层包(宽松版本),用pip-tools编译出锁定的requirements.txt(精确版本)。这样既方便升级,又能保证复现:

pip install pip-tools pip-compile requirements.in -o requirements.txt pip-sync requirements.txt

pip-sync比pip install更狠,它会把你环境里多余的包删掉,保证环境和锁定文件完全一致。这在 CI 和部署时特别有用,能避免“本地能跑、服务器不行”的经典问题。如果项目更复杂,直接上poetry或uv,它们把依赖、虚拟环境、打包一体化管理,省心不少。

3. 模型与推理封装:让 AI 能力变成可调用的积木

3.1 封装的意义:隔离变化,统一接口

AI 项目里,模型是最容易变的部分。今天用这个 API,明天换本地部署,后天加个缓存层。如果业务代码直接调用模型 SDK,那每次换模型都要改遍全项目。正确做法是定义一个抽象接口,把模型调用包在models/层里:

# src/models/base.py from abc import ABC, abstractmethod class BaseLLM(ABC): @abstractmethod def generate(self, prompt: str, **kwargs) -> str: ... @abstractmethod def batch_generate(self, prompts: list[str], **kwargs) -> list[str]: ...

然后针对不同后端实现这个接口。业务层只依赖BaseLLM,不关心底层是哪个模型。这样换模型就是加一个实现类,改一行配置,业务代码纹丝不动。这就是依赖倒置在 AI 项目里的实际价值。

3.2 重试、超时与降级:AI 调用必须假设它会失败

AI 调用跟普通函数调用最大的不同,是它天然不稳定。网络会抖、服务会限流、模型会返回空、会超时。如果你不处理这些,线上就是随机崩。我的封装里必带三件套:超时、重试、降级。

import time from tenacity import retry, stop_after_attempt, wait_exponential class RemoteLLM(BaseLLM): def __init__(self, config): self.config = config @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), ) def generate(self, prompt: str, **kwargs) -> str: try: resp = self._call_api(prompt, timeout=self.config.timeout, **kwargs) if not resp or not resp.strip(): raise ValueError("empty response") return resp except Exception as e: # 记录日志,便于排查 logger.warning(f"generate failed: {e}") raise

tenacity的指数退避很关键,第一次等 1 秒,第二次 2 秒,第三次 4 秒,避免瞬间打爆下游。降级策略则看业务:可以返回缓存结果、返回兜底文案、或者抛给上层决定。关键是不能让一次失败直接冒泡到用户。

实操心得:重试次数不是越多越好。我一般设 3 次,因为超过 3 次还失败,大概率是服务真挂了,再重试只是浪费时间和配额。同时一定要给重试加日志,否则线上出问题你根本不知道重试了多少次。

3.3 缓存:省钱又提速的隐形功臣

AI 调用又慢又贵,缓存是性价比最高的优化。最简单的做法是用functools.lru_cache,但它只对纯函数有效,且进程重启就没了。生产环境我推荐用文件缓存或 Redis,key 用“模型名 + 提示词哈希 + 参数哈希”:

import hashlib, json, os def cache_key(model: str, prompt: str, params: dict) -> str: raw = json.dumps({"m": model, "p": prompt, "k": params}, sort_keys=True) return hashlib.sha256(raw.encode()).hexdigest()

这里有个坑:参数里如果有随机种子或时间戳,缓存永远命中不了。所以缓存 key 要只包含影响输出的确定性参数。另外,缓存要设过期时间,因为模型可能更新,旧结果未必还适用。我一般设 7 天,兼顾命中率和新鲜度。

4. 服务化与接口设计:把能力交出去

4.1 从函数到服务:什么时候该上 Web 框架

当你的 AI 能力需要被别的系统调用、被前端调用、或者被多个用户同时用时,就该服务化了。Python 里最轻的选择是 FastAPI,它自带类型校验和自动文档,对 AI 项目特别友好,因为 AI 接口的参数往往很多,类型校验能挡掉大量脏输入。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() class SummarizeRequest(BaseModel): text: str max_length: int = 200 class SummarizeResponse(BaseModel): summary: str model: str @app.post("/summarize", response_model=SummarizeResponse) def summarize(req: SummarizeRequest): if len(req.text) > 10000: raise HTTPException(status_code=400, detail="text too long") result = service.summarize(req.text, max_length=req.max_length) return SummarizeResponse(summary=result, model=config.model.name)

用 pydantic 定义请求和响应模型,好处是接口契约显式化。前端一看文档就知道要传什么、会收到什么,减少沟通成本。而且 FastAPI 会自动生成/docs页面,联调时特别省事。

4.2 同步还是异步:AI 接口的性能关键

AI 调用是 IO 密集型(等网络、等模型),如果用同步接口,一个请求卡住,整个线程就堵着。FastAPI 支持async def,配合异步 HTTP 客户端,能大幅提升并发。但要注意:如果你的模型调用是同步阻塞的(比如本地推理),放进 async 函数里反而会阻塞事件循环。这时候要么用线程池,要么老老实实写同步接口让框架自己调度。

import asyncio from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(max_workers=4) @app.post("/summarize") async def summarize(req: SummarizeRequest): loop = asyncio.get_event_loop() result = await loop.run_in_executor( executor, service.summarize, req.text ) return {"summary": result}

这个模式我用了很多次,把阻塞的推理丢进线程池,事件循环继续处理其他请求,并发能力立刻上一个台阶。max_workers要根据模型的实际并发能力设,设太大反而会拖垮模型服务。

4.3 限流与鉴权:别让服务被薅秃

服务一旦暴露,就会有人(或脚本)疯狂调用。限流是必须的,最简单的做法是按 IP 或 API Key 计数。FastAPI 可以用中间件实现:

from collections import defaultdict import time rate_limit = defaultdict(list) @app.middleware("http") async def limit_middleware(request, call_next): ip = request.client.host now = time.time() rate_limit[ip] = [t for t in rate_limit[ip] if now - t < 60] if len(rate_limit[ip]) >= 60: return JSONResponse(status_code=429, content={"detail": "too many requests"}) rate_limit[ip].append(now) return await call_next(request)

这段代码实现的是“每分钟 60 次”的滑动窗口限流。生产环境建议用 Redis 存计数,因为多进程部署时内存计数不共享。鉴权则用 API Key 或 Token,放在请求头里校验,别放 URL 参数里,否则会进日志泄露。

5. 测试与可观测性:让问题在爆发前被发现

5.1 AI 项目怎么测:确定性测试与不确定性测试分开

AI 项目的测试比普通项目难,因为输出不确定。我的策略是分层测试:确定性逻辑(数据清洗、参数校验、缓存 key 生成)用传统单元测试,断言精确值;不确定性逻辑(模型输出)用“属性测试”,只断言输出满足某些性质,比如非空、长度在范围内、包含关键字段。

def test_cache_key_stable(): k1 = cache_key("m", "hello", {"t": 0.7}) k2 = cache_key("m", "hello", {"t": 0.7}) assert k1 == k2 # 确定性 def test_summarize_not_empty(): result = service.summarize("一段测试文本") assert isinstance(result, str) assert len(result) > 0 # 属性断言,不比对具体内容

对于提示词效果,硬断言没用,得靠评测集。我会维护一个小规模的人工标注集,每次改提示词就跑一遍,看准确率、召回率有没有退化。这套评测脚本放scripts/里,不进生产,但每次发版前必跑。

5.2 日志与追踪:线上出问题靠什么定位

AI 项目的日志要记三样东西:输入、输出、耗时。输入用于复现,输出用于分析,耗时用于性能优化。但要注意脱敏,用户输入里可能有敏感信息,日志里要过滤。我一般用结构化日志:

import logging, json logger = logging.getLogger("ai") def log_call(prompt, output, elapsed, model): logger.info(json.dumps({ "event": "llm_call", "model": model, "prompt_len": len(prompt), "output_len": len(output), "elapsed": round(elapsed, 3), }, ensure_ascii=False))

用 JSON 格式是为了方便后续接入日志系统做检索和统计。prompt_len而不是完整 prompt,是出于隐私和体积考虑。如果确实需要完整记录,单独存到受控的存储里,并设访问权限。

常见问题:很多人日志只记“调用成功”,不记失败原因。结果线上报错率上升,却不知道是超时、限流还是模型返回异常。我的做法是失败日志里带上异常类型和重试次数,这样一眼就能看出问题分布。

5.3 成本监控:AI 项目独有的账本

AI 调用是要花钱的,token 用量、GPU 时长都是成本。工程化项目必须能回答“这个月花了多少、哪个功能最费钱”。做法是在封装层记录每次调用的 token 数(或估算值),汇总到监控系统。我一般按“功能模块 + 模型”两个维度统计,这样能快速定位成本大头。如果某个功能成本异常高,要么优化提示词,要么换更便宜的模型,要么加缓存。

6. 部署与持续迭代:让项目活得久

6.1 容器化:一次构建,到处运行

AI 项目部署最大的痛点是环境。CUDA 版本、Python 版本、系统库,任何一处不一致都可能跑不起来。Docker 是标准解法。一个精简的 Dockerfile:

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY src/ ./src/ COPY configs/ ./configs/ COPY prompts/ ./prompts/ ENV PYTHONPATH=/app EXPOSE 8000 CMD ["uvicorn", "src.services.api:app", "--host", "0.0.0.0", "--port", "8000"]

关键点是先复制依赖文件再复制代码,这样改代码不会触发依赖重装,构建快很多。--no-cache-dir减小镜像体积。如果模型文件很大,不要打进镜像,用挂载卷或对象存储,镜像只放代码。

6.2 配置注入:同一份镜像跑多环境

镜像应该只有一份,环境差异靠配置注入。用环境变量指定加载哪个配置文件:

import os, yaml def load_config(): env = os.getenv("APP_ENV", "dev") with open(f"configs/{env}.yaml") as f: return yaml.safe_load(f)

这样 dev、prod 用同一个镜像,只是APP_ENV不同。密钥也走环境变量,绝不进镜像。这是十二要素应用的基本原则,AI 项目同样适用。

6.3 灰度与回滚:改提示词也要能撤回

AI 项目迭代频繁,提示词、模型、参数都可能变。每次变更都应该能灰度、能回滚。最简单的做法是配置里加一个“流量比例”,新版本先接 10% 流量,观察指标没问题再全量。回滚就是把配置改回去,重启服务。因为提示词和参数都在配置里,回滚不需要重新构建镜像,几秒钟就能完成。

我个人的经验是,任何影响输出的变更都要有回滚预案。有一次我改了个提示词,本地测着挺好,上线后某个边界场景输出全乱了,幸好配置能秒回滚,没造成大影响。从那以后,我坚持所有提示词变更都走配置,绝不硬编码。

7. 常见问题与排查速查

实际做工程化的过程中,问题往往集中在几个地方。我把踩过的坑整理成表,方便对照排查:

现象可能原因排查方向
本地能跑,服务器报错依赖版本不一致用锁定文件,检查 Python 版本
模型输出为空提示词问题或超时看日志里的原始响应和耗时
并发上不去同步阻塞事件循环检查是否用了线程池
缓存命中率低key 含随机参数检查 key 生成逻辑
成本突然飙升某功能调用量激增按模块统计 token 用量
配置改了不生效缓存了配置对象检查配置加载时机

排查的核心思路是先定位层次:是配置问题、依赖问题、还是代码逻辑问题。我一般从日志入手,看最后一次成功和第一次失败的差异,往往能快速缩小范围。另外,本地复现是王道,如果本地复现不了,说明环境差异是主因,优先查依赖和配置。

避坑技巧:给项目加一个scripts/healthcheck.py,启动时自检配置、依赖、模型连通性。部署后先跑一遍,能挡掉大部分低级问题。

8. 我在这条路上的一些真实体会

工程化这东西,刚开始做会觉得繁琐,觉得不如直接写脚本痛快。但项目一旦要活过三个月、要交给别人维护、要天天稳定跑,这些“繁琐”就是救命稻草。我最大的转变,是从“能跑就行”变成“改了还能跑”。这个转变的代价是前期多花时间搭骨架,收益是后期改需求时不用推倒重来。

还有一点,工程化不是一次性任务,是持续习惯。每次加新功能,都问自己:配置抽出去了吗?有测试吗?失败会怎样?日志够定位吗?养成这个习惯后,项目自然就稳了。至于工具选型,别追新,选团队熟悉的、社区活跃的,能省很多事。FastAPI、pydantic、tenacity、Docker 这套组合,我用了几年,没出过大问题,推荐给刚起步的人。

最后分享一个小技巧:把项目的“运行手册”写进 README,包括怎么装、怎么跑、怎么测、怎么部署、出问题找谁。这份文档的价值,在你休假时别人要接手的那一刻,会体现得淋漓尽致。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询