agents.base.hooks¶

Hook system for agent lifecycle events.

This module provides a flexible hook system that allows users to inject custom logic at various points in agent execution.

The hook system supports the following event types: - Lifecycle: setup, graph building - Execution: before/after run and arun - Node execution: before/after individual nodes - Error handling: errors and retries - State management: state updates

Examples

Basic hook usage:

from haive.agents.simple import SimpleAgent
from haive.agents.base.hooks import HookEvent, logging_hook

agent = SimpleAgent(name="my_agent")

# Add logging hook
agent.add_hook(HookEvent.BEFORE_RUN, logging_hook)

# Add custom hook
def my_hook(context):
    print(f"Starting {context.agent_name}")

agent.add_hook(HookEvent.BEFORE_RUN, my_hook)

Using decorators:

agent = SimpleAgent(name="my_agent")

@agent.before_run
def log_start(context):
    print(f"Starting execution of {context.agent_name}")

@agent.after_run
def log_end(context):
    print(f"Completed: {context.output_data}")

Error handling:

@agent.on_error
def handle_error(context):
    logger.error(f"Error in {context.agent_name}: {context.error}")
    # Could send alerts, retry, etc.

Note

Hooks are executed in the order they were added. Hook errors are caught and logged but don’t interrupt agent execution. Use hooks for monitoring, logging, metrics, validation, and other cross-cutting concerns.

Classes¶

`HookContext`	Context passed to hook functions.
`HookEvent`	Events where hooks can be attached.
`HooksMixin`	Mixin that adds hook functionality to agents.

Functions¶

`comprehensive_workflow_hook`(context)	Comprehensive hook that logs all workflow events.
`create_multi_stage_hook`(stages)	Create a hook that tracks multi-stage agent workflows.
`grading_hook`(context)	Log grading events and track quality metrics.
`logging_hook`(context)	Log hook events.
`message_transformation_hook`(context)	Log message transformation events.
`pre_post_processing_hook`(context)	Log pre/post processing events.
`reflection_hook`(context)	Log reflection events and provide insights.
`retry_limit_hook`([max_retries])	Create a hook that limits retries.
`state_validation_hook`(context)	Validate state updates.
`structured_output_hook`(context)	Log structured output processing events.
`timing_hook`(context)	Track execution time.

Module Contents¶

class agents.base.hooks.HookContext(/, **data)¶

Bases: pydantic.BaseModel

Context passed to hook functions.

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)

model_config¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.base.hooks.HookEvent¶

Bases: str, enum.Enum

Events where hooks can be attached.

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

class agents.base.hooks.HooksMixin(*args, **kwargs)¶

Mixin that adds hook functionality to agents.

Init .

add_hook(event, hook)¶

Add a hook function for an event.

Parameters:

event (HookEvent) – The event to hook into
hook (HookFunction) – The function to call on the event

Return type:

None

Examples

agent.add_hook(HookEvent.BEFORE_RUN, lambda ctx: print(f”Running {ctx.agent_name}”))

after_grading(func)¶

Decorator to add an after_grading hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

after_message_transform(func)¶

Decorator to add an after_message_transform hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

after_reflection(func)¶

Decorator to add an after_reflection hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

after_run(func)¶

Decorator to add an after_run hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

after_setup(func)¶

Decorator to add an after_setup hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

after_structured_output(func)¶

Decorator to add an after_structured_output hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

before_grading(func)¶

Decorator to add a before_grading hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

before_message_transform(func)¶

Decorator to add a before_message_transform hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

before_reflection(func)¶

Decorator to add a before_reflection hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

before_run(func)¶

Decorator to add a before_run hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

before_setup(func)¶

Decorator to add a before_setup hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

before_structured_output(func)¶

Decorator to add a before_structured_output hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

clear_hooks(event=None)¶

Clear hooks for an event or all events.

Parameters:: event (HookEvent | None) – Specific event to clear hooks for, or None for all events
Return type:: None

on_error(func)¶

Decorator to add an on_error hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

post_process(func)¶

Decorator to add a post_process hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

pre_process(func)¶

Decorator to add a pre_process hook.

Parameters:: func (HookFunction)
Return type:: HookFunction

remove_hook(event, hook)¶

Remove a hook function.

Parameters:

event (HookEvent) – The event to remove the hook from
hook (HookFunction) – The hook function to remove

Return type:

None

agents.base.hooks.comprehensive_workflow_hook(context)¶

Comprehensive hook that logs all workflow events.

A single hook that can handle all types of workflow events for complete monitoring and debugging of agent execution.

Parameters:: context (HookContext) – Hook context with event details.
Return type:: None

agents.base.hooks.create_multi_stage_hook(stages)¶

Create a hook that tracks multi-stage agent workflows.

Factory function for creating hooks that monitor complex workflows like reflection, grading, and structured output processing.

Parameters:: stages (list[str]) – List of stage names to track (e.g., [“grading”, “reflection”, “improvement”])
Returns:: A hook function that tracks multi-stage workflows.
Return type:: HookFunction

Examples

hook = create_multi_stage_hook([“grading”, “reflection”, “improvement”]) agent.add_hook(HookEvent.PRE_PROCESS, hook) agent.add_hook(HookEvent.POST_PROCESS, hook)

agents.base.hooks.grading_hook(context)¶

Log grading events and track quality metrics.

Monitors grading processes and logs quality assessment results.

Parameters:: context (HookContext) – Hook context with grading details.
Return type:: None

agents.base.hooks.logging_hook(context)¶

Log hook events.

A general-purpose logging hook that logs event details at appropriate levels.

Parameters:: context (HookContext) – Hook context with event details.
Return type:: None

Note

Uses INFO level for events, DEBUG for data, ERROR for errors.

agents.base.hooks.message_transformation_hook(context)¶

Log message transformation events.

Logs details about message transformations including the transformation type and number of messages processed.

Parameters:: context (HookContext) – Hook context with transformation details.
Return type:: None

agents.base.hooks.pre_post_processing_hook(context)¶

Log pre/post processing events.

Monitors pre and post processing stages in multi-agent workflows.

Parameters:: context (HookContext) – Hook context with processing details.
Return type:: None

agents.base.hooks.reflection_hook(context)¶

Log reflection events and provide insights.

Tracks reflection processing including grading and improvement suggestions.

Parameters:: context (HookContext) – Hook context with reflection details.
Return type:: None

agents.base.hooks.retry_limit_hook(max_retries=3)¶

Create a hook that limits retries.

Factory function that creates a hook to limit retry attempts per agent/node.

Parameters:: max_retries (int) – Maximum number of retries allowed per agent/node combination.
Returns:: A hook function that tracks and limits retries.
Raises:: Exception – When retry limit is exceeded.
Return type:: HookFunction

Examples

agent.add_hook(HookEvent.ON_RETRY, retry_limit_hook(max_retries=5))

agents.base.hooks.state_validation_hook(context)¶

Validate state updates.

Parameters:: context (HookContext)
Return type:: None

agents.base.hooks.structured_output_hook(context)¶

Log structured output processing events.

Tracks structural output parsing and validation.

Parameters:: context (HookContext) – Hook context with structured output details.
Return type:: None

agents.base.hooks.timing_hook(context)¶

Track execution time.

Measures and logs the execution time between BEFORE_RUN/ARUN and AFTER_RUN/ARUN events.

Parameters:: context (HookContext) – Hook context. Uses context.metadata to store start time.
Return type:: None

Note

Must be added to both BEFORE and AFTER events to work properly.

Examples

agent.add_hook(HookEvent.BEFORE_RUN, timing_hook) agent.add_hook(HookEvent.AFTER_RUN, timing_hook)