Doc Pipeline Skill

Overview

This skill enables building document processing pipelines - chain multiple operations (extract, transform, convert) into reusable workflows with data flowing between stages.

How to Use

1. Describe what you want to accomplish
2. Provide any required input data or files
3. I'll execute the appropriate operations

Example prompts:

• "PDF → Extract Text → Translate → Generate DOCX"
• "Image → OCR → Summarize → Create Report"
• "Excel → Analyze → Generate Charts → Create PPT"
• "Multiple inputs → Merge → Format → Output"

Domain Knowledge

Pipeline Architecture

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

Pipeline DSL (Domain Specific Language)

yaml
1# pipeline.yaml
2name: contract-review-pipeline
3description: Extract, analyze, and report on contracts
4
5stages:
6  - name: extract
7    operation: pdf-extraction
8    input: $input_file
9    output: $extracted_text
10    
11  - name: analyze
12    operation: ai-analyze
13    input: $extracted_text
14    prompt: "Review this contract for risks..."
15    output: $analysis
16    
17  - name: report
18    operation: docx-generation
19    input: $analysis
20    template: templates/review_report.docx
21    output: $output_file

Python Implementation

python
1from typing import Callable, Any
2from dataclasses import dataclass
3
4@dataclass
5class Stage:
6    name: str
7    operation: Callable
8    
9class Pipeline:
10    def __init__(self, name: str):
11        self.name = name
12        self.stages: list[Stage] = []
13    
14    def add_stage(self, name: str, operation: Callable):
15        self.stages.append(Stage(name, operation))
16        return self  # Fluent API
17    
18    def run(self, input_data: Any) -> Any:
19        data = input_data
20        for stage in self.stages:
21            print(f"Running stage: {stage.name}")
22            data = stage.operation(data)
23        return data
24
25# Example usage
26pipeline = Pipeline("contract-review")
27pipeline.add_stage("extract", extract_pdf_text)
28pipeline.add_stage("analyze", analyze_with_ai)
29pipeline.add_stage("generate", create_docx_report)
30
31result = pipeline.run("/path/to/contract.pdf")

Advanced: Conditional Pipelines

python
1class ConditionalPipeline(Pipeline):
2    def add_conditional_stage(self, name: str, condition: Callable, 
3                               if_true: Callable, if_false: Callable):
4        def conditional_op(data):
5            if condition(data):
6                return if_true(data)
7            return if_false(data)
8        return self.add_stage(name, conditional_op)
9
10# Usage
11pipeline.add_conditional_stage(
12    "ocr_if_needed",
13    condition=lambda d: d.get("has_images"),
14    if_true=run_ocr,
15    if_false=lambda d: d
16)

Best Practices

1. Keep stages focused (single responsibility)
2. Use intermediate outputs for debugging
3. Implement stage-level error handling
4. Make pipelines configurable via YAML/JSON

Installation

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

Resources

• Custom Repository
• Claude Office Skills Hub