Skip to content

API.md

Latest commit

 

History

History
380 lines (278 loc) · 10.8 KB

API.md

File metadata and controls

380 lines (278 loc) · 10.8 KB
Raw

graphexosuit.core API Documentation

Complete reference for the public API of graphexosuit.core, the core runtime and execution layer for LangGraph-based stateful workflows with interrupt/resume capabilities.

Table of Contents

Core Runtime Classes

ExosuitCore

Thin runtime wrapper around a LangGraph workflow that enables execution, pausing, resuming, and retrying of graph operations with checkpoint-based state management.

Constructor

ExosuitCore(*, graph: Any, checkpointer_cm: Any)

Parameters:

  • graph – A LangGraph StateGraph or compiled CompiledStateGraph representing the workflow
  • checkpointer_cm – A context manager that yields a BaseCheckpointSaver for checkpoint persistence

Raises:

  • ValueError – If checkpointer setup fails

Example:

from graphexosuit.core import ExosuitCore
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

graph = StateGraph({"value": int})
# ... define graph nodes and edges
compiled_graph = graph.compile(checkpointer=MemorySaver())

class CheckpointerContextManager:
    def __enter__(self):
        return MemorySaver()
    def __exit__(self, *args):
        pass

core = ExosuitCore(graph=compiled_graph, checkpointer_cm=CheckpointerContextManager())
    
    def get_checkpointer_cm(self):
        checkpointer = MemorySaver()
        return checkpointer

workflow = MyWorkflow()
core = ExosuitCore(workflow)

Methods

run(initial_state: dict, thread_id: Optional[str] = None) -> RunResult

Execute the graph from the beginning with fresh state.

Parameters:

  • initial_state – Initial state dict passed to the graph. Passed through transform_initial_state() if available.
  • thread_id – Optional identifier for this execution. A UUID is generated if omitted.

Returns:

  • RunResult – The execution outcome, either an interrupt or final result.

Raises:

  • GraphExecutionError – If the graph raises an exception during execution.

Example:

result = core.run({"value": 42}, thread_id="session-1")
if result.interrupt_value:
    print(f"Paused: {result.interrupt_value.message}")
else:
    print(f"Completed: {result.final_result}")
resume(thread_id: str, checkpoint_id: str, resume_value: Any) -> RunResult

Resume a paused graph execution from a checkpoint.

Parameters:

  • thread_id – Thread identifier of the paused execution.
  • checkpoint_id – Checkpoint to resume from (from a prior interrupt).
  • resume_value – The payload to send back to the paused node. Passed through transform_resume_value() if available. Typically a dict.

Returns:

  • RunResult – The next execution outcome.

Raises:

  • GraphExecutionError – If the graph raises an exception during execution.

Example:

# After receiving an interrupt
if result.interrupt_value:
    # User selects option 0
    selected_option = result.interrupt_value.options[0]
    
    # Resume with the selected payload
    result = core.resume(
        thread_id=result.thread_id,
        checkpoint_id=result.checkpoint_id,
        resume_value=selected_option.payload
    )
retry(thread_id: str, checkpoint_id: str) -> RunResult

Retry a failed graph node from its last checkpoint.

Parameters:

  • thread_id – Thread identifier of the failed execution.
  • checkpoint_id – Checkpoint at which the failure occurred.

Returns:

  • RunResult – The next execution outcome.

Raises:

  • GraphExecutionError – If the graph raises an exception again during the retry.

Example:

try:
    result = core.run({"value": 42})
except GraphExecutionError as exc:
    print(f"Execution failed at checkpoint {exc.get_checkpoint_id()}")
    # Retry from the checkpoint
    result = core.retry(
        thread_id=exc.get_thread_id(),
        checkpoint_id=exc.get_checkpoint_id()
    )

Data Models

InterruptOption

A selectable option presented to the user during a graph interrupt.

Duck-typed: any object with label and payload attributes works.

@dataclass
class InterruptOption:
    label: str      # Human-readable label for the option
    payload: Any    # Data to send back if this option is selected

Fields:

  • label: str – Human-readable display text for the option.
  • payload: Any – Data structure sent back to the graph if selected (typically a dict or primitive value).

Example:

from graphexosuit.core import InterruptOption

option_a = InterruptOption(
    label="Use default strategy",
    payload={"strategy": "default"}
)
option_b = InterruptOption(
    label="Use aggressive strategy",
    payload={"strategy": "aggressive"}
)

StandardizedInterrupt

The interrupt value that graph nodes must pass to interrupt().

Duck-typed: any object with message and options attributes works.

@dataclass
class StandardizedInterrupt:
    message: str                    # Human-readable pause reason
    options: list[InterruptOption]  # Available resumption choices

Fields:

  • message: str – Human-readable message explaining why execution paused (e.g., "User decision required").
  • options: list[InterruptOption] – List of available options the user can select to resume.

Example:

from graphexosuit.core import StandardizedInterrupt, InterruptOption

interrupt = StandardizedInterrupt(
    message="Choose a strategy to continue",
    options=[
        InterruptOption(label="Default", payload={"strategy": "default"}),
        InterruptOption(label="Aggressive", payload={"strategy": "aggressive"}),
        InterruptOption(label="Conservative", payload={"strategy": "conservative"}),
    ]
)

RunResult

The outcome of a graph execution, pause, or error.

Invariant: Exactly one of interrupt_value or final_result is non-None. If interrupt_value is set, checkpoint_id must also be set.

@dataclass
class RunResult:
    thread_id: str                                    # Execution thread identifier
    checkpoint_id: Optional[str] = None               # Checkpoint ID (set if interrupted)
    interrupt_value: Optional[StandardizedInterrupt] = None  # Interrupt data if paused
    final_result: Optional[dict] = None               # Final output if completed

Fields:

  • thread_id: str – Unique identifier for this execution thread. Used to resume or retry.
  • checkpoint_id: Optional[str] – Checkpoint identifier; set when interrupted, used for resume() and retry().
  • interrupt_value: Optional[StandardizedInterrupt] – The interrupt object if execution paused, otherwise None.
  • final_result: Optional[dict] – Final output dict if execution completed, otherwise None.

Example:

result = core.run({"value": 42})

if result.interrupt_value:
    print(f"Paused: {result.interrupt_value.message}")
    print(f"Options: {[o.label for o in result.interrupt_value.options]}")
    print(f"Resume with thread={result.thread_id}, checkpoint={result.checkpoint_id}")
else:
    print(f"Completed: {result.final_result}")

Exceptions

GraphExecutionError

Raised when the graph throws an exception during execution.

Wraps the original exception and provides thread and checkpoint context for recovery.

class GraphExecutionError(Exception):
    def __init__(self,
                 message: str,
                 original_exception: Exception,
                 thread_id: str,
                 checkpoint_id: str) -> None:
        ...

Parameters:

  • message: str – Descriptive message about the error (e.g., "Graph execution failed").
  • original_exception: Exception – The original exception raised by the graph.
  • thread_id: str – Thread identifier for recovery.
  • checkpoint_id: str – Checkpoint identifier for recovery.

Methods:

  • get_thread_id() -> str – Returns the thread identifier.
  • get_checkpoint_id() -> str – Returns the checkpoint identifier.

Example:

from graphexosuit.core import GraphExecutionError

try:
    result = core.run({"value": 42})
except GraphExecutionError as exc:
    print(f"Execution failed: {exc}")
    thread_id = exc.get_thread_id()
    checkpoint_id = exc.get_checkpoint_id()
    # Retry or log the error
    result = core.retry(thread_id, checkpoint_id)

InvalidInterruptError

Raised when a graph node returns an interrupt value that does not satisfy the StandardizedInterrupt duck type.

An interrupt must have message and options attributes, and each option must have label and payload attributes.

class InvalidInterruptError(ValueError):
    """Raised when an interrupt value does not satisfy the StandardizedInterrupt interface."""

Example:

# This will raise InvalidInterruptError
invalid_interrupt = {"message": "Paused"}  # Missing 'options' attribute

GraphLoaderError

Raised when a graph module cannot be loaded or is missing required functions.

class GraphLoaderError(Exception):
    """Raised when the graph module cannot be loaded or is missing required functions."""

Workflow Example

End-to-end example demonstrating interrupt/resume flow:

from graphexosuit.core import (
    ExosuitCore, ExosuitLiner, StandardizedInterrupt, InterruptOption
)
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from contextlib import contextmanager

class MyWorkflow(ExosuitLiner):
    def get_graph(self) -> StateGraph:
        graph = StateGraph({"value": int, "strategy": str})
        
        def process_node(state):
            # Pause and ask for user input
            return state
        
        def decide_node(state):
            if state.get("strategy") == "default":
                state["value"] *= 2
            elif state.get("strategy") == "aggressive":
                state["value"] *= 10
            return state
        
        graph.add_node("process", process_node)
        graph.add_node("decide", decide_node)
        
        # Interrupt after process_node
        graph.add_edge("process", "decide")
        graph.set_entry_point("process")
        graph.set_finish_point("decide")
        
        return graph
    
    def get_checkpointer_cm(self):
        return MemorySaver()

# Execute
workflow = MyWorkflow()
core = ExosuitCore(workflow)

# Initial run
result = core.run({"value": 5})
print(f"Thread: {result.thread_id}")
print(f"Interrupt: {result.interrupt_value.message}")

# User selects an option
if result.interrupt_value:
    selected = result.interrupt_value.options[0]
    result = core.resume(
        thread_id=result.thread_id,
        checkpoint_id=result.checkpoint_id,
        resume_value=selected.payload,
    )
    print(f"Final result: {result.final_result}")