Skip to main content

Table of Contents

Module spoon_ai.agents.graph_agent

Graph-based agent implementation for SpoonOS.

This module provides the GraphAgent class that executes StateGraph workflows, integrating the graph execution system with the existing agent architecture.

GraphAgent Objects​

class GraphAgent(BaseAgent)

An agent that executes StateGraph workflows.

This agent provides a bridge between the existing SpoonOS agent architecture and the new graph-based execution system. It allows complex, stateful workflows to be defined as graphs and executed with proper state management.

Key Features:

  • Executes StateGraph workflows
  • Maintains compatibility with existing agent interfaces
  • Provides detailed execution logging and error handling
  • Supports both sync and async node functions

__init__​

def __init__(**kwargs)

Initialize the GraphAgent.

Arguments:

  • graph - StateGraph instance to execute
  • **kwargs - Additional arguments passed to BaseAgent

Raises:

  • ValueError - If no graph is provided

validate_graph​

@validator('graph')
def validate_graph(cls, v)

Validate that the provided graph is a StateGraph instance.

run​

async def run(request: Optional[str] = None) -> str

Execute the graph workflow.

This method overrides the base run method to invoke the compiled graph instead of the traditional step-based execution loop.

Arguments:

  • request - Optional input request to include in initial state

Returns:

String representation of the execution result

Raises:

  • RuntimeError - If agent is not in IDLE state
  • GraphExecutionError - If graph execution fails

step​

async def step() -> str

Step method for compatibility with BaseAgent.

Since GraphAgent uses graph execution instead of step-based execution, this method is not used in normal operation but is required by the BaseAgent interface.

Returns:

Status message indicating graph-based execution

get_execution_history​

def get_execution_history() -> list

Get the execution history from the last graph run.

Returns:

List of execution steps with metadata

get_execution_metadata​

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

Get metadata from the last execution.

Returns:

Dictionary containing execution metadata

clear_state​

def clear_state()

Clear preserved state and execution history.

update_initial_state​

def update_initial_state(updates: Dict[str, Any])

Update the initial state for future executions.

Arguments:

  • updates - Dictionary of state updates to merge

set_preserve_state​

def set_preserve_state(preserve: bool)

Enable or disable state preservation between runs.

Arguments:

  • preserve - Whether to preserve state between runs