我是 HolySheep AI 的技术布道师，在过去一年里帮助超过 200 家企业完成了 AI 能力的接入与落地。今天我要分享的是如何用 AI API 构建一套员工培训内容智能生成系统，从 0 到 1 搭建生产级别的完整方案。
这套系统的核心价值在于：将传统的课程设计周期从 2 周压缩到 2 小时，同时保持内容质量的稳定性。通过本文，你将获得可直接部署的架构设计、经过压力测试的代码实现，以及真实的性能基准数据。
系统架构设计
员工培训内容生成系统的核心挑战在于：需要同时满足内容质量、生成速度、成本控制三个维度。我在设计时采用了三层架构：
- 接入层：统一 API 网关，负责认证、限流、日志
- 业务层：内容生成引擎，包含模板管理、多模型路由、输出校验
- 数据层：向量知识库、生成历史缓存、企业知识索引
# 项目目录结构
training-content-api/
├── config/
│ ├── __init__.py
│ └── settings.py # 配置管理
├── core/
│ ├── __init__.py
│ ├── generator.py # 核心生成器
│ ├── router.py # 模型路由
│ └── validator.py # 内容校验
├── api/
│ ├── __init__.py
│ └── routes.py # FastAPI 路由
├── utils/
│ ├── __init__.py
│ └── cache.py # 缓存工具
├── tests/
│ └── test_generator.py # 单元测试
└── main.py # 入口文件
生产级代码实现
配置管理
# config/settings.py
from pydantic_settings import BaseSettings
from typing import Optional
import os
class Settings(BaseSettings):
# HolySheep API 配置 - ¥1=$1 无损汇率
HOLYSHEEP_API_KEY: str = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
# 模型配置 - 根据内容类型选择合适模型
MODEL_FOR_COURSE: str = "gpt-4.1" # 课程设计：高质量
MODEL_FOR_QUIZ: str = "gpt-4.1" # 题目生成：高质量
MODEL_FOR_SUMMARY: str = "deepseek-chat" # 摘要总结：低成本
# 性能配置
MAX_CONCURRENT_REQUESTS: int = 50 # 最大并发
REQUEST_TIMEOUT: int = 60 # 超时时间（秒）
CACHE_TTL: int = 3600 # 缓存有效期
# 内容配置
MAX_TOKENS_PER_REQUEST: int = 8192
TEMPERATURE: float = 0.7
class Config:
env_file = ".env"
settings = Settings()
核心生成器实现
# core/generator.py
import aiohttp
import json
import hashlib
from typing import Dict, List, Optional
from datetime import datetime
from .router import ModelRouter
from .validator import ContentValidator
from utils.cache import RedisCache
class TrainingContentGenerator:
"""
员工培训内容生成器
支持：课程大纲、章节内容、测试题目、学习总结
"""
def __init__(self, api_key: str, base_url: str):
self.api_key = api_key
self.base_url = base_url
self.router = ModelRouter()
self.validator = ContentValidator()
self.cache = RedisCache()
async def generate_course_outline(
self,
topic: str,
target_level: str = "中级",
duration_hours: int = 4
) -> Dict:
"""
生成课程大纲
Args:
topic: 培训主题
target_level: 目标受众水平 (初级/中级/高级)
duration_hours: 课程时长（小时）
"""
cache_key = self._generate_cache_key("outline", topic, target_level)
# 尝试从缓存获取
cached = await self.cache.get(cache_key)
if cached:
return json.loads(cached)
prompt = f"""请为以下培训主题设计完整的课程大纲：
主题：{topic}
目标受众：{target_level}员工
课程时长：{duration_hours}小时
请按以下JSON格式输出：
{{
"course_name": "课程名称",
"total_duration": "总时长",
"modules": [
{{
"title": "模块标题",
"duration": "分钟数",
"objectives": ["学习目标1", "学习目标2"],
"content_points": ["内容要点1", "内容要点2"],
"practical_exercises": ["实操练习1"]
}}
],
"assessment_method": "考核方式"
}}"""
response = await self._call_api(
model=self.router.select_model("course_outline"),
messages=[{"role": "user", "content": prompt}],
max_tokens=4096
)
result = self.validator.validate_outline(response)
await self.cache.set(cache_key, json.dumps(result, ensure_ascii=False))
return result
async def generate_module_content(
self,
module: Dict,
company_context: Optional[str] = None
) -> str:
"""
生成模块详细内容
"""
prompt = f"""请为以下课程模块生成详细内容：
模块标题：{module['title']}
学习目标：{', '.join(module['objectives'])}
内容要点：{', '.join(module['content_points'])}
{f'公司背景信息：{company_context}' if company_context else ''}
要求：
1. 内容专业、实用，紧密结合工作场景
2. 包含具体案例和最佳实践
3. 语言通俗易懂，适合培训场景
4. 使用 Markdown 格式输出，包含适当的标题层级"""
response = await self._call_api(
model=self.router.select_model("module_content"),
messages=[{"role": "user", "content": prompt}],
max_tokens=8192
)
return response
async def generate_quiz(
self,
content: str,
question_count: int = 10,
question_types: List[str] = ["选择题", "判断题", "简答题"]
) -> Dict:
"""
生成测试题目
"""
prompt = f"""请根据以下培训内容生成测试题目：
{content}
要求：
- 共生成 {question_count} 道题目
- 题型分布：{', '.join(question_types)}
- 每道题目都要有标准答案和解析
- 题目难度适中，考察核心知识点
按以下JSON格式输出：
{{
"questions": [
{{
"type": "选择题",
"question": "题目内容",
"options": ["A. 选项1", "B. 选项2", "C. 选项3", "D. 选项4"],
"answer": "B",
"explanation": "答案解析"
}}
]
}}"""
response = await self._call_api(
model=self.router.select_model("quiz"),
messages=[{"role": "user", "content": prompt}],
max_tokens=4096
)
return self.validator.validate_quiz(response)
async def _call_api(
self,
model: str,
messages: List[Dict],
max_tokens: int = 2048
) -> str:
"""
调用 HolySheep API
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type":