haive.core.schema.base_state_schemas

Base state schemas with clear inheritance hierarchy.

This module provides a cleaner inheritance structure for state schemas, separating concerns between different types of agents and workflows.

Classes

AgentState	State for a single agent with a primary engine (usually LLM).
DataProcessingState	State for data processing workflows.
EngineState	State that can hold engines (serializable components).
HierarchicalAgentState	State for hierarchical agent systems (parent-child relationships).
MessagingState	State that includes message handling.
MetaAgentState	State for meta-agents that can spawn and manage other agents.
MinimalState	Absolute minimal state - just data, no engines or agents.
MultiAgentState	State for multi-agent systems with proper isolation.
ToolExecutorState	Specialized state for pure tool execution workflows.
ToolState	State that includes tool management.
WorkflowState	State for workflow agents that can modify their own execution graph.

Functions

create_agent_state(agent_type[, with_workflow, with_meta])	Factory to create appropriate agent state class.
create_multi_agent_state([hierarchical])	Factory to create appropriate multi-agent state class.

Module Contents

class haive.core.schema.base_state_schemas.AgentState(/, **data)

Bases: ToolState

State for a single agent with a primary engine (usually LLM).

This is the base for traditional agents that have a main decision-making engine.

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)

property primary_engine: haive.core.engine.base.Engine | None

Get the primary engine.

Return type:

haive.core.engine.base.Engine | None

class haive.core.schema.base_state_schemas.DataProcessingState(/, **data)

Bases: EngineState

State for data processing workflows.

Focuses on data transformation engines rather than LLMs.

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)

class haive.core.schema.base_state_schemas.EngineState(/, **data)

Bases: MessagingState

State that can hold engines (serializable components).

This is the base for states that need engines but aren’t necessarily “agents” in the LLM sense.

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)

get_engine(name)

Get an engine, deserializing if needed.

Parameters:

name (str)

Return type:

haive.core.engine.base.Engine | None

register_engine(name, engine)

Register an engine (can be serialized dict or instance).

Parameters:

• name (str)

• engine (haive.core.engine.base.Engine | dict[str, Any])

Return type:

None

serialize_engines()

Ensure all engines are in serialized form.

Return type:

None

class haive.core.schema.base_state_schemas.HierarchicalAgentState(/, **data)

Bases: MultiAgentState

State for hierarchical agent systems (parent-child relationships).

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)

add_child_agent(agent_name)

Add a child agent.

Parameters:

agent_name (str)

Return type:

None

aggregate_child_results()

Aggregate results from child agents based on strategy.

Return type:

dict[str, Any]

class haive.core.schema.base_state_schemas.MessagingState(/, **data)

Bases: MinimalState

State that includes message handling.

For workflows that need conversation/message tracking but not necessarily LLM capabilities (e.g., routing, logging, monitoring).

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)

add_message(message)

Add a message to the conversation.

Parameters:

message (langchain_core.messages.BaseMessage)

Return type:

None

get_last_message()

Get the most recent message.

Return type:

langchain_core.messages.BaseMessage | None

class haive.core.schema.base_state_schemas.MetaAgentState(/, **data)

Bases: WorkflowState

State for meta-agents that can spawn and manage other agents.

This is for advanced scenarios where agents create and coordinate other agents dynamically.

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)

spawn_sub_agent(name, agent_type, initial_state=None)

Spawn a new sub-agent.

Parameters:

• name (str)

• agent_type (str)

• initial_state (dict[str, Any] | None)

Return type:

None

update_sub_agent_result(name, result)

Update results from a sub-agent.

Parameters:

• name (str)

• result (Any)

Return type:

None

class haive.core.schema.base_state_schemas.MinimalState(/, **data)

Bases: pydantic.BaseModel

Absolute minimal state - just data, no engines or agents.

Use this for simple data transformations, validations, or workflows that don’t need LLM or other engine capabilities.

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)

class haive.core.schema.base_state_schemas.MultiAgentState(/, **data)

Bases: MessagingState

State for multi-agent systems with proper isolation.

This provides a clean separation between shared state and per-agent private state.

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)

broadcast_to_agents(data)

Broadcast data to all agents via shared context.

Parameters:

data (dict[str, Any])

Return type:

None

collect_agent_results()

Collect results from all agents.

Return type:

dict[str, Any]

get_agent_state(agent_name)

Get or create state for an agent.

Parameters:

agent_name (str)

Return type:

AgentState

class haive.core.schema.base_state_schemas.ToolExecutorState(/, **data)

Bases: ToolState

Specialized state for pure tool execution workflows.

No LLM needed - just tool orchestration based on rules or configs.

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)

add_execution_step(tool_name, inputs)

Add a step to the execution plan.

Parameters:

• tool_name (str)

• inputs (dict[str, Any])

Return type:

None

mark_step_complete(result)

Mark current step as complete with result.

Parameters:

result (Any)

Return type:

None

class haive.core.schema.base_state_schemas.ToolState(/, **data)

Bases: EngineState

State that includes tool management.

For workflows that use tools but might not have a primary LLM (e.g., pure tool orchestration, data processing pipelines).

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)

class haive.core.schema.base_state_schemas.WorkflowState(/, **data)

Bases: AgentState

State for workflow agents that can modify their own execution graph.

This enables meta-programming where agents can inspect and modify their own workflow based on results.

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)

get_compiled_graph()

Get compiled graph, recompiling if needed.

Return type:

haive.core.graph.BaseGraph | None

modify_graph(modifications)

Apply modifications to the workflow graph.

Parameters:

modifications (dict[str, Any])

Return type:

None

haive.core.schema.base_state_schemas.create_agent_state(agent_type, with_workflow=False, with_meta=False)

Factory to create appropriate agent state class.

Parameters:

• agent_type (str) – Type of agent (llm, tool_executor, data_processor, etc.)

• with_workflow (bool) – Whether agent can modify its workflow

• with_meta (bool) – Whether agent can spawn sub-agents

Returns:

Appropriate state class

Return type:

type[AgentState]

haive.core.schema.base_state_schemas.create_multi_agent_state(hierarchical=False)

Factory to create appropriate multi-agent state class.

Parameters:

hierarchical (bool) – Whether agents have parent-child relationships

Returns:

Appropriate state class

Return type:

type[MultiAgentState]