doc-pipeline

ドキュメント処理パイプラインスキル

概要

このスキルを使うと、ドキュメント処理パイプラインを構築できます。複数の操作（抽出、変換、形式変換）をつなぎ、各段階でデータを受け渡す再利用可能なワークフローを作成できます。

使い方

1. 実現したいことを説明します
2. 必要な入力データやファイルを提供します
3. 適切な操作を実行します

プロンプトの例：

• 「PDF → テキスト抽出 → 翻訳 → DOCX を生成」
• 「画像 → OCR → 要約 → レポートを作成」
• 「Excel → 分析 → グラフを生成 → PPT を作成」
• 「複数の入力 → 結合 → 書式設定 → 出力」

ドメイン知識

パイプラインアーキテクチャ

Stage 1      Stage 2      Stage 3      Stage 4
┌──────┐    ┌──────┐    ┌──────┐    ┌──────┐
│Extract│ → │Transform│ → │ AI   │ → │Output│
│ PDF  │    │  Data  │    │Analyze│   │ DOCX │
└──────┘    └──────┘    └──────┘    └──────┘
     │           │           │           │
     └───────────┴───────────┴───────────┘
                 Data Flow

パイプライン DSL（ドメイン固有言語）

# pipeline.yaml
name: contract-review-pipeline
description: Extract, analyze, and report on contracts

stages:
  - name: extract
    operation: pdf-extraction
    input: $input_file
    output: $extracted_text
    
  - name: analyze
    operation: ai-analyze
    input: $extracted_text
    prompt: "Review this contract for risks..."
    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"Running stage: {stage.name}")
            data = stage.operation(data)
        return data

# Example usage
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)

# Usage
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 でパイプラインを設定可能にする

インストール

# Install required dependencies
pip install python-docx openpyxl python-pptx reportlab jinja2

リソース

• カスタムリポジトリ
• Claude Office Skills Hub