agents.planning.enhanced_plan_execute_v6¶

Enhanced Plan & Execute V6 - Using MultiAgent with Universal Structured Output.

This module provides the next-generation Plan & Execute implementation using: - MultiAgent for multi-agent orchestration - Universal structured output pattern (Agent.with_structured_output) - BasePlannerAgent and BaseExecutorAgent from base planning components - Modern Haive architecture with hooks, recompilation, and state management

## Key Features

  • Universal Structured Output: Any agent can have structured output via mixins

  • MultiAgent: Modern multi-agent coordination with conditional routing

  • Base Planning Components: Uses BasePlannerAgent and BaseExecutorAgent

  • Real Component Testing: No mocks, actual LLM execution and tool usage

  • Modern Architecture: Hooks, recompilation, state management, type safety

## Architecture

BasePlannerAgent.with_structured_output(BasePlan)

↓ (structured BasePlan output)

BaseExecutorAgent.with_structured_output(ExecutionResult)

↓ (structured ExecutionResult output)

Conditional Routing (should_continue)

↓

MultiAgent orchestration

## Usage

### Basic Usage

from haive.agents.planning.enhanced_plan_execute_v6 import create_enhanced_plan_execute_v6

# Create with default configuration agent = create_enhanced_plan_execute_v6() result = await agent.arun(“Calculate compound interest on $1000 at 5% for 10 years”)

# Create with custom tools from haive.tools.tools.search_tools import tavily_search_tool agent = create_enhanced_plan_execute_v6(

name=”research_planner”, executor_tools=[tavily_search_tool]

) result = await agent.arun(“Research Tesla stock performance and calculate ROI”)

### Advanced Configuration
agent = create_enhanced_plan_execute_v6(

name=”advanced_planner”, planner_config=AugLLMConfig(

model=”gpt-4”, temperature=0.2, system_message=”You are an expert strategic planner.”

), executor_config=AugLLMConfig(

model=”gpt-4-turbo”, temperature=0.1

), executor_tools=[tavily_search_tool, calculator_tool], max_iterations=15, enable_hooks=True

)

# Add custom hooks @agent.before_run def track_execution(context):

print(f”Starting planning workflow: {context.agent_name}”)

result = await agent.arun(“Complex multi-step research task”)

## Status: Production Ready

This implementation showcases the full power of modern Haive architecture: - MultiAgent coordination - Universal structured output patterns - Base planning components - Real component integration

Classes¶

ExecutionResult

Structured execution result from executor agent.

PlanExecuteState

Enhanced state for plan and execute workflow coordination.

PlanningDecision

Decision model for routing and coordination.

Functions¶

create_enhanced_plan_execute_v6([name, ...])

Create enhanced Plan & Execute V6 using modern Haive patterns.

create_research_plan_execute_v6([executor_tools])

Create a plan and execute agent optimized for research tasks.

create_simple_plan_execute_v6([executor_tools])

Create a simple plan and execute agent with default settings.

get_next_step_v6(plan, completed_steps)

Get the next step ID to execute based on plan and completed steps.

process_executor_output_v6(state, executor_result)

Process executor output and update state.

process_planner_output_v6(state, planner_result)

Process planner output and update state.

should_continue_v6(state)

Enhanced routing logic for V6 workflow.

test_enhanced_plan_execute_v6()

Test the enhanced plan and execute V6 system.

Module Contents¶

class agents.planning.enhanced_plan_execute_v6.ExecutionResult(/, **data)¶

Bases: pydantic.BaseModel

Structured execution result from executor agent.

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)

class agents.planning.enhanced_plan_execute_v6.PlanExecuteState(/, **data)¶

Bases: haive.core.schema.prebuilt.multi_agent_state.MultiAgentState

Enhanced state for plan and execute workflow coordination.

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)

class agents.planning.enhanced_plan_execute_v6.PlanningDecision(/, **data)¶

Bases: pydantic.BaseModel

Decision model for routing and coordination.

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)

agents.planning.enhanced_plan_execute_v6.create_enhanced_plan_execute_v6(name='EnhancedPlanExecuteV6', planner_config=None, executor_config=None, executor_tools=None, max_iterations=20, enable_hooks=True)¶

Create enhanced Plan & Execute V6 using modern Haive patterns.

Parameters:

  • name (str) – Name for the multi-agent system

  • planner_config (haive.core.engine.aug_llm.AugLLMConfig | None) – Configuration for the planning agent

  • executor_config (haive.core.engine.aug_llm.AugLLMConfig | None) – Configuration for the execution agent

  • executor_tools (list | None) – Tools available to the executor

  • max_iterations (int) – Maximum planning iterations

  • enable_hooks (bool) – Whether to enable the hooks system

Returns:

Complete planning workflow system

Return type:

MultiAgent

Examples

Basic usage:

agent = create_enhanced_plan_execute_v6()
result = await agent.arun("Calculate compound interest")

With custom configuration:

agent = create_enhanced_plan_execute_v6(
    name="research_planner",
    planner_config=AugLLMConfig(model="gpt-4", temperature=0.2),
    executor_tools=[tavily_search_tool],
    max_iterations=15
)
result = await agent.arun("Research and analyze market trends")
agents.planning.enhanced_plan_execute_v6.create_research_plan_execute_v6(executor_tools=None)¶

Create a plan and execute agent optimized for research tasks.

Parameters:

executor_tools (list | None)

Return type:

haive.agents.multi.agent.MultiAgent

agents.planning.enhanced_plan_execute_v6.create_simple_plan_execute_v6(executor_tools=None)¶

Create a simple plan and execute agent with default settings.

Parameters:

executor_tools (list | None)

Return type:

haive.agents.multi.agent.MultiAgent

agents.planning.enhanced_plan_execute_v6.get_next_step_v6(plan, completed_steps)¶

Get the next step ID to execute based on plan and completed steps.

Parameters:

  • plan (haive.agents.planning.base.models.BasePlan[haive.agents.planning.base.models.PlanContent])

  • completed_steps (list[str])

Return type:

str | None

agents.planning.enhanced_plan_execute_v6.process_executor_output_v6(state, executor_result)¶

Process executor output and update state.

Parameters:
Return type:

dict

agents.planning.enhanced_plan_execute_v6.process_planner_output_v6(state, planner_result)¶

Process planner output and update state.

Parameters:

  • state (PlanExecuteState)

  • planner_result (haive.agents.planning.base.models.BasePlan[haive.agents.planning.base.models.PlanContent])

Return type:

dict

agents.planning.enhanced_plan_execute_v6.should_continue_v6(state)¶

Enhanced routing logic for V6 workflow.

Parameters:

state (PlanExecuteState)

Return type:

str

async agents.planning.enhanced_plan_execute_v6.test_enhanced_plan_execute_v6()¶

Test the enhanced plan and execute V6 system.