# 旅游内容自动生成系统 - 开发指南 ## 1. 系统架构 旅游内容自动生成系统采用模块化架构设计，主要分为核心模块(core)和工具模块(utils)两大部分。系统通过Pipeline模式协调各个组件的工作，实现从选题生成到内容审核的完整流程。 ### 1.1 架构图 ``` ┌─────────────┐ │ 主程序 │ │ main.py │ └──────┬──────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ PipelineManager │ └───────┬───────────────────┬──────────────────┬──────────────┘ │ │ │ ▼ ▼ ▼ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ TopicGenerator │ │ContentGenerator│ │ ContentJudger │ └───────┬───────┘ └───────┬───────┘ └───────┬───────┘ │ │ │ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────────────────────────────┐ │ AIAgent │ └─────────────────────────────────────────────────────────────┘ ▲ ▲ ▲ │ │ │ │ │ │ ┌───────┴───────┐ ┌───────┴───────┐ ┌──────┴────────┐ │PromptBuilders │ │ ResourceLoader│ │ OutputManager │ └───────────────┘ └───────────────┘ └───────────────┘ ``` ### 1.2 数据流 1. 配置加载: 系统启动时，ConfigManager加载所有配置文件 2. 选题生成: TopicGenerator调用AIAgent生成选题 3. 内容生成: ContentGenerator为每个选题调用AIAgent生成内容 4. 内容审核: ContentJudger审核生成的内容 5. 结果保存: OutputManager保存所有生成的内容和中间产物 ## 2. 代码结构 ### 2.1 目录结构 ``` . ├── main.py # 主程序入口 ├── config/ # 配置文件目录 │ ├── ai_model.json # AI模型配置 │ ├── content_gen.json # 内容生成配置 │ ├── resource.json # 资源配置 │ ├── system.json # 系统配置 │ └── topic_gen.json # 选题生成配置 ├── core/ # 核心模块 │ ├── ai/ # AI代理 │ │ ├── __init__.py │ │ └── ai_agent.py # AI代理实现 │ ├── config/ # 配置管理 │ │ ├── __init__.py │ │ └── config_manager.py # 配置管理器 │ └── exception/ # 异常处理 │ ├── __init__.py │ └── exceptions.py # 自定义异常类 ├── utils/ # 工具模块 │ ├── file_io.py # 文件IO │ ├── pipeline.py # 流程管理 │ ├── prompts.py # 提示词构建 │ └── tweet/ # 内容生成相关 │ ├── __init__.py │ ├── content_generator.py # 内容生成器 │ ├── content_judger.py # 内容审核器 │ ├── topic_generator.py # 选题生成器 │ └── topic_parser.py # 选题解析器 ├── resource/ # 资源文件 │ ├── data/ # 数据文件 │ └── prompt/ # 提示词模板 ├── result/ # 输出结果 └── requirements.txt # 依赖包 ``` ### 2.2 主要模块说明 #### 2.2.1 核心模块 (core) - **ai_agent.py**: 负责与AI模型的交互，封装API调用逻辑 - **config_manager.py**: 负责加载和管理配置文件 - **exceptions.py**: 定义系统自定义异常类 #### 2.2.2 工具模块 (utils) - **file_io.py**: 提供文件读写功能，包括ResourceLoader和OutputManager - **pipeline.py**: 实现整个生成流程的管理 - **prompts.py**: 提供提示词模板和构建功能 - **tweet/**: 内容生成相关的具体实现 - **topic_generator.py**: 选题生成器 - **content_generator.py**: 内容生成器 - **content_judger.py**: 内容审核器 - **topic_parser.py**: 选题解析器 ## 3. 核心类说明 ### 3.1 AIAgent 类 `AIAgent` 类是系统与AI模型交互的核心，负责发送请求并处理响应。 ```python class AIAgent: def __init__(self, config: AIModelConfig): # 初始化AI代理，设置API客户端 async def generate_text( self, system_prompt: str, user_prompt: str, use_stream: bool = False, temperature: Optional[float] = None, top_p: Optional[float] = None, presence_penalty: Optional[float] = None, stage: str = "" ) -> Tuple[str, int, int, float]: # 生成文本，支持流式和非流式输出 # 返回: (生成文本, 输入token数, 输出token数, 耗时) ``` ### 3.2 ConfigManager 类 `ConfigManager` 类负责加载和管理配置文件。 ```python class ConfigManager: def __init__(self): # 初始化配置管理器 def load_from_directory(self, config_dir: str): # 从目录加载所有配置文件 def get_config(self, config_name: str, config_class: Type[T]) -> T: # 获取指定名称的配置对象 ``` ### 3.3 PipelineManager 类 `PipelineManager` 类负责协调整个内容生成流程。 ```python class PipelineManager: def __init__(self, config_dir: str, run_id: Optional[str] = None): # 初始化管道管理器 async def process_content_generation(self, topics: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 处理内容生成 async def process_content_judging(self, generated_contents: List[Dict[str, Any]]): # 处理内容审核 async def run_pipeline(self): # 按顺序执行整个流程 ``` ### 3.4 提示词构建器类 系统包含多个提示词构建器，负责构建不同阶段的提示词。 ```python class BasePromptBuilder(PromptTemplate): # 基础提示词构建器，提供通用方法 class TopicPromptBuilder(BasePromptBuilder): # 选题生成提示词构建器 class ContentPromptBuilder(BasePromptBuilder): # 内容生成提示词构建器 class JudgerPromptBuilder(BasePromptBuilder): # 内容审核提示词构建器 ``` ## 4. 关键流程 ### 4.1 选题生成流程 1. 加载配置和资源 2. 构建系统提示和用户提示 3. 调用AI模型生成选题 4. 解析AI响应，提取选题信息 5. 保存选题结果 ```python # 在TopicGenerator.generate_topics方法中实现 async def generate_topics(self) -> Optional[List[Dict[str, Any]]]: # 构建提示 system_prompt = self.prompt_builder.get_system_prompt() user_prompt = self.prompt_builder.build_user_prompt(...) # 调用AI raw_result = await self.ai_agent.generate_text(...) # 解析结果 topics = self.parser.parse(raw_result) # 保存结果 self.output_manager.save_json(topics, "topics_generated.json") return topics ``` ### 4.2 内容生成流程 1. 为每个选题构建提示 2. 调用AI模型生成内容 3. 解析AI响应，提取内容信息 4. 保存生成的内容 ```python # 在ContentGenerator.generate_content_for_topic方法中实现 async def generate_content_for_topic(self, topic: Dict[str, Any]) -> Dict[str, Any]: # 构建提示 system_prompt = self.prompt_builder.get_system_prompt() user_prompt = self.prompt_builder.build_user_prompt(topic=topic) # 调用AI raw_result = await self.ai_agent.generate_text(...) # 解析和保存结果 content_data = json_repair.loads(raw_result) self.output_manager.save_json(content_data, "article.json", ...) return content_data ``` ### 4.3 内容审核流程 1. 为每个生成的内容构建审核提示 2. 调用AI模型进行审核 3. 解析审核结果，提取修改后的内容 4. 保存审核后的内容 ```python # 在ContentJudger.judge_content方法中实现 async def judge_content(self, generated_content: str, topic: Dict[str, Any]) -> Dict[str, Any]: # 构建提示 system_prompt = self.prompt_builder.get_system_prompt() user_prompt = self.prompt_builder.build_user_prompt(...) # 调用AI raw_result = await self.ai_agent.generate_text(...) # 解析结果 judged_data = json_repair_loads(raw_result) # 保存结果 self.output_manager.save_json(judged_data, "article_judged.json", ...) return judged_data ``` ## 5. 扩展指南 ### 5.1 添加新的生成阶段 要添加新的生成阶段（例如图像生成），需要以下步骤： 1. 创建新的生成器类，例如 `ImageGenerator` 2. 创建对应的提示词构建器，例如 `ImagePromptBuilder` 3. 在 `PipelineManager` 中添加新的处理方法 4. 在 `run_pipeline` 方法中调用新的处理方法 ```python # 1. 创建新的生成器类 class ImageGenerator: def __init__(self, ai_agent, config_manager, output_manager): # 初始化 async def generate_image_for_content(self, content): # 生成图像的逻辑 # 2. 在PipelineManager中添加 def __init__(self, ...): # 现有初始化代码 self.image_generator = ImageGenerator(...) async def process_image_generation(self, contents): # 处理图像生成的逻辑 async def run_pipeline(self): # 现有流程代码 # 添加新阶段 if generated_contents: await self.process_image_generation(generated_contents) ``` ### 5.2 添加新的AI模型 要支持新的AI模型，需要修改 `AIAgent` 类： 1. 更新 `__init__` 方法，支持新的模型配置 2. 在 `generate_text` 方法中添加对新模型的支持 3. 更新配置文件结构 ```python class AIAgent: def __init__(self, config: AIModelConfig): self.config = config if config.provider == "openai": self.client = AsyncOpenAI(...) elif config.provider == "new_provider": self.client = NewProviderClient(...) async def generate_text(self, ...): # 根据provider选择不同的API调用方式 if self.config.provider == "openai": # 现有OpenAI调用逻辑 elif self.config.provider == "new_provider": # 新提供商的API调用逻辑 ``` ### 5.3 自定义资源加载器 要支持新的资源类型或加载方式，可以扩展 `ResourceLoader` 类： ```python class ResourceLoader: # 现有方法 @staticmethod def load_special_format(file_path: str) -> Optional[Dict]: # 加载特殊格式文件的逻辑 @staticmethod def load_from_database(query: str) -> Optional[str]: # 从数据库加载资源的逻辑 ``` ### 5.4 添加新的输出格式 要支持新的输出格式，可以扩展 `OutputManager` 类： ```python class OutputManager: # 现有方法 def save_html(self, content: str, filename: str, subdir: Optional[str] = None): # 保存HTML格式的逻辑 def export_to_cms(self, content: Dict, api_endpoint: str): # 导出到CMS系统的逻辑 ``` ## 6. 最佳实践 ### 6.1 代码风格 - 遵循PEP 8规范 - 使用类型注解提高代码可读性 - 为所有公共方法和类提供文档字符串 - 使用异步编程处理IO密集型操作 ### 6.2 错误处理 - 使用自定义异常类区分不同类型的错误 - 实现重试机制处理临时性故障 - 记录详细的错误信息，便于调试 ```python try: result = await self.ai_agent.generate_text(...) except RetryableError as e: # 处理可重试错误 logger.warning(f"遇到可重试错误: {e}") # 实现重试逻辑 except NonRetryableError as e: # 处理不可重试错误 logger.error(f"遇到不可重试错误: {e}") # 返回错误信息 except Exception as e: # 处理未预期的错误 logger.critical(f"遇到未知错误: {e}", exc_info=True) # 返回通用错误信息 ``` ### 6.3 性能优化 - 使用异步IO提高并发性能 - 实现缓存机制减少重复计算 - 优化提示词减少token使用量 - 使用采样率控制参考内容的使用量 ### 6.4 测试策略 - 为核心功能编写单元测试 - 使用模拟对象测试AI调用 - 编写集成测试验证完整流程 - 使用参数化测试覆盖多种场景 ```python # 单元测试示例 async def test_topic_parser(): parser = TopicParser() raw_json = '{"topics": [{"index": "1", "date": "2024-07-01", ...}]}' topics = parser.parse(raw_json) assert len(topics) == 1 assert topics[0]["index"] == "1" assert topics[0]["date"] == "2024-07-01" ``` ## 7. 常见问题与解决方案 ### 7.1 AI调用超时 **问题**: AI模型调用经常超时。 **解决方案**: 1. 增加超时设置 2. 实现指数退避重试 3. 减少提示词长度 4. 考虑使用流式响应 ```python # 实现指数退避重试 backoff_time = 1.0 for attempt in range(max_retries): try: response = await client.chat.completions.create(...) return response except APITimeoutError: logger.warning(f"尝试 {attempt + 1}/{max_retries} 超时") await asyncio.sleep(backoff_time) backoff_time *= 2 # 指数增长 ``` ### 7.2 内存使用过高 **问题**: 处理大量内容时内存使用过高。 **解决方案**: 1. 使用生成器处理大量数据 2. 实现批处理机制 3. 及时释放不再需要的资源 ```python # 使用生成器处理大量数据 async def process_topics_in_batches(self, topics, batch_size=5): for i in range(0, len(topics), batch_size): batch = topics[i:i+batch_size] results = await self.process_batch(batch) for result in results: yield result ``` ### 7.3 JSON解析错误 **问题**: AI返回的内容无法解析为JSON。 **解决方案**: 1. 使用更健壮的JSON解析库（如json_repair） 2. 在提示词中强调JSON格式要求 3. 实现后处理逻辑修复常见错误 ```python # 使用json_repair修复JSON try: data = json.loads(raw_result) except json.JSONDecodeError: try: data = json_repair_loads(raw_result) logger.warning("使用json_repair修复了JSON格式") except Exception as e: logger.error(f"JSON修复失败: {e}") # 返回错误信息 ``` ## 8. 未来扩展方向 1. **多模态支持**: 添加图像生成和处理能力 2. **用户反馈系统**: 收集和利用用户对生成内容的反馈 3. **内容分发集成**: 与社交媒体平台API集成，自动发布内容 4. **A/B测试框架**: 比较不同提示词和参数的效果 5. **本地模型支持**: 支持本地部署的开源模型 ## 9. 贡献指南 1. Fork仓库并创建功能分支 2. 编写代码并添加测试 3. 确保所有测试通过 4. 提交Pull Request并描述变更 5. 等待代码审查和合并