haive.core.schema.schema_manager

StateSchemaManager for creating and manipulating state schemas.

This module provides the StateSchemaManager class, which offers a low-level API for dynamically creating, modifying, and managing state schemas at runtime. Unlike the SchemaComposer, which provides a higher-level builder-style API focused on composition, the StateSchemaManager offers fine-grained control over schema construction and modification with support for advanced features like computed properties, validators, and custom methods.

The StateSchemaManager is particularly useful for: - Programmatically building schemas with complex interdependencies - Adding validators, properties, and methods to schemas - Performing schema transformations and modifications at runtime - Providing a programmatic interface for schema manipulation - Creating specialized schema variants with custom behaviors

Key capabilities include: - Field creation and manipulation with comprehensive type handling - Support for field sharing, reducers, and engine I/O relationships - Addition of validators, properties, and computed properties - Dynamic method addition (instance, class, and static methods) - Schema finalization with proper metadata configuration - Integration with SchemaComposer for seamless conversion

Examples

from haive.core.schema import StateSchemaManager from typing import List from langchain_core.messages import BaseMessage

# Create a manager manager = StateSchemaManager(name=”ConversationState”)

# Add fields manager.add_field(

“messages”, List[BaseMessage], default_factory=list, description=”Conversation history”, shared=True

)

# Add a computed property def get_last_message(self):

if not self.messages:

return None

return self.messages[-1]

manager.add_computed_property(“last_message”, get_last_message)

# Add a method def add_message(self, message):

self.messages.append(message)

manager.add_method(“add_message”, add_message)

# Build the schema ConversationState = manager.build()

This module is part of the Haive Schema System, providing the lower-level foundation for schema manipulation that complements the higher-level SchemaComposer.

Classes

StateSchemaManager

Manager for dynamically creating and manipulating state schemas.

Module Contents

class haive.core.schema.schema_manager.StateSchemaManager(data=None, name=None, config=None)

Manager for dynamically creating and manipulating state schemas.

The StateSchemaManager provides a comprehensive, low-level API for dynamically creating, modifying, and managing state schemas at runtime. It offers granular control over schema construction and modification with advanced features like computed properties, validators, and custom methods.

This class serves as a layer over Pydantic’s model creation functionality with additional features specific to the Haive framework, including field sharing, reducer functions, and engine I/O tracking. Unlike SchemaComposer, which provides a higher-level builder-style API focused on composition from components, StateSchemaManager offers fine-grained control over schema construction.

Key capabilities include:

• Field management: Add, modify, and remove fields with comprehensive type handling

• Field sharing: Configure which fields are shared between parent and child graphs

• Reducer functions: Set up field-specific reducers for state merging

• Engine I/O: Track which fields are inputs and outputs for which engines

• Validators: Add custom validation functions for field values

• Properties and computed properties: Define dynamic properties with getters/setters

• Methods: Add instance, class, and static methods to the schema

• Schema finalization: Build and finalize the schema with proper metadata

This class is particularly useful for programmatically building schemas with complex interdependencies, adding validation logic, and creating specialized schema variants with custom behaviors.

Initialize a new StateSchemaManager.

Parameters:

• data (dict[str, Any] | type[pydantic.BaseModel] | pydantic.BaseModel | haive.core.schema.schema_composer.SchemaComposer | None) – Source data to initialize from

• name (str | None) – Custom schema name

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

add_field(name, field_type, default=None, default_factory=None, description=None, shared=False, reducer=None, input_for=None, output_from=None, optional=True, **kwargs)

Add a field to the schema with comprehensive options.

Parameters:

• name (str) – Field name

• field_type (type[T]) – Type of the field

• default (Any) – Default value for the field

• default_factory (collections.abc.Callable[[], T] | None) – Optional factory function for default value

• description (str | None) – Optional field description

• shared (bool) – Whether field is shared with parent graph

• reducer (collections.abc.Callable | None) – Optional reducer function for this field

• input_for (list[str] | None) – List of engines this field serves as input for

• output_from (list[str] | None) – List of engines this field is output from

• optional (bool) – Whether to make the field optional (default: True)

• **kwargs – Additional field parameters

Returns:

Self for chaining

Return type:

StateSchemaManager

add_method(method, method_name=None, method_type='instance')

Add a method to the schema.

Parameters:

• method (collections.abc.Callable) – Method callable to add

• method_name (str | None) – Optional method name (defaults to method.__name__)

• method_type (str) – Type of method: “instance”, “class”, or “static”

Returns:

Self for chaining

Return type:

StateSchemaManager

classmethod create_message_state(additional_fields=None, name='MessageState')

Create a schema with messages field and additional fields.

Parameters:

• additional_fields (dict[str, Any] | None) – Optional dictionary of additional fields to add

• name (str) – Name for the schema

Returns:

StateSchema subclass with messages field

Return type:

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

classmethod from_components(components, name='ComponentSchema', include_messages_field=True)

Create a schema manager from a list of components.

Parameters:

• components (list[Any]) – List of components to extract fields from

• name (str) – Name for the resulting schema

• include_messages_field (bool) – Whether to ensure a messages field exists

Returns:

StateSchemaManager with extracted fields

Return type:

StateSchemaManager

get_model(lock=False, as_state_schema=True, name=None, use_annotated=True)

Create a Pydantic model with all configured options.

Parameters:

• lock (bool) – Whether to lock the schema against further modifications

• as_state_schema (bool) – Whether to use StateSchema as the base class

• name (str | None) – Optional name for the schema class

• use_annotated (bool) – Whether to use Annotated types for metadata

Returns:

Created model class

Return type:

type[pydantic.BaseModel]

has_field(name)

Check if the schema has a specific field.

Parameters:

name (str) – Field name to check

Returns:

True if field exists, False otherwise

Return type:

bool

mark_as_input_field(field_name, engine_name)

Mark a field as an input field for an engine.

Parameters:

• field_name (str) – Name of the field

• engine_name (str) – Name of the engine

Returns:

Self for chaining

Return type:

StateSchemaManager

mark_as_output_field(field_name, engine_name)

Mark a field as an output field for an engine.

Parameters:

• field_name (str) – Name of the field

• engine_name (str) – Name of the engine

Returns:

Self for chaining

Return type:

StateSchemaManager

merge(other)

Merge with another schema, preserving first occurrences.

Parameters:

other (StateSchemaManager | type[pydantic.BaseModel] | pydantic.BaseModel | haive.core.schema.schema_composer.SchemaComposer) – Another object to merge with

Returns:

New merged StateSchemaManager

Return type:

StateSchemaManager

modify_field(name, new_type=None, new_default=None, new_default_factory=None, new_description=None, new_shared=None, new_reducer=None, add_input_for=None, add_output_from=None, remove_input_for=None, remove_output_from=None, **kwargs)

Modify an existing field’s properties.

Parameters:

• name (str) – Name of the field to modify

• new_type (type | None) – New type for the field

• new_default (Any) – New default value

• new_default_factory (collections.abc.Callable | None) – New default factory function

• new_description (str | None) – New description

• new_shared (bool | None) – Whether field should be shared

• new_reducer (collections.abc.Callable | None) – New reducer function

• add_input_for (list[str] | None) – List of engines to add as input consumers

• add_output_from (list[str] | None) – List of engines to add as output producers

• remove_input_for (list[str] | None) – List of engines to remove as input consumers

• remove_output_from (list[str] | None) – List of engines to remove as output producers

• **kwargs – Additional field parameters

Returns:

Self for chaining

Return type:

StateSchemaManager

remove_field(name)

Remove a field from the schema.

Parameters:

name (str) – Name of the field to remove

Returns:

Self for chaining

Return type:

StateSchemaManager

to_composer()

Convert to a SchemaComposer instance.

Returns:

SchemaComposer with current fields and metadata

Return type:

haive.core.schema.schema_composer.SchemaComposer