docs(04): complete phase planning
This commit is contained in:
@@ -36,11 +36,11 @@ must_haves:
|
||||
- from: "kb_pipeline/processors/parse.py"
|
||||
to: "kb_pipeline/link_parser.py"
|
||||
via: "LinkParser.parse()"
|
||||
pattern: "LinkParser\\(\\)\\.parse"
|
||||
pattern: "LinkParser\(\)\.parse"
|
||||
- from: "kb_pipeline/processors/base.py"
|
||||
to: "kb_pipeline/state.py"
|
||||
via: "WorkflowStateManager integration"
|
||||
pattern: "state_manager\\."
|
||||
pattern: "state_manager\."
|
||||
---
|
||||
|
||||
<objective>
|
||||
@@ -198,13 +198,14 @@ Add tests to existing test_schemas.py file.
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 3: Implement ParseProcessor</name>
|
||||
<files>kb_pipeline/processors/parse.py, tests/test_processors/test_parse.py</files>
|
||||
<name>Task 3: Implement ParseProcessor with __init__.py export</name>
|
||||
<files>kb_pipeline/processors/parse.py, kb_pipeline/processors/__init__.py, tests/test_processors/test_parse.py</files>
|
||||
<behavior>
|
||||
- Test 1: ParseProcessor parses Bilibili video URL correctly
|
||||
- Test 2: ParseProcessor parses Zhihu answer URL correctly
|
||||
- Test 3: ParseProcessor returns ParseOutput with success=False for unknown URLs
|
||||
- Test 4: ParseProcessor records state when state_manager is provided
|
||||
- Test 5: ParseProcessor can be imported from kb_pipeline.processors
|
||||
</behavior>
|
||||
<action>
|
||||
Create `kb_pipeline/processors/parse.py`:
|
||||
@@ -222,18 +223,8 @@ Create `kb_pipeline/processors/parse.py`:
|
||||
|
||||
4. Keep implementation simple - just wrap LinkParser
|
||||
5. Handle both success and failure cases with proper state recording
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_processors/test_parse.py -v -x</automated>
|
||||
</verify>
|
||||
<done>ParseProcessor correctly parses URLs and returns structured ParseOutput</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 4: Update processors __init__.py</name>
|
||||
<files>kb_pipeline/processors/__init__.py</files>
|
||||
<action>
|
||||
Update `kb_pipeline/processors/__init__.py` to export new classes:
|
||||
Then update `kb_pipeline/processors/__init__.py` to export new classes:
|
||||
|
||||
```python
|
||||
from .base import BaseProcessor
|
||||
@@ -246,9 +237,9 @@ __all__ = ["BaseProcessor", "ParseProcessor", "ASRProcessor"]
|
||||
Keep existing ASRProcessor export and add new ones.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python -c "from kb_pipeline.processors import BaseProcessor, ParseProcessor, ASRProcessor; print('OK')"</automated>
|
||||
<automated>pytest tests/test_processors/test_parse.py -v -x && python -c "from kb_pipeline.processors import BaseProcessor, ParseProcessor, ASRProcessor; print('OK')"</automated>
|
||||
</verify>
|
||||
<done>All processors can be imported from kb_pipeline.processors</done>
|
||||
<done>ParseProcessor correctly parses URLs, returns structured ParseOutput, and is exported from processors module</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
@@ -258,6 +249,7 @@ Keep existing ASRProcessor export and add new ones.
|
||||
- BaseProcessor is abstract and cannot be instantiated
|
||||
- ParseProcessor correctly wraps LinkParser
|
||||
- State recording works when state_manager is provided
|
||||
- All processors can be imported from kb_pipeline.processors
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
|
||||
@@ -6,6 +6,7 @@ wave: 2
|
||||
depends_on: [04-01]
|
||||
files_modified:
|
||||
- kb_pipeline/processors/download.py
|
||||
- kb_pipeline/processors/__init__.py
|
||||
- kb_pipeline/schemas.py
|
||||
- tests/test_processors/test_download.py
|
||||
autonomous: true
|
||||
@@ -37,7 +38,7 @@ must_haves:
|
||||
- from: "kb_pipeline/processors/download.py"
|
||||
to: "kb_pipeline/state.py"
|
||||
via: "WorkflowStateManager"
|
||||
pattern: "state_manager\\."
|
||||
pattern: "state_manager\."
|
||||
---
|
||||
|
||||
<objective>
|
||||
@@ -138,7 +139,7 @@ Follow StepInput/StepOutput pattern from Phase 1.
|
||||
Create test file with schema validation tests.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_processors/test_download.py -v -x</automated>
|
||||
<automated>pytest tests/test_processors/test_download.py -v -x -k "test_schema"</automated>
|
||||
</verify>
|
||||
<done>DownloadInput/DownloadOutput schemas defined and validated with tests</done>
|
||||
</task>
|
||||
@@ -175,21 +176,22 @@ Create `kb_pipeline/processors/download.py` with DownloadProcessor class.
|
||||
Create comprehensive tests mocking adapters and state_manager.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_processors/test_download.py -v -x</automated>
|
||||
<automated>pytest tests/test_processors/test_download.py -v -x -k "test_metadata"</automated>
|
||||
</verify>
|
||||
<done>DownloadProcessor fetches metadata with state recording</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 3: Add content downloading to DownloadProcessor</name>
|
||||
<files>kb_pipeline/processors/download.py, tests/test_processors/test_download.py</files>
|
||||
<name>Task 3: Add content downloading and __init__.py export</name>
|
||||
<files>kb_pipeline/processors/download.py, kb_pipeline/processors/__init__.py, tests/test_processors/test_download.py</files>
|
||||
<behavior>
|
||||
- Test 1: DownloadProcessor downloads Bilibili article content
|
||||
- Test 2: DownloadProcessor skips video download (handled by ASR)
|
||||
- Test 3: DownloadProcessor creates correct output file paths
|
||||
- Test 4: DownloadProcessor can be imported from kb_pipeline.processors
|
||||
</behavior>
|
||||
<action>
|
||||
Extend DownloadProcessor to download content.
|
||||
Extend DownloadProcessor to download content and update __init__.py.
|
||||
|
||||
**For articles/text content:**
|
||||
- Call adapter.fetch_content() to get text
|
||||
@@ -205,18 +207,8 @@ Extend DownloadProcessor to download content.
|
||||
- Content: `{platform}_{content_id}_content.md` (for articles)
|
||||
|
||||
Update tests to verify content downloading behavior.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_processors/test_download.py -v -x</automated>
|
||||
</verify>
|
||||
<done>DownloadProcessor downloads content for text-based content types</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 4: Update processors __init__.py</name>
|
||||
<files>kb_pipeline/processors/__init__.py</files>
|
||||
<action>
|
||||
Update `kb_pipeline/processors/__init__.py` to export DownloadProcessor.
|
||||
Then update `kb_pipeline/processors/__init__.py` to export DownloadProcessor:
|
||||
|
||||
```python
|
||||
from .base import BaseProcessor
|
||||
@@ -228,9 +220,9 @@ __all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python -c "from kb_pipeline.processors import DownloadProcessor; print('OK')"</automated>
|
||||
<automated>pytest tests/test_processors/test_download.py -v -x && python -c "from kb_pipeline.processors import DownloadProcessor; print('OK')"</automated>
|
||||
</verify>
|
||||
<done>DownloadProcessor exported from processors module</done>
|
||||
<done>DownloadProcessor downloads content for text-based content types and is exported from processors module</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
@@ -240,6 +232,7 @@ __all__ = ["BaseProcessor", "ParseProcessor", "DownloadProcessor", "ASRProcessor
|
||||
- DownloadProcessor correctly uses PlatformAdapter implementations
|
||||
- State recording works for both success and failure cases
|
||||
- Content files are saved to correct locations
|
||||
- DownloadProcessor can be imported from kb_pipeline.processors
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
|
||||
@@ -5,51 +5,41 @@ type: execute
|
||||
wave: 4
|
||||
depends_on: [04-01, 04-02, 04-03]
|
||||
files_modified:
|
||||
- kb_pipeline/schemas.py
|
||||
- kb_pipeline/pipeline.py
|
||||
- kb_pipeline/processors/__init__.py
|
||||
- kb-workflow/kb_workflow.py
|
||||
- tests/test_pipeline/test_pipeline.py
|
||||
- tests/test_cli/test_process_command.py
|
||||
autonomous: true
|
||||
requirements: [MNT-01]
|
||||
user_setup: []
|
||||
|
||||
must_haves:
|
||||
truths:
|
||||
- "Pipeline orchestrestr executes full workflow from URL to analysis"
|
||||
- "Single command triggers entire workflow"
|
||||
- "Progress is visible through simple logging"
|
||||
- "New developers can understand code in 30 minutes"
|
||||
- "Pipeline schemas (PipelineInput/Output/Config) are defined"
|
||||
- "Pipeline class initializes with processors list"
|
||||
- "Pipeline orchestrates processor execution"
|
||||
artifacts:
|
||||
- path: "kb_pipeline/schemas.py"
|
||||
provides: "PipelineInput/Output/Config schemas"
|
||||
contains: "class PipelineInput"
|
||||
- path: "kb_pipeline/pipeline.py"
|
||||
provides: "Pipeline engine class"
|
||||
min_lines: 100
|
||||
- path: "kb-workflow/kb_workflow.py"
|
||||
provides: "CLI entry point with process subcommand"
|
||||
contains: "class Pipeline"
|
||||
min_lines: 50
|
||||
- path: "tests/test_pipeline/test_pipeline.py"
|
||||
provides: "Pipeline tests"
|
||||
min_lines: 100
|
||||
- path: "tests/test_cli/test_process_command.py"
|
||||
provides: "CLI tests"
|
||||
provides: "Pipeline unit tests"
|
||||
contains: "test_pipeline"
|
||||
key_links:
|
||||
- from: "kb_pipeline/pipeline.py"
|
||||
to: "kb_pipeline/processors"
|
||||
via: "Pipeline.run()"
|
||||
pattern: "Pipeline\\("
|
||||
- from: "kb-workflow/kb_workflow.py"
|
||||
to: "kb_pipeline.processors"
|
||||
via: "import"
|
||||
pattern: "from.* import"
|
||||
- from: "kb_pipeline/processors" import ParseProcessor, DownloadProcessor, ASrProcessor, AnalyzeProcessor
|
||||
pattern: "from.* import.*"
|
||||
via: "process() calls"
|
||||
pattern: "processor\.process\("
|
||||
---
|
||||
|
||||
<objective>
|
||||
Create Pipeline engine to orchest all processors and execute the complete workflow from URL to analysis report.
|
||||
Create Pipeline schemas and Pipeline class foundation to orchestrate processor execution.
|
||||
|
||||
Purpose: Enable single-command workflow execution with automatic state management and error recovery, and and new developer to easily understand and code结构.
|
||||
Output: Pipeline engine class, process subcommand, and CLI entry point.
|
||||
Purpose: Establish the core orchestration engine that coordinates all processors in a unified workflow.
|
||||
Output: PipelineInput/Output/Config schemas, Pipeline class with processor initialization and step execution logic.
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@@ -61,24 +51,24 @@ Output: Pipeline engine class, process subcommand, and CLI entry point.
|
||||
@.planning/PROJECT.md
|
||||
@.planning/ROADMAP.md
|
||||
@.planning/STATE.md
|
||||
@.planning/phases/04-analysis-orchestration/04-CONTEXT.md
|
||||
@.planning/phases/04-analysis-orchestration/04-RESEARCH.md
|
||||
@.planning/phases/04-analysis-orchestration/04-context.md
|
||||
@.planning/phases/04-analysis-orchestration/04-research.md
|
||||
|
||||
@.planning/phases/04-analysis-orchestration/04-01-SUMMARY.md
|
||||
@.planning/phases/04-analysis-orchestration/04-02-SUMMARY.md
|
||||
@.planning/phases/04-analysis-orchestration/04-03-SUMMARY.md
|
||||
@.planning/phases/04-analysis-orchestration/04-01-summary.md
|
||||
@.planning/phases/04-analysis-orchestration/04-02-summary.md
|
||||
@.planning/phases/04-analysis-orchestration/04-03-summary.md
|
||||
|
||||
<interfaces>
|
||||
<!-- Key types and contracts from existing code. Executor should use these directly. -->
|
||||
<!-- Key types and contracts from existing codebase. Executor should use these directly. -->
|
||||
|
||||
From kb_pipeline/processors/base.py:
|
||||
```python
|
||||
class BaseProcessor(ABC):
|
||||
STEP_NAME: str
|
||||
async def process(self, input_data: StepInput) -> StepOutput:
|
||||
async def process(self, input_data: StepInput) -> StepOutput
|
||||
```
|
||||
|
||||
From kb_pipeline/processors/parse.py:
|
||||
from kb_pipeline/processors/parse.py:
|
||||
```python
|
||||
class ParseProcessor(BaseProcessor):
|
||||
STEP_NAME = "parse"
|
||||
@@ -103,264 +93,45 @@ from kb_pipeline/processors/analyze.py:
|
||||
```python
|
||||
class AnalyzeProcessor(BaseProcessor):
|
||||
STEP_NAME = "analyze"
|
||||
async def process(self, input_data: AnalyInput) -> AnalyOutput
|
||||
async def process(self, input_data: AnalyzeInput) -> AnalyzeOutput
|
||||
```
|
||||
|
||||
from kb_pipeline/schemas.py:
|
||||
```python
|
||||
class StepInput(BaseModel):
|
||||
content_id: str = platform: Literal["bilibili", "zhihu"]
|
||||
content_id: str = Field(..., min_length=1)
|
||||
platform: Literal["bilibili", "zhihu"]
|
||||
|
||||
class StepOutput(BaseModel):
|
||||
success: bool
|
||||
error_message: Optional[str]
|
||||
completed_at: datetime
|
||||
error_message: Optional[str] = None
|
||||
completed_at: datetime = Field(default_factory=datetime.now)
|
||||
|
||||
class ParseInput(StepInput):
|
||||
url: str
|
||||
url: str = Field(..., min_length=1, description="URL to parse")
|
||||
|
||||
class ParseOutput(StepOutput):
|
||||
parsed: ParsedLink
|
||||
content_dir: Path
|
||||
metadata_file: Optional[str] = None
|
||||
content_file: Optional[str] = None
|
||||
content_type: str = "video" # video needs AS
|
||||
|
||||
# article不需要
|
||||
|
||||
content_file: Optional[str] = None # 从 download步骤获取
|
||||
|
||||
content_type: str # "video" or "article"
|
||||
download_input.content_dir
|
||||
download_output(
|
||||
success=True,
|
||||
metadata_file=f"{content_dir}/{platform}_{content_id}_metadata.json",
|
||||
content_file=content_file,
|
||||
title=metadata.get('title', '未知'),
|
||||
author=metadata.get('author', '未知'),
|
||||
duration=metadata.get('duration') # seconds, )
|
||||
|
||||
return state.mark_step_completed(
|
||||
self.STEP_NAME, f"title={title}, author={author}, type={content_type}"
|
||||
)
|
||||
self.state_manager.save(state)
|
||||
return output
|
||||
|
||||
platform: str
|
||||
content_type: str
|
||||
content_id: str
|
||||
resolved_url: str
|
||||
original_url: str
|
||||
```
|
||||
|
||||
from kb_pipeline.state import WorkflowStateManager
|
||||
from kb_pipeline.config import get_settings
|
||||
from kb_pipeline.errors import PipelineError
|
||||
|
||||
from 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 typing import List, Optional, Dict, Any, TYPE_CHECKING
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PipelineConfig:
|
||||
"""Pipeline configuration"""
|
||||
steps: List[str] = Field(default_factory=lambda: ["parse", "download", "asr", "analyze"])
|
||||
output_dir: Path
|
||||
skip_completed: bool = Field(default=True, description="Skip completed steps")
|
||||
max_retries: int = Field(default=3, description="Max retries for each step")
|
||||
|
||||
|
||||
class Pipeline:
|
||||
"""Pipeline engine - orchestrates multiple processors
|
||||
|
||||
Orchestrates the workflow:
|
||||
1. Parse URL to extract platform and content ID
|
||||
2. Run processors in order (parse, download, asr, analyze)
|
||||
3. Handle errors and retry on failures
|
||||
4. Return results
|
||||
|
||||
Key decisions:
|
||||
- Sequential execution (not parallel - simpler to easier to debug
|
||||
- State persistence after each step (via WorkflowStateManager)
|
||||
- Progress logging to simple, clear
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
processors: List[BaseProcessor],
|
||||
config: Optional[PipelineConfig] = None,
|
||||
output_dir: Optional[Path] = None,
|
||||
skip_completed: bool = False
|
||||
):
|
||||
self.processors = processors
|
||||
self.config = config or PipelineConfig.default()
|
||||
self.output_dir = output_dir
|
||||
|
||||
self.skip_completed = skip_completed
|
||||
|
||||
# Validate processors
|
||||
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[str, Any]:
|
||||
"""Execute full workflow
|
||||
|
||||
Args:
|
||||
url: URL to process
|
||||
|
||||
Returns:
|
||||
Dict containing all step results
|
||||
"""
|
||||
results: Dict[str, Any] = {}
|
||||
current_step = 0
|
||||
total_steps = len(self.config.steps)
|
||||
|
||||
state: Optional[WorkflowState] = None
|
||||
|
||||
try:
|
||||
for step_name in self.config.steps:
|
||||
current_step += 1
|
||||
logger.info(f"[Step {current_step}/{total_steps}] {step_name}...")
|
||||
|
||||
|
||||
# Get processor
|
||||
processor = self.processors.get(step_name)
|
||||
if not processor:
|
||||
raise ValueError(f"Processor not found: {step_name}")
|
||||
|
||||
# Get content directory from parsed URL
|
||||
parsed = await self._parse_url(url)
|
||||
if not parsed:
|
||||
results["error"] = "Failed to parse URL"
|
||||
return results
|
||||
|
||||
# Get or create content directory
|
||||
content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}"
|
||||
content_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Initialize state manager for this content
|
||||
state_manager = WorkflowStateManager(content_dir)
|
||||
|
||||
# Run each step
|
||||
previous_result = None
|
||||
for step_name in self.config.steps:
|
||||
# Check if already completed
|
||||
if state and state.is_step_completed(step_name):
|
||||
current_step += 1
|
||||
logger.info(f"[Step {current_step}/{total_steps}] {step_name} - skipping (already completed)")
|
||||
results[step_name] = {"status": "skipped", "message": "Already completed"}
|
||||
continue
|
||||
|
||||
# Create step input
|
||||
input_data = self._create_step_input(
|
||||
step_name=step_name,
|
||||
parsed=parsed,
|
||||
state=state,
|
||||
previous_result=previous_result
|
||||
)
|
||||
|
||||
# Run processor
|
||||
try:
|
||||
output = await processor.process(input_data)
|
||||
results[step_name] = {
|
||||
"status": "success" if output.success else "failed",
|
||||
"output": output.model_dump() if output.success else None,
|
||||
"error": output.error_message
|
||||
}
|
||||
previous_result = output
|
||||
except Exception as e:
|
||||
results[step_name] = {
|
||||
"status": "failed",
|
||||
"error": str(e)
|
||||
}
|
||||
logger.error(f"[Step {current_step}/{total_steps}] {step_name} - failed: {e}")
|
||||
# Don't stop pipeline on failure
|
||||
raise
|
||||
|
||||
return results
|
||||
|
||||
def _parse_url(self, url: str) -> Optional[ParsedLink]:
|
||||
"""Parse URL using LinkParser"""
|
||||
if not url:
|
||||
return None
|
||||
|
||||
parser = LinkParser()
|
||||
parsed = parser.parse(url)
|
||||
if not parsed:
|
||||
logger.warning(f"Failed to parse URL: {url}")
|
||||
return None
|
||||
|
||||
return ParsedLink
|
||||
|
||||
def _create_step_input(
|
||||
self,
|
||||
step_name: str,
|
||||
parsed: ParsedLink,
|
||||
state: Optional[WorkflowState],
|
||||
previous_result: Optional[Dict[str, Any]],
|
||||
) -> Dict[str, Any]:
|
||||
"""Create step input based on step name and previous result"""
|
||||
step_map = {
|
||||
"parse": ParseProcessor,
|
||||
"download": DownloadProcessor,
|
||||
"asr": ASRProcessor,
|
||||
"analyze": AnalyzeProcessor,
|
||||
}
|
||||
|
||||
# Get processor instance
|
||||
processor = self.processors[step_name]
|
||||
if not processor:
|
||||
raise ValueError(f"Unknown processor: {step_name}")
|
||||
|
||||
return step_map
|
||||
|
||||
async def _run_step(
|
||||
self,
|
||||
step_name: str,
|
||||
input_data: Dict[str, Any],
|
||||
state: Optional[WorkflowState],
|
||||
previous_result: Optional[Dict[str, Any]]
|
||||
) -> StepOutput:
|
||||
"""Run a single step"""
|
||||
processor = self.processors[step_name]
|
||||
if not processor:
|
||||
raise ValueError(f"No processor for {step_name}")
|
||||
|
||||
# Check if already completed
|
||||
if state and state.is_step_completed(step_name):
|
||||
logger.info(f"Step {step_name} already completed, skipping")
|
||||
return self._create_skipped_output()
|
||||
|
||||
|
||||
try:
|
||||
# Run the
|
||||
if previous_result and previous_result.get(step_name):
|
||||
input_data = previous_result[step_name]
|
||||
else:
|
||||
raise ValueError(f"No previous result for {step_name}")
|
||||
|
||||
return await processor.process(input_data)
|
||||
except Exception as e:
|
||||
logger.error(f"Step {step_name} failed: {e}")
|
||||
# Record failure
|
||||
if state:
|
||||
state.mark_step_failed(step_name, str(e))
|
||||
self.state_manager.save(state)
|
||||
raise
|
||||
|
||||
# Update previous result
|
||||
previous_result[step_name] = output.model_dump()
|
||||
|
||||
return results
|
||||
|
||||
def get_output_dir(self) -> Path:
|
||||
return Path(self.output_dir) / Path(self.config.output_dir)
|
||||
from kb_pipeline/state.py:
|
||||
```python
|
||||
class WorkflowStateManager:
|
||||
STATE_FILENAME = "workflow_state.json"
|
||||
def __init__(self, content_dir: Path): ...
|
||||
def load(self) -> Optional[WorkflowState]: ...
|
||||
def save(self, state: WorkflowState) -> None: ...
|
||||
def get_or_create(self, content_id: str, platform: str, source_url: str) -> WorkflowState: ...
|
||||
```
|
||||
|
||||
from kb_pipeline/config.py:
|
||||
```python
|
||||
def get_settings() -> KBSettings: ...
|
||||
```
|
||||
|
||||
</interfaces>
|
||||
</context>
|
||||
@@ -371,29 +142,52 @@ class Pipeline:
|
||||
<name>Task 1: Define Pipeline Schemas</name>
|
||||
<files>kb_pipeline/schemas.py, tests/test_pipeline/test_pipeline.py</files>
|
||||
<behavior>
|
||||
- Test 1: PipelineInput validates required fields
|
||||
- Test 2: PipelineOutput validates required fields
|
||||
- Test 3: PipelineConfig default steps are correct
|
||||
- Test 1: PipelineInput validates url field (required)
|
||||
- Test 2: PipelineOutput contains results dict
|
||||
- Test 3: PipelineConfig has default steps list
|
||||
- Test 4: PipelineConfig allows custom output_dir
|
||||
</behavior>
|
||||
<action>
|
||||
Add PipelineInput, PipelineOutput, and PipelineConfig to kb_pipeline/schemas.py.
|
||||
|
||||
**PipelineInput fields (extends nothing - all fields inherited from StepInput):**
|
||||
- content_dir: str - Directory to save downloaded content
|
||||
**PipelineInput fields:**
|
||||
```python
|
||||
class PipelineInput(BaseModel):
|
||||
"""Pipeline input - URL to process"""
|
||||
url: str = Field(..., min_length=1, description="URL to process")
|
||||
output_dir: Optional[Path] = Field(None, description="Output directory for content")
|
||||
skip_completed: bool = Field(default=True, description="Skip already completed steps")
|
||||
```
|
||||
|
||||
- content_type: str - Type of content (video, article, answer, zhuanlan)
|
||||
**PipelineOutput fields:**
|
||||
```python
|
||||
class PipelineOutput(BaseModel):
|
||||
"""Pipeline output - results from all steps"""
|
||||
success: bool
|
||||
error_message: Optional[str] = None
|
||||
results: Dict[str, Any] = Field(default_factory=dict)
|
||||
content_dir: Optional[Path] = None
|
||||
```
|
||||
|
||||
- skip_completed: bool - Whether to skip already completed steps (default True, content_type not to ["video", "article", "answer", "zhuanlan", "download", "asr", "analyze"]
|
||||
**PipelineConfig fields:**
|
||||
```python
|
||||
class PipelineConfig(BaseModel):
|
||||
"""Pipeline configuration"""
|
||||
steps: List[str] = Field(
|
||||
default_factory=lambda: ["parse", "download", "asr", "analyze"],
|
||||
description="Steps to execute in order"
|
||||
)
|
||||
output_dir: Optional[Path] = Field(None, description="Base output directory")
|
||||
skip_completed: bool = Field(default=True, description="Skip completed steps")
|
||||
max_retries: int = Field(default=3, description="Max retries per step")
|
||||
```
|
||||
|
||||
- steps: List[str] = Field(...)
|
||||
|
||||
output_dir: Path
|
||||
skip_completed: bool = True
|
||||
Create test file tests/test_pipeline/test_pipeline.py with schema validation tests.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_pipeline/test_pipeline.py -v -x -k"test_schema" in</automated>
|
||||
<automated>pytest tests/test_pipeline/test_pipeline.py -v -x -k "test_schema"</automated>
|
||||
</verify>
|
||||
<done>PipelineInput/Output schemas defined and validated</done>
|
||||
<done>PipelineInput/Output/Config schemas defined and validated</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
@@ -402,257 +196,102 @@ Add PipelineInput, PipelineOutput, and PipelineConfig to kb_pipeline/schemas.py.
|
||||
<behavior>
|
||||
- Test 1: Pipeline initializes with all processors
|
||||
- Test 2: Pipeline.parse_url creates correct content directory
|
||||
- Test 3: Pipeline.run() executes all steps in correct order
|
||||
- Test 4: Pipeline.run() handles step skipping correctly
|
||||
- Test 5: Pipeline.run() handles errors and continues to next step
|
||||
- Test 6: Pipeline.run() records state for each step
|
||||
- Test 3: Pipeline validates processor STEP_NAME matches config.steps
|
||||
- Test 4: Pipeline handles missing processor gracefully
|
||||
</behavior>
|
||||
<action>
|
||||
Create `kb_pipeline/pipeline.py` with Pipeline class.
|
||||
|
||||
**Implementation:**
|
||||
1. Initialize with processors list (ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor)
|
||||
2. Set config with PipelineConfig.default()
|
||||
3. Set output_dir
|
||||
4. Initialize state_manager (WorkflowStateManager)
|
||||
5. Implement run() method that:
|
||||
- Creates content directory
|
||||
- Runs steps in order (parse, download, asr, analyze)
|
||||
- Returns results dict
|
||||
```python
|
||||
from typing import List, Optional, Dict, Any
|
||||
from pathlib import Path
|
||||
import logging
|
||||
|
||||
6. Add helper methods:
|
||||
- _parse_url(url) -> ParsedLink
|
||||
- _get_content_dir(parsed)
|
||||
- _run_step(step_name, parsed, state, previous_result)
|
||||
- _record_state
|
||||
from kb_pipeline.config import get_settings
|
||||
from kb_pipeline.schemas import PipelineConfig, PipelineInput, PipelineOutput
|
||||
from kb_pipeline.state import WorkflowStateManager
|
||||
from kb_pipeline.link_parser import LinkParser, ParsedLink
|
||||
from kb_pipeline.processors.base import BaseProcessor
|
||||
|
||||
7. Add error handling with try/except
|
||||
- logger.error for error message
|
||||
if self.state_manager:
|
||||
state.mark_step_failed(step_name, str(e))
|
||||
self.state_manager.save(state)
|
||||
raise
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def _parse_url(self, url: str) -> Optional[ParsedLink]:
|
||||
|
||||
class Pipeline:
|
||||
"""Pipeline engine - orchestrates multiple processors
|
||||
|
||||
Orchestrates the workflow:
|
||||
1. Parse URL to extract platform and content ID
|
||||
2. Run processors in order (parse, download, asr, analyze)
|
||||
3. Handle errors and state persistence
|
||||
4. Return results
|
||||
|
||||
Key design decisions:
|
||||
- Sequential execution (simpler, easier to debug)
|
||||
- State persistence after each step (via WorkflowStateManager)
|
||||
- Simple progress logging
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
processors: List[BaseProcessor],
|
||||
config: Optional[PipelineConfig] = None,
|
||||
output_dir: Optional[Path] = None,
|
||||
):
|
||||
self.processors = {p.STEP_NAME: p for p in processors}
|
||||
self.config = config or PipelineConfig()
|
||||
self.output_dir = output_dir or Path.cwd()
|
||||
self.settings = get_settings()
|
||||
|
||||
# Validate processors
|
||||
for step_name in self.processors:
|
||||
if step_name not in self.config.steps:
|
||||
raise ValueError(f"Unknown processor: {step_name}")
|
||||
|
||||
def _parse_url(self, url: str) -> Optional[ParsedLink]:
|
||||
"""Parse URL using LinkParser"""
|
||||
if not url:
|
||||
return None
|
||||
|
||||
parser = LinkParser()
|
||||
parsed = parser.parse(url)
|
||||
if not parsed:
|
||||
logger.warning(f"Failed to parse URL: {url}")
|
||||
return None
|
||||
|
||||
return Parsed
|
||||
return parsed
|
||||
|
||||
def _get_content_dir(self, parsed: ParsedLink) -> Path:
|
||||
"""Get content directory from parsed URL"""
|
||||
content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}"
|
||||
content_dir.mkdir(parents=True, exist_ok=True)
|
||||
return content_dir
|
||||
```
|
||||
|
||||
def _create_step_input(
|
||||
self,
|
||||
step_name: str,
|
||||
parsed: ParsedLink,
|
||||
state: Optional[WorkflowState],
|
||||
previous_result: Optional[Dict[str, Any]]
|
||||
) -> Dict[str, Any]:
|
||||
"""Create step input based on step name and previous result"""
|
||||
step_map = {
|
||||
"parse": ParseProcessor,
|
||||
"download": DownloadProcessor,
|
||||
"asr": ASRProcessor,
|
||||
"analyze": AnalyzeProcessor
|
||||
}
|
||||
|
||||
# Get processor instance
|
||||
processor = self.processors[step_name]
|
||||
if not processor:
|
||||
raise ValueError(f"Unknown processor: {step_name}")
|
||||
|
||||
return step_map
|
||||
|
||||
async def _run_step(
|
||||
self,
|
||||
step_name: str,
|
||||
input_data: Dict[str, Any],
|
||||
state: Optional[WorkflowState],
|
||||
previous_result: Optional[Dict[str, Any]],
|
||||
) -> StepOutput:
|
||||
"""Run a single step"""
|
||||
processor = self.processors[step_name]
|
||||
if not processor:
|
||||
raise ValueError(f"No processor for {step_name}")
|
||||
|
||||
# Check if already completed
|
||||
if state and state.is_step_completed(step_name):
|
||||
logger.info(f"Step {step_name} already completed, skipping")
|
||||
return self._create_skipped_output()
|
||||
|
||||
try:
|
||||
# Run step
|
||||
if previous_result and previous_result.get(step_name):
|
||||
input_data = previous_result[step_name]
|
||||
else:
|
||||
raise ValueError(f"No previous result for {step_name}")
|
||||
|
||||
return await processor.process(input_data)
|
||||
except Exception as e:
|
||||
logger.error(f"Step {step_name} failed: {e}")
|
||||
if state:
|
||||
state.mark_step_failed(step_name, str(e))
|
||||
self.state_manager.save(state)
|
||||
raise
|
||||
|
||||
# Update previous result
|
||||
previous_result[step_name] = output.model_dump()
|
||||
return results
|
||||
|
||||
def _create_skipped_output(self) -> StepOutput:
|
||||
return StepOutput(success=True, error_message="Skipped: already completed")
|
||||
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 3: Add step execution logic</name>
|
||||
<files>kb_pipeline/pipeline.py</files>
|
||||
<action>
|
||||
Extend Pipeline.run() to implement step execution logic:
|
||||
|
||||
1. Create content directory
|
||||
2. Run steps in order
|
||||
3. Handle step skipping (check state)
|
||||
4. Handle errors (try/except, log and continue)
|
||||
5. Record final state
|
||||
6. Return results
|
||||
|
||||
**Step execution logic:**
|
||||
```python
|
||||
async def _execute_steps(
|
||||
self,
|
||||
parsed: ParsedLink,
|
||||
state: WorkflowState,
|
||||
previous_result: Dict[str, Any]
|
||||
) -> None:
|
||||
for step_name in self.config.steps:
|
||||
step_start = datetime.now()
|
||||
state.mark_step_started(step_name)
|
||||
|
||||
logger.info(f"[Step {step_num}/{total_steps}] {step_name}...")
|
||||
|
||||
# Get or run processor
|
||||
processor = self.processors[step_name]
|
||||
if not processor:
|
||||
raise ValueError(f"Unknown processor: {step_name}")
|
||||
|
||||
# Get step input
|
||||
input_data = self._create_step_input(
|
||||
step_name=step_name,
|
||||
parsed=parsed,
|
||||
state=state,
|
||||
previous_result=previous_result.get(step_name)
|
||||
)
|
||||
|
||||
# Run processor
|
||||
try:
|
||||
output = await processor.process(input_data)
|
||||
previous_result[step_name] = output.model_dump()
|
||||
results[step_name] = {
|
||||
"status": "success" if output.success else "failed",
|
||||
"output": output.model_dump() if output.success else None,
|
||||
"error": output.error_message
|
||||
}
|
||||
previous_result = output
|
||||
except Exception as e:
|
||||
results[step_name] = {
|
||||
"status": "failed",
|
||||
"error": str(e)
|
||||
}
|
||||
logger.error(f"[Step {step_num}/{total_steps}] {step_name} - failed: {e}")
|
||||
# Record failure
|
||||
if state:
|
||||
state.mark_step_failed(step_name, str(e))
|
||||
self.state_manager.save(state)
|
||||
# Stop on failure
|
||||
raise
|
||||
|
||||
return results
|
||||
Create comprehensive tests with mocked processors.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_pipeline/test_pipeline.py -v -x</automated>
|
||||
<automated>pytest tests/test_pipeline/test_pipeline.py -v -x -k "test_init or test_parse"</automated>
|
||||
</verify>
|
||||
<done>Pipeline engine executes workflow with state persistence</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 4: Add CLI process command</name>
|
||||
<files>kb-workflow/kb_workflow.py</files>
|
||||
<action>
|
||||
Extend kb-workflow/kb_workflow.py to add process subcommand.
|
||||
|
||||
**CLI additions:**
|
||||
- New argument: 'url' with help='URL to process'
|
||||
- New argument: '--output-dir', '-o' with help='Output directory (default: KB_BASE_PATH/content)
|
||||
- New argument: '--no-analyze', action='store_true', help='Skip analysis step')
|
||||
- Set default to process function that handles the command
|
||||
|
||||
- Call Pipeline().run(url, output_dir, skip_completed=no)
|
||||
- Print success message
|
||||
- If error: print error message and exit(1)
|
||||
|
||||
**Update existing help text and kb_workflow.py to document the new command.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python kb-workflow/kb_workflow.py --help | grep process</automated>
|
||||
</verify>
|
||||
<done>CLI process command added and functional</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 5: Create Pipeline tests</name>
|
||||
<files>tests/test_pipeline/test_pipeline.py,files>
|
||||
<action>
|
||||
Create comprehensive tests for Pipeline class.
|
||||
|
||||
**Test cases:**
|
||||
1. Test pipeline initialization
|
||||
2. Test URL parsing and content dir creation
|
||||
3. Test step execution order
|
||||
4. Test step skipping
|
||||
5. Test error handling and6. Test state persistence
|
||||
|
||||
7. Test full workflow (mocked)
|
||||
|
||||
Use fixtures for mocking processors, state_manager, settings.
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_pipeline/test_pipeline.py -v</automated>
|
||||
</verify>
|
||||
<done>Pipeline class fully tested</done>
|
||||
<done>Pipeline class initializes correctly and processor validation</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- All tests pass: `pytest tests/test_pipeline/test_pipeline.py tests/test_cli/test_process_command.py -v`
|
||||
- Pipeline.run() executes workflow correctly
|
||||
- State persistence works
|
||||
- Error handling works
|
||||
- Step skipping works
|
||||
- All tests pass: `pytest tests/test_pipeline/test_pipeline.py -v -x`
|
||||
- PipelineInput/Output/Config schemas exist and validate
|
||||
- Pipeline class initializes with processors
|
||||
- URL parsing creates correct content directory
|
||||
- Processor validation works
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- Pipeline class orchestrates all processors
|
||||
- Pipeline.run() executes workflow from URL to analysis
|
||||
- Content directory created correctly
|
||||
- Steps executed in order
|
||||
- State persisted after each step
|
||||
- Errors handled with retry
|
||||
- Completed steps skipped
|
||||
- CLI process command works
|
||||
- PipelineInput/Output/Config schemas defined in schemas.py
|
||||
- Pipeline class exists in pipeline.py
|
||||
- Pipeline initializes with processor list
|
||||
- URL parsing creates content directory
|
||||
- All tests pass
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/04-analysis-orchestration/04-04-SUMMARY.md`
|
||||
After completion, create `.planning/phases/04-analysis-orchestration/04-04-SUMmary.md`
|
||||
</output>
|
||||
|
||||
@@ -38,24 +38,26 @@ created: 2026-03-13
|
||||
|
||||
| Task ID | Plan | Wave | Requirement | Test Type | Automated Command | File Exists | Status |
|
||||
|---------|------|------|-------------|-----------|-------------------|-------------|--------|
|
||||
| 04-01-01 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_base.py -v` | ⬜ W0 | ⬜ pending |
|
||||
| 04-01-02 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_parse.py -v` | ⬜ W0 | ⬜ pending |
|
||||
| 04-02-01 | 02 | 2 | MNT-01 | unit | `pytest tests/test_processors/test_download.py -v` | ⬜ W0 | ⬜ pending |
|
||||
| 04-03-01 | 03 | 3 | MNT-01 | unit | `pytest tests/test_processors/test_analyze.py -v` | ⬜ W0 | ⬜ pending |
|
||||
| 04-04-01 | 04 | 4 | MNT-01 | unit | `pytest tests/test_pipeline/test_pipeline.py -v` | ⬜ W0 | ⬜ pending |
|
||||
| 04-04-02 | 04 | 4 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | ⬜ W0 | ⬜ pending |
|
||||
| 04-05-01 | 05 | 5 | MNT-01 | integration | `pytest tests/test_integration/test_e2e_workflow.py -v` | ⬜ W0 | ⬜ pending |
|
||||
|
||||
*Status: ⬜ pending · ✅ green · ❌ red · ⚠️ flaky*
|
||||
| 04-01-01 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_base.py -v` | W0 | pending |
|
||||
| 04-01-02 | 01 | 1 | MNT-01 | unit | `pytest tests/test_processors/test_parse.py -v` | W0 | pending |
|
||||
| 04-02-01 | 02 | 2 | MNT-01 | unit | `pytest tests/test_processors/test_download.py -v` | W0 | pending |
|
||||
| 04-03-01 | 03 | 3 | MNT-01 | unit | `pytest tests/test_processors/test_analyze.py -v` | W0 | pending |
|
||||
| 04-04-01 | 04 | 4 | MNT-01 | unit | `pytest tests/test_pipeline/test_pipeline.py -v` | W0 | pending |
|
||||
| 04-04-02 | 04 | 4 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | W0 | pending |
|
||||
| 04-04-03 | 04 | 4 | MNT-01 | unit | `pytest tests/test_pipeline/test_pipeline.py -v` | W0 | pending |
|
||||
| 04-04-04 | 04 | 5 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | W0 | pending |
|
||||
| 04-05-01 | 05 | 5 | MNT-01 | integration | `pytest tests/test_integration/test_e2e_workflow.py -v` | W0 | pending |
|
||||
04-04-05 | 04 | 5 | MNT-01 | unit | `pytest tests/test_cli/test_process_command.py -v` | W0 | pending |
|
||||
|
||||
*Status: pending - green - red - flaky*
|
||||
---
|
||||
|
||||
## Wave 0 Requirements
|
||||
|
||||
Existing infrastructure covers all phase requirements:
|
||||
- `pytest.ini` — configured with pythonpath
|
||||
- `tests/conftest.py` — shared fixtures exist
|
||||
- `tests/test_processors/` — directory exists for processor tests
|
||||
- `pytest.ini` -- configured with pythonpath
|
||||
- `tests/conftest.py` -- shared fixtures exist
|
||||
- `tests/test_processors/` -- directory exists for processor tests
|
||||
|
||||
---
|
||||
|
||||
|
||||
Reference in New Issue
Block a user