haive.core.schema.field_utils

Field utilities for the Haive Schema System.

from typing import Any This module provides a comprehensive set of utilities for creating, extracting, and manipulating Pydantic fields within the Haive Schema System. It ensures consistent handling of field metadata, types, and defaults across the entire framework.

The utilities in this module serve as the low-level foundation for the Schema System, handling technical details like: - Creating fields with standardized metadata - Working with Annotated types for metadata embedding - Extracting metadata from type annotations - Type inference and manipulation - Resolver functions for reducers

Core functions include: - create_field: Create a standard Pydantic field with metadata - create_annotated_field: Create a field using Python’s Annotated type for metadata - extract_type_metadata: Extract base type and metadata from annotations - infer_field_type: Intelligently determine types from values - get_common_reducers: Access standard reducer functions - resolve_reducer: Convert reducer names to functions

These utilities are primarily used by FieldDefinition, SchemaComposer, and StateSchemaManager to implement higher-level functionality.

Examples

from haive.core.schema.field_utils import (

create_field, create_annotated_field, get_common_reducers

) from typing import List import operator

# Create a standard field field_type, field_info = create_field(

field_type=List[str], default_factory=list, description=”List of items”, shared=True, reducer=operator.add

)

# Create an annotated field with embedded metadata field_type, field_info = create_annotated_field(

field_type=List[str], default_factory=list, description=”List of items”, shared=True, reducer=operator.add

)

# Get common reducer functions reducers = get_common_reducers() add_messages = reducers[“add_messages”] # LangGraph’s message list combiner

Classes

FieldMetadata

Standardized container for field metadata in the Haive Schema System.

Functions

camel_to_snake_case(name)	Convert CamelCase to snake_case.
create_annotated_field(field_type[, default, ...])	Create a Pydantic field using Python's Annotated type for embedded metadata.
create_field(field_type[, default, default_factory, ...])	Create a standardized Pydantic field with consistent metadata handling.
create_field_name_from_model(model_class[, ...])	Create a proper field name from a Pydantic model class.
extract_field_info(field_info)	Extract useful information from a Pydantic Field object.
extract_type_metadata(type_annotation)	Extract base type and metadata from a type annotation.
field_config(**config)	Decorator to set field configuration for schema integration.
field_description(description)	Decorator to set the field description for schema integration.
field_name(name)	Simple decorator to set the field name for schema integration.
format_type_annotation(type_annotation)	Format a type annotation for display or documentation.
get_common_reducers()	Get a registry of common reducer functions.
get_field_info_from_model(model_class)	Get field info from a model class, checking for annotations.
infer_field_type(value)	Infer the field type from a value.
resolve_reducer(reducer_name)	Resolve a reducer function from its name.

Module Contents

class haive.core.schema.field_utils.FieldMetadata(description=None, shared=False, reducer=None, source=None, input_for=None, output_from=None, structured_model=None, title=None, **extra)

Standardized container for field metadata in the Haive Schema System.

This class encapsulates all metadata associated with a field, serving as a comprehensive representation of field properties beyond what Pydantic directly supports. It provides a structured way to manage:

• Basic field properties (description, title, etc.)

• Haive-specific properties (shared status, reducer functions, etc.)

• Engine I/O tracking (input/output relationships with engines)

• Structured output model associations

FieldMetadata provides methods for converting between different metadata representations, including dictionaries for Field instantiation and annotation objects for Annotated types. It also supports merging metadata from different sources and serializing reducer functions.

This class serves as a single source of truth for field metadata throughout the Schema System, ensuring consistent handling of field properties across schema composition, manipulation, and serialization operations.

description

Human-readable description of the field

Type:

Optional[str]

shared

Whether the field is shared with parent graphs

Type:

bool

reducer

Function to combine field values during updates

Type:

Optional[Callable]

source

Component that provided this field

Type:

Optional[str]

input_for

Engines this field serves as input for

Type:

List[str]

output_from

Engines this field is output from

Type:

List[str]

structured_model

Name of structured model this field belongs to

Type:

Optional[str]

title

Field title (for OpenAPI/Schema generation)

Type:

Optional[str]

extra

Additional metadata properties

Type:

Dict[str, Any]

Initialize field metadata with comprehensive properties.

Parameters:

• description (str | None) – Human-readable description of the field

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

• reducer (collections.abc.Callable | None) – Function to combine field values during state updates

• source (str | None) – Component that provided 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

• structured_model (str | None) – Name of structured model this field belongs to

• title (str | None) – Optional field title (for OpenAPI/Schema generation)

• **extra – Additional metadata properties

classmethod from_annotation(annotation)

Extract field metadata from an annotated type.

Parameters:

annotation (type) – Type annotation to extract metadata from

Returns:

FieldMetadata if metadata was found, None otherwise

Return type:

Optional[FieldMetadata]

get_reducer_name()

Get serializable name for the reducer function.

Returns:

String name of the reducer function or None

Return type:

str | None

merge(other)

Merge with another FieldMetadata instance.

Parameters:

other (FieldMetadata) – FieldMetadata instance to merge with

Returns:

New FieldMetadata instance with merged data

Return type:

FieldMetadata

to_annotation_metadata()

Convert to a list of metadata objects for Annotated types.

Returns:

List of metadata objects for use in Annotated[Type, …]

Return type:

list[Any]

to_dict()

Convert metadata to dictionary for Field instantiation.

Returns:

Dictionary of metadata suitable for pydantic.Field constructor

Return type:

dict[str, Any]

haive.core.schema.field_utils.camel_to_snake_case(name)

Convert CamelCase to snake_case.

Parameters:

name (str) – CamelCase string to convert

Returns:

snake_case version of the string

Return type:

str

Examples

>>> camel_to_snake_case("QueryRefinementResponse")
'query_refinement_response'
>>> camel_to_snake_case("UserProfile")
'user_profile'
>>> camel_to_snake_case("APIKey")
'api_key'

haive.core.schema.field_utils.create_annotated_field(field_type, default=None, default_factory=None, metadata=None, description=None, shared=False, reducer=None, make_optional=True, **kwargs)

Create a Pydantic field using Python’s Annotated type for embedded metadata.

This function creates a field for Pydantic models using the Annotated type to embed metadata directly in the type annotation. This approach aligns with Pydantic v2’s design and provides better support for schema composition and manipulation.

By embedding metadata in the Annotated type, field properties like shared status and reducer functions stay attached to the field type itself, which allows them to be preserved during operations like schema composition and subclassing.

The function supports both direct metadata parameters (description, shared, reducer) and a comprehensive FieldMetadata object for more complex metadata.

Parameters:

• field_type (Type[T]) – The Python type of the field (e.g., str, List[int])

• default (Any, optional) – Default value for the field. Used if default_factory is not provided. Defaults to None.

• default_factory (Optional[Callable[[], T]], optional) – Factory function that returns the default value. Takes precedence over default if both are provided. Defaults to None.

• metadata (Optional[FieldMetadata], optional) – Comprehensive field metadata object. If provided, other metadata parameters (description, shared, reducer) are ignored. Defaults to None.

• description (Optional[str], optional) – Human-readable description of the field. Ignored if metadata is provided. Defaults to None.

• shared (bool, optional) – Whether the field is shared with parent graphs. Ignored if metadata is provided. Defaults to False.

• reducer (Optional[Callable], optional) – Function to combine field values during updates. Ignored if metadata is provided. Defaults to None.

• make_optional (bool, optional) – Whether to make the field Optional[T] if it’s not already. This ensures the field can be None, which is important for state management. Defaults to True.

• **kwargs – Additional field parameters passed to FieldMetadata or Field.

Returns:

A tuple containing:

• field_type: The annotated type with embedded metadata

• field_info: The Pydantic Field object with standard properties

Return type:

Tuple[Type, Field]

Examples

from typing import List from pydantic import create_model, Field import operator

# Create an annotated field with shared status and reducer field_type, field_info = create_annotated_field(

field_type=List[str], default_factory=list, description=”List of items”, shared=True, reducer=operator.add

)

# Create a model using the field MyModel = create_model(

“MyModel”, items=(field_type, field_info)

)

# The model will have “items” as a shared field with an add reducer # The metadata stays attached to the field type

haive.core.schema.field_utils.create_field(field_type, default=None, default_factory=None, metadata=None, description=None, shared=False, reducer=None, make_optional=True, **kwargs)

Create a standardized Pydantic field with consistent metadata handling.

Parameters:

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

• default (Any) – Default value (used if default_factory is None)

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

• metadata (FieldMetadata | None) – Optional FieldMetadata object for comprehensive metadata

• description (str | None) – Optional field description (ignored if metadata is provided)

• shared (bool) – Whether field is shared with parent (ignored if metadata is provided)

• reducer (collections.abc.Callable | None) – Optional reducer function (ignored if metadata is provided)

• make_optional (bool) – Whether to make the field Optional if it’s not already

• **kwargs – Additional field parameters

Returns:

Tuple of (field_type, field_info) ready for Pydantic model creation

Return type:

tuple[type, pydantic.Field]

haive.core.schema.field_utils.create_field_name_from_model(model_class, remove_suffixes=False)

Create a proper field name from a Pydantic model class.

Parameters:

• model_class (type[pydantic.BaseModel]) – The Pydantic model class

• remove_suffixes (bool) – Whether to remove common suffixes like “Response”, “Result”

Returns:

A properly formatted snake_case field name

Return type:

str

Examples

>>> class QueryRefinementResponse(BaseModel): pass
>>> create_field_name_from_model(QueryRefinementResponse)
'query_refinement_response'
>>> create_field_name_from_model(QueryRefinementResponse, remove_suffixes=True)
'query_refinement'

haive.core.schema.field_utils.extract_field_info(field_info)

Extract useful information from a Pydantic Field object.

Parameters:

field_info (pydantic.Field) – Pydantic Field object

Returns:

Tuple of (default_value, default_factory, metadata_dict)

Return type:

tuple[Any, collections.abc.Callable | None, dict[str, Any]]

haive.core.schema.field_utils.extract_type_metadata(type_annotation)

Extract base type and metadata from a type annotation.

Parameters:

type_annotation (type) – Type annotation to extract from

Returns:

Tuple of (base_type, field_metadata)

Return type:

tuple[type, FieldMetadata | None]

haive.core.schema.field_utils.field_config(**config)

Decorator to set field configuration for schema integration.

Parameters:

**config – Field configuration options

Return type:

Any

Examples

>>> @field_config(required=True, default=None)
... class QueryRefinementResponse(BaseModel):
...     pass

haive.core.schema.field_utils.field_description(description)

Decorator to set the field description for schema integration.

Parameters:

description (str) – The field description to use

Examples

>>> @field_description("Refined query results")
... class QueryRefinementResponse(BaseModel):
...     pass

haive.core.schema.field_utils.field_name(name)

Simple decorator to set the field name for schema integration.

Parameters:

name (str) – The field name to use in schema integration

Examples

>>> @field_name("query_refinement")
... class QueryRefinementResponse(BaseModel):
...     pass

haive.core.schema.field_utils.format_type_annotation(type_annotation)

Format a type annotation for display or documentation.

Parameters:

type_annotation (type) – Type annotation to format

Returns:

Formatted string representation of the type

Return type:

str

haive.core.schema.field_utils.get_common_reducers()

Get a registry of common reducer functions.

Returns:

Dictionary of reducer name -> reducer function

Return type:

dict[str, collections.abc.Callable]

haive.core.schema.field_utils.get_field_info_from_model(model_class)

Get field info from a model class, checking for annotations.

This function looks for field integration annotations on the model class and returns field information for schema integration.

Parameters:

model_class (type[pydantic.BaseModel]) – The Pydantic model class

Returns:

Dictionary with field configuration for schema integration

Return type:

dict[str, Any]

haive.core.schema.field_utils.infer_field_type(value)

Infer the field type from a value.

Parameters:

value (Any) – Value to infer type from

Returns:

Inferred type

Return type:

type

haive.core.schema.field_utils.resolve_reducer(reducer_name)

Resolve a reducer function from its name.

Parameters:

reducer_name (str) – Name of the reducer to resolve

Returns:

Callable reducer function or None if not found

Return type:

collections.abc.Callable | None