Archived

This repository has been archived on 2026-08-07. You can view files and clone it. You cannot open issues or pull requests or push a commit.

Files

T

starmen1265 408a02e615 docs: complete project research

2026-03-13 15:45:47 +08:00

16 KiB

Raw Blame History

Architecture Research

Domain: Skill-based Workflow Orchestration System Researched: 2026-03-13 Confidence: HIGH

Standard Architecture

System Overview

技能型工作流编排系统采用分层架构，将处理逻辑、状态管理和执行协调分离。基于当前研究和行业最佳实践，推荐以下四层架构：

+---------------------------------------------------------------+
|                    Orchestration Layer                        |
|    [Workflow Coordinator] [State Manager] [Error Handler]     |
+---------------------------------------------------------------+
|                    Skill Definition Layer                     |
|    [SKILL.md specs] [Schema Definitions] [Trigger Rules]      |
+---------------------------------------------------------------+
|                    Execution Layer                            |
|    [Platform Skills] [Processing Skills] [Analysis Skills]    |
+---------------------------------------------------------------+
|                    Infrastructure Layer                       |
|    [File System] [External APIs] [AI Services] [State Store]  |
+---------------------------------------------------------------+

Component Responsibilities

Component	Responsibility	Typical Implementation
Workflow Coordinator	协调多个技能的执行顺序、依赖管理	kb-workflow 主控模块
State Manager	步骤级状态持久化、断点续传支持	workflow_state.json + 原子写入
Error Handler	错误分类、降级策略、重试逻辑	统一错误处理模块
Skill Registry	技能发现、能力声明、Schema 注册	SKILL.md + JSON Schema
Platform Skills	平台特定内容获取（B站/知乎）	kb-bilibili-, kb-zhihu-
Processing Skills	内容转换（语音识别/OCR）	kb-whisper, kb-funasr, kb-ocr
Analysis Skills	AI 深度分析、报告生成	kb-deep-analyzer

Recommended Project Structure

kb-skills/
+-- skills/                      # 技能模块目录
|   +-- core/                    # 核心基础设施
|   |   +-- state_manager.py     # 状态管理
|   |   +-- schema_registry.py   # Schema 注册
|   |   +-- pipeline.py          # 管道执行引擎
|   |   +-- error_handler.py     # 错误处理
|   |
|   +-- platforms/               # 平台技能
|   |   +-- bilibili/            # B站技能组
|   |   |   +-- SKILL.md
|   |   |   +-- metadata.py
|   |   |   +-- content.py
|   |   +-- zhihu/               # 知乎技能组
|   |       +-- SKILL.md
|   |       +-- metadata.py
|   |       +-- content.py
|   |
|   +-- processing/              # 处理技能
|   |   +-- whisper/             # 语音识别
|   |   +-- funasr/              # 中文语音识别
|   |   +-- ocr/                 # OCR 处理
|   |
|   +-- analysis/                # 分析技能
|       +-- deep_analyzer/       # 深度分析
|
+-- schemas/                     # JSON Schema 定义
|   +-- workflow_state.schema.json
|   +-- skill_output.schema.json
|   +-- platform_metadata.schema.json
|
+-- workflows/                   # 工作流定义
|   +-- video_pipeline.yaml      # 视频处理流程
|   +-- article_pipeline.yaml    # 文章处理流程
|
+-- cli.py                       # 统一 CLI 入口
+-- config.py                    # 配置管理

Structure Rationale

skills/core/: 核心基础设施，所有技能共享的状态管理、Schema 验证和管道执行
skills/platforms/: 按平台组织，每个平台有自己的元数据提取和内容下载技能
skills/processing/: 内容转换技能，独立于平台
skills/analysis/: AI 分析技能，作为管道的最后阶段
schemas/: 统一的 JSON Schema 定义，确保接口一致性
workflows/: 声明式工作流定义，指定技能组合和执行顺序

Architectural Patterns

Pattern 1: Pipeline Orchestration (管道编排)

What: 将处理流程建模为一系列串联的步骤，每个步骤接收输入、执行处理、产生输出。

When to use: 可预测的、重复性的任务，如内容处理管道。

Trade-offs:

优点：确定性执行、易于调试、支持断点续传
缺点：灵活性有限，不适合需要动态决策的场景

Example:

# Pipeline definition
class ContentPipeline:
    steps = [
        ("parse", LinkParserSkill),
        ("metadata", MetadataSkill),      # depends on parse
        ("download", ContentSkill),       # depends on metadata
        ("transcribe", TranscribeSkill),  # conditional: only for video
        ("analyze", AnalyzeSkill),        # depends on download + transcribe
    ]

    def run(self, url: str, state: WorkflowState):
        for step_name, skill in self.steps:
            if state.is_completed(step_name):
                continue  # Skip completed steps

            result = skill.execute(state.get_step_input(step_name))
            state.mark_completed(step_name, result)

            if not result.success:
                return self.handle_error(result.error)

Pattern 2: Step-Level State Management (步骤级状态管理)

What: 每个主要步骤完成后立即持久化状态，支持从任意断点恢复执行。

When to use: 长时间运行的批处理任务，可能因网络、API 限制等原因中断。

Trade-offs:

优点：断点续传、执行追踪、幂等性保证
缺点：状态管理开销、需要设计状态 Schema

Example:

# workflow_state.json
{
    "url": "https://www.bilibili.com/video/BV1xx411c7mD",
    "platform": "bilibili",
    "content_type": "video",
    "created_at": "2026-03-13T10:00:00Z",
    "updated_at": "2026-03-13T10:15:00Z",
    "steps": {
        "parse": {
            "status": "completed",
            "output": {"video_id": "BV1xx411c7mD"},
            "completed_at": "2026-03-13T10:00:05Z"
        },
        "metadata": {
            "status": "completed",
            "output": {"title": "...", "duration": 1800},
            "completed_at": "2026-03-13T10:00:30Z"
        },
        "download": {
            "status": "in_progress",
            "output": null,
            "started_at": "2026-03-13T10:01:00Z"
        }
    }
}

Pattern 3: Schema-First Interface Design (Schema 优先接口设计)

What: 先定义输入输出的 JSON Schema，再实现技能逻辑。所有技能遵守统一的接口规范。

When to use: 多人协作、需要接口稳定性和可测试性的项目。

Trade-offs:

优点：接口清晰、易于测试、支持自动验证
缺点：前期设计成本、Schema 维护负担

Example:

// schemas/skill_output.schema.json
{
    "$schema": "http://json-schema.org/draft-07/schema#",
    "type": "object",
    "required": ["success", "data"],
    "properties": {
        "success": {"type": "boolean"},
        "data": {"type": "object"},
        "error": {
            "type": "object",
            "properties": {
                "code": {"type": "string"},
                "message": {"type": "string"},
                "detail": {"type": "string"}
            }
        },
        "metadata": {
            "type": "object",
            "properties": {
                "duration_ms": {"type": "number"},
                "tokens_used": {"type": "number"}
            }
        }
    }
}

Pattern 4: Idempotent Execution (幂等执行)

What: 相同输入重复执行产生相同结果，不会产生副作用（如重复文件）。

When to use: 任何可能需要重试的处理步骤。

Trade-offs:

优点：安全重试、简化错误恢复
缺点：需要设计去重逻辑

Example:

class ContentSkill:
    def execute(self, input: dict) -> SkillOutput:
        # Generate deterministic filename from input
        content_hash = hashlib.sha256(input['url'].encode()).hexdigest()[:12]
        output_path = self.get_output_dir() / f"{content_hash}.mp4"

        # Check if already exists
        if output_path.exists():
            return SkillOutput(success=True, data={"path": str(output_path), "cached": True})

        # Download content
        self.download_content(input['url'], output_path)
        return SkillOutput(success=True, data={"path": str(output_path), "cached": False})

Data Flow

Content Processing Pipeline

[URL Input]
    |
    v
+-------------+     +----------------+
| Link Parser | --> | Platform/Type  |
+-------------+     | Detection      |
                    +----------------+
                           |
           +---------------+---------------+
           |                               |
           v                               v
    +-------------+                 +-------------+
    | Bilibili    |                 | Zhihu       |
    | Pipeline    |                 | Pipeline    |
    +-------------+                 +-------------+
           |                               |
    +------+------+                +-------+-------+
    |             |                |               |
    v             v                v               v
[Metadata]   [Download]       [Metadata]     [Download]
    |             |                |               |
    +------+------+                +-------+-------+
           |                               |
           v                               v
    +-------------+                 +-------------+
    | Video?      |                 | Article     |
    | - Whisper   |                 | - Direct    |
    | - FunASR    |                 |   Extract   |
    | - OCR       |                 +-------------+
    +-------------+                       |
           |                              |
           +---------------+--------------+
                           |
                           v
                  +-----------------+
                  | Deep Analyzer   |
                  | (AI Analysis)   |
                  +-----------------+
                           |
                           v
                  +-----------------+
                  | Archive to      |
                  | Knowledge Base  |
                  +-----------------+

State Management Flow

[Workflow Start]
       |
       v
+------------------+
| Load/Init State  | <-- workflow_state.json
+------------------+
       |
       v
+------------------+     +------------------+
| For Each Step    | --> | Check Status     |
+------------------+     | - completed? skip|
       |                 | - failed? retry  |
       v                 | - pending? run   |
+------------------+     +------------------+
| Execute Step     |            |
+------------------+            v
       |               +------------------+
       v               | Mark Completed   |
+------------------+    | Save State       |
| Validate Output  | -->| (atomic write)   |
+------------------+    +------------------+
       |
       v
[Next Step / Complete]

Key Data Flows

URL to Metadata: URL -> Link Parser -> (platform, type, id) -> Platform Metadata Skill -> Metadata JSON
Metadata to Content: Metadata JSON -> Content Skill -> (video file / article text)
Content to Transcript: Video File -> Transcribe Skill (Whisper/FunASR) -> Transcript JSON
Content + Transcript to Analysis: Content + Transcript -> Deep Analyzer -> Analysis Report

Scaling Considerations

Scale	Architecture Adjustments
1-100 items/day	当前架构足够，单进程执行
100-1000 items/day	添加任务队列，并行处理独立任务
1000+ items/day	分布式任务调度，API 请求限流，缓存层

Scaling Priorities

First bottleneck: 外部 API 限流（B站/知乎）。解决方案：请求队列 + 指数退避重试
Second bottleneck: AI 分析耗时。解决方案：批处理 + 并行分析

Anti-Patterns

Anti-Pattern 1: Global State Mutation

What people do: 在多个技能间共享可变全局状态

Why it's wrong: 难以追踪状态变化、难以测试、难以恢复

Do this instead: 每个步骤从 state 读取输入，将输出写回 state，状态变更通过 state manager 集中管理

Anti-Pattern 2: Implicit Dependencies

What people do: 技能之间通过隐式约定传递数据（如固定文件路径）

Why it's wrong: 脆弱的耦合，修改一个技能可能破坏其他技能

Do this instead: 在 workflow_state.json 中显式声明依赖关系，通过 state manager 传递数据

Anti-Pattern 3: Silent Failure

What people do: 捕获异常后只打印日志，继续执行

Why it's wrong: 下游步骤收到无效输入，产生难以调试的错误

Do this instead: 明确的失败处理策略 - 要么终止并记录，要么降级并提供默认值

Anti-Pattern 4: Non-Idempotent Operations

What people do: 每次执行都创建新文件（如 output_1.mp4, output_2.mp4）

Why it's wrong: 重试产生垃圾文件，磁盘占用增长

Do this instead: 基于输入生成确定性文件名，已存在则跳过

Integration Points

External Services

Service	Integration Pattern	Notes
Bilibili API	HTTP + Cookie Auth	需要 BILIBILI_COOKIE 环境变量
Zhihu API	HTTP + Cookie Auth	需要 ZHIHU_COOKIE 环境变量
AI Provider	OpenAI-compatible API	需要 AI_API_KEY 环境变量
HuggingFace	API + Local Cache	需要 HF_TOKEN 环境变量
Git	Command Line	需要 KB_GIT_USER/EMAIL 配置

Internal Boundaries

Boundary	Communication	Notes
Orchestrator <-> Skills	JSON via stdin/stdout	标准化输入输出格式
Skills <-> State Manager	File I/O	workflow_state.json
Skills <-> External APIs	HTTP	统一错误处理和重试

Build Order Implications

基于组件依赖关系，建议的构建顺序：

Phase 1: Foundation (基础设施)

State Manager - 所有后续组件的基础
Schema Registry - 定义统一的输入输出格式
Error Handler - 统一错误处理模式

Phase 2: Core Skills (核心技能)

Link Parser - 入口点，无依赖
Platform Metadata Skills - 依赖 Link Parser 输出
Platform Content Skills - 依赖 Metadata 输出

Phase 3: Processing Skills (处理技能)

Whisper/FunASR - 依赖视频文件
OCR Processor - 依赖视频截图

Phase 4: Analysis Skills (分析技能)

Deep Analyzer - 依赖内容和转录文本

Phase 5: Orchestration (编排层)

Pipeline Engine - 组合所有技能
Workflow Definitions - 声明式流程定义

Sources

AI Architecture Patterns 101: Workflows, Agents, MCPs, and A2A Systems - HIGH confidence, comprehensive pattern overview
Data Pipeline Architecture: 5 Design Patterns with Examples - Dagster - HIGH confidence, pipeline patterns
Agentic Workflows in 2026: The ultimate guide - Vellum - HIGH confidence, agent architecture patterns
State of Open Source Workflow Orchestration Systems 2025 - MEDIUM confidence, ecosystem overview
ML Pipeline Architecture Design Patterns - Neptune.ai - MEDIUM confidence, ML-specific patterns

Architecture research for: Skill-based Workflow Orchestration System Researched: 2026-03-13

16 KiB Raw Blame History Unescape Escape

Architecture Research

Standard Architecture

System Overview

Component Responsibilities

Recommended Project Structure

Structure Rationale

Architectural Patterns

Pattern 1: Pipeline Orchestration (管道编排)

Pattern 2: Step-Level State Management (步骤级状态管理)

Pattern 3: Schema-First Interface Design (Schema 优先接口设计)

Pattern 4: Idempotent Execution (幂等执行)

Data Flow

Content Processing Pipeline

State Management Flow

Key Data Flows

Scaling Considerations

Scaling Priorities

Anti-Patterns

Anti-Pattern 1: Global State Mutation

Anti-Pattern 2: Implicit Dependencies

Anti-Pattern 3: Silent Failure

Anti-Pattern 4: Non-Idempotent Operations

Integration Points

External Services

Internal Boundaries

Build Order Implications

Phase 1: Foundation (基础设施)

Phase 2: Core Skills (核心技能)

Phase 3: Processing Skills (处理技能)

Phase 4: Analysis Skills (分析技能)

Phase 5: Orchestration (编排层)

Sources

16 KiB

Raw Blame History