第13课第二阶段工程化与部署学习目标掌握AI应用项目的标准目录结构学会使用FastAPI构建模型服务接口了解Docker容器化部署和CI/CD基本流程理解团队协作中的工程化规范1. 为什么需要工程化前几课我们写的代码都是单文件脚本。但真实的设备维修系统需要多人协作前后端、算法工程师并行开发持续迭代新增功能不影响已有功能可靠部署开发、测试、生产环境一致可观测性日志、监控、问题排查2. 项目标准结构maintenance-ai/ ├── src/ # 源代码 │ ├── __init__.py │ ├── main.py # FastAPI入口 │ ├── config.py # 配置管理 │ ├── agent/ # Agent模块 │ │ ├── __init__.py │ │ ├── tools.py # 工具定义 │ │ └── agent.py # Agent构建 │ ├── rag/ # RAG模块 │ │ ├── __init__.py │ │ ├── loader.py # 文档加载 │ │ └── retriever.py # 检索逻辑 │ └── models/ # 数据模型 │ └── schemas.py ├── tests/ # 测试 │ ├── test_agent.py │ └── test_rag.py ├── configs/ # 配置文件 │ └── settings.yaml ├── knowledge/ # 知识库原始文件 ├── Dockerfile ├── docker-compose.yml ├── requirements.txt ├── .env.example └── README.md关键原则src/按功能模块划分子目录不按文件类型tests/与src/平行测试文件命名test_*.py敏感配置放.env不提交到Gitknowledge/存放原始文档向量库自动生成3. 配置管理# src/config.pyfromdataclassesimportdataclass,fieldimportosdataclassclassSettings:# API配置 - 从环境变量读取openai_api_key:strfield(default_factorylambda:os.getenv(OPENAI_API_KEY,))openai_base_url:strfield(default_factorylambda:os.getenv(OPENAI_BASE_URL,https://api.deepseek.com))model_name:strdeepseek-chatembedding_model:strtext-embedding-3-small# 服务配置host:str0.0.0.0port:int8000# RAG配置chunk_size:int500chunk_overlap:int50top_k:int3settingsSettings()4. FastAPI服务接口# src/main.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelimportlogging logging.basicConfig(levellogging.INFO,format%(asctime)s [%(levelname)s] %(name)s: %(message)s)loggerlogging.getLogger(maintenance_ai)appFastAPI(title设备维修AI助手,version1.0.0)classQueryRequest(BaseModel):question:strsession_id:strdefaultclassDiagnoseRequest(BaseModel):device_id:strfault_description:strpriority:str普通app.post(/diagnose)asyncdefdiagnose(req:DiagnoseRequest):logger.info(f诊断请求:{req.device_id}-{req.fault_description})resultagent_executor.invoke({input:f诊断{req.device_id}{req.fault_description}})return{status:ok,diagnosis:result[output]}app.post(/query)asyncdefquery(req:QueryRequest):logger.info(f问答请求:{req.question[:50]}...)resultqa_chain({question:req.question})return{status:ok,answer:result[answer]}5. Docker容器化DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, src.main:app, --host, 0.0.0.0, --port, 8000]docker-compose.ymlversion:3.8services:app:build:.ports:-8000:8000env_file:.envvolumes:-./knowledge:/app/knowledge-./chroma_data:/app/chroma_datarestart:unless-stopped关键理解volumes挂载知识库目录容器重建不丢数据env_file从.env读取API Key等敏感配置restart: unless-stopped保证服务自动重启6. CI/CD基本流程代码提交 → 自动测试 → 代码审查 → 构建镜像 → 部署测试 → 验证 → 部署生产最小可用的GitHub Actions配置name:CIon:[push,pull_request]jobs:test:runs-on:ubuntu-lateststeps:-uses:actions/checkoutv4-uses:actions/setup-pythonv5with:{python-version:3.11}-run:pip install-r requirements.txt-run:pytest tests/7. 日志与监控importloggingimporttime logging.basicConfig(levellogging.INFO,format%(asctime)s [%(levelname)s] %(name)s: %(message)s)loggerlogging.getLogger(maintenance_ai)# 在关键操作前后记录日志logger.info(f收到诊断请求: device{req.device_id})starttime.time()resultagent_executor.invoke(...)elapsedtime.time()-start logger.info(f诊断完成, 耗时:{elapsed:.1f}s)日志级别使用规范DEBUG调试信息生产环境关闭INFO正常操作记录请求、完成、耗时WARNING异常但可继续重试、降级ERROR错误需要关注API失败、数据异常8. 团队协作规范Git分支策略main(生产) → develop(开发) → feature/*(功能)代码审查所有代码必须PR审核接口文档FastAPI自动生成Swagger文档/docs环境变量使用.env.example模板确保团队成员知道需要哪些配置9. 练习基础运行脚本生成项目脚手架浏览生成的目录结构理解每个文件的用途进阶修改生成的main.py添加一个/health健康检查接口和一个/stats统计接口挑战编写docker-compose.yml加入Redis作为对话缓存服务10. 验证清单能画出项目的标准目录结构并说明每个目录的作用理解FastAPI接口的设计思路和Pydantic模型的作用能解释Dockerfile中每条指令的作用知道CI/CD流程的各个环节及其目的能配置基本的日志记录code# -*- coding: utf-8 -*- 第13课项目脚手架生成器 - 设备维修AI系统 依赖: 无额外依赖纯Python标准库 运行后会生成完整的项目目录结构 importos PROJECT_NAMEmaintenance-ai# 文件模板定义 MAIN_PY\ 设备维修AI助手 - FastAPI服务 import logging from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from pydantic import BaseModel from src.config import settings logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s) logger logging.getLogger(maintenance_ai) # --- 请求/响应模型 --- class QueryRequest(BaseModel): question: str session_id: str default class DiagnoseRequest(BaseModel): device_id: str fault_description: str priority: str 普通 class ReportRequest(BaseModel): device_id: str report_type: str 状态报告 class AIResponse(BaseModel): status: str data: str session_id: str # --- Agent初始化占位 --- agent_executor None # 实际项目中在此初始化Agent asynccontextmanager async def lifespan(app: FastAPI): 应用生命周期管理 logger.info(设备维修AI系统启动...) # global agent_executor # agent_executor build_agent() yield logger.info(设备维修AI系统关闭) app FastAPI(title设备维修AI助手, version1.0.0, lifespanlifespan) app.get(/health) async def health(): return {status: ok, version: 1.0.0} app.post(/diagnose, response_modelAIResponse) async def diagnose(req: DiagnoseRequest): 设备故障诊断接口 logger.info(f诊断请求: {req.device_id} - {req.fault_description}) if not agent_executor: raise HTTPException(503, Agent未初始化) result agent_executor.invoke( {input: f诊断设备{req.device_id}{req.fault_description}} ) return AIResponse(statusok, dataresult[output]) app.post(/query, response_modelAIResponse) async def query(req: QueryRequest): 知识库问答接口 logger.info(f问答请求: {req.question[:50]}...) return AIResponse(statusok, data接口已就绪请接入RAG链, session_idreq.session_id) app.post(/report, response_modelAIResponse) async def report(req: ReportRequest): 生成设备报告接口 logger.info(f报告请求: {req.device_id} - {req.report_type}) return AIResponse(statusok, dataf设备{req.device_id}的{req.report_type}生成中...) if __name__ __main__: import uvicorn uvicorn.run(app, hostsettings.host, portsettings.port) CONFIG_PY\ 配置管理模块 - 从环境变量读取配置 import os from dataclasses import dataclass, field dataclass class Settings: 应用配置 # API配置 openai_api_key: str field( default_factorylambda: os.getenv(OPENAI_API_KEY, )) openai_base_url: str field( default_factorylambda: os.getenv(OPENAI_BASE_URL, https://api.deepseek.com)) model_name: str deepseek-chat embedding_model: str text-embedding-3-small # 服务配置 host: str 0.0.0.0 port: int 8000 # RAG配置 chunk_size: int 500 chunk_overlap: int 50 top_k: int 3 chroma_dir: str ./chroma_data settings Settings() SCHEMAS_PY\ 数据模型定义 from pydantic import BaseModel class Device(BaseModel): device_id: str name: str model: str status: str run_hours: int location: str class WorkOrder(BaseModel): order_id: str device_id: str fault_description: str priority: str status: str created_at: str DOCKERFILE\ FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple COPY . . EXPOSE 8000 CMD [uvicorn, src.main:app, --host, 0.0.0.0, --port, 8000] DOCKER_COMPOSE\ version: 3.8 services: maintenance-ai: build: . container_name: maintenance-ai ports: - 8000:8000 env_file: - .env volumes: - ./knowledge:/app/knowledge - ./chroma_data:/app/chroma_data restart: unless-stopped healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 10s retries: 3 REQUIREMENTS\ fastapi0.109.0 uvicorn0.27.0 pydantic2.5.3 langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.13 chromadb0.4.22 sentence-transformers2.3.1 python-dotenv1.0.0 ENV_EXAMPLE\ # OpenAI/DeepSeek API配置 OPENAI_API_KEYsk-your-api-key-here OPENAI_BASE_URLhttps://api.deepseek.com # 服务配置 HOST0.0.0.0 PORT8000 # 知识库配置 CHUNK_SIZE500 CHUNK_OVERLAP50 TOP_K3 SETTINGS_YAML\ # 设备维修AI系统配置 app: name: maintenance-ai version: 1.0.0 llm: model: deepseek-chat temperature: 0 base_url: https://api.deepseek.com rag: chunk_size: 500 chunk_overlap: 50 top_k: 3 embedding_model: text-embedding-3-small server: host: 0.0.0.0 port: 8000 README\ # 设备维修AI助手 ## 快速开始 1. 复制配置文件并填入API Key cp .env.example .env 2. 安装依赖 pip install -r requirements.txt 3. 启动服务 python -m src.main 4. Docker部署 docker-compose up -d ## API文档 启动后访问 http://localhost:8000/docs 查看Swagger文档 # 文件清单 FILES{src/__init__.py:# 设备维修AI系统\n__version__ 1.0.0\n,src/main.py:MAIN_PY,src/config.py:CONFIG_PY,src/agent/__init__.py:,src/agent/tools.py:# 参考第12课的5个工具定义\n# from langchain_classic.tools import tool\n,src/agent/agent.py:# 参考第12课的build_agent函数\n,src/rag/__init__.py:,src/rag/loader.py:# 参考第11课的文档加载逻辑\n,src/rag/retriever.py:# 参考第11课的向量检索逻辑\n,src/models/__init__.py:,src/models/schemas.py:SCHEMAS_PY,tests/test_agent.py:# Agent模块测试\ndef test_placeholder():\n assert True\n,tests/test_rag.py:# RAG模块测试\ndef test_placeholder():\n assert True\n,configs/settings.yaml:SETTINGS_YAML,knowledge/README.md:# 知识库文件\n将设备维修手册、故障记录等文件放在此目录\n,requirements.txt:REQUIREMENTS,Dockerfile:DOCKERFILE,docker-compose.yml:DOCKER_COMPOSE,.env.example:ENV_EXAMPLE,README.md:README,}defgenerate_project(base_dir):生成项目脚手架project_diros.path.join(base_dir,PROJECT_NAME)print(f️ 生成项目脚手架{project_dir}\n)created[]forrel_path,contentinFILES.items():full_pathos.path.join(project_dir,rel_path)os.makedirs(os.path.dirname(full_path),exist_okTrue)withopen(full_path,w,encodingutf-8)asf:f.write(content)created.append(rel_path)returnproject_dir,createddefprint_tree(path,prefix):打印目录树ifos.path.isfile(path):returnentriessorted(os.listdir(path))dirs[eforeinentriesifos.path.isdir(os.path.join(path,e))]files[eforeinentriesifos.path.isfile(os.path.join(path,e))]items[(d,True)fordindirs][(f,False)forfinfiles]fori,(name,is_dir)inenumerate(items):is_last(ilen(items)-1)connector└── ifis_lastelse├── displayf{name}/ifis_direlsenameprint(f{prefix}{connector}{display})ifis_dir:ext ifis_lastelse│ print_tree(os.path.join(path,name),prefixext)defmain():print( 第13课项目工程化 - 脚手架生成器\n)baseos.path.dirname(os.path.dirname(os.path.abspath(__file__)))project_dir,createdgenerate_project(base)print(f✅ 共生成{len(created)}个文件\n)print( 项目结构)print(f{─*40})print(f{PROJECT_NAME}/)print_tree(project_dir)print(f{─*40})print(\n 生成文件清单)forfincreated:print(f ✓{f})print(\n 下一步操作)print( 1. cd maintenance-ai)print( 2. cp .env.example .env (填入API Key))print( 3. pip install -r requirements.txt)print( 4. python -m src.main)print( 5. 打开 http://localhost:8000/docs 查看API文档)if__name____main__:main()