youtube-summarizer/backend/integrations

Agent Framework Integrations

This module provides comprehensive integration support for YouTube Summarizer with popular AI agent frameworks, enabling seamless integration with LangChain, CrewAI, AutoGen, and other agent orchestration systems.

🚀 Quick Start

from integrations.agent_framework import create_youtube_agent_orchestrator

# Create orchestrator with all available frameworks
orchestrator = create_youtube_agent_orchestrator()

# Process a video
result = await orchestrator.process_video(
    "https://youtube.com/watch?v=dQw4w9WgXcQ",
    task_type="summarize"
)

📦 Installation

Core Dependencies

pip install fastapi uvicorn pydantic

Framework-Specific Dependencies

# LangChain
pip install langchain langchain-openai langchain-anthropic

# CrewAI  
pip install crewai

# AutoGen
pip install pyautogen

# Optional: All frameworks
pip install langchain crewai pyautogen

🛠️ Components

1. LangChain Tools (langchain_tools.py)

Pre-built LangChain tools for YouTube processing:

from integrations.langchain_tools import get_youtube_langchain_tools

# Get all tools
tools = get_youtube_langchain_tools()

# Available tools:
# - youtube_transcript: Extract transcripts (YouTube captions or Whisper AI)
# - youtube_summarize: Generate AI summaries with customizable options
# - youtube_batch: Process multiple videos efficiently
# - youtube_search: Search processed videos and summaries

# Use with LangChain agents
from langchain.agents import create_react_agent, AgentExecutor

agent = create_react_agent(llm=your_llm, tools=tools, prompt=your_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = await agent_executor.ainvoke({
    "input": "Summarize this YouTube video: https://youtube.com/watch?v=abc123"
})

Tool Details

YouTube Transcript Tool

• Name: youtube_transcript
• Purpose: Extract transcripts using YouTube captions or Whisper AI
• Inputs: video_url (required), source (youtube/whisper/both), whisper_model (tiny/base/small/medium/large)
• Output: JSON with transcript text and quality metrics

YouTube Summarization Tool

• Name: youtube_summarize
• Purpose: Generate AI-powered video summaries
• Inputs: video_url (required), summary_type (brief/standard/comprehensive/detailed), format (structured/bullet_points/paragraph/narrative)
• Output: Structured summary with key points and insights

YouTube Batch Tool

• Name: youtube_batch
• Purpose: Process multiple videos efficiently
• Inputs: video_urls (list), batch_name, processing_type (transcribe/summarize)
• Output: Batch job details with progress tracking

YouTube Search Tool

• Name: youtube_search
• Purpose: Search processed videos and summaries
• Inputs: query (required), limit (default: 10)
• Output: Ranked search results with relevance scores

2. Agent Framework Support (agent_framework.py)

Framework-agnostic agent implementations:

from integrations.agent_framework import AgentFactory, FrameworkType

# Create framework-specific agents
langchain_agent = AgentFactory.create_agent(FrameworkType.LANGCHAIN, llm=your_llm)
crewai_agent = AgentFactory.create_agent(FrameworkType.CREWAI, role="YouTube Specialist")
autogen_agent = AgentFactory.create_agent(FrameworkType.AUTOGEN, name="YouTubeProcessor")

# Process videos with any framework
result = await langchain_agent.process_video("https://youtube.com/watch?v=xyz", "summarize")

Supported Frameworks

LangChain Integration

• Full ReAct agent support with custom tools
• Memory management and conversation tracking
• Async execution with proper error handling
• Tool chaining and complex workflows

CrewAI Integration

• Role-based agent creation with specialized capabilities
• Task delegation and crew coordination
• Multi-agent collaboration for complex video processing
• Structured task execution with expected outputs

AutoGen Integration

• Conversational agent interaction patterns
• Group chat support for batch processing
• Multi-turn conversations for iterative refinement
• Integration with AutoGen's proxy patterns

3. Agent Orchestrator

Unified interface for managing multiple frameworks:

from integrations.agent_framework import AgentOrchestrator

orchestrator = AgentOrchestrator()

# Register agents
orchestrator.register_agent(FrameworkType.LANGCHAIN, langchain_agent)
orchestrator.register_agent(FrameworkType.CREWAI, crewai_agent)

# Process with specific framework
result = await orchestrator.process_video(
    "https://youtube.com/watch?v=abc", 
    framework=FrameworkType.LANGCHAIN
)

# Compare frameworks
comparison = await orchestrator.compare_frameworks("https://youtube.com/watch?v=abc")

📋 Usage Examples

Basic Video Processing

import asyncio
from integrations.agent_framework import quick_process_video

async def basic_example():
    # Quick video summarization
    result = await quick_process_video(
        "https://youtube.com/watch?v=dQw4w9WgXcQ",
        task_type="summarize",
        framework="langchain"
    )
    print(f"Summary: {result}")

asyncio.run(basic_example())

Advanced Multi-Framework Processing

import asyncio
from integrations.agent_framework import create_youtube_agent_orchestrator

async def advanced_example():
    # Create orchestrator with all available frameworks
    orchestrator = create_youtube_agent_orchestrator()
    
    # Process video with default framework
    result = await orchestrator.process_video(
        "https://youtube.com/watch?v=educational_video",
        task_type="summarize",
        summary_type="comprehensive"
    )
    
    # Batch process multiple videos
    video_urls = [
        "https://youtube.com/watch?v=video1",
        "https://youtube.com/watch?v=video2",
        "https://youtube.com/watch?v=video3"
    ]
    
    batch_result = await orchestrator.process_batch(
        video_urls,
        task_type="transcribe",
        source="whisper"
    )
    
    # Compare different frameworks
    comparison = await orchestrator.compare_frameworks(
        "https://youtube.com/watch?v=test_video"
    )
    
    return result, batch_result, comparison

asyncio.run(advanced_example())

Custom LangChain Agent

from langchain.llms import OpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from integrations.langchain_tools import get_youtube_langchain_tools

# Create custom LangChain agent
llm = OpenAI(temperature=0)
tools = get_youtube_langchain_tools()
memory = ConversationBufferMemory(memory_key="chat_history")

# Custom prompt
prompt_template = """
You are a YouTube video analysis expert with access to advanced processing tools.

Your tools:
{tools}

Use these tools to help users with:
- Extracting accurate transcripts from YouTube videos
- Creating comprehensive summaries with key insights
- Processing multiple videos efficiently in batches
- Searching through previously processed content

Always provide detailed, well-structured responses with actionable insights.

{agent_scratchpad}
"""

agent = create_react_agent(llm=llm, tools=tools, prompt=prompt_template)
agent_executor = AgentExecutor(
    agent=agent, 
    tools=tools, 
    memory=memory,
    verbose=True,
    max_iterations=5
)

# Use the agent
async def use_custom_agent():
    result = await agent_executor.ainvoke({
        "input": "Please analyze this educational video and provide a comprehensive summary with key takeaways: https://youtube.com/watch?v=educational_content"
    })
    return result

asyncio.run(use_custom_agent())

CrewAI Crew Setup

from crewai import Agent, Task, Crew
from integrations.agent_framework import CrewAIYouTubeAgent

# Create specialized agents
transcript_specialist = CrewAIYouTubeAgent(
    role="Transcript Extraction Specialist",
    goal="Extract accurate and comprehensive transcripts from YouTube videos",
    backstory="Expert in audio processing and transcript quality analysis"
)

summary_specialist = CrewAIYouTubeAgent(
    role="Content Summarization Specialist", 
    goal="Create insightful and actionable video summaries",
    backstory="Experienced content analyst with expertise in educational material synthesis"
)

# Create tasks
extract_task = Task(
    description="Extract high-quality transcript from the provided YouTube video",
    agent=transcript_specialist,
    expected_output="Clean, accurate transcript with quality metrics"
)

summarize_task = Task(
    description="Create a comprehensive summary based on the extracted transcript",
    agent=summary_specialist,
    expected_output="Structured summary with key points and actionable insights"
)

# Create and run crew
crew = Crew(
    agents=[transcript_specialist, summary_specialist],
    tasks=[extract_task, summarize_task],
    verbose=True
)

result = crew.kickoff()

🔧 Configuration

Environment Variables

# Backend API Configuration
ANTHROPIC_API_KEY=sk-ant-...           # For AI summarization
OPENAI_API_KEY=sk-...                  # For Whisper transcription
DATABASE_URL=sqlite:///./data/app.db   # Database connection

# Framework-specific configuration
LANGCHAIN_VERBOSE=true                 # Enable LangChain debugging
CREWAI_LOG_LEVEL=info                 # CrewAI logging level
AUTOGEN_TIMEOUT=60                    # AutoGen conversation timeout

Agent Capabilities

from integrations.agent_framework import AgentCapabilities, AgentContext

# Configure agent capabilities
capabilities = AgentCapabilities(
    can_extract_transcripts=True,
    can_summarize_videos=True, 
    can_batch_process=True,
    can_search_content=True,
    requires_async=True,
    max_concurrent_videos=3,
    supported_video_length_minutes=120
)

# Set agent context
context = AgentContext(
    user_id="user_123",
    session_id="session_456",
    preferences={
        "summary_type": "comprehensive",
        "transcript_source": "whisper",
        "output_format": "structured"
    },
    rate_limits={"videos_per_hour": 10, "batch_size": 5},
    cost_budget=5.00
)

# Apply to agent
agent.capabilities = capabilities
agent.set_context(context)

🧪 Testing

Run Example Integration

cd backend/integrations
python example_integration.py

This will demonstrate:

• LangChain tools functionality
• Agent factory capabilities
• Orchestrator features
• Framework comparisons
• Advanced parameter handling

Unit Testing

# Test LangChain tools
python -m pytest tests/test_langchain_tools.py -v

# Test agent framework
python -m pytest tests/test_agent_framework.py -v

# Test integration examples
python -m pytest tests/test_integrations.py -v

🚨 Error Handling

The integration modules include comprehensive error handling:

Graceful Framework Fallbacks

# Frameworks are imported with try/catch
try:
    from langchain.tools import BaseTool
    LANGCHAIN_AVAILABLE = True
except ImportError:
    LANGCHAIN_AVAILABLE = False
    # Mock implementations provided

# Check availability before use
if LANGCHAIN_AVAILABLE:
    # Use real LangChain functionality
else:
    # Fall back to mock implementations

Service Availability Checks

# Backend services checked at runtime
try:
    from ..services.dual_transcript_service import DualTranscriptService
    transcript_service = DualTranscriptService()
    SERVICES_AVAILABLE = True
except ImportError:
    # Use mock services for development/testing
    SERVICES_AVAILABLE = False

Comprehensive Error Responses

# All methods return structured error information
{
    "success": False,
    "error": "Detailed error message",
    "error_code": "SERVICE_UNAVAILABLE",
    "retry_after": 30,
    "suggestions": ["Check API keys", "Verify service status"]
}

🔌 Extension Points

Adding New Frameworks

1. Inherit from BaseYouTubeAgent:

class MyFrameworkAgent(BaseYouTubeAgent):
    def __init__(self):
        super().__init__(FrameworkType.CUSTOM)
    
    async def process_video(self, video_url, task_type, **kwargs):
        # Implementation
        pass

2. Register in AgentFactory:

# Update AgentFactory.create_agent() method
elif framework == FrameworkType.MY_FRAMEWORK:
    return MyFrameworkAgent(**kwargs)

3. Add to FrameworkType enum:

class FrameworkType(Enum):
    LANGCHAIN = "langchain"
    CREWAI = "crewai" 
    AUTOGEN = "autogen"
    MY_FRAMEWORK = "my_framework"  # Add here

Custom Tool Development

from integrations.langchain_tools import BaseTool

class CustomYouTubeTool(BaseTool):
    name = "custom_youtube_tool"
    description = "Custom tool for specialized YouTube processing"
    
    async def _arun(self, video_url: str, **kwargs) -> str:
        # Implementation
        return json.dumps({"result": "custom processing"})

📚 Additional Resources

• Backend Services: See backend/services/ for core YouTube processing
• API Documentation: OpenAPI spec at /docs when running the server
• MCP Server: See backend/mcp_server.py for Model Context Protocol integration
• Frontend Integration: React components in frontend/src/components/

🤝 Contributing

1. Follow the existing code patterns and error handling
2. Add comprehensive docstrings and type hints
3. Include both real and mock implementations
4. Add tests for new functionality
5. Update this README with new features

📄 License

This integration module is part of the YouTube Summarizer project and follows the same licensing terms.