AI 技术正在从实验室走向工程化从算法研究变成可复用的开发组件。但很多团队在引入 AI 能力时往往面临一个现实问题如何把 AI 模型、提示词工程、业务逻辑和传统软件工程方法结合起来形成一套可持续迭代的 AI 工程实践体系。本文将以一个实际项目为例展示如何从零搭建一个具备完整 AI 能力的应用涵盖环境准备、模型集成、提示词设计、异常处理和部署上线全流程。1. 理解 AI 工程化的核心挑战AI 工程化不是简单调用 API而是要把非确定性的 AI 输出整合到确定性的软件系统中。这需要解决几个关键问题1.1 模型输出的不确定性处理传统软件开发中相同的输入必然产生相同的输出。但 AI 模型特别是大语言模型的响应存在随机性即使使用相同的提示词也可能得到不同结构和内容的回答。工程上需要设计校验机制、重试策略和降级方案。1.2 提示词版本化管理提示词本质上是另一种形式的代码需要像管理源代码一样进行版本控制、测试和迭代优化。但提示词又不同于传统代码它的效果评估更主观需要建立评估标准和 A/B 测试机制。1.3 成本与延迟的平衡直接调用云端大模型 API 虽然方便但存在成本高、网络延迟、数据隐私等问题。本地部署的小模型虽然可控性强但能力有限。工程上需要根据场景选择合适的模型策略甚至设计分层调用方案。1.4 与传统系统的集成AI 功能很少独立存在通常需要与现有的用户系统、数据库、业务流程集成。这涉及到数据格式转换、异步处理、状态管理等复杂问题。2. 项目环境与工具链准备2.1 基础开发环境建议使用 Python 3.9 作为主要开发语言这是目前 AI 生态最完善的环境。同时需要准备以下核心工具# 创建虚拟环境 python -m venv ai_project source ai_project/bin/activate # Linux/Mac # ai_project\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain fastapi uvicorn pydantic2.2 模型访问配置根据项目需求选择模型提供商。如果使用 OpenAI 系列模型需要配置 API Key# config.py import os from typing import Optional class AIConfig: OPENAI_API_KEY: str os.getenv(OPENAI_API_KEY, ) MODEL_NAME: str gpt-3.5-turbo # 默认模型 MAX_RETRIES: int 3 # API 调用重试次数 TIMEOUT: int 30 # 请求超时时间 classmethod def validate_config(cls): if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY 环境变量未设置)2.3 项目结构设计清晰的目录结构有助于维护 AI 项目的复杂性ai_project/ ├── src/ │ ├── core/ # 核心AI功能 │ │ ├── __init__.py │ │ ├── models.py # 数据模型 │ │ ├── prompts/ # 提示词模板 │ │ └── clients.py # AI客户端封装 │ ├── api/ # Web接口层 │ ├── utils/ # 工具函数 │ └── config.py # 配置文件 ├── tests/ # 测试用例 ├── requirements.txt # 依赖列表 └── main.py # 应用入口3. 构建可维护的 AI 客户端直接裸调用 AI API 会导致代码分散、难以维护。应该封装统一的客户端来处理重试、异常、日志等通用逻辑。3.1 基础客户端实现# src/core/clients.py import logging import time from typing import Dict, Any, Optional from openai import OpenAI, APIError, RateLimitError logger logging.getLogger(__name__) class AIClient: def __init__(self, config): self.client OpenAI(api_keyconfig.OPENAI_API_KEY) self.config config self.model config.MODEL_NAME def chat_completion(self, messages: list, temperature: float 0.7) - Optional[str]: 带重试机制的聊天补全调用 for attempt in range(self.config.MAX_RETRIES): try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, timeoutself.config.TIMEOUT ) return response.choices[0].message.content except RateLimitError as e: wait_time 2 ** attempt # 指数退避 logger.warning(f速率限制{wait_time}秒后重试: {e}) time.sleep(wait_time) except APIError as e: logger.error(fAPI错误: {e}) if attempt self.config.MAX_RETRIES - 1: raise time.sleep(1) except Exception as e: logger.error(f未知错误: {e}) break return None3.2 提示词模板管理将提示词从代码中分离出来便于管理和迭代# src/core/prompts/__init__.py from string import Template from typing import Dict, Any class PromptTemplate: def __init__(self, template: str, required_vars: list): self.template Template(template) self.required_vars required_vars def format(self, **kwargs) - str: # 检查必需变量 missing_vars [var for var in self.required_vars if var not in kwargs] if missing_vars: raise ValueError(f缺少必需变量: {missing_vars}) return self.template.substitute(**kwargs) # 具体提示词定义 SUMMARY_PROMPT PromptTemplate( template请为以下文本生成一个简洁的摘要要求 1. 保留核心信息 2. 长度控制在100字以内 3. 语言流畅自然 文本内容 $content 摘要, required_vars[content] ) CLASSIFICATION_PROMPT PromptTemplate( template请对以下文本进行分类从预定义类别中选择最合适的一项。 预定义类别$categories 文本内容 $content 请只返回类别名称不要额外解释。, required_vars[content, categories] )4. 实现业务逻辑层AI 能力需要封装成具体的业务功能而不是直接暴露给上层调用。4.1 文本处理服务示例# src/core/services.py from .clients import AIClient from .prompts import SUMMARY_PROMPT, CLASSIFICATION_PROMPT import re class TextProcessingService: def __init__(self, ai_client: AIClient): self.client ai_client def generate_summary(self, content: str, max_length: int 100) - dict: 生成文本摘要 if len(content.strip()) 10: return {success: False, error: 文本内容过短} try: prompt SUMMARY_PROMPT.format(contentcontent) messages [{role: user, content: prompt}] result self.client.chat_completion(messages, temperature0.3) if result and len(result) max_length 20: # 允许一定误差 return {success: True, summary: result.strip()} else: # 结果过长尝试截断或重试 truncated result[:max_length] ... if result else 生成失败 return {success: True, summary: truncated, truncated: True} except Exception as e: return {success: False, error: str(e)} def classify_text(self, content: str, categories: list) - dict: 文本分类 try: prompt CLASSIFICATION_PROMPT.format( contentcontent, categories、.join(categories) ) messages [{role: user, content: prompt}] result self.client.chat_completion(messages, temperature0.1) # 清理响应提取类别 if result: cleaned re.sub(r[^\w\u4e00-\u9fff], , result.strip()) for category in categories: if category in cleaned: return {success: True, category: category} return {success: False, error: 无法识别类别, raw_response: result} except Exception as e: return {success: False, error: str(e)}4.2 数据模型定义使用 Pydantic 确保数据验证# src/core/models.py from pydantic import BaseModel, Field from typing import Optional, List class SummaryRequest(BaseModel): content: str Field(..., min_length10, description待摘要文本) max_length: Optional[int] Field(100, ge50, le500, description摘要最大长度) class SummaryResponse(BaseModel): success: bool summary: Optional[str] None error: Optional[str] None truncated: Optional[bool] None class ClassificationRequest(BaseModel): content: str Field(..., min_length5, description待分类文本) categories: List[str] Field(..., min_items2, description候选类别列表) class ClassificationResponse(BaseModel): success: bool category: Optional[str] None error: Optional[str] None5. 构建 Web API 接口使用 FastAPI 提供 RESTful 接口便于前端和其他服务调用。5.1 API 路由实现# src/api/routes.py from fastapi import APIRouter, HTTPException from src.core.services import TextProcessingService from src.core.models import SummaryRequest, SummaryResponse, ClassificationRequest, ClassificationResponse from src.core.clients import AIClient from src.config import AIConfig router APIRouter(prefix/api/v1/ai, tags[AI服务]) # 初始化AI客户端 ai_config AIConfig() ai_client AIClient(ai_config) text_service TextProcessingService(ai_client) router.post(/summary, response_modelSummaryResponse) async def generate_summary(request: SummaryRequest): 生成文本摘要 result text_service.generate_summary(request.content, request.max_length) if not result[success]: raise HTTPException(status_code400, detailresult[error]) return SummaryResponse(**result) router.post(/classify, response_modelClassificationResponse) async def classify_text(request: ClassificationRequest): 文本分类 if len(request.categories) 2: raise HTTPException(status_code400, detail至少需要2个分类类别) result text_service.classify_text(request.content, request.categories) if not result[success]: raise HTTPException(status_code400, detailresult[error]) return ClassificationResponse(**result) router.get(/health) async def health_check(): 健康检查端点 return {status: healthy, model: ai_config.MODEL_NAME}5.2 应用入口配置# main.py from fastapi import FastAPI from src.api.routes import router from src.config import AIConfig import uvicorn # 验证配置 try: AIConfig.validate_config() except ValueError as e: print(f配置错误: {e}) exit(1) app FastAPI( titleAI工程实践示例, description展示AI能力集成的最佳实践, version1.0.0 ) app.include_router(router) app.on_event(startup) async def startup_event(): print(AI服务启动完成) if __name__ __main__: uvicorn.run(main:app, host0.0.0.0, port8000, reloadTrue)6. 测试与验证策略6.1 单元测试编写# tests/test_services.py import pytest from src.core.services import TextProcessingService from src.core.clients import AIClient from src.config import AIConfig class MockAIClient: 模拟AI客户端用于测试 def chat_completion(self, messages, temperature0.7): content messages[0][content] if 摘要 in content: return 这是一个测试摘要。 elif 分类 in content: return 科技 return 默认响应 def test_summary_service(): config AIConfig() mock_client MockAIClient() service TextProcessingService(mock_client) # 正常情况测试 result service.generate_summary(这是一段需要摘要的长文本内容。 * 10) assert result[success] True assert 测试摘要 in result[summary] # 异常情况测试 result service.generate_summary(短) assert result[success] False assert 文本内容过短 in result[error] def test_classification_service(): config AIConfig() mock_client MockAIClient() service TextProcessingService(mock_client) categories [科技, 体育, 娱乐] result service.classify_text(人工智能技术发展, categories) assert result[success] True assert result[category] 科技6.2 API 接口测试使用 curl 或 httpie 测试接口# 测试摘要接口 curl -X POST http://localhost:8000/api/v1/ai/summary \ -H Content-Type: application/json \ -d {content: 这里是需要摘要的长文本内容..., max_length: 150} # 测试分类接口 curl -X POST http://localhost:8000/api/v1/ai/classify \ -H Content-Type: application/json \ -d {content: NBA总决赛精彩回顾, categories: [体育, 科技, 财经]}7. 生产环境部署考虑7.1 环境配置管理生产环境需要使用环境变量或配置中心# .env.production OPENAI_API_KEYsk-prod-... MODEL_NAMEgpt-4 MAX_RETRIES5 TIMEOUT60 LOG_LEVELINFO7.2 容器化部署使用 Docker 确保环境一致性# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]7.3 监控与日志添加详细的日志记录和性能监控# 在客户端中添加日志 def chat_completion(self, messages: list, temperature: float 0.7) - Optional[str]: start_time time.time() logger.info(fAI请求开始: model{self.model}, messages_count{len(messages)}) try: # ... 原有逻辑 end_time time.time() logger.info(fAI请求成功: duration{end_time-start_time:.2f}s) return response except Exception as e: logger.error(fAI请求失败: {e}, exc_infoTrue) raise8. 常见问题与排查指南8.1 API 调用问题排查问题现象可能原因检查步骤解决方案请求超时网络问题/模型负载高检查网络连接测试其他接口增加超时时间实现重试机制认证失败API Key 错误或过期验证 API Key 格式和权限重新生成 API Key检查余额速率限制请求频率过高查看响应头中的限制信息实现指数退避重试控制请求频率内容过滤触发安全策略检查输入内容是否合规清理输入内容调整提示词8.2 响应质量优化# 响应质量检查函数 def validate_ai_response(response: str, expected_type: str) - bool: 验证AI响应是否符合预期格式 if not response or len(response.strip()) 0: return False if expected_type summary: return 10 len(response) 200 # 摘要长度检查 elif expected_type classification: return len(response.strip()) 50 # 分类结果应简洁 return True # 在服务层添加验证 def generate_summary(self, content: str, max_length: int 100) - dict: # ... 原有逻辑 if result and validate_ai_response(result, summary): return {success: True, summary: result.strip()} else: # 响应质量不合格记录日志并返回错误 logger.warning(fAI响应质量不合格: {result}) return {success: False, error: AI响应不符合要求}8.3 成本控制策略# 成本监控装饰器 def cost_monitor(func): def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() # 记录调用统计可接入监控系统 logger.info(fAI调用统计: function{func.__name__}, fduration{end_time-start_time:.2f}s) return result return wrapper # 应用监控 cost_monitor def chat_completion(self, messages: list, temperature: float 0.7): # ... 原有实现9. 最佳实践总结9.1 提示词工程实践将提示词模板化与代码分离管理为不同场景设计专用的提示词模板建立提示词版本管理和A/B测试机制在提示词中明确输出格式要求和约束条件9.2 错误处理与降级实现完整的重试机制指数退避为关键功能设计降级方案如规则引擎备用记录详细的请求日志便于问题排查设置合理的超时时间和响应验证9.3 性能优化建议对频繁请求的结果进行缓存批量处理可以合并的AI请求根据业务场景选择合适的模型规模监控API调用延迟和成功率指标9.4 安全与合规对用户输入进行内容检查和过滤敏感数据避免直接发送给第三方API遵守模型提供商的使用条款建立数据隐私保护机制这个实践框架展示了如何将AI能力系统化地集成到软件工程流程中。实际项目中还需要根据具体业务需求调整架构设计但核心的工程化思路——模块化、可测试、可监控、可维护——是通用的。随着AI技术的快速发展建立良好的工程实践基础比追求最新模型更重要。
AI 技术正在从实验室走向工程化从算法研究变成可复用的开发组件。但很多团队在引入 AI 能力时往往面临一个现实问题如何把 AI 模型、提示词工程、业务逻辑和传统软件工程方法结合起来形成一套可持续迭代的 AI 工程实践体系。本文将以一个实际项目为例展示如何从零搭建一个具备完整 AI 能力的应用涵盖环境准备、模型集成、提示词设计、异常处理和部署上线全流程。1. 理解 AI 工程化的核心挑战AI 工程化不是简单调用 API而是要把非确定性的 AI 输出整合到确定性的软件系统中。这需要解决几个关键问题1.1 模型输出的不确定性处理传统软件开发中相同的输入必然产生相同的输出。但 AI 模型特别是大语言模型的响应存在随机性即使使用相同的提示词也可能得到不同结构和内容的回答。工程上需要设计校验机制、重试策略和降级方案。1.2 提示词版本化管理提示词本质上是另一种形式的代码需要像管理源代码一样进行版本控制、测试和迭代优化。但提示词又不同于传统代码它的效果评估更主观需要建立评估标准和 A/B 测试机制。1.3 成本与延迟的平衡直接调用云端大模型 API 虽然方便但存在成本高、网络延迟、数据隐私等问题。本地部署的小模型虽然可控性强但能力有限。工程上需要根据场景选择合适的模型策略甚至设计分层调用方案。1.4 与传统系统的集成AI 功能很少独立存在通常需要与现有的用户系统、数据库、业务流程集成。这涉及到数据格式转换、异步处理、状态管理等复杂问题。2. 项目环境与工具链准备2.1 基础开发环境建议使用 Python 3.9 作为主要开发语言这是目前 AI 生态最完善的环境。同时需要准备以下核心工具# 创建虚拟环境 python -m venv ai_project source ai_project/bin/activate # Linux/Mac # ai_project\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain fastapi uvicorn pydantic2.2 模型访问配置根据项目需求选择模型提供商。如果使用 OpenAI 系列模型需要配置 API Key# config.py import os from typing import Optional class AIConfig: OPENAI_API_KEY: str os.getenv(OPENAI_API_KEY, ) MODEL_NAME: str gpt-3.5-turbo # 默认模型 MAX_RETRIES: int 3 # API 调用重试次数 TIMEOUT: int 30 # 请求超时时间 classmethod def validate_config(cls): if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY 环境变量未设置)2.3 项目结构设计清晰的目录结构有助于维护 AI 项目的复杂性ai_project/ ├── src/ │ ├── core/ # 核心AI功能 │ │ ├── __init__.py │ │ ├── models.py # 数据模型 │ │ ├── prompts/ # 提示词模板 │ │ └── clients.py # AI客户端封装 │ ├── api/ # Web接口层 │ ├── utils/ # 工具函数 │ └── config.py # 配置文件 ├── tests/ # 测试用例 ├── requirements.txt # 依赖列表 └── main.py # 应用入口3. 构建可维护的 AI 客户端直接裸调用 AI API 会导致代码分散、难以维护。应该封装统一的客户端来处理重试、异常、日志等通用逻辑。3.1 基础客户端实现# src/core/clients.py import logging import time from typing import Dict, Any, Optional from openai import OpenAI, APIError, RateLimitError logger logging.getLogger(__name__) class AIClient: def __init__(self, config): self.client OpenAI(api_keyconfig.OPENAI_API_KEY) self.config config self.model config.MODEL_NAME def chat_completion(self, messages: list, temperature: float 0.7) - Optional[str]: 带重试机制的聊天补全调用 for attempt in range(self.config.MAX_RETRIES): try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, timeoutself.config.TIMEOUT ) return response.choices[0].message.content except RateLimitError as e: wait_time 2 ** attempt # 指数退避 logger.warning(f速率限制{wait_time}秒后重试: {e}) time.sleep(wait_time) except APIError as e: logger.error(fAPI错误: {e}) if attempt self.config.MAX_RETRIES - 1: raise time.sleep(1) except Exception as e: logger.error(f未知错误: {e}) break return None3.2 提示词模板管理将提示词从代码中分离出来便于管理和迭代# src/core/prompts/__init__.py from string import Template from typing import Dict, Any class PromptTemplate: def __init__(self, template: str, required_vars: list): self.template Template(template) self.required_vars required_vars def format(self, **kwargs) - str: # 检查必需变量 missing_vars [var for var in self.required_vars if var not in kwargs] if missing_vars: raise ValueError(f缺少必需变量: {missing_vars}) return self.template.substitute(**kwargs) # 具体提示词定义 SUMMARY_PROMPT PromptTemplate( template请为以下文本生成一个简洁的摘要要求 1. 保留核心信息 2. 长度控制在100字以内 3. 语言流畅自然 文本内容 $content 摘要, required_vars[content] ) CLASSIFICATION_PROMPT PromptTemplate( template请对以下文本进行分类从预定义类别中选择最合适的一项。 预定义类别$categories 文本内容 $content 请只返回类别名称不要额外解释。, required_vars[content, categories] )4. 实现业务逻辑层AI 能力需要封装成具体的业务功能而不是直接暴露给上层调用。4.1 文本处理服务示例# src/core/services.py from .clients import AIClient from .prompts import SUMMARY_PROMPT, CLASSIFICATION_PROMPT import re class TextProcessingService: def __init__(self, ai_client: AIClient): self.client ai_client def generate_summary(self, content: str, max_length: int 100) - dict: 生成文本摘要 if len(content.strip()) 10: return {success: False, error: 文本内容过短} try: prompt SUMMARY_PROMPT.format(contentcontent) messages [{role: user, content: prompt}] result self.client.chat_completion(messages, temperature0.3) if result and len(result) max_length 20: # 允许一定误差 return {success: True, summary: result.strip()} else: # 结果过长尝试截断或重试 truncated result[:max_length] ... if result else 生成失败 return {success: True, summary: truncated, truncated: True} except Exception as e: return {success: False, error: str(e)} def classify_text(self, content: str, categories: list) - dict: 文本分类 try: prompt CLASSIFICATION_PROMPT.format( contentcontent, categories、.join(categories) ) messages [{role: user, content: prompt}] result self.client.chat_completion(messages, temperature0.1) # 清理响应提取类别 if result: cleaned re.sub(r[^\w\u4e00-\u9fff], , result.strip()) for category in categories: if category in cleaned: return {success: True, category: category} return {success: False, error: 无法识别类别, raw_response: result} except Exception as e: return {success: False, error: str(e)}4.2 数据模型定义使用 Pydantic 确保数据验证# src/core/models.py from pydantic import BaseModel, Field from typing import Optional, List class SummaryRequest(BaseModel): content: str Field(..., min_length10, description待摘要文本) max_length: Optional[int] Field(100, ge50, le500, description摘要最大长度) class SummaryResponse(BaseModel): success: bool summary: Optional[str] None error: Optional[str] None truncated: Optional[bool] None class ClassificationRequest(BaseModel): content: str Field(..., min_length5, description待分类文本) categories: List[str] Field(..., min_items2, description候选类别列表) class ClassificationResponse(BaseModel): success: bool category: Optional[str] None error: Optional[str] None5. 构建 Web API 接口使用 FastAPI 提供 RESTful 接口便于前端和其他服务调用。5.1 API 路由实现# src/api/routes.py from fastapi import APIRouter, HTTPException from src.core.services import TextProcessingService from src.core.models import SummaryRequest, SummaryResponse, ClassificationRequest, ClassificationResponse from src.core.clients import AIClient from src.config import AIConfig router APIRouter(prefix/api/v1/ai, tags[AI服务]) # 初始化AI客户端 ai_config AIConfig() ai_client AIClient(ai_config) text_service TextProcessingService(ai_client) router.post(/summary, response_modelSummaryResponse) async def generate_summary(request: SummaryRequest): 生成文本摘要 result text_service.generate_summary(request.content, request.max_length) if not result[success]: raise HTTPException(status_code400, detailresult[error]) return SummaryResponse(**result) router.post(/classify, response_modelClassificationResponse) async def classify_text(request: ClassificationRequest): 文本分类 if len(request.categories) 2: raise HTTPException(status_code400, detail至少需要2个分类类别) result text_service.classify_text(request.content, request.categories) if not result[success]: raise HTTPException(status_code400, detailresult[error]) return ClassificationResponse(**result) router.get(/health) async def health_check(): 健康检查端点 return {status: healthy, model: ai_config.MODEL_NAME}5.2 应用入口配置# main.py from fastapi import FastAPI from src.api.routes import router from src.config import AIConfig import uvicorn # 验证配置 try: AIConfig.validate_config() except ValueError as e: print(f配置错误: {e}) exit(1) app FastAPI( titleAI工程实践示例, description展示AI能力集成的最佳实践, version1.0.0 ) app.include_router(router) app.on_event(startup) async def startup_event(): print(AI服务启动完成) if __name__ __main__: uvicorn.run(main:app, host0.0.0.0, port8000, reloadTrue)6. 测试与验证策略6.1 单元测试编写# tests/test_services.py import pytest from src.core.services import TextProcessingService from src.core.clients import AIClient from src.config import AIConfig class MockAIClient: 模拟AI客户端用于测试 def chat_completion(self, messages, temperature0.7): content messages[0][content] if 摘要 in content: return 这是一个测试摘要。 elif 分类 in content: return 科技 return 默认响应 def test_summary_service(): config AIConfig() mock_client MockAIClient() service TextProcessingService(mock_client) # 正常情况测试 result service.generate_summary(这是一段需要摘要的长文本内容。 * 10) assert result[success] True assert 测试摘要 in result[summary] # 异常情况测试 result service.generate_summary(短) assert result[success] False assert 文本内容过短 in result[error] def test_classification_service(): config AIConfig() mock_client MockAIClient() service TextProcessingService(mock_client) categories [科技, 体育, 娱乐] result service.classify_text(人工智能技术发展, categories) assert result[success] True assert result[category] 科技6.2 API 接口测试使用 curl 或 httpie 测试接口# 测试摘要接口 curl -X POST http://localhost:8000/api/v1/ai/summary \ -H Content-Type: application/json \ -d {content: 这里是需要摘要的长文本内容..., max_length: 150} # 测试分类接口 curl -X POST http://localhost:8000/api/v1/ai/classify \ -H Content-Type: application/json \ -d {content: NBA总决赛精彩回顾, categories: [体育, 科技, 财经]}7. 生产环境部署考虑7.1 环境配置管理生产环境需要使用环境变量或配置中心# .env.production OPENAI_API_KEYsk-prod-... MODEL_NAMEgpt-4 MAX_RETRIES5 TIMEOUT60 LOG_LEVELINFO7.2 容器化部署使用 Docker 确保环境一致性# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]7.3 监控与日志添加详细的日志记录和性能监控# 在客户端中添加日志 def chat_completion(self, messages: list, temperature: float 0.7) - Optional[str]: start_time time.time() logger.info(fAI请求开始: model{self.model}, messages_count{len(messages)}) try: # ... 原有逻辑 end_time time.time() logger.info(fAI请求成功: duration{end_time-start_time:.2f}s) return response except Exception as e: logger.error(fAI请求失败: {e}, exc_infoTrue) raise8. 常见问题与排查指南8.1 API 调用问题排查问题现象可能原因检查步骤解决方案请求超时网络问题/模型负载高检查网络连接测试其他接口增加超时时间实现重试机制认证失败API Key 错误或过期验证 API Key 格式和权限重新生成 API Key检查余额速率限制请求频率过高查看响应头中的限制信息实现指数退避重试控制请求频率内容过滤触发安全策略检查输入内容是否合规清理输入内容调整提示词8.2 响应质量优化# 响应质量检查函数 def validate_ai_response(response: str, expected_type: str) - bool: 验证AI响应是否符合预期格式 if not response or len(response.strip()) 0: return False if expected_type summary: return 10 len(response) 200 # 摘要长度检查 elif expected_type classification: return len(response.strip()) 50 # 分类结果应简洁 return True # 在服务层添加验证 def generate_summary(self, content: str, max_length: int 100) - dict: # ... 原有逻辑 if result and validate_ai_response(result, summary): return {success: True, summary: result.strip()} else: # 响应质量不合格记录日志并返回错误 logger.warning(fAI响应质量不合格: {result}) return {success: False, error: AI响应不符合要求}8.3 成本控制策略# 成本监控装饰器 def cost_monitor(func): def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() # 记录调用统计可接入监控系统 logger.info(fAI调用统计: function{func.__name__}, fduration{end_time-start_time:.2f}s) return result return wrapper # 应用监控 cost_monitor def chat_completion(self, messages: list, temperature: float 0.7): # ... 原有实现9. 最佳实践总结9.1 提示词工程实践将提示词模板化与代码分离管理为不同场景设计专用的提示词模板建立提示词版本管理和A/B测试机制在提示词中明确输出格式要求和约束条件9.2 错误处理与降级实现完整的重试机制指数退避为关键功能设计降级方案如规则引擎备用记录详细的请求日志便于问题排查设置合理的超时时间和响应验证9.3 性能优化建议对频繁请求的结果进行缓存批量处理可以合并的AI请求根据业务场景选择合适的模型规模监控API调用延迟和成功率指标9.4 安全与合规对用户输入进行内容检查和过滤敏感数据避免直接发送给第三方API遵守模型提供商的使用条款建立数据隐私保护机制这个实践框架展示了如何将AI能力系统化地集成到软件工程流程中。实际项目中还需要根据具体业务需求调整架构设计但核心的工程化思路——模块化、可测试、可监控、可维护——是通用的。随着AI技术的快速发展建立良好的工程实践基础比追求最新模型更重要。
