⚙️ Configuration Guide¶

This guide covers advanced configuration options for Haive Agents, from basic agent setup to enterprise-scale deployments.

Engine Configuration ¶

AugLLMConfig Options ¶

The AugLLMConfig class provides comprehensive configuration for agent engines.

from haive.core.engine.aug_llm import AugLLMConfig
from pydantic import BaseModel, Field

# Basic configuration
basic_config = AugLLMConfig(
    provider="openai",           # openai, azure, anthropic, custom
    model="gpt-4",              # Model identifier
    temperature=0.7,            # 0.0-2.0, creativity level
    max_tokens=2000,            # Response length limit
    stream=False,               # Enable streaming responses
    timeout=30                  # Request timeout in seconds
)

# Advanced configuration
advanced_config = AugLLMConfig(
    provider="azure",
    model="gpt-4-32k",
    api_base="https://your-resource.openai.azure.com/",
    api_version="2024-02-15-preview",
    deployment_name="gpt-4-deployment",

    # Generation parameters
    temperature=0.5,
    top_p=0.9,
    frequency_penalty=0.1,
    presence_penalty=0.1,

    # System behavior
    system_message="You are an expert AI assistant.",
    enable_function_calling=True,
    max_function_call_rounds=5,

    # Performance
    stream=True,
    enable_caching=True,
    cache_ttl=3600,
    max_retries=3,
    backoff_factor=2.0
)

Provider-Specific Configuration ¶

OpenAI Configuration

openai_config = AugLLMConfig(
    provider="openai",
    model="gpt-4-turbo-preview",
    api_key=os.getenv("OPENAI_API_KEY"),
    organization=os.getenv("OPENAI_ORG_ID"),  # Optional

    # Model-specific options
    temperature=0.7,
    max_tokens=4000,
    top_p=1.0,
    frequency_penalty=0.0,
    presence_penalty=0.0,

    # Advanced options
    seed=12345,                 # For reproducible outputs
    response_format="text",     # text, json_object
    logit_bias={},             # Token bias adjustments
    user="user_123"            # User identifier for tracking
)

Azure OpenAI Configuration

azure_config = AugLLMConfig(
    provider="azure",
    model="gpt-4",
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    api_base=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_version="2024-02-15-preview",
    deployment_name="my-gpt4-deployment",

    # Azure-specific options
    azure_ad_token=None,        # Use AD authentication
    azure_ad_token_provider=None,

    # Same generation options as OpenAI
    temperature=0.6,
    max_tokens=3000
)

Anthropic Claude Configuration

anthropic_config = AugLLMConfig(
    provider="anthropic",
    model="claude-3-opus-20240229",
    api_key=os.getenv("ANTHROPIC_API_KEY"),

    # Claude-specific options
    temperature=0.5,
    max_tokens=4000,
    top_p=0.9,
    top_k=40,

    # System prompts (Claude handles these differently)
    system_message="You are Claude, an AI assistant created by Anthropic."
)

Custom Provider Configuration

from haive.core.engine.custom import CustomLLMProvider

custom_config = AugLLMConfig(
    provider="custom",
    model="my-custom-model",
    custom_provider=CustomLLMProvider(
        api_base="https://my-api.example.com",
        auth_token=os.getenv("CUSTOM_API_KEY"),
        headers={"User-Agent": "MyApp/1.0"}
    ),

    # Standard options still apply
    temperature=0.8,
    max_tokens=2000
)

Structured Output Configuration ¶

Configure agents to return structured, typed data:

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class Sentiment(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"

class AnalysisResult(BaseModel):
    """Structured analysis output."""

    sentiment: Sentiment = Field(description="Overall sentiment")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score")
    key_phrases: List[str] = Field(description="Important phrases identified")
    word_count: int = Field(ge=0, description="Number of words analyzed")
    language: Optional[str] = Field(default=None, description="Detected language")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "sentiment": "positive",
                    "confidence": 0.89,
                    "key_phrases": ["excellent product", "highly recommend"],
                    "word_count": 156,
                    "language": "english"
                }
            ]
        }

# Use structured output in agent
structured_config = AugLLMConfig(
    model="gpt-4",
    structured_output_model=AnalysisResult,
    temperature=0.3,  # Lower for consistent structure
    system_message="You are a sentiment analysis expert. Always provide complete, accurate analysis."
)

Tool Configuration ¶

Configure tools and function calling behavior:

from langchain_core.tools import tool

@tool
def web_search(query: str) -> str:
    """Search the web for current information."""
    # Implementation here
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Calculate mathematical expressions safely."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

# Configure tools in engine
tool_config = AugLLMConfig(
    model="gpt-4",
    tools=[web_search, calculator],

    # Tool behavior options
    tool_choice="auto",          # auto, none, or specific tool name
    parallel_tool_calls=True,    # Allow multiple tools simultaneously
    max_tool_calls_per_turn=5,   # Limit tool usage
    tool_call_timeout=30,        # Timeout per tool call

    # Function calling options
    enable_function_calling=True,
    max_function_call_rounds=3,
    function_call_strategy="react"  # react, plan_and_execute, custom
)

Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action … (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question

Begin!

Question: {input} Thought:{agent_scratchpad}”””

)

MultiAgent Configuration ¶

from haive.agents import MultiAgent

# Create specialized agents
planner = ReactAgent(name="planner", engine=planning_config, tools=planning_tools)
executor = SimpleAgent(name="executor", engine=execution_config)
reviewer = SimpleAgent(name="reviewer", engine=review_config)

# Sequential workflow
sequential_workflow = MultiAgent(
    name="content_pipeline",
    agents=[planner, executor, reviewer],
    execution_mode="sequential",

    # Coordination options
    share_state=True,
    aggregate_results=True,
    max_workflow_time=600,

    # Error handling
    continue_on_error=False,
    fallback_strategy="retry_failed",
    max_agent_failures=2
)

# Parallel execution
parallel_analysis = MultiAgent(
    name="analysis_team",
    agents=[analyst1, analyst2, analyst3],
    execution_mode="parallel",

    # Parallel-specific options
    max_concurrent_agents=3,
    wait_for_all=True,
    timeout_per_agent=120,

    # Result aggregation
    result_aggregation_strategy="merge"  # merge, select_best, vote
)

Memory Configuration ¶

Basic Memory Setup ¶

from haive.agents.memory_reorganized import SimpleMemoryAgent

# Basic vector memory
basic_memory_config = {
    "store_type": "chroma",
    "persist_directory": "./agent_memory",
    "collection_name": "conversations",

    # Memory behavior
    "enable_classification": True,
    "max_memory_items": 10000,
    "similarity_threshold": 0.7,
    "memory_decay_factor": 0.95
}

memory_agent = SimpleMemoryAgent(
    name="remembering_assistant",
    engine=AugLLMConfig(),
    memory_config=basic_memory_config
)

Advanced Integrated Memory ¶

from haive.agents.memory_reorganized.coordination import IntegratedMemorySystem

# Comprehensive memory system
integrated_memory_config = {
    "mode": "INTEGRATED",  # SIMPLE, VECTOR, GRAPH, INTEGRATED

    # Classification system
    "enable_classification": True,
    "classification_model": "gpt-3.5-turbo",
    "memory_types": [
        "EPISODIC", "SEMANTIC", "PROCEDURAL", "CONTEXTUAL",
        "EMOTIONAL", "SOCIAL", "SPATIAL", "TEMPORAL",
        "CAUSAL", "CONDITIONAL", "PERSONAL"
    ],

    # Vector store configuration
    "chroma_config": {
        "persist_directory": "./memory/chroma",
        "collection_name": "agent_knowledge",
        "embedding_model": "sentence-transformers/all-MiniLM-L6-v2",
        "similarity_metric": "cosine",
        "n_results": 10
    },

    # Graph store configuration
    "neo4j_config": {
        "uri": "neo4j://localhost:7687",
        "username": "neo4j",
        "password": os.getenv("NEO4J_PASSWORD"),
        "database": "agent_memory"
    },

    # Memory consolidation
    "enable_consolidation": True,
    "consolidation_schedule": "0 2 * * *",  # Daily at 2 AM
    "consolidation_threshold": 100,  # Trigger after N new memories

    # Importance scoring
    "enable_importance_scoring": True,
    "importance_model": "gpt-4",
    "decay_factor": 0.98,
    "boost_recent": True,

    # Performance options
    "enable_caching": True,
    "cache_ttl": 3600,
    "max_concurrent_retrievals": 5,
    "retrieval_timeout": 10
}

Environment Configuration ¶

Development Environment ¶

# haive_config.yaml - Development
environment: development

agents:
  default_provider: openai
  default_model: gpt-3.5-turbo
  default_temperature: 0.7
  enable_debug_logging: true

memory:
  store_type: chroma
  persist_locally: true
  enable_classification: false  # Faster for development

monitoring:
  enable_metrics: false
  enable_tracing: true
  log_level: DEBUG

performance:
  max_concurrent_agents: 2
  enable_caching: false  # Disable for testing
  timeout_seconds: 60

Production Environment ¶

# haive_config.yaml - Production
environment: production

agents:
  default_provider: azure
  default_model: gpt-4
  default_temperature: 0.5
  enable_debug_logging: false

memory:
  store_type: integrated
  enable_classification: true
  enable_consolidation: true
  backup_interval: 3600

monitoring:
  enable_metrics: true
  enable_tracing: true
  log_level: INFO
  metrics_endpoint: "http://prometheus:9090"

performance:
  max_concurrent_agents: 20
  enable_caching: true
  cache_type: redis
  timeout_seconds: 300

security:
  enable_input_validation: true
  enable_output_sanitization: true
  rate_limiting:
    max_requests_per_minute: 60
    max_requests_per_hour: 1000

Loading Configuration ¶

import yaml
import os
from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class HaiveConfig:
    """Central configuration management."""

    environment: str
    agents: Dict[str, Any]
    memory: Dict[str, Any]
    monitoring: Dict[str, Any]
    performance: Dict[str, Any]
    security: Dict[str, Any] = None

    @classmethod
    def load_from_file(cls, config_path: str) -> "HaiveConfig":
        """Load configuration from YAML file."""
        with open(config_path, 'r') as f:
            config_dict = yaml.safe_load(f)
        return cls(**config_dict)

    @classmethod
    def load_from_env(cls) -> "HaiveConfig":
        """Load configuration from environment variables."""
        return cls(
            environment=os.getenv("HAIVE_ENV", "development"),
            agents={
                "default_provider": os.getenv("HAIVE_DEFAULT_PROVIDER", "openai"),
                "default_model": os.getenv("HAIVE_DEFAULT_MODEL", "gpt-3.5-turbo"),
                "default_temperature": float(os.getenv("HAIVE_DEFAULT_TEMPERATURE", "0.7")),
            },
            memory={
                "store_type": os.getenv("HAIVE_MEMORY_STORE", "chroma"),
                "enable_classification": os.getenv("HAIVE_MEMORY_CLASSIFICATION", "false").lower() == "true",
            },
            monitoring={
                "enable_metrics": os.getenv("HAIVE_ENABLE_METRICS", "false").lower() == "true",
                "log_level": os.getenv("HAIVE_LOG_LEVEL", "INFO"),
            },
            performance={
                "max_concurrent_agents": int(os.getenv("HAIVE_MAX_CONCURRENT", "5")),
                "timeout_seconds": int(os.getenv("HAIVE_TIMEOUT", "120")),
            }
        )

# Usage
config = HaiveConfig.load_from_file("haive_config.yaml")
# or
config = HaiveConfig.load_from_env()

# Create agents with configuration
agent_config = AugLLMConfig(
    provider=config.agents["default_provider"],
    model=config.agents["default_model"],
    temperature=config.agents["default_temperature"]
)

Security Configuration ¶

API Key Management ¶

import os
from cryptography.fernet import Fernet

class SecureConfig:
    """Secure configuration management."""

    def __init__(self, encryption_key: bytes = None):
        self.cipher = Fernet(encryption_key or Fernet.generate_key())

    def get_api_key(self, provider: str) -> str:
        """Get API key securely."""
        encrypted_key = os.getenv(f"{provider.upper()}_API_KEY_ENCRYPTED")
        if encrypted_key:
            return self.cipher.decrypt(encrypted_key.encode()).decode()

        # Fallback to plain environment variable (less secure)
        return os.getenv(f"{provider.upper()}_API_KEY")

    def encrypt_api_key(self, api_key: str) -> str:
        """Encrypt API key for storage."""
        return self.cipher.encrypt(api_key.encode()).decode()

# Usage
secure_config = SecureConfig()
openai_key = secure_config.get_api_key("openai")

agent_config = AugLLMConfig(
    provider="openai",
    model="gpt-4",
    api_key=openai_key
)

Input Validation ¶

from pydantic import BaseModel, Field, validator
import re

class SecureAgentInput(BaseModel):
    """Validated input for agents."""

    content: str = Field(..., min_length=1, max_length=10000)
    user_id: str = Field(..., regex=r"^[a-zA-Z0-9_-]+$")
    session_id: str = Field(..., regex=r"^[a-fA-F0-9-]+$")

    @validator('content')
    def validate_content(cls, v):
        # Check for potential injection attempts
        dangerous_patterns = [
            r'<script.*?>.*?</script>',
            r'javascript:',
            r'data:text/html',
            r'\bexec\b',
            r'\beval\b'
        ]

        for pattern in dangerous_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError("Potentially dangerous content detected")

        return v

# Use in agent
def secure_agent_run(agent, raw_input: dict):
    try:
        validated_input = SecureAgentInput(**raw_input)
        return agent.arun(validated_input.content)
    except ValueError as e:
        return f"Input validation failed: {e}"

Rate Limiting ¶

import time
import asyncio
from collections import defaultdict
from typing import Dict

class RateLimiter:
    """Rate limiting for agent requests."""

    def __init__(self,
                 max_requests_per_minute: int = 60,
                 max_requests_per_hour: int = 1000):
        self.max_per_minute = max_requests_per_minute
        self.max_per_hour = max_requests_per_hour
        self.minute_counts: Dict[str, list] = defaultdict(list)
        self.hour_counts: Dict[str, list] = defaultdict(list)

    async def check_rate_limit(self, user_id: str) -> bool:
        """Check if user has exceeded rate limits."""
        now = time.time()

        # Clean old entries
        minute_cutoff = now - 60
        hour_cutoff = now - 3600

        self.minute_counts[user_id] = [
            ts for ts in self.minute_counts[user_id] if ts > minute_cutoff
        ]
        self.hour_counts[user_id] = [
            ts for ts in self.hour_counts[user_id] if ts > hour_cutoff
        ]

        # Check limits
        if len(self.minute_counts[user_id]) >= self.max_per_minute:
            return False
        if len(self.hour_counts[user_id]) >= self.max_per_hour:
            return False

        # Record this request
        self.minute_counts[user_id].append(now)
        self.hour_counts[user_id].append(now)

        return True

# Usage with agents
rate_limiter = RateLimiter()

async def rate_limited_agent_call(agent, user_id: str, input_data: str):
    if not await rate_limiter.check_rate_limit(user_id):
        raise Exception("Rate limit exceeded")

    return await agent.arun(input_data)

Configuration Best Practices ¶

Development Guidelines

Environment Variables: Use .env files for sensitive data
Configuration Files: Use YAML/JSON for complex configurations
Validation: Always validate configuration values
Defaults: Provide sensible defaults for all options
Documentation: Document all configuration options

Production Guidelines

Security: Encrypt sensitive data, use secure key management
Monitoring: Enable comprehensive metrics and logging
Performance: Optimize for your workload and scale
Reliability: Configure timeouts, retries, and fallbacks
Compliance: Follow data protection and privacy requirements

Configuration Hierarchy

Configuration values are resolved in this order:

Runtime Parameters: Direct method arguments
Environment Variables: OS environment variables
Configuration Files: YAML/JSON configuration files
Class Defaults: Default values in class definitions

Example configuration validation:

def validate_config(config: HaiveConfig):
    """Validate configuration for common issues."""

    # Check required fields
    if not config.agents.get("default_provider"):
        raise ValueError("default_provider is required")

    # Validate temperature range
    temp = config.agents.get("default_temperature", 0.7)
    if not 0.0 <= temp <= 2.0:
        raise ValueError("temperature must be between 0.0 and 2.0")

    # Check memory configuration
    if config.memory.get("store_type") == "neo4j":
        if not config.memory.get("neo4j_config"):
            raise ValueError("neo4j_config required when using Neo4j store")

    # Validate performance settings
    max_concurrent = config.performance.get("max_concurrent_agents", 5)
    if max_concurrent < 1 or max_concurrent > 100:
        raise ValueError("max_concurrent_agents must be between 1 and 100")

This comprehensive configuration guide ensures your Haive Agents are properly configured for any environment, from development to large-scale production deployments.