# Pixelle-Video 优秀设计模式分析

> 📅 分析日期：2026-06-05
> 📌 项目：https://github.com/AIDC-AI/Pixelle-Video
> 💡 用途：后期决定是否借鉴到自己的项目中

---

## 📋 设计模式清单

| # | 设计模式 | 评分 | 复杂度 | 可借鉴度 |
|---|---------|------|--------|---------|
| 1 | Pipeline 架构模式 | ⭐⭐⭐⭐⭐ | 高 | 高 |
| 2 | 配置管理单例模式 | ⭐⭐⭐⭐⭐ | 中 | 高 |
| 3 | 任务管理异步模式 | ⭐⭐⭐⭐⭐ | 中 | 高 |
| 4 | Core 服务层聚合 | ⭐⭐⭐⭐⭐ | 中 | 高 |
| 5 | LLM 结构化输出 | ⭐⭐⭐⭐ | 中 | 中 |
| 6 | Storyboard 数据模型 | ⭐⭐⭐⭐ | 低 | 高 |
| 7 | Prompt 模板分离 | ⭐⭐⭐⭐ | 低 | 高 |
| 8 | API Schema 规范化 | ⭐⭐⭐⭐ | 低 | 中 |
| 9 | Lazy Initialization | ⭐⭐⭐⭐ | 低 | 高 |

---

## 1️⃣ Pipeline 架构模式 ⭐⭐⭐⭐⭐

**文件位置**: `pixelle_video/pipelines/base.py`

### 核心代码
```python
from abc import ABC, abstractmethod
from typing import Optional, Callable

class BasePipeline(ABC):
    """所有自定义 Pipeline 必须继承此基类"""
    
    def __init__(self, pixelle_video_core):
        self.core = pixelle_video_core
        self.llm = pixelle_video_core.llm
        self.tts = pixelle_video_core.tts
        self.media = pixelle_video_core.media
    
    @abstractmethod
    async def __call__(
        self,
        text: str,
        progress_callback: Optional[Callable[[ProgressEvent], None]] = None,
        **kwargs
    ) -> VideoGenerationResult:
        pass
```

### 使用示例
```python
# 内置 Pipeline
self.pipelines = {
    "standard": StandardPipeline(self),
    "custom": CustomPipeline(self),
    "asset_based": AssetBasedPipeline(self),
}

# 调用
result = await self.pipelines["standard"](text="...", n_scenes=5)
```

### 优点
- ✅ 解耦业务逻辑，每个 Pipeline 独立
- ✅ 支持多种生成模式
- ✅ 通过 `progress_callback` 实现进度追踪
- ✅ 统一接口，易于扩展

### 对比你的项目
- 你的 `book-generator` 模块可以用 Pipeline 重构
- TTS 生成、图片生成可以各自作为 Pipeline

---

## 2️⃣ 配置管理单例模式 ⭐⭐⭐⭐⭐

**文件位置**: `pixelle_video/config/manager.py`

### 核心代码
```python
class ConfigManager:
    _instance: Optional['ConfigManager'] = None
    
    def __new__(cls, config_path: str = "config.yaml"):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, config_path: str = "config.yaml"):
        if hasattr(self, '_initialized'):
            return
        self._initialized = True
    
    def update(self, updates: dict):
        """深度合并更新"""
        def deep_merge(base: dict, updates: dict) -> dict:
            for key, value in updates.items():
                if key in base and isinstance(base[key], dict):
                    deep_merge(base[key], value)
                else:
                    base[key] = value
            return base
```

### 优点
- ✅ 全局单例，避免重复加载
- ✅ 配置热重载支持
- ✅ 深度合并更新
- ✅ 配置验证机制

### 对比你的项目
- 你的 `.env` + `config.ts` 可参考此模式
- 需要统一配置管理入口

---

## 3️⃣ 任务管理异步模式 ⭐⭐⭐⭐⭐

**文件位置**: `api/tasks/manager.py`

### 核心代码
```python
class TaskManager:
    def __init__(self):
        self._tasks: Dict[str, Task] = {}
        self._task_futures: Dict[str, asyncio.Task] = {}
    
    async def execute_task(self, task_id: str, coro_func, *args, **kwargs):
        async def _execute():
            try:
                task.status = TaskStatus.RUNNING
                result = await coro_func(*args, **kwargs)
                task.status = TaskStatus.COMPLETED
                task.result = result
            except Exception as e:
                task.status = TaskStatus.FAILED
                task.error = str(e)
        
        future = asyncio.create_task(_execute())
        self._task_futures[task_id] = future
    
    def update_progress(self, task_id: str, current: int, total: int, message: str = ""):
        percentage = (current / total * 100) if total > 0 else 0
        task.progress = TaskProgress(current=current, total=total, percentage=percentage)
    
    def cancel_task(self, task_id: str) -> bool:
        """取消运行中的任务"""
        future = self._task_futures.get(task_id)
        if future and not future.done():
            future.cancel()
        task.status = TaskStatus.CANCELLED
        return True
```

### 优点
- ✅ 异步任务生命周期管理
- ✅ 自动清理过期任务
- ✅ 进度追踪 + 取消机制
- ✅ 状态持久化

### 对比你的项目
- 你的 Redis 队列可参考任务状态设计
- 可增加任务取消功能

---

## 4️⃣ Core 服务层聚合模式 ⭐⭐⭐⭐⭐

**文件位置**: `pixelle_video/service.py`

### 核心代码
```python
class PixelleVideoCore:
    async def initialize(self):
        # 核心服务
        self.llm = LLMService(self.config)
        self.tts = TTSService(self.config, core=self)
        self.media = MediaService(self.config, core=self)
        self.api_media = APIProviderMediaService(self.config, core=self)
        self.video = VideoService()
        
        # Pipeline 注册
        self.pipelines = {
            "standard": StandardPipeline(self),
            "custom": CustomPipeline(self),
        }
    
    # 每个服务都可以访问 core
    async def some_method(self):
        audio = await self.tts("hello")
        image = await self.media.generate(prompt="...")
```

### 优点
- ✅ 统一入口，封装所有能力
- ✅ 依赖注入（每个服务可访问 core）
- ✅ Pipeline 可插拔
- ✅ 懒加载支持

### 对比你的项目
- 你的 `server/src/services/` 可以整合到 Core
- TTS 服务、存储服务统一管理

---

## 5️⃣ LLM 结构化输出 ⭐⭐⭐⭐

**文件位置**: `pixelle_video/services/llm_service.py`

### 核心代码
```python
from pydantic import BaseModel
from typing import TypeVar, Type

T = TypeVar("T", bound=BaseModel)

async def __call__(
    self,
    prompt: str,
    response_type: Optional[Type[T]] = None,
    **kwargs
) -> Union[str, T]:
    
    if response_type is not None:
        # 自动生成 JSON Schema 指令
        schema = response_type.model_json_schema()
        json_instruction = f"""## JSON Output Required
```json
{json.dumps(schema, indent=2)}
```
Only output JSON, no other text."""
        
        response = await client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": f"{prompt}\n\n{json_instruction}"}],
        )
        
        return self._parse_response_as_model(content, response_type)
```

### 使用示例
```python
from pydantic import BaseModel

class BookOutline(BaseModel):
    title: str
    chapters: List[str]
    total_words: int

outline = await llm(
    prompt="为《Atomic Habits》生成大纲",
    response_type=BookOutline
)
print(outline.title)  # 强类型访问
```

### 优点
- ✅ Pydantic 强类型输出
- ✅ 自动生成 JSON Schema 指令
- ✅ 兼容所有 OpenAI 兼容 API
- ✅ 多层容错解析

---

## 6️⃣ Storyboard 数据模型 ⭐⭐⭐⭐

**文件位置**: `pixelle_video/models/storyboard.py`

### 核心代码
```python
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional

@dataclass
class StoryboardFrame:
    index: int
    narration: str
    image_prompt: str
    audio_path: Optional[str] = None
    media_type: Optional[str] = None
    video_segment_path: Optional[str] = None
    duration: float = 0.0
    created_at: Optional[datetime] = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()

@dataclass
class Storyboard:
    title: str
    config: StoryboardConfig
    frames: List[StoryboardFrame] = field(default_factory=list)
    final_video_path: Optional[str] = None
    
    @property
    def progress(self) -> float:
        """自动计算进度"""
        if not self.frames:
            return 0.0
        completed = sum(1 for f in self.frames if f.video_segment_path)
        return completed / len(self.frames)
    
    @property
    def is_completed(self) -> bool:
        return all(f.video_segment_path for f in self.frames)
```

### 优点
- ✅ `@dataclass` 简洁定义
- ✅ 嵌套层级清晰
- ✅ `@property` 计算属性
- ✅ 类型安全

---

## 7️⃣ Prompt 模板分离 ⭐⭐⭐⭐

**文件位置**: `pixelle_video/prompts/*.py`

### 示例：标题生成
```python
# prompts/title_generation.py

TITLE_GENERATION_PROMPT = """Please generate a short, attractive title.

Requirements:
1. **Language Consistency (CRITICAL)**: The title MUST be in the same language as input
2. **Character Limit (CRITICAL)**: MUST NOT exceed {max_length} characters
3. **Core Message (CRITICAL)**: MUST capture the MAIN POINT

Title:"""

def build_title_generation_prompt(content: str, max_length: int = 15) -> str:
    return TITLE_GENERATION_PROMPT.format(
        content=content[:500],  # 限制长度
        max_length=max_length
    )
```

### 目录结构
```
prompts/
├── title_generation.py
├── content_narration.py
├── image_generation.py
├── topic_narration.py
└── video_generation.py
```

### 优点
- ✅ Prompt 代码与业务逻辑分离
- ✅ 参数化模板，易于复用
- ✅ 注释详细，便于维护
- ✅ 支持多语言

---

## 8️⃣ API Schema 规范化 ⭐⭐⭐⭐

**文件位置**: `api/schemas/video.py`

### 核心代码
```python
from pydantic import BaseModel, Field
from typing import Optional, Literal, Dict, Any

class VideoGenerateRequest(BaseModel):
    text: str = Field(..., description="Source text for video generation")
    
    mode: Literal["generate", "fixed"] = Field(
        "generate",
        description="Processing mode"
    )
    
    n_scenes: Optional[int] = Field(
        5, 
        ge=1, le=20,  # 范围约束
        description="Number of scenes"
    )
    
    template_params: Optional[Dict[str, Any]] = Field(
        None,
        description="Custom template parameters"
    )
    
    class Config:
        json_schema_extra = {
            "example": {
                "text": "Atomic Habits teaches us...",
                "n_scenes": 5,
            }
        }
```

### 优点
- ✅ Pydantic 验证 + 自动文档
- ✅ 范围约束（`ge`, `le`）
- ✅ 示例 JSON 可直接用于测试
- ✅ 类型提示完善

---

## 9️⃣ Lazy Initialization 懒加载 ⭐⭐⭐⭐

**文件位置**: `pixelle_video/service.py`

### 核心代码
```python
async def _get_or_create_comfykit(self) -> ComfyKit:
    current_config = self._get_comfykit_config()
    current_hash = self._compute_comfykit_config_hash(current_config)
    
    # 配置变更检测
    if self._comfykit is None or self._comfykit_config_hash != current_hash:
        if self._comfykit is not None:
            await self._comfykit.close()  # 清理旧实例
        
        self._comfykit = ComfyKit(**current_config)
        self._comfykit_config_hash = current_hash
    
    return self._comfykit
```

### 优点
- ✅ 首次使用时才初始化
- ✅ 配置变更自动重建
- ✅ 内存优化
- ✅ 支持热重载

---

## 📊 借鉴优先级建议

### 🔴 高优先级（可直接借鉴）

| 模式 | 理由 | 借鉴方式 |
|------|------|---------|
| **Pipeline 架构** | 解耦能力强，适合复杂业务流程 | 重构 `book-generator` 模块 |
| **Storyboard 模型** | 简单实用，可追踪复杂任务状态 | 新建视频生成模块 |
| **Prompt 模板分离** | 低侵入，易实施 | 新建 `prompts/` 目录 |
| **Lazy Loading** | 性能优化，减少启动时间 | 重构配置加载 |

### 🟡 中优先级（有条件借鉴）

| 模式 | 理由 | 借鉴条件 |
|------|------|---------|
| **Core 服务聚合** | 需要重构较大范围 | 先小范围试点 |
| **配置单例** | 需统一配置入口 | 评估现有配置复杂度 |

### 🟢 低优先级（了解即可）

| 模式 | 理由 |
|------|------|
| LLM 结构化输出 | 需要 Pydantic 基础 |
| API Schema | 项目已有类似方案 |

---

## 📝 实施建议

### Phase 1：低成本高收益（1-2天）
1. 创建 `server/src/prompts/` 目录，分离 Prompt 模板
2. 引入 `Storyboard` 数据模型用于视频生成状态追踪

### Phase 2：中等投入（3-5天）
3. 实现 Pipeline 基类，重构 TTS 生成流程
4. 实现配置热重载

### Phase 3：长期优化（1-2周）
5. Core 服务层聚合
6. 任务管理增强（取消、进度）

---

## 🔗 相关文件索引

| 设计模式 | 源文件 | 行数 |
|---------|--------|------|
| Pipeline | `pixelle_video/pipelines/base.py` | 117 |
| 配置单例 | `pixelle_video/config/manager.py` | 182 |
| 任务管理 | `api/tasks/manager.py` | 270 |
| Core 聚合 | `pixelle_video/service.py` | 316 |
| LLM 服务 | `pixelle_video/services/llm_service.py` | 340 |
| 数据模型 | `pixelle_video/models/storyboard.py` | 144 |
| Prompt 模板 | `pixelle_video/prompts/title_generation.py` | 85 |
| API Schema | `api/schemas/video.py` | 117 |