arshai 1.9.0
A powerful agent framework for building conversational AI systems
License: MIT
Requires Python: <3.13,>=3.11
Keywords: ai, agents, llm, workflow, rag, conversational-ai, multi-agent, vector-db, embeddings

# Arshai

**A powerful AI application framework built on clean architecture principles with complete developer control.**

```
┌────────────────────────────────────────────────────────────┐
│          Layer 3: Agentic Systems                          │
├────────────────────────────────────────────────────────────┤
│  Workflows  →┐                                             │
│  Memory     →├──→  Agents                                  │
│  Orchestration→┘                                           │
└─────────────────────┬──────────────────────────────────────┘
                      │
┌─────────────────────▼──────────────────────────────────────┐
│          Layer 2: Agents                                   │
├────────────────────────────────────────────────────────────┤
│              Agents  →  Tool Integration                   │
│                      →  Custom Logic                       │
└─────────────────────┬──────────────────────────────────────┘
                      │
┌─────────────────────▼──────────────────────────────────────┐
│          Layer 1: LLM Clients                              │
├────────────────────────────────────────────────────────────┤
│              Custom Logic  →  OpenAI                       │
│                            →  Azure OpenAI                 │
│                            →  Google Gemini                │
│                            →  OpenRouter                   │
└────────────────────────────────────────────────────────────┘
```

## 📚 Documentation

**Complete documentation is available at: [Arshai Documentation](https://felesh-ai.github.io/arshai/)** (or build locally with `cd docs_sphinx && make html`)

The comprehensive documentation covers:
- **Getting Started**: Quick installation and your first agent
- **Framework Guide**: Deep dive into agents, tools, memory, and orchestration
- **Implementations**: LLM clients, vector databases, and memory backends
- **Tutorials**: Step-by-step guides for common use cases
- **API Reference**: Complete interface and class documentation
- **Extending**: Custom agents, LLM providers, and components

For a quick overview, continue reading this README. For detailed guides and API documentation, visit the full documentation site.

## Philosophy: Developer Authority First

Arshai empowers you to build exactly what you need, when you need it, how you need it. **You control every component, dependency, and interaction.** No magic, no hidden behavior, no framework lock-in.

### The Arshai Way vs Traditional Frameworks

**❌ Traditional Framework Approach:**
```python
# Framework controls everything - what's happening inside?
framework = AIFramework()
framework.load_config("config.yaml") 
agent = framework.create_agent("chatbot")  # Hidden creation
response = agent.chat("Hello")  # Opaque behavior
```

**✅ Arshai Approach - You're In Control:**
```python
# You create, configure, and control everything
llm_config = ILLMConfig(model="gpt-4", temperature=0.7)
llm_client = OpenAIClient(llm_config)  # You create it

agent = ChatAgent(
    llm_client=llm_client,  # You inject dependencies  
    system_prompt="Be helpful",  # You configure behavior
    memory=memory_manager  # You manage state
)

response = await agent.process(input_data)  # You control the flow
```

## Three-Layer Architecture

### Layer 1: LLM Clients (Foundation)
**Minimal Developer Authority** - Standardized access to AI providers

```python
from arshai.llms.openai import OpenAIClient
from arshai.core.interfaces.illm import ILLMConfig, ILLMInput

# Direct instantiation - you create when needed
config = ILLMConfig(model="gpt-4", temperature=0.7)
llm = OpenAIClient(config)

# Unified interface across all providers
llm_input = ILLMInput(
    system_prompt="You are a helpful assistant",
    user_message="What's the weather like?",
    regular_functions={"get_weather": get_weather_func},
    background_tasks={"log_interaction": log_func}  # Fire-and-forget
)

response = await llm.chat(llm_input)
```

**Supported Providers:**
- **OpenAI** - Direct API with streaming and function calling
- **Azure OpenAI** - Enterprise deployment with native parsing
- **Google Gemini** - Native SDK with dual authentication  
- **OpenRouter** - Access to Claude, Llama, and 200+ models
- **Extensible** - Add new providers by implementing `ILLM` interface

### Layer 2: Agents (Business Logic)
**Moderate Developer Authority** - Purpose-driven AI components

```python
from arshai.agents.base import BaseAgent
from arshai.core.interfaces.iagent import IAgentInput

class CustomerServiceAgent(BaseAgent):
    """Your custom agent with your business logic."""
    
    def __init__(self, llm_client: ILLM, knowledge_base: KnowledgeBase):
        super().__init__(llm_client, "You are a customer service expert")
        self.knowledge_base = knowledge_base
    
    async def process(self, input: IAgentInput) -> Dict[str, Any]:
        # Your custom preprocessing
        context = await self._analyze_request(input.message)
        
        # Define tools for this specific agent
        async def search_kb(query: str) -> str:
            return await self.knowledge_base.search(query)
        
        async def log_issue(issue: str) -> None:
            """BACKGROUND TASK: Log customer issue."""
            await self.issue_tracker.create_ticket(issue)
        
        # Your LLM integration with custom tools
        llm_input = ILLMInput(
            system_prompt=self.system_prompt,
            user_message=f"Context: {context}\nUser: {input.message}",
            regular_functions={"search_knowledge": search_kb},
            background_tasks={"log_issue": log_issue}
        )
        
        result = await self.llm_client.chat(llm_input)
        
        # Your custom post-processing
        return {
            "response": result["llm_response"],
            "confidence": self._calculate_confidence(result),
            "escalate": self._should_escalate(result)
        }
```

### Layer 3: Agentic Systems (Complete Control) 
**Maximum Developer Authority** - Orchestrate everything exactly as you need

```python
class CustomerSupportSystem:
    """Your complete system - you control everything."""
    
    def __init__(self):
        # You create all components
        llm = OpenAIClient(ILLMConfig(model="gpt-4"))
        memory = RedisMemoryManager(redis_url=os.getenv("REDIS_URL"))
        knowledge_base = VectorKnowledgeBase(milvus_client, embeddings)
        
        # You compose the system
        self.triage_agent = TriageAgent(llm, "Route customer requests")
        self.support_agent = CustomerServiceAgent(llm, knowledge_base)
        self.escalation_agent = EscalationAgent(llm, ticket_system)
        
        # You define the workflow
        self.memory = memory
        
    async def handle_request(self, user_message: str, user_id: str) -> Dict[str, Any]:
        """Your orchestration logic."""
        
        # Step 1: Retrieve user history
        history = await self.memory.retrieve({"user_id": user_id})
        
        # Step 2: Triage the request
        triage_input = IAgentInput(
            message=user_message,
            metadata={"history": history, "user_id": user_id}
        )
        triage_result = await self.triage_agent.process(triage_input)
        
        # Step 3: Route based on triage (your business logic)
        if triage_result["urgency"] == "high":
            agent = self.escalation_agent
        else:
            agent = self.support_agent
        
        # Step 4: Process with selected agent
        support_input = IAgentInput(
            message=user_message,
            metadata={
                "user_id": user_id,
                "triage": triage_result,
                "history": history
            }
        )
        final_result = await agent.process(support_input)
        
        # Step 5: Update memory (your state management)
        await self.memory.store({
            "user_id": user_id,
            "interaction": {
                "request": user_message,
                "response": final_result["response"],
                "agent_used": agent.__class__.__name__,
                "timestamp": datetime.now().isoformat()
            }
        })
        
        # Your final response format
        return {
            "response": final_result["response"],
            "agent_used": agent.__class__.__name__,
            "escalated": isinstance(agent, EscalationAgent),
            "confidence": final_result.get("confidence", 0.0)
        }
```

## Comprehensive Feature Overview

### 🎯 **Developer Authority**
- **Direct Instantiation**: You create every component explicitly
- **Dependency Injection**: All dependencies passed through constructors
- **No Hidden Magic**: Every behavior is visible in your code
- **Complete Control**: You decide architecture, not the framework
- **Interface-First Design**: Well-defined contracts for easy testing
- **Zero Framework Lock-in**: Use what you need, ignore the rest

### 🤖 **Advanced LLM Client System**

#### **Multi-Provider Support**
- **OpenAI Integration**: Native API with GPT-4o, GPT-4o-mini, and all models
  - Streaming with real-time function execution
  - Advanced function calling with parallel execution
  - Structured outputs with Pydantic models
  - Token usage tracking and cost optimization
  - Rate limiting and error handling

- **Azure OpenAI Enterprise**: Full Azure deployment support
  - Azure authentication and endpoint management
  - Content filtering and compliance features
  - Regional deployment flexibility
  - Enterprise security and audit logging

- **Google Gemini**: Native SDK integration
  - Gemini Pro, Flash, and specialized models
  - Dual authentication (API key and Service Account)
  - Advanced safety settings and content filtering
  - Image and multimodal capabilities
  - Real-time streaming with function calls

- **OpenRouter**: Access to 200+ models including:
  - Anthropic Claude 3.5 Sonnet, Haiku, Opus
  - Meta Llama 3.1, 3.2 (8B, 70B, 405B)
  - Mistral Large, 8x7B, 8x22B
  - Cohere Command R+
  - Specialized models for coding, reasoning, and creativity

#### **Unified Function Calling Architecture**
```python
# Works identically across all providers
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"temperature": 72, "condition": "sunny"}

async def send_notification(message: str) -> None:
    """BACKGROUND TASK: Send notification."""
    # Runs independently, doesn't block conversation
    print(f"📧 Notification: {message}")

# Same interface for any provider
llm_input = ILLMInput(
    system_prompt="You are a helpful assistant",
    user_message="What's the weather in San Francisco?",
    regular_functions={"get_weather": get_weather},
    background_tasks={"send_notification": send_notification}
)

# Works with OpenAI, Azure, Gemini, OpenRouter
response = await llm_client.chat(llm_input)
```

#### **Progressive Streaming**
- **Real-time Function Execution**: Functions execute immediately during streaming
- **Parallel Tool Execution**: Regular tools run concurrently for speed
- **Background Task Management**: Fire-and-forget operations tracked automatically
- **Enhanced Context Building**: Function results integrated into conversation
- **Safe Usage Accumulation**: Robust token and cost tracking during streaming

### 🧠 **Sophisticated Agent System**

#### **Base Agent Capabilities**
- **Flexible Return Types**: Return strings, dicts, generators, or custom objects
- **Stateless Design**: Pure functions with no hidden state
- **Tool Integration**: Easy integration of custom functions and APIs
- **Memory Compatibility**: Works with any memory backend
- **Error Resilience**: Graceful handling of LLM failures

#### **Specialized Agent Types**
```python
# Working Memory Agent - Maintains conversation context
from arshai.agents.working_memory import WorkingMemoryAgent

# Base Agent - Flexible foundation for custom agents
from arshai.agents.base import BaseAgent

# Custom domain-specific agents
class CustomerServiceAgent(BaseAgent):
    def __init__(self, llm_client, knowledge_base, escalation_system):
        super().__init__(llm_client, "Customer service expert")
        self.knowledge_base = knowledge_base
        self.escalation_system = escalation_system
    
    async def process(self, input: IAgentInput) -> Dict[str, Any]:
        # Your custom business logic
        # Access to knowledge base, escalation, metrics
        # Return structured responses with confidence scores
        pass
```

#### **Agent Development Patterns**
- **Domain Specialization**: Create agents for specific business domains
- **Tool Composition**: Combine multiple tools and APIs per agent
- **Response Formatting**: Structure outputs exactly as needed
- **Confidence Scoring**: Built-in or custom confidence assessment
- **Escalation Logic**: Automatic escalation to human agents or specialized systems

### 🔧 **Comprehensive Tool Ecosystem**

#### **Built-in Tools**
- **Web Search**: SearxNG integration with connection pooling
  - Multiple search engines aggregation
  - Result filtering and ranking
  - Performance optimization for production

- **Knowledge Base**: Vector database integration
  - Milvus support with hybrid search
  - Document ingestion and chunking
  - Semantic search with embedding models
  - Metadata filtering and faceted search

- **MCP (Model Context Protocol) Tools**: Dynamic tool loading
  - File system operations (read, write, search)
  - Database connections and queries  
  - API integrations and webhooks
  - Custom tool development framework
  - Thread pool management for performance

#### **Custom Tool Development**
```python
# Function-based tool pattern - simple and powerful
class DatabaseTools:
    """Collection of database tool functions."""

    def __init__(self, db_connection):
        self.db = db_connection

    def search_products(self, query: str, category: str = None) -> List[Dict]:
        """Search products in database."""
        # Your database logic
        return results

    async def log_interaction(self, user_id: str, action: str) -> None:
        """BACKGROUND TASK: Log user interaction."""
        # Background logging, doesn't affect conversation
        await self.db.log(user_id, action)

# Easy integration with agents
db_tools = DatabaseTools(connection)
agent = CustomAgent(
    llm_client,
    regular_functions={"search_products": db_tools.search_products},
    background_tasks={"log_interaction": db_tools.log_interaction}
)
```

### 🧬 **Agentic System Orchestration**

#### **Workflow Management**
- **State-Driven Workflows**: Complex multi-step processes
- **Conditional Branching**: Dynamic execution paths based on results
- **Error Recovery**: Automatic retry and fallback mechanisms
- **Parallel Execution**: Concurrent agent execution for performance
- **State Persistence**: Resume workflows across sessions

#### **Multi-Agent Coordination**
```python
class IntelligentHelpdesk:
    def __init__(self):
        # Create specialized agents
        self.triage_agent = TriageAgent(llm_client)
        self.technical_agent = TechnicalSupportAgent(llm_client, knowledge_base)
        self.billing_agent = BillingAgent(llm_client, billing_system)
        self.escalation_agent = EscalationAgent(llm_client, human_queue)
        
        # Shared resources
        self.memory_manager = RedisMemoryManager()
        self.metrics_collector = MetricsCollector()
    
    async def handle_request(self, request: str, user_id: str) -> Dict[str, Any]:
        # Step 1: Triage determines routing
        triage_result = await self.triage_agent.classify(request)
        
        # Step 2: Route to appropriate specialist
        if triage_result['category'] == 'technical':
            agent = self.technical_agent
        elif triage_result['category'] == 'billing':
            agent = self.billing_agent
        elif triage_result['urgency'] == 'critical':
            agent = self.escalation_agent
        
        # Step 3: Process with context and memory
        context = await self.memory_manager.get_user_context(user_id)
        result = await agent.process_with_context(request, context)
        
        # Step 4: Update memory and metrics
        await self.memory_manager.update_context(user_id, result)
        await self.metrics_collector.record_interaction(result)
        
        return result
```

### 💾 **Advanced Memory Management**

#### **Memory Backend Options**
- **Redis Memory Manager**: Production-ready distributed memory
  - Automatic TTL management
  - Connection pooling and failover
  - Pub/sub for real-time updates
  - Memory cleanup and optimization

- **In-Memory Manager**: Fast development and testing
  - LRU eviction policies
  - Background cleanup tasks
  - Memory usage monitoring

- **Working Memory System**: Context-aware conversation tracking
  - Automatic context summarization
  - Relevance scoring and pruning
  - Cross-session continuity

#### **Memory Usage Patterns**
```python
# Environment-specific memory selection
def create_memory_system(environment: str) -> IMemoryManager:
    if environment == "production":
        return RedisMemoryManager(
            redis_url=os.getenv("REDIS_URL"),
            ttl=3600,  # 1 hour TTL
            key_prefix="app_prod",
            connection_pool_size=10
        )
    elif environment == "development":
        return InMemoryManager(
            ttl=1800,  # 30 minutes
            max_size=1000,
            cleanup_interval=300  # 5 minutes
        )
    else:
        return InMemoryManager(ttl=300)  # Testing

# Advanced conversation memory
class ConversationMemory:
    def __init__(self, memory_manager, summarizer_agent):
        self.memory = memory_manager
        self.summarizer = summarizer_agent
    
    async def store_conversation(self, user_id: str, conversation: List[Dict]):
        # Automatic summarization for long conversations
        if len(conversation) > 10:
            summary = await self.summarizer.summarize(conversation[-10:])
            conversation = conversation[:-10] + [summary]
        
        await self.memory.store({
            "user_id": user_id,
            "conversation": conversation,
            "timestamp": datetime.now().isoformat()
        })
```

### 📊 **Production-Grade Observability**

#### **OpenTelemetry Integration**
- **Automatic Instrumentation**: No code changes required
- **Distributed Tracing**: Track requests across services
- **Custom Metrics**: Business-specific monitoring
- **Export Flexibility**: Send to any OTLP-compatible backend

#### **Built-in Metrics**
```python
# Automatic metrics collection
metrics_collected = {
    "llm_requests_total": "Counter of LLM API calls",
    "llm_request_duration_seconds": "Histogram of request latencies",
    "llm_tokens_total": "Counter of tokens used (input/output/total)",
    "llm_cost_total": "Counter of API costs by provider",
    "llm_time_to_first_token_seconds": "Time to start streaming",
    "llm_time_to_last_token_seconds": "Total generation time",
    "agent_processing_duration_seconds": "Agent processing time",
    "memory_operations_total": "Memory store/retrieve operations",
    "tool_execution_duration_seconds": "Tool execution timing",
    "background_task_duration_seconds": "Background task timing"
}

# Zero-fallback monitoring - always works
from arshai.observability import ObservabilityConfig

# Optional: Advanced configuration
obs_config = ObservabilityConfig(
    endpoint="http://jaeger:14268/api/traces",
    service_name="arshai_app",
    environment="production",
    custom_metrics={"business_metric": "counter"}
)

# Automatic instrumentation
client = OpenAIClient(
    config=llm_config,
    observability_config=obs_config  # Optional
)

# Metrics automatically exported to your monitoring system
```

#### **Performance Monitoring**
- **Real-time Dashboards**: Track system health and performance
- **Alert Configuration**: Custom alerts for SLA violations
- **Cost Tracking**: Detailed cost breakdown by model and usage
- **Error Rate Monitoring**: Track and analyze failure patterns
- **Capacity Planning**: Usage trends and scaling recommendations

### 🗄️ **Vector Database Integration**

#### **Milvus Support**
- **Hybrid Search**: Dense + sparse vector combination
  - Semantic similarity with dense vectors
  - Keyword matching with sparse vectors
  - Weighted ranking and reranking

- **Production Features**:
  - Connection pooling and async operations
  - Batch insertion with configurable batch sizes
  - Automatic index creation and optimization
  - Metadata filtering and complex queries
  - Schema versioning and migration support

```python
# Advanced vector operations
from arshai.vector_db.milvus_client import MilvusClient
from arshai.core.interfaces.ivector_db_client import ICollectionConfig

# Async operations for performance
client = MilvusClient()
config = ICollectionConfig(
    collection_name="knowledge_base",
    dense_dim=1536,  # OpenAI embeddings
    is_hybrid=True,  # Enable sparse + dense
    schema_model=CustomDocument  # Structured metadata
)

# Hybrid search with filtering
results = await client.hybrid_search(
    config=config,
    dense_vectors=dense_embeddings,
    sparse_vectors=sparse_embeddings,
    expr="metadata['category'] == 'technical'",
    limit=10
)
```

### 📦 **Document Processing Pipeline**

#### **Ingestion and Chunking**
- **Multi-format Support**: PDF, DOCX, TXT, HTML, Markdown
- **Intelligent Chunking**: Semantic and structural chunking
- **Metadata Extraction**: Automatic metadata detection
- **Batch Processing**: Efficient large-scale ingestion
- **Deduplication**: Content-based duplicate detection

#### **Embedding Generation**
- **Multiple Providers**: OpenAI, Google, Hugging Face
- **Model Selection**: Choose optimal models per use case
- **Batch Processing**: Efficient embedding generation
- **Caching**: Avoid re-processing unchanged content

### 🚀 **Enterprise Performance & Scalability**

#### **Connection Management**
- **HTTP Connection Pooling**: Configurable limits prevent resource exhaustion
  - Default: 100 max connections, 10 per host
  - Configurable via environment variables
  - Automatic connection reuse and cleanup

- **Thread Pool Management**: Bounded execution prevents deadlocks
  - CPU-aware thread limits: `min(32, cpu_count * 2)`
  - Shared pools across MCP tools
  - Graceful shutdown and cleanup

#### **Async Architecture**
- **Non-blocking Operations**: All I/O operations are async
- **Concurrent Processing**: Parallel function execution
- **Background Tasks**: Fire-and-forget operations
- **Resource Management**: Automatic cleanup and monitoring

#### **Production Optimizations**
```python
# Environment-based configuration
os.environ.update({
    'ARSHAI_MAX_CONNECTIONS': '200',        # HTTP connection pool
    'ARSHAI_MAX_CONNECTIONS_PER_HOST': '20', # Per-host limit
    'ARSHAI_MAX_THREADS': '64',             # Thread pool size
    'ARSHAI_MAX_MEMORY_MB': '4096',         # Memory limit
    'ARSHAI_CLEANUP_INTERVAL': '300',       # Background cleanup
    'ARSHAI_BATCH_SIZE': '100'              # Vector DB batch size
})

# Production deployment patterns
class ProductionApp:
    def __init__(self):
        # Shared resources for efficiency
        self.llm_pool = {
            'fast': [OpenAIClient(fast_config) for _ in range(3)],
            'powerful': [OpenAIClient(powerful_config) for _ in range(2)]
        }
        
        self.memory_manager = RedisMemoryManager(
            redis_url=os.getenv("REDIS_URL"),
            connection_pool_size=20,
            ttl=7200
        )
        
        # Load balancing and circuit breakers
        self.load_balancer = LoadBalancer(self.llm_pool)
        self.circuit_breaker = CircuitBreaker(failure_threshold=5)
    
    async def handle_concurrent_requests(self, requests: List[str]) -> List[str]:
        # Process up to 100 concurrent requests
        semaphore = asyncio.Semaphore(100)
        
        async def process_one(request):
            async with semaphore:
                agent = self.create_agent()
                return await agent.process(IAgentInput(message=request))
        
        tasks = [process_one(req) for req in requests]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
```

### 🛠️ **Development and Testing**

#### **Testing Framework**
- **Mock Support**: Easy mocking of LLM clients and dependencies
- **Integration Testing**: Real LLM integration tests
- **Load Testing**: Performance and scalability validation
- **Async Testing**: Full async/await test support

```python
# Easy testing with dependency injection
async def test_customer_agent():
    mock_llm = AsyncMock()
    mock_llm.chat.return_value = {
        "llm_response": "Test response",
        "usage": {"total_tokens": 50}
    }
    
    mock_kb = AsyncMock()
    mock_kb.search.return_value = ["relevant info"]
    
    # Create agent with mocked dependencies
    agent = CustomerServiceAgent(mock_llm, mock_kb)
    
    result = await agent.process(IAgentInput(
        message="Help with billing",
        metadata={"user_id": "123"}
    ))
    
    assert "Test response" in result["response"]
    mock_llm.chat.assert_called_once()
    mock_kb.search.assert_called_once()
```

#### **Development Patterns**
- **Configuration Management**: Environment-based settings
- **Hot Reloading**: Development server with auto-reload
- **Debugging Tools**: Comprehensive logging and tracing
- **Performance Profiling**: Built-in performance analysis

## Installation & Setup

### Basic Installation
```bash
# Core framework only
pip install arshai

# With all optional dependencies
pip install arshai[all]  # Includes all features below

# Specific feature sets
pip install arshai[redis,milvus,observability]  # Production essentials
pip install arshai[docs]                        # Documentation building
```

### Feature-Specific Installations
```bash
# Memory backends
pip install arshai[redis]     # Redis memory manager

# Vector databases  
pip install arshai[milvus]    # Milvus vector database

# Advanced search
pip install arshai[flashrank] # Re-ranking capabilities

# Monitoring
pip install arshai[observability] # OpenTelemetry integration

# Development tools
pip install arshai[dev]       # Testing and development utilities
```

### Environment Setup
```bash
# LLM Provider API Keys
export OPENAI_API_KEY="your-openai-key"
export AZURE_OPENAI_API_KEY="your-azure-key"
export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
export GOOGLE_API_KEY="your-gemini-key"
export OPENROUTER_API_KEY="your-openrouter-key"

# Optional: Google Service Account (alternative to API key)
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

# Memory backends
export REDIS_URL="redis://localhost:6379"
export REDIS_TTL="3600"  # 1 hour

# Vector database
export MILVUS_HOST="localhost"
export MILVUS_PORT="19530"
export MILVUS_DB_NAME="default"

# Performance tuning
export ARSHAI_MAX_CONNECTIONS="100"
export ARSHAI_MAX_THREADS="32"
export ARSHAI_BATCH_SIZE="50"

# Observability
export OTEL_EXPORTER_OTLP_ENDPOINT="http://jaeger:14268"
export OTEL_SERVICE_NAME="arshai_app"
```

### Docker Setup
```dockerfile
# Production Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install production dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Performance configuration
ENV ARSHAI_MAX_CONNECTIONS=200
ENV ARSHAI_MAX_CONNECTIONS_PER_HOST=20
ENV ARSHAI_MAX_THREADS=64
ENV ARSHAI_MAX_MEMORY_MB=4096

# Redis connection
ENV REDIS_URL=redis://redis:6379
ENV REDIS_TTL=7200

# Milvus connection
ENV MILVUS_HOST=milvus
ENV MILVUS_PORT=19530

COPY . .
CMD ["python", "app.py"]
```

### Docker Compose for Development
```yaml
# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379
      - MILVUS_HOST=milvus
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - redis
      - milvus
      - jaeger
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  
  milvus:
    image: milvusdb/milvus:latest
    ports:
      - "19530:19530"
    environment:
      - ETCD_ENDPOINTS=etcd:2379
  
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"
      - "14268:14268"
```

## Quick Start Guide 🚀

### 1. Quick Start - Your First Agent

```python
from arshai.llms.openai import OpenAIClient
from arshai.agents.base import BaseAgent
from arshai.core.interfaces.illm import ILLMConfig
from arshai.core.interfaces.iagent import IAgentInput

# Step 1: Create your LLM client (you control this)
llm_config = ILLMConfig(
    model="gpt-4o",           # Latest model
    temperature=0.7,         # Creativity level
    max_tokens=2000          # Response length
)
llm_client = OpenAIClient(llm_config)

# Step 2: Create your agent (you define the behavior)
agent = BaseAgent(
    llm_client=llm_client, 
    system_prompt="You are an expert Python programming assistant. Provide clear, practical examples with explanations."
)

# Step 3: Use your agent (you control the interaction)
response = await agent.process(IAgentInput(
    message="How do I handle async/await in Python? Show me a practical example."
))

print(response)
# Output: Detailed explanation with code examples
```

### 1b. Agent with Memory (Conversation Context)

```python
from arshai.agents.working_memory import WorkingMemoryAgent
from arshai.memory.working_memory.in_memory_manager import InMemoryManager

# Create memory system for conversation context
memory = InMemoryManager(ttl=1800)  # 30-minute memory

# Create agent with memory capabilities
conversation_agent = WorkingMemoryAgent(
    llm_client=llm_client,
    memory_manager=memory,
    system_prompt="You are a helpful assistant that remembers our conversation context."
)

# First interaction
response1 = await conversation_agent.process(IAgentInput(
    message="My name is Alex and I'm learning Python.",
    metadata={"conversation_id": "conv_123"}
))

# Second interaction - agent remembers context
response2 = await conversation_agent.process(IAgentInput(
    message="What was my name again?",
    metadata={"conversation_id": "conv_123"}
))

print(response2)  # Will remember "Alex"
```

### 1c. Multi-Provider Setup

```python
# Use different providers for different tasks
from arshai.llms.openai import OpenAIClient
from arshai.llms.google_genai import GeminiClient
from arshai.llms.openrouter import OpenRouterClient

# Fast model for simple tasks
fast_llm = OpenAIClient(ILLMConfig(model="gpt-4o-mini", temperature=0.3))

# Powerful model for complex analysis
strong_llm = OpenAIClient(ILLMConfig(model="gpt-4o", temperature=0.5))

# Alternative provider for variety
gemini_llm = GeminiClient(ILLMConfig(model="gemini-1.5-pro", temperature=0.7))

# Access to specialized models
openrouter_llm = OpenRouterClient(ILLMConfig(
    model="anthropic/claude-3.5-sonnet",  # Claude via OpenRouter
    temperature=0.6
))

# Create specialized agents
quick_agent = BaseAgent(fast_llm, "Quick responses for simple questions")
analysis_agent = BaseAgent(strong_llm, "Deep analysis and complex reasoning")
creative_agent = BaseAgent(gemini_llm, "Creative writing and brainstorming")
research_agent = BaseAgent(openrouter_llm, "Research and detailed explanations")

# Use the right agent for each task
quick_response = await quick_agent.process(IAgentInput(
    message="What's 15% tip on $42?"
))

analysis_response = await analysis_agent.process(IAgentInput(
    message="Analyze the pros and cons of microservices vs monolithic architecture for a startup"
))
```

### 2. Advanced Agent with Multiple Tool Types

```python
# Comprehensive restaurant agent with multiple capabilities
class SmartRestaurantAgent(BaseAgent):
    """Intelligent restaurant agent with real-time data and background tasks."""
    
    def __init__(self, llm_client: ILLM, reservation_system, pos_system, notification_service):
        super().__init__(llm_client, "You are an expert restaurant assistant with access to real-time data")
        self.reservation_system = reservation_system
        self.pos_system = pos_system
        self.notification_service = notification_service
    
    async def process(self, input: IAgentInput) -> Dict[str, Any]:
        # Define regular tools (return results to conversation)
        def get_current_time() -> Dict[str, str]:
            """Get current time with timezone info."""
            from datetime import datetime
            import pytz
            
            now = datetime.now(pytz.UTC)
            local_time = now.astimezone(pytz.timezone('America/New_York'))
            
            return {
                "utc_time": now.isoformat(),
                "local_time": local_time.strftime("%I:%M %p"),
                "date": local_time.strftime("%A, %B %d, %Y"),
                "timezone": "Eastern Time"
            }
        
        def calculate_bill_details(subtotal: float, tax_rate: float = 0.08, tip_percent: float = 20) -> Dict[str, float]:
            """Calculate comprehensive bill breakdown."""
            tax = subtotal * tax_rate
            tip = subtotal * (tip_percent / 100)
            total = subtotal + tax + tip
            
            return {
                "subtotal": round(subtotal, 2),
                "tax": round(tax, 2),
                "tip": round(tip, 2),
                "total": round(total, 2),
                "tip_percentage": tip_percent
            }
        
        async def check_table_availability(party_size: int, preferred_time: str) -> Dict[str, Any]:
            """Check real-time table availability."""
            # Your reservation system integration
            available_times = await self.reservation_system.check_availability(
                party_size=party_size,
                date=preferred_time
            )
            
            return {
                "available": len(available_times) > 0,
                "available_times": available_times[:5],  # Show top 5 options
                "party_size": party_size,
                "wait_time_estimate": "15-20 minutes" if available_times else "45+ minutes"
            }
        
        async def get_menu_recommendations(dietary_restrictions: str = None, budget_range: str = "moderate") -> List[Dict[str, Any]]:
            """Get personalized menu recommendations."""
            # Your menu system integration with filtering
            menu_items = await self.pos_system.get_filtered_menu(
                dietary_restrictions=dietary_restrictions,
                price_range=budget_range
            )
            
            return [
                {
                    "name": item["name"],
                    "description": item["description"],
                    "price": item["price"],
                    "dietary_tags": item["tags"],
                    "popularity_score": item["popularity"]
                }
                for item in menu_items[:8]  # Top 8 recommendations
            ]
        
        # Define background tasks (fire-and-forget)
        async def log_customer_interaction(user_id: str, interaction_type: str, details: Dict[str, Any]) -> None:
            """BACKGROUND TASK: Log interaction for analytics."""
            await self.pos_system.log_interaction({
                "user_id": user_id,
                "type": interaction_type,
                "details": details,
                "timestamp": datetime.now().isoformat()
            })
        
        async def send_follow_up_notification(user_id: str, message: str, delay_minutes: int = 60) -> None:
            """BACKGROUND TASK: Schedule follow-up notification."""
            await self.notification_service.schedule_notification(
                user_id=user_id,
                message=message,
                delay_minutes=delay_minutes
            )
        
        async def update_customer_preferences(user_id: str, preferences: Dict[str, Any]) -> None:
            """BACKGROUND TASK: Update customer profile."""
            await self.pos_system.update_customer_profile(
                user_id=user_id,
                preferences=preferences
            )
        
        async def trigger_marketing_follow_up(user_id: str, interaction_context: str) -> None:
            """BACKGROUND TASK: Trigger marketing automation."""
            # Your CRM/marketing automation integration
            await self.notification_service.trigger_campaign(
                user_id=user_id,
                campaign_type="restaurant_follow_up",
                context=interaction_context
            )
        
        # Prepare LLM input with all tools
        llm_input = ILLMInput(
            system_prompt="""You are a sophisticated restaurant assistant with access to real-time systems.
            You can check availability, provide menu recommendations, calculate bills, and help with reservations.
            Always be helpful, accurate, and proactive in offering relevant information.
            When customers show interest, set up appropriate follow-ups.""",
            
            user_message=input.message,
            
            # Regular functions return results to continue conversation
            regular_functions={
                "get_current_time": get_current_time,
                "calculate_bill_details": calculate_bill_details,
                "check_table_availability": check_table_availability,
                "get_menu_recommendations": get_menu_recommendations
            },
            
            # Background tasks run independently
            background_tasks={
                "log_customer_interaction": log_customer_interaction,
                "send_follow_up_notification": send_follow_up_notification,
                "update_customer_preferences": update_customer_preferences,
                "trigger_marketing_follow_up": trigger_marketing_follow_up
            }
        )
        
        # Get LLM response with tool usage
        result = await self.llm_client.chat(llm_input)
        
        # Return structured response
        return {
            "response": result["llm_response"],
            "tools_used": result.get("tools_used", []),
            "background_tasks_triggered": result.get("background_tasks", []),
            "user_id": input.metadata.get("user_id"),
            "session_id": input.metadata.get("session_id"),
            "timestamp": datetime.now().isoformat()
        }

# Production usage
async def main():
    # Create dependencies