doc-pipeline

文档流水线技能

概述

本技能支持构建文档处理流水线——将多个操作（提取、转换、转换）链接成可复用的工作流，数据在阶段之间流动。

使用方法

1. 描述你想要完成的任务
2. 提供所需的输入数据或文件
3. 我将执行相应的操作

示例提示：

• "PDF → 提取文本 → 翻译 → 生成 DOCX"
• "图片 → OCR → 总结 → 创建报告"
• "Excel → 分析 → 生成图表 → 创建 PPT"
• "多个输入 → 合并 → 格式化 → 输出"

领域知识

流水线架构

阶段 1      阶段 2      阶段 3      阶段 4
┌──────┐    ┌──────┐    ┌──────┐    ┌──────┐
│提取  │ → │转换  │ → │ AI   │ → │输出  │
│ PDF  │    │ 数据  │    │分析  │   │ DOCX │
└──────┘    └──────┘    └──────┘    └──────┘
     │           │           │           │
     └───────────┴───────────┴───────────┘
                 数据流

流水线 DSL（领域特定语言）

# pipeline.yaml
name: contract-review-pipeline
description: 提取、分析并报告合同

stages:
  - name: extract
    operation: pdf-extraction
    input: $input_file
    output: $extracted_text
    
  - name: analyze
    operation: ai-analyze
    input: $extracted_text
    prompt: "审查此合同的风险..."
    output: $analysis
    
  - name: report
    operation: docx-generation
    input: $analysis
    template: templates/review_report.docx
    output: $output_file

Python 实现

from typing import Callable, Any
from dataclasses import dataclass

@dataclass
class Stage:
    name: str
    operation: Callable
    
class Pipeline:
    def __init__(self, name: str):
        self.name = name
        self.stages: list[Stage] = []
    
    def add_stage(self, name: str, operation: Callable):
        self.stages.append(Stage(name, operation))
        return self  # Fluent API
    
    def run(self, input_data: Any) -> Any:
        data = input_data
        for stage in self.stages:
            print(f"运行阶段: {stage.name}")
            data = stage.operation(data)
        return data

# 示例用法
pipeline = Pipeline("contract-review")
pipeline.add_stage("extract", extract_pdf_text)
pipeline.add_stage("analyze", analyze_with_ai)
pipeline.add_stage("generate", create_docx_report)

result = pipeline.run("/path/to/contract.pdf")

高级：条件流水线

class ConditionalPipeline(Pipeline):
    def add_conditional_stage(self, name: str, condition: Callable, 
                               if_true: Callable, if_false: Callable):
        def conditional_op(data):
            if condition(data):
                return if_true(data)
            return if_false(data)
        return self.add_stage(name, conditional_op)

# 用法
pipeline.add_conditional_stage(
    "ocr_if_needed",
    condition=lambda d: d.get("has_images"),
    if_true=run_ocr,
    if_false=lambda d: d
)

最佳实践

1. 保持阶段聚焦（单一职责）
2. 使用中间输出进行调试
3. 实现阶段级错误处理
4. 通过 YAML/JSON 使流水线可配置

安装

# 安装所需依赖
pip install python-docx openpyxl python-pptx reportlab jinja2

资源

• 自定义仓库
• Claude Office Skills Hub