windlass.agent.graph

graph

LangGraph-backed agent runtime.

When you need more than a loop — conditional routing, subgraphs, explicit state machines, LangGraph's own checkpointers — build the agent with .graph(). This runtime compiles the same configuration into a real StateGraph and hands it to you via :meth:LangGraphRuntime.native_graph.

Install with::

pip install "windlass[agent]"

The Level 3 promise in full::

agent = Windlass.agent().llm("openai").tool(search).graph()

graph = agent.native_graph()          # the uncompiled StateGraph
graph.add_node("review", review_fn)
graph.add_edge("tools", "review")
graph.add_conditional_edges("review", route_fn, {"retry": "agent", "done": END})
agent.recompile()                     # pick up the changes

Example

from windlass import Windlass # doctest: +SKIP agent = Windlass.agent().llm("openai").graph() # doctest: +SKIP agent.run("hello").output # doctest: +SKIP

LangGraphRuntime

LangGraphRuntime(**kwargs: Any)

Bases: AgentRuntime

An agent whose execution is a compiled LangGraph state machine.

Behaves exactly like :class:~windlass.agent.runtime.AgentRuntime — same constructor, same run/stream/resume — but the loop is a graph you can inspect, extend and visualise.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Everything :class:~windlass.agent.runtime.AgentRuntime accepts, including llm, tools, memory, guardrail, checkpointer and tracer.	{}

Raises:

Type	Description
MissingDependencyError	When langgraph is not installed.

Note

Passing checkpointer enables both savers: LangGraph's own keeps the graph state, and the Windlass one keeps the snapshot that makes :meth:~windlass.agent.runtime.AgentRuntime.resume behave identically across the two runtimes.

Note

The graph is compiled lazily on first run. After mutating the graph returned by :meth:native_graph, call :meth:recompile so the change takes effect.

Source code in src\windlass\agent\graph.py

def __init__(self, **kwargs: Any) -> None:
    super().__init__(**kwargs)
    self._langgraph = require("langgraph.graph", extra="agent", feature="The LangGraph runtime")
    self._graph: Any = None
    self._compiled: Any = None
    self._saver: Any = None

native_graph

native_graph() -> Any

Return the uncompiled StateGraph for direct manipulation.

Returns:

Type	Description
Any	The LangGraph StateGraph, with the agent and tools nodes
Any	already wired.

Example

graph = agent.native_graph() # doctest: +SKIP graph.add_node("audit", audit_fn) # doctest: +SKIP agent.recompile() # doctest: +SKIP

Source code in src\windlass\agent\graph.py

def native_graph(self) -> Any:
    """Return the uncompiled ``StateGraph`` for direct manipulation.

    Returns:
        The LangGraph ``StateGraph``, with the ``agent`` and ``tools`` nodes
        already wired.

    Example:
        >>> graph = agent.native_graph()             # doctest: +SKIP
        >>> graph.add_node("audit", audit_fn)        # doctest: +SKIP
        >>> agent.recompile()                        # doctest: +SKIP
    """
    if self._graph is None:
        self._graph = self._build_graph()
    return self._graph

compiled_graph

compiled_graph() -> Any

Return the compiled graph, compiling it if necessary.

Source code in src\windlass\agent\graph.py

def compiled_graph(self) -> Any:
    """Return the compiled graph, compiling it if necessary."""
    if self._compiled is None:
        self._compiled = self._compile()
    return self._compiled

recompile

recompile() -> Any

Recompile after mutating the graph.

Returns:

Type	Description
Any	The freshly compiled graph.

Source code in src\windlass\agent\graph.py

def recompile(self) -> Any:
    """Recompile after mutating the graph.

    Returns:
        The freshly compiled graph.
    """
    self._compiled = self._compile()
    return self._compiled

arun async

arun(
    prompt: Any,
    *,
    thread_id: str | None = None,
    max_iterations: int | None = None,
    **llm_kwargs: Any
) -> AgentResponse

Run the compiled graph to completion.

Parameters:

Name	Type	Description	Default
prompt	Any	A string, a message, or a full transcript.	required
thread_id	str | None	Conversation thread, also used as LangGraph's thread key.	None
max_iterations	int | None	Override for the recursion limit.	None
**llm_kwargs	Any	Accepted for signature parity; graph nodes use the model's configured defaults.	{}

Returns:

Type	Description
AgentResponse	The answer, transcript, per-step trace and aggregate usage.

Raises:

Type	Description
MaxIterationsExceeded	When the recursion limit is hit.
InterruptedError_	When execution paused before a tool node.
AgentError	When graph execution fails.

Source code in src\windlass\agent\graph.py

async def arun(
    self,
    prompt: Any,
    *,
    thread_id: str | None = None,
    max_iterations: int | None = None,
    **llm_kwargs: Any,
) -> AgentResponse:
    """Run the compiled graph to completion.

    Args:
        prompt: A string, a message, or a full transcript.
        thread_id: Conversation thread, also used as LangGraph's thread key.
        max_iterations: Override for the recursion limit.
        **llm_kwargs: Accepted for signature parity; graph nodes use the
            model's configured defaults.

    Returns:
        The answer, transcript, per-step trace and aggregate usage.

    Raises:
        MaxIterationsExceeded: When the recursion limit is hit.
        AgentInterrupt: When execution paused before a tool node.
        AgentError: When graph execution fails.
    """
    from windlass.core.exceptions import InterruptedError_ as AgentInterrupt
    from windlass.core.exceptions import MaxIterationsExceeded

    thread = thread_id or new_id("thread")
    started = time.perf_counter()
    messages = await self._prepare(prompt, thread)
    graph = self.compiled_graph()

    config = {
        "configurable": {"thread_id": thread},
        "recursion_limit": (max_iterations or self.max_iterations) * 2 + 2,
    }

    with self.tracer.span(f"agent.{self.name}", kind="agent", inputs=str(prompt)) as span:
        try:
            final = await graph.ainvoke({"messages": messages, "steps": []}, config=config)
        except Exception as exc:
            if "recursion" in str(exc).lower():
                raise MaxIterationsExceeded(max_iterations or self.max_iterations) from exc
            raise AgentError(f"Graph execution failed: {exc}") from exc

        state = graph.get_state(config) if self.checkpointer is not None else None
        if state is not None and getattr(state, "next", None):
            pending = _pending_calls(final.get("messages", []))
            self._checkpoint(
                thread,
                [_to_windlass(m) for m in final.get("messages", [])],
                list(final.get("steps", [])),
                pending=pending,
            )
            raise AgentInterrupt(
                f"{len(pending)} tool call(s) need approval.",
                payload=[c.model_dump() for c in pending],
                thread_id=thread,
            )

        transcript = [_to_windlass(m) for m in final.get("messages", [])]
        steps = [
            s if isinstance(s, AgentStep) else AgentStep.model_validate(s)
            for s in final.get("steps", [])
        ]
        output = transcript[-1].content if transcript else ""
        output = await self._finish(output, thread, prompt, transcript)

        usage = Usage(calls=0)
        for step in steps:
            usage = usage + step.usage

        span.set_output(output)
        span.set_usage(usage)

    return AgentResponse(
        output=output,
        messages=transcript,
        steps=steps,
        usage=usage,
        latency_ms=(time.perf_counter() - started) * 1000,
        thread_id=thread,
        metadata={"runtime": "langgraph"},
    )

astream async

astream(
    prompt: Any, *, thread_id: str | None = None, **llm_kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream graph execution.

LangGraph streams at node granularity rather than token granularity, so you receive one text event per completed assistant turn plus tool-call events. For token-level streaming use the built-in runtime.

Parameters:

Name	Type	Description	Default
prompt	Any	A string, a message, or a full transcript.	required
thread_id	str | None	Conversation thread.	None
**llm_kwargs	Any	Accepted for signature parity.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	class:~windlass.core.types.StreamEvent values.

Source code in src\windlass\agent\graph.py

async def astream(
    self, prompt: Any, *, thread_id: str | None = None, **llm_kwargs: Any
) -> AsyncIterator[StreamEvent]:
    """Stream graph execution.

    LangGraph streams at node granularity rather than token granularity, so
    you receive one text event per completed assistant turn plus tool-call
    events. For token-level streaming use the built-in runtime.

    Args:
        prompt: A string, a message, or a full transcript.
        thread_id: Conversation thread.
        **llm_kwargs: Accepted for signature parity.

    Yields:
        :class:`~windlass.core.types.StreamEvent` values.
    """
    thread = thread_id or new_id("thread")
    messages = await self._prepare(prompt, thread)
    graph = self.compiled_graph()
    config = {
        "configurable": {"thread_id": thread},
        "recursion_limit": self.max_iterations * 2 + 2,
    }

    async for update in graph.astream({"messages": messages, "steps": []}, config=config):
        for node, payload in (update or {}).items():
            for raw in (payload or {}).get("messages", []):
                message = _to_windlass(raw)
                if message.content:
                    yield StreamEvent(type="text", delta=message.content, raw={"node": node})
                for call in message.tool_calls:
                    yield StreamEvent(type="tool_call", tool_call=call)
    yield StreamEvent(type="done")

describe

describe() -> dict[str, Any]

Return a JSON-safe description, including the graph's node names.

Source code in src\windlass\agent\graph.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe description, including the graph's node names."""
    parts = super().describe()
    parts["runtime"] = "langgraph"
    with contextlib.suppress(Exception):  # description must never fail
        parts["nodes"] = sorted(self.native_graph().nodes)
    return parts

draw

draw() -> str

Return a Mermaid diagram of the compiled graph.

Returns:

Type	Description
str	Mermaid source, ready to paste into documentation.

Raises:

Type	Description
AgentError	When the installed LangGraph cannot render diagrams.

Source code in src\windlass\agent\graph.py

def draw(self) -> str:
    """Return a Mermaid diagram of the compiled graph.

    Returns:
        Mermaid source, ready to paste into documentation.

    Raises:
        AgentError: When the installed LangGraph cannot render diagrams.
    """
    try:
        return str(self.compiled_graph().get_graph().draw_mermaid())
    except Exception as exc:
        raise AgentError(f"Could not render the graph: {exc}") from exc