haive.core.engine.agent.protocols

Agent protocols - Protocol definitions for Agent capabilities.

This module defines protocol interfaces that describe the expected functionality of Agent classes in the Haive framework. These protocols enable runtime type checking and promote consistent interfaces across different agent implementations.

Classes

AgentProtocol	Protocol defining the core functionality of an Agent.
ExtensibilityAgentProtocol	Protocol defining pattern-based extensibility for Agents.
FullAgentProtocol	Protocol combining all agent capabilities.
PersistentAgentProtocol	Protocol defining persistence functionality for Agents.
StreamingAgentProtocol	Protocol defining streaming functionality for Agents.
VisualizationAgentProtocol	Protocol defining visualization capabilities for Agents.

Functions

assert_agent_protocols(agent_class)

Assert that Agent class implements all protocols.

Module Contents

class haive.core.engine.agent.protocols.AgentProtocol

Bases: Protocol[TIn, TOut, TState]

Protocol defining the core functionality of an Agent.

This protocol specifies the minimum interface requirements for Agent implementations in the Haive framework.

async arun(input_data, thread_id=None, config=None, **kwargs)

Asynchronously run the agent with input data.

Parameters:

• input_data (TIn) – Input data for the agent

• thread_id (str | None) – Optional thread ID for persistence

• config (langchain_core.runnables.RunnableConfig | None) – Optional runtime configuration

• **kwargs – Additional runtime configuration

Returns:

Output from the agent

Return type:

TOut

compile()

Compile the agent’s workflow graph.

Return type:

None

run(input_data, thread_id=None, debug=True, config=None, **kwargs)

Synchronously run the agent with input data.

Parameters:

• input_data (TIn) – Input data for the agent

• thread_id (str | None) – Optional thread ID for persistence

• debug (bool) – Whether to enable debug mode

• config (langchain_core.runnables.RunnableConfig | None) – Optional runtime configuration

• **kwargs – Additional runtime configuration

Returns:

Output from the agent

Return type:

TOut

setup_workflow()

Set up the workflow graph for this agent.

Return type:

None

property app: Any

Return the compiled agent application.

Return type:

Any

class haive.core.engine.agent.protocols.ExtensibilityAgentProtocol

Bases: Protocol

Protocol defining pattern-based extensibility for Agents.

This protocol specifies methods related to pattern application and graph modification.

apply_pattern(pattern_name, **kwargs)

Apply a graph pattern to the agent’s workflow.

Parameters:

• pattern_name (str) – Name of the pattern to apply

• **kwargs – Pattern-specific parameters

Return type:

None

class haive.core.engine.agent.protocols.FullAgentProtocol

Bases: AgentProtocol, StreamingAgentProtocol, PersistentAgentProtocol, VisualizationAgentProtocol, ExtensibilityAgentProtocol, Protocol

Protocol combining all agent capabilities.

This protocol represents a fully-featured agent with all available capabilities in the Haive framework.

class haive.core.engine.agent.protocols.PersistentAgentProtocol

Bases: Protocol

Protocol defining persistence functionality for Agents.

This protocol specifies methods related to state persistence and thread management in agents.

inspect_state(thread_id=None, config=None)

Inspect the current state of the agent.

Parameters:

• thread_id (str | None) – Optional thread ID for persistence

• config (langchain_core.runnables.RunnableConfig | None) – Optional runtime configuration

Return type:

None

reset_state(thread_id=None, config=None)

Reset the agent’s state for a thread.

Parameters:

• thread_id (str | None) – Optional thread ID for persistence

• config (langchain_core.runnables.RunnableConfig | None) – Optional runtime configuration

Returns:

True if successful, False otherwise

Return type:

bool

save_state_history(runnable_config=None)

Save the current agent state to a JSON file.

Parameters:

runnable_config (langchain_core.runnables.RunnableConfig | None) – Optional runnable configuration

Returns:

True if successful, False otherwise

Return type:

bool

class haive.core.engine.agent.protocols.StreamingAgentProtocol

Bases: Protocol[TIn, TOut]

Protocol defining streaming functionality for Agents.

This protocol extends the core agent functionality with streaming capabilities for real-time outputs.

async astream(input_data, thread_id=None, stream_mode='values', config=None, **kwargs)

Asynchronously stream agent execution with input data.

Parameters:

• input_data (TIn) – Input data for the agent

• thread_id (str | None) – Optional thread ID for persistence

• stream_mode (str) – Stream mode (values, updates, debug, etc.)

• config (langchain_core.runnables.RunnableConfig | None) – Optional runtime configuration

• **kwargs – Additional runtime configuration

Yields:

Async iterator of state updates during execution

Return type:

collections.abc.AsyncGenerator[dict[str, Any], None]

stream(input_data, thread_id=None, stream_mode='values', config=None, debug=True, **kwargs)

Stream agent execution with input data.

Parameters:

• input_data (TIn) – Input data for the agent

• thread_id (str | None) – Optional thread ID for persistence

• stream_mode (str) – Stream mode (values, updates, debug, etc.)

• config (langchain_core.runnables.RunnableConfig | None) – Optional runtime configuration

• debug (bool) – Whether to enable debug mode

• **kwargs – Additional runtime configuration

Yields:

State updates during execution

Return type:

collections.abc.Generator[dict[str, Any], None, None]

class haive.core.engine.agent.protocols.VisualizationAgentProtocol

Bases: Protocol

Protocol defining visualization capabilities for Agents.

This protocol specifies methods related to graph visualization and debugging.

visualize_graph(output_path=None)

Generate and save a visualization of the agent’s graph.

Parameters:

output_path (str | None) – Optional custom path for visualization output

Return type:

None

haive.core.engine.agent.protocols.assert_agent_protocols(agent_class)

Assert that Agent class implements all protocols.

Parameters:

agent_class (type[haive.core.engine.agent.agent.Agent])