haive.agents.base.hooks ======================= .. py:module:: haive.agents.base.hooks .. autoapi-nested-parse:: 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 .. rubric:: 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 ------- .. autoapisummary:: haive.agents.base.hooks.HookContext haive.agents.base.hooks.HookEvent haive.agents.base.hooks.HooksMixin Functions --------- .. autoapisummary:: haive.agents.base.hooks.comprehensive_workflow_hook haive.agents.base.hooks.create_multi_stage_hook haive.agents.base.hooks.grading_hook haive.agents.base.hooks.logging_hook haive.agents.base.hooks.message_transformation_hook haive.agents.base.hooks.pre_post_processing_hook haive.agents.base.hooks.reflection_hook haive.agents.base.hooks.retry_limit_hook haive.agents.base.hooks.state_validation_hook haive.agents.base.hooks.structured_output_hook haive.agents.base.hooks.timing_hook Module Contents --------------- .. py:class:: HookContext(/, **data) Bases: :py:obj:`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. .. py:attribute:: model_config Configuration for the model, should be a dictionary conforming to [`ConfigDict`][pydantic.config.ConfigDict]. .. py:class:: HookEvent Bases: :py:obj:`str`, :py:obj:`enum.Enum` Events where hooks can be attached. Initialize self. See help(type(self)) for accurate signature. .. py:class:: HooksMixin(*args, **kwargs) Mixin that adds hook functionality to agents. Init . .. py:method:: add_hook(event, hook) Add a hook function for an event. :param event: The event to hook into :param hook: The function to call on the event .. rubric:: Examples agent.add_hook(HookEvent.BEFORE_RUN, lambda ctx: print(f"Running {ctx.agent_name}")) .. py:method:: after_grading(func) Decorator to add an after_grading hook. .. py:method:: after_message_transform(func) Decorator to add an after_message_transform hook. .. py:method:: after_reflection(func) Decorator to add an after_reflection hook. .. py:method:: after_run(func) Decorator to add an after_run hook. .. py:method:: after_setup(func) Decorator to add an after_setup hook. .. py:method:: after_structured_output(func) Decorator to add an after_structured_output hook. .. py:method:: before_grading(func) Decorator to add a before_grading hook. .. py:method:: before_message_transform(func) Decorator to add a before_message_transform hook. .. py:method:: before_reflection(func) Decorator to add a before_reflection hook. .. py:method:: before_run(func) Decorator to add a before_run hook. .. py:method:: before_setup(func) Decorator to add a before_setup hook. .. py:method:: before_structured_output(func) Decorator to add a before_structured_output hook. .. py:method:: clear_hooks(event = None) Clear hooks for an event or all events. :param event: Specific event to clear hooks for, or None for all events .. py:method:: on_error(func) Decorator to add an on_error hook. .. py:method:: post_process(func) Decorator to add a post_process hook. .. py:method:: pre_process(func) Decorator to add a pre_process hook. .. py:method:: remove_hook(event, hook) Remove a hook function. :param event: The event to remove the hook from :param hook: The hook function to remove .. py:function:: 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. :param context: Hook context with event details. .. py:function:: 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. :param stages: List of stage names to track (e.g., ["grading", "reflection", "improvement"]) :returns: A hook function that tracks multi-stage workflows. .. rubric:: Examples hook = create_multi_stage_hook(["grading", "reflection", "improvement"]) agent.add_hook(HookEvent.PRE_PROCESS, hook) agent.add_hook(HookEvent.POST_PROCESS, hook) .. py:function:: grading_hook(context) Log grading events and track quality metrics. Monitors grading processes and logs quality assessment results. :param context: Hook context with grading details. .. py:function:: logging_hook(context) Log hook events. A general-purpose logging hook that logs event details at appropriate levels. :param context: Hook context with event details. .. note:: Uses INFO level for events, DEBUG for data, ERROR for errors. .. py:function:: message_transformation_hook(context) Log message transformation events. Logs details about message transformations including the transformation type and number of messages processed. :param context: Hook context with transformation details. .. py:function:: pre_post_processing_hook(context) Log pre/post processing events. Monitors pre and post processing stages in multi-agent workflows. :param context: Hook context with processing details. .. py:function:: reflection_hook(context) Log reflection events and provide insights. Tracks reflection processing including grading and improvement suggestions. :param context: Hook context with reflection details. .. py:function:: 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. :param max_retries: 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. .. rubric:: Examples agent.add_hook(HookEvent.ON_RETRY, retry_limit_hook(max_retries=5)) .. py:function:: state_validation_hook(context) Validate state updates. .. py:function:: structured_output_hook(context) Log structured output processing events. Tracks structural output parsing and validation. :param context: Hook context with structured output details. .. py:function:: timing_hook(context) Track execution time. Measures and logs the execution time between BEFORE_RUN/ARUN and AFTER_RUN/ARUN events. :param context: Hook context. Uses context.metadata to store start time. .. note:: Must be added to both BEFORE and AFTER events to work properly. .. rubric:: Examples agent.add_hook(HookEvent.BEFORE_RUN, timing_hook) agent.add_hook(HookEvent.AFTER_RUN, timing_hook)