agents.base.pre_post_agent_mixin

Pre/Post Agent Processing Mixin.

This mixin generalizes the pre/post agent pattern from the reflection agents to the enhanced base agent, allowing any agent to have pre-processing and post-processing stages with message transformation support.

The pattern supports: - Optional pre-processing agent - Main agent processing - Optional post-processing agent - Message transformation between stages - Hook integration for monitoring

Examples

Basic usage with reflection:

class MyReflectionAgent(Agent, PrePostAgentMixin):
    def setup_agent(self):
        # Set up main agent
        self.main_agent = SimpleAgent(name="writer", engine=config)

        # Set up post-processing with reflection
        self.post_agent = SimpleAgent(name="reflector", engine=reflection_config)
        self.use_post_transform = True
        self.post_transform_type = "reflection"

Graded reflection pattern:

class MyGradedAgent(Agent, PrePostAgentMixin):
    def setup_agent(self):
        self.pre_agent = SimpleAgent(name="grader", engine=grading_config)
        self.main_agent = SimpleAgent(name="responder", engine=main_config)
        self.post_agent = SimpleAgent(name="improver", engine=reflection_config)

        self.use_pre_transform = False
        self.use_post_transform = True

Factory pattern:

agent = create_reflection_agent(
    main_agent=SimpleAgent(name="writer", engine=config),
    reflection_type="graded"
)

Classes

MessageTransformer

Simple message transformer for reflection patterns.

PrePostAgentMixin

Mixin that adds pre/post agent processing capabilities.

Functions

create_graded_reflection_agent(main_agent[, ...])

Create an agent with grading and reflection processing.

create_reflection_agent(main_agent[, ...])

Create an agent with reflection post-processing.

create_structured_output_agent(main_agent, output_model)

Create an agent with structured output post-processing.

Module Contents

class agents.base.pre_post_agent_mixin.MessageTransformer(transformation_type='reflection', preserve_first=True)

Simple message transformer for reflection patterns.

Initialize transformer.

Parameters:
  • transformation_type (str) – Type of transformation (“reflection”, “ai_to_human”, etc.)

  • preserve_first (bool) – Whether to preserve the first message

transform_messages(messages)

Transform messages according to the transformation type.

Parameters:

messages (list[langchain_core.messages.BaseMessage]) – Input messages to transform

Returns:

Transformed messages

Return type:

list[langchain_core.messages.BaseMessage]

class agents.base.pre_post_agent_mixin.PrePostAgentMixin

Mixin that adds pre/post agent processing capabilities.

This mixin generalizes the PrePostMultiAgent pattern from reflection agents to work with any enhanced agent. It provides:

  • Optional pre-processing agent

  • Main agent processing (the agent this mixin is applied to)

  • Optional post-processing agent

  • Message transformation between stages

  • Hook integration for monitoring

  • Configurable transformation types

async arun(input_data)

Override arun to use pre/post processing if agents are configured.

Parameters:

input_data (Any) – Input for the agent

Returns:

Result from pre/post processing or standard arun

Return type:

Any

model_post_init(__context)

Initialize the mixin after Pydantic validation.

Parameters:

__context (Any)

Return type:

None

async run_with_pre_post_processing(input_data)

Execute the agent with pre/post processing stages.

This method orchestrates the full pre → main → post workflow with proper message transformation and hook integration.

Parameters:

input_data (Any) – Input for the agent workflow

Returns:

Combined results from all processing stages

Return type:

dict[str, Any]

setup_transformers()

Set up message transformers based on configuration.

Return type:

None

agents.base.pre_post_agent_mixin.create_graded_reflection_agent(main_agent, grading_agent=None, reflection_agent=None, name=None, **kwargs)

Create an agent with grading and reflection processing.

Parameters:
  • main_agent (haive.agents.base.agent.Agent) – The primary agent that generates responses

  • grading_agent (haive.agents.base.agent.Agent | None) – Optional custom grading agent

  • reflection_agent (haive.agents.base.agent.Agent | None) – Optional custom reflection agent

  • name (str | None) – Name for the enhanced agent

  • **kwargs – Additional configuration

Returns:

Agent with grading and reflection capabilities

Return type:

haive.agents.base.agent.Agent

agents.base.pre_post_agent_mixin.create_reflection_agent(main_agent, reflection_agent=None, name=None, **kwargs)

Create an agent with reflection post-processing.

Parameters:
  • main_agent (haive.agents.base.agent.Agent) – The primary agent that generates responses

  • reflection_agent (haive.agents.base.agent.Agent | None) – Optional custom reflection agent

  • name (str | None) – Name for the enhanced agent

  • **kwargs – Additional configuration

Returns:

Agent with reflection capabilities

Return type:

haive.agents.base.agent.Agent

agents.base.pre_post_agent_mixin.create_structured_output_agent(main_agent, output_model, name=None, **kwargs)

Create an agent with structured output post-processing.

Parameters:
  • main_agent (haive.agents.base.agent.Agent) – The primary agent that generates responses

  • output_model (type[pydantic.BaseModel]) – Pydantic model for structured output

  • name (str | None) – Name for the enhanced agent

  • **kwargs – Additional configuration

Returns:

Agent with structured output capabilities

Return type:

haive.agents.base.agent.Agent