haive.core.schema.prebuilt.meta_state

Meta state schema with embedded agent and graph composition support.

This module provides MetaStateSchema, a specialized state schema for graph-level agent composition and recompilation management. It focuses on agent lifecycle, graph coordination, and dynamic recompilation rather than tool routing.

The meta state pattern enables: - Agent embedding within graph states - Graph composition and coordination - Recompilation tracking and management - Agent lifecycle management - Dynamic agent modification

Examples

from haive.core.schema.prebuilt.meta_state import MetaStateSchema from haive.agents.simple.agent import SimpleAgent

# Create a contained agent inner_agent = SimpleAgent()

# Create meta state with embedded agent meta_state = MetaStateSchema(

agent=inner_agent, agent_state={“initialized”: True}, graph_context={“composition”: “nested”}

)

# Agent can be executed and recompiled within graph nodes result = meta_state.execute_agent()

Classes

MetaStateSchema

State schema with embedded agent and graph composition support.

Module Contents

class haive.core.schema.prebuilt.meta_state.MetaStateSchema(/, **data)

Bases: haive.core.schema.state_schema.StateSchema, haive.core.common.mixins.recompile_mixin.RecompileMixin

State schema with embedded agent and graph composition support.

MetaStateSchema extends StateSchema and RecompileMixin to provide graph-level agent composition and recompilation management. It focuses on agent lifecycle, graph coordination, and dynamic recompilation.

Key Features:

• Agent embedding: Store agents as state fields

• Graph composition: Coordinate nested agent execution

• Recompilation tracking: Track when agents need recompilation

• Agent lifecycle: Manage agent state and execution

• Dynamic modification: Support runtime agent changes

Fields:

agent: The contained agent instance graph_context: Graph-level execution context agent_state: Current state of the contained agent execution_result: Result from agent execution composition_metadata: Metadata about graph composition

The meta state tracks agent changes and manages recompilation automatically using the RecompileMixin.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

check_agent_recompilation()

Check if the contained agent needs recompilation.

Returns:

True if agent needs recompilation

Return type:

bool

clone_with_agent(new_agent, reset_history=True)

Create a clone of this meta state with a different agent.

Parameters:

• new_agent (Any) – The new agent to use

• reset_history (bool) – Whether to reset execution history

Returns:

New MetaStateSchema instance with the new agent

Return type:

MetaStateSchema

async execute_agent(input_data=None, config=None, update_state=True)

Execute the contained agent with graph-focused execution.

Parameters:

• input_data (dict[str, Any] | None) – Input data for the agent (defaults to messages)

• config (dict[str, Any] | None) – Execution configuration

• update_state (bool) – Whether to update the meta state with results

Returns:

Dictionary containing execution results

Raises:

• ValueError – If no agent is configured

• RuntimeError – If agent execution fails

Return type:

dict[str, Any]

classmethod from_agent(agent, initial_state=None, graph_context=None)

Create a MetaStateSchema from an agent for graph composition.

Parameters:

• agent (Any) – The agent to embed

• initial_state (dict[str, Any] | None) – Initial agent state

• graph_context (dict[str, Any] | None) – Initial graph context

Returns:

New MetaStateSchema instance

Return type:

MetaStateSchema

get_agent_engine(engine_name)

Get an engine from the contained agent for graph composition.

Parameters:

engine_name (str) – Name of the engine to retrieve

Returns:

Engine instance if found, None otherwise

Return type:

Any | None

get_execution_summary()

Get a summary of agent execution and graph composition status.

Returns:

Dictionary with execution statistics and graph status

Return type:

dict[str, Any]

prepare_agent_input(additional_input=None, include_agent_state=True, include_context=True)

Prepare input data for agent execution with graph context.

Parameters:

• additional_input (dict[str, Any] | None) – Additional input data to include

• include_agent_state (bool) – Whether to include agent state

• include_context (bool) – Whether to include graph context

Returns:

Dictionary with prepared input data

Return type:

dict[str, Any]

reset_execution_state()

Reset execution state for the contained agent.

Return type:

None

setup_graph_composition()

Setup graph composition with the contained agent.

This validator: 1. Sets agent metadata (name, type) 2. Initializes graph context 3. Sets up recompilation tracking 4. Initializes composition metadata

Return type:

Self

update_agent(new_agent)

Update the contained agent and handle recompilation.

Parameters:

new_agent (Any) – The new agent to use

Return type:

None