API Reference

SuperDialog ships two engines, one entry point. DialogMachine is the recommended way in; it drives either engine behind the same Agent protocol - the Playbook engine by default, the legacy graph runtime with engine="flow".

DialogMachine (unified entry point)

Construction

DialogMachine(
    source: Flow | FlowSet | Playbook | str | dict,  # path, parsed dict, or object
    llm: str | None = None,             # model URI (Talker, and Director unless split)
    tools: list[Tool] | None = None,    # any Tool; both engines
    memory: ContextStore | None = None, # graph-only; default in-memory
    config: dict | None = None,         # graph-only: max_tokens, temperature, etc.
    traversal_dir: str | Path | None = None,  # graph-only: auto-save traversal JSON
    adapter: str = "toolcall",          # graph-only adapter selection
    *,
    engine: str = "auto",               # "auto" | "playbook" | "flow"
    director_llm: str | None = None,    # Playbook: strong-Director override
)

source is the artifact: a path string, a parsed dict, or an object. The unified loader auto-detects full playbooks, the simple format, and legacy flow JSON (compiled transparently), so you never route by format. Engine selection (engine="auto", the default): a Flow / FlowSet object keeps the legacy graph engine (back-compat); a Playbook object, a path string, or a parsed dict runs the Playbook engine. Override with:

engine="playbook" - force the Playbook engine (compiling a flow if needed)
engine="flow" - force the legacy graph runtime (ValueError on a Playbook)

In Playbook mode, llm is the Talker and the Director unless director_llm splits them; a missing llm raises a clear ValueError. Graph-only methods (switch_flow, seed, load_flow_state) raise NotImplementedError in Playbook mode, and flow_state returns None.

from superdialog import DialogMachine

agent = DialogMachine("booking.yaml", llm="openai/gpt-4.1-mini")          # Playbook (default)
agent = DialogMachine("booking.yaml", llm="anthropic/claude-haiku-4-5",
                      director_llm="anthropic/claude-opus-4-7")           # split Talker/Director
dm = DialogMachine(Flow.load("kyc.json"), llm="...", engine="flow")        # legacy graph

`turn`

async turn(text, context=None, stream=False) -> Turn | AsyncIterator[StreamChunk]

The primary method. Always async - drive it from asyncio.run(...) or any async runtime.

reply = await agent.turn("hello")
print(reply.text)

stream = await agent.turn("hello", stream=True)   # await the coroutine, then iterate
async for chunk in stream:
    print(chunk.text, end="")

Turn carries text, tool_calls, and metadata. On the Playbook engine, metadata includes checkpoint, version, ended, and (on terminal checkpoints) outcome.

Streaming is real on the Playbook engine. PlaybookAgent yields live provider tokens as the Talker produces them. The legacy graph engine resolves the turn in one shot and surfaces whitespace-delimited chunks (the StreamChunk(text, done, turn) shape is stable).

`start` / `reset` / `set_llm` / `assist`

opening = await agent.start()       # agent greets first, no user input
agent.reset()                       # clear memory, restart from the beginning
agent.set_llm("anthropic/claude-haiku-4-5")   # hot-swap; applies next turn
agent.assist("Customer is upset. Be especially empathetic.")  # system inject

`switch_flow` (graph engine only)

dm.switch_flow("escalation")                      # FlowSet, engine="flow"
dm.switch_flow("billing", preserve_memory=True)

Raises NotImplementedError on the Playbook engine - use multiple journeys and advance rules instead.

Creating agents

`generate_simple_playbook` (default)

async generate_simple_playbook(prompt, llm, *, max_attempts=3) -> str

Bootstrap a validated simple-format playbook from a description. The output is parsed and compiled before return, so a successful return is always loadable. CLI equivalent: superdialog generate "<prompt>" --output playbook.yaml.

from superdialog.playbook import generate_simple_playbook

yaml_text = await generate_simple_playbook(
    "An agent that books demo calls and captures a day and time.",
    director,                       # any CompletesLLM
)
open("playbook.yaml", "w").write(yaml_text)

`create_dialog_flow` (legacy)

async create_dialog_flow(prompt, llm, **kwargs) -> Flow

Bootstrap a legacy flow graph from a prompt. Prefer generate_simple_playbook for new agents; generated flows still run on the Playbook engine by default. The llm is used only at construction.

Playbook engine

Import from superdialog.playbook. Checkpoint-compound journeys: a fast Talker streams every spoken turn while an async Director extracts slots, judges advancement, and runs tools over an append-only event log.

`Playbook`

The authored artifact. Load it - all loaders auto-detect the simple format and legacy flow JSON:

from superdialog.playbook import Playbook

pb = Playbook.load(path)         # YAML for .yaml/.yml, else JSON
pb = Playbook.from_yaml(text)
pb = Playbook.from_json(text)

Top-level fields: persona, journeys (≥1), dispatch, tools, pipelines, handlers, interrupts, policies, middleware, env, views, initial. Validation runs on construction and raises on unknown checkpoint/pipeline/tool refs, duplicate ids, undeclared requires keys, and the reserved pipeline result key.

`PlaybookAgent`

The engine behind the Agent protocol - drop it into SessionWorker and every host adapter unchanged. Use it directly when you want explicit Talker/Director LLMs or a custom HTTP executor.

from superdialog.playbook import Playbook, PlaybookAgent, httpx_http

agent = PlaybookAgent(
    playbook=Playbook.load("booking.yaml"),
    talker_llm=talker,          # StreamsLLM
    director_llm=director,      # CompletesLLM
    http=httpx_http,            # HttpFn
    python_tools=None,          # dict[str, PythonToolFn] | None
    token_budget=4000,          # Talker view budget (estimated tokens)
    barrier_timeout=0.4,        # hard gate: wait this long for the Director
    hold_timeout=None,          # then filler + this much more before degrading
)

async turn(text, *, stream=False) - Agent protocol. With stream=True the iterator yields live provider tokens, then any pass-through say_verbatim lines, then the done=True chunk.
Barge-in safety - aborting the stream interrupts speech, not the state machine: the Director runs to completion in a shielded scope.
assist, chat_ctx / load_chat_ctx, event_log / load_event_log, runtime - system inject, transcript view, the lossless log, and the PlaybookRuntime.

The artifact model

`Checkpoint`

Field	Meaning
`id`	Unique within its journey; referenced as `"journey.id"`
`goal`	What “done” means; shown to Talker and Director
`slots`	`dict[str, SlotSpec]` to extract while here
`guidance`	Talker prose; Jinja over `{slots, views, results}`
`say_verbatim`	Exact line; bypasses the Talker LLM
`never_say`	Hard prohibitions injected into the Talker view
`exit_say`	One-shot line spoken on the turn that leaves this checkpoint via a director rule (post-capture pitch); never on interrupt/policy advances. Simple format authors it as `then_say`
`advance_when`	Ordered `AdvanceRule` list (outcome gates)
`gate`	`soft` (default) or `hard` (barrier; `requires` need confirmed slots)
`auto`	Speak verbatim once, then advance without user input
`pipeline`	Pipeline run once on entry; routes on `pipeline.ok` / `pipeline.failed`
`on_failure`	Route on pipeline failure or turn-budget exhaustion
`terminal` + `outcome`	Entering ends the session with this outcome label
`turn_budget`	User turns before a wrap-up steering note

`SlotSpec`

type (str int float bool date enum array object), required, values (enum), authoritative (tool / pipeline / expr-set writes only), invalidates (slots/results cleared when this value changes), description.

`AdvanceRule`

when (prose for judge: llm, expr text for judge: expr), judge (llm/expr), to (target ref), requires (slots that must be filled at soft gates, confirmed at hard gates), set (confirmed slot writes on advance). Verdict-extracted slots are provisional at hard gates - a single verdict can never confirm its own requires in one shot.

`ToolSpec` / `PipelineSpec`

ToolSpec: type (http/python), method/url/headers/body (sandboxed Jinja over {slots, env, results}), store_response_as, env_updates, run_once, when, timeout, args. PipelineSpec: ordered steps, each with typed on: {ok, failed, http_<code>} branches and a capped RetrySpec (retry ≤ 10). Tool failures are recorded as failed result events - data, never a crash.

Protocols: `CompletesLLM`, `StreamsLLM`, `HttpFn`, `PythonToolFn`

class CompletesLLM(Protocol):            # Director seam
    async def complete(self, messages: list[dict[str, str]], **kwargs) -> str: ...

class StreamsLLM(Protocol):              # Talker seam
    def stream(self, messages: list[dict[str, str]], **kwargs) -> AsyncIterator[str]: ...

HttpFn = Callable[..., Awaitable[tuple[int, Any]]]   # (method=, url=, headers=,
                                                     #  body=, timeout=) -> (status, json)

class PythonToolFn(Protocol):            # registered via python_tools={id: fn}
    async def __call__(self, args: dict[str, Any], state: ConversationState) -> Any: ...

httpx_http is the production HttpFn backed by httpx. Any superdialog.llm.LLMProvider (the litellm-backed one behind model URIs) adapts to the Talker/Director protocols in a few lines - see Embedding Guides.

The expr language

Used by judge: expr rules, ToolSpec.when, and Playbook.views. A safe, LLM-free, restricted Python expression over state:

slots.city == "Pune"                    # slot value; None when unset
results.hold.ok                         # tool result: ok / status / data / error
results.search.status == 404
env.BOOKING_API                         # env lane (not available in views)
pipeline.ok                             # pipeline-owned checkpoints only
len(results.search.data.slots) > 0
first(pluck(results.search.data.slots, "time"))

Helpers (the only callables): len, first, last, pluck, unique, min, max, any, all. Forbidden (raises ExprError): arithmetic, comprehensions, lambdas, dict literals, f-strings, any _-prefixed name, non-whitelisted calls, expressions over 4096 chars. Missing values evaluate to None (falsy), never an exception.

Migrating flows

from superdialog import Flow
from superdialog.playbook import compile_flow, coverage_report

flow = Flow.load("golf_booking.json")
pb = compile_flow(flow)                # single-journey "main" Playbook
report = coverage_report(flow, pb)     # CoverageReport - the lossless proof
assert not report.unmapped_nodes
assert not report.unmapped_edges
assert not report.unmapped_actions

compile_flow is lossless by construction:

Legacy construct	Becomes
Conversational nodes	Checkpoints in journey `"main"`
Tool-free computational nodes	Folded into their sources’ advance rules
Tool-bearing computational chains	A `PipelineSpec` + synthetic checkpoint
Hub routers (≥4-exit)	`dispatch` entries merged into inbound checkpoints
Silence nodes	`policies.silence`
Token-expiry global edge + refresh	`middleware`
Other global edges	`interrupts`
Webhook/timer system nodes	`handlers`
`global_actions`	`tools`, 1:1

Deterministic edge conditions (X.success == true, X.status == 404) compile to judge: expr; everything else stays judge: llm with the prose verbatim. coverage_report lists anything that didn’t map (any unmapped_* entry is a compiler bug) - run it in CI.

`EventLog` and `ConversationState`

The event log is the single source of truth; state is a pure fold over it.

from superdialog.playbook import ConversationState, EventLog

text = agent.event_log.to_jsonl()                 # persist (JSONL, one event/line)
agent.load_event_log(EventLog.from_jsonl(text))   # lossless restore
state = ConversationState.fold(agent.event_log, playbook)

EventLog is append-only with contiguous versions from 1. Events are frozen, discriminated on type: utterance, slot_write, advance, steering_note, tool_call, tool_result, env_write, scratchpad, summary, external, degraded, session_end. ConversationState.fold derives checkpoint_id, slots (value + provisional/confirmed status), transcript, env, tool_results, ended, outcome, and helpers slot_value, confirmed, filled.

`replay` and the eval bridge

from superdialog.playbook import replay, run_session, run_eval, PersonaSpec

report = await replay(log, playbook, director_llm)   # pure: never mutates log
report.stable                                        # every decision matched?

metrics = await run_session(agent, persona, user_llm)
report = await run_eval(playbook_factory=make_agent, personas=personas,
                        user_llm=user_llm, n=1)
report.completion_rate, report.mean_slot_accuracy

replay re-runs the Director over recorded utterances under a (possibly edited) playbook and diffs decisions - regression evidence for prompt or model changes. The eval bridge scores persona self-play from the same logs.

Sessions (v0.2)

Sessions add lifecycle and persistence on top of any Agent-protocol brain - PlaybookAgent (default) and the legacy DialogMachine alike.

Agent Protocol

class Agent(Protocol):
    async def turn(text: str, *, stream: bool = False) -> TurnResult | AsyncIterator[StreamChunk]
    def assist(text: str) -> None
    @property
    def chat_ctx(self) -> ChatContext
    def load_chat_ctx(ctx: ChatContext) -> None

`SessionWorker` / `SessionHandle`

from superdialog import DialogMachine, SessionWorker, InMemorySessionStore

worker = SessionWorker(
    agent_factory=lambda: DialogMachine("booking.yaml", llm="openai/gpt-4.1-mini"),
    store=InMemorySessionStore(),
    lock_backend=None,        # default: AsyncioLockBackend
    max_sessions=1000,
)

async with worker.acquire("user-42") as h:
    result = await h.turn("hello")
    h.assist("Customer sounds upset; be empathetic.")

agent_factory is called once per new session; acquire(session_id) loads or creates the session, locks it for the block, and persists on exit. Different ids run in parallel; same id serialises. SessionHandle exposes turn, assist, state.

Session stores and lock backends

Store	Ships	Use case
`InMemorySessionStore`	✅ v0.2	Single-process; state lives for process lifetime
`NullSessionStore`	✅ v0.2	Voice calls; no persistence wanted
`RedisSessionStore`, `FileSessionStore`, `SQLiteSessionStore`	🔜 planned	Distributed / durable

Backend	Ships	Use case
`AsyncioLockBackend`	✅ v0.2	Single-process
`RedisLockBackend`	🔜 planned	Multi-process / distributed

SessionWorker’s SessionRecord persists chat_ctx / flow_state only. For durable Playbook-engine resume, persist agent.event_log.to_jsonl() and restore with load_event_log - see Sessions.

Other agent brains

from superdialog import LLMAgent, LangChainAgent, SessionWorker, InMemorySessionStore

worker = SessionWorker(
    agent_factory=lambda: LLMAgent(llm="openai/gpt-4.1-mini", system_prompt="Be helpful."),
    store=InMemorySessionStore(),
)
# LangChainAgent(runnable=...) requires: pip install superdialog[langchain]

Tools

PythonTool, HttpTool, and MCPTool implement the Tool ABC and are passed through DialogMachine(tools=[...]) on either engine. On the Playbook engine, prefer declaring tools in the playbook’s process layer (see Tools).

from superdialog import PythonTool, HttpTool, MCPTool, Tool
import os

PythonTool.of(lookup_customer)                # infer id/name/schema
PythonTool(id="lookup", name="lookup", description="...", fn=lookup_customer)

HttpTool(id="lookup", name="lookup", description="Look up a customer",
         url="https://api.company.io/lookup", method="POST",
         auth={"type": "bearer", "token": os.environ["KEY"]})

MCPTool(id="search", name="search", description="Search the KB",
        server="https://mcp.company.io")

Tool.from_dict({"type": "http", "id": "lookup", "name": "lookup",
                "description": "...", "url": "https://api.company.io/lookup"})

HTTP auth accepts {"type": "bearer", "token": "..."} in v0.2 (basic, api_key, callable planned).

LLM provider registration

register_llm_provider(
    name="internal",
    base_url="https://llm.company.io/v1",
    api_key=os.environ["INTERNAL_KEY"],
    api_style="openai",
)
agent = DialogMachine("booking.yaml", llm="custom/internal/llama-3-70b-tuned")

Process-global. Once registered, custom/<name>/<model> works in DialogMachine(llm=...), set_llm(), and create_dialog_flow(llm=...).

Adapters

Import	Purpose
`superdialog.adapters.livekit.DialogMachineLLM`	LiveKit `Agent(llm=...)` plugin (accepts any Agent)
`superdialog.adapters.pipecat.make_processor`	Factory for PipeCat `FrameProcessor`
`superdialog.adapters.fastapi.FastAPIRouter`	Mountable router: `/turn`, `/stream`, `/reset`
`superdialog.adapters.websocket.WebSocketRunner`	Standalone WSS server for Unpod Voice Infra

See Embedding Guides for complete integration examples per host.

Getting Started

Core Concepts

Reference

Embedding Guides

DialogMachine (unified entry point)

Construction

`turn`

`start` / `reset` / `set_llm` / `assist`

`switch_flow` (graph engine only)

Creating agents

`generate_simple_playbook` (default)

`create_dialog_flow` (legacy)

Playbook engine

`Playbook`

`PlaybookAgent`

The artifact model

`Checkpoint`

`SlotSpec`

`AdvanceRule`

`ToolSpec` / `PipelineSpec`

Protocols: `CompletesLLM`, `StreamsLLM`, `HttpFn`, `PythonToolFn`

The expr language

Migrating flows

`EventLog` and `ConversationState`

`replay` and the eval bridge

Sessions (v0.2)

Agent Protocol

`SessionWorker` / `SessionHandle`

Session stores and lock backends

Other agent brains

Tools

LLM provider registration

Adapters

​DialogMachine (unified entry point)

​Construction

​turn

​start / reset / set_llm / assist

​switch_flow (graph engine only)

​Creating agents

​generate_simple_playbook (default)

​create_dialog_flow (legacy)

​Playbook engine

​Playbook

​PlaybookAgent

​The artifact model

​Checkpoint

​SlotSpec

​AdvanceRule

​ToolSpec / PipelineSpec

​Protocols: CompletesLLM, StreamsLLM, HttpFn, PythonToolFn

​The expr language

​Migrating flows

​EventLog and ConversationState

​replay and the eval bridge

​Sessions (v0.2)

​Agent Protocol

​SessionWorker / SessionHandle

​Session stores and lock backends

​Other agent brains

​Tools

​LLM provider registration

​Adapters

DialogMachine (unified entry point)

Construction

`turn`

`start` / `reset` / `set_llm` / `assist`

`switch_flow` (graph engine only)

Creating agents

`generate_simple_playbook` (default)

`create_dialog_flow` (legacy)

Playbook engine

`Playbook`

`PlaybookAgent`

The artifact model

`Checkpoint`

`SlotSpec`

`AdvanceRule`

`ToolSpec` / `PipelineSpec`

Protocols: `CompletesLLM`, `StreamsLLM`, `HttpFn`, `PythonToolFn`

The expr language

Migrating flows

`EventLog` and `ConversationState`

`replay` and the eval bridge

Sessions (v0.2)

Agent Protocol

`SessionWorker` / `SessionHandle`

Session stores and lock backends

Other agent brains

Tools

LLM provider registration

Adapters