​
Managing State for Individual Agents

xpander.ai’s agent system uses a sophisticated state management approach that integrates agent configuration, workflow graphs, and execution state. This guide explains how to work with single agent state using the xpander.ai SDK.

​
Understanding xpander.ai’s Agent State Model

xpander.ai’s agent state is built around several key components:

  • Agent Configuration: Core properties and settings for the agent
  • Graph Workflow: The connected operations that define agent capabilities
  • Execution Context: The current state of task execution
  • Memory & Messages: Conversation history and contextual information

​
Setting Up an Agent with Proper State

First, you need to initialize a client and connect to an existing agent:

from xpander_sdk import XpanderClient, GraphItem
import os

# Initialize client
xpander_client = XpanderClient(api_key=os.environ['XPANDER_API_KEY'])

# Connect to an existing agent
AGENT_ID = 'your-agent-id'
agent = xpander_client.agents.get(AGENT_ID)
print(f'Agent: {agent.id}')

​
Building the Agent’s Workflow Graph

A key part of xpander.ai’s state management is the workflow graph. This defines how operations connect and how data flows between them:

# First, retrieve available interfaces
agentic_interfaces = agent.retrieve_agentic_interfaces()
linkedin_interface = next((interface for interface in agentic_interfaces 
                           if 'linkedin' in interface.name.lower()), None)

# Then, retrieve operations from the interface
linkedin_operations = agent.retrieve_agentic_operations(agentic_interface=linkedin_interface)

# Find specific operations you want to use
search_profile_operation = next((operation for operation in linkedin_operations 
                                if operation.path == "/search-people"), None)
get_profile_operation = next((operation for operation in linkedin_operations 
                             if operation.path == "/get-profile-data-by-url"), None)

# Attach these operations to the agent
agent.attach_operations(operations=[search_profile_operation, get_profile_operation])

# Build the workflow graph
search_profile_node = agent.graph.add_node(
    GraphItem(
        agent=agent,
        item_id=search_profile_operation.id_to_use_on_graph,
        name=search_profile_operation.name
    )
)

get_profile_node = agent.graph.add_node(
    GraphItem(
        agent=agent,
        item_id=get_profile_operation.id_to_use_on_graph,
        name=get_profile_operation.name
    )
)

# Connect the nodes to define the workflow
search_profile_node.connect([get_profile_node])

# Deploy the changes to save state
agent.sync()

This graph structure is a fundamental part of the agent’s state - it defines how operations are connected and how data flows between them during execution.

​
Managing Execution State

When executing tasks with an agent, the xpander.ai SDK manages execution state through several key methods:

from openai import OpenAI

# Initialize OpenAI client for LLM calls
openai_client = OpenAI(api_key=os.environ['OPENAI_API_KEY'])

# Add a task for the agent to execute
agent.add_task("Find information about John Smith on LinkedIn")

# Initialize the execution context
agent.memory.init_messages(
    input=agent.execution.input_message,
    instructions=agent.instructions
)

# Execute the agent's task
while not agent.is_finished():
    # Generate LLM response using current state
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=agent.messages,
        tools=agent.get_tools(),
        tool_choice=agent.tool_choice,
        temperature=0.0
    )
    
    # Update state with LLM response
    agent.add_messages(response.model_dump())
    
    # Extract and execute tool calls
    tool_calls = XpanderClient.extract_tool_calls(llm_response=response.model_dump())
    agent.run_tools(tool_calls=tool_calls)

# Get the final result
execution_result = agent.retrieve_execution_result()
print(f"Result: {execution_result.result}")

The execution state includes:

  • Current point in the workflow graph
  • Message history with the LLM
  • Results from operation executions
  • Overall task status

​
Working with Agent Memory

xpander.ai’s memory system maintains context throughout the agent’s operation:

# Access the current messages
current_messages = agent.messages

# Add a custom message
agent.add_messages({
    "role": "user",
    "content": "Also search for his experience in AI research."
})

# Check if certain information is already in memory
has_name = any("John Smith" in msg.get("content", "") for msg in agent.messages)

# Add context information
if not has_name:
    agent.memory.add_context("The user is looking for John Smith, an AI researcher.")

​
Visualizing and Debugging Agent State

Understanding the current state of your agent is crucial for debugging:

# Define a helper function to inspect agent state
def inspect_agent_state(agent):
    print(f"Agent ID: {agent.id}")
    print(f"Current task: {agent.execution.input_message if hasattr(agent, 'execution') else 'None'}")
    
    # Show available operations
    print("\nAttached operations:")
    for op in agent.list_attached_operations():
        print(f"- {op.name} ({op.id})")
    
    # Show graph structure
    print("\nWorkflow graph:")
    for node in agent.graph.list_nodes():
        connections = agent.graph.get_connections(node.id)
        print(f"- {node.name} connects to: {[conn.name for conn in connections]}")
    
    # Show execution state if available
    if hasattr(agent, 'execution') and agent.execution:
        print(f"\nExecution status: {agent.execution.status}")
        print(f"Is finished: {agent.is_finished()}")
    
    # Show recent messages if available
    if hasattr(agent, 'messages') and agent.messages:
        print("\nRecent messages:")
        for msg in agent.messages[-3:]:
            print(f"- {msg['role']}: {msg.get('content', '')[:50]}...")

# Use the function to debug
inspect_agent_state(agent)

​
Persisting State Changes

After modifying your agent’s state, you need to persist these changes:

# Apply all pending changes
agent.sync()

# Check if changes were successful
refreshed_agent = xpander_client.agents.get(AGENT_ID)
print(f"Agent has {len(refreshed_agent.list_attached_operations())} operations")

​
Managing Multiple Sessions (Tentative)

For handling multiple users or conversation threads:

# Create or retrieve a session for a specific user
user_session = agent.create_session(user_id="user-123")

# Use a specific session for this execution
agent.add_task(
    "Continue our previous discussion about LinkedIn profiles",
    session_id=user_session.id
)

# Initialize with session context
agent.memory.init_messages(
    input=agent.execution.input_message,
    instructions=agent.instructions,
    session=user_session
)

The session management functionality shown here is a standard pattern in agent systems. Verify specific session API details in the xpander.ai SDK documentation.

​
Best Practices for Agent State Management

  1. Clear workflow definition: Design your agent’s graph carefully to ensure proper data flow
  2. Synchronized changes: Always call agent.sync() after modifying the agent configuration
  3. Error handling: Implement try/except blocks around all state-changing operations
  4. Test execution flow: Validate that your agent properly navigates the workflow graph
  5. Inspect state: Use debugging helpers to understand the current state during development

​
Next Steps