PromptGraph

Flowchart-driven LLM control through YAML state machines.

Features

• YAML-First - Define workflows declaratively, version control them
• Provider Agnostic - Works with Anthropic, OpenAI, Google, or any completion function
• Framework Optional - Use standalone OR integrate with Vel
• Conditional Branches - Route based on LLM output with safe expression evaluation
• Loop Control - Prevent infinite loops with max_loops limits
• Full Tracing - Debug with complete execution history
• Mermaid Export - Visualize workflows as diagrams

Installation

pip install promptgraph

Quick Start

Define a Workflow

# research.yaml
name: research_agent

inputs:
  topic:
    type: string
    required: true

states:
  research:
    start: true
    prompt: |
      Research {topic} and list 3 key findings.
    output: findings
    transitions:
      - to: summarize

  summarize:
    prompt: |
      Summarize these findings:
      {findings}
    output: summary
    transitions:
      - to: review

  review:
    prompt: |
      Review this summary. Respond with JSON:
      {"approved": true/false, "feedback": "..."}
    output: review_result
    parse: json
    transitions:
      - to: done
        when: "approved == true"
      - to: summarize
        when: "approved == false"
        max_loops: 3

  done:
    prompt: "Finalize: {summary}"
    output: final_report
    end: true

outputs:
  - final_report

Execute with Any Provider

from promptgraph import Workflow
import anthropic

# Load workflow
workflow = Workflow.from_yaml("research.yaml")

# Create completion function
client = anthropic.Anthropic()

def complete(prompt: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

# Run
result = workflow.run(
    complete=complete,
    inputs={"topic": "quantum computing"}
)

print(result.outputs["final_report"])
print(f"Completed in {len(result.trace.steps)} steps")

Or Use Built-in Helpers

from promptgraph import Workflow
from promptgraph.integrations.native import anthropic_complete
import anthropic

workflow = Workflow.from_yaml("research.yaml")
client = anthropic.Anthropic()

result = workflow.run(
    complete=anthropic_complete(client),
    inputs={"topic": "quantum computing"}
)

Vel Integration

PromptGraph integrates seamlessly with Vel:

from vel import Agent
from promptgraph.integrations.vel import WorkflowMixin

class ResearchAgent(WorkflowMixin, Agent):
    workflow = "research.yaml"

agent = ResearchAgent(
    id='research:v1',
    model={'provider': 'anthropic', 'model': 'claude-sonnet-4-20250514'}
)

result = await agent.run_workflow({"topic": "AI safety"})

Streaming Execution

async for event in workflow.run_stream(complete, inputs):
    match event["type"]:
        case "state_enter":
            print(f"→ {event['state']}")
        case "state_output":
            print(f"  Output: {event['output'][:50]}...")
        case "transition":
            print(f"  → {event['to']}")
        case "workflow_complete":
            print(f"✓ Done!")

Mermaid Visualization

print(workflow.to_mermaid())

Output:

stateDiagram-v2
    [*] --> research
    research --> summarize
    summarize --> review
    review --> done: approved == true
    review --> summarize: approved == false
    done --> [*]

Loading

YAML Schema

States

states:
  my_state:
    start: true # Mark as starting state
    end: true # Mark as ending state
    prompt: | # Prompt template with {variables}
      Do something with {input_var}
    output: result_var # Variable name for output
    parse: json # Parse mode: text, json, yaml
    transitions:
      - to: next_state
        when: "condition == true" # Optional condition
        max_loops: 3 # Optional loop limit

Conditions

Safe expression evaluation (no eval()):

# Comparisons
when: "approved == true"
when: "score >= 0.8"
when: "status != 'error'"

# Boolean logic
when: "approved and score > 0.5"
when: "not failed"

# Nested access
when: "result.approved == true"

Inputs

inputs:
  topic:
    type: string
    required: true
    description: What to research

  depth:
    type: string
    default: "moderate"
    enum: [shallow, moderate, deep]

API Reference

Workflow

class Workflow:
    @classmethod
    def from_yaml(cls, source: str | Path) -> Workflow

    def run(self, complete: Callable, inputs: dict = None) -> WorkflowResult
    async def run_async(self, complete: Callable, inputs: dict = None) -> WorkflowResult
    async def run_stream(self, complete: Callable, inputs: dict = None) -> AsyncIterator

    def to_mermaid(self, highlight_state: str = None) -> str

WorkflowResult

@dataclass
class WorkflowResult:
    success: bool
    outputs: dict[str, Any]
    context: ExecutionContext
    trace: ExecutionTrace
    error: str | None

ExecutionTrace

result.trace.steps          # List of TraceStep
result.trace.to_json()      # Export to JSON
result.trace.duration_ms    # Total execution time

Examples

See the examples/ directory:

• simple_example.py - Basic usage
• anthropic_example.py - With Anthropic SDK
• openai_example.py - With OpenAI SDK
• vel_example.py - Vel integration
• workflows/ - Sample workflow YAML files

Development

git clone https://github.com/rscheiwe/promptgraph
cd promptgraph
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy promptgraph --strict

# Linting
ruff check promptgraph

License

MIT

Credits

Inspired by fladdict/promptgraph prompting technique.

Built to integrate with Vel agent runtime.

promptgraph