workflow-ai

Multi-phase workflow orchestration, storage, and tracking for LLM agent pipelines.

Zero external dependencies at runtime -- uses only the Python standard library.

Installation

pip install workflow-ai

For development (includes pytest):

pip install workflow-ai[dev]

Quickstart

Define and execute a workflow

import asyncio
from workflow_ai import SimpleWorkflowEngine, InMemoryWorkflowStorage, WorkflowDefinition

storage = InMemoryWorkflowStorage()
engine = SimpleWorkflowEngine(storage)

workflow = WorkflowDefinition(
    id="wf-1",
    name="review-pipeline",
    description="Multi-phase code review",
    phases=["analyze", "review", "synthesize"],
    config={"depth": "thorough"},
)

result = asyncio.run(engine.execute(workflow))
print(result.status)           # "completed"
print(result.phases_completed)  # ["analyze", "review", "synthesize"]
print(result.duration_ms)       # elapsed time in ms

Custom phase handler

async def handler(phase, args, config):
    if phase == "analyze":
        return {"findings": ["bug in auth.py", "unused import"]}
    elif phase == "review":
        return {"verdict": "needs-work"}
    return {"status": "done"}

engine = SimpleWorkflowEngine(storage, phase_handler=handler)
result = asyncio.run(engine.execute(workflow))
print(result.outputs["analyze"])  # {"findings": [...]}

Decorators

from workflow_ai import as_workflow_phase, with_workflow_tracking, InMemoryWorkflowStorage

storage = InMemoryWorkflowStorage()

@with_workflow_tracking(workflow_name="my-pipeline", storage=storage)
async def run_pipeline(data: str) -> str:
    """Execute the full pipeline."""
    return f"processed: {data}"

@as_workflow_phase(phase_name="analyze", storage=storage, execution_id="exec-1")
async def analyze(code: str) -> dict:
    return {"issues": 0}

Resume and status

# Check status
status = asyncio.run(engine.status(result.run_id))
print(status.phase, status.progress)  # "synthesize", 1.0

# Resume an interrupted run
resumed = asyncio.run(engine.resume(result.run_id))
print(resumed.status)  # "completed"

API Overview

Types

• WorkflowDefinition(id, name, description, phases, config) -- workflow blueprint
• WorkflowResult(workflow_id, run_id, status, phases_completed, outputs, duration_ms) -- execution outcome
• WorkflowStatus(run_id, phase, progress, started_at) -- live progress snapshot
• WorkflowExecution(id, workflow_name, task_description, total_workers, total_duration_ms, outcome) -- persisted record
• WorkerResult(id, execution_id, model, content, latency_ms, tokens_used, success) -- individual worker output

Protocols

• WorkflowStorageBackend -- runtime-checkable protocol for workflow persistence
• WorkflowEngine -- runtime-checkable protocol for workflow execution engines

Classes

• InMemoryWorkflowStorage -- dict-backed storage for testing and development
• SimpleWorkflowEngine -- iterates through phases, persists results, supports resume

Decorators (ADR-0006)

from workflow_ai import as_workflow_phase, with_workflow_tracking

@as_workflow_phase(phase_name="analyze", storage=s, execution_id="e1")
async def analyze(code: str) -> dict:
    return {"result": "clean"}

@with_workflow_tracking(workflow_name="pipeline", storage=s)
async def pipeline() -> str:
    return "done"

FlossWare Engineering Standards

This package complies with FlossWare/engineering-standards:

ADR	Title	How
ADR-0001	Explicit Opt-In	Nothing activates automatically; decorators require explicit application
ADR-0006	Cross-Cutting Decorators	@as_workflow_phase, @with_workflow_tracking
ADR-0008	Free-First	Zero external runtime dependencies (stdlib only)
ADR-0009	Core Principles	Modular, composable, contracts over implementations
ADR-0017	Agent-Neutral	Works with any agent runtime via Protocols
ADR-0020	Capability-Protocol Separation	Transport-independent workflow capabilities

See STANDARDS.md for detailed compliance notes.

License

MIT