LLM Pipeline

https://github.com/yugen-au/llm-pipeline
Admin
LLM Pipeline is a declarative framework for orchestrating complex LLM workflows with real-time event...
Tokens:1,101,782
Snippets:8,904
Trust Score:1.5
Update:6 months ago
Show doc for...
Tokens:
Raw
### Getting Started: Installation and Quickstart (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/PLAN.md

Provides instructions for installing the LLM Pipeline framework and a quickstart guide to create your first pipeline. It covers auto-SQLite initialization for development and explicit database setup for production environments, along with provider configuration using GeminiProvider.

```python
from yugen.pipeline import Pipeline
from yugen.providers.gemini import GeminiProvider

# Installation (example using pip):
# pip install yugen-llm-pipeline

# Quickstart Example:
# Initialize provider
provider = GeminiProvider(api_key="YOUR_GEMINI_API_KEY")

# Initialize pipeline (auto-SQLite for development)
pipeline = Pipeline(provider=provider)

# Define pipeline steps, etc. (omitted for brevity)

# Run pipeline
# result = pipeline.run()

# Production Database Setup (example):
# from sqlalchemy import create_engine
# engine = create_engine("postgresql://user:password@host:port/database")
# pipeline = Pipeline(provider=provider, db_engine=engine)

```

--------------------------------

### Update Quick-Start Example in README.md (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/implementation/step-20-review-fixes.md

Replaces an incorrect and overly complex quick-start example in README.md with a simplified essential flow. The previous example used non-existent API parameters and incorrect function signatures. This fix provides a more concise example and directs users to a comprehensive guide for full code.

```python
# Simplified essential flow - full example in guides/basic-pipeline.md

class MyPipeline(
    PipelineConfig,
    registry=MyRegistry,
    strategies=MyStrategies
):
    def sanitize(self, data: str) -> str:
        return data.strip()[:10000]

pipeline = MyPipeline(provider=provider)
pipeline.execute(
    data="your input data",
    initial_context={'key': 'value'}
)
```

--------------------------------

### Install and Run LLM Pipeline UI

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-57-documentation-examples/PLAN.md

Provides command-line instructions for installing the LLM Pipeline UI and starting it in different modes. Includes options for development mode with hot reloading and connecting to a specific database file.

```bash
pip install llm-pipeline[ui]

# Start UI (default port 8642)
llm-pipeline ui

# Development mode with hot reload
llm-pipeline ui --dev --port 8642

# Connect to a specific database
llm-pipeline ui --db /path/to/pipeline.db
```

--------------------------------

### Complete LLM Pipeline Integration Example

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-1-codebase-architecture.md

A comprehensive example demonstrating the setup of an LLM pipeline, including defining domain models with SQLModel, setting up a registry with FK dependency validation, defining instruction and context classes, creating extraction steps, and defining a strategy. This example covers vendor and lane extraction.

```python
# 1. Define Domain Models
class Vendor(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

class RateCard(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    vendor_id: int = Field(foreign_key="vendor.id")
    effective_date: date

class Lane(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    rate_card_id: int = Field(foreign_key="ratecard.id")
    origin: str
    destination: str


# 2. Define Registry (FK order)
class MyRegistry(PipelineDatabaseRegistry, models=[
    Vendor,     # No FKs
    RateCard,   # FK to Vendor
    Lane,       # FK to RateCard
]):
    pass


# 3. Define Instruction Classes
class VendorDetectionInstructions(LLMResultMixin):
    vendor_name: str
    confidence_score: float = 0.95

    example: ClassVar[dict] = {
        "vendor_name": "ACME Logistics",
        "confidence_score": 0.98,
    }

class LaneExtractionInstructions(LLMResultMixin):
    lanes: List[Dict[str, str]]

    example: ClassVar[dict] = {
        "lanes": [
            {"origin": "Sydney", "destination": "Melbourne"},
        ],
        "confidence_score": 0.95,
    }


# 4. Define Context Classes
class VendorDetectionContext(PipelineContext):
    vendor_name: str


# 5. Define Extractions
class VendorExtraction(PipelineExtraction, model=Vendor):
    def default(self, results):
        instruction = results[0]
        return [Vendor(name=instruction.vendor_name)]

class LaneExtraction(PipelineExtraction, model=Lane):
    def default(self, results):
        instruction = results[0]
        rate_card = self.pipeline.get_extractions(RateCard)[0]

        lanes = []
        for lane_dict in instruction.lanes:
            lanes.append(Lane(
                rate_card_id=rate_card.id,
                origin=lane_dict["origin"],
                destination=lane_dict["destination"],
            ))
        return lanes


# 6. Define Steps
@step_definition(
    instructions=VendorDetectionInstructions,
    default_system_key="vendor_detection",
    default_user_key="vendor_detection",
    default_extractions=[VendorExtraction],
    context=VendorDetectionContext,
)
class VendorDetectionStep(LLMStep):
    def prepare_calls(self) -> List[StepCallParams]:
        return [
            self.create_llm_call(
                variables={"data": self.pipeline.get_sanitized_data()}
            )
        ]

    def process_instructions(self, instructions):
        return VendorDetectionContext(
            vendor_name=instructions[0].vendor_name
        )

@step_definition(
    instructions=LaneExtractionInstructions,
    default_system_key="lane_extraction",
    default_user_key="lane_extraction",
    default_extractions=[LaneExtraction],
)
class LaneExtractionStep(LLMStep):
    def prepare_calls(self) -> List[StepCallParams]:
        vendor_name = self.pipeline.context["vendor_name"]
        return [
            self.create_llm_call(
                variables={
                    "data": self.pipeline.get_sanitized_data(),
                    "vendor": vendor_name,
                }
            )
        ]


# 7. Define Strategy
class DefaultStrategy(PipelineStrategy):
    def can_handle(self, context):
        return True  # Always handles

    def get_steps(self):
        return [
            VendorDetectionStep.create_definition(),
            LaneExtractionStep.create_definition(),
        ]

```

--------------------------------

### Install and Run Document Classifier Example (Bash)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/basic-pipeline.md

This code block provides instructions for installing the 'llm-pipeline' library and running a complete working example for a document classifier. It involves navigating to the example directory and executing a Python script.

```bash
# Install dependencies
pip install llm-pipeline

# Run example
cd examples/document_classifier
python run.py
```

--------------------------------

### UI CLI Examples (Bash)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-57-documentation-examples/VALIDATED_RESEARCH.md

Provides examples of how to install and run the LLM pipeline UI using command-line interface commands. Includes flags for development mode, port specification, and database configuration.

```bash
pip install llm-pipeline[ui]
llm-pipeline ui --dev --port 8080 --db "sqlite:///db.sqlite"
```

--------------------------------

### Prompt Key Discovery Example - Python

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/prompts.md

Provides a Python code example illustrating the prompt key auto-discovery process. It shows how the framework searches for strategy-specific and step-level prompts based on the step name and current strategy.

```python
# Step name: SemanticMappingStep → "semantic_mapping"
# Current strategy: "lane_based"

# Search order:
# 1. Look for "semantic_mapping.lane_based" (strategy-specific)
# 2. Look for "semantic_mapping" (step-level fallback)
# 3. Fail if neither found and no explicit key provided
```

--------------------------------

### Rate Card Parser Pipeline Usage Example (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/multi-strategy.md

Demonstrates how to use the RateCardParserPipeline with sample data. It covers setup, execution, checking the strategy, accessing results, and saving to a database.

```python
from sqlmodel import Session, create_engine
import pandas as pd

# Setup
engine = create_engine("sqlite:///rates.db")
session = Session(engine)

# Sample data
df = pd.DataFrame({
    'Origin': ['NYC', 'LAX', 'ORD'],
    'Destination': ['SFO', 'JFK', 'ATL'],
    'Standard Rate': [100, 150, 120],
    'Express Rate': [150, 200, 180],
})

# Execute pipeline
pipeline = RateCardParserPipeline()
pipeline.execute(df, {
    'rate_card_id': 1,
    'vendor_id': 1,
})

# Check which strategy was used
print(f"Table type: {pipeline.context['table_type']}")
print(f"Strategy: {pipeline._current_strategy.name}")

# Access results
lanes = pipeline.get_extractions(Lane)
print(f"Extracted {len(lanes)} lanes")

# Save to database
results = pipeline.save(session)
```

--------------------------------

### Install Core Frontend Dependencies

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-29-init-frontend-structure/research/step-1-frontend-stack-research.md

Installs the core frontend libraries including React, ReactDOM, TanStack libraries for routing and data fetching, and Zustand for state management.

```bash
# Core
npm install react@^19.2.0 react-dom@^19.2.0
npm install @tanstack/react-router @tanstack/react-query zustand

# Tailwind v4
npm install tailwindcss @tailwindcss/vite

# Dev
npm install -D @tanstack/router-plugin @types/node typescript
```

--------------------------------

### Configure LLM Models

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Shows how to specify different LLM providers using pydantic-ai model strings during pipeline construction.

```python
# Google Gemini
pipeline = TextClassifierPipeline(model='google-gla:gemini-2.0-flash-lite')

# OpenAI
pipeline = TextClassifierPipeline(model='openai:gpt-4o')

# Anthropic
pipeline = TextClassifierPipeline(model='anthropic:claude-3-5-sonnet-latest')
```

--------------------------------

### Environment and Database Configuration

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Provides shell commands for setting API keys and configuring the database file location.

```bash
export GEMINI_API_KEY="your-key"
ls -la .llm_pipeline/
export LLM_PIPELINE_DB="/tmp/pipeline.db"
```

--------------------------------

### Complete YAML Prompt Example - YAML

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/prompts.md

A comprehensive example of a user prompt for table extraction. It includes all required fields and optional fields like 'description' and 'is_active', along with specific content for lane-based extraction.

```yaml
# prompts/extraction/table_extraction.yaml
prompt_key: table_extraction.lane_based
name: Lane-Based Table Extraction User Prompt
type: user
category: extraction
step: table_extraction
version: "1.2"
description: |
  User prompt for extracting lane-based rate tables.
  Uses schema hints for structured output.
is_active: true
content: |
  Extract rate information from the following table data:

  {table_data}

  Expected columns: {column_hints}

  Return JSON array matching the schema.
```

--------------------------------

### Configure Production Database with SQLAlchemy (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Illustrates setting up a production database, specifically PostgreSQL, using SQLAlchemy's `create_engine`. It initializes framework tables via `init_pipeline_db` and then creates domain tables, followed by establishing a database session.

```python
from sqlalchemy import create_engine
from sqlmodel import Session, SQLModel
from llm_pipeline.db import init_pipeline_db

# Assuming Category and ClassifiedText are defined SQLModel classes
# class Category:
#     __table__ = ...
# class ClassifiedText:
#     __table__ = ...

# Create engine
engine = create_engine(
    "postgresql://user:pass@localhost/dbname",
    echo=False
)

# Initialize framework tables
init_pipeline_db(engine)

# Create domain tables
SQLModel.metadata.create_all(engine, tables=[
    Category.__table__,
    ClassifiedText.__table__,
])

# Create session
session = Session(engine)

```

--------------------------------

### Automatic Prompt Key Discovery Example

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/implementation/step-15-prompt-management-guide.md

Shows how prompt keys are automatically discovered based on class names and strategy-specific resolutions. It covers the hierarchy of discovery, from strategy-level to step-level and explicit overrides, ensuring the correct prompt is selected.

```python
# Example: SemanticMappingStep class automatically maps to 'semantic_mapping' prompt key
# If a specific prompt key is defined in YAML, it overrides the auto-discovered key.
```

--------------------------------

### Rate Card Parser Pipeline Setup (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/multi-strategy.md

Defines the RateCardParserPipeline, a multi-strategy parser that uses the RateCardParserRegistry and RateCardParserStrategies. Includes a sanitize method for data conversion.

```python
class RateCardParserPipeline(
    PipelineConfig,
    registry=RateCardParserRegistry,
    strategies=RateCardParserStrategies
):
    """Multi-strategy rate card parser."""

    def sanitize(self, data: pd.DataFrame) -> str:
        return data.to_csv(sep='|', index=False)
```

--------------------------------

### Implement Pipeline Extraction Logic

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Creating extraction classes to transform structured LLM output into persistent database models.

```python
class CategoryExtraction(PipelineExtraction, extraction_type=Category):
    def default(self, instructions_list, context):
        categories = []
        for inst in instructions_list:
            category = Category(
                name=inst.category_name,
                description=inst.reasoning
            )
            categories.append(category)
        return categories
```

--------------------------------

### Define LLM Instruction Models

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Using Pydantic models to enforce structured output from LLMs, including validation examples.

```python
from pydantic import BaseModel, Field
from llm_pipeline import LLMResultMixin

class CategoryInstructions(BaseModel, LLMResultMixin):
    category_name: str = Field(description="Name of the category")
    confidence: float = Field(description="Confidence score 0-1")
    reasoning: str = Field(description="Why this category was chosen")

    example = {
        "category_name": "Technology",
        "confidence": 0.95,
        "reasoning": "Text discusses software development"
    }
```

--------------------------------

### JavaScript Testing Setup Example

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-33-run-list-view/VALIDATED_RESEARCH.md

An example setup file for Vitest, often used to import global testing utilities like those provided by @testing-library/jest-dom. This ensures necessary assertion methods are available globally.

```javascript
// vitest.setup.ts
import '@testing-library/jest-dom/vitest';

```

--------------------------------

### Initialize Pipeline Components

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-3-c4-architecture.md

Demonstrates the core imports required to set up the pipeline environment, including context management, extraction, transformation, database registry, and LLM provider interfaces.

```python
from llm_pipeline.context import PipelineContext
from llm_pipeline.extraction import PipelineExtraction
from llm_pipeline.transformation import PipelineTransformation
from llm_pipeline.registry import PipelineDatabaseRegistry
from llm_pipeline.state import PipelineStepState, PipelineRunInstance
from llm_pipeline.db import init_pipeline_db
from llm_pipeline.session import ReadOnlySession
from llm_pipeline.llm import LLMProvider
from llm_pipeline.llm.gemini import GeminiProvider
```

--------------------------------

### Define and Execute Pipeline

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-1-codebase-architecture.md

Demonstrates how to define a pipeline class with registry and strategies, then execute it using a database engine and an LLM provider.

```python
class MyStrategies(PipelineStrategies, strategies=[DefaultStrategy]):
    pass

class MyPipeline(PipelineConfig,
                 registry=MyRegistry,
                 strategies=MyStrategies):
    pass

engine = create_engine("sqlite:///pipeline.db")
provider = GeminiProvider()

pipeline = MyPipeline(engine=engine, provider=provider)
pipeline.execute(
    data="<CSV content>",
    initial_context={},
    use_cache=True,
)

results = pipeline.save()
```

--------------------------------

### Configure Pipeline Registry and Pipeline

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/multi-strategy.md

Sets up the database registry for models and the main pipeline configuration. The pipeline orchestrates the execution flow and handles data sanitization for LLM consumption.

```python
from llm_pipeline import PipelineConfig, PipelineDatabaseRegistry
from .models import Vendor, RateCard, Lane, ChargeType, Rate

class RateCardParserRegistry(
    PipelineDatabaseRegistry,
    models=[
        Vendor,
        RateCard,
        Lane,
        ChargeType,
        Rate,
    ]
):
    """Database models managed by this pipeline."""

class RateCardParserPipeline(
    PipelineConfig,
    registry=RateCardParserRegistry,
    strategies=RateCardParserStrategies
):
    def sanitize(self, data: pd.DataFrame) -> str:
        """Convert DataFrame to LLM-friendly format."""
        return data.to_csv(sep='|', index=False)
```

--------------------------------

### Define Query Parameters Model for GET /runs

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-20-runs-api-endpoints/PLAN.md

Defines a Pydantic model for query parameters used in the GET /runs endpoint. This model includes filters for pipeline name, status, start time, and pagination parameters like offset and limit.

```python
from typing import Optional
from datetime import datetime
from fastapi import Query
from pydantic import BaseModel

class RunListParams(BaseModel):
    pipeline_name: Optional[str] = None
    status: Optional[str] = None
    started_after: Optional[datetime] = None
    started_before: Optional[datetime] = None
    offset: int = Query(default=0, ge=0)
    limit: int = Query(default=50, ge=1, le=200)
```

--------------------------------

### Install shadcn/ui components

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-34-run-detail-view/PLAN.md

Commands to initialize and add required UI components to the frontend project using the shadcn CLI.

```bash
npx shadcn@latest add card
npx shadcn@latest add separator
npx shadcn@latest add scroll-area
```

--------------------------------

### GET /api/runs

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-33-run-list-view/VALIDATED_RESEARCH.md

Retrieves a paginated list of pipeline runs, sorted by start time in descending order.

```APIDOC
## GET /api/runs

### Description
Retrieves a list of pipeline runs with support for pagination. The backend defaults to sorting by `started_at DESC`.

### Method
GET

### Endpoint
/api/runs

### Parameters
#### Query Parameters
- **page** (number) - Optional - Current page number (used to calculate offset: (page - 1) * 25)
- **status** (string) - Optional - Filter by status ('running', 'completed', 'failed')

### Request Example
GET /api/runs?page=1&status=running

### Response
#### Success Response (200)
- **items** (array) - List of pipeline run objects
- **total** (number) - Total count of runs matching criteria
- **offset** (number) - Current offset value
- **limit** (number) - Items per page (fixed at 25)

#### Response Example
{
  "items": [
    {
      "id": 1,
      "status": "completed",
      "started_at": "2023-10-27T10:00:00Z"
    }
  ],
  "total": 100,
  "offset": 0,
  "limit": 25
}
```

--------------------------------

### Pipeline Instance Traceability

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-1-codebase-architecture.md

Demonstrates how to link created database instances to a specific pipeline run, enabling auditability and state reconstruction.

```python
for instance in pipeline.get_extractions(Lane):
    run_instance = PipelineRunInstance(
        run_id=pipeline.run_id,
        model_type="Lane",
        model_id=instance.id,
    )
    session.add(run_instance)
```

--------------------------------

### Basic Pipeline Example (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/PLAN.md

Demonstrates a complete, working example of a basic LLM pipeline. This includes defining domain models, setting up the registry with FK order validation, creating instruction and context classes, defining extractions and steps, and configuring a strategy. It shows pipeline execution with caching and saving results to the database, following patterns from projects like logistics-intelligence.

```python
from yugen.pipeline import Pipeline
from yugen.providers.gemini import GeminiProvider
from yugen.core.registry import PipelineDatabaseRegistry
from yugen.core.steps import LLMStep
from yugen.core.instructions import Instruction
from yugen.core.context import Context
from yugen.core.extraction import Extraction
from yugen.core.strategy import Strategy
from pydantic import BaseModel
from sqlalchemy import create_engine

# Define domain models (example)
class RateCardItem(BaseModel):
    item_name: str
    price: float

# Define context class
class RateCardContext(Context):
    table_type: str = "rate_card"

# Define extraction
class RateCardExtraction(Extraction):
    def extract(self, text: str) -> list[RateCardItem]:
        # ... extraction logic ...
        return []

# Define instruction
class RateCardInstruction(Instruction):
    def get_extraction(self) -> Extraction:
        return RateCardExtraction()

# Define step
class RateCardStep(LLMStep):
    instruction = RateCardInstruction()

# Define strategy
class RateCardStrategy(Strategy):
    def can_handle(self, context: Context) -> bool:
        return isinstance(context, RateCardContext) and context.table_type == "rate_card"

    def get_step(self, context: Context) -> LLMStep:
        return RateCardStep()

# Setup database and registry
db_engine = create_engine("sqlite:///:memory:")
registry = PipelineDatabaseRegistry(db_engine)

# Setup provider
provider = GeminiProvider(api_key="YOUR_GEMINI_API_KEY")

# Create pipeline
pipeline = Pipeline(provider=provider, db_engine=db_engine)
pipeline.add_strategy(RateCardStrategy())

# Create context and run pipeline
context = RateCardContext()
result = pipeline.run(context=context)

# Save results
# pipeline.save(result)

```

--------------------------------

### Install shadcn UI components

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-35-step-detail-panel/PLAN.md

Commands to add the Sheet and Tabs components from the shadcn/ui library to the project.

```bash
npx shadcn@latest add sheet
npx shadcn@latest add tabs
```

--------------------------------

### Register Database Models

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Creating a registry class to manage database models and enforce dependency ordering for foreign keys.

```python
class TextClassifierRegistry(
    PipelineDatabaseRegistry,
    models=[
        Category,
        ClassifiedText,
    ]
):
    pass
```

--------------------------------

### Pipeline Execution Sequence

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-3-c4-architecture.md

Illustrates the step-by-step execution of the pipeline, including initialization, building the execution order, the main processing loop with caching and LLM calls, and final persistence and tracking. It highlights conditional logic like cache checks and optional consensus polling.

```text
PipelineConfig.execute(data, initial_context, use_cache, consensus_polling)
│
├─ 1. Initialize: Create execution context, load strategies
│
├─ 2. Build execution order: _build_execution_order()
│   └─ Determine steps from strategies respecting FK dependencies
│
├─ 3. Main loop: for step_index in range(max_steps):
│   │
│   ├─ A. Select strategy for this step
│   ├─ B. Get StepDefinition from strategy
│   ├─ C. Create step instance: step_class.create_step(pipeline)
│   │
│   ├─ D. Check cache: _find_cached_state(step, input_hash)
│   │   └─ If cached and valid → skip execution
│   │
│   ├─ E. If not cached:
│   │   │
│   │   ├─ E1. Get instructions: prepare_calls() → ExecuteLLMStepParams
│   │   ├─ E2. Call LLM: execute_llm_step(params) → results
│   │   ├─ E3. Validate instructions: process_instructions(results)
│   │   ├─ E4. Save execution state: _save_step_state()
│   │   │
│   │   ├─ E5. Transformation: PipelineTransformation.transform(data, instructions)
│   │   ├─ E6. Extraction: For each PipelineExtraction:
│   │   │    extract(results) → List[SQLModel]
│   │   │    store in self.extractions[model_class]
│   │   │
│   │   ├─ E7. Optional consensus: If consensus_polling:
│   │   │    _execute_with_consensus() → validate agreement
│   │   │
│   │   └─ E8. Optional action: Call action_after() if defined
│   │
│   └─ F. Move to next step
│
├─ 4. Persist: save(session, engine)
│   └─ For each extracted model, insert to DB in registry order
│
└─ 5. Track: Track created instances in PipelineRunInstance

```

--------------------------------

### Define Database Registry with Dependencies

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Ensures database models are registered in the correct dependency order to avoid foreign key errors.

```python
class MyRegistry(
    PipelineDatabaseRegistry,
    models=[
        ParentModel,  # No FK dependencies first
        ChildModel,   # Depends on ParentModel second
    ]
):
    pass
```

--------------------------------

### Initialize shadcn/ui

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-29-init-frontend-structure/research/step-1-frontend-stack-research.md

Initializes the shadcn/ui library using the CLI and defines the resulting configuration file. This setup is optimized for a client-side Vite application with specific directory aliases.

```bash
npx shadcn@latest init
```

```json
{
  "$schema": "https://ui.shadcn.com/schema.json",
  "style": "new-york",
  "rsc": false,
  "tsx": true,
  "tailwind": {
    "config": "",
    "css": "src/index.css",
    "baseColor": "neutral",
    "cssVariables": true,
    "prefix": ""
  },
  "aliases": {
    "components": "@/components",
    "utils": "@/lib/utils",
    "ui": "@/components/ui",
    "lib": "@/lib",
    "hooks": "@/hooks"
  },
  "iconLibrary": "lucide"
}
```

--------------------------------

### Enable Caching for Pipeline Steps

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Configures caching for a specific step definition to improve performance by hashing inputs and prompt versions.

```python
@step_definition
class ExpensiveStep(LLMStep, ...):
    use_cache = True  # Default: False
```

--------------------------------

### Define Pipeline Call Parameters

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Demonstrates how to define retry logic and custom indicators for 'not found' scenarios within the prepare_calls method.

```python
def prepare_calls(self, pipeline: PipelineConfig) -> List[Dict[str, Any]]:
    return [{
        "prompt": "...",
        "system_instruction": "...",
        "max_retries": 5,
        "not_found_indicators": ["not found", "no data"]
    }]
```

--------------------------------

### Reordering Strategies for Specificity (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/multi-strategy.md

Illustrates how to resolve the 'Multiple Strategies Match' problem by reordering the strategies list to prioritize more specific strategies over general ones.

```python
class MyStrategies(
    PipelineStrategies,
    strategies=[
        SpecificStrategy,    # Try specific first
        GeneralStrategy,     # Then general
    ]
):
    pass
```

--------------------------------

### Access Pipeline Data and Extractions

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Illustrates the three-tier data model for accessing pipeline context, intermediate data, and database model extractions.

```python
# 1. Context
pipeline.context["table_type"] = "lane_based"
pipeline.context["detected_category"] = "Technology"

# 2. Data
pipeline.data["TableDetectionStep"] = dataframe
pipeline.data["SanitizedStep"] = cleaned_text

# 3. Extractions
categories = pipeline.extractions[Category]
texts = pipeline.extractions[ClassifiedText]
```

--------------------------------

### Installation

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/api/index.md

Instructions for installing the llm-pipeline framework, including basic and development dependencies.

```APIDOC
## Installation

### Basic Installation

Install the core framework with required dependencies:

```bash
pip install llm-pipeline
```

### Optional Dependencies

#### Development Tools

Install testing and development dependencies:

```bash
pip install llm-pipeline[dev]
```
```

--------------------------------

### Initialize TextClassifierPipeline with Session

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Demonstrates how to instantiate a pipeline with a database session. The pipeline automatically wraps the session in a ReadOnlySession to ensure data integrity during execution.

```python
pipeline = TextClassifierPipeline(
    model='google-gla:gemini-2.0-flash-lite',
    session=session
)
```

--------------------------------

### Retrieve System and User Prompts

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-1-codebase-architecture.md

Demonstrates how to fetch system and user prompts using the prompt service, injecting variables for template rendering and error reporting.

```python
system_instruction = prompt_service.get_system_prompt(
    "my_step.system_instruction",
    variables={"context_var": "value"},
    variable_instance=system_vars,
)

user_prompt = prompt_service.get_user_prompt(
    "my_step.user_prompt",
    variables={"data": sanitized_data},
    variable_instance=user_vars,
)
```

--------------------------------

### Initialize and Use Prompt Model

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/api/prompts.md

Demonstrates how to instantiate a Prompt object with required metadata, template content, and variable definitions for storage in the system.

```python
from llm_pipeline.db.prompt import Prompt

prompt = Prompt(
    prompt_key="semantic_mapping.system_instruction",
    prompt_name="Semantic Mapping System Instruction",
    prompt_type="system",
    category="extraction",
    step_name="semantic_mapping",
    content="Extract {entity_type} from the document: {document_text}",
    required_variables=["entity_type", "document_text"],
    description="System instruction for semantic mapping extraction",
    version="1.2",
    is_active=True
)
```

--------------------------------

### LLM Pipeline UI Package Import Guard

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/master-27-cli-entry-point/research/step-1-codebase-structure-research.md

This Python code snippet from `llm_pipeline/ui/__init__.py` ensures that FastAPI is installed before importing UI-related components. It raises an `ImportError` if FastAPI is not found, guiding the user to install the necessary dependencies. This pattern is important for deferred imports in the CLI.

```python
try:
    import fastapi
except ImportError:
    raise ImportError(
        "llm_pipeline.ui requires FastAPI. "
        "Install with: pip install llm-pipeline[ui]"
    )
from llm_pipeline.ui.app import create_app
__all__ = ["create_app"]
```

--------------------------------

### Define Pipeline Database Models

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/guides/getting-started.md

Using SQLModel to create database schemas for storing classified data and categories, ensuring proper foreign key relationships.

```python
from sqlmodel import SQLModel, Field as SQLField
from typing import Optional

class Category(SQLModel, table=True):
    __tablename__ = "categories"
    id: Optional[int] = SQLField(default=None, primary_key=True)
    name: str = SQLField(index=True)
    description: Optional[str] = None

class ClassifiedText(SQLModel, table=True):
    __tablename__ = "classified_texts"
    id: Optional[int] = SQLField(default=None, primary_key=True)
    text: str
    category_id: int = SQLField(foreign_key="categories.id")
    confidence: float
```

--------------------------------

### Auto-SQLite Initialization Configuration (Python)

Source: https://github.com/yugen-au/llm-pipeline/blob/main/docs/tasks/completed/adhoc-20260210-docs-generation/research/step-1-codebase-architecture.md

Explains how the system automatically initializes an SQLite database. It defaults to `.llm_pipeline/pipeline.db` in the current working directory but can be overridden via an environment variable or by explicitly providing a SQLAlchemy engine.

```python
# Default: .llm_pipeline/pipeline.db in cwd
db_path = Path.cwd() / ".llm_pipeline" / "pipeline.db"

# Override via environment
export LLM_PIPELINE_DB="/path/to/custom.db"

# Or explicit engine
pipeline = MyPipeline(engine=create_engine("postgresql://..."))
```
Tokens:
Raw