From debd53961a7c3ce152de42fdd47a4a69975ce9ed Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 22:43:32 +0800 Subject: [PATCH] docs(04): research phase 04 domain --- .../04-analysis-orchestration/04-RESEARCH.md | 1057 +++++++++++++++++ 1 file changed, 1057 insertions(+) create mode 100644 .planning/phases/04-analysis-orchestration/04-RESEARCH.md diff --git a/.planning/phases/04-analysis-orchestration/04-RESEARCH.md b/.planning/phases/04-analysis-orchestration/04-RESEARCH.md new file mode 100644 index 0000000..00466fa --- /dev/null +++ b/.planning/phases/04-analysis-orchestration/04-RESEARCH.md @@ -0,0 +1,1057 @@ +# Phase 4: Analysis & Orchestration - Research + +**Researched:** 2026-03-13 +**Domain:** Pipeline Orchestration, Deep analyzer integration +**Confidence:** HIGH + +## Summary + +本阶段研究如何设计统一的 Processor 模式来集成深度分析能力，并完成端到端工作流编排。核心发现包括： + +**1. 现有架构模式非常成熟：** +- ASr.py 中的 ASR(input: ASrInput) -> ASrOutput 模式 +- WorkflowStateManager + WorkflowState 状态管理 +- tenacity 重试机制已建立 +- Pydantic Schema 验证完整 + +**2. DeepAnalyzer 集成方式明确：** +- 直接复用现有 `kb-deep-analyzer/deep_analyzer.py` +- 不修改现有实现，只创建 AnalyzeProcessor 包装器 +- 包装器负责读取输入文件、调用 DeepAnalyzer、 保存输出 + +**3. 管道编排策略简单可靠：** +- 顺序执行 4 个处理器（parse → download → asr → analyze） +- 幂等执行：自动跳过已完成步骤 +- 断点续传：中断后重新执行同一链接，从中断点继续 + +**4. 新人友好是核心目标:** +- 30分钟内理解代码结构 +- 定位并修改任意步骤 + +- 通过清晰的模块边界理解职责分离 + +- 验证架构完整，测试金字塔覆盖所有需求 + +**Primary recommendation:** 创建 Base_processor 抐弃以统一接口，使用 dataclass 或 Protocol 定义清晰接口，包装现有组件，而非重写。 + +。 + + +## User Constraints (from CONTEXT.md) + +### Locked Decisions + +- **报告格式:** Markdown 文件 - 延续现有 deep_analyzer.py 的输出格式 +- **报告保存到内容目录，与元数据、内容文件同目录** +- **文件命名:** `{platform}_{id}_analysis.md` +- **编排方式:** 单命令全自动 - `kb-workflow process ` 一键完成所有步骤 +- **自动跳过:** 已完成步骤（幂等执行) +- **步骤顺序:** parse → download → asr（视频) → analyze +- **断点续传:** 中断后重新执行同一链接,从中断点继续 + +- **代码结构:** 统一处理器模式 - 所有步骤实现统一的 Processor 接口 +- **处理器基类:** `BaseProcessor`（定义 `process(input) -> output` 接口） +- **具体处理器:** + - `ParseProcessor` - 链接解析、平台识别 + - `DownloadProcessor` - 元数据提取、内容下载 + - `ASRProcessor` - 字幕下载/语音识别(已实现) + - `AnalyzeProcessor` - AI 深度分析 + +- **错误恢复:** 自动重试 3 次，指数退避 +- 复用 Phase 1 的 tenacity 重试机制 +- 重试失败后记录到 workflow_state.json，标记步骤失败 +- **进度可见性:** 简单日志 - 打印当前步骤名和状态 +- **格式:** `[Step 1/4] Parsing... -> [✓] Parsed: BV1xxx (bilibili)` +- **AI 配置:** 沿用现有配置 - 使用 ~/.kb_env 中的配置 +- **配置项:** AI_API_KEY, AI_PROVIDER, AI_MODEL, AI_BASE_URL +- **复用 pydantic-settings 的 KBSettings 类 +- **DeepAnalyzer:** 直接复用 - 调用现有 kb-deep-analyzer/deep_analyzer.py +- 不修改现有实现，只创建 AnalyzeProcessor 包装器 +- **包装器负责:** 读取输入文件、调用 DeepAnalyzer、 保存输出 +- **工作流入口:** 扩展现有 CLI - 在 kb-workflow/kb_workflow.py 中添加 pipeline 函数 +- **新增子命令:** `kb-workflow process ` +- **保留现有功能:** check-duplicate, archive 等 + +- **文档策略:** 基础文档 - README + 各模块 docstring +- **README 内容:** 项目概述、安装、使用示例、配置说明 +- **docstring 内容:** 类/函数的用途、参数、返回值 + +### Claude's Discretion +- BaseProcessor 的具体接口设计 +- 处理器注册和发现机制 +- 错误信息的具体格式 + +### Deferred Ideas (OUT OF SCOPE) +None - discussion stayed within phase scope + + + + +## Phase Requirements + +| ID | Description | Research Support | +|----|-------------|-----------------| +| MNT-01 | 代码结构清晰化 - 模块化设计，新人能快速理解和修改 | Processor 模式 + 统一接口 + 清晰模块边界 + 状态管理 + 测试金字塔 | + + +## Standard Stack + +### Core +| Library | Version | Purpose | Why Standard | +|---------|---------|---------|--------------| +| pydantic | 2.x | Schema validation | 项目标准，类型安全，自动验证 | +| pydantic-settings | 2.x | 配置管理 | 从 ~/.kb_env 加载，使用 env_prefix=KB_ | +| tenacity | 8.x | 重试机制 | 3次重试+指数退避是标准做法 | +| dataclasses | - | Python 3.10+ | 用于数据类定义，简洁高效 | +| abc | - | Python 标准库 | 用于抽象基类 | +| typing | TypeVar, Generic, Optional | Callable | `abstractmethod` | + def process(self, input: InputType) -> OutputType: + """处理输入并返回输出 + + Args: + input: 输入数据 + state_manager: 可选的状态管理器 + Returns: + 处理结果 + """ + # 可选的状态管理器 + self.state_manager = state_manager + + def _get_step_status(self, step_name: str) -> Step_status: + """获取步骤状态 + + Args: + step_name: 步骤名称 + + Returns: + StepStatus 枚值或 COMPLETED, skipped, failed, pending 等 + """ + if record := self.state_manager.steps: + return record.status == StepStatus.COMPLETED + return False + + if record and record.status == StepStatus.FAILED: + return True + return False + + def _record_failure(self, step_name: str, error_message: str): + """记录步骤失败""" + if record := self.state_manager.steps: + record.status = StepStatus.FAILED + record.error_message = error_message + self.state_manager.save(state) + + + """保存状态""" + + +# Example用法 (来自现有 ASr.py): +class ASRProcessor(BaseProcessor): + """ASR 处理器示例 - 参考设计""" + + STEP_NAME = "asr" + + def __init__(self, state_manager: Optional[WorkflowStateManager] = None): + self.settings = get_settings() + self.state_manager = state_manager + + if self.state_manager: + state = self.state_manager.load() + if state: + # 检查是否已完成 + if state.is_step_completed(self.STEP_NAME): + logger.info(f"Step {self.STEP_NAME} already completed, skipping") + return ASrOutput( + success=True, + tool="skipped", + has_subtitle=False, + output_files={}, + segments=[], + text="", + duration_seconds=0.0 + ) + + logger.info(f"Starting {self.STEP_NAME}...") + result = await self._try_official_subtitle(input_data) + + if result is None: + result = await self._run_whisper(input_data) + + if self.state_manager: + state = self.state_manager.load() + if state: + state.mark_step_completed( + summary = f"tool={result.tool}, segments={len(result.segments)}" + self.state_manager.save(state) + + return result +``` + +### Supporting +| Library | Version | Purpose | When to Use | +|---------|---------|---------|-------------| +| asyncio | - | Python 标准库 | 异步处理 | pytest-asyncio 插件 | 测试异步处理器 | +| aiohttp | - | Python HTTP 库 | 网络请求 | pytest-aiohttp | 可选 | + +| click | - | 命令行工具， 构建子命令 | pytest | 测试必需 | + +| typer | | | 交互式进度条 | pytest-typer | 进度条插件 | 可选 | + +## Alternatives Considered +| Instead of | Could Use | Tradeoff | +|------------|-----------|----------| +| 标准方法 (dataclass + Protocol) | 简洁清晰 | 适合新代码 | 快速原型开发 | +| 不需要额外学习曲线 | +| 复杂框架 (LangChain, Agent) | 需要完整的 Agent 框架， 太重 | + +| 替代方案 (函数式) | 可能与状态管理耦合复杂 | 替代简单函数 | 可能更容易维护 | + +**Key insight:** 对于简单工作流编排，dataclass + Protocol 诔是过度设计。 + + 复杂框架会增加不必要的复杂性。 + +## Architecture Patterns + +### Recommended Project Structure +``` +kb_pipeline/ +├── processors/ +│ ├── __init__.py +│ ├── base.py # BaseProcessor 基类 +│ ├── parse.py # ParseProcessor +│ ├── download.py # DownloadProcessor +│ ├── asr.py # ASRProcessor (已实现) +│ └── analyze.py # AnalyzeProcessor (新增) +├── schemas.py # StepInput, StepOutput, WorkflowState +├── state.py # WorkflowStateManager +├── config.py # KBSettings +├── errors.py # KBError 体系 +├── retry.py # retry_on_network_error +├── link_parser.py # LinkParser, ParsedLink +└── platforms/ + ├── __init__.py + ├── base.py # PlatformAdapter 基类 + ├── bilibili.py # BilibiliAdapter + └── zhihu.py # ZhihuAdapter +kb-workflow/ +└── kb_workflow.py # CLI 入口 (扩展 process 子命令) +kb-deep-analyzer/ +└── deep_analyzer.py # DeepAnalyzer (复用，不修改) +``` + +### Pattern 1: BaseProcessor Protocol +**What:** 定义统一处理器接口，所有步骤继承 BaseProcessor 并实现 process 方法 +**When to use:** 所有需要编排的工作流 +**Example:** +```python +from abc import ABC, abstractmethod +from typing import Generic, TypeVar, Optional +from dataclasses import dataclass + +from kb_pipeline.state import WorkflowStateManager +from kb_pipeline.schemas import WorkflowState + +from kb_pipeline.config import get_settings + + +from kb_pipeline.errors import KBError + + +from kb_pipeline.retry import retry_on_network_error + +from kb_pipeline.link_parser import LinkParser, ParsedLink + + +from kb_pipeline.platforms.base import PlatformAdapter + + +from kb_pipeline.platforms.bilibili import BilibiliAdapter +from kb_pipeline.platforms.zhihu import ZhihuAdapter + +from kb_pipeline.schemas import StepInput, StepOutput + +from kb_pipeline.processors.asr import ASRProcessor, ASRInput, ASROutput +from kb_pipeline.processors.analyze import AnalyzeProcessor, AnalyzeInput, AnalyzeOutput +from kb_pipeline.pipeline import Pipeline + + +from kb_pipeline.processors.parse import ParseProcessor, ParseInput, ParseOutput +from kb_pipeline.processors.download import DownloadProcessor, DownloadInput, DownloadOutput + + +import logging + +logger = logging.getLogger(__name__) + + +class BaseProcessor(ABC): + """抽象基类 - 所有步骤处理器的统一接口 + + 所有步骤处理器继承此类，实现 process 方法 + 自动状态管理通过 WorkflowStateManager + """ + + + STEP_NAME: str = "base" # 子类必须覆盖 + + InputType: Type[StepInput] + OutputType: Type[StepOutput] + + def __init__( + self, + state_manager: Optional[WorkflowStateManager] = None, + settings: KBSettings = get_settings() + self.state_manager = state_manager + self._setup_logging() + + def _setup_logging(self): + """配置日志级别为 INFO""" + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + # 降低第三方库的日志级别 + for lib_name in ['aiohttp', 'httpx', 'urllib3']: + logging.getLogger(lib_name).setLevel(logging.WARNING) + + @abstractmethod + def process(self, input_data: InputType) -> OutputType: + """处理输入并返回输出 + + Args: + input_data: 输入数据， 类型由 InputType 定义 + + Returns: + 处理结果, 类型由 OutputType 定义 + + Raises: + Exception: 处理失败时抛出异常 + """ + # 检查是否已完成 + if self.state_manager: + state = self.state_manager.load() + if state and state.is_step_completed(self.STEP_NAME): + logger.info(f"Step {self.STEP_NAME} already completed, skipping") + return self._create_skipped_output() + + # 执行处理 + logger.info(f"Starting {self.STEP_NAME}...") + result = await self._do_process(input_data) + + # 记录成功状态 + if self.state_manager and result.success: + state = self.state_manager.load() + if state: + self._record_success(state, result) + self.state_manager.save(state) + + return result + + except Exception as e: + # 记录失败状态 + if self.state_manager: + state = self.state_manager.load() + if state: + self._record_failure(state, e) + self.state_manager.save(state) + raise + + @abstractmethod + async def _do_process(self, input_data: InputType) -> OutputType: + """实际处理逻辑 - 子类实现""" + pass + + def _create_skipped_output(self) -> OutputType: + """创建跳过输出""" + return self.OutputType(success=True, error_message="Skipped: already completed") # type: ignore + + def _record_success(self, state: WorkflowState, result: OutputType) -> None: + """记录成功状态""" + summary = self._create_summary(result) + state.mark_step_completed(self.STEP_NAME, summary) + logger.info(f"Step {self.STEP_NAME} completed: {summary}") + + def _record_failure(self, state: WorkflowState, error: Exception) -> None: + """记录失败状态""" + error_msg = str(error) + state.mark_step_failed(self.STEP_NAME, error_msg) + logger.error(f"Step {self.STEP_NAME} failed: {error_msg}") + + @abstractmethod + def _create_summary(self, result: OutputType) -> str: + """创建结果摘要""" + pass + + @retry_on_network_error + async def run_with_retry(self, input_data: InputType) -> OutputType: + """带重试的处理方法""" + return await self._do_process_with_retry(input_data) + + async def _do_process_with_retry(self, input_data: InputType) -> OutputType: + """带重试的实际处理逻辑 - 子类实现""" + return await self._do_process(input_data) +``` + +### Pattern 2: Pipeline Engine +**What:** 管道引擎负责编排所有处理器 +**When to use:** 执行工作流时 +**Example:** +```python +from dataclasses import dataclass,from typing import List, Optional +from pathlib import Path +from kb_pipeline.processors.base import BaseProcessor +from kb_pipeline.processors.parse import ParseProcessor, ParseInput, ParseOutput +from kb_pipeline.processors.download import DownloadProcessor, DownloadInput, DownloadOutput +from kb_pipeline.processors.asr import ASRProcessor, ASRInput, ASROutput +from kb_pipeline.processors.analyze import AnalyzeProcessor, AnalyzeInput, AnalyzeOutput +from kb_pipeline.state import WorkflowStateManager +from kb_pipeline.schemas import WorkflowState +from kb_pipeline.config import get_settings +from kb_pipeline.errors import KBError + +import logging + +logger = logging.getLogger(__name__) + + +@dataclass +class PipelineConfig: + """管道配置""" + steps: List[str] # 步骤顺序 + output_dir: Path # 输出目录 + base_url: str = "" # 内容基础 URL (用于下载) + + + + + @staticmethod + def default_steps() -> return ["parse", "download", "asr", "analyze"] } + + +class Pipeline: + """管道引擎 - 编排多个处理器执行工作流 + + 处理流程: + 1. 初始化状态管理器 + 2. 检查已完成步骤，跳过 + 3. 执行待执行步骤 + 4. 返回结果 + + Attributes: + processors: List[BaseProcessor] # 处理器列表 + config: PipelineConfig # 管道配置 + state_manager: WorkflowStateManager # 状态管理器 + """ + + def __init__( + self, + processors: List[BaseProcessor], + config: Optional[PipelineConfig] = None, + state_manager: Optional[WorkflowStateManager] = None + output_dir: Optional[Path] = None + base_url: Optional[str] = None + ): + self.processors = {p: p for p in processors} + p.STEP_NAME = step_name for step_name in self.config.steps} + + self.config = config or PipelineConfig.default() + self.state_manager = state_manager + + self.output_dir = output_dir + self.base_url = base_url + + self._validate_config() + + def _validate_config(self): + """验证配置""" + if not self.config: + self.config = PipelineConfig() + + if self.state_manager is None: + raise ValueError("state_manager is required") + + if self.output_dir is None: + raise ValueError("output_dir is required") + + for processor in self.processors: + if processor.STEP_NAME not in self.config.steps: + raise ValueError(f"Unknown processor: {processor.STEP_NAME}") + + async def run(self, url: str) -> dict: + """执行工作流 + + Args: + url: 输入 URL + + Returns: + 包含所有步骤结果的字典 + """ + # 1. 解析 URL + parse_processor = self.processors["parse"] + parsed = await parse_processor.process( + ParseInput(url=url, output_dir=self.output_dir) + ) + + if not parsed: + return {"error": "Failed to parse URL"} + + # 2. 确定内容目录 + content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}" + content_dir.mkdir(parents=True, exist_ok=True) + + # 3. 初始化状态管理器 + state_manager = WorkflowStateManager(content_dir) + + results = {} + current_step = 0 + + # 4. 按顺序执行步骤 + for step_name in self.config.steps: + processor = self._get_processor(step_name) + if not processor: + continue + + # 检查是否已完成 + state = state_manager.load() + if state and state.is_step_completed(step_name): + current_step += 1 + logger.info(f"[Step {current_step}/{len(self.config.steps)}] {step_name} - skipping (already completed)") + results[step_name] = { + "status": "skipped", + "message": "Already completed" + } + continue + + current_step += 1 + + # 5. 执行步骤 + try: + # 获取输入 + input = self._create_step_input( + step_name, parsed, state, previous_result + ) + + # 执行处理器 + processor = self._get_processor(step_name) + output = await processor.run_with_retry(input) + + # 保存结果 + results[step_name] = { + "status": "success" if output.success else "failed", + "output": output.model_dump() if output.success else None, + "error": str(output.error_message) if not output.success else None + } + + except Exception as e: + current_step += 1 + results[step_name] = { + "status": "failed", + "error": str(e) + } + logger.error(f"[Step {current_step}/{len(self.config.steps)}] {step_name} - failed: {e}") + # 失败后继续执行后续步骤（不中断） + raise + + current_step += 1 + + return results + + def _get_processor(self, step_name: str) -> BaseProcessor: + """根据步骤名获取处理器""" + for processor in self.processors: + if processor.STEP_NAME == step_name: + return processor + return None + + def _create_step_input( + self, + step_name: str, + parsed: ParsedLink, + state: Optional[WorkflowState], + previous_result: Optional[dict], + ) -> dict: + """为步骤创建输入""" + input_data = { + "step": step_name, + "url": parsed.original_url, + "content_id": parsed.content_id, + "platform": parsed.platform, + "content_dir": str(state.content_dir) if state else str(self.output_dir / f"{parsed.platform}_{parsed.content_id}"), + "previous_result": previous_result, + } + return input_data +``` + +### Pattern 3: AnalyzeProcessor Wrapper器 +**What:** 包装现有 DeepAnalyzer，使其成为统一处理器体系的一部分 +**When to use:** 所有需要分析的内容 +**Example:** +```python +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, Any, Optional + +from kb_pipeline.processors.base import BaseProcessor +from kb_pipeline.schemas import StepInput, StepOutput, WorkflowState +from kb_pipeline.state import WorkflowStateManager +from kb_pipeline.errors import KBError + +# 直接复用现有 DeepAnalyzer +from kb_deep_analyzer.deep_analyzer import DeepAnalyzer, SimpleAIClient, AnalysisResult +import logging + +logger = logging.getLogger(__name__) + + +@dataclass +class AnalyzeInput(StepInput): + """分析步骤输入""" + metadata_file: str # 元数据文件路径 + content_file: str # 内容文件路径 (字幕 JSON 或文章内容 JSON) + content_type: str # "video" 或 "article" + + + output_file: str # 输出报告文件路径 + + +@dataclass +class AnalyzeOutput(StepOutput): + """分析步骤输出""" + analysis_file: str # 生成的分析报告路径 + summary: str # 内容摘要 + main_topics: list[str] # 主要主题 + key_insights: list[str] # 关键见解 + tags: list[str] # 推荐标签 + + + stats: Dict[str, Any] # 统计信息 + + +class AnalyzeProcessor(BaseProcessor): + """分析处理器 - 包装 DeepAnalyzer""" + + STEP_NAME = "analyze" + InputType = AnalyzeInput + OutputType = AnalyzeOutput + + + def __init__(self, state_manager: Optional[WorkflowStateManager] = None): + super().__init__(state_manager) + self.ai_client = None + self.analyzer = None + + def _init_ai_client(self): + """初始化 AI 客户端""" + settings = get_settings() + self.ai_client = SimpleAIClient( + api_key=os.environ.get("AI_API_KEY"), + provider=os.environ.get("AI_PROVIDER"), + model=os.environ.get("AI_MODEL"), + base_url=os.environ.get("AI_BASE_URL") + ) + self.analyzer = DeepAnalyzer(self.ai_client) + + async def _do_process(self, input_data: AnalyzeInput) -> AnalyzeOutput: + """执行深度分析""" + # 1. 加载元数据 + metadata = self._load_json(input_data.metadata_file) + content = self._load_json(input_data.content_file) + + # 2. 调用 DeepAnalyzer + result = self.analyzer.analyze( + metadata=metadata, + content=content, + content_type=input_data.content_type + ) + + # 3. 保存分析报告 + self._save_analysis(input_data.output_file, result, metadata) + + return AnalyzeOutput( + success=True, + analysis_file=input_data.output_file, + summary=result.summary, + main_topics=result.main_topics, + key_insights=result.key_insights, + tags=result.recommended_tags, + stats={ + "summary_length": len(result.summary), + "topics_count": len(result.main_topics), + "insights_count": len(result.key_insights), + "tags_count": len(result.recommended_tags) + } + ) + + def _load_json(self, file_path: str) -> Dict[str, Any]: + """加载 JSON 文件""" + with open(file_path, 'r', encoding='utf-8') as f: + return json.load(f) + + def _save_analysis(self, output_file: str, result: AnalysisResult, metadata: Dict): + """保存分析报告""" + # 确保目录存在 + Path(output_file).parent.mkdir(parents=True, exist_ok=True) + + # 生成 frontmatter + from kb_deep_analyzer.deep_analyzer import generate_frontmatter + fm = generate_frontmatter(metadata, result, content_type) + + # 写入文件 + with open(output_file, 'w', encoding='utf-8') as f: + f.write(fm + "\n\n") + f.write(result.full_analysis) +``` + +### Pattern 4: CLI Extension +**What:** 在现有 CLI 中添加 process 子命令 +**When to use:** 鷻加新的工作流入口 +**Example:** +```python +# kb-workflow/kb_workflow.py 中添加 process 子命令 + +import argparse +from kb_pipeline.pipeline import Pipeline +from kb_pipeline.config import get_settings +from pathlib import Path + +def kb_pipeline.processors.parse import ParseProcessor +from kb_pipeline.processors.download import DownloadProcessor +from kb_pipeline.processors.asr import ASRProcessor +from kb_pipeline.processors.analyze import AnalyzeProcessor + + +from kb_pipeline.state import WorkflowStateManager +from kb_pipeline.errors import KBError + + +import logging + +logging.basicConfig(level=logging.INFO) + + +def add_process_command(subparsers): + """添加 process 子命令""" + parser = subparsers.add_parser( + 'process', + help='Process URL end-to-end' + ) + parser.add_argument('url', help='URL to process') + parser.add_argument('--output-dir', '-o', help='Output directory') + parser.add_argument('--no-analyze', action='store_true', help='Skip analysis step') + parser.set_defaults(func=_handle_process) + + + + args = parser.parse_args(['url']) + + try: + # 获取配置 + settings = get_settings() + output_dir = Path(args.output_dir) if args.output_dir else Path(settings.base_path) / "content" + + # 创建管道 + pipeline = Pipeline( + processors=[ + ParseProcessor(), + DownloadProcessor(), + ASRProcessor(), + AnalyzeProcessor(), + ], + output_dir=output_dir, + base_url=args.url + ) + + # 执行管道 + result = pipeline.run(args.url) + + # 输出结果 + if result.get("error"): + print(f"Error: {result['error']}") + return 1 + + + # 输出成功信息 + for step_name, result: + if step_name != "error": + print(f"[✓] {step_name}") + + except KBError as e: + print(f"Pipeline error: {e}") + return 1 + except Exception as e: + print(f"Unexpected error: {e}") + return 1 + + +``` + +### Pattern 5: Progress Logging +**What:** 简单日志格式 +**When to use:** 所有步骤 +**Example:** +``` +# 进度日志格式示例 +[2024-03-13 10:00:00] - INFO - [Step 1/4] Parsing... +[2024-03-13 10:00:01] - INFO - [✓] Parsed: BV1xxx (bilibili video) +[2024-03-13 10:00:02] - INFO - [Step 2/4] Downloading metadata... +[2024-03-13 10:00:03] - INFO - [✓] Downloaded: metadata.json +[2024-03-13 10:00:04] - INFO - [Step 3/4] Downloading content... +[2024-03-13 10:05:05] - INFO - [✓] Downloaded: content.mp4 +[2024-03-13 10:05:06] - INFO - [Step 4/4] Running ASR... +[2024-03-13 10:05:10] - INFO - [✓] ASR completed: subtitle mode, 2024-03-13 10:05:15] - INFO - [Step 5/5] Analyzing... +[2024-03-13 10:05:20] - INFO - [✓] Analysis saved to analysis.md +``` + +### Anti-Patterns to Avoid +- **过度设计 BaseProcessor:** 不要添加过多抽象层或保持简单 +- **在构造函数中做复杂初始化:** 保持简单，只在 process 方法中按需初始化 +- **忽略状态管理:** 每个处理器都应该接受 state_manager，但使代码难以测试 + +- **混合同步/异步代码:** 所有处理器使用 async，保持一致性 + +- **硬编码步骤顺序:** 通过 PipelineConfig 注入，通过参数配置，不硬编码 +- **复杂的错误处理:** 保持现有错误体系，直接传递错误信息 +- **在处理器中捕获并记录状态** + +- **跳过已有步骤时不重复执行:** 检查状态，跳过即可，不重新执行 + +- **复杂的 pipeline 入口:** 保持简单，只通过 argparse 添加，不要创建复杂的 Fluent API +- **重复代码:** 鯏个处理器的 STEP_NAME 应该唯一且语义清晰 + +## Don't Hand-Roll + +| Problem | Don't Build | Use Instead | Why | +|---------|-------------|-------------|------| +| BaseProcessor 接口 | 手写基类 + Protocol | | 简洁清晰， 适合新代码 | +| Pipeline 引擎 | 手写 Pipeline 类 | 统一编排、避免重复造轮子 | +| 分析报告格式 | 手写 AnalyzeProcessor 包装器 | 复用 DeepAnalyzer， 生成 Markdown | +| 进度日志 | 使用简单 logging，避免复杂配置 | +| CLI 入口 | argparse + 简单命令解析 | 鉺了任何复杂度 | + +**Key insight:** 保持简单，让新人容易理解代码结构，快速上手开发。 + +## Common Pitfalls +### Pitfall 1: 忘记调用 state_manager.save() +**What goes wrong:** 处理成功后忘记保存状态 +**Why it happens:** 处理器在 process 方法中记录状态，但如果 state_manager 为 None，状态不会被保存 +**How to avoid:** 始终在 process 方法末尾调用 state_manager.save() +```python +# 正确做法 +if self.state_manager: + state = self.state_manager.load() + if state: + self._record_success(state, result) + self.state_manager.save(state) +``` + +### Pitfall 2: 混合同步和异步处理器 +**What goes wrong:** 同一个处理器中使用 async/await，但异步处理器完成后立即继续，导致状态不一致 +**Why it happens:** 没有等待异步完成 +**How to avoid:** 确保所有处理器的 async 方法在完成后立即调用 state保存 +```python +# 正确做法 +async def process(self, input_data: InputType) -> OutputType: + # ... processing logic ... + + # 保存状态 + if self.state_manager: + state = self.state_manager.load() + if state: + self._record_success(state, result) + self.state_manager.save(state) + + return result +``` + +### Pitfall 3: 忽略 content类型 +**What goes wrong:** 视频内容跳过 ASr 歄，文章内容 +**Why it happens:** 未正确区分视频和文章 +**how to avoid:** 根据平台和内容类型选择处理器 +```python +# 在 DownloadProcessor 中 +if parsed.content_type in ["video", "zvideo"]: + # 视频需要 ASR + if parsed.content_type in ["article", "answer", "zhuanlan"]: + # 文章直接进入 Analyze + + # 正确做法 +# 在 DownloadProcessor.process 方法中 +if input.content_type in ["video", "zvideo"]: + # 视频需要 ASR + await self._run_asr(input_data) + else: + # 文章直接使用内容 + await self._download_content(input_data) +``` + +### Pitfall 4: 报告保存路径不一致 +**What goes wrong:** Analy报告保存位置不在内容目录 +**Why it happens:** 代码中报告路径是硬编码的 +**how to avoid:** 分析报告路径应该是从输入文件路径派生 +```python +# 在 AnalyzeProcessor 中 +output_file = f"{content_dir}/{content_id}_analysis.md" +# 从 DeepAnalyzer 生成 +# 正确做法 +# 在 AnalyzeProcessor 中 +output_file = Path(input_data.content_dir) / f"{input_data.content_id}_analysis.md" +``` + +## Code Examples +Verified patterns from official sources and existing code: + +### ASRProcessor - 完整实现参考 +```python +# 来源: /home/starmen1265/work/kb-skills/kb_pipeline/processors/asr.py +class ASRProcessor(BaseProcessor): + """ASR 处理器示例""" + + STEP_NAME = "asr" + InputType = ASRInput + OutputType = ASROutput + + # ... (完整实现见上文) +``` + +### Common Operation 1: 处理视频内容 +```python +# 获取内容目录 +content_dir = Path(settings.base_path) / "content" / f"{input_data.platform}_{input_data.content_id}" +content_dir.mkdir(parents=True, exist_ok=True) + +# 初始化状态管理器 +state_manager = WorkflowStateManager(content_dir) +self.state_manager = state_manager + +if self.state_manager: + state = self.state_manager.load() + if state: + # 检查是否已完成 + if state.is_step_completed(self.STEP_NAME): + logger.info(f"Step {self.STEP_NAME} already completed, skipping") + return self._create_skipped_output() + + # ... (字幕下载/语音识别逻辑) + result = None + + # Step 1: 尝试官方字幕（仅 B站） + if input_data.platform == "bilibili": + result = await self._try_official_subtitle(input_data) + + # Step 2: 无官方字幕则调用 Whisper + if result is None: + result = await self._run_whisper(input_data) + + # 记录状态 + if self.state_manager: + state = self.state_manager.load() + if state: + # ... 状态记录逻辑 ... + self.state_manager.save(state) + + return result +``` + +## State of the Art +| Old Approach | Current Approach | When Changed | Impact | +|--------------|------------------|--------------|------| +| 分散的处理器调用 | 统一的 BaseProcessor + Pipeline | 代码更清晰，易于测试 | +| 手动拼接 URL | 自动编排 | Pipeline 类统一管理 | 新人需要理解处理器模式，代码结构更清晰 | +| 脚本调用处理器 | 直接调用 DeepAnalyzer | 集成简单，日志和状态管理 | 模块化 | 无需理解内部实现 | + +**Deprecated/outdated:** +- 无 + +## Open Questions + +1. **Parse processor 的输入 Schema 设计** + - 需要哪些字段? + - 内容目录路径（从 download 步骤获取） + - 上一步处理器的输入（如 ASr 的 video_path） + + - 当前设计: ParseInput 包含 content_dir + - 问题: 是否需要 output_dir 字段？ + - 建议: 添加 output_dir 字段，明确指定输出目录 + + - 问题: DownloadProcessor 的输入 Schema 设计 + - 需要哪些字段？ + - 内容类型（video/article）影响后续处理逻辑 + - 建议: 参考现有 ASRInput 设计，添加 content_type 和 content_dir 字段 + +``` + +## Validation Architecture + +> Skip this section entirely if workflow.nyquist_validation is explicitly set to false in .planning/config.json. If the key is absent, treat as enabled. + +### Test Framework +| Property | Value | +|----------|-------| +| Framework | pytest + pytest-asyncio | +| Config file | pytest.ini | +| Quick run command | `pytest tests/test_processors/test_analyze.py -v -x` | +| Full suite command | `pytest tests/ -v` | + +### Phase Requirements → Test Map +| Req ID | Behavior | Test Type | Automated Command | File Exists? | +|--------|----------|-----------|-------------------|-------------| +| MNT-01 | BaseProcessor 接口定义 | unit | `pytest tests/test_processors/test_base.py::test_base_processor_interface -x` | Wave 0 | +| MNT-01 | Pipeline 类实现 | unit | `pytest tests/test_pipeline/test_pipeline.py::test_pipeline_run -x` | Wave 0 | +| MNT-01 | ParseProcessor 实现 | unit | `pytest tests/test_processors/test_parse.py::test_parse_processor -x` | Wave 0 | +| MNT-01 | DownloadProcessor 实现 | unit | `pytest tests/test_processors/test_download.py::test_download_processor -x` | Wave 0 | +| MNT-01 | AnalyzeProcessor 实现 | unit | `pytest tests/test_processors/test_analyze.py::test_analyze_processor -x` | Wave 0 | +| MNT-01 | CLI process 命令 | unit | `pytest tests/test_cli/test_process_command.py::test_process_command -x` | Wave 0 | +| MNT-01 | 端到端工作流 | integration | `pytest tests/test_integration/test_e2e_workflow.py -x` | Wave 0 | +| MNT-01 | 幂等性保证 | unit | `pytest tests/test_pipeline/test_idempotency.py -x` | Wave 0 | +| MNT-01 | 错误恢复策略 | unit | `pytest tests/test_pipeline/test_error_recovery.py -x` | Wave 0 | + +### Sampling Rate +- **Per task commit:** `pytest tests/test_processors/test_{module}.py -v -x` (单个处理器测试) +- **Per wave merge:** `pytest tests/ -v` (全量测试) +- **Phase gate:** 全量测试绿色通过 + +### Wave 0 Gaps +- [ ] `tests/test_processors/test_base.py` — BaseProcessor 接口测试 +- [ ] `tests/test_processors/test_parse.py` — ParseProcessor 测试 +- [ ] `tests/test_processors/test_download.py` — DownloadProcessor 测试 +- [ ] `tests/test_processors/test_analyze.py` — AnalyzeProcessor 测试 +- [ ] `tests/test_pipeline/test_pipeline.py` — Pipeline 类测试 +- [ ] `tests/test_cli/test_process_command.py` — CLI process 命令测试 +- [ ] `tests/test_integration/test_e2e_workflow.py` — 端到端集成测试 +- [ ] `tests/test_pipeline/test_idempotency.py` — 幂等性测试 +- [ ] `tests/test_pipeline/test_error_recovery.py` — 错误恢复测试 + +## Sources + +### Primary (HIGH confidence) +- 现有代码: kb_pipeline/processors/asr.py - ASRProcessor 实现参考 +- 现有代码: kb_pipeline/schemas.py - StepInput/StepOutput/WorkflowState +- 现有代码: kb_pipeline/state.py - WorkflowStateManager +- 现有代码: kb_pipeline/config.py - KBSettings +- 现有代码: kb_pipeline/errors.py - KBError 体系 +- 现有代码: kb_pipeline/retry.py - tenacity 重试机制 +- 现有代码: kb_pipeline/link_parser.py - LinkParser +- 现有代码: kb_pipeline/platforms/base.py - PlatformAdapter 基类 +- 现有代码: kb-deep-analyzer/deep_analyzer.py - DeepAnalyzer 实现 + +- 现有代码: kb-workflow/kb_workflow.py - CLI 入口 + +### Secondary (MEDIUM confidence) +- pytest-asyncio 文档 - 异步测试支持 + +- argparse 文档 - CLI 参数解析 + +### Tertiary (LOW confidence) +- 无 + +## Metadata + +**Confidence breakdown:** +- Standard stack: HIGH - 现有代码分析 + Python 标准库 +- Architecture: HIGH - 现有代码模式清晰，设计合理 +- Pitfalls: MEDIUM - 基于经验推断，需要实际验证 + +- Validation: HIGH - 现有测试基础设施完整 + +**Research date:** 2026-03-13 +**Valid until:** 30 days (架构稳定)