企业官网建设流程全解析

第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，不提交到Git
• knowledge/存放原始文档，向量库自动生成

3. 配置管理

# src/config.pyfromdataclassesimportdataclass,fieldimportos@dataclassclassSettings:# API配置 - 从环境变量读取openai_api_key:str=field(default_factory=lambda:os.getenv("OPENAI_API_KEY",""))openai_base_url:str=field(default_factory=lambda: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=500chunk_overlap:int=50top_k:int=3settings=Settings()

4. FastAPI服务接口

# src/main.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelimportlogging logging.basicConfig(level=logging.INFO,format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")logger=logging.getLogger("maintenance_ai")app=FastAPI(title="设备维修AI助手",version="1.0.0")classQueryRequest(BaseModel):question:strsession_id:str="default"classDiagnoseRequest(BaseModel):device_id:strfault_description:strpriority:str="普通"@app.post("/diagnose")asyncdefdiagnose(req:DiagnoseRequest):logger.info(f"诊断请求:{req.device_id}-{req.fault_description}")result=agent_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]}...")result=qa_chain({"question":req.question})return{"status":"ok","answer":result["answer"]}

5. Docker容器化

Dockerfile

FROM 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.yml

version:"3.8"services:app:build:.ports:-"8000:8000"env_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/checkout@v4-uses:actions/setup-python@v5with:{python-version:"3.11"}-run:pip install-r requirements.txt-run:pytest tests/

7. 日志与监控

importloggingimporttime logging.basicConfig(level=logging.INFO,format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")logger=logging.getLogger("maintenance_ai")# 在关键操作前后记录日志logger.info(f"收到诊断请求: device={req.device_id}")start=time.time()result=agent_executor.invoke(...)elapsed=time.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. 练习

1. 基础：运行脚本生成项目脚手架，浏览生成的目录结构，理解每个文件的用途
2. 进阶：修改生成的main.py，添加一个/health健康检查接口和一个/stats统计接口
3. 挑战：编写docker-compose.yml加入Redis作为对话缓存服务

10. 验证清单

• 能画出项目的标准目录结构并说明每个目录的作用
• 理解FastAPI接口的设计思路和Pydantic模型的作用
• 能解释Dockerfile中每条指令的作用
• 知道CI/CD流程的各个环节及其目的
• 能配置基本的日志记录

code

# -*- coding: utf-8 -*-""" 第13课：项目脚手架生成器 - 设备维修AI系统 依赖: 无额外依赖（纯Python标准库） 运行后会生成完整的项目目录结构 """importos PROJECT_NAME="maintenance-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(level=logging.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助手", version="1.0.0", lifespan=lifespan) @app.get("/health") async def health(): return {"status": "ok", "version": "1.0.0"} @app.post("/diagnose", response_model=AIResponse) 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(status="ok", data=result["output"]) @app.post("/query", response_model=AIResponse) async def query(req: QueryRequest): """知识库问答接口""" logger.info(f"问答请求: {req.question[:50]}...") return AIResponse(status="ok", data="接口已就绪，请接入RAG链", session_id=req.session_id) @app.post("/report", response_model=AIResponse) async def report(req: ReportRequest): """生成设备报告接口""" logger.info(f"报告请求: {req.device_id} - {req.report_type}") return AIResponse(status="ok", data=f"设备{req.device_id}的{req.report_type}生成中...") if __name__ == "__main__": import uvicorn uvicorn.run(app, host=settings.host, port=settings.port) '''CONFIG_PY='''\ """配置管理模块 - 从环境变量读取配置""" import os from dataclasses import dataclass, field @dataclass class Settings: """应用配置""" # API配置 openai_api_key: str = field( default_factory=lambda: os.getenv("OPENAI_API_KEY", "")) openai_base_url: str = field( default_factory=lambda: 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="""\ fastapi==0.109.0 uvicorn==0.27.0 pydantic==2.5.3 langchain==0.1.0 langchain-openai==0.0.5 langchain-community==0.0.13 chromadb==0.4.22 sentence-transformers==2.3.1 python-dotenv==1.0.0 """ENV_EXAMPLE="""\ # OpenAI/DeepSeek API配置 OPENAI_API_KEY=sk-your-api-key-here OPENAI_BASE_URL=https://api.deepseek.com # 服务配置 HOST=0.0.0.0 PORT=8000 # 知识库配置 CHUNK_SIZE=500 CHUNK_OVERLAP=50 TOP_K=3 """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_dir=os.path.join(base_dir,PROJECT_NAME)print(f"🏗️ 生成项目脚手架：{project_dir}\n")created=[]forrel_path,contentinFILES.items():full_path=os.path.join(project_dir,rel_path)os.makedirs(os.path.dirname(full_path),exist_ok=True)withopen(full_path,"w",encoding="utf-8")asf:f.write(content)created.append(rel_path)returnproject_dir,createddefprint_tree(path,prefix=""):"""打印目录树"""ifos.path.isfile(path):returnentries=sorted(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=(i==len(items)-1)connector="└── "ifis_lastelse"├── "display=f"{name}/"ifis_direlsenameprint(f"{prefix}{connector}{display}")ifis_dir:ext=" "ifis_lastelse"│ "print_tree(os.path.join(path,name),prefix+ext)defmain():print("🏭 第13课：项目工程化 - 脚手架生成器\n")base=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))project_dir,created=generate_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()