haive.core.schema.agent_schema_composer¶

Agent Schema Composer for the Haive Framework.

This module provides the AgentSchemaComposer class, which extends SchemaComposer with agent-specific functionality for building dynamic state schemas from multiple agents. It handles intelligent field separation, message preservation, and multi-agent coordination patterns.

The AgentSchemaComposer is the cornerstone of multi-agent state management in Haive, enabling seamless composition of schemas with proper field sharing, reducers, and engine I/O mappings while preserving critical message fields like tool_call_id.

Examples

Basic usage for composing schemas from agents:

from haive.core.schema.agent_schema_composer import AgentSchemaComposer
from haive.agents.react.agent import ReactAgent
from haive.agents.simple.agent import SimpleAgent

# Create agents with their engines
react_agent = ReactAgent(name="Calculator", engine=calc_engine)
simple_agent = SimpleAgent(name="Planner", engine=plan_engine)

# Compose schema from agents
MultiAgentState = AgentSchemaComposer.from_agents(
    agents=[react_agent, simple_agent],
    name="MultiAgentState",
    separation="smart",  # Intelligent field separation
    build_mode=BuildMode.SEQUENCE  # Sequential execution
)

# Create state instance
state = MultiAgentState()
# state.messages will use preserve_messages_reducer automatically

haive.core.schema.agent_schema_composer.logger¶: Module-level logger for debugging schema composition

haive.core.schema.agent_schema_composer.BuildMode[source]¶: Enum defining execution patterns for multi-agent systems

Note

This module uses a custom message reducer (preserve_messages_reducer) instead of LangGraph’s default to prevent loss of fields like tool_call_id in ToolMessage objects during multi-agent execution.

Classes¶

`AgentSchemaComposer`	Enhanced schema composer with agent-specific intelligence.
`BuildMode`	Build modes for agent schema composition.

Functions¶

add_messages(current_msgs, new_msgs)

Combine message lists while preserving BaseMessage objects.

Module Contents¶

class haive.core.schema.agent_schema_composer.AgentSchemaComposer(name='ComposedSchema', base_state_schema=None)[source]¶

Bases: haive.core.schema.schema_composer.SchemaComposer

Enhanced schema composer with agent-specific intelligence.

Extends SchemaComposer to provide intelligent schema composition from agent instances, handling field separation strategies, message preservation, and multi-agent coordination patterns. This class is the primary interface for building state schemas in multi-agent systems.

The composer automatically: - Extracts fields from agent state schemas and engines - Applies intelligent field separation based on usage patterns - Preserves engine I/O mappings for proper routing - Adds message fields with custom reducers to preserve tool_call_id - Includes meta state for agent coordination when needed

Examples

>>> # Compose from multiple agents
>>> schema = AgentSchemaComposer.from_agents(
...     agents=[research_agent, writer_agent],
...     separation="smart"
... )
>>> state = schema()  # Create state instance

Initialize a new SchemaComposer.

Parameters:

name (str) – The name for the composed schema class. Defaults to “ComposedSchema”.
base_state_schema (type[haive.core.schema.state_schema.StateSchema] | None) – Optional custom base state schema to use. If not provided, the composer will auto-detect the appropriate base class.

Examples

Creating a schema composer for a conversational agent:

composer = SchemaComposer(name="ConversationState")
composer.add_field("messages", List[BaseMessage], default_factory=list)
schema_class = composer.build()

Using a custom base schema:

from haive.core.schema.prebuilt import MessagesStateWithTokenUsage
composer = SchemaComposer(
    name="TokenAwareState",
    base_state_schema=MessagesStateWithTokenUsage
)

classmethod from_agents(agents, name=None, include_meta=None, separation='smart', build_mode=None)[source]¶

Compose a state schema from multiple agents with intelligent defaults.

This method is the primary entry point for creating multi-agent state schemas. It analyzes the provided agents and creates a unified schema with proper field sharing, reducers, and engine mappings.

Parameters:

agents (list[haive.agents.base.agent.Agent]) – List of agent instances to compose the schema from. Each agent should have a state_schema and optionally engines with I/O fields.
name (str | None) – Optional name for the composed schema class. If None, generates a name based on the agents (e.g., “MultiAgentState”).
include_meta (bool | None) – Whether to include MetaAgentState for coordination. If None, auto-detects based on number of agents and meta_state usage.
separation (str) – Field separation strategy: - “smart”: Automatically determine sharing based on usage patterns - “shared”: All fields are shared between agents - “namespaced”: Each agent’s fields get a unique prefix
build_mode (BuildMode | None) – Execution pattern for the multi-agent system. If None, defaults to PARALLEL for single agent, SEQUENCE for multiple.

Returns:

A new StateSchema class composed from the agent specifications.

Raises:

ValueError – If no agents are provided or if agents have invalid schemas.

Return type:

type[haive.core.schema.state_schema.StateSchema]

Examples

>>> # Smart separation with auto-detected build mode
>>> schema = AgentSchemaComposer.from_agents(
...     agents=[agent1, agent2],
...     name="CustomMultiState"
... )
>>>
>>> # Explicit configuration
>>> schema = AgentSchemaComposer.from_agents(
...     agents=[supervisor, worker1, worker2],
...     separation="namespaced",
...     build_mode=BuildMode.HIERARCHICAL
... )

Note

The messages field is always added with preserve_messages_reducer to ensure tool_call_id and other fields are maintained across agent boundaries.

classmethod from_agents_with_multiagent_base(agents, name=None, separation='smart', build_mode=None)[source]¶

Compose a state schema that inherits from MultiAgentState with agent fields.

This method creates a schema that MARRIES AgentSchemaComposer with MultiAgentState: - Inherits ALL MultiAgentState functionality (hierarchical agents, agent_states, agent_outputs) - Adds ALL agent-specific fields from the composed agents on top - Preserves the hierarchical big daddy grand state architecture

This is the solution for multi-agent systems that need both: 1. Hierarchical state management (MultiAgentState) 2. Access to all agent-specific fields (composed from agents)

Parameters:

agents (list[haive.agents.base.agent.Agent]) – List of agent instances to compose fields from
name (str | None) – Optional name for the composed schema class
separation (str) – Field separation strategy for agent fields
build_mode (BuildMode | None) – Execution pattern for the multi-agent system

Returns:

A StateSchema class that inherits from MultiAgentState and includes all composed agent fields

Return type:

type[haive.core.schema.state_schema.StateSchema]

Examples

>>> # Self-discovery agents example
>>> schema = AgentSchemaComposer.from_agents_with_multiagent_base(
...     agents=[self_discovery_agent],
...     name="SelfDiscoveryMultiState"
... )
>>> # This schema has BOTH:
>>> # - MultiAgentState fields: agents, agent_states, agent_outputs
>>> # - SelfDiscoveryState fields: reasoning_modules, task_description, etc.

class haive.core.schema.agent_schema_composer.BuildMode[source]¶

Bases: str, enum.Enum

Build modes for agent schema composition.

Defines execution patterns for multi-agent systems, determining how agents are orchestrated and how their fields are composed.

PARALLEL¶: All agents execute independently with separate state spaces

SEQUENCE¶: Agents execute in order, with state flowing between them

HIERARCHICAL¶: Parent-child relationships with supervisor patterns

CUSTOM¶: User-defined execution mode requiring custom implementation

Initialize self. See help(type(self)) for accurate signature.

haive.core.schema.agent_schema_composer.add_messages(current_msgs, new_msgs)[source]¶

Combine message lists while preserving BaseMessage objects.

This function uses preserve_messages_reducer to maintain all fields in BaseMessage objects, including tool_call_id which is critical for multi-agent tool coordination.

Parameters:

current_msgs (list[langchain_core.messages.AnyMessage]) – Existing messages in the state
new_msgs (list[langchain_core.messages.AnyMessage]) – New messages to add to the state

Returns:

Combined list of messages with BaseMessage objects preserved intact

Return type:

list[langchain_core.messages.AnyMessage]

Examples

>>> from langchain_core.messages import ToolMessage
>>> tool_msg = ToolMessage(content="Result", tool_call_id="123")
>>> messages = add_messages([], [tool_msg])
>>> messages[0].tool_call_id  # Preserved!
'123'