API Reference

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

import asyncio
from superdialog import create_dialog_flow

# Inside async code
flow = await create_dialog_flow(
    prompt="Confirm appointment. Ask if Friday 4pm works; offer 5pm if not.",
    llm="openai/gpt-5.1",
)

# From a sync entry point
flow = asyncio.run(create_dialog_flow(prompt=..., llm=...))

flow.save("kyc.json")       # Serialize to JSON
flow = Flow.load("kyc.json") # Deserialize from JSON

from superdialog import FlowSet

flowset = FlowSet({
    "main": main_flow,
    "escalation": escalation_flow,
    "billing": billing_flow,
})

DialogMachine(
    flow: Flow | FlowSet,
    llm: str,                                    # model URI
    tools: list[Tool] | None = None,
    memory: Memory | None = None,                # default: in-memory
    config: dict | None = None,                  # max_tokens, temperature, etc.
    traversal_dir: str | Path | None = None,     # auto-save traversal JSON on completion
    adapter: str = "toolcall",                   # "toolcall" (1 LLM call/turn) or "llm" (2 calls/turn)
)

async start() -> Turn

opening = await dialog_machine.start()
print(opening.text)   # "Hello! I'm here to help verify your KYC. Could you provide..."

async turn(
    text: str,
    context: dict | None = None,
    stream: bool | Literal["text"] = False,
) -> Turn | AsyncIterator[StreamChunk]

# Non-streaming - returns complete Turn
reply = await dialog_machine.turn("hello")
print(reply.text)
print(reply.tool_calls)   # list[ToolCall]
print(reply.metadata)     # {"latency_ms": ..., "tokens": ..., "model": ...}

# Streaming - await coroutine first, then iterate
stream = await dialog_machine.turn("hello", stream=True)
async for chunk in stream:
    print(chunk.text, end="")

dialog_machine.reset()

dialog_machine.set_llm("anthropic/claude-haiku-4-5")

dialog_machine.switch_flow("escalation")
dialog_machine.switch_flow("billing", preserve_memory=True)

dialog_machine.assist("Customer is upset. Be especially empathetic.")

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

from superdialog import DialogMachine, SessionWorker, InMemorySessionStore

worker = SessionWorker(
    agent_factory=lambda: DialogMachine(flow=flow, llm="openai/gpt-5.1", tools=tools),
    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.")

from superdialog import LLMAgent, LangChainAgent, SessionWorker

# Raw chat brain - no flow, no slots
worker = SessionWorker(
    agent_factory=lambda: LLMAgent(llm="openai/gpt-5.1", system_prompt="Be helpful."),
    store=InMemorySessionStore(),
)

# LangChain runnable (requires: pip install superdialog[langchain])
worker = SessionWorker(
    agent_factory=lambda: LangChainAgent(runnable=my_chain),
    store=InMemorySessionStore(),
)

from superdialog.tools import tool

@tool
def lookup_customer(customer_id: str) -> dict:
    """Look up customer record by ID."""
    return crm.get(customer_id)

# Use directly in DialogMachine
dialog_machine = DialogMachine(flow=flow, llm="...", tools=[lookup_customer])

# Convenience constructor - infers id, name, description, schema
tool = PythonTool.of(my_function)

# Explicit constructor
tool = PythonTool(
    id="lookup",
    name="lookup",
    description="Look up customer by ID",
    fn=lookup_customer,
)

from superdialog import Tool

tool = Tool.from_dict(
    {"type": "http", "id": "lookup", "name": "lookup",
     "description": "...", "url": "https://api.company.io/lookup"},
    handler_registry={"my_fn": my_python_fn},  # optional, for type=python
)

tool = HttpTool(
    id="lookup",
    name="lookup",
    description="Look up a customer by partial Aadhaar",
    url="https://api.company.io/customer/lookup",
    method="POST",                    # default: POST
    auth={"type": "bearer", "token": os.environ["KEY"]},
    input_schema=None,                # inferred if not provided
)

tool = MCPTool(
    id="search",
    name="search",
    description="Search the knowledge base",
    server="https://mcp.company.io",
    input_schema=None,
)

register_llm_provider(
    name="internal",
    base_url="https://llm.company.io/v1",
    api_key=os.environ["INTERNAL_KEY"],
    api_style="openai",               # only "openai" supported in v0.2
)

# Use anywhere after registration
dialog_machine = DialogMachine(flow=flow, llm="custom/internal/llama-3-70b-tuned")

from superdialog.eval import Eval

eval = Eval(
    flow=flow,
    corpus="tests/kyc-corpus.jsonl",
    llms=["openai/gpt-5.1", "anthropic/claude-haiku-4-5", "groq/llama-3.3-70b"],
    metrics=["accuracy", "latency_p95", "tool_call_correctness"],
)
report = eval.run()
report.save("reports/2026-05-19.md")