haive.core.graph.node.composer.protocols

Protocols for extract and update functions in NodeSchemaComposer.

This module defines the protocol interfaces that extract and update functions must implement to be compatible with the NodeSchemaComposer system.

Classes

ExtractFunction	Protocol for extract functions.
TransformFunction	Protocol for transform functions.
UpdateFunction	Protocol for update functions.

Module Contents

class haive.core.graph.node.composer.protocols.ExtractFunction

Bases: Protocol[TState, TInput]

Protocol for extract functions.

Extract functions take a state object and configuration, returning the extracted input data that will be passed to the node’s processing.

Examples

def extract_messages(state: MessagesState, config: Dict[str, Any]) -> List[BaseMessage]:

field_name = config.get(“field_name”, “messages”) return getattr(state, field_name, [])

def extract_with_projection(state: MultiAgentState, config: Dict[str, Any]) -> Dict[str, Any]:

# Complex projection logic return projected_state

class haive.core.graph.node.composer.protocols.TransformFunction

Bases: Protocol

Protocol for transform functions.

Transform functions are used in field mapping pipelines to modify values during extraction or update operations.

Examples

def uppercase(value: str) -> str:

return value.upper() if isinstance(value, str) else str(value).upper()

def parse_json(value: str) -> Any:

import json return json.loads(value)

class haive.core.graph.node.composer.protocols.UpdateFunction

Bases: Protocol[TState, TOutput]

Protocol for update functions.

Update functions take the result from node processing along with the original state and configuration, returning a dictionary of state updates.

Examples

def update_messages(result: AIMessage, state: MessagesState, config: Dict[str, Any]) -> Dict[str, Any]:

messages = list(getattr(state, “messages”, [])) messages.append(result) return {“messages”: messages}

def update_type_aware(result: Any, state: BaseModel, config: Dict[str, Any]) -> Dict[str, Any]:

# Smart type-based updates return update_dict