Skip to main content

Table of Contents

Module spoon_ai.graph.engine

Graph engine: StateGraph, CompiledGraph, and interrupt API implementation.

BaseNode Objects​

class BaseNode(ABC, Generic[State])

Base class for all graph nodes

__call__​

@abstractmethod
async def __call__(state: State,
                   config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Execute the node logic

RunnableNode Objects​

class RunnableNode(BaseNode[State])

Runnable node that wraps a function

__call__​

async def __call__(state: State,
                   config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Execute the wrapped function

ToolNode Objects​

class ToolNode(BaseNode[State])

Tool node for executing tools

__call__​

async def __call__(state: State,
                   config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Execute tools based on state

ConditionNode Objects​

class ConditionNode(BaseNode[State])

Conditional node for routing decisions

__call__​

async def __call__(state: State,
                   config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Execute condition and return routing decision

interrupt​

def interrupt(data: Dict[str, Any]) -> Any

Interrupt execution and wait for human input.

RouteRule Objects​

class RouteRule()

Advanced routing rule for automatic path selection

matches​

def matches(state: Dict[str, Any], query: str = "") -> bool

Check if this rule matches the current state/query

RunningSummary Objects​

@dataclass
class RunningSummary()

Rolling conversation summary used by the summarisation node.

SummarizationNode Objects​

class SummarizationNode(BaseNode[Dict[str, Any]])

Node that summarises conversation history before model invocation.

StateGraph Objects​

class StateGraph(Generic[State])

add_node​

def add_node(
        node_name: str, node: Union[BaseNode[State],
                                    Callable[[State], Any]]) -> "StateGraph"

Add a node to the graph

add_edge​

def add_edge(
        start_node: str,
        end_node: str,
        condition: Optional[Callable[[State], bool]] = None) -> "StateGraph"

Add an edge. When condition is provided, edge becomes conditional.

add_conditional_edges​

def add_conditional_edges(start_node: str, condition: Callable[[State], str],
                          path_map: Dict[str, str]) -> "StateGraph"

Add conditional edges

set_entry_point​

def set_entry_point(node_name: str) -> "StateGraph"

Set the entry point

add_tool_node​

def add_tool_node(tools: List[Any], name: str = "tools") -> "StateGraph"

Add a tool node

add_conditional_node​

def add_conditional_node(condition_func: Callable[[State], str],
                         name: str = "condition") -> "StateGraph"

Add a conditional node

add_parallel_group​

def add_parallel_group(
    group_name: str,
    nodes: List[str],
    config: Optional[Union[Dict[str, Any], ParallelGroupConfig]] = None
) -> "StateGraph"

Add a parallel execution group

add_routing_rule​

def add_routing_rule(source_node: str,
                     condition: Union[str, Callable[[State, str], bool]],
                     target_node: str,
                     priority: int = 0) -> "StateGraph"

Add an intelligent routing rule

get_state​

def get_state(
        config: Optional[Dict[str, Any]] = None) -> Optional[StateSnapshot]

Fetch the latest (or specified) checkpoint snapshot for a thread.

get_state_history​

def get_state_history(
        config: Optional[Dict[str, Any]] = None) -> Iterable[StateSnapshot]

Return all checkpoints for the given thread, ordered by creation time.

add_pattern_routing​

def add_pattern_routing(source_node: str,
                        pattern: str,
                        target_node: str,
                        priority: int = 0) -> "StateGraph"

Add pattern-based routing rule

set_intelligent_router​

def set_intelligent_router(
        router_func: Callable[[Dict[str, Any], str], str]) -> "StateGraph"

Set the intelligent router function

set_llm_router​

def set_llm_router(router_func: Optional[Callable[[Dict[str, Any], str],
                                                  str]] = None,
                   config: Optional[Dict[str, Any]] = None) -> "StateGraph"

Set the LLM-powered router function

Arguments:

  • router_func - Custom LLM router function. If None, uses default LLM router.
  • config - Configuration for LLM router (model, temperature, max_tokens, etc.)

enable_llm_routing​

def enable_llm_routing(
        config: Optional[Dict[str, Any]] = None) -> "StateGraph"

Enable LLM-powered natural language routing

This automatically sets up LLM routing for the graph entry point.

compile​

def compile(checkpointer: Optional[Any] = None) -> "CompiledGraph"

Compile the graph

get_graph​

def get_graph() -> Dict[str, Any]

Get graph structure for visualization/debugging

CompiledGraph Objects​

class CompiledGraph(Generic[State])

Compiled graph for execution

get_execution_metrics​

def get_execution_metrics() -> Dict[str, Any]

Get aggregated execution metrics