llmflow-core

Deterministic LLM workflow engine for file-defined, schema-validated pipelines.

Overview

llmflow-core executes explicit workflow DAGs from YAML and prompt files. It is designed for predictable execution, strict output contracts, and replayable artifacts.

Core guarantees:

• Stable topological execution order
• Fail-fast behavior on first step error
• JSON-schema validation for LLM outputs
• Run artifacts for audit and replay

Why this exists

Production LLM pipelines often fail on three basics:

• Step order is implicit and hard to reason about
• Outputs drift from expected structure
• Runs are difficult to reproduce and debug

llmflow-core addresses this with file-defined workflows, strict validation, and deterministic run traces.

Installation

python -m pip install -e .

Install with test dependencies:

python -m pip install -e .[dev]

Quickstart

1) Run with Python API

from llmflow import MockProvider, RunConfig, Runner, Workflow

workflow = Workflow.load("examples/blog_pipeline/workflow.yaml")

runner = Runner(
    provider=MockProvider(default_output="{}", strict=False),
    config=RunConfig(artifacts_dir=".runs", provider_name="mock"),
)

result = runner.run(
    workflow,
    inputs={"topic": "Deterministic AI", "audience": "Engineering managers"},
)

print(result.outputs)
print(result.run_dir)

Note:

• MockProvider(default_output=...) returns the same JSON for every LLM step.
• If your workflow has different per-step schemas, use per-prompt mock responses (see Example Workflow below).

2) Run with CLI

llmflow run examples/blog_pipeline/workflow.yaml \
  --input topic="Deterministic AI" \
  --input audience="Engineering managers" \
  --mock-output '{"title":"Draft","summary":"S","body":"B"}'

3) Inspect and replay

llmflow graph examples/blog_pipeline/workflow.yaml
llmflow replay .runs/run_YYYYMMDD_HHMMSS_<shortid>

CLI

The CLI is a thin wrapper over the library API.

Commands:

• llmflow run <workflow.yaml> --input key=value ...
• llmflow graph <workflow.yaml>
• llmflow replay <run_dir>

Example:

llmflow run examples/blog_pipeline/workflow.yaml \
  --input topic="Deterministic AI" \
  --input audience="Engineering managers" \
  --mock-output '{"title":"Draft","summary":"S","body":"B"}'

llmflow graph examples/blog_pipeline/workflow.yaml
llmflow replay .runs/run_YYYYMMDD_HHMMSS_<shortid>

CLI mock behavior:

• --mock-output and --mock-output-file are applied to all LLM steps in the run.
• For workflows with heterogeneous per-step schemas, prefer Python API tests with prompt-specific mock responses.

Example workflow

Repository example: examples/blog_pipeline

Contents:

• examples/blog_pipeline/workflow.yaml
• examples/blog_pipeline/prompts/outline.md
• examples/blog_pipeline/prompts/critique.md
• examples/blog_pipeline/prompts/revise.md
• examples/blog_pipeline/schemas/outline.json
• examples/blog_pipeline/schemas/critique.json
• examples/blog_pipeline/schemas/final_article.json
• examples/blog_pipeline/tools.py

The end-to-end deterministic example run is validated by:

• tests/test_examples_blog_pipeline.py

Workflow YAML format

Minimal shape:

workflow:
  name: blog_post_pipeline
  version: "1.0"

inputs:
  topic:
    type: string
  audience:
    type: string

steps:
  - id: outline
    type: llm
    prompt: prompts/outline.md
    output_schema: schemas/outline.json
    llm:
      model: mock-model
      temperature: 0

  - id: critique
    type: llm
    depends_on: [outline]
    prompt: prompts/critique.md
    output_schema: schemas/critique.json
    llm:
      model: mock-model
      temperature: 0

outputs:
  article: critique

Rules:

• workflow.name and workflow.version are required.
• inputs is a mapping of input names to type declarations.
• Each step needs a unique id and type.
• depends_on must reference existing step ids.
• outputs maps final output names to step ids.
• For llm steps, prompt, output_schema, and llm.model are required.

Replay

Replay reconstructs outputs from recorded artifacts and verifies they match the recorded outputs.json exactly.

from llmflow import replay

result = replay(".runs/run_YYYYMMDD_HHMMSS_<shortid>")
print(result.outputs)

Optional workflow override:

result = replay(
    ".runs/run_YYYYMMDD_HHMMSS_<shortid>",
    workflow_path="examples/blog_pipeline/workflow.yaml",
)

Artifacts

Each run writes a folder under .runs/:

.runs/
  run_YYYYMMDD_HHMMSS_<shortid>/
    metadata.json
    inputs.json
    outputs.json
    steps/
      <step_id>/
        output.json
        rendered_prompt.md
        llm_call.json
    logs.txt

metadata.json includes:

• artifacts_version
• engine version
• workflow name, version, and hash
• provider name
• execution order
• prompt hashes and step output hashes
• timestamps

Typical step artifacts:

• steps/<step_id>/output.json: Validated step output payload
• steps/<step_id>/rendered_prompt.md: Rendered prompt text for LLM steps
• steps/<step_id>/llm_call.json: Provider request/response metadata for LLM steps

Extending the engine

Providers

Implement Provider.call(request) -> ProviderResponse and pass the provider to Runner.

from llmflow.providers import Provider, ProviderRequest, ProviderResponse


class StaticProvider(Provider):
    def call(self, request: ProviderRequest) -> ProviderResponse:
        return ProviderResponse(
            model=request.model,
            output_text='{"title":"Draft","summary":"S","body":"B"}',
            raw={"provider": "static"},
        )

Tools

Register Python functions in ToolRegistry. Tool functions accept merged step inputs and must return a dict.

from llmflow.registry import ToolRegistry

tools = ToolRegistry()
tools.register("summarize_topic", lambda inputs: {"topic_slug": inputs["topic"].lower()})

Validators

Register custom validators in ValidatorRegistry. Validator functions accept merged step inputs and should return True/None on success, or False on failure.

from llmflow.registry import ValidatorRegistry

validators = ValidatorRegistry()
validators.register("has_summary", lambda inputs: bool(inputs.get("summary")))

Custom steps

Register custom step classes in StepRegistry when you need a new execution primitive beyond llm, tool, and validate.

Testing

Run all tests:

pytest

Run tests with coverage for core modules:

pytest --cov=llmflow --cov-report=term-missing

Determinism and replay checks:

pytest tests/test_replay.py tests/test_examples_blog_pipeline.py

Run a local command equivalent to CI:

python -m pip install -e .[dev]
pytest --cov=llmflow --cov-report=term-missing

Current status

Implemented through Phase 11:

• Core workflow loading and validation
• Graph ordering and cycle detection
• LLM/tool/validate steps
• Artifacts and metadata
• Runner and replay
• CLI (run, graph, replay)
• Example workflow (examples/blog_pipeline)
• CI workflow and coverage command (pytest --cov=llmflow)