5.3 内容生成平台案例

5.3 内容生成平台案例

本节导读

本节通过实际案例讲解内容生成平台的设计与实现，涵盖批量文章生成、多语言翻译、SEO优化、质量审核和发布自动化等核心功能，帮助读者构建完整的内容生成系统。

学习目标

• 掌握内容生成平台的整体架构设计
• 学会批量文章生成的技术实现
• 理解多语言翻译系统的设计与优化
• 掌握SEO优化的关键技术和实现方法
• 学会质量审核和发布自动化的完整流程

核心概念

内容生成平台架构

内容生成平台是一个复杂的系统，需要处理内容创作、翻译、优化、审核和发布等多个环节。其核心架构包括：

关键技术组件

1. 内容生成引擎：基于大模型API生成原始内容
2. 多语言翻译：支持多种语言之间的翻译服务
3. SEO优化：优化内容以提升搜索引擎排名
4. 质量审核：检查内容质量和合规性
5. 内容存储：管理内容和版本信息
6. 发布自动化：自动发布到各个平台
7. 监控分析：监控各环节性能和质量

分步实战

步骤1：内容生成引擎实现

构建内容生成引擎，支持批量内容生成：

# content_generator.py
import asyncio
import time
import json
import aiohttp
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
import uuid
import logging
from concurrent.futures import ThreadPoolExecutor
import re

class ContentFormat(Enum):
    """内容格式枚举"""
    ARTICLE = "article"
    BLOG = "blog"
    NEWS = "news"
    PRODUCT = "product"
    SOCIAL = "social"
    REPORT = "report"
    GUIDE = "guide"

class ContentStatus(Enum):
    """内容状态枚举"""
    PENDING = "pending"
    GENERATING = "generating"
    COMPLETED = "completed"
    FAILED = "failed"
    TRANSLATING = "translating"
    TRANSLATED = "translated"
    OPTIMIZING = "optimizing"
    OPTIMIZED = "optimized"
    REVIEWING = "reviewing"
    APPROVED = "approved"
    REJECTED = "rejected"
    SCHEDULED = "scheduled"
    PUBLISHED = "published"

@dataclass
class ContentRequest:
    """内容请求"""
    id: str
    title: str
    format: ContentFormat
    topic: str
    keywords: List[str]
    target_audience: str
    word_count: int
    style: str
    language: str
    requirements: Dict[str, Any]
    created_at: float
    updated_at: float
    priority: int = 1
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}
        self.created_at = time.time()
        self.updated_at = time.time()
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'ContentRequest':
        """从字典创建"""
        return cls(**data)

@dataclass
class ContentItem:
    """内容项"""
    id: str
    request_id: str
    content: str
    format: ContentFormat
    language: str
    status: ContentStatus
    word_count: int
    generated_at: float
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}
        self.generated_at = time.time()

class ContentGenerator:
    """内容生成引擎"""
    
    def __init__(self, 
                 llm_api_clients: Dict[str, Any],
                 max_concurrent: int = 5,
                 retry_attempts: int = 3,
                 timeout: int = 60):
        
        self.llm_api_clients = llm_api_clients
        self.max_concurrent = max_concurrent
        self.retry_attempts = retry_attempts
        self.timeout = timeout
        
        # 内容请求存储（实际项目中应使用数据库）
        self.content_requests: Dict[str, ContentRequest] = {}
        self.content_items: Dict[str, ContentItem] = {}
        
        # 线程池
        self.executor = ThreadPoolExecutor(max_workers=max_concurrent)
        
        # 配置日志
        self.logger = logging.getLogger(__name__)
    
    async def create_content_request(self, request_data: Dict[str, Any]) -> ContentRequest:
        """创建内容请求"""
        request_id = str(uuid.uuid4())
        
        # 创建内容请求对象
        request = ContentRequest(
            id=request_id,
            title=request_data.get("title", ""),
            format=ContentFormat(request_data.get("format", "article")),
            topic=request_data.get("topic", ""),
            keywords=request_data.get("keywords", []),
            target_audience=request_data.get("target_audience", "general"),
            word_count=request_data.get("word_count", 1000),
            style=request_data.get("style", "professional"),
            language=request_data.get("language", "zh-CN"),
            requirements=request_data.get("requirements", {}),
            priority=request_data.get("priority", 1)
        )
        
        # 存储请求
        self.content_requests[request_id] = request
        
        self.logger.info(f"创建内容请求: {request_id} - {request.title}")
        return request
    
    async def generate_content(self, request_id: str) -> ContentItem:
        """生成内容"""
        if request_id not in self.content_requests:
            raise ValueError(f"请求不存在: {request_id}")
        
        request = self.content_requests[request_id]
        
        # 创建内容项
        content_id = str(uuid.uuid4())
        content_item = ContentItem(
            id=content_id,
            request_id=request_id,
            format=request.format,
            language=request.language,
            status=ContentStatus.GENERATING,
            word_count=0
        )
        
        self.content_items[content_id] = content_item
        
        try:
            # 生成内容
            content = await self._generate_content_by_api(request)
            
            # 更新内容项
            content_item.content = content
            content_item.word_count = len(content.split())
            content_item.status = ContentStatus.COMPLETED
            content_item.generated_at = time.time()
            
            self.logger.info(f"内容生成完成: {content_id} - {request.title}")
            return content_item
            
        except Exception as e:
            content_item.status = ContentStatus.FAILED
            content_item.metadata = {"error": str(e)}
            self.logger.error(f"内容生成失败: {content_id} - {str(e)}")
            raise
    
    async def _generate_content_by_api(self, request: ContentRequest) -> str:
        """通过API生成内容"""
        # 根据语言选择API客户端
        language_code = request.language.split("-")[0]
        
        if language_code == "zh":
            api_key = "chinese_api_key"
        elif language_code == "en":
            api_key = "english_api_key"
        else:
            api_key = "general_api_key"
        
        if api_key not in self.llm_api_clients:
            raise ValueError(f"不支持的语言: {request.language}")
        
        api_client = self.llm_api_clients[api_key]
        
        # 构建提示词
        prompt = self._build_generation_prompt(request)
        
        # 调用API
        response = await api_client.chat_completion([
            {"role": "system", "content": "你是一个专业的内容创作助手，根据用户要求生成高质量内容。"},
            {"role": "user", "content": prompt}
        ])
        
        content = response.choices[0]["message"]["content"]
        
        # 内容后处理
        content = self._post_process_content(content, request)
        
        return content
    
    def _build_generation_prompt(self, request: ContentRequest) -> str:
        """构建生成提示词"""
        prompt = f"""
请根据以下要求生成{request.value}内容：

**主题**: {request.topic}
**标题**: {request.title}
**目标受众**: {request.target_audience}
**风格**: {request.style}
**字数要求**: {request.word_count}字
**关键词**: {', '.join(request.keywords)}

**具体要求**:
{json.dumps(request.requirements, ensure_ascii=False)}

**内容格式要求**:
- 文章结构清晰，包含引言、主体和结论
- 段落简洁，每段3-5句话
- 逻辑连贯，层次分明
- 语言准确，避免口语化

请直接生成内容，不要包含任何解释或说明。
"""
        return prompt
    
    def _post_process_content(self, content: str, request: ContentRequest) -> str:
        """内容后处理"""
        # 清理多余的空格和换行
        content = re.sub(r'\n\s*\n', '\n\n', content)
        content = re.sub(r'\s+', ' ', content)
        
        # 确保标题存在
        if not content.startswith("#"):
            content = f"# {request.title}\n\n{content}"
        
        # 调整字数
        words = content.split()
        if len(words) > request.word_count:
            # 截断内容
            content = ' '.join(words[:request.word_count])
        elif len(words) < request.word_count * 0.8:
            # 扩展内容（简单方法，实际可以添加相关段落）
            content += f"\n\n（本文由{request.style}风格撰写，包含{request.keywords}等关键词内容。）"
        
        return content
    
    async def batch_generate_content(self, request_ids: List[str]) -> List[ContentItem]:
        """批量生成内容"""
        results = []
        
        # 使用异步控制并发
        semaphore = asyncio.Semaphore(self.max_concurrent)
        
        async def generate_with_semaphore(request_id: str) -> ContentItem:
            async with semaphore:
                return await self.generate_content(request_id)
        
        # 创建批量任务
        tasks = [generate_with_semaphore(request_id) for request_id in request_ids]
        
        # 等待所有任务完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理结果
        successful_results = []
        for result in results:
            if isinstance(result, Exception):
                self.logger.error(f"批量生成失败: {str(result)}")
            else:
                successful_results.append(result)
        
        return successful_results
    
    async def get_content_status(self, content_id: str) -> Optional[ContentStatus]:
        """获取内容状态"""
        if content_id in self.content_items:
            return self.content_items[content_id].status
        return None
    
    async def get_content_by_request(self, request_id: str) -> Optional[ContentItem]:
        """根据请求获取内容"""
        for content_id, content_item in self.content_items.items():
            if content_item.request_id == request_id:
                return content_item
        return None
    
    async def get_generation_statistics(self) -> Dict[str, Any]:
        """获取生成统计信息"""
        total_requests = len(self.content_requests)
        total_content = len(self.content_items)
        
        status_counts = {}
        for content_item in self.content_items.values():
            status = content_item.status.value
            status_counts[status] = status_counts.get(status, 0) + 1
        
        return {
            "total_requests": total_requests,
            "total_content": total_content,
            "status_counts": status_counts,
            "generation_rate": status_counts.get("completed", 0) / total_content if total_content > 0 else 0
        }

# 使用示例
async def content_generator_example():
    """内容生成器使用示例"""
    
    # 模拟LLM客户端
    class MockLLMClient:
        async def chat_completion(self, messages):
            # 模拟生成内容
            topic = messages[1]["content"]
            if "人工智能" in topic:
                content = """
# 人工智能的发展与应用

人工智能是计算机科学的一个重要分支，致力于开发能够像人类一样思考的智能系统。

## 技术基础

人工智能的技术基础包括机器学习、深度学习、自然语言处理等多个领域。机器学习是人工智能的核心技术之一，通过算法让计算机从数据中学习模式。

## 实际应用

人工智能在医疗、金融、教育、制造等多个领域都有广泛应用。在医疗领域，AI可以帮助医生进行疾病诊断和治疗方案制定。

## 未来展望

随着技术的不断发展，人工智能将在更多领域发挥重要作用，为人类社会带来更多便利和价值。
"""
            else:
                content = f"""
# {topic}

这是根据要求生成的内容，包含相关的信息和分析内容。内容结构清晰，层次分明，符合要求的格式和风格要求。

## 核心要点

1. 主题概述
2. 详细分析
3. 应用实例
4. 总结展望

内容长度适中，符合字数要求。
"""
            return type('Response', (), {
                'choices': [{'message': {'content': content}}]
            })()
    
    # 创建内容生成器
    generator = ContentGenerator(
        llm_api_clients={
            "chinese_api_key": MockLLMClient(),
            "english_api_key": MockLLMClient(),
            "general_api_key": MockLLMClient()
        },
        max_concurrent=3
    )
    
    # 创建内容请求
    request1_data = {
        "title": "人工智能的未来发展趋势",
        "format": "article",
        "topic": "人工智能技术发展",
        "keywords": ["AI", "机器学习", "深度学习", "未来趋势"],
        "target_audience": "技术爱好者",
        "word_count": 1500,
        "style": "专业科普",
        "language": "zh-CN",
        "requirements": {
            "include_charts": True,
            "examples_needed": True
        }
    }
    
    request2_data = {
        "title": "Climate Change and Renewable Energy",
        "format": "article",
        "topic": "climate change",
        "keywords": ["climate", "renewable", "energy", "sustainability"],
        "target_audience": "general public",
        "word_count": 1000,
        "style": "informative",
        "language": "en-US",
        "requirements": {
            "include_statistics": True,
            "references_needed": False
        }
    }
    
    # 创建请求
    request1 = await generator.create_content_request(request1_data)
    request2 = await generator.create_content_request(request2_data)
    
    print(f"创建请求1: {request1.id}")
    print(f"创建请求2: {request2.id}")
    
    # 批量生成内容
    request_ids = [request1.id, request2.id]
    generated_contents = await generator.batch_generate_content(request_ids)
    
    print(f"生成内容数量: {len(generated_contents)}")
    
    # 显示生成结果
    for content_item in generated_contents:
        print(f"内容ID: {content_item.id}")
        print(f"状态: {content_item.status}")
        print(f"字数: {content_item.word_count}")
        print(f"内容预览: {content_item.content[:200]}...")
        print("-" * 50)
    
    # 获取统计信息
    stats = await generator.get_generation_statistics()
    print(f"生成统计: {stats}")

if __name__ == "__main__":
    asyncio.run(content_generator_example())
​

步骤2：多语言翻译系统实现

构建多语言翻译系统，支持批量翻译：

# translation_system.py
import asyncio
import time
import aiohttp
import json
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
import uuid
​