Iworkflow

Classes

class arshai.core.interfaces.iworkflow.INode(node_id, name, **kwargs)

Bases: Protocol

Interface for workflow nodes that wrap agents.

Nodes are callable objects that process workflow state and return updated state along with any other results. They form the building blocks of workflows.

__init__(node_id, name, **kwargs)

Initialize the node with its identifier and name.

get_id()

Get the node’s unique identifier.

Return type:

str

get_name()

Get the node’s name.

Return type:

str

async process(input_data)

Process the input data and return updated state and results.

Parameters:

input_data (Dict[str, Any]) – Dictionary containing the input data including state

Return type:

Dict[str, Any]

Returns:

Dictionary with updated state and processing results

get_agent_settings()

Get the settings associated with the agent for this node.

Return type:

Dict[str, Any]

__call__(input_data)

Make the node callable, directly delegating to the process method.

This allows nodes to be called as functions, matching the pattern from the previous project.

Parameters:

input_data (Dict[str, Any]) – Dictionary containing the input data including state

Return type:

Dict[str, Any]

Returns:

Dictionary with processing results

class arshai.core.interfaces.iworkflow.IUserContext(**data)

Bases: IDTO

User context information.

user_id: str

last_active: datetime

interaction_count: int

dict(*args, **kwargs)

Convert to dictionary with serializable datetime.

Return type:

dict

model_config: ClassVar[ConfigDict] = {'allow_mutation': False, 'arbitrary_types_allowed': True, 'smart_union': True, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class arshai.core.interfaces.iworkflow.IWorkflowConfig(settings, debug_mode=False, **kwargs)

Bases: Protocol

Configuration for workflow orchestration.

This interface defines how workflows are configured, including their structure (nodes and edges) and routing logic.

__init__(settings, debug_mode=False, **kwargs)

Initialize workflow configuration.

create_workflow()

Create and configure the workflow.

Return type:

IWorkflowOrchestrator

class arshai.core.interfaces.iworkflow.IWorkflowOrchestrator(debug_mode=False)

Bases: Protocol

Orchestrates the workflow execution.

The orchestrator manages the flow of control and data between nodes in a workflow. It is responsible for routing input to the appropriate entry node, executing nodes in sequence based on edge connections, and collecting the final results.

__init__(debug_mode=False)

Initialize workflow orchestrator.

add_node(name, node)

Add a node to the workflow.

Return type:

None

add_edge(from_node, to_node)

Add an edge between nodes.

Return type:

None

set_entry_points(router_func, entry_mapping)

Set the workflow entry points with routing logic.

Return type:

None

async execute(input_data, callbacks, is_streaming=False)

Execute the workflow with given input.

Parameters:

input_data (Dict[str, Any]) – Dictionary containing the input data including state

Return type:

Dict[str, Any]

Returns:

Dictionary with final state and workflow results

class arshai.core.interfaces.iworkflow.IWorkflowState(**data)

Bases: IDTO

Complete workflow state.

This state model follows the pattern from the previous project’s WorkflowState, providing a central object that flows through all nodes in the workflow and captures the complete state of the workflow execution.

user_context: IUserContext

working_memories: Dict[str, IWorkingMemory]

notification_state: INotificationState

agent_data: Dict[str, Any]

current_step: str

step_count: int

errors: List[Dict]

processing_path: str

workflow_data: Dict[str, Any]

class Config

Bases: object

arbitrary_types_allowed = True

update_from(other)

Update this state from another state instance.

This method allows for easy copying of state from one instance to another, which is useful when nodes create new state instances.

Parameters:

other (IWorkflowState) – The other state instance to copy from

Return type:

None

model_config: ClassVar[ConfigDict] = {'allow_mutation': False, 'arbitrary_types_allowed': True, 'smart_union': True, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].