Node

Classes

class arshai.workflows.node.BaseNode(node_id, name, agent, node_config=None, **kwargs)

Bases: INode

Base implementation of the INode interface.

Wraps an agent and provides the node interface for workflow integration. This implementation follows the direct dependency injection pattern where the agent is provided directly rather than through Settings.

__init__(node_id, name, agent, node_config=None, **kwargs)

Initialize a node with an agent.

Parameters:

• node_id (str) – Unique identifier for this node

• name (str) – Descriptive name for this node

• agent (IAgent) – The agent implementation this node will use

• node_config (Optional[Dict[str, Any]]) – Optional configuration dictionary for this node

• **kwargs (Any) – Additional node-specific settings

Example

# Create agent first from arshai.agents.hub.working_memory import WorkingMemoryAgent from arshai.llms.openai import OpenAIClient

llm_client = OpenAIClient(config) agent = WorkingMemoryAgent(llm_client, memory_manager, “You are helpful”)

# Create node with agent node = BaseNode(

node_id=”process_user_query”, name=”Process User Query”, agent=agent, node_config={“timeout”: 30, “retries”: 3}

)

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 using the underlying agent.

Following the workflow state pattern, this method: 1. Extracts the workflow state from input data 2. Processes the input using the agent 3. Updates the state with the results 4. Returns the 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 workflow state and processing results

get_agent_settings()

Get the configuration associated with 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