ambivo-agents 1.4.0
Multi-Agent AI System for automation
License: MIT
Requires Python: >=3.11

# Ambivo Agents - Multi-Agent AI System

A toolkit for AI-powered automation including data analytics with DuckDB, media processing, knowledge base operations, web scraping, YouTube downloads, HTTP/REST API integration, and more.

## Alpha Release Disclaimer

**This library is currently in alpha stage.** While functional, it may contain bugs, undergo breaking changes, and lack complete documentation. Developers should thoroughly evaluate and test the library before considering it for production use.

For production scenarios, we recommend:
- Extensive testing in your specific environment
- Implementing proper error handling and monitoring
- Having rollback plans in place
- Staying updated with releases for critical fixes

## Table of Contents

- [Quick Start](#quick-start)
- [Agent Creation](#agent-creation)
- [Features](#features)
- [Available Agents](#available-agents)
- [Workflow System](#workflow-system)
- [System Messages](#system-messages)
- [Prerequisites](#prerequisites)
- [Installation](#installation)
- [Configuration](#configuration)
- [Configuration Methods](#configuration-methods)
- [Project Structure](#project-structure)
- [Usage Examples](#usage-examples)
- [Streaming System](#streaming-system)
- [Session Management](#session-management)
- [Web API Integration](#web-api-integration)
- [Command Line Interface](#command-line-interface)
- [Architecture](#architecture)
- [Docker Setup](#docker-setup)
  - [Agent Handoff Mechanism](#agent-handoff-mechanism)
  - [File Access Security Configuration](#file-access-security-configuration)
- [Troubleshooting](#troubleshooting)
- [Security Considerations](#security-considerations)
- [Contributing](#contributing)
- [License](#license)
- [Author](#author)
- [Support](#support)

## Quick Start

### ModeratorAgent Example

The **ModeratorAgent** automatically routes queries to specialized agents:

```python
from ambivo_agents import ModeratorAgent
import asyncio

async def main():
    # Create the moderator
    moderator, context = ModeratorAgent.create(user_id="john")
    
    print(f"Session: {context.session_id}")
    
    # Auto-routing examples
    response1 = await moderator.chat("Download audio from https://youtube.com/watch?v=example")
    response2 = await moderator.chat("Search for latest AI trends")
    response3 = await moderator.chat("Extract audio from video.mp4 as MP3")
    response4 = await moderator.chat("GET https://api.github.com/users/octocat")
    response5 = await moderator.chat("Load data from sales.csv and analyze it")
    response6 = await moderator.chat("What is machine learning?")
    
    # Check available agents
    status = await moderator.get_agent_status()
    print(f"Available agents: {list(status['active_agents'].keys())}")
    
    # Cleanup
    await moderator.cleanup_session()

asyncio.run(main())
```

### KnowledgeSynthesisAgent Example

The **KnowledgeSynthesisAgent** combines multiple knowledge sources with quality assessment:

```python
from ambivo_agents.agents.knowledge_synthesis import KnowledgeSynthesisAgent
from ambivo_agents.agents.response_quality_assessor import QualityLevel
import asyncio

async def main():
    # Create synthesis agent directly (NOT through ModeratorAgent)
    synthesis_agent, context = KnowledgeSynthesisAgent.create(
        user_id="researcher",
        quality_threshold=QualityLevel.GOOD,
        max_iterations=3,
        enable_auto_scraping=True
    )
    
    print(f"Session: {context.session_id}")
    
    # Multi-source synthesis examples using process_with_quality_assessment
    response1 = await synthesis_agent.process_with_quality_assessment(
        "search across knowledge bases: kb1, kb2 for: AI trends"
    )
    
    response2 = await synthesis_agent.process_with_quality_assessment(
        "prioritize web search - latest developments in robotics 2025"
    )
    
    response3 = await synthesis_agent.process_with_quality_assessment(
        "comprehensive search - research blockchain technology"
    )
    
    # Check quality assessment
    print(f"Quality: {response1.get('quality_assessment', {}).get('quality_level', 'unknown')}")
    print(f"Sources: {response1.get('quality_assessment', {}).get('sources_used', [])}")
    
    # Cleanup
    await synthesis_agent.cleanup_session()

asyncio.run(main())
```

### Simple KnowledgeSynthesisAgent with Specific Knowledge Bases

For targeted knowledge base queries with automatic collection detection:

```python
from ambivo_agents.agents.knowledge_synthesis import KnowledgeSynthesisAgent
import asyncio

async def simple_example():
    # Setup - Create agent and configure knowledge bases
    agent, context = KnowledgeSynthesisAgent.create(user_id="user123")
    agent.update_context_metadata(available_knowledge_bases=[
        'research_trends_in_cryptocurrency_20250816_193439',
        'research_trends_in_robotics_tech_2025_20250812_172007'
    ])
    agent._load_available_collections()  # Required after setting knowledge bases
    
    try:
        # Method 1: Full quality assessment (recommended)
        result = await agent.process_with_quality_assessment(
            "What are cryptocurrency trends?"
        )
        print(f"Response: {result['response']}")
        print(f"Quality: {result['quality_assessment']['quality_level']}")
        print(f"Sources: {result['quality_assessment']['sources_used']}")
        
        # Method 2: Simple chat interface
        response = await agent.chat("Tell me about robotics developments")
        print(f"Chat response: {response}")
        
    finally:
        # Always cleanup
        await agent.cleanup_session()

asyncio.run(simple_example())
```

### Command Line Usage

```bash
# Install and run
pip install ambivo-agents

# Interactive mode
ambivo-agents

# Single commands
ambivo-agents -q "Download audio from https://youtube.com/watch?v=example"
ambivo-agents -q "Search for Python tutorials"
ambivo-agents -q "Load data from sales.csv and analyze it"
ambivo-agents -q "GET https://jsonplaceholder.typicode.com/posts/1"
```

## Agent Creation

### ModeratorAgent (Recommended)

```python
from ambivo_agents import ModeratorAgent

# Create moderator with auto-routing
moderator, context = ModeratorAgent.create(user_id="john")

# Chat with automatic agent selection
result = await moderator.chat("Download audio from https://youtube.com/watch?v=example")

# Cleanup
await moderator.cleanup_session()
```

**Use ModeratorAgent for:**
- Multi-purpose applications
- Intelligent routing between capabilities
- Context-aware conversations
- Simplified development

### Direct Agent Creation

```python
from ambivo_agents import YouTubeDownloadAgent

# Create specific agent
agent, context = YouTubeDownloadAgent.create(user_id="john")

# Use agent directly
result = await agent._download_youtube_audio("https://youtube.com/watch?v=example")

# Cleanup
await agent.cleanup_session()
```

**Use Direct Creation for:**
- Single-purpose applications
- Specific workflows with known requirements
- Performance-critical applications
- Custom integrations

## Features

### Core Capabilities
- **ModeratorAgent**: Intelligent multi-agent orchestrator with automatic routing
- **Smart Routing**: Automatically routes queries to appropriate specialized agents
- **Data Analytics**: In-memory DuckDB integration with CSV/XLS ingestion and text-based visualizations
- **File Ingestion & Processing**: All agents can read/parse JSON, CSV, XML, YAML files and insert into databases
- **Context Memory**: Maintains conversation history across interactions
- **Docker Integration**: Secure, isolated execution environment
- **Redis Memory**: Persistent conversation memory with compression
- **Multi-Provider LLM**: Automatic failover between OpenAI, Anthropic, and AWS Bedrock
- **Configuration-Driven**: All features controlled via `agent_config.yaml`
- **Workflow System**: Multi-agent workflows with parallel and sequential execution
- **System Messages**: Customizable system prompts for agent behavior control

## Available Agents

### ModeratorAgent 
**Enhanced with Skill Assignment System**
- Intelligent orchestrator that routes to specialized agents
- Context-aware multi-turn conversations
- Automatic agent selection based on query analysis
- Session management and cleanup
- Workflow execution and coordination
- ✨ **NEW: Skill Assignment** - Assign external capabilities that take priority over agent routing
- ✨ **Smart Skill Routing** - Assigned skills are checked first, then falls back to normal agent routing
- ✨ **Unified Interface** - Single agent that can handle both assigned skills and general orchestration
- 🎯 **Use Cases**: Custom integrations with existing agent orchestration, priority skill handling

### KnowledgeSynthesisAgent 🧠
**Advanced Multi-Source Orchestrator with Quality Assessment**

**🚨 IMPORTANT**: KnowledgeSynthesisAgent is **NOT** a specialized agent that gets routed to by ModeratorAgent. It **IS** an enhanced ModeratorAgent that you use **directly** as your primary orchestrator.

**Key Capabilities:**
- 🔗 **Multi-Source Synthesis** - Intelligently combines knowledge bases, web search, and web scraping
- 📊 **Quality Assessment** - Built-in ResponseQualityAssessor evaluates and improves responses
- 🎯 **Adaptive Strategy** - Automatically selects optimal search strategy based on query analysis
- 📚 **Multiple Knowledge Base Support** - Query across multiple knowledge bases simultaneously
- 🌐 **Web Search Integration** - Supplements knowledge with real-time web search results
- 🕷️ **Smart Web Scraping** - Automatically scrapes relevant URLs for deeper information
- ⚡ **Iterative Improvement** - Refines responses until quality threshold is met
- 🧠 **Intelligent Routing** - Inherits all ModeratorAgent capabilities plus synthesis features

**Architecture:**
```
KnowledgeSynthesisAgent extends ModeratorAgent
├── Inherits: Agent routing, session management, workflows
└── Adds: Multi-source synthesis, quality assessment, intelligent orchestration
```

**Key Differences:**
| Feature | ModeratorAgent | KnowledgeSynthesisAgent |
|---------|----------------|------------------------|
| **Purpose** | Routes queries to specialized agents | Multi-source information synthesis |
| **Response** | Simple agent routing | Quality-assessed, synthesized responses |
| **Sources** | Single agent per query | Multiple sources per query |
| **Quality Control** | Basic agent responses | Built-in quality assessment & iteration |
| **Use Case** | General orchestration | Research & information synthesis |
| **Usage Pattern** | Router/orchestrator | Direct primary agent |

**❌ Wrong Usage Pattern:**
```python
# DON'T DO THIS - Won't work!
moderator = ModeratorAgent.create(...)
# Expecting to route to KnowledgeSynthesisAgent - this fails
response = await moderator.chat("synthesize information...")  # ❌ Fails
```

**✅ Correct Usage Pattern:**
```python
# DO THIS - Works correctly!
from ambivo_agents.agents.knowledge_synthesis import KnowledgeSynthesisAgent

# Use KnowledgeSynthesisAgent directly as your orchestrator
synthesis_agent, context = KnowledgeSynthesisAgent.create(
    user_id="user123",
    quality_threshold=QualityLevel.GOOD,
    max_iterations=3,
    enable_auto_scraping=True
)

# Multi-knowledge base query
response = await synthesis_agent.process_with_quality_assessment(
    "search across knowledge bases: kb1, kb2 for: AI trends"
)

# Web-supplemented query  
response = await synthesis_agent.process_with_quality_assessment(
    "prioritize web search - latest developments in robotics 2025"
)

# Comprehensive research
response = await synthesis_agent.process_with_quality_assessment(
    "comprehensive search - research blockchain technology trends"
)

await synthesis_agent.cleanup_session()
```

**Quality-Assessed Response Format:**
```python
{
    'success': True,
    'response': 'Synthesized answer from multiple sources...',
    'quality_assessment': {
        'quality_level': 'good',           # excellent|good|fair|poor|unacceptable
        'confidence_score': 0.85,         # 0.0 - 1.0
        'sources_used': ['knowledge_base', 'web_search'],
        'strengths': ['Comprehensive coverage', 'Current information'],
        'weaknesses': ['Could use more specific examples']
    },
    'query_analysis': {
        'query_type': 'current_events',
        'strategy_used': 'web_first',
        'sources_consulted': 2
    },
    'metadata': {
        'iterations': 1,
        'total_sources_consulted': 2,
        'user_preferences': {}
    }
}
```

**Query Patterns:**
- `"search across knowledge bases: kb1, kb2, kb3 for: [query]"` - Multi-KB search
- `"prioritize web search - [query]"` - Web-first strategy
- `"check knowledge base first - [query]"` - KB-first strategy  
- `"comprehensive search - [query]"` - Use all sources
- `"synthesize information about [topic]"` - Full synthesis mode

**🎯 Use Cases:**
- Research across multiple knowledge bases
- Web-supplemented knowledge queries
- Quality-assured information synthesis
- Multi-source fact checking
- Comprehensive topic research

**📋 Configuration:**
- **Enabled by default** (inherits from ModeratorAgent)
- **Environment Control**: `AMBIVO_AGENTS_MODERATOR_ENABLED=true/false`
- **No separate enablement needed** - uses existing ModeratorAgent configuration

---

## 📚 Multi-Knowledge Base Processing Logic

Understanding how multiple knowledge bases are handled across different agents:

### **ModeratorAgent - Simple Routing**
```
User Query → Intent Analysis → Route to KnowledgeBaseAgent → Return Result
```
- **No multi-KB logic** - simply routes to KnowledgeBaseAgent
- **Single hop** - passes query through without modification
- **Fallback routing** - falls back to AssistantAgent if KB agent fails

### **KnowledgeBaseAgent - Advanced Multi-KB Selection**
```
Query → Normalize KB List → Score KBs → Select Top 2 → Query Each → Pick Best by Source Count → Return
```

**🔍 Detailed Flow (Single-Pass, No Iteration):**

**Step 1: KB Input Normalization**
```python
# Accepts multiple input formats:
kb_names = ["kb1", "kb2"]                           # List of strings
kb_names = [{"kb_name": "kb1", "description": "..."}, ...] # List of objects
kb_names = '["kb1", "kb2"]'                         # JSON string

# Normalizes to: [{"kb_name": "kb1", "description": None}, ...]
```

**Step 2: KB Scoring Algorithm**
```python
def score_kb_for_query(kb_entry, user_message, topics):
    # Extract meaningful keywords (>3 chars, no stopwords)
    keywords = extract_keywords(user_message + topics)
    
    score = 0.0
    kb_text = f"{kb_name} {kb_description}".lower()
    
    # Base scoring: keyword overlap
    for keyword in keywords:
        if keyword in kb_text:
            score += 1.0
    
    # Domain bonuses
    if "legal" in kb_text and legal_keywords_present:
        score += 1.5
    if "finance" in kb_text and finance_keywords_present:  
        score += 1.0
    
    return score

# Example scoring:
# Query: "blockchain technology trends"
# - crypto_research_kb: score = 2.0 (matches "blockchain", "technology")
# - legal_documents_kb: score = 0.0 (no matches)
# - tech_trends_kb: score = 2.0 (matches "technology", "trends")
```

**Step 3: KB Selection (Performance Optimization)**
```python
# Sort by score and select top 2 candidates
scored_kbs = [(kb, score) for kb, score in kb_scores]
scored_kbs.sort(key=lambda x: x[1], reverse=True)

# Select top 2 with positive scores
top_kbs = [kb for kb, score in scored_kbs if score > 0][:2]

# Fallback: use first 2 if no positive scores
if not top_kbs:
    top_kbs = original_kb_list[:2]
```

**Step 4: Multi-KB Querying (Single Pass - No Iteration Loop)**
```python
async def query_multiple_kbs(selected_kbs, query):
    results = {}
    best_answer = None
    best_source_count = -1
    
    # Process each KB exactly once (no retries or iterations)
    for kb in selected_kbs:  # Usually 2 KBs
        # Single attempt per KB
        result = await query_single_kb(kb.name, query)
        results[kb.name] = result
        
        if result.success:
            source_count = len(result.source_details)
            # Winner = KB with most document sources
            if source_count > best_source_count:
                best_answer = result.answer
                best_source_count = source_count
                best_kb = kb.name
    
    return best_answer, full_metadata
```

**Step 5: Best Answer Selection**
- **Criteria**: Number of source documents (not quality assessment)
- **Winner takes all**: Returns single best answer (no synthesis)
- **Metadata**: Includes results from all queried KBs

**🚨 Important Characteristics:**
- ❌ **No iteration loops** - each KB queried exactly once
- ❌ **No quality assessment** - purely source count based
- ❌ **No answer synthesis** - returns single best answer
- ✅ **Performance optimized** - maximum 2 KBs processed
- ✅ **Deterministic** - same query always produces same KB selection

### **KnowledgeSynthesisAgent - Multi-Source with Quality Iteration**
```
Query → Analyze Strategy → Gather from All Sources → Quality Assessment → [Iterate if Needed] → Synthesize Final Answer
```

**🔄 Detailed Flow (Iterative with Quality Control):**

**Step 1: Query Analysis & Strategy**
```python
analysis = await analyze_query(query)
# Determines: knowledge_first, web_first, parallel, or adaptive strategy
# Based on: query type, time sensitivity, complexity
```

**Step 2: Multi-Source Gathering**
```python
# Parallel execution across all source types
sources = await asyncio.gather(
    gather_from_knowledge_base(query),    # Routes to KnowledgeBaseAgent
    gather_from_web_search(query),        # Routes to WebSearchAgent  
    gather_from_web_scraping(query)       # Routes to WebScraperAgent
)
```

**Step 3: Quality Assessment & Iteration Loop**
```python
for iteration in range(max_iterations):  # Usually 3 iterations max
    assessment = await quality_assessor.assess_response(sources, query)
    
    # Check if quality meets threshold
    if assessment.quality_level >= quality_threshold:
        break  # Success - exit iteration loop
        
    # If quality insufficient, gather additional sources
    if assessment.needs_additional_sources:
        for suggested_source in assessment.suggested_sources:
            additional_response = await gather_from_source(suggested_source, query)
            sources.append(additional_response)

# Final synthesis of all gathered information
final_answer = await synthesize_responses(sources, query)
```

**Step 4: Quality-Driven Source Integration**
- **Quality metrics**: accuracy, completeness, relevance, currency
- **Iterative refinement**: adds sources until quality threshold met
- **Intelligent synthesis**: combines information from all sources
- **Source attribution**: tracks which sources contributed to answer

### **Comparison Summary**

| **Aspect** | **ModeratorAgent** | **KnowledgeBaseAgent** | **KnowledgeSynthesisAgent** |
|------------|-------------------|------------------------|----------------------------|
| **Multi-KB Handling** | None (routes only) | Advanced selection logic | Routes to KB + other sources |
| **KB Selection** | No selection | Score-based top 2 selection | Uses KB agent's selection |
| **Iteration** | No | No (single-pass) | Yes (quality-driven) |
| **Answer Selection** | First response | Most sources wins | Quality assessment |
| **Synthesis** | None | None (single answer) | Multi-source synthesis |
| **Performance** | Fast (single route) | Fast (max 2 KBs) | Slower (comprehensive) |
| **Quality Control** | Basic | Source count based | Advanced quality assessment |
| **Use Case** | General routing | Multi-KB queries | Research & comprehensive answers |

### **When to Use Which Agent**

**Use ModeratorAgent when:**
- Simple routing to appropriate agents
- General-purpose applications
- Fast response times needed

**Use KnowledgeBaseAgent directly when:**  
- Specifically need to query multiple knowledge bases
- Want deterministic KB selection logic
- Need metadata about which KBs were used

**Use KnowledgeSynthesisAgent when:**
- Need comprehensive, high-quality answers
- Want information from multiple source types
- Quality assessment and iteration is important
- Research-oriented tasks requiring synthesis

### Database Agent (Optional)
**Best for: Database connections, data exploration, and basic queries**
- ✅ **Multi-Database Support**: MongoDB, MySQL, and PostgreSQL connections
- ✅ **Schema Analysis**: Automatic database structure discovery and exploration  
- ✅ **Natural Language Queries**: Convert conversational requests to SQL/MongoDB queries
- ✅ **File Ingestion**: Direct JSON/CSV import into database tables
- ✅ **Safety-First Design**: Read-only mode by default, simple SELECT queries only
- ✅ **Export Integration**: Seamless handoff to AnalyticsAgent for complex analysis
- ⚠️ **Intentionally Limited**: Simple queries only (no JOINs, window functions, CTEs)
- 📋 **Use Cases**: Data exploration, basic queries, file imports, database connections
- **Note**: Requires installation with `pip install ambivo-agents[database]`

### Analytics Agent  
**Best for: Complex data analysis, advanced SQL, and statistical operations**
- 🚀 **Advanced SQL Engine**: Full DuckDB support with complex operations
- ✅ **Complex JOINs**: INNER, LEFT, RIGHT, OUTER joins across multiple datasets
- ✅ **Window Functions**: ROW_NUMBER(), RANK(), SUM() OVER(), statistical analysis
- ✅ **Advanced Aggregations**: GROUP BY with HAVING, complex statistical functions
- ✅ **CTEs & Subqueries**: WITH clauses, correlated subqueries, complex logic
- ✅ **UNION Operations**: Combine result sets with UNION/UNION ALL
- ✅ **Multi-File Analysis**: Load and join multiple CSV/XLSX files simultaneously
- ✅ **Statistical Functions**: Percentiles, correlations, trend analysis, outlier detection
- ✅ **Visualization**: Text-based charts and intelligent chart recommendations
- ✅ **Docker Security**: All operations run in isolated containers
- 📊 **Use Cases**: Business intelligence, complex analytics, statistical modeling, data science

**When to Use Which Agent:**
- **DatabaseAgent**: Simple queries, database connections, data exploration
- **AnalyticsAgent**: Complex analysis, joins, statistical operations, advanced SQL

### Assistant Agent
**Enhanced with Skill Assignment System**
- General purpose conversational AI with intelligent skill routing
- Context-aware responses and multi-turn conversations
- Customizable system messages
- ✨ **NEW: Skill Assignment** - Assign external capabilities like API specs, databases, and knowledge bases
- ✨ **Smart Intent Detection** - Automatically detects when to use assigned skills vs normal conversation
- ✨ **Dynamic Agent Spawning** - Internally creates specialized agents (APIAgent, DatabaseAgent, etc.) on-demand
- ✨ **Natural Language Translation** - Converts technical responses to conversational language
- 🎯 **Use Cases**: Custom API integration, database access, document search, while maintaining conversational interface

### Code Executor Agent
- Secure Python and Bash execution in Docker
- Isolated environment with resource limits
- Real-time output streaming

### Web Search Agent
- Multi-provider search (Brave, AVES APIs)
- Academic search capabilities
- Automatic provider failover

### Web Scraper Agent
- Proxy-enabled scraping (ScraperAPI compatible)
- Playwright and requests-based scraping
- Batch URL processing with rate limiting

### Knowledge Base Agent
- Document ingestion (PDF, DOCX, TXT, web URLs)
- Vector similarity search with Qdrant
- Semantic question answering

### Media Editor Agent
- Audio/video processing with FFmpeg
- Format conversion, resizing, trimming
- Audio extraction and volume adjustment

### YouTube Download Agent
- Download videos and audio from YouTube
- Docker-based execution with pytubefix
- Automatic title sanitization and metadata extraction

### Gather Agent
**Intelligent conversational form-filling with natural language understanding**
- ✅ **Conversational Interface**: Ask questions one at a time with natural flow
- ✅ **Multiple Question Types**: free-text, yes-no, single-select, multi-select
- ✅ **Smart Questionnaire Loading**: JSON/YAML from chat, files, or URLs
- ✅ **Conditional Logic**: Advanced dependent question workflows
- 🚀 **Natural Language Parsing** (NEW): Understand conversational responses
  - "Absolutely!" → "Yes" for yes/no questions
  - "I'd prefer email" → maps to email option in single-select
  - "Both AWS and Azure" → maps to multiple selections
  - "We have about 4 people" → maps to "3-5 people" range
- ✅ **Graceful Fallback**: Standard exact matching when NLP is disabled
- ✅ **Session Persistence**: Remember answers across conversation (~1 hour)
- ✅ **API Submission**: Configurable endpoint with collection status tracking

**Configuration:**
```yaml
# Enable natural language understanding (requires LLM)
gather:
  enable_natural_language_parsing: true  # Default: false
  
# Or via environment variable:
# export AMBIVO_AGENTS_GATHER_ENABLE_NATURAL_LANGUAGE_PARSING=true
```

Usage
```bash
# Interactive CLI demo (waits for your input after each question)
python examples/gather_cli.py

# Load questionnaire from a file path or URL
python examples/gather_cli.py --path ./questionnaire.yaml
python examples/gather_cli.py --path https://example.com/questionnaire.json

# Do a real HTTP submission (configure endpoint in agent_config.yaml)
python examples/gather_cli.py --real-submit

# Minimal scripted example
python examples/gather_simple.py
```

Configuration
```yaml
agent_capabilities:
  enable_gather: true

gather:
  submission_endpoint: "https://your-api.example.com/submit"
  submission_method: "POST"
  submission_headers:
    Authorization: "Bearer your-api-token"
    Content-Type: "application/json"
  memory_ttl_seconds: 3600
  # Optional: Validate free-text answers with LLM before proceeding
  enable_llm_answer_validation: false
  answer_validation:
    default_min_length: 1
  # Optional: Let LLM rewrite prompts (off by default for predictability)
  enable_llm_prompt_rewrite: false
```

See also: examples/gather_cli.py and examples/gather_simple.py for end-to-end demos. You can also wrap GatherAgent behind a small REST API for chatbot UIs (start/reply/finish/abort pattern).

### API Agent
- Comprehensive HTTP/REST API integration
- Multiple authentication methods (Bearer, API Key, Basic, OAuth2)
- Pre-fetch authentication with automatic token refresh
- Built-in retry logic with exponential backoff
- Security features (domain filtering, SSL verification)
- Support for all HTTP methods (GET, POST, PUT, DELETE, etc.)

#### API Agent Configuration

**Timeout Settings:**

The API Agent supports configurable timeout settings for different use cases:

```bash
# Environment variable configuration
export AMBIVO_AGENTS_API_AGENT_TIMEOUT_SECONDS=46  # Custom timeout in seconds

# Or in agent_config.yaml
api_agent:
  timeout_seconds: 46                    # Request timeout
  max_safe_timeout: 8                    # Requests above this use Docker for safety
  force_docker_above_timeout: true       # Enable Docker for long-running requests
```

**Localhost and Domain Access:**

For testing with local services or specific domain restrictions:

```bash
# Allow localhost access (disabled by default for security)
export AMBIVO_AGENTS_API_AGENT_ALLOWED_DOMAINS="127.0.0.1,localhost,api.example.com"
export AMBIVO_AGENTS_API_AGENT_BLOCKED_DOMAINS=""  # Clear default blocks

# Or in agent_config.yaml
api_agent:
  allowed_domains:
    - "127.0.0.1"
    - "localhost" 
    - "api.example.com"
    - "*.trusted-domain.com"     # Wildcards supported
  blocked_domains: []            # Override default localhost blocks
  
  # Default security settings (recommended for production)
  # allowed_domains: null        # Allows all except blocked
  # blocked_domains:
  #   - "localhost"
  #   - "127.0.0.1"
  #   - "0.0.0.0"
  #   - "169.254.169.254"        # AWS metadata service
```

**Usage Examples:**

```python
from ambivo_agents import APIAgent

# Test with local transcription service
async def test_local_api():
    agent = APIAgent.create_simple(user_id="tester")
    
    # Natural language API call
    response = await agent.chat("""
    Make a POST request to http://127.0.0.1:8002/kh/transcribe with:
    - Authorization: Bearer your-jwt-token
    - Content-Type: application/json  
    - Body: {"s3_url": "https://your-bucket.s3.amazonaws.com/audio.wav"}
    """)
    
    await agent.cleanup_session()

# Direct API request with custom timeout
async def direct_api_call():
    from ambivo_agents.agents.api_agent import APIRequest, HTTPMethod
    
    agent = APIAgent.create_simple(user_id="api_user")
    
    request = APIRequest(
        url="http://127.0.0.1:8002/kh/transcribe",
        method=HTTPMethod.POST,
        headers={
            "Authorization": "Bearer your-jwt-token",
            "Content-Type": "application/json"
        },
        json_data={"s3_url": "https://your-bucket.s3.amazonaws.com/audio.wav"},
        timeout=46  # Custom timeout in seconds
    )
    
    response = await agent.make_api_request(request)
    print(f"Status: {response.status_code}")
    print(f"Duration: {response.duration_ms}ms")
    
    await agent.cleanup_session()
```

**Security Notes:**
- Localhost access is blocked by default for security
- Always use `allowed_domains` in production to restrict API access
- Set appropriate timeouts to prevent long-running requests
- Consider using `max_safe_timeout` for requests that might take longer

## Workflow System

The workflow system enables multi-agent orchestration with sequential and parallel execution patterns:

### Basic Workflow Usage

```python
from ambivo_agents.core.workflow import WorkflowBuilder, WorkflowPatterns
from ambivo_agents import ModeratorAgent

async def workflow_example():
    # Create moderator with agents
    moderator, context = ModeratorAgent.create(
        user_id="workflow_user",
        enabled_agents=['web_search', 'web_scraper', 'knowledge_base']
    )
    
    # Create search -> scrape -> ingest workflow
    workflow = WorkflowPatterns.create_search_scrape_ingest_workflow(
        moderator.specialized_agents['web_search'],
        moderator.specialized_agents['web_scraper'], 
        moderator.specialized_agents['knowledge_base']
    )
    
    # Execute workflow
    result = await workflow.execute(
        "Research renewable energy trends and store in knowledge base",
        context.to_execution_context()
    )
    
    if result.success:
        print(f"Workflow completed in {result.execution_time:.2f}s")
        print(f"Nodes executed: {result.nodes_executed}")
    
    await moderator.cleanup_session()
```

### Advanced Workflow Features

```python
from ambivo_agents.core.enhanced_workflow import (
    AdvancedWorkflowBuilder, EnhancedModeratorAgent
)

async def advanced_workflow():
    # Create enhanced moderator
    base_moderator, context = ModeratorAgent.create(user_id="advanced_user")
    enhanced_moderator = EnhancedModeratorAgent(base_moderator)
    
    # Natural language workflow triggers
    response1 = await enhanced_moderator.process_message_with_workflows(
        "I need agents to reach consensus on climate solutions"
    )
    
    response2 = await enhanced_moderator.process_message_with_workflows(
        "Create a debate between agents about AI ethics"
    )
    
    # Check workflow status
    status = await enhanced_moderator.get_workflow_status()
    print(f"Available workflows: {status['advanced_workflows']['registered']}")
```

### Workflow Patterns

- **Sequential Workflows**: Execute agents in order, passing results between them
- **Parallel Workflows**: Execute multiple agents simultaneously
- **Consensus Workflows**: Agents collaborate to reach agreement
- **Debate Workflows**: Structured multi-agent discussions
- **Error Recovery**: Automatic fallback to backup agents
- **Map-Reduce**: Parallel processing with result aggregation

## System Messages

System messages control agent behavior and responses. Each agent supports custom system prompts:

### Default System Messages

```python
# Agents come with role-specific system messages
assistant_agent = AssistantAgent.create_simple(user_id="user")
# Default: "You are a helpful AI assistant. Provide accurate, thoughtful responses..."

code_agent = CodeExecutorAgent.create_simple(user_id="user") 
# Default: "You are a code execution specialist. Write clean, well-commented code..."
```

### Custom System Messages

```python
from ambivo_agents import AssistantAgent

# Create agent with custom system message
custom_system = """You are a technical documentation specialist. 
Always provide detailed explanations with code examples. 
Use professional terminology and structured responses."""

agent, context = AssistantAgent.create(
    user_id="doc_specialist",
    system_message=custom_system
)

response = await agent.chat("Explain REST API design principles")
```

### Moderator System Messages

```python
from ambivo_agents import ModeratorAgent

# Custom moderator behavior
moderator_system = """You are a project management assistant.
Route technical queries to appropriate agents and provide 
executive summaries of complex multi-agent interactions."""

moderator, context = ModeratorAgent.create(
    user_id="pm_user",
    system_message=moderator_system
)
```

### System Message Features

- **Context Integration**: System messages work with conversation history
- **Agent-Specific**: Each agent type has optimized default prompts
- **Workflow Aware**: System messages adapt to workflow contexts
- **Provider Compatibility**: Works with all LLM providers (OpenAI, Anthropic, Bedrock)

## Prerequisites

### Required
- **Python 3.11+**
- **Docker** (for code execution, media processing, YouTube downloads)
- **Redis** (Cloud Redis recommended)

### API Keys (Optional - based on enabled features)
- **OpenAI API Key** (for GPT models)
- **Anthropic API Key** (for Claude models)
- **AWS Credentials** (for Bedrock models)
- **Brave Search API Key** (for web search)
- **AVES API Key** (for web search)
- **ScraperAPI/Proxy credentials** (for web scraping)
- **Qdrant Cloud API Key** (for Knowledge Base operations)
- **Redis Cloud credentials** (for memory management)

## Installation

### 1. Install Dependencies

**Core Installation (without database support):**
```bash
pip install -r requirements.txt
```

**With Optional Database Support:**
```bash
# Install with database capabilities (MongoDB, MySQL, PostgreSQL)
pip install ambivo-agents[database]

# Or install all optional features including database support
pip install ambivo-agents[all]
```

The database agents are optional and require additional dependencies:
- **MongoDB**: `pymongo>=4.0.0`
- **MySQL**: `mysql-connector-python>=8.0.0`
- **PostgreSQL**: `psycopg2-binary>=2.9.0`

### 2. Setup Docker Images
```bash
docker pull sgosain/amb-ubuntu-python-public-pod
```

### 3. Setup Redis

**Recommended: Cloud Redis**
```yaml
# In agent_config.yaml
redis:
  host: "your-redis-cloud-endpoint.redis.cloud"
  port: 6379
  password: "your-redis-password"
```

**Alternative: Local Redis**
```bash
# Using Docker
docker run -d --name redis -p 6379:6379 redis:latest
```

## Configuration

Create `agent_config.yaml` in your project root:

```yaml
# Redis Configuration (Required)
redis:
  host: "your-redis-cloud-endpoint.redis.cloud"
  port: 6379
  db: 0
  password: "your-redis-password"

# LLM Configuration (Required - at least one provider)
llm:
  preferred_provider: "openai"
  temperature: 0.7
  openai_api_key: "your-openai-key"
  anthropic_api_key: "your-anthropic-key"
  aws_access_key_id: "your-aws-key"
  aws_secret_access_key: "your-aws-secret"
  aws_region: "us-east-1"

# Agent Capabilities
agent_capabilities:
  enable_knowledge_base: true
  enable_web_search: true
  enable_code_execution: true
  enable_file_processing: true
  enable_web_ingestion: true
  enable_api_calls: true
  enable_web_scraping: true
  enable_proxy_mode: true
  enable_media_editor: true
  enable_youtube_download: true
  enable_analytics: true

# ModeratorAgent default agents
moderator:
  default_enabled_agents:
    - knowledge_base
    - web_search
    - assistant
    - media_editor
    - youtube_download
    - code_executor
    - web_scraper
    - analytics

# Service-specific configurations
web_search:
  brave_api_key: "your-brave-api-key"
  avesapi_api_key: "your-aves-api-key"

knowledge_base:
  qdrant_url: "https://your-cluster.qdrant.tech"
  qdrant_api_key: "your-qdrant-api-key"
  chunk_size: 1024
  chunk_overlap: 20
  similarity_top_k: 5

youtube_download:
  docker_image: "sgosain/amb-ubuntu-python-public-pod"
  download_dir: "./youtube_downloads"
  timeout: 600
  memory_limit: "1g"
  default_audio_only: true

analytics:
  docker_image: "sgosain/amb-ubuntu-python-public-pod"
  timeout: 300
  memory_limit: "1g"
  enable_url_ingestion: true
  max_file_size_mb: 100

docker:
  timeout: 60
  memory_limit: "512m"
  images: ["sgosain/amb-ubuntu-python-public-pod"]
```

## Configuration Methods

The library supports two configuration methods:

### 1. Environment Variables (Recommended for Production)

**Quick Start with Environment Variables:**

```bash
# Download and edit the full template
curl -o set_env.sh https://github.com/ambivo-corp/ambivo-agents/raw/main/set_env_template.sh
chmod +x set_env.sh

# Edit the template with your credentials, then source it
source set_env.sh
```

**Replace ALL placeholder values** with your actual credentials:
- Redis connection details
- LLM API keys (OpenAI/Anthropic)
- Web Search API keys (Brave/AVES)
- Knowledge Base credentials (Qdrant)
- Web Scraping proxy (ScraperAPI)

**Minimal Environment Setup:**
```bash
# Required - Redis
export AMBIVO_AGENTS_REDIS_HOST="your-redis-host.redis.cloud"
export AMBIVO_AGENTS_REDIS_PORT="6379"
export AMBIVO_AGENTS_REDIS_PASSWORD="your-redis-password"

# Required - At least one LLM provider
export AMBIVO_AGENTS_OPENAI_API_KEY="sk-your-openai-key"

# Optional - Enable specific agents
export AMBIVO_AGENTS_ENABLE_YOUTUBE_DOWNLOAD="true"
export AMBIVO_AGENTS_ENABLE_WEB_SEARCH="true"
export AMBIVO_AGENTS_ENABLE_ANALYTICS="true"
export AMBIVO_AGENTS_MODERATOR_ENABLED_AGENTS="youtube_download,web_search,analytics,assistant"

# Run your application
python your_app.py
```

### 2. YAML Configuration (Traditional)

**Use the full YAML template:**

```bash
# Download and edit the full template
curl -o agent_config_sample.yaml https://github.com/ambivo-corp/ambivo-agents/raw/main/agent_config_sample.yaml

# Copy to your config file and edit with your credentials
cp agent_config_sample.yaml agent_config.yaml
```

**Replace ALL placeholder values** with your actual credentials, then create `agent_config.yaml` in your project root.

### Docker Deployment with Environment Variables

```yaml
# docker-compose.yml
version: '3.8'
services:
  ambivo-app:
    image: your-app:latest
    environment:
      - AMBIVO_AGENTS_REDIS_HOST=redis
      - AMBIVO_AGENTS_REDIS_PORT=6379
      - AMBIVO_AGENTS_OPENAI_API_KEY=${OPENAI_API_KEY}
      - AMBIVO_AGENTS_ENABLE_YOUTUBE_DOWNLOAD=true
      - AMBIVO_AGENTS_ENABLE_ANALYTICS=true
    volumes:
      - ./downloads:/app/downloads
      - /var/run/docker.sock:/var/run/docker.sock
    depends_on:
      - redis
  
  redis:
    image: redis:latest
    ports:
      - "6379:6379"
```

**Note:** Environment variables take precedence over YAML configuration. The `agent_config.yaml` file is optional when using environment variables.

## Project Structure

```
ambivo_agents/
├── agents/          # Agent implementations
│   ├── analytics.py     # Analytics Agent (DuckDB da