agents.base.typed_agent¶

Typed agent base classes with clear separation of concerns.

This module provides a cleaner agent hierarchy that matches the state schema hierarchy, with better separation between different types of agents.

Classes¶

`AdaptiveAgent`	Agent that adapts its behavior based on performance.
`BaseAgent`	Base class for agents with primary decision-making engine.
`BaseExecutor`	Base class for all executors (not necessarily agents).
`DataProcessor`	Executor for data processing workflows.
`LLMAgent`	Standard LLM-based agent.
`MetaAgent`	Agent that can spawn and manage other agents.
`ReactiveAgent`	Agent that reacts to specific patterns or triggers.
`ToolExecutor`	Executor for tool-based workflows without LLM.
`WorkflowAgent`	Agent that can modify its own workflow graph.

Functions¶

`create_agent`(agent_type, name[, engine])	Factory to create appropriate agent.
`create_executor`(executor_type, name, **kwargs)	Factory to create appropriate executor.

Module Contents¶

class agents.base.typed_agent.AdaptiveAgent(name, performance_metrics, adaptation_threshold=0.7, **kwargs)¶

Bases: WorkflowAgent

Agent that adapts its behavior based on performance.

Tracks its own performance and modifies strategy accordingly.

Init .

Parameters:

name (str) – [TODO: Add description]
performance_metrics (list[str]) – [TODO: Add description]
adaptation_threshold (float) – [TODO: Add description]

async adapt_strategy(state)¶

Adapt the agent’s strategy.

Parameters:: state (haive.core.schema.base_state_schemas.WorkflowState)
Return type:: None

calculate_performance(state)¶

Calculate performance metrics.

Parameters:: state (haive.core.schema.base_state_schemas.WorkflowState)
Return type:: dict[str, float]

async execute(state)¶

Execute with performance tracking.

Parameters:: state (haive.core.schema.base_state_schemas.WorkflowState)
Return type:: haive.core.schema.base_state_schemas.WorkflowState

needs_adaptation(metrics)¶

Check if adaptation is needed.

Parameters:: metrics (dict[str, float])
Return type:: bool

class agents.base.typed_agent.BaseAgent(name, primary_engine=None, state_schema=AgentState, **kwargs)¶

Bases: BaseExecutor[haive.core.schema.base_state_schemas.AgentState]

Base class for agents with primary decision-making engine.

Agents are executors that have a primary engine (usually LLM) for making decisions.

Init .

Parameters:

name (str) – [TODO: Add description]
primary_engine (haive.core.engine.base.Engine | None) – [TODO: Add description]
state_schema (type[haive.core.schema.base_state_schemas.AgentState]) – [TODO: Add description]

async execute(state)¶

Execute agent logic.

Parameters:: state (haive.core.schema.base_state_schemas.AgentState)
Return type:: haive.core.schema.base_state_schemas.AgentState

abstractmethod run_engine(engine, state)¶

Async:

Parameters:

engine (haive.core.engine.base.Engine)
state (haive.core.schema.base_state_schemas.AgentState)

Return type:

Any

Run the primary engine with state.

update_state_with_result(state, result)¶

Update state with engine result.

Parameters:

state (haive.core.schema.base_state_schemas.AgentState)
result (Any)

Return type:

haive.core.schema.base_state_schemas.AgentState

class agents.base.typed_agent.BaseExecutor(name, state_schema, **kwargs)¶

Bases: abc.ABC, Generic[TState]

Base class for all executors (not necessarily agents).

Executors are components that process state but don’t necessarily have LLM capabilities. This includes tool executors, data processors, routers, validators, etc.

Init .

Parameters:

name (str) – [TODO: Add description]
state_schema (type[TState]) – [TODO: Add description]

abstractmethod execute(state)¶

Async:
Parameters:: state (TState)
Return type:: TState

Execute the processing logic.

get_required_engines()¶

Get list of required engine names.

Return type:: list[str]

validate_state(state)¶

Validate that state has required components.

Parameters:: state (TState)
Return type:: bool

class agents.base.typed_agent.DataProcessor(name, required_engines, **kwargs)¶

Bases: BaseExecutor[haive.core.schema.base_state_schemas.DataProcessingState]

Executor for data processing workflows.

Processes data through various transformation engines.

Init .

Parameters:

name (str) – [TODO: Add description]
required_engines (list[str]) – [TODO: Add description]

async execute(state)¶

Process data through stages.

Parameters:: state (haive.core.schema.base_state_schemas.DataProcessingState)
Return type:: haive.core.schema.base_state_schemas.DataProcessingState

get_required_engines()¶

Get list of required engine names.

Return type:: list[str]

class agents.base.typed_agent.LLMAgent(name, primary_engine=None, state_schema=AgentState, **kwargs)¶

Bases: BaseAgent

Standard LLM-based agent.

Uses an LLM engine for conversation and decision making.

Init .

Parameters:

name (str) – [TODO: Add description]
primary_engine (haive.core.engine.base.Engine | None) – [TODO: Add description]
state_schema (type[haive.core.schema.base_state_schemas.AgentState]) – [TODO: Add description]

async run_engine(engine, state)¶

Run LLM engine.

Parameters:

engine (haive.core.engine.base.Engine)
state (haive.core.schema.base_state_schemas.AgentState)

Return type:

Any

update_state_with_result(state, result)¶

Update state with LLM result.

Parameters:

state (haive.core.schema.base_state_schemas.AgentState)
result (Any)

Return type:

haive.core.schema.base_state_schemas.AgentState

class agents.base.typed_agent.MetaAgent(name, primary_engine=None, agent_factory=None, **kwargs)¶

Bases: WorkflowAgent

Agent that can spawn and manage other agents.

This is for advanced scenarios where agents need to dynamically create and coordinate other agents.

Init .

Parameters:

name (str) – [TODO: Add description]
primary_engine (haive.core.engine.base.Engine | None) – [TODO: Add description]
agent_factory (dict[str, type[BaseAgent]] | None) – [TODO: Add description]

aggregate_results(state)¶

Aggregate sub-agent results.

Parameters:: state (haive.core.schema.base_state_schemas.MetaAgentState)
Return type:: None

async execute(state)¶

Execute meta-agent logic.

Parameters:: state (haive.core.schema.base_state_schemas.MetaAgentState)
Return type:: haive.core.schema.base_state_schemas.MetaAgentState

async execute_sub_agents(state)¶

Execute active sub-agents.

Parameters:: state (haive.core.schema.base_state_schemas.MetaAgentState)
Return type:: None

should_aggregate(state)¶

Determine if results should be aggregated.

Parameters:: state (haive.core.schema.base_state_schemas.MetaAgentState)
Return type:: bool

should_spawn_agents(state)¶

Determine if new agents should be spawned.

Parameters:: state (haive.core.schema.base_state_schemas.MetaAgentState)
Return type:: bool

async spawn_agents(state)¶

Spawn new sub-agents.

Parameters:: state (haive.core.schema.base_state_schemas.MetaAgentState)
Return type:: None

class agents.base.typed_agent.ReactiveAgent(name, triggers, **kwargs)¶

Bases: LLMAgent

Agent that reacts to specific patterns or triggers.

Useful for monitoring, alerting, or event-driven workflows.

Init .

Parameters:

name (str) – [TODO: Add description]
triggers (list[dict[str, Any]]) – [TODO: Add description]

check_triggers(state)¶

Check which triggers are activated.

Parameters:: state (haive.core.schema.base_state_schemas.AgentState)
Return type:: list[str]

evaluate_trigger(trigger, state)¶

Evaluate a single trigger.

Parameters:

trigger (dict[str, Any])
state (haive.core.schema.base_state_schemas.AgentState)

Return type:

bool

async execute(state)¶

Check triggers before normal execution.

Parameters:: state (haive.core.schema.base_state_schemas.AgentState)
Return type:: haive.core.schema.base_state_schemas.AgentState

class agents.base.typed_agent.ToolExecutor(name, execution_strategy='sequential', **kwargs)¶

Bases: BaseExecutor[haive.core.schema.base_state_schemas.ToolExecutorState]

Executor for tool-based workflows without LLM.

Executes tools based on predefined plans or rules.

Init .

Parameters:

name (str) – [TODO: Add description]
execution_strategy (str) – [TODO: Add description]

async execute(state)¶

Execute tools according to plan.

Parameters:: state (haive.core.schema.base_state_schemas.ToolExecutorState)
Return type:: haive.core.schema.base_state_schemas.ToolExecutorState

class agents.base.typed_agent.WorkflowAgent(name, primary_engine=None, initial_graph=None, **kwargs)¶

Bases: BaseAgent

Agent that can modify its own workflow graph.

This agent can inspect results and dynamically modify its execution graph.

Init .

Parameters:

name (str) – [TODO: Add description]
primary_engine (haive.core.engine.base.Engine | None) – [TODO: Add description]
initial_graph (dict[str, Any] | None) – [TODO: Add description]

async determine_graph_modifications(state)¶

Determine what graph modifications to make.

Parameters:: state (haive.core.schema.base_state_schemas.WorkflowState)
Return type:: dict[str, Any]

async execute(state)¶

Execute with potential graph modification.

Parameters:: state (haive.core.schema.base_state_schemas.WorkflowState)
Return type:: haive.core.schema.base_state_schemas.WorkflowState

should_modify_graph(state)¶

Determine if graph should be modified.

Parameters:: state (haive.core.schema.base_state_schemas.WorkflowState)
Return type:: bool

agents.base.typed_agent.create_agent(agent_type, name, engine=None, **kwargs)¶

Factory to create appropriate agent.

Parameters:

agent_type (str) – Type of agent
name (str) – Name for the agent
engine (haive.core.engine.base.Engine | None) – Primary engine for the agent
**kwargs – Additional arguments

Returns:

Agent instance

Return type:

BaseAgent

agents.base.typed_agent.create_executor(executor_type, name, **kwargs)¶

Factory to create appropriate executor.

Parameters:

executor_type (str) – Type of executor
name (str) – Name for the executor
**kwargs – Additional arguments

Returns:

Executor instance

Return type:

BaseExecutor