# 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())
```
### 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
- 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
### Database Agent (Optional)
- Secure connections to MongoDB, MySQL, and PostgreSQL databases
- Schema inspection and table structure exploration
- Natural language to SQL/MongoDB query conversion
- Safe query execution with read-only mode by default
- **File ingestion support**: Insert JSON/CSV files directly into database tables
- Data export to CSV for analytics handoff
- Automatic integration with Analytics Agent for visualization
- Query result formatting with tables and statistics
- **Note**: Requires installation with `pip install ambivo-agents[database]`
### Analytics Agent
- CSV/XLS file ingestion into in-memory DuckDB
- Schema exploration and data quality assessment
- Natural language to SQL query conversion
- Text-based chart generation (bar charts, line charts, tables)
- Chart recommendations based on data characteristics
- Docker-only execution for security
- Business intelligence and data insights
### Assistant Agent
- General purpose conversational AI
- Context-aware responses
- Multi-turn conversations
- Customizable system messages
### 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
### 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.)
## 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 data analysis)
│ ├── api_agent.py # API Agent (HTTP/REST integration)
│ ├── assistant.py # Assistant Agent (general conversation)
│ ├── code_executor.py # Code Executor Agent (Docker-based execution)
│ ├── database_agent.py # Database Agent (MongoDB, MySQL, PostgreSQL)
│ ├── knowledge_base.py # Knowledge Base Agent (Qdrant vector search)
│ ├── media_editor.py # Media Editor Agent (FFmpeg processing)
│ ├── moderator.py # ModeratorAgent (main orchestrator)
│ ├── web_scraper.py # Web Scraper Agent (Playwright-based)
│ ├── web_search.py # Web Search Agent (Brave/AVES search)
│ └── youtube_download.py # YouTube Download Agent (pytubefix)
├── config/ # Configuration management
├── core/ # Core functionality
│ ├── base.py
│ ├── llm.py
│ ├── memory.py
│ ├── workflow.py # Basic workflow system
│ └── enhanced_workflow.py # Advanced workflow patterns
├── executors/ # Execution environments
├── services/ # Service layer
├── __init__.py # Package initialization
└── cli.py # Command line interface
```
## Usage Examples
### ModeratorAgent with Auto-Routing
```python
from ambivo_agents import ModeratorAgent
import asyncio
async def basic_moderator():
moderator, context = ModeratorAgent.create(user_id="demo_user")
# Auto-routing examples
examples = [
"Download audio from https://youtube.com/watch?v=example",
"Search for latest artificial intelligence news",
"Load data from sales.csv and analyze trends",
"Extract audio from video.mp4 as high quality MP3",
"What is machine learning and how does it work?",
]
for query in examples:
response = await moderator.chat(query)
print(f"Response: {response[:100]}...")
await moderator.cleanup_session()
asyncio.run(basic_moderator())
```
### Context-Aware Conversations
```python
async def context_conversation():
moderator, context = ModeratorAgent.create(user_id="context_demo")
# Initial request
response1 = await moderator.chat("Download audio from https://youtube.com/watch?v=example")
# Follow-up using context
response2 = await moderator.chat("Actually, download the video instead of just audio")
# Another follow-up
response3 = await moderator.chat("Get information about that video")
await moderator.cleanup_session()
```
### YouTube Downloads
```python
from ambivo_agents import YouTubeDownloadAgent
async def download_youtube():
agent, context = YouTubeDownloadAgent.create(user_id="media_user")
# Download audio
result = await agent._download_youtube_audio(
"https://youtube.com/watch?v=example"
)
if result['success']:
print(f"Audio downloaded: {result['filename']}")
print(f"Path: {result['file_path']}")
await agent.cleanup_session()
```
### Database Operations
```python
from ambivo_agents import DatabaseAgent
async def database_demo():
agent, context = DatabaseAgent.create(user_id="db_user")
# Connect to MySQL database
response = await agent.chat("Connect to MySQL database at localhost:3306, database: mydb, username: user, password: pass")
print(f"Connection: {response}")
# Explore database schema
schema = await agent.chat("show me the database schema")
print(f"Schema: {schema}")
# Natural language queries
result = await agent.chat("show me all users from the users table")
print(f"Users: {result}")
# Query with analytics handoff
analytics_result = await agent.chat("get sales data and create visualizations")
print(f"Analytics: {analytics_result}")
await agent.cleanup_session()
# Database to Analytics Workflow
async def database_analytics_workflow():
from ambivo_agents import ModeratorAgent
# Use ModeratorAgent for automatic routing
moderator, context = ModeratorAgent.create(
user_id="workflow_user",
enabled_agents=["database_agent", "analytics", "assistant"]
)
# Connect and query database
await moderator.chat("Connect to MySQL localhost:3306 database mydb user admin password secret")
# Query data with automatic analytics handoff
response = await moderator.chat("Get sales data from orders table and create charts showing trends")
print(response)
await moderator.cleanup_session()
```
### File Reading and Database Ingestion
All agents have built-in file reading capabilities for JSON, CSV, XML, and YAML files. Database insertion requires the optional database package.
```python
from ambivo_agents import AssistantAgent
async def read_file_and_insert_to_database():
"""Reads a JSON file and attempts database insertion with graceful fallback"""
# Step 1: Read and parse file (always available)
agent = AssistantAgent.create_simple(user_id="file_user")
result = await agent.read_and_parse_file("./data/users.json")
if not result['success']:
print(f"❌ Failed to read file: {result.get('error', 'Unknown error')}")
await agent.cleanup_session()
return
json_data = result['parse_result']['data']
print(f"✅ Successfully loaded {len(json_data)} records from users.json")
# Step 2: Attempt database insertion
try:
from ambivo_agents import DatabaseAgent
# DatabaseAgent is available - proceed with insertion
db_agent = DatabaseAgent.create_simple(user_id="db_user")
# Connect to MongoDB
await db_agent.chat("connect to mongodb://localhost:27017 database myapp")
# Insert the data
response = await db_agent.chat(f"insert this data into users collection: {json_data}")
print(f"✅ Successfully inserted data into MongoDB: {response}")
await db_agent.cleanup_session()
except ImportError:
# DatabaseAgent not available - provide polite warning and alternatives
print("\n⚠️ Database insertion not available")
print("💡 To enable database features, install with: pip install ambivo-agents[database]")
print("\n📁 Available alternatives:")
print(" • File successfully read and parsed")
print(" • Data can be transformed to other formats")
# Show what we can still do
csv_result = await agent.convert_json_to_csv(json_data)
if csv_result['success']:
print(" • ✅ Converted to CSV format (available for export)")
await agent.cleanup_session()
# Alternative: Natural language approach with graceful handling
async def natural_language_file_ingestion():
"""Uses natural language commands with automatic fallback"""
try:
from ambivo_agents import DatabaseAgent
agent = DatabaseAgent.create_simple(user_id="user")
# Full database workflow available
await agent.chat("connect to mongodb://localhost:27017 database myapp")
response = await agent.chat("read users.json file and insert all records into users collection")
print(f"✅ Database ingestion completed: {response}")
await agent.cleanup_session()
except ImportError:
# Fallback to file reading only
from ambivo_agents import AssistantAgent
agent = AssistantAgent.create_simple(user_id="user")
print("⚠️ DatabaseAgent not installed. Reading file only...")
response = await agent.chat("read and analyze the users.json file structure")
print(f"📁 File analysis: {response}")
print("💡 Install database support with: pip install ambivo-agents[database]")
await agent.cleanup_session()
```
### Data Analytics
```python
from ambivo_agents import AnalyticsAgent
async def analytics_demo():
agent, context = AnalyticsAgent.create(user_id="analyst_user")
# Load and analyze CSV data
response = await agent.chat("load data from sales.csv and analyze it")
print(f"Analysis: {response}")
# Schema exploration
schema = await agent.chat("show me the schema of the current dataset")
print(f"Schema: {schema}")
# Natural language queries
top_sales = await agent.chat("what are the top 5 products by sales?")
print(f"Top Sales: {top_sales}")
# SQL queries
sql_result = await agent.chat("SELECT region, SUM(sales) as total FROM data GROUP BY region")
print(f"SQL Result: {sql_result}")
# Visualizations
chart = await agent.chat("create a bar chart showing sales by region")
print(f"Chart: {chart}")
await agent.cleanup_session()
# Context Preservation Example
async def context_preservation_demo():
"""Demonstrates context/state preservation between chat messages"""
agent = AnalyticsAgent.create_simple(user_id="user123")
try:
# Load data once
await agent.chat("load data from transactions.xlsx and analyze it")
# Multiple queries without reload - uses cached context
schema = await agent.chat("show schema") # ✅ Uses cached data
top_items = await agent.chat("what are the top 5 amounts?") # ✅ Uses cached data
summary = await agent.chat("summary statistics") # ✅ Uses cached data
counts = await agent.chat("count by category") # ✅ Uses cached data
print("All queries executed using cached dataset - no reload needed!")
finally:
await agent.cleanup_session() # Clean up resources
```
### Knowledge Base Operations
```python
from ambivo_agents import KnowledgeBaseAgent
async def knowledge_base_demo():
agent, context = KnowledgeBaseAgent.create(user_id="kb_user")
# Ingest document
result = await agent._ingest_document(
kb_name="company_kb",
doc_path="/path/to/document.pdf",
custom_meta={"department": "HR", "type": "policy"}
)
if result['success']:
# Query the knowledge base
answer = await agent._query_knowledge_base(
kb_name="company_kb",
query="What is the remote work policy?"
)
if answer['success']:
print(f"Answer: {answer['answer']}")
await agent.cleanup_session()
```
### API Integration
```python
from ambivo_agents import APIAgent
from ambivo_agents.agents.api_agent import APIRequest, AuthConfig, HTTPMethod, AuthType
async def api_integration_demo():
agent, context = APIAgent.create(user_id="api_user")
# Basic GET request
request = APIRequest(
url="https://jsonplaceholder.typicode.com/posts/1",
method=HTTPMethod.GET
)
response = await agent.make_api_request(request)
if not response.error:
print(f"Status: {response.status_code}")
print(f"Data: {response.json_data}")
# POST with authentication
auth_config = AuthConfig(
auth_type=AuthType.BEARER,
token="your-api-token"
)
post_request = APIRequest(
url="https://api.example.com/data",
method=HTTPMethod.POST,
auth_config=auth_config,
json_data={"name": "test", "value": "123"}
)
post_response = await agent.make_api_request(post_request)
# Google OAuth2 with pre-fetch
google_auth = AuthConfig(
auth_type=AuthType.BEARER,
pre_auth_url="https://www.googleapis.com/oauth2/v4/token",
pre_auth_method=HTTPMethod.POST,
pre_auth_payload={
"client_id": "your-client-id",
"client_secret": "your-secret",
"refresh_token": "your-refresh-token",
"grant_type": "refresh_token"
},
token_path="access_token"
)
sheets_request = APIRequest(
url="https://sheets.googleapis.com/v4/spreadsheets/sheet-id/values/A1:B10",
method=HTTPMethod.GET,
auth_config=google_auth
)
# APIAgent will automatically fetch access_token first, then make the request
sheets_response = await agent.make_api_request(sheets_request)
await agent.cleanup_session()
# Conversational API usage
async def conversational_api():
agent = APIAgent.create_simple(user_id="chat_user")
# Use natural language for API requests
response = await agent.chat("GET https://jsonplaceholder.typicode.com/users/1")
print(response)
response = await agent.chat(
"POST https://api.example.com/data with headers: {\"Authorization\": \"Bearer token\"} "
"and data: {\"message\": \"Hello API\"}"
)
print(response)
await agent.cleanup_session()
```
### Context Manager Pattern
```python
from ambivo_agents import ModeratorAgent, AgentSession
import asyncio
async def main():
# Auto-cleanup with context manager
async with AgentSession(ModeratorAgent, user_id="sarah") as moderator:
response = await moderator.chat("Download audio from https://youtube.com/watch?v=example")
print(response)
# Moderator automatically cleaned up
asyncio.run(main())
```
### Workflow Examples
```python
from ambivo_agents.core.workflow import WorkflowBuilder
async def custom_workflow():
# Create agents
moderator, context = ModeratorAgent.create(user_id="workflow_demo")
# Build custom workflow
builder = WorkflowBuilder()
builder.add_agent(moderator.specialized_agents['web_search'], "search")
builder.add_agent(moderator.specialized_agents['assistant'], "analyze")
builder.add_edge("search", "analyze")
builder.set_start_node("search")
builder.set_end_node("analyze")
workflow = builder.build()
# Execute workflow
result = await workflow.execute(
"Research AI safety and provide analysis",
context.to_execution_context()
)
print(f"Workflow result: {result.success}")
await moderator.cleanup_session()
```
## Streaming System
The library features a modern **StreamChunk** system that provides structured, type-safe streaming responses with rich metadata.
### StreamChunk Overview
All agents now return `StreamChunk` objects instead of raw strings, enabling:
- **Type-safe content classification** with `StreamSubType` enum
- **Rich metadata** for debugging, analytics, and context
- **Programmatic filtering** without string parsing
- **Consistent interface** across all agents
### StreamSubType Categories
```python
from ambivo_agents.core.base import StreamSubType
# Available sub-types:
StreamSubType.CONTENT # Actual response content from LLMs
StreamSubType.STATUS # Progress updates, thinking, interim info
StreamSubType.RESULT # Search results, processing outputs
StreamSubType.ERROR # Error messages and failures
StreamSubType.METADATA # Additional context and metadata
```
### Basic Streaming Usage
```python
from ambivo_agents import ModeratorAgent
from ambivo_agents.core.base import StreamSubType
async def streaming_example():
moderator, context = ModeratorAgent.create(user_id="stream_user")
# Stream with filtering
print("🤖 Assistant: ", end='', flush=True)
async for chunk in moderator.chat_stream("Search for Python tutorials"):
# Filter by content type
if chunk.sub_type == StreamSubType.CONTENT:
print(chunk.text, end='', flush=True)
elif chunk.sub_type == StreamSubType.STATUS:
print(f"\n[{chunk.text.strip()}]", end='', flush=True)
elif chunk.sub_type == StreamSubType.ERROR:
print(f"\n[ERROR: {chunk.text}]", end='', flush=True)
await moderator.cleanup_session()
```
### Advanced Streaming with Metadata
```python
async def advanced_streaming():
moderator, context = ModeratorAgent.create(user_id="advanced_user")
# Collect and analyze stream
content_chunks = []
status_chunks = []
result_chunks = []
async for chunk in moderator.chat_stream("Download audio from YouTube"):
# Categorize by type
if chunk.sub_type == StreamSubType.CONTENT:
content_chunks.append(chunk)
elif chunk.sub_type == StreamSubType.STATUS:
status_chunks.append(chunk)
elif chunk.sub_type == StreamSubType.RESULT:
result_chunks.append(chunk)
# Access metadata
agent_info = chunk.metadata.get('agent')
operation = chunk.metadata.get('operation')
phase = chunk.metadata.get('phase')
print(f"[{chunk.sub_type.value}] {chunk.text[:50]}...")
if agent_info:
print(f" Agent: {agent_info}")
if operation:
print(f" Operation: {operation}")
# Analysis
print(f"\nStream Analysis:")
print(f"Content chunks: {len(content_chunks)}")
print(f"Status updates: {len(status_chunks)}")
print(f"Results: {len(result_chunks)}")
await moderator.cleanup_session()
```
### Streaming in Web APIs
```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
app = FastAPI()
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
moderator, context = ModeratorAgent.create(user_id=request.user_id)
async def generate_stream():
async for chunk in moderator.chat_stream(request.message):
# Convert StreamChunk to JSON
chunk_data = {
'type': 'chunk',
'sub_type': chunk.sub_type.value,
'text': chunk.text,
'metadata': chunk.metadata,
'timestamp': chunk.timestamp.isoformat()
}
yield f"data: {json.dumps(chunk_data)}\n\n"
yield "data: {\"type\": \"done\"}\n\n"
return StreamingResponse(generate_stream(), media_type="text/stream")
```
### Real-time UI Integration
```javascript
// Frontend streaming handler
const eventSource = new EventSource('/chat/stream');
eventSource.onmessage = function(event) {
const data = JSON.parse(event.data);
if (data.type === 'chunk') {
switch(data.sub_type) {
case 'content':
// Display main response content
appendToChat(data.text);
break;
case 'status':
// Show progress indicator
updateStatus(data.text);
break;
case 'result':
// Display search results/outputs
addResult(data.text, data.metadata);
break;
case 'error':
// Handle errors
showError(data.text);
break;
}
}
};
```
### StreamChunk Benefits
**For Developers:**
- **Type Safety** - No string parsing for content classification
- **Rich Context** - Access agent info, operation details, timing
- **Easy Filtering** - Filter streams by content type programmatically
- **Debugging** - Detailed metadata for troubleshooting
**For Applications:**
- **Smart UIs** - Show different content types appropriately
- **Progress Tracking** - Real-time operation status updates
- **Error Handling** - Structured error information
- **Analytics** - Performance metrics and usage tracking
## Session Management
### Understanding Session vs Conversation IDs
The library uses two identifiers for context management:
- **session_id**: Represents a broader user session or application context
- **conversation_id**: Represents a specific conversation thread within a session
```python
# Single conversation (most common)
moderator, context = ModeratorAgent.create(
user_id="john",
session_id="user_john_main",
conversation_id="user_john_main" # Same as session_id
)
# Multiple conversations per session
session_key = "user_john_tenant_abc"
# Conversation 1: Data Analysis
moderator1, context1 = ModeratorAgent.create(
user_id="john",
session_id=session_key,
conversation_id="john_data_analysis_conv"
)
# Conversation 2: YouTube Downloads
moderator2, context2 = ModeratorAgent.create(
user_id="john",
session_id=session_key,
conversation_id="john_youtube_downloads_conv"
)
```
## Web API Integration
```python
from ambivo_agents import ModeratorAgent
import asyncio
import time
class ChatAPI:
def __init__(self):
self.active_moderators = {}
async def chat_endpoint(self, request_data):
user_message = request_data.get('message', '')
user_id = request_data.get('user_id', f"user_{uuid.uuid4()}")
session_id = request_data.get('session_id', f"session_{user_id}_{int(time.time())}")
try:
if session_id not in self.active_moderators:
moderator, context = ModeratorAgent.create(
user_id=user_id,
session_id=session_id
)
self.active_moderators[session_id] = moderator
else:
moderator = self.active_moderators[session_id]
response_content = await moderator.chat(user_message)
return {
'success': True,
'response': response_content,
'session_id': session_id,
'timestamp': time.time()
}
except Exception as e:
return {
'success': False,
'error': str(e),
'timestamp': time.time()
}
async def cleanup_session(self, session_id):
if session_id in self.active_moderators:
await self.active_moderators[session_id].cleanup_session()
del self.active_moderators[session_id]
```
## Command Line Interface
```bash
# Interactive mode with auto-routing
ambivo-agents
# Single queries
ambivo-agents -q "Download audio from https://youtube.com/watch?v=example"
ambivo-agents -q "Search for latest AI trends"
ambivo-agents -q "Load data from sales.csv and show top products"
ambivo-agents -q "Extract audio from video.mp4"
# Check agent status
ambivo-agents status
# Test all agents
ambivo-agents --test
# Debug mode
ambivo-agents --debug -q "test query"
```
## Architecture
### ModeratorAgent Architecture
The **ModeratorAgent** acts as an intelligent orchestrator:
```
[User Query]
↓
[ModeratorAgent] ← Analyzes intent and context
↓
[Intent Analysis] ← Uses LLM + patterns + keywords
↓
[Route Selection] ← Chooses best agent(s)
↓
[Specialized Agent] ← YouTubeAgent, SearchAgent, etc.
↓
[Response] ← Combined and contextualized
↓
[User]
```
### Workflow Architecture
```
[WorkflowBuilder] → [Workflow Definition]
↓ ↓
[Workflow Executor] → [Sequential/Parallel Execution]
↓ ↓
[State Management] → [Persistent Checkpoints]
↓ ↓
[Result Aggregation] → [Final Response]
```
### Memory System
- Redis-based persistence with compression and caching
- Built-in conversation history for every agent
- Session-aware context with automatic cleanup
- Multi-session support with isolation
### LLM Provider Management
- Automatic failover between OpenAI, Anthropic, AWS Bedrock
- Rate limiting and error handling
- Provider rotation based on availability and performance
## Docker Setup
### Consolidated Docker File Sharing System
Ambivo Agents uses a **consolidated Docker-shared directory structure** for all file operations across agents. This provides consistent, secure, and efficient file sharing between your host filesystem and Docker containers.
#### Consolidated Directory Structure
All Docker-based agents (AnalyticsAgent, MediaEditorAgent, CodeExecutorAgent, WebScraperAgent, APIAgent) use the same base structure:
```
Your Project Directory/
└── docker_shared/ # Consolidated base directory
├── input/ # Read-only input files
│ ├── analytics/ → /docker_shared/input/analytics (AnalyticsAgent)
│ ├── media/ → /docker_shared/input/media (MediaEditorAgent)
│ ├── code/ → /docker_shared/input/code (CodeExecutorAgent)
│ └── scraper/ → /docker_shared/input/scraper (WebScraperAgent)
├── output/ # Read-write output files
│ ├── analytics/ → /docker_shared/output/analytics (Analysis results)
│ ├── media/ → /docker_shared/output/media (Processed media)
│ ├── code/ → /docker_shared/output/code (Code execution results)
│ └── scraper/ → /docker_shared/output/scraper (Scraped data)
├── temp/ # Read-write temporary workspace
│ ├── analytics/ → /docker_shared/temp/analytics (Analytics temp files)
│ ├── media/ → /docker_shared/temp/media (Media processing temp)
│ └── code/ → /docker_shared/temp/code (Code execution temp)
├── handoff/ # Read-write inter-agent file sharing
│ ├── database/ → /docker_shared/handoff/database (Database exports)
│ ├── analytics/ → /docker_shared/handoff/analytics (Analytics results)
│ └── media/ → /docker_shared/handoff/media (Media for processing)
└── work/ → /docker_shared/work # General workspace
```
#### How the System Works
When you request operations like "convert data.csv to xlsx" or "process video.mp4":
1. **File Detection**: System detects file paths in your request
2. **Directory Setup**: Auto-creates agent-specific subdirectories in `./docker_shared/`
3. **File Copying**: Copies your files to appropriate input directories
4. **Docker Execution**: Runs containers with consistent `/docker_shared/` mount points
5. **Result Retrieval**: Outputs appear in agent-specific output directories
#### Inter-Agent Workflows
The consolidated structure enables seamless workflows between agents:
**Database → Analytics Workflow:**
```
1. DatabaseAgent exports data → ./docker_shared/handoff/database/export.csv
2. AnalyticsAgent automatically → reads from /docker_shared/handoff/database/
3. AnalyticsAgent processes data → outputs to /docker_shared/output/analytics/
4. Results available at → ./docker_shared/output/analytics/results.json
```
#### Agent Handoff Mechanism
The handoff system uses the `handoff_subdir` parameter to enable seamless file transfers between agents:
**DatabaseAgent → AnalyticsAgent Handoff:**
```python
# DatabaseAgent automatically exports to handoff directory
result = await db_agent.chat("export sales data for analytics",
handoff_subdir="sales_analysis_2024")
# Creates: ./docker_shared/handoff/database/sales_analysis_2024/
# AnalyticsAgent automatically detects handoff files
analytics_result = await analytics_agent.chat("analyze sales data",
handoff_subdir="sales_analysis_2024")
# Reads from: ./docker_shared/handoff/database/sales_analysis_2024/
```
**Handoff Directory Management:**
- **Automatic creation**: Subdirectories created based on `handoff_subdir` parameter
- **File naming**: `{agent_type}_{timestamp}_{operation}.{ext}`
- **Cleanup**: Handoff files older than 24 hours automatically removed
- **Thread-safe**: Multiple concurrent handoffs supported
- **Cross-platform**: Works consistently across Windows, macOS, and Linux
**Configuration in `agent_config.yaml`:**
```yaml
docker:
container_mounts:
handoff: "/docker_shared/handoff"
agent_subdirs:
database: ["handoff/database"]
analytics: ["input/analytics", "output/analytics", "handoff/analytics"]
media: ["input/media", "output/media", "handoff/media"]
```
**Enhanced Fallback (CSV→XLSX Conversion):**
```
1. User: "convert sales.csv to xlsx"
2. ModeratorAgent detects file operation need
3. Copies sales.csv → ./docker_shared/input/code/sales.csv
4. CodeExecutorAgent processes → from /docker_shared/input/code/sales.csv
5. Outputs converted file → to /docker_shared/output/code/sales.xlsx
6. User accesses result at → ./docker_shared/output/code/sales.xlsx
```
**Media Processing Workflow:**
```
1. User places video → ./docker_shared/input/media/input.mp4
2. MediaEditorAgent processes → from /docker_shared/input/media/input.mp4
3. Outputs processed file → to /docker_shared/output/media/output.mp3
4. User gets result from → ./docker_shared/output/media/output.mp3
```
#### Third-Party Developer Integration
For developers building custom agents:
```python
from ambivo_agents.core import get_shared_manager
# Get the consolidated shared manager
shared_manager = get_shared_manager()
# Prepare environment for your custom agent
input_path, output_path = shared_manager.prepare_agent_environment(
agent="my_custom_agent",
input_files=["./my_data.csv"]
)
# Get Docker volume configuration
volumes = shared_manager.get_docker_volumes()
# volumes = {
# '/path/to/docker_shared/input': {'bind': '/docker_shared/input', 'mode': 'ro'},
# '/path/to/docker_shared/output': {'bind': '/docker_shared/output', 'mode': 'rw'},
# # ... other mounts
# }
# In your Docker container, access files at:
# - Input: /docker_shared/input/my_custom_agent/
# - Output: /docker_shared/output/my_custom_agent/
# - Temp: /docker_shared/temp/my_custom_agent/
# - Handoff: /docker_shared/handoff/my_custom_agent/
# After processing, check results:
output_files = shared_manager.list_outputs("my_custom_agent")
latest_output = shared_manager.get_latest_output("my_custom_agent", ".xlsx")
```
#### Example Usage
```python
import asyncio
from ambivo_agents import ModeratorAgent
async def process_files_with_consolidated_structure():
# Create moderator with auto-routing
moderator, context = ModeratorAgent.create(user_id="file_processor")
# File operations use consolidated Docker structure
await moderator.chat("convert sales_data.csv to xlsx format") # → ./docker_shared/output/code/
await moderator.chat("extract audio from video.mp4 as MP3") # → ./docker_shared/output/media/
await moderator.chat("analyze customer_data.csv and chart") # → ./docker_shared/output/analytics/
# All results organized by agent type in docker_shared/output/
await moderator.cleanup_session()
# Run the example
asyncio.run(process_files_with_consolidated_structure())
```
#### File Locations After Operations
```bash
# Directory structure after various operations
your-project/
├── sales_data.csv # Your original files
├── video.mp4
├── customer_data.csv
└── docker_shared/ # Consolidated results
└── output/
├── code/
│ └── sales_data.xlsx # CSV→XLSX conversion
├── media/
│ └── video_audio.mp3 # Audio extraction
└── analytics/
├── analysis_report.json # Data analysis
└── customer_charts.png # Generated charts
```
#### Configuration
The consolidated structure is configured in `agent_config.yaml`:
```yaml
docker:
shared_base_dir: "./docker_shared" # Host base directory
container_mounts:
input: "/docker_shared/input" # Read-only input
output: "/docker_shared/output" # Read-write output
temp: "/docker_shared/temp" # Read-write temp
handoff: "/docker_shared/handoff" # Read-write handoffs
work: "/docker_shared/work" # Read-write workspace
agent_subdirs:
analytics: ["input/analytics", "output/analytics", "temp/analytics", "handoff/analytics"]
media: ["input/media", "output/media", "temp/media", "handoff/media"]
code: ["input/code", "output/code", "temp/code", "handoff/code"]
# ... other agents
```
#### Third-Party Developer Project Structure
When developers install `ambivo-agents` via `pip install ambivo-agents`, the Docker shared directory is created relative to their project root. Here's how the directory structure would look:
```
my-ai-project/ # Third-party developer's project
├── main.py # Their application code
├── requirements.txt # Including ambivo-agents
├── agent_config.yaml # Their configuration file
├── data/ # Their project data
│ ├── input_files.csv
│ └── documents.pdf
├── docker_shared/ # Auto-created by ambivo-agents
│ ├── input/ # Container read-only mounts
│ │ ├── analytics/ # For data analysis tasks
│ │ │ └── uploaded_data.csv
│ │ ├── media/ # For media processing
│ │ │ └── video_to_process.mp4
│ │ └── code/ # For code execution
│ │ └── user_script.py
│ ├── output/ # Container write-enabled results
│ │ ├── analytics/ # Analysis results
│ │ │ ├── report.json
│ │ │ └── charts.png
│ │ ├── media/ # Processed media
│ │ │ ├── audio_extracted.mp3
│ │ │ └── compressed_video.mp4
│ │ └── code/ # Code execution results
│ │ └── execution_results.txt
│ ├── temp/ # Temporary files during processing
│ │ ├── analytics/
│ │ ├── media/
│ │ └── code/
│ ├── handoff/ # Cross-agent file sharing
│ │ ├── analytics/ # Database → Analytics
│ │ ├── database/ # Database exports
│ │ ├── media/ # Media processing handoffs
│ │ └── scraper/ # Web scraper results
│ └── work/ # Container workspace
└── venv/ # Their virtual environment
└── lib/python3.x/site-packages/
└── ambivo_agents/ # Installed package
```
**Environment Variable Configuration:**
Developers can customize the shared directory location:
```bash
# In their .env or environment
export AMBIVO_AGENTS_DOCKER_SHARED_BASE_DIR="/custom/path/shared"
```
**Example Usage in Developer's Code:**
```python
# my-ai-project/main.py
from ambivo_agents.agents.analytics import AnalyticsAgent
from ambivo_agents.agents.moderator import ModeratorAgent
# Create agents - they automatically use configured shared directory
moderator = ModeratorAgent.create_simple(user_id="developer123")
# Process data - files are managed in docker_shared/
response = await moderator.chat("analyze the sales data in my CSV file")
# The docker_shared/ directory is automatically created and managed
# Input files are accessible at docker_shared/input/analytics/
# Results appear in docker_shared/output/analytics/
```
**Benefits for Third-Party Developers:**
- **Isolated**: Each project gets its own `docker_shared/` directory
- **Portable**: Directory structure is relative to project root
- **Configurable**: Can be customized via environment variables
- **Auto-managed**: Created and organized automatically
- **Secure**: Container access is properly restricted
#### Security & Permissions
- **Input Security**: All input directories mounted read-only (`ro`)
- **Output Isolation**: Each agent has isolated output directories
- **Network Isolation**: Docker containers run with `network_disabled=True`
- **Memory Limits**: Configurable memory restrictions per agent
- **Auto-Cleanup**: Temporary files cleaned based on age (configurable)
- **Permission Control**: Directory permissions managed automatically
#### File Access Security Configuration
**Restricted Directories Protection:**
The system includes built-in protection against accessing sensitive system directories:
```yaml
# agent_config.yaml
security:
file_access:
restricted_directories:
- "/etc" # System configuration
- "/root" # Root user directory
- "/var/log" # System logs
- "/proc" # Process information
- "/sys" # System information
- "/dev" # Device files
- "/boot" # Boot files
- "~/.ssh" # SSH keys
- "~/.aws" # AWS credentials
- "~/.config" # User configuration
- "/usr/bin" # System binaries
- "/usr/sbin" # System admin binaries
```
**Environment Variable Configuration:**
```bash
# Alternative to YAML configuration
export AMBIVO_AGENTS_FILE_ACCESS_RESTRICTED_DIRS="/etc,/var/log,/sys,/proc,/dev"
```
**How Restricted Directories Work:**
- **Path Resolution**: Uses `Path.expanduser().resolve()` for proper path handling
- **Security by Default**: Common sensitive directories blocked by default
- **Symbolic Link Protection**: Resolves symbolic links to prevent bypass attempts
- **Cross-Platform**: Works on Windows, macOS, and Linux
- **Agent Coverage**: Protects both `BaseAgent.read_file()` and `DatabaseAgent.ingest_file_to_mongodb()`
**Example Usage:**
```python
from ambivo_agents import DatabaseAgent
# This will be blocked by default security settings
result = await db_agent.chat("ingest data from /etc/passwd")
# Returns: {"success": False, "error": "Access denied: File path '/etc/passwd' is in a restricted directory"}
# This works normally (assuming file exists)
result = await db_agent.chat("ingest data from ./data/users.csv")
# Returns: {"success": True, ...}
```
**Security Best Practices:**
- **Always use** the default restricted directories in production
- **Add custom** restricted paths for your specific environment
- **Test security** settings before deployment
- **Monitor access** attempts to restricted directories
- **Regular audits** of file access patterns
#### Monitoring & Maintenance
```python
from ambivo_agents.core import get_shared_manager
shared_manager = get_shared_manager()
# Monitor disk usage
usage = shared_manager.get_disk_usage()
print(f"Total usage: {usage['total_bytes'] / (1024**3):.2f} GB")
# Cleanup old temporary files
cleaned_count = shared_manager.cleanup_temp_files(max_age_hours=24)
print(f"Cleaned {cleaned_count} temporary files")
# List outputs for specific agent
output_files = shared_manager.list_outputs("analytics")
print(f"Analytics outputs: {output_files}")
```
#### Supported File Types & Detection
The system automatically detects file paths in natural language and supports:
**Data Files**: `.csv`, `.xlsx`, `.xls`, `.json`, `.xml`, `.parquet`
**Media Files**: `.mp4`, `.avi`, `.mov`, `.mp3`, `.wav`, `.flac`
**Text Files**: `.txt`, `.md`, `.log`, `.py`, `.js`, `.sql`
**Documents**: `.pdf` (read-only)
```
# These requests automatically trigger file sharing:
"convert data.csv to xlsx" → Detects: data.csv → ./docker_shared/input/code/
"extract audio from video.mp4" → Detects: video.mp4 → ./docker_shared/input/media/
"analyze quarterly_report.xlsx" → Detects: quarterly_report.xlsx → ./docker_shared/input/analytics/
"scrape data from website" → No file detected → ./docker_shared/output/scraper/
```
#### Docker Image Configuration
**Default Image**: `sgosain/amb-ubuntu-python-public-pod`
**Custom Docker Image for Consolidated Structure**:
```dockerfile
FROM sgosain/amb-ubuntu-python-public-pod
# Install additional packages for your use case
RUN pip install openpyxl xlsxwriter plotly seaborn
# Create consolidated mount points
RUN mkdir -p /docker_shared/{input,output,temp,handoff,work}
# Add custom scripts that work with consolidated structure
COPY your-scripts/ /opt/scripts/
# Set working directory
WORKDIR /docker_shared/work
# Example script that uses consolidated paths
RUN echo '#!/bin/bash\n\
echo "Input files: $(ls -la /docker_shared/input/)"\n\
echo "Output directory: /docker_shared/output/"\n\
echo "Temp directory: /docker_shared/temp/"\n\
echo "Handoff directory: /docker_shared/handoff/"' > /opt/scripts/show_structure.sh
RUN chmod +x /opt/scripts/show_structure.sh
```
#### Troubleshooting
**Directory Issues:**
```bash
# Check if docker_shared structure exists
ls -la docker_shared/
# Verify agent subdirectories
ls -la docker_shared/output/
```
**File Access Issues:**
```bash
# Check permissions
chmod 755 docker_shared/
find docker_shared/ -type d -exec chmod 755 {} \;
# Verify Docker can access the directory
docker run --rm -v $(pwd)/docker_shared:/docker_shared alpine ls -la /docker_shared
```
**Volume Mount Issues:**
```bash
# Test consolidated volume mounting
docker run --rm \
-v $(pwd)/docker_shared/input:/docker_shared/input:ro \
-v $(pwd)/docker_shared/output:/docker_shared/output:rw \
alpine ls -la /docker_shared/
```
#### Benefits of Consolidated Structure
✅ **Consistency**: All agents use the same directory structure
✅ **Inter-Agent Workflows**: Seamless file handoffs between agents
✅ **Security**: Proper read-only/read-write permissions
✅ **Organization**: Files organized by agent and purpose
✅ **Monitoring**: Centralized disk usage and cleanup
✅ **Third-Party Integration**: Easy for custom agent development
✅ **Auto-Management**: Directories created and managed automatically
## Troubleshooting
### Common Issues
1. **Redis Connection Failed**
```bash
# Check if Redis is running
redis-cli ping # Should return "PONG"
```
2. **Docker Not Available**
```bash
# Check Docker is running
docker ps
```
3. **Agent Creation Errors**
```python
from ambivo_agents import ModeratorAgent
try:
moderator, context = ModeratorAgent.create(user_id="test")
print(f"Success: {context.session_id}")
await moderator.cleanup_session()
except Exception as e:
print(f"Error: {e}")
```
4. **Import Errors**
```bash
python -c "from ambivo_agents import ModeratorAgent; print('Import success')"
```
### Debug Mode
Enable verbose logging:
```yaml
service:
log_level: "DEBUG"
log_to_file: true
```
## Security Considerations
- **Docker Isolation**: All code execution happens in isolated containers
- **Network Restrictions**: Containers run with `network_disabled=True` by default
- **Resource Limits**: Memory and CPU limits prevent resource exhaustion
- **API Key Management**: Store sensitive keys in environment variables
- **Input Sanitization**: All user inputs are validated and sanitized
- **Session Isolation**: Each agent session is completely isolated
## Contributing
We welcome contributions! Please see our [Contributing Guidelines](CONTRIBUTING.md) for details.
### Development Setup
```bash
# Clone repository
git clone https://github.com/ambivo-corp/ambivo-agents.git
cd ambivo-agents
# Install in development mode
pip install -e .
# Test ModeratorAgent
python -c "
from ambivo_agents import ModeratorAgent
import asyncio
async def test():
moderator, context = ModeratorAgent.create(user_id='test')
response = await moderator.chat('Hello!')
print(f'Response: {response}')
await moderator.cleanup_session()
asyncio.run(test())
"
```
## License
MIT License - see [LICENSE](LICENSE) file for details.
## Author
**Hemant Gosain 'Sunny'**
- Company: [Ambivo](https://www.ambivo.com)
- Email: info@ambivo.com
## Support
- Email: info@ambivo.com
- Website: https://www.ambivo.com
- Issues: [GitHub Issues](https://github.com/ambivo-corp/ambivo-agents/issues)
## Attributions & Third-Party Technologies
This project leverages several open-source libraries and commercial services:
### Core Technologies
- **Docker**: Container runtime for secure code execution
- **Redis**: In-memory data store for session management
- **Python**: Core programming language
### AI/ML Frameworks
- **OpenAI**: GPT models and API services
- **Anthropic**: Claude models and API services
- **AWS Bedrock**: Cloud-based AI model access
- **LangChain**: Framework for building AI applications (by LangChain, Inc.)
- **LlamaIndex**: Data framework for LLM applications (by Jerry Liu)
- **Hugging Face**: Model hub and transformers library
### Data Processing
- **pandas**: Data analysis and manipulation library
- **DuckDB**: In-process SQL OLAP database
- **Qdrant**: Vector database for semantic search
- **tabulate**: ASCII table formatting library
### Media & Web
- **FFmpeg**: Multimedia processing framework
- **YouTube**: Video platform (via public APIs)
- **pytubefix**: YouTube video downloader library
- **Brave Search**: Web search API service
- **Beautiful Soup**: HTML/XML parsing library
### Development Tools
- **pytest**: Testing framework
- **black**: Code formatting tool
- **Docker Hub**: Container image repository
## Legal Disclaimer
**IMPORTANT**: This software is provided "as is" without warranty of any kind. Users are responsible for:
1. **API Compliance**: Ensuring compliance with all third-party service terms (OpenAI, Anthropic, AWS, YouTube, etc.)
2. **Data Privacy**: Protecting user data and complying with applicable privacy laws
3. **Usage Limits**: Respecting rate limits and usage policies of external services
4. **Security**: Implementing appropriate security measures for production use
5. **Licensing**: Ensuring compliance with all third-party library licenses
The authors and contributors are not liable for any damages arising from the use of this software. Users should thoroughly test and validate the software before production deployment.
**Third-Party Services**: This library integrates with external services that have their own terms of service, privacy policies, and usage limitations. Users must comply with all applicable terms.
**Web Scraping & Content Access**: Users must practice ethical web scraping by respecting robots.txt, rate limits, and website terms of service. YouTube content access must comply with YouTube's Terms of Service and API policies - downloading copyrighted content without permission is prohibited.
---
*Developed by the Ambivo team.*
Raw data
{
"_id": null,
"home_page": "https://github.com/ambivo-corp/ambivo-agents",
"name": "ambivo-agents",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": null,
"author": "Hemant Gosain 'Sunny'",
"author_email": "Hemant Gosain 'Sunny' <info@ambivo.com>",
"download_url": "https://files.pythonhosted.org/packages/c2/ce/c10fb77a9fc26c9abe6b76c107ce9752b5f00761aba469fbfd138d8834b8/ambivo_agents-1.2.5.tar.gz",
"platform": null,
"description": "# Ambivo Agents - Multi-Agent AI System\n\nA 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.\n\n## Alpha Release Disclaimer\n\n**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.\n\nFor production scenarios, we recommend:\n- Extensive testing in your specific environment\n- Implementing proper error handling and monitoring\n- Having rollback plans in place\n- Staying updated with releases for critical fixes\n\n## Table of Contents\n\n- [Quick Start](#quick-start)\n- [Agent Creation](#agent-creation)\n- [Features](#features)\n- [Available Agents](#available-agents)\n- [Workflow System](#workflow-system)\n- [System Messages](#system-messages)\n- [Prerequisites](#prerequisites)\n- [Installation](#installation)\n- [Configuration](#configuration)\n- [Configuration Methods](#configuration-methods)\n- [Project Structure](#project-structure)\n- [Usage Examples](#usage-examples)\n- [Streaming System](#streaming-system)\n- [Session Management](#session-management)\n- [Web API Integration](#web-api-integration)\n- [Command Line Interface](#command-line-interface)\n- [Architecture](#architecture)\n- [Docker Setup](#docker-setup)\n - [Agent Handoff Mechanism](#agent-handoff-mechanism)\n - [File Access Security Configuration](#file-access-security-configuration)\n- [Troubleshooting](#troubleshooting)\n- [Security Considerations](#security-considerations)\n- [Contributing](#contributing)\n- [License](#license)\n- [Author](#author)\n- [Support](#support)\n\n## Quick Start\n\n### ModeratorAgent Example\n\nThe **ModeratorAgent** automatically routes queries to specialized agents:\n\n```python\nfrom ambivo_agents import ModeratorAgent\nimport asyncio\n\nasync def main():\n # Create the moderator\n moderator, context = ModeratorAgent.create(user_id=\"john\")\n \n print(f\"Session: {context.session_id}\")\n \n # Auto-routing examples\n response1 = await moderator.chat(\"Download audio from https://youtube.com/watch?v=example\")\n response2 = await moderator.chat(\"Search for latest AI trends\")\n response3 = await moderator.chat(\"Extract audio from video.mp4 as MP3\")\n response4 = await moderator.chat(\"GET https://api.github.com/users/octocat\")\n response5 = await moderator.chat(\"Load data from sales.csv and analyze it\")\n response6 = await moderator.chat(\"What is machine learning?\")\n \n # Check available agents\n status = await moderator.get_agent_status()\n print(f\"Available agents: {list(status['active_agents'].keys())}\")\n \n # Cleanup\n await moderator.cleanup_session()\n\nasyncio.run(main())\n```\n\n### Command Line Usage\n\n```bash\n# Install and run\npip install ambivo-agents\n\n# Interactive mode\nambivo-agents\n\n# Single commands\nambivo-agents -q \"Download audio from https://youtube.com/watch?v=example\"\nambivo-agents -q \"Search for Python tutorials\"\nambivo-agents -q \"Load data from sales.csv and analyze it\"\nambivo-agents -q \"GET https://jsonplaceholder.typicode.com/posts/1\"\n```\n\n## Agent Creation\n\n### ModeratorAgent (Recommended)\n\n```python\nfrom ambivo_agents import ModeratorAgent\n\n# Create moderator with auto-routing\nmoderator, context = ModeratorAgent.create(user_id=\"john\")\n\n# Chat with automatic agent selection\nresult = await moderator.chat(\"Download audio from https://youtube.com/watch?v=example\")\n\n# Cleanup\nawait moderator.cleanup_session()\n```\n\n**Use ModeratorAgent for:**\n- Multi-purpose applications\n- Intelligent routing between capabilities\n- Context-aware conversations\n- Simplified development\n\n### Direct Agent Creation\n\n```python\nfrom ambivo_agents import YouTubeDownloadAgent\n\n# Create specific agent\nagent, context = YouTubeDownloadAgent.create(user_id=\"john\")\n\n# Use agent directly\nresult = await agent._download_youtube_audio(\"https://youtube.com/watch?v=example\")\n\n# Cleanup\nawait agent.cleanup_session()\n```\n\n**Use Direct Creation for:**\n- Single-purpose applications\n- Specific workflows with known requirements\n- Performance-critical applications\n- Custom integrations\n\n## Features\n\n### Core Capabilities\n- **ModeratorAgent**: Intelligent multi-agent orchestrator with automatic routing\n- **Smart Routing**: Automatically routes queries to appropriate specialized agents\n- **Data Analytics**: In-memory DuckDB integration with CSV/XLS ingestion and text-based visualizations\n- **File Ingestion & Processing**: All agents can read/parse JSON, CSV, XML, YAML files and insert into databases\n- **Context Memory**: Maintains conversation history across interactions\n- **Docker Integration**: Secure, isolated execution environment\n- **Redis Memory**: Persistent conversation memory with compression\n- **Multi-Provider LLM**: Automatic failover between OpenAI, Anthropic, and AWS Bedrock\n- **Configuration-Driven**: All features controlled via `agent_config.yaml`\n- **Workflow System**: Multi-agent workflows with parallel and sequential execution\n- **System Messages**: Customizable system prompts for agent behavior control\n\n## Available Agents\n\n### ModeratorAgent \n- Intelligent orchestrator that routes to specialized agents\n- Context-aware multi-turn conversations\n- Automatic agent selection based on query analysis\n- Session management and cleanup\n- Workflow execution and coordination\n\n### Database Agent (Optional)\n- Secure connections to MongoDB, MySQL, and PostgreSQL databases\n- Schema inspection and table structure exploration\n- Natural language to SQL/MongoDB query conversion\n- Safe query execution with read-only mode by default\n- **File ingestion support**: Insert JSON/CSV files directly into database tables\n- Data export to CSV for analytics handoff\n- Automatic integration with Analytics Agent for visualization\n- Query result formatting with tables and statistics\n- **Note**: Requires installation with `pip install ambivo-agents[database]`\n\n### Analytics Agent\n- CSV/XLS file ingestion into in-memory DuckDB\n- Schema exploration and data quality assessment \n- Natural language to SQL query conversion\n- Text-based chart generation (bar charts, line charts, tables)\n- Chart recommendations based on data characteristics\n- Docker-only execution for security\n- Business intelligence and data insights\n\n### Assistant Agent\n- General purpose conversational AI\n- Context-aware responses\n- Multi-turn conversations\n- Customizable system messages\n\n### Code Executor Agent\n- Secure Python and Bash execution in Docker\n- Isolated environment with resource limits\n- Real-time output streaming\n\n### Web Search Agent\n- Multi-provider search (Brave, AVES APIs)\n- Academic search capabilities\n- Automatic provider failover\n\n### Web Scraper Agent\n- Proxy-enabled scraping (ScraperAPI compatible)\n- Playwright and requests-based scraping\n- Batch URL processing with rate limiting\n\n### Knowledge Base Agent\n- Document ingestion (PDF, DOCX, TXT, web URLs)\n- Vector similarity search with Qdrant\n- Semantic question answering\n\n### Media Editor Agent\n- Audio/video processing with FFmpeg\n- Format conversion, resizing, trimming\n- Audio extraction and volume adjustment\n\n### YouTube Download Agent\n- Download videos and audio from YouTube\n- Docker-based execution with pytubefix\n- Automatic title sanitization and metadata extraction\n\n### API Agent\n- Comprehensive HTTP/REST API integration\n- Multiple authentication methods (Bearer, API Key, Basic, OAuth2)\n- Pre-fetch authentication with automatic token refresh\n- Built-in retry logic with exponential backoff\n- Security features (domain filtering, SSL verification)\n- Support for all HTTP methods (GET, POST, PUT, DELETE, etc.)\n\n## Workflow System\n\nThe workflow system enables multi-agent orchestration with sequential and parallel execution patterns:\n\n### Basic Workflow Usage\n\n```python\nfrom ambivo_agents.core.workflow import WorkflowBuilder, WorkflowPatterns\nfrom ambivo_agents import ModeratorAgent\n\nasync def workflow_example():\n # Create moderator with agents\n moderator, context = ModeratorAgent.create(\n user_id=\"workflow_user\",\n enabled_agents=['web_search', 'web_scraper', 'knowledge_base']\n )\n \n # Create search -> scrape -> ingest workflow\n workflow = WorkflowPatterns.create_search_scrape_ingest_workflow(\n moderator.specialized_agents['web_search'],\n moderator.specialized_agents['web_scraper'], \n moderator.specialized_agents['knowledge_base']\n )\n \n # Execute workflow\n result = await workflow.execute(\n \"Research renewable energy trends and store in knowledge base\",\n context.to_execution_context()\n )\n \n if result.success:\n print(f\"Workflow completed in {result.execution_time:.2f}s\")\n print(f\"Nodes executed: {result.nodes_executed}\")\n \n await moderator.cleanup_session()\n```\n\n### Advanced Workflow Features\n\n```python\nfrom ambivo_agents.core.enhanced_workflow import (\n AdvancedWorkflowBuilder, EnhancedModeratorAgent\n)\n\nasync def advanced_workflow():\n # Create enhanced moderator\n base_moderator, context = ModeratorAgent.create(user_id=\"advanced_user\")\n enhanced_moderator = EnhancedModeratorAgent(base_moderator)\n \n # Natural language workflow triggers\n response1 = await enhanced_moderator.process_message_with_workflows(\n \"I need agents to reach consensus on climate solutions\"\n )\n \n response2 = await enhanced_moderator.process_message_with_workflows(\n \"Create a debate between agents about AI ethics\"\n )\n \n # Check workflow status\n status = await enhanced_moderator.get_workflow_status()\n print(f\"Available workflows: {status['advanced_workflows']['registered']}\")\n```\n\n### Workflow Patterns\n\n- **Sequential Workflows**: Execute agents in order, passing results between them\n- **Parallel Workflows**: Execute multiple agents simultaneously\n- **Consensus Workflows**: Agents collaborate to reach agreement\n- **Debate Workflows**: Structured multi-agent discussions\n- **Error Recovery**: Automatic fallback to backup agents\n- **Map-Reduce**: Parallel processing with result aggregation\n\n## System Messages\n\nSystem messages control agent behavior and responses. Each agent supports custom system prompts:\n\n### Default System Messages\n\n```python\n# Agents come with role-specific system messages\nassistant_agent = AssistantAgent.create_simple(user_id=\"user\")\n# Default: \"You are a helpful AI assistant. Provide accurate, thoughtful responses...\"\n\ncode_agent = CodeExecutorAgent.create_simple(user_id=\"user\") \n# Default: \"You are a code execution specialist. Write clean, well-commented code...\"\n```\n\n### Custom System Messages\n\n```python\nfrom ambivo_agents import AssistantAgent\n\n# Create agent with custom system message\ncustom_system = \"\"\"You are a technical documentation specialist. \nAlways provide detailed explanations with code examples. \nUse professional terminology and structured responses.\"\"\"\n\nagent, context = AssistantAgent.create(\n user_id=\"doc_specialist\",\n system_message=custom_system\n)\n\nresponse = await agent.chat(\"Explain REST API design principles\")\n```\n\n### Moderator System Messages\n\n```python\nfrom ambivo_agents import ModeratorAgent\n\n# Custom moderator behavior\nmoderator_system = \"\"\"You are a project management assistant.\nRoute technical queries to appropriate agents and provide \nexecutive summaries of complex multi-agent interactions.\"\"\"\n\nmoderator, context = ModeratorAgent.create(\n user_id=\"pm_user\",\n system_message=moderator_system\n)\n```\n\n### System Message Features\n\n- **Context Integration**: System messages work with conversation history\n- **Agent-Specific**: Each agent type has optimized default prompts\n- **Workflow Aware**: System messages adapt to workflow contexts\n- **Provider Compatibility**: Works with all LLM providers (OpenAI, Anthropic, Bedrock)\n\n## Prerequisites\n\n### Required\n- **Python 3.11+**\n- **Docker** (for code execution, media processing, YouTube downloads)\n- **Redis** (Cloud Redis recommended)\n\n### API Keys (Optional - based on enabled features)\n- **OpenAI API Key** (for GPT models)\n- **Anthropic API Key** (for Claude models)\n- **AWS Credentials** (for Bedrock models)\n- **Brave Search API Key** (for web search)\n- **AVES API Key** (for web search)\n- **ScraperAPI/Proxy credentials** (for web scraping)\n- **Qdrant Cloud API Key** (for Knowledge Base operations)\n- **Redis Cloud credentials** (for memory management)\n\n## Installation\n\n### 1. Install Dependencies\n\n**Core Installation (without database support):**\n```bash\npip install -r requirements.txt\n```\n\n**With Optional Database Support:**\n```bash\n# Install with database capabilities (MongoDB, MySQL, PostgreSQL)\npip install ambivo-agents[database]\n\n# Or install all optional features including database support\npip install ambivo-agents[all]\n```\n\nThe database agents are optional and require additional dependencies:\n- **MongoDB**: `pymongo>=4.0.0`\n- **MySQL**: `mysql-connector-python>=8.0.0`\n- **PostgreSQL**: `psycopg2-binary>=2.9.0`\n\n### 2. Setup Docker Images\n```bash\ndocker pull sgosain/amb-ubuntu-python-public-pod\n```\n\n### 3. Setup Redis\n\n**Recommended: Cloud Redis**\n```yaml\n# In agent_config.yaml\nredis:\n host: \"your-redis-cloud-endpoint.redis.cloud\"\n port: 6379\n password: \"your-redis-password\"\n```\n\n**Alternative: Local Redis**\n```bash\n# Using Docker\ndocker run -d --name redis -p 6379:6379 redis:latest\n```\n\n## Configuration\n\nCreate `agent_config.yaml` in your project root:\n\n```yaml\n# Redis Configuration (Required)\nredis:\n host: \"your-redis-cloud-endpoint.redis.cloud\"\n port: 6379\n db: 0\n password: \"your-redis-password\"\n\n# LLM Configuration (Required - at least one provider)\nllm:\n preferred_provider: \"openai\"\n temperature: 0.7\n openai_api_key: \"your-openai-key\"\n anthropic_api_key: \"your-anthropic-key\"\n aws_access_key_id: \"your-aws-key\"\n aws_secret_access_key: \"your-aws-secret\"\n aws_region: \"us-east-1\"\n\n# Agent Capabilities\nagent_capabilities:\n enable_knowledge_base: true\n enable_web_search: true\n enable_code_execution: true\n enable_file_processing: true\n enable_web_ingestion: true\n enable_api_calls: true\n enable_web_scraping: true\n enable_proxy_mode: true\n enable_media_editor: true\n enable_youtube_download: true\n enable_analytics: true\n\n# ModeratorAgent default agents\nmoderator:\n default_enabled_agents:\n - knowledge_base\n - web_search\n - assistant\n - media_editor\n - youtube_download\n - code_executor\n - web_scraper\n - analytics\n\n# Service-specific configurations\nweb_search:\n brave_api_key: \"your-brave-api-key\"\n avesapi_api_key: \"your-aves-api-key\"\n\nknowledge_base:\n qdrant_url: \"https://your-cluster.qdrant.tech\"\n qdrant_api_key: \"your-qdrant-api-key\"\n chunk_size: 1024\n chunk_overlap: 20\n similarity_top_k: 5\n\nyoutube_download:\n docker_image: \"sgosain/amb-ubuntu-python-public-pod\"\n download_dir: \"./youtube_downloads\"\n timeout: 600\n memory_limit: \"1g\"\n default_audio_only: true\n\nanalytics:\n docker_image: \"sgosain/amb-ubuntu-python-public-pod\"\n timeout: 300\n memory_limit: \"1g\"\n enable_url_ingestion: true\n max_file_size_mb: 100\n\ndocker:\n timeout: 60\n memory_limit: \"512m\"\n images: [\"sgosain/amb-ubuntu-python-public-pod\"]\n```\n\n## Configuration Methods\n\nThe library supports two configuration methods:\n\n### 1. Environment Variables (Recommended for Production)\n\n**Quick Start with Environment Variables:**\n\n```bash\n# Download and edit the full template\ncurl -o set_env.sh https://github.com/ambivo-corp/ambivo-agents/raw/main/set_env_template.sh\nchmod +x set_env.sh\n\n# Edit the template with your credentials, then source it\nsource set_env.sh\n```\n\n**Replace ALL placeholder values** with your actual credentials:\n- Redis connection details\n- LLM API keys (OpenAI/Anthropic)\n- Web Search API keys (Brave/AVES)\n- Knowledge Base credentials (Qdrant)\n- Web Scraping proxy (ScraperAPI)\n\n**Minimal Environment Setup:**\n```bash\n# Required - Redis\nexport AMBIVO_AGENTS_REDIS_HOST=\"your-redis-host.redis.cloud\"\nexport AMBIVO_AGENTS_REDIS_PORT=\"6379\"\nexport AMBIVO_AGENTS_REDIS_PASSWORD=\"your-redis-password\"\n\n# Required - At least one LLM provider\nexport AMBIVO_AGENTS_OPENAI_API_KEY=\"sk-your-openai-key\"\n\n# Optional - Enable specific agents\nexport AMBIVO_AGENTS_ENABLE_YOUTUBE_DOWNLOAD=\"true\"\nexport AMBIVO_AGENTS_ENABLE_WEB_SEARCH=\"true\"\nexport AMBIVO_AGENTS_ENABLE_ANALYTICS=\"true\"\nexport AMBIVO_AGENTS_MODERATOR_ENABLED_AGENTS=\"youtube_download,web_search,analytics,assistant\"\n\n# Run your application\npython your_app.py\n```\n\n### 2. YAML Configuration (Traditional)\n\n**Use the full YAML template:**\n\n```bash\n# Download and edit the full template\ncurl -o agent_config_sample.yaml https://github.com/ambivo-corp/ambivo-agents/raw/main/agent_config_sample.yaml\n\n# Copy to your config file and edit with your credentials\ncp agent_config_sample.yaml agent_config.yaml\n```\n\n**Replace ALL placeholder values** with your actual credentials, then create `agent_config.yaml` in your project root.\n\n### Docker Deployment with Environment Variables\n\n```yaml\n# docker-compose.yml\nversion: '3.8'\nservices:\n ambivo-app:\n image: your-app:latest\n environment:\n - AMBIVO_AGENTS_REDIS_HOST=redis\n - AMBIVO_AGENTS_REDIS_PORT=6379\n - AMBIVO_AGENTS_OPENAI_API_KEY=${OPENAI_API_KEY}\n - AMBIVO_AGENTS_ENABLE_YOUTUBE_DOWNLOAD=true\n - AMBIVO_AGENTS_ENABLE_ANALYTICS=true\n volumes:\n - ./downloads:/app/downloads\n - /var/run/docker.sock:/var/run/docker.sock\n depends_on:\n - redis\n \n redis:\n image: redis:latest\n ports:\n - \"6379:6379\"\n```\n\n**Note:** Environment variables take precedence over YAML configuration. The `agent_config.yaml` file is optional when using environment variables.\n\n## Project Structure\n\n```\nambivo_agents/\n\u251c\u2500\u2500 agents/ # Agent implementations\n\u2502 \u251c\u2500\u2500 analytics.py # Analytics Agent (DuckDB data analysis)\n\u2502 \u251c\u2500\u2500 api_agent.py # API Agent (HTTP/REST integration)\n\u2502 \u251c\u2500\u2500 assistant.py # Assistant Agent (general conversation)\n\u2502 \u251c\u2500\u2500 code_executor.py # Code Executor Agent (Docker-based execution)\n\u2502 \u251c\u2500\u2500 database_agent.py # Database Agent (MongoDB, MySQL, PostgreSQL)\n\u2502 \u251c\u2500\u2500 knowledge_base.py # Knowledge Base Agent (Qdrant vector search)\n\u2502 \u251c\u2500\u2500 media_editor.py # Media Editor Agent (FFmpeg processing)\n\u2502 \u251c\u2500\u2500 moderator.py # ModeratorAgent (main orchestrator)\n\u2502 \u251c\u2500\u2500 web_scraper.py # Web Scraper Agent (Playwright-based)\n\u2502 \u251c\u2500\u2500 web_search.py # Web Search Agent (Brave/AVES search)\n\u2502 \u2514\u2500\u2500 youtube_download.py # YouTube Download Agent (pytubefix)\n\u251c\u2500\u2500 config/ # Configuration management\n\u251c\u2500\u2500 core/ # Core functionality\n\u2502 \u251c\u2500\u2500 base.py\n\u2502 \u251c\u2500\u2500 llm.py\n\u2502 \u251c\u2500\u2500 memory.py\n\u2502 \u251c\u2500\u2500 workflow.py # Basic workflow system\n\u2502 \u2514\u2500\u2500 enhanced_workflow.py # Advanced workflow patterns\n\u251c\u2500\u2500 executors/ # Execution environments\n\u251c\u2500\u2500 services/ # Service layer\n\u251c\u2500\u2500 __init__.py # Package initialization\n\u2514\u2500\u2500 cli.py # Command line interface\n```\n\n## Usage Examples\n\n### ModeratorAgent with Auto-Routing\n\n```python\nfrom ambivo_agents import ModeratorAgent\nimport asyncio\n\nasync def basic_moderator():\n moderator, context = ModeratorAgent.create(user_id=\"demo_user\")\n \n # Auto-routing examples\n examples = [\n \"Download audio from https://youtube.com/watch?v=example\",\n \"Search for latest artificial intelligence news\", \n \"Load data from sales.csv and analyze trends\",\n \"Extract audio from video.mp4 as high quality MP3\",\n \"What is machine learning and how does it work?\",\n ]\n \n for query in examples:\n response = await moderator.chat(query)\n print(f\"Response: {response[:100]}...\")\n \n await moderator.cleanup_session()\n\nasyncio.run(basic_moderator())\n```\n\n### Context-Aware Conversations\n\n```python\nasync def context_conversation():\n moderator, context = ModeratorAgent.create(user_id=\"context_demo\")\n \n # Initial request \n response1 = await moderator.chat(\"Download audio from https://youtube.com/watch?v=example\")\n \n # Follow-up using context\n response2 = await moderator.chat(\"Actually, download the video instead of just audio\")\n \n # Another follow-up\n response3 = await moderator.chat(\"Get information about that video\")\n \n await moderator.cleanup_session()\n```\n\n### YouTube Downloads\n\n```python\nfrom ambivo_agents import YouTubeDownloadAgent\n\nasync def download_youtube():\n agent, context = YouTubeDownloadAgent.create(user_id=\"media_user\")\n \n # Download audio\n result = await agent._download_youtube_audio(\n \"https://youtube.com/watch?v=example\"\n )\n \n if result['success']:\n print(f\"Audio downloaded: {result['filename']}\")\n print(f\"Path: {result['file_path']}\")\n \n await agent.cleanup_session()\n```\n\n### Database Operations\n\n```python\nfrom ambivo_agents import DatabaseAgent\n\nasync def database_demo():\n agent, context = DatabaseAgent.create(user_id=\"db_user\")\n \n # Connect to MySQL database\n response = await agent.chat(\"Connect to MySQL database at localhost:3306, database: mydb, username: user, password: pass\")\n print(f\"Connection: {response}\")\n \n # Explore database schema\n schema = await agent.chat(\"show me the database schema\")\n print(f\"Schema: {schema}\")\n \n # Natural language queries\n result = await agent.chat(\"show me all users from the users table\")\n print(f\"Users: {result}\")\n \n # Query with analytics handoff\n analytics_result = await agent.chat(\"get sales data and create visualizations\")\n print(f\"Analytics: {analytics_result}\")\n \n await agent.cleanup_session()\n\n# Database to Analytics Workflow\nasync def database_analytics_workflow():\n from ambivo_agents import ModeratorAgent\n \n # Use ModeratorAgent for automatic routing\n moderator, context = ModeratorAgent.create(\n user_id=\"workflow_user\",\n enabled_agents=[\"database_agent\", \"analytics\", \"assistant\"]\n )\n \n # Connect and query database\n await moderator.chat(\"Connect to MySQL localhost:3306 database mydb user admin password secret\")\n \n # Query data with automatic analytics handoff\n response = await moderator.chat(\"Get sales data from orders table and create charts showing trends\")\n print(response)\n \n await moderator.cleanup_session()\n```\n\n### File Reading and Database Ingestion\n\nAll agents have built-in file reading capabilities for JSON, CSV, XML, and YAML files. Database insertion requires the optional database package.\n\n```python\nfrom ambivo_agents import AssistantAgent\n\nasync def read_file_and_insert_to_database():\n \"\"\"Reads a JSON file and attempts database insertion with graceful fallback\"\"\"\n \n # Step 1: Read and parse file (always available)\n agent = AssistantAgent.create_simple(user_id=\"file_user\")\n \n result = await agent.read_and_parse_file(\"./data/users.json\")\n if not result['success']:\n print(f\"\u274c Failed to read file: {result.get('error', 'Unknown error')}\")\n await agent.cleanup_session()\n return\n \n json_data = result['parse_result']['data']\n print(f\"\u2705 Successfully loaded {len(json_data)} records from users.json\")\n \n # Step 2: Attempt database insertion\n try:\n from ambivo_agents import DatabaseAgent\n \n # DatabaseAgent is available - proceed with insertion\n db_agent = DatabaseAgent.create_simple(user_id=\"db_user\")\n \n # Connect to MongoDB\n await db_agent.chat(\"connect to mongodb://localhost:27017 database myapp\")\n \n # Insert the data\n response = await db_agent.chat(f\"insert this data into users collection: {json_data}\")\n print(f\"\u2705 Successfully inserted data into MongoDB: {response}\")\n \n await db_agent.cleanup_session()\n \n except ImportError:\n # DatabaseAgent not available - provide polite warning and alternatives\n print(\"\\n\u26a0\ufe0f Database insertion not available\")\n print(\"\ud83d\udca1 To enable database features, install with: pip install ambivo-agents[database]\")\n print(\"\\n\ud83d\udcc1 Available alternatives:\")\n print(\" \u2022 File successfully read and parsed\")\n print(\" \u2022 Data can be transformed to other formats\")\n \n # Show what we can still do\n csv_result = await agent.convert_json_to_csv(json_data)\n if csv_result['success']:\n print(\" \u2022 \u2705 Converted to CSV format (available for export)\")\n \n await agent.cleanup_session()\n\n# Alternative: Natural language approach with graceful handling\nasync def natural_language_file_ingestion():\n \"\"\"Uses natural language commands with automatic fallback\"\"\"\n \n try:\n from ambivo_agents import DatabaseAgent\n agent = DatabaseAgent.create_simple(user_id=\"user\")\n \n # Full database workflow available\n await agent.chat(\"connect to mongodb://localhost:27017 database myapp\")\n response = await agent.chat(\"read users.json file and insert all records into users collection\")\n print(f\"\u2705 Database ingestion completed: {response}\")\n \n await agent.cleanup_session()\n \n except ImportError:\n # Fallback to file reading only\n from ambivo_agents import AssistantAgent\n agent = AssistantAgent.create_simple(user_id=\"user\")\n \n print(\"\u26a0\ufe0f DatabaseAgent not installed. Reading file only...\")\n response = await agent.chat(\"read and analyze the users.json file structure\")\n print(f\"\ud83d\udcc1 File analysis: {response}\")\n print(\"\ud83d\udca1 Install database support with: pip install ambivo-agents[database]\")\n \n await agent.cleanup_session()\n```\n\n### Data Analytics\n\n```python\nfrom ambivo_agents import AnalyticsAgent\n\nasync def analytics_demo():\n agent, context = AnalyticsAgent.create(user_id=\"analyst_user\")\n \n # Load and analyze CSV data\n response = await agent.chat(\"load data from sales.csv and analyze it\")\n print(f\"Analysis: {response}\")\n \n # Schema exploration\n schema = await agent.chat(\"show me the schema of the current dataset\")\n print(f\"Schema: {schema}\")\n \n # Natural language queries\n top_sales = await agent.chat(\"what are the top 5 products by sales?\")\n print(f\"Top Sales: {top_sales}\")\n \n # SQL queries\n sql_result = await agent.chat(\"SELECT region, SUM(sales) as total FROM data GROUP BY region\")\n print(f\"SQL Result: {sql_result}\")\n \n # Visualizations\n chart = await agent.chat(\"create a bar chart showing sales by region\")\n print(f\"Chart: {chart}\")\n \n await agent.cleanup_session()\n\n# Context Preservation Example\nasync def context_preservation_demo():\n \"\"\"Demonstrates context/state preservation between chat messages\"\"\"\n agent = AnalyticsAgent.create_simple(user_id=\"user123\")\n \n try:\n # Load data once\n await agent.chat(\"load data from transactions.xlsx and analyze it\")\n \n # Multiple queries without reload - uses cached context\n schema = await agent.chat(\"show schema\") # \u2705 Uses cached data\n top_items = await agent.chat(\"what are the top 5 amounts?\") # \u2705 Uses cached data\n summary = await agent.chat(\"summary statistics\") # \u2705 Uses cached data\n counts = await agent.chat(\"count by category\") # \u2705 Uses cached data\n \n print(\"All queries executed using cached dataset - no reload needed!\")\n \n finally:\n await agent.cleanup_session() # Clean up resources\n```\n\n### Knowledge Base Operations\n\n```python\nfrom ambivo_agents import KnowledgeBaseAgent\n\nasync def knowledge_base_demo():\n agent, context = KnowledgeBaseAgent.create(user_id=\"kb_user\")\n \n # Ingest document\n result = await agent._ingest_document(\n kb_name=\"company_kb\",\n doc_path=\"/path/to/document.pdf\",\n custom_meta={\"department\": \"HR\", \"type\": \"policy\"}\n )\n \n if result['success']:\n # Query the knowledge base\n answer = await agent._query_knowledge_base(\n kb_name=\"company_kb\",\n query=\"What is the remote work policy?\"\n )\n \n if answer['success']:\n print(f\"Answer: {answer['answer']}\")\n \n await agent.cleanup_session()\n```\n\n### API Integration\n\n```python\nfrom ambivo_agents import APIAgent\nfrom ambivo_agents.agents.api_agent import APIRequest, AuthConfig, HTTPMethod, AuthType\n\nasync def api_integration_demo():\n agent, context = APIAgent.create(user_id=\"api_user\")\n \n # Basic GET request\n request = APIRequest(\n url=\"https://jsonplaceholder.typicode.com/posts/1\",\n method=HTTPMethod.GET\n )\n \n response = await agent.make_api_request(request)\n if not response.error:\n print(f\"Status: {response.status_code}\")\n print(f\"Data: {response.json_data}\")\n \n # POST with authentication\n auth_config = AuthConfig(\n auth_type=AuthType.BEARER,\n token=\"your-api-token\"\n )\n \n post_request = APIRequest(\n url=\"https://api.example.com/data\",\n method=HTTPMethod.POST,\n auth_config=auth_config,\n json_data={\"name\": \"test\", \"value\": \"123\"}\n )\n \n post_response = await agent.make_api_request(post_request)\n \n # Google OAuth2 with pre-fetch\n google_auth = AuthConfig(\n auth_type=AuthType.BEARER,\n pre_auth_url=\"https://www.googleapis.com/oauth2/v4/token\",\n pre_auth_method=HTTPMethod.POST,\n pre_auth_payload={\n \"client_id\": \"your-client-id\",\n \"client_secret\": \"your-secret\",\n \"refresh_token\": \"your-refresh-token\",\n \"grant_type\": \"refresh_token\"\n },\n token_path=\"access_token\"\n )\n \n sheets_request = APIRequest(\n url=\"https://sheets.googleapis.com/v4/spreadsheets/sheet-id/values/A1:B10\",\n method=HTTPMethod.GET,\n auth_config=google_auth\n )\n \n # APIAgent will automatically fetch access_token first, then make the request\n sheets_response = await agent.make_api_request(sheets_request)\n \n await agent.cleanup_session()\n\n# Conversational API usage\nasync def conversational_api():\n agent = APIAgent.create_simple(user_id=\"chat_user\")\n \n # Use natural language for API requests\n response = await agent.chat(\"GET https://jsonplaceholder.typicode.com/users/1\")\n print(response)\n \n response = await agent.chat(\n \"POST https://api.example.com/data with headers: {\\\"Authorization\\\": \\\"Bearer token\\\"} \"\n \"and data: {\\\"message\\\": \\\"Hello API\\\"}\"\n )\n print(response)\n \n await agent.cleanup_session()\n```\n\n### Context Manager Pattern\n\n```python\nfrom ambivo_agents import ModeratorAgent, AgentSession\nimport asyncio\n\nasync def main():\n # Auto-cleanup with context manager\n async with AgentSession(ModeratorAgent, user_id=\"sarah\") as moderator:\n response = await moderator.chat(\"Download audio from https://youtube.com/watch?v=example\")\n print(response)\n # Moderator automatically cleaned up\n\nasyncio.run(main())\n```\n\n### Workflow Examples\n\n```python\nfrom ambivo_agents.core.workflow import WorkflowBuilder\n\nasync def custom_workflow():\n # Create agents\n moderator, context = ModeratorAgent.create(user_id=\"workflow_demo\")\n \n # Build custom workflow\n builder = WorkflowBuilder()\n builder.add_agent(moderator.specialized_agents['web_search'], \"search\")\n builder.add_agent(moderator.specialized_agents['assistant'], \"analyze\")\n builder.add_edge(\"search\", \"analyze\")\n builder.set_start_node(\"search\")\n builder.set_end_node(\"analyze\")\n \n workflow = builder.build()\n \n # Execute workflow\n result = await workflow.execute(\n \"Research AI safety and provide analysis\",\n context.to_execution_context()\n )\n \n print(f\"Workflow result: {result.success}\")\n await moderator.cleanup_session()\n```\n\n## Streaming System\n\nThe library features a modern **StreamChunk** system that provides structured, type-safe streaming responses with rich metadata.\n\n### StreamChunk Overview\n\nAll agents now return `StreamChunk` objects instead of raw strings, enabling:\n- **Type-safe content classification** with `StreamSubType` enum\n- **Rich metadata** for debugging, analytics, and context\n- **Programmatic filtering** without string parsing\n- **Consistent interface** across all agents\n\n### StreamSubType Categories\n\n```python\nfrom ambivo_agents.core.base import StreamSubType\n\n# Available sub-types:\nStreamSubType.CONTENT # Actual response content from LLMs\nStreamSubType.STATUS # Progress updates, thinking, interim info \nStreamSubType.RESULT # Search results, processing outputs\nStreamSubType.ERROR # Error messages and failures\nStreamSubType.METADATA # Additional context and metadata\n```\n\n### Basic Streaming Usage\n\n```python\nfrom ambivo_agents import ModeratorAgent\nfrom ambivo_agents.core.base import StreamSubType\n\nasync def streaming_example():\n moderator, context = ModeratorAgent.create(user_id=\"stream_user\")\n \n # Stream with filtering\n print(\"\ud83e\udd16 Assistant: \", end='', flush=True)\n \n async for chunk in moderator.chat_stream(\"Search for Python tutorials\"):\n # Filter by content type\n if chunk.sub_type == StreamSubType.CONTENT:\n print(chunk.text, end='', flush=True)\n elif chunk.sub_type == StreamSubType.STATUS:\n print(f\"\\n[{chunk.text.strip()}]\", end='', flush=True)\n elif chunk.sub_type == StreamSubType.ERROR:\n print(f\"\\n[ERROR: {chunk.text}]\", end='', flush=True)\n \n await moderator.cleanup_session()\n```\n\n### Advanced Streaming with Metadata\n\n```python\nasync def advanced_streaming():\n moderator, context = ModeratorAgent.create(user_id=\"advanced_user\")\n \n # Collect and analyze stream\n content_chunks = []\n status_chunks = []\n result_chunks = []\n \n async for chunk in moderator.chat_stream(\"Download audio from YouTube\"):\n # Categorize by type\n if chunk.sub_type == StreamSubType.CONTENT:\n content_chunks.append(chunk)\n elif chunk.sub_type == StreamSubType.STATUS:\n status_chunks.append(chunk)\n elif chunk.sub_type == StreamSubType.RESULT:\n result_chunks.append(chunk)\n \n # Access metadata\n agent_info = chunk.metadata.get('agent')\n operation = chunk.metadata.get('operation')\n phase = chunk.metadata.get('phase')\n \n print(f\"[{chunk.sub_type.value}] {chunk.text[:50]}...\")\n if agent_info:\n print(f\" Agent: {agent_info}\")\n if operation:\n print(f\" Operation: {operation}\")\n \n # Analysis\n print(f\"\\nStream Analysis:\")\n print(f\"Content chunks: {len(content_chunks)}\")\n print(f\"Status updates: {len(status_chunks)}\")\n print(f\"Results: {len(result_chunks)}\")\n \n await moderator.cleanup_session()\n```\n\n### Streaming in Web APIs\n\n```python\nfrom fastapi import FastAPI\nfrom fastapi.responses import StreamingResponse\nimport json\n\napp = FastAPI()\n\n@app.post(\"/chat/stream\")\nasync def chat_stream(request: ChatRequest):\n moderator, context = ModeratorAgent.create(user_id=request.user_id)\n \n async def generate_stream():\n async for chunk in moderator.chat_stream(request.message):\n # Convert StreamChunk to JSON\n chunk_data = {\n 'type': 'chunk',\n 'sub_type': chunk.sub_type.value,\n 'text': chunk.text,\n 'metadata': chunk.metadata,\n 'timestamp': chunk.timestamp.isoformat()\n }\n yield f\"data: {json.dumps(chunk_data)}\\n\\n\"\n \n yield \"data: {\\\"type\\\": \\\"done\\\"}\\n\\n\"\n \n return StreamingResponse(generate_stream(), media_type=\"text/stream\")\n```\n\n### Real-time UI Integration\n\n```javascript\n// Frontend streaming handler\nconst eventSource = new EventSource('/chat/stream');\n\neventSource.onmessage = function(event) {\n const data = JSON.parse(event.data);\n \n if (data.type === 'chunk') {\n switch(data.sub_type) {\n case 'content':\n // Display main response content\n appendToChat(data.text);\n break;\n case 'status':\n // Show progress indicator\n updateStatus(data.text);\n break;\n case 'result':\n // Display search results/outputs\n addResult(data.text, data.metadata);\n break;\n case 'error':\n // Handle errors\n showError(data.text);\n break;\n }\n }\n};\n```\n\n### StreamChunk Benefits\n\n**For Developers:**\n- **Type Safety** - No string parsing for content classification\n- **Rich Context** - Access agent info, operation details, timing\n- **Easy Filtering** - Filter streams by content type programmatically\n- **Debugging** - Detailed metadata for troubleshooting\n\n**For Applications:**\n- **Smart UIs** - Show different content types appropriately\n- **Progress Tracking** - Real-time operation status updates\n- **Error Handling** - Structured error information\n- **Analytics** - Performance metrics and usage tracking\n\n\n## Session Management\n\n### Understanding Session vs Conversation IDs\n\nThe library uses two identifiers for context management:\n\n- **session_id**: Represents a broader user session or application context\n- **conversation_id**: Represents a specific conversation thread within a session\n\n```python\n# Single conversation (most common)\nmoderator, context = ModeratorAgent.create(\n user_id=\"john\",\n session_id=\"user_john_main\", \n conversation_id=\"user_john_main\" # Same as session_id\n)\n\n# Multiple conversations per session\nsession_key = \"user_john_tenant_abc\"\n\n# Conversation 1: Data Analysis\nmoderator1, context1 = ModeratorAgent.create(\n user_id=\"john\",\n session_id=session_key,\n conversation_id=\"john_data_analysis_conv\"\n)\n\n# Conversation 2: YouTube Downloads \nmoderator2, context2 = ModeratorAgent.create(\n user_id=\"john\", \n session_id=session_key,\n conversation_id=\"john_youtube_downloads_conv\"\n)\n```\n\n## Web API Integration\n\n```python\nfrom ambivo_agents import ModeratorAgent\nimport asyncio\nimport time\n\nclass ChatAPI:\n def __init__(self):\n self.active_moderators = {}\n \n async def chat_endpoint(self, request_data):\n user_message = request_data.get('message', '')\n user_id = request_data.get('user_id', f\"user_{uuid.uuid4()}\")\n session_id = request_data.get('session_id', f\"session_{user_id}_{int(time.time())}\")\n \n try:\n if session_id not in self.active_moderators:\n moderator, context = ModeratorAgent.create(\n user_id=user_id,\n session_id=session_id\n )\n self.active_moderators[session_id] = moderator\n else:\n moderator = self.active_moderators[session_id]\n \n response_content = await moderator.chat(user_message)\n \n return {\n 'success': True,\n 'response': response_content,\n 'session_id': session_id,\n 'timestamp': time.time()\n }\n \n except Exception as e:\n return {\n 'success': False,\n 'error': str(e),\n 'timestamp': time.time()\n }\n \n async def cleanup_session(self, session_id):\n if session_id in self.active_moderators:\n await self.active_moderators[session_id].cleanup_session()\n del self.active_moderators[session_id]\n```\n\n## Command Line Interface\n\n```bash\n# Interactive mode with auto-routing\nambivo-agents\n\n# Single queries\nambivo-agents -q \"Download audio from https://youtube.com/watch?v=example\"\nambivo-agents -q \"Search for latest AI trends\"\nambivo-agents -q \"Load data from sales.csv and show top products\"\nambivo-agents -q \"Extract audio from video.mp4\"\n\n# Check agent status\nambivo-agents status\n\n# Test all agents\nambivo-agents --test\n\n# Debug mode\nambivo-agents --debug -q \"test query\"\n```\n\n## Architecture\n\n### ModeratorAgent Architecture\n\nThe **ModeratorAgent** acts as an intelligent orchestrator:\n\n```\n[User Query] \n \u2193\n[ModeratorAgent] \u2190 Analyzes intent and context\n \u2193\n[Intent Analysis] \u2190 Uses LLM + patterns + keywords\n \u2193\n[Route Selection] \u2190 Chooses best agent(s)\n \u2193\n[Specialized Agent] \u2190 YouTubeAgent, SearchAgent, etc.\n \u2193\n[Response] \u2190 Combined and contextualized\n \u2193\n[User]\n```\n\n### Workflow Architecture\n\n```\n[WorkflowBuilder] \u2192 [Workflow Definition]\n \u2193 \u2193\n[Workflow Executor] \u2192 [Sequential/Parallel Execution]\n \u2193 \u2193\n[State Management] \u2192 [Persistent Checkpoints]\n \u2193 \u2193\n[Result Aggregation] \u2192 [Final Response]\n```\n\n### Memory System\n- Redis-based persistence with compression and caching\n- Built-in conversation history for every agent\n- Session-aware context with automatic cleanup\n- Multi-session support with isolation\n\n### LLM Provider Management\n- Automatic failover between OpenAI, Anthropic, AWS Bedrock\n- Rate limiting and error handling\n- Provider rotation based on availability and performance\n\n## Docker Setup\n\n### Consolidated Docker File Sharing System\n\nAmbivo Agents uses a **consolidated Docker-shared directory structure** for all file operations across agents. This provides consistent, secure, and efficient file sharing between your host filesystem and Docker containers.\n\n#### Consolidated Directory Structure\n\nAll Docker-based agents (AnalyticsAgent, MediaEditorAgent, CodeExecutorAgent, WebScraperAgent, APIAgent) use the same base structure:\n\n```\nYour Project Directory/\n\u2514\u2500\u2500 docker_shared/ # Consolidated base directory\n \u251c\u2500\u2500 input/ # Read-only input files\n \u2502 \u251c\u2500\u2500 analytics/ \u2192 /docker_shared/input/analytics (AnalyticsAgent)\n \u2502 \u251c\u2500\u2500 media/ \u2192 /docker_shared/input/media (MediaEditorAgent)\n \u2502 \u251c\u2500\u2500 code/ \u2192 /docker_shared/input/code (CodeExecutorAgent)\n \u2502 \u2514\u2500\u2500 scraper/ \u2192 /docker_shared/input/scraper (WebScraperAgent)\n \u251c\u2500\u2500 output/ # Read-write output files\n \u2502 \u251c\u2500\u2500 analytics/ \u2192 /docker_shared/output/analytics (Analysis results)\n \u2502 \u251c\u2500\u2500 media/ \u2192 /docker_shared/output/media (Processed media)\n \u2502 \u251c\u2500\u2500 code/ \u2192 /docker_shared/output/code (Code execution results)\n \u2502 \u2514\u2500\u2500 scraper/ \u2192 /docker_shared/output/scraper (Scraped data)\n \u251c\u2500\u2500 temp/ # Read-write temporary workspace\n \u2502 \u251c\u2500\u2500 analytics/ \u2192 /docker_shared/temp/analytics (Analytics temp files)\n \u2502 \u251c\u2500\u2500 media/ \u2192 /docker_shared/temp/media (Media processing temp)\n \u2502 \u2514\u2500\u2500 code/ \u2192 /docker_shared/temp/code (Code execution temp)\n \u251c\u2500\u2500 handoff/ # Read-write inter-agent file sharing\n \u2502 \u251c\u2500\u2500 database/ \u2192 /docker_shared/handoff/database (Database exports)\n \u2502 \u251c\u2500\u2500 analytics/ \u2192 /docker_shared/handoff/analytics (Analytics results)\n \u2502 \u2514\u2500\u2500 media/ \u2192 /docker_shared/handoff/media (Media for processing)\n \u2514\u2500\u2500 work/ \u2192 /docker_shared/work # General workspace\n```\n\n#### How the System Works\n\nWhen you request operations like \"convert data.csv to xlsx\" or \"process video.mp4\":\n\n1. **File Detection**: System detects file paths in your request\n2. **Directory Setup**: Auto-creates agent-specific subdirectories in `./docker_shared/`\n3. **File Copying**: Copies your files to appropriate input directories\n4. **Docker Execution**: Runs containers with consistent `/docker_shared/` mount points\n5. **Result Retrieval**: Outputs appear in agent-specific output directories\n\n#### Inter-Agent Workflows\n\nThe consolidated structure enables seamless workflows between agents:\n\n**Database \u2192 Analytics Workflow:**\n```\n1. DatabaseAgent exports data \u2192 ./docker_shared/handoff/database/export.csv\n2. AnalyticsAgent automatically \u2192 reads from /docker_shared/handoff/database/\n3. AnalyticsAgent processes data \u2192 outputs to /docker_shared/output/analytics/\n4. Results available at \u2192 ./docker_shared/output/analytics/results.json\n```\n\n#### Agent Handoff Mechanism\n\nThe handoff system uses the `handoff_subdir` parameter to enable seamless file transfers between agents:\n\n**DatabaseAgent \u2192 AnalyticsAgent Handoff:**\n```python\n# DatabaseAgent automatically exports to handoff directory\nresult = await db_agent.chat(\"export sales data for analytics\", \n handoff_subdir=\"sales_analysis_2024\")\n# Creates: ./docker_shared/handoff/database/sales_analysis_2024/\n\n# AnalyticsAgent automatically detects handoff files\nanalytics_result = await analytics_agent.chat(\"analyze sales data\",\n handoff_subdir=\"sales_analysis_2024\")\n# Reads from: ./docker_shared/handoff/database/sales_analysis_2024/\n```\n\n**Handoff Directory Management:**\n- **Automatic creation**: Subdirectories created based on `handoff_subdir` parameter\n- **File naming**: `{agent_type}_{timestamp}_{operation}.{ext}`\n- **Cleanup**: Handoff files older than 24 hours automatically removed\n- **Thread-safe**: Multiple concurrent handoffs supported\n- **Cross-platform**: Works consistently across Windows, macOS, and Linux\n\n**Configuration in `agent_config.yaml`:**\n```yaml\ndocker:\n container_mounts:\n handoff: \"/docker_shared/handoff\"\n \n agent_subdirs:\n database: [\"handoff/database\"]\n analytics: [\"input/analytics\", \"output/analytics\", \"handoff/analytics\"]\n media: [\"input/media\", \"output/media\", \"handoff/media\"]\n```\n\n**Enhanced Fallback (CSV\u2192XLSX Conversion):**\n```\n1. User: \"convert sales.csv to xlsx\"\n2. ModeratorAgent detects file operation need\n3. Copies sales.csv \u2192 ./docker_shared/input/code/sales.csv\n4. CodeExecutorAgent processes \u2192 from /docker_shared/input/code/sales.csv\n5. Outputs converted file \u2192 to /docker_shared/output/code/sales.xlsx\n6. User accesses result at \u2192 ./docker_shared/output/code/sales.xlsx\n```\n\n**Media Processing Workflow:**\n```\n1. User places video \u2192 ./docker_shared/input/media/input.mp4\n2. MediaEditorAgent processes \u2192 from /docker_shared/input/media/input.mp4\n3. Outputs processed file \u2192 to /docker_shared/output/media/output.mp3\n4. User gets result from \u2192 ./docker_shared/output/media/output.mp3\n```\n\n#### Third-Party Developer Integration\n\nFor developers building custom agents:\n\n```python\nfrom ambivo_agents.core import get_shared_manager\n\n# Get the consolidated shared manager\nshared_manager = get_shared_manager()\n\n# Prepare environment for your custom agent\ninput_path, output_path = shared_manager.prepare_agent_environment(\n agent=\"my_custom_agent\",\n input_files=[\"./my_data.csv\"]\n)\n\n# Get Docker volume configuration\nvolumes = shared_manager.get_docker_volumes()\n# volumes = {\n# '/path/to/docker_shared/input': {'bind': '/docker_shared/input', 'mode': 'ro'},\n# '/path/to/docker_shared/output': {'bind': '/docker_shared/output', 'mode': 'rw'},\n# # ... other mounts\n# }\n\n# In your Docker container, access files at:\n# - Input: /docker_shared/input/my_custom_agent/\n# - Output: /docker_shared/output/my_custom_agent/\n# - Temp: /docker_shared/temp/my_custom_agent/\n# - Handoff: /docker_shared/handoff/my_custom_agent/\n\n# After processing, check results:\noutput_files = shared_manager.list_outputs(\"my_custom_agent\")\nlatest_output = shared_manager.get_latest_output(\"my_custom_agent\", \".xlsx\")\n```\n\n#### Example Usage\n\n```python\nimport asyncio\nfrom ambivo_agents import ModeratorAgent\n\nasync def process_files_with_consolidated_structure():\n # Create moderator with auto-routing\n moderator, context = ModeratorAgent.create(user_id=\"file_processor\")\n \n # File operations use consolidated Docker structure\n await moderator.chat(\"convert sales_data.csv to xlsx format\") # \u2192 ./docker_shared/output/code/\n await moderator.chat(\"extract audio from video.mp4 as MP3\") # \u2192 ./docker_shared/output/media/\n await moderator.chat(\"analyze customer_data.csv and chart\") # \u2192 ./docker_shared/output/analytics/\n \n # All results organized by agent type in docker_shared/output/\n await moderator.cleanup_session()\n\n# Run the example\nasyncio.run(process_files_with_consolidated_structure())\n```\n\n#### File Locations After Operations\n\n```bash\n# Directory structure after various operations\nyour-project/\n\u251c\u2500\u2500 sales_data.csv # Your original files\n\u251c\u2500\u2500 video.mp4\n\u251c\u2500\u2500 customer_data.csv\n\u2514\u2500\u2500 docker_shared/ # Consolidated results\n \u2514\u2500\u2500 output/\n \u251c\u2500\u2500 code/\n \u2502 \u2514\u2500\u2500 sales_data.xlsx # CSV\u2192XLSX conversion\n \u251c\u2500\u2500 media/\n \u2502 \u2514\u2500\u2500 video_audio.mp3 # Audio extraction\n \u2514\u2500\u2500 analytics/\n \u251c\u2500\u2500 analysis_report.json # Data analysis\n \u2514\u2500\u2500 customer_charts.png # Generated charts\n```\n\n#### Configuration\n\nThe consolidated structure is configured in `agent_config.yaml`:\n\n```yaml\ndocker:\n shared_base_dir: \"./docker_shared\" # Host base directory\n container_mounts:\n input: \"/docker_shared/input\" # Read-only input\n output: \"/docker_shared/output\" # Read-write output\n temp: \"/docker_shared/temp\" # Read-write temp\n handoff: \"/docker_shared/handoff\" # Read-write handoffs\n work: \"/docker_shared/work\" # Read-write workspace\n agent_subdirs:\n analytics: [\"input/analytics\", \"output/analytics\", \"temp/analytics\", \"handoff/analytics\"]\n media: [\"input/media\", \"output/media\", \"temp/media\", \"handoff/media\"]\n code: [\"input/code\", \"output/code\", \"temp/code\", \"handoff/code\"]\n # ... other agents\n```\n\n#### Third-Party Developer Project Structure\n\nWhen developers install `ambivo-agents` via `pip install ambivo-agents`, the Docker shared directory is created relative to their project root. Here's how the directory structure would look:\n\n```\nmy-ai-project/ # Third-party developer's project\n\u251c\u2500\u2500 main.py # Their application code\n\u251c\u2500\u2500 requirements.txt # Including ambivo-agents\n\u251c\u2500\u2500 agent_config.yaml # Their configuration file\n\u251c\u2500\u2500 data/ # Their project data\n\u2502 \u251c\u2500\u2500 input_files.csv\n\u2502 \u2514\u2500\u2500 documents.pdf\n\u251c\u2500\u2500 docker_shared/ # Auto-created by ambivo-agents\n\u2502 \u251c\u2500\u2500 input/ # Container read-only mounts\n\u2502 \u2502 \u251c\u2500\u2500 analytics/ # For data analysis tasks\n\u2502 \u2502 \u2502 \u2514\u2500\u2500 uploaded_data.csv\n\u2502 \u2502 \u251c\u2500\u2500 media/ # For media processing\n\u2502 \u2502 \u2502 \u2514\u2500\u2500 video_to_process.mp4\n\u2502 \u2502 \u2514\u2500\u2500 code/ # For code execution\n\u2502 \u2502 \u2514\u2500\u2500 user_script.py\n\u2502 \u251c\u2500\u2500 output/ # Container write-enabled results\n\u2502 \u2502 \u251c\u2500\u2500 analytics/ # Analysis results\n\u2502 \u2502 \u2502 \u251c\u2500\u2500 report.json\n\u2502 \u2502 \u2502 \u2514\u2500\u2500 charts.png\n\u2502 \u2502 \u251c\u2500\u2500 media/ # Processed media\n\u2502 \u2502 \u2502 \u251c\u2500\u2500 audio_extracted.mp3\n\u2502 \u2502 \u2502 \u2514\u2500\u2500 compressed_video.mp4\n\u2502 \u2502 \u2514\u2500\u2500 code/ # Code execution results\n\u2502 \u2502 \u2514\u2500\u2500 execution_results.txt\n\u2502 \u251c\u2500\u2500 temp/ # Temporary files during processing\n\u2502 \u2502 \u251c\u2500\u2500 analytics/\n\u2502 \u2502 \u251c\u2500\u2500 media/\n\u2502 \u2502 \u2514\u2500\u2500 code/\n\u2502 \u251c\u2500\u2500 handoff/ # Cross-agent file sharing\n\u2502 \u2502 \u251c\u2500\u2500 analytics/ # Database \u2192 Analytics\n\u2502 \u2502 \u251c\u2500\u2500 database/ # Database exports\n\u2502 \u2502 \u251c\u2500\u2500 media/ # Media processing handoffs\n\u2502 \u2502 \u2514\u2500\u2500 scraper/ # Web scraper results\n\u2502 \u2514\u2500\u2500 work/ # Container workspace\n\u2514\u2500\u2500 venv/ # Their virtual environment\n \u2514\u2500\u2500 lib/python3.x/site-packages/\n \u2514\u2500\u2500 ambivo_agents/ # Installed package\n```\n\n**Environment Variable Configuration:**\n\nDevelopers can customize the shared directory location:\n\n```bash\n# In their .env or environment\nexport AMBIVO_AGENTS_DOCKER_SHARED_BASE_DIR=\"/custom/path/shared\"\n```\n\n**Example Usage in Developer's Code:**\n\n```python\n# my-ai-project/main.py\nfrom ambivo_agents.agents.analytics import AnalyticsAgent\nfrom ambivo_agents.agents.moderator import ModeratorAgent\n\n# Create agents - they automatically use configured shared directory\nmoderator = ModeratorAgent.create_simple(user_id=\"developer123\")\n\n# Process data - files are managed in docker_shared/\nresponse = await moderator.chat(\"analyze the sales data in my CSV file\")\n\n# The docker_shared/ directory is automatically created and managed\n# Input files are accessible at docker_shared/input/analytics/\n# Results appear in docker_shared/output/analytics/\n```\n\n**Benefits for Third-Party Developers:**\n\n- **Isolated**: Each project gets its own `docker_shared/` directory\n- **Portable**: Directory structure is relative to project root\n- **Configurable**: Can be customized via environment variables\n- **Auto-managed**: Created and organized automatically\n- **Secure**: Container access is properly restricted\n\n#### Security & Permissions\n\n- **Input Security**: All input directories mounted read-only (`ro`)\n- **Output Isolation**: Each agent has isolated output directories\n- **Network Isolation**: Docker containers run with `network_disabled=True`\n- **Memory Limits**: Configurable memory restrictions per agent\n- **Auto-Cleanup**: Temporary files cleaned based on age (configurable)\n- **Permission Control**: Directory permissions managed automatically\n\n#### File Access Security Configuration\n\n**Restricted Directories Protection:**\n\nThe system includes built-in protection against accessing sensitive system directories:\n\n```yaml\n# agent_config.yaml\nsecurity:\n file_access:\n restricted_directories:\n - \"/etc\" # System configuration\n - \"/root\" # Root user directory\n - \"/var/log\" # System logs\n - \"/proc\" # Process information\n - \"/sys\" # System information\n - \"/dev\" # Device files\n - \"/boot\" # Boot files\n - \"~/.ssh\" # SSH keys\n - \"~/.aws\" # AWS credentials\n - \"~/.config\" # User configuration\n - \"/usr/bin\" # System binaries\n - \"/usr/sbin\" # System admin binaries\n```\n\n**Environment Variable Configuration:**\n```bash\n# Alternative to YAML configuration\nexport AMBIVO_AGENTS_FILE_ACCESS_RESTRICTED_DIRS=\"/etc,/var/log,/sys,/proc,/dev\"\n```\n\n**How Restricted Directories Work:**\n- **Path Resolution**: Uses `Path.expanduser().resolve()` for proper path handling\n- **Security by Default**: Common sensitive directories blocked by default\n- **Symbolic Link Protection**: Resolves symbolic links to prevent bypass attempts\n- **Cross-Platform**: Works on Windows, macOS, and Linux\n- **Agent Coverage**: Protects both `BaseAgent.read_file()` and `DatabaseAgent.ingest_file_to_mongodb()`\n\n**Example Usage:**\n```python\nfrom ambivo_agents import DatabaseAgent\n\n# This will be blocked by default security settings\nresult = await db_agent.chat(\"ingest data from /etc/passwd\")\n# Returns: {\"success\": False, \"error\": \"Access denied: File path '/etc/passwd' is in a restricted directory\"}\n\n# This works normally (assuming file exists)\nresult = await db_agent.chat(\"ingest data from ./data/users.csv\")\n# Returns: {\"success\": True, ...}\n```\n\n**Security Best Practices:**\n- **Always use** the default restricted directories in production\n- **Add custom** restricted paths for your specific environment\n- **Test security** settings before deployment\n- **Monitor access** attempts to restricted directories\n- **Regular audits** of file access patterns\n\n#### Monitoring & Maintenance\n\n```python\nfrom ambivo_agents.core import get_shared_manager\n\nshared_manager = get_shared_manager()\n\n# Monitor disk usage\nusage = shared_manager.get_disk_usage()\nprint(f\"Total usage: {usage['total_bytes'] / (1024**3):.2f} GB\")\n\n# Cleanup old temporary files\ncleaned_count = shared_manager.cleanup_temp_files(max_age_hours=24)\nprint(f\"Cleaned {cleaned_count} temporary files\")\n\n# List outputs for specific agent\noutput_files = shared_manager.list_outputs(\"analytics\")\nprint(f\"Analytics outputs: {output_files}\")\n```\n\n#### Supported File Types & Detection\n\nThe system automatically detects file paths in natural language and supports:\n\n**Data Files**: `.csv`, `.xlsx`, `.xls`, `.json`, `.xml`, `.parquet` \n**Media Files**: `.mp4`, `.avi`, `.mov`, `.mp3`, `.wav`, `.flac` \n**Text Files**: `.txt`, `.md`, `.log`, `.py`, `.js`, `.sql` \n**Documents**: `.pdf` (read-only)\n\n```\n# These requests automatically trigger file sharing:\n\"convert data.csv to xlsx\" \u2192 Detects: data.csv \u2192 ./docker_shared/input/code/\n\"extract audio from video.mp4\" \u2192 Detects: video.mp4 \u2192 ./docker_shared/input/media/\n\"analyze quarterly_report.xlsx\" \u2192 Detects: quarterly_report.xlsx \u2192 ./docker_shared/input/analytics/\n\"scrape data from website\" \u2192 No file detected \u2192 ./docker_shared/output/scraper/\n```\n\n#### Docker Image Configuration\n\n**Default Image**: `sgosain/amb-ubuntu-python-public-pod`\n\n**Custom Docker Image for Consolidated Structure**:\n\n```dockerfile\nFROM sgosain/amb-ubuntu-python-public-pod\n\n# Install additional packages for your use case\nRUN pip install openpyxl xlsxwriter plotly seaborn\n\n# Create consolidated mount points\nRUN mkdir -p /docker_shared/{input,output,temp,handoff,work}\n\n# Add custom scripts that work with consolidated structure\nCOPY your-scripts/ /opt/scripts/\n\n# Set working directory\nWORKDIR /docker_shared/work\n\n# Example script that uses consolidated paths\nRUN echo '#!/bin/bash\\n\\\necho \"Input files: $(ls -la /docker_shared/input/)\"\\n\\\necho \"Output directory: /docker_shared/output/\"\\n\\\necho \"Temp directory: /docker_shared/temp/\"\\n\\\necho \"Handoff directory: /docker_shared/handoff/\"' > /opt/scripts/show_structure.sh\n\nRUN chmod +x /opt/scripts/show_structure.sh\n```\n\n#### Troubleshooting\n\n**Directory Issues:**\n```bash\n# Check if docker_shared structure exists\nls -la docker_shared/\n\n# Verify agent subdirectories\nls -la docker_shared/output/\n```\n\n**File Access Issues:**\n```bash\n# Check permissions\nchmod 755 docker_shared/\nfind docker_shared/ -type d -exec chmod 755 {} \\;\n\n# Verify Docker can access the directory\ndocker run --rm -v $(pwd)/docker_shared:/docker_shared alpine ls -la /docker_shared\n```\n\n**Volume Mount Issues:**\n```bash\n# Test consolidated volume mounting\ndocker run --rm \\\n -v $(pwd)/docker_shared/input:/docker_shared/input:ro \\\n -v $(pwd)/docker_shared/output:/docker_shared/output:rw \\\n alpine ls -la /docker_shared/\n```\n\n#### Benefits of Consolidated Structure\n\n\u2705 **Consistency**: All agents use the same directory structure \n\u2705 **Inter-Agent Workflows**: Seamless file handoffs between agents \n\u2705 **Security**: Proper read-only/read-write permissions \n\u2705 **Organization**: Files organized by agent and purpose \n\u2705 **Monitoring**: Centralized disk usage and cleanup \n\u2705 **Third-Party Integration**: Easy for custom agent development \n\u2705 **Auto-Management**: Directories created and managed automatically\n\n## Troubleshooting\n\n### Common Issues\n\n1. **Redis Connection Failed**\n ```bash\n # Check if Redis is running\n redis-cli ping # Should return \"PONG\"\n ```\n\n2. **Docker Not Available**\n ```bash\n # Check Docker is running\n docker ps\n ```\n\n3. **Agent Creation Errors**\n ```python\n from ambivo_agents import ModeratorAgent\n try:\n moderator, context = ModeratorAgent.create(user_id=\"test\")\n print(f\"Success: {context.session_id}\")\n await moderator.cleanup_session()\n except Exception as e:\n print(f\"Error: {e}\")\n ```\n\n4. **Import Errors**\n ```bash\n python -c \"from ambivo_agents import ModeratorAgent; print('Import success')\"\n ```\n\n### Debug Mode\n\nEnable verbose logging:\n```yaml\nservice:\n log_level: \"DEBUG\"\n log_to_file: true\n```\n\n## Security Considerations\n\n- **Docker Isolation**: All code execution happens in isolated containers\n- **Network Restrictions**: Containers run with `network_disabled=True` by default\n- **Resource Limits**: Memory and CPU limits prevent resource exhaustion \n- **API Key Management**: Store sensitive keys in environment variables\n- **Input Sanitization**: All user inputs are validated and sanitized\n- **Session Isolation**: Each agent session is completely isolated\n\n## Contributing\n\nWe welcome contributions! Please see our [Contributing Guidelines](CONTRIBUTING.md) for details.\n\n### Development Setup\n\n```bash\n# Clone repository\ngit clone https://github.com/ambivo-corp/ambivo-agents.git\ncd ambivo-agents\n\n# Install in development mode\npip install -e .\n\n# Test ModeratorAgent\npython -c \"\nfrom ambivo_agents import ModeratorAgent\nimport asyncio\n\nasync def test():\n moderator, context = ModeratorAgent.create(user_id='test')\n response = await moderator.chat('Hello!')\n print(f'Response: {response}')\n await moderator.cleanup_session()\n\nasyncio.run(test())\n\"\n```\n\n## License\n\nMIT License - see [LICENSE](LICENSE) file for details.\n\n## Author\n\n**Hemant Gosain 'Sunny'**\n- Company: [Ambivo](https://www.ambivo.com)\n- Email: info@ambivo.com\n\n## Support\n\n- Email: info@ambivo.com\n- Website: https://www.ambivo.com\n- Issues: [GitHub Issues](https://github.com/ambivo-corp/ambivo-agents/issues)\n\n## Attributions & Third-Party Technologies\n\nThis project leverages several open-source libraries and commercial services:\n\n### Core Technologies\n- **Docker**: Container runtime for secure code execution\n- **Redis**: In-memory data store for session management\n- **Python**: Core programming language\n\n### AI/ML Frameworks\n- **OpenAI**: GPT models and API services\n- **Anthropic**: Claude models and API services \n- **AWS Bedrock**: Cloud-based AI model access\n- **LangChain**: Framework for building AI applications (by LangChain, Inc.)\n- **LlamaIndex**: Data framework for LLM applications (by Jerry Liu)\n- **Hugging Face**: Model hub and transformers library\n\n### Data Processing\n- **pandas**: Data analysis and manipulation library\n- **DuckDB**: In-process SQL OLAP database\n- **Qdrant**: Vector database for semantic search\n- **tabulate**: ASCII table formatting library\n\n### Media & Web\n- **FFmpeg**: Multimedia processing framework\n- **YouTube**: Video platform (via public APIs)\n- **pytubefix**: YouTube video downloader library\n- **Brave Search**: Web search API service\n- **Beautiful Soup**: HTML/XML parsing library\n\n### Development Tools\n- **pytest**: Testing framework\n- **black**: Code formatting tool\n- **Docker Hub**: Container image repository\n\n## Legal Disclaimer\n\n**IMPORTANT**: This software is provided \"as is\" without warranty of any kind. Users are responsible for:\n\n1. **API Compliance**: Ensuring compliance with all third-party service terms (OpenAI, Anthropic, AWS, YouTube, etc.)\n2. **Data Privacy**: Protecting user data and complying with applicable privacy laws\n3. **Usage Limits**: Respecting rate limits and usage policies of external services\n4. **Security**: Implementing appropriate security measures for production use\n5. **Licensing**: Ensuring compliance with all third-party library licenses\n\nThe authors and contributors are not liable for any damages arising from the use of this software. Users should thoroughly test and validate the software before production deployment.\n\n**Third-Party Services**: This library integrates with external services that have their own terms of service, privacy policies, and usage limitations. Users must comply with all applicable terms.\n\n**Web Scraping & Content Access**: Users must practice ethical web scraping by respecting robots.txt, rate limits, and website terms of service. YouTube content access must comply with YouTube's Terms of Service and API policies - downloading copyrighted content without permission is prohibited.\n\n---\n\n*Developed by the Ambivo team.*\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Multi-Agent AI System for automation",
"version": "1.2.5",
"project_urls": {
"Company": "https://www.ambivo.com",
"Documentation": "https://github.com/ambivo-corp/ambivo-agents/blob/main/README.md",
"Homepage": "https://github.com/ambivo-corp/ambivo-agents",
"Issues": "https://github.com/ambivo-corp/ambivo-agents/issues",
"Repository": "https://github.com/ambivo-corp/ambivo-agents"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "6b4da1a5ddf1ed921ce823919423f2fbdf29cfbe34461098aaaa111cbe77e105",
"md5": "7e160d17dc6900c14609657236903ef5",
"sha256": "eb314332c4885459baac7747bcb4e28ffebe859846a255a62ac950be5bf3a308"
},
"downloads": -1,
"filename": "ambivo_agents-1.2.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7e160d17dc6900c14609657236903ef5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 321325,
"upload_time": "2025-07-18T23:45:12",
"upload_time_iso_8601": "2025-07-18T23:45:12.205484Z",
"url": "https://files.pythonhosted.org/packages/6b/4d/a1a5ddf1ed921ce823919423f2fbdf29cfbe34461098aaaa111cbe77e105/ambivo_agents-1.2.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "c2cec10fb77a9fc26c9abe6b76c107ce9752b5f00761aba469fbfd138d8834b8",
"md5": "28c53e409ced0d5f5617afb4fe5ab15d",
"sha256": "47dbc1165a92dc4453964c89f93703d688644128866b2e070e23d8930c2a6cfd"
},
"downloads": -1,
"filename": "ambivo_agents-1.2.5.tar.gz",
"has_sig": false,
"md5_digest": "28c53e409ced0d5f5617afb4fe5ab15d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 337973,
"upload_time": "2025-07-18T23:45:14",
"upload_time_iso_8601": "2025-07-18T23:45:14.032706Z",
"url": "https://files.pythonhosted.org/packages/c2/ce/c10fb77a9fc26c9abe6b76c107ce9752b5f00761aba469fbfd138d8834b8/ambivo_agents-1.2.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-18 23:45:14",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "ambivo-corp",
"github_project": "ambivo-agents",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [
{
"name": "redis",
"specs": [
[
"~=",
"6.2.0"
]
]
},
{
"name": "redis",
"specs": []
},
{
"name": "docker",
"specs": [
[
">=",
"6.0.0"
]
]
},
{
"name": "asyncio-mqtt",
"specs": [
[
">=",
"0.11.0"
]
]
},
{
"name": "cachetools",
"specs": [
[
"~=",
"6.1.0"
]
]
},
{
"name": "lz4",
"specs": [
[
"~=",
"4.4.4"
]
]
},
{
"name": "openai",
"specs": [
[
"==",
"1.84.0"
]
]
},
{
"name": "langchain",
"specs": [
[
"==",
"0.3.25"
]
]
},
{
"name": "langchain-community",
"specs": [
[
"==",
"0.3.24"
]
]
},
{
"name": "langchain-core",
"specs": [
[
"==",
"0.3.63"
]
]
},
{
"name": "langchain-openai",
"specs": [
[
"==",
"0.3.19"
]
]
},
{
"name": "langchainhub",
"specs": [
[
"==",
"0.1.21"
]
]
},
{
"name": "langchain-text-splitters",
"specs": [
[
"==",
"0.3.8"
]
]
},
{
"name": "langchain-anthropic",
"specs": [
[
"==",
"0.3.15"
]
]
},
{
"name": "langchain-aws",
"specs": [
[
"~=",
"0.2.24"
]
]
},
{
"name": "langchain-voyageai",
"specs": []
},
{
"name": "huggingface-hub",
"specs": []
},
{
"name": "llama-index-core",
"specs": [
[
"~=",
"0.12.43"
]
]
},
{
"name": "llama-index-embeddings-langchain",
"specs": []
},
{
"name": "llama-index-llms-langchain",
"specs": []
},
{
"name": "llama-index-llms-openai",
"specs": []
},
{
"name": "llama-index-vector-stores-qdrant",
"specs": []
},
{
"name": "llama-index-readers-smart-pdf-loader",
"specs": []
},
{
"name": "pydantic",
"specs": [
[
"==",
"2.11.7"
]
]
},
{
"name": "boto3",
"specs": [
[
"==",
"1.38.42"
]
]
},
{
"name": "python-dotenv",
"specs": [
[
">=",
"1.1.1"
]
]
},
{
"name": "pyyaml",
"specs": [
[
">=",
"6.0.2"
]
]
},
{
"name": "psutil",
"specs": [
[
">=",
"7.0.0"
]
]
},
{
"name": "qdrant-client",
"specs": []
},
{
"name": "numexpr",
"specs": []
},
{
"name": "tabulate",
"specs": [
[
">=",
"0.9.0"
]
]
},
{
"name": "aiosqlite",
"specs": [
[
">=",
"0.19.0"
]
]
},
{
"name": "aiohttp",
"specs": [
[
">=",
"3.8.0"
]
]
},
{
"name": "aiofiles",
"specs": [
[
">=",
"22.1.0"
]
]
},
{
"name": "pytest",
"specs": [
[
">=",
"8.4.1"
]
]
},
{
"name": "pytest-asyncio",
"specs": [
[
">=",
"1.0.0"
]
]
},
{
"name": "black",
"specs": [
[
">=",
"25.1.0"
]
]
},
{
"name": "isort",
"specs": [
[
">=",
"6.0.1"
]
]
},
{
"name": "unstructured",
"specs": []
},
{
"name": "langchain-unstructured",
"specs": []
},
{
"name": "requests",
"specs": [
[
"~=",
"2.32.4"
]
]
},
{
"name": "setuptools",
"specs": [
[
"~=",
"80.9.0"
]
]
},
{
"name": "urllib3",
"specs": [
[
"~=",
"2.5.0"
]
]
},
{
"name": "beautifulsoup4",
"specs": [
[
"~=",
"4.13.4"
]
]
},
{
"name": "click",
"specs": [
[
"~=",
"8.2.1"
]
]
},
{
"name": "python-docx",
"specs": [
[
"~=",
"1.2.0"
]
]
},
{
"name": "python-pptx",
"specs": [
[
"~=",
"1.0.2"
]
]
},
{
"name": "pytesseract",
"specs": [
[
"~=",
"0.3.13"
]
]
},
{
"name": "pandas",
"specs": [
[
"~=",
"2.3.1"
]
]
},
{
"name": "aioconsole",
"specs": [
[
"~=",
"0.8.1"
]
]
},
{
"name": "httpx",
"specs": [
[
"~=",
"0.28.1"
]
]
},
{
"name": "pymongo",
"specs": [
[
"~=",
"4.13.2"
]
]
},
{
"name": "mysql-connector-python",
"specs": [
[
"~=",
"9.3.0"
]
]
},
{
"name": "psycopg2-binary",
"specs": [
[
"~=",
"2.9.10"
]
]
},
{
"name": "pypdf",
"specs": [
[
"~=",
"5.6.1"
]
]
},
{
"name": "pillow",
"specs": [
[
"~=",
"11.2.1"
]
]
}
],
"lcname": "ambivo-agents"
}
