jinye_huang/TravelContentCreator

Template

Fork 0

jinye_huang 8326485d8a 增加了文档，完成了生文模块

2025-07-09 16:33:38 +08:00

16 KiB

Raw Blame History

旅游内容自动生成系统 - 开发指南

1. 系统架构

旅游内容自动生成系统采用模块化架构设计，主要分为核心模块(core)和工具模块(utils)两大部分。系统通过Pipeline模式协调各个组件的工作，实现从选题生成到内容审核的完整流程。

1.1 架构图

                                 ┌─────────────┐
                                 │    主程序    │
                                 │   main.py   │
                                 └──────┬──────┘
                                        │
                                        ▼
┌─────────────────────────────────────────────────────────────┐
│                     PipelineManager                         │
└───────┬───────────────────┬──────────────────┬──────────────┘
        │                   │                  │
        ▼                   ▼                  ▼
┌───────────────┐   ┌───────────────┐   ┌───────────────┐
│ TopicGenerator │   │ContentGenerator│   │ ContentJudger │
└───────┬───────┘   └───────┬───────┘   └───────┬───────┘
        │                   │                  │
        │                   │                  │
        ▼                   ▼                  ▼
┌─────────────────────────────────────────────────────────────┐
│                        AIAgent                              │
└─────────────────────────────────────────────────────────────┘
        ▲                   ▲                  ▲
        │                   │                  │
        │                   │                  │
┌───────┴───────┐   ┌───────┴───────┐   ┌──────┴────────┐
│PromptBuilders │   │  ResourceLoader│   │ OutputManager │
└───────────────┘   └───────────────┘   └───────────────┘

1.2 数据流

配置加载: 系统启动时，ConfigManager加载所有配置文件
选题生成: TopicGenerator调用AIAgent生成选题
内容生成: ContentGenerator为每个选题调用AIAgent生成内容
内容审核: ContentJudger审核生成的内容
结果保存: 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模型交互的核心，负责发送请求并处理响应。

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 类负责加载和管理配置文件。

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 类负责协调整个内容生成流程。

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 提示词构建器类

系统包含多个提示词构建器，负责构建不同阶段的提示词。

class BasePromptBuilder(PromptTemplate):
    # 基础提示词构建器，提供通用方法
    
class TopicPromptBuilder(BasePromptBuilder):
    # 选题生成提示词构建器
    
class ContentPromptBuilder(BasePromptBuilder):
    # 内容生成提示词构建器
    
class JudgerPromptBuilder(BasePromptBuilder):
    # 内容审核提示词构建器

4. 关键流程

4.1 选题生成流程

加载配置和资源
构建系统提示和用户提示
调用AI模型生成选题
解析AI响应，提取选题信息
保存选题结果

# 在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 内容生成流程

为每个选题构建提示
调用AI模型生成内容
解析AI响应，提取内容信息
保存生成的内容

# 在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.loads(raw_result)
    self.output_manager.save_json(content_data, "article.json", ...)
    
    return content_data

4.3 内容审核流程

为每个生成的内容构建审核提示
调用AI模型进行审核
解析审核结果，提取修改后的内容
保存审核后的内容

# 在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 添加新的生成阶段

要添加新的生成阶段（例如图像生成），需要以下步骤：

创建新的生成器类，例如 ImageGenerator
创建对应的提示词构建器，例如 ImagePromptBuilder
在 PipelineManager 中添加新的处理方法
在 run_pipeline 方法中调用新的处理方法

# 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 类：

更新 __init__ 方法，支持新的模型配置
在 generate_text 方法中添加对新模型的支持
更新配置文件结构

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 类：

class ResourceLoader:
    # 现有方法
    
    @staticmethod
    def load_special_format(file_path: str) -> Optional[Dict]:
        # 加载特殊格式文件的逻辑
        
    @staticmethod
    def load_from_database(query: str) -> Optional[str]:
        # 从数据库加载资源的逻辑

5.4 添加新的输出格式

要支持新的输出格式，可以扩展 OutputManager 类：

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 错误处理

使用自定义异常类区分不同类型的错误
实现重试机制处理临时性故障
记录详细的错误信息，便于调试

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调用
编写集成测试验证完整流程
使用参数化测试覆盖多种场景

# 单元测试示例
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模型调用经常超时。

解决方案:

增加超时设置
实现指数退避重试
减少提示词长度
考虑使用流式响应

# 实现指数退避重试
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 内存使用过高

问题: 处理大量内容时内存使用过高。

解决方案:

使用生成器处理大量数据
实现批处理机制
及时释放不再需要的资源

# 使用生成器处理大量数据
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。

解决方案:

使用更健壮的JSON解析库（如json_repair）
在提示词中强调JSON格式要求
实现后处理逻辑修复常见错误

# 使用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. 未来扩展方向

多模态支持: 添加图像生成和处理能力
用户反馈系统: 收集和利用用户对生成内容的反馈
内容分发集成: 与社交媒体平台API集成，自动发布内容
A/B测试框架: 比较不同提示词和参数的效果
本地模型支持: 支持本地部署的开源模型

9. 贡献指南

Fork仓库并创建功能分支
编写代码并添加测试
确保所有测试通过
提交Pull Request并描述变更
等待代码审查和合并

16 KiB Raw Blame History Unescape Escape

旅游内容自动生成系统 - 开发指南

1. 系统架构

1.1 架构图

1.2 数据流

2. 代码结构

2.1 目录结构

2.2 主要模块说明

2.2.1 核心模块 (core)

2.2.2 工具模块 (utils)

3. 核心类说明

3.1 AIAgent 类

3.2 ConfigManager 类

3.3 PipelineManager 类

3.4 提示词构建器类

4. 关键流程

4.1 选题生成流程

4.2 内容生成流程

4.3 内容审核流程

5. 扩展指南

5.1 添加新的生成阶段

5.2 添加新的AI模型

5.3 自定义资源加载器

5.4 添加新的输出格式

6. 最佳实践

6.1 代码风格

6.2 错误处理

6.3 性能优化

6.4 测试策略

7. 常见问题与解决方案

7.1 AI调用超时

7.2 内存使用过高

7.3 JSON解析错误

8. 未来扩展方向

9. 贡献指南

16 KiB

Raw Blame History