patterns

Reusable Agent Patterns

Pattern: Basic Agent Structure

from haive.agents.base import Agent
from haive.core.schema import StateSchema
from haive.core.graph import BaseGraph
from pydantic import Field
from typing import List, Dict, Any, Optional

class CustomAgentState(StateSchema):
    """State schema for custom agent."""
    messages: List[str] = Field(default_factory=list)
    context: Dict[str, Any] = Field(default_factory=dict)

class CustomAgent(Agent[CustomAgentState]):
    """Custom agent implementation."""

    def __init__(
        self,
        name: str,
        engine: Any,
        **kwargs
    ) -> None:
        super().__init__(
            name=name,
            state_schema=CustomAgentState,
            **kwargs
        )
        self.engine = engine

    def setup_agent(self) -> None:
        """Initialize agent components."""
        self._sync_fields_from_engine()
        self._setup_schemas()
        self._build_initial_graph()

    def build_graph(self) -> BaseGraph:
        """Define agent workflow."""
        graph = BaseGraph()

        # Add your nodes here
        graph.add_node("start", self._process_input)
        graph.add_node("end", self._generate_output)

        # Define edges
        graph.add_edge("start", "end")
        graph.set_entry_point("start")

        return graph.compile()

Pattern: Multi-Engine Agent

class MultiEngineAgent(Agent[MultiEngineState]):
    """Agent using multiple engines for different tasks."""

    def __init__(
        self,
        name: str,
        main_engine: Any,
        tool_engine: Any,
        **kwargs
    ) -> None:
        super().__init__(name=name, **kwargs)
        self.main_engine = main_engine
        self.tool_engine = tool_engine

        # Register engines
        registry = EngineRegistry.get_instance()
        registry.register(main_engine)
        registry.register(tool_engine)

    def build_graph(self) -> BaseGraph:
        """Build graph with engine routing."""
        graph = BaseGraph()

        # Router node decides which engine
        graph.add_node("router", self._route_request)
        graph.add_node("main", self._use_main_engine)
        graph.add_node("tools", self._use_tool_engine)

        # Conditional routing
        graph.add_conditional_edge(
            "router",
            self._needs_tools,
            {
                True: "tools",
                False: "main"
            }
        )

        return graph.compile()