quartermaster-graph

Framework-agnostic graph schema for defining AI agent workflows as directed graphs.

Features

• Pydantic-based models: Agent, GraphSpec, GraphNode, GraphEdge with full validation
• GraphBuilder fluent API for programmatic graph construction (Graph convenience alias)
• 40+ node types covering LLM, control flow, data, user interaction, memory, and utility
• YAML/JSON serialization with round-trip fidelity
• Graph validation: start/end nodes, cycle detection, orphan detection, edge label checks
• Graph traversal: topological sort, path finding, successor/predecessor queries
• Pre-built templates: simple chat, decision tree, multi-agent supervisor, and more
• Typed metadata schemas for each node type

New in v0.5.0

• program_runner(program=<callable>) -- pass a @tool()-decorated function directly instead of its name string; the graph builder auto-registers it. Parity with .agent(tools=[...]).

Installation

pip install quartermaster-graph

Dependencies: pydantic>=2.0, pyyaml>=6.0

Quick Start

Build a Graph with the Fluent API

from quartermaster_graph import Graph

graph = (
    Graph("Customer Support Agent")
    .user("How can I help you?")
    .instruction(
        "Classify intent", model="gpt-4o", system_instruction="Classify the user's intent."
    )
    .decision("Route by intent", options=["billing", "technical", "general"])
    .on("billing")
    .instruction("Handle billing", model="gpt-4o")
    .end()
    .on("technical")
    .instruction("Handle technical", model="gpt-4o")
    .end()
    .on("general")
    .instruction("Handle general", model="gpt-4o")
    .end()
    .end()
)

Graph is a convenience alias for GraphBuilder. Both .build() and .to_graph() return a GraphSpec:

from quartermaster_graph import GraphBuilder

builder = GraphBuilder("My Agent")
builder.instruction("Process", model="gpt-4o").end()

spec = builder.build()  # returns GraphSpec
spec = builder.to_graph()  # same thing

Load a Graph from YAML

# agent.yaml
agent_id: "550e8400-e29b-41d4-a716-446655440000"
start_node_id: "00000000-0000-0000-0000-000000000001"
nodes:
  - id: "00000000-0000-0000-0000-000000000001"
    type: "Start1"
    name: "Start"
  - id: "00000000-0000-0000-0000-000000000002"
    type: "User1"
    name: "Input"
  - id: "00000000-0000-0000-0000-000000000003"
    type: "Instruction1"
    name: "Process"
    metadata:
      llm_system_instruction: "Analyze user input"
      llm_model: "gpt-4o"
  - id: "00000000-0000-0000-0000-000000000004"
    type: "End1"
    name: "End"
edges:
  - source_id: "00000000-0000-0000-0000-000000000001"
    target_id: "00000000-0000-0000-0000-000000000002"
  - source_id: "00000000-0000-0000-0000-000000000002"
    target_id: "00000000-0000-0000-0000-000000000003"
  - source_id: "00000000-0000-0000-0000-000000000003"
    target_id: "00000000-0000-0000-0000-000000000004"

from quartermaster_graph import from_yaml

with open("agent.yaml") as f:
    spec = from_yaml(f.read())

API Reference

Core Models

AgentGraph and AgentVersion both exist as deprecated backward-compatibility aliases for GraphSpec.

GraphSpec Methods

GraphBuilder Methods

Start / End

LLM Nodes

| .summarize(name, model, provider, temperature, system_instruction, **kwargs) | GraphBuilder | LLM condenses conversation history | | .agent(name, model, provider, system_instruction, tools, max_iterations, **kwargs) | GraphBuilder | Agentic loop with tools, up to max_iterations | | .vision(name, model, provider, system_instruction, **kwargs) | GraphBuilder | Image vision/analysis node | | .merge(name, model, provider, temperature, system_instruction, prefix_message, suffix_message, **kwargs) | GraphBuilder | LLM merges parallel branch outputs |

Control Flow

User Interaction

Data Nodes

Memory

Composition / Utility

Build / Export

Node Configuration Kwargs

All builder methods accept optional keyword arguments:

Loops and Cycles

Use connect() to create back-edges for iterative flows:

agent = (
    Graph("Refiner")
    .user("Input")
    .var("Init", variable="round", expression="1")
    .text("Header", template="Round {{round}}", traverse_in=TraverseIn.AWAIT_FIRST)
    .instruction("Process", system_instruction="Improve the text")
    .var("Increment", variable="round", expression="round + 1")
    .if_node("Done?", expression="round > 3")
    .on("true")
    .text("Done", template="Complete")
    .end()
    .on("false")
    .text("Continue", template="", show_output=False)
    .end()
    .end()
)
agent.connect("Continue", "Header", label="loop")
graph = agent.build(validate=False)  # skip validation for intentional cycles

Validation

from quartermaster_graph import validate_graph

errors = validate_graph(agent_graph)
for err in errors:
    print(f"[{err.severity}] {err.code}: {err.message}")

Validation checks:

• Exactly one Start node, at least one End node
• All edge source/target IDs reference existing nodes
• Orphan detection (unreachable from Start)
• Cycle detection (DAG property)
• Decision/If/Switch nodes have proper edge labels

Serialization

from quartermaster_graph import to_json, from_json, to_yaml, from_yaml, json_schema

# JSON round-trip
data = to_json(agent_graph)
restored = from_json(data)

# YAML round-trip
yaml_str = to_yaml(agent_graph)
restored = from_yaml(yaml_str)

# JSON Schema for cross-language validation
schema = json_schema()

Traversal Utilities

from quartermaster_graph import (
    get_start_node,
    get_successors,
    get_predecessors,
    get_path,
    topological_sort,
    find_merge_points,
    find_decision_points,
)

start = get_start_node(agent_graph)
ordered = topological_sort(agent_graph)
path = get_path(agent_graph, start_id, end_id)

Enums

Integration with Sibling Packages

# Build a graph (quartermaster-graph)
from quartermaster_graph import Graph

graph = Graph("Agent").user("Input").instruction("Process").end()

# Execute it (quartermaster-engine)
from quartermaster_engine import FlowRunner

runner = FlowRunner(graph=graph.build(), node_registry=registry)
result = runner.run("Hello!")

Contributing

See CONTRIBUTING.md for guidelines.

License

Apache License 2.0 -- see LICENSE for details.