Build a LangChain Agent That Summarizes YouTube Videos

pip install langchain langchain-community langchain-openai
pip install youtube-transcript-api pytube
pip install openai-whisper  # For Whisper fallback
pip install yt-dlp           # Better than pytube for audio download

pip install fastapi uvicorn python-dotenv

from langchain_community.document_loaders import YoutubeLoader

def load_youtube_transcript(
    url: str,
    language: str = "en",
    translation: str = None
) -> list:
    """
    Load transcript from a YouTube URL.
    
    Args:
        url: Full YouTube URL or video ID
        language: Transcript language code (e.g., 'en', 'es', 'fr')
        translation: Translate to this language if original not available
    
    Returns:
        List of Document objects with transcript text
    """
    loader = YoutubeLoader.from_youtube_url(
        url,
        add_video_info=True,     # Includes title, author, length in metadata
        language=[language],
        translation=translation
    )
    
    documents = loader.load()
    
    if not documents:
        raise ValueError(f"No transcript found for: {url}")
    
    print(f"Loaded transcript: {len(documents[0].page_content)} characters")
    print(f"Video: {documents[0].metadata.get('title', 'Unknown')}")
    print(f"Author: {documents[0].metadata.get('author', 'Unknown')}")
    print(f"Length: {documents[0].metadata.get('length', 0)} seconds")
    
    return documents


# Try it
try:
    docs = load_youtube_transcript(
        "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
        language="en"
    )
    print("Transcript loaded successfully")
except Exception as e:
    print(f"Transcript loading failed: {e}")

import os
import tempfile
import whisper
import yt_dlp


def download_audio(url: str, output_dir: str = None) -> str:
    """
    Download audio from a YouTube URL using yt-dlp.
    Returns the path to the downloaded audio file.
    """
    if output_dir is None:
        output_dir = tempfile.mkdtemp()
    
    output_template = os.path.join(output_dir, "%(id)s.%(ext)s")
    
    ydl_opts = {
        "format": "bestaudio/best",
        "outtmpl": output_template,
        "postprocessors": [{
            "key": "FFmpegExtractAudio",
            "preferredcodec": "mp3",
            "preferredquality": "192",
        }],
        "quiet": True,
        "no_warnings": True,
    }
    
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=True)
        video_id = info.get("id", "audio")
        audio_path = os.path.join(output_dir, f"{video_id}.mp3")
    
    print(f"Audio downloaded: {audio_path}")
    return audio_path


def transcribe_with_whisper(
    audio_path: str,
    model_size: str = "base",
    language: str = None
) -> str:
    """
    Transcribe audio file using OpenAI Whisper.
    
    model_size options: tiny, base, small, medium, large
    Larger models are more accurate but slower and use more memory.
    """
    print(f"Loading Whisper model: {model_size}")
    model = whisper.load_model(model_size)
    
    print("Transcribing audio...")
    transcribe_opts = {"verbose": False}
    if language:
        transcribe_opts["language"] = language
    
    result = model.transcribe(audio_path, **transcribe_opts)
    
    transcript = result["text"]
    detected_language = result.get("language", "unknown")
    
    print(f"Transcription complete: {len(transcript)} characters")
    print(f"Detected language: {detected_language}")
    
    return transcript


def load_transcript_with_fallback(
    url: str,
    language: str = "en",
    whisper_model: str = "base",
    cleanup_audio: bool = True
) -> tuple:
    """
    Load transcript via YoutubeLoader with Whisper fallback.
    
    Returns: (transcript_text, metadata_dict, source)
    source is either 'youtube_captions' or 'whisper'
    """
    # Try YouTube captions first
    try:
        loader = YoutubeLoader.from_youtube_url(
            url,
            add_video_info=True,
            language=[language]
        )
        docs = loader.load()
        
        if docs and docs[0].page_content.strip():
            return (
                docs[0].page_content,
                docs[0].metadata,
                "youtube_captions"
            )
        
        raise ValueError("Empty transcript returned")
        
    except Exception as e:
        print(f"YouTube captions unavailable ({e}). Falling back to Whisper...")
    
    # Whisper fallback
    audio_path = None
    try:
        audio_path = download_audio(url)
        transcript_text = transcribe_with_whisper(
            audio_path,
            model_size=whisper_model,
            language=language if language != "en" else None
        )
        
        # Extract basic metadata via yt-dlp (no download)
        with yt_dlp.YoutubeDL({"quiet": True}) as ydl:
            info = ydl.extract_info(url, download=False)
            metadata = {
                "title": info.get("title", "Unknown"),
                "author": info.get("uploader", "Unknown"),
                "length": info.get("duration", 0),
                "source": url
            }
        
        return transcript_text, metadata, "whisper"
        
    finally:
        if cleanup_audio and audio_path and os.path.exists(audio_path):
            os.remove(audio_path)
            print("Audio file cleaned up")

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document


def chunk_transcript(
    transcript_text: str,
    metadata: dict,
    chunk_size: int = 4000,
    chunk_overlap: int = 200
) -> list:
    """
    Split a transcript into overlapping chunks suitable for map-reduce.
    
    chunk_size: characters per chunk (4000 chars ≈ 1000 tokens)
    chunk_overlap: characters of overlap between adjacent chunks
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=[". ", "? ", "! ", "\n", " ", ""]
    )
    
    texts = splitter.split_text(transcript_text)
    
    # Wrap in Document objects with metadata
    chunks = [
        Document(
            page_content=text,
            metadata={
                **metadata,
                "chunk_index": i,
                "total_chunks": len(texts)
            }
        )
        for i, text in enumerate(texts)
    ]
    
    print(f"Split transcript into {len(chunks)} chunks")
    print(f"Avg chunk size: {sum(len(c.page_content) for c in chunks) // len(chunks)} chars")
    
    return chunks

from langchain.chains.summarize import load_summarize_chain
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate


def build_summarization_chain(
    model: str = "gpt-4o-mini",
    temperature: float = 0,
    chain_type: str = "map_reduce",
    output_format: str = "standard"
):
    """
    Build a summarization chain for YouTube transcripts.
    
    chain_type: 'map_reduce' (long videos), 'stuff' (short videos), 'refine'
    output_format: 'standard', 'bullets', 'structured'
    """
    llm = ChatOpenAI(model=model, temperature=temperature)
    
    if chain_type == "stuff":
        # For short videos where the full transcript fits in context
        prompt = PromptTemplate(
            template="""You are summarizing a YouTube video transcript.

Transcript:
{text}

Write a clear, informative summary that covers:
1. The main topic and purpose of the video
2. Key points and takeaways
3. Any specific techniques, tools, or conclusions mentioned

Summary:""",
            input_variables=["text"]
        )
        return load_summarize_chain(
            llm,
            chain_type="stuff",
            prompt=prompt
        )
    
    # Map-reduce for long videos
    map_prompt = PromptTemplate(
        template="""Summarize this section of a YouTube video transcript.
Be concise but capture all important points, facts, and conclusions.

Transcript section:
{text}

Section summary:""",
        input_variables=["text"]
    )
    
    if output_format == "bullets":
        combine_prompt = PromptTemplate(
            template="""You are combining section summaries from a YouTube video into a final summary.

Section summaries:
{text}

Write a final summary as bullet points grouped by topic.
Format: Start each group with a bold topic heading, followed by 2-4 bullet points.

Final summary:""",
            input_variables=["text"]
        )
    
    elif output_format == "structured":
        combine_prompt = PromptTemplate(
            template="""You are combining section summaries from a YouTube video into a structured summary.

Section summaries:
{text}

Write a final summary with these sections:
**Overview**: 2-3 sentences on what the video is about
**Key Points**: numbered list of the 5-7 most important takeaways
**Conclusion**: what the video recommends or concludes

Final summary:""",
            input_variables=["text"]
        )
    
    else:
        combine_prompt = PromptTemplate(
            template="""You are combining section summaries from a YouTube video into a final summary.

Section summaries:
{text}

Write a cohesive, well-structured summary covering the video's main points.
Aim for 3-5 paragraphs. Be specific about techniques, tools, or data mentioned.

Final summary:""",
            input_variables=["text"]
        )
    
    return load_summarize_chain(
        llm,
        chain_type="map_reduce",
        map_prompt=map_prompt,
        combine_prompt=combine_prompt,
        verbose=False
    )


def summarize_transcript(
    chunks: list,
    model: str = "gpt-4o-mini",
    chain_type: str = "map_reduce",
    output_format: str = "structured"
) -> str:
    """
    Summarize transcript chunks using the specified chain type.
    """
    # For very short transcripts, use 'stuff' directly
    total_chars = sum(len(c.page_content) for c in chunks)
    if total_chars < 6000 and chain_type == "map_reduce":
        print("Short transcript detected, using 'stuff' chain")
        chain_type = "stuff"
    
    chain = build_summarization_chain(
        model=model,
        chain_type=chain_type,
        output_format=output_format
    )
    
    print(f"Summarizing {len(chunks)} chunks with {chain_type} chain...")
    result = chain.invoke({"input_documents": chunks})
    
    summary = result.get("output_text", result.get("text", ""))
    print(f"Summary generated: {len(summary)} characters")
    
    return summary

from typing import Optional
import time


def summarize_youtube_video(
    url: str,
    language: str = "en",
    whisper_model: str = "base",
    summarization_model: str = "gpt-4o-mini",
    combine_model: str = "gpt-4o",
    output_format: str = "structured",
    chunk_size: int = 4000,
    chunk_overlap: int = 200
) -> dict:
    """
    Full pipeline: URL → transcript → chunks → summary.
    
    Returns dict with summary, metadata, timing, and source info.
    """
    start_time = time.time()
    
    # Step 1: Load transcript
    print(f"\n{'='*50}")
    print(f"Processing: {url}")
    print(f"{'='*50}")
    
    transcript_text, metadata, source = load_transcript_with_fallback(
        url,
        language=language,
        whisper_model=whisper_model
    )
    
    load_time = time.time() - start_time
    print(f"Transcript loaded in {load_time:.1f}s (source: {source})")
    
    # Step 2: Chunk transcript
    chunks = chunk_transcript(
        transcript_text,
        metadata,
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    
    # Step 3: Determine chain type
    total_chars = sum(len(c.page_content) for c in chunks)
    chain_type = "stuff" if total_chars < 6000 else "map_reduce"
    
    # Step 4: Summarize
    summary = summarize_transcript(
        chunks,
        model=summarization_model,
        chain_type=chain_type,
        output_format=output_format
    )
    
    total_time = time.time() - start_time
    
    return {
        "url": url,
        "title": metadata.get("title", "Unknown"),
        "author": metadata.get("author", "Unknown"),
        "duration_seconds": metadata.get("length", 0),
        "transcript_source": source,
        "num_chunks": len(chunks),
        "chain_type": chain_type,
        "summary": summary,
        "processing_time_seconds": round(total_time, 1)
    }


# Test run
if __name__ == "__main__":
    result = summarize_youtube_video(
        "https://www.youtube.com/watch?v=YOUR_VIDEO_ID",
        output_format="structured"
    )
    
    print(f"\nVideo: {result['title']}")
    print(f"Author: {result['author']}")
    print(f"Duration: {result['duration_seconds'] // 60} minutes")
    print(f"Processed in: {result['processing_time_seconds']}s")
    print(f"\n{'='*50}")
    print("SUMMARY:")
    print('='*50)
    print(result['summary'])

from langchain.tools import tool
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from typing import Optional
import re


@tool
def summarize_video(url: str, output_format: str = "structured") -> str:
    """
    Summarize a YouTube video given its URL.
    
    Use this tool when the user provides a YouTube URL and asks for a summary,
    key points, or overview of the video content.
    
    Args:
        url: Full YouTube URL (e.g., https://www.youtube.com/watch?v=...)
        output_format: 'standard', 'bullets', or 'structured' (default: structured)
    
    Returns:
        Structured summary of the video content
    """
    try:
        result = summarize_youtube_video(
            url=url,
            output_format=output_format
        )
        
        output = f"""**Video**: {result['title']}
**Author**: {result['author']}
**Duration**: {result['duration_seconds'] // 60} minutes
**Transcript source**: {result['transcript_source']}

{result['summary']}

*Processed in {result['processing_time_seconds']}s using {result['chain_type']} summarization*"""
        
        return output
        
    except Exception as e:
        return f"Failed to summarize video: {str(e)}"


@tool
def extract_video_id(url: str) -> str:
    """
    Extract the video ID from a YouTube URL.
    
    Use this when the user provides a YouTube URL and you need the video ID.
    """
    patterns = [
        r"youtube\.com/watch\?v=([a-zA-Z0-9_-]{11})",
        r"youtu\.be/([a-zA-Z0-9_-]{11})",
        r"youtube\.com/embed/([a-zA-Z0-9_-]{11})",
        r"youtube\.com/shorts/([a-zA-Z0-9_-]{11})"
    ]
    
    for pattern in patterns:
        match = re.search(pattern, url)
        if match:
            return f"Video ID: {match.group(1)}"
    
    return "Could not extract video ID from URL"


def build_youtube_agent(
    model: str = "gpt-4o",
    enable_memory: bool = True
) -> AgentExecutor:
    """
    Build a YouTube summarization agent with tool use.
    """
    llm = ChatOpenAI(model=model, temperature=0)
    
    tools = [summarize_video, extract_video_id]
    
    system_message = """You are a helpful assistant that specializes in summarizing 
YouTube videos. When given a YouTube URL, you use the summarize_video tool to 
generate a comprehensive summary. 

You can also:
- Explain specific parts of a video when asked
- Compare key points from multiple videos
- Extract the video ID from a URL
- Generate summaries in different formats (standard, bullets, structured)

Always use the summarize_video tool when the user provides a YouTube URL.
Be helpful and informative in your responses."""

    if enable_memory:
        prompt = ChatPromptTemplate.from_messages([
            ("system", system_message),
            MessagesPlaceholder("chat_history", optional=True),
            ("human", "{input}"),
            MessagesPlaceholder("agent_scratchpad")
        ])
        
        memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
    else:
        prompt = ChatPromptTemplate.from_messages([
            ("system", system_message),
            ("human", "{input}"),
            MessagesPlaceholder("agent_scratchpad")
        ])
        memory = None
    
    agent = create_openai_tools_agent(llm, tools, prompt)
    
    return AgentExecutor(
        agent=agent,
        tools=tools,
        memory=memory,
        verbose=True,
        max_iterations=3,
        handle_parsing_errors=True
    )


# Run the agent
if __name__ == "__main__":
    agent = build_youtube_agent(model="gpt-4o")
    
    # Single video summary
    response = agent.invoke({
        "input": "Please summarize this video: https://www.youtube.com/watch?v=YOUR_VIDEO_ID"
    })
    print(response["output"])
    
    # Follow-up question (uses memory)
    response2 = agent.invoke({
        "input": "Can you give me just the bullet points from that video?"
    })
    print(response2["output"])

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains.summarize import load_summarize_chain
from langchain.schema import Document


def hierarchical_summarize(
    chunks: list,
    map_model: str = "gpt-4o-mini",
    combine_model: str = "gpt-4o",
    intermediate_batch_size: int = 5
) -> str:
    """
    Hierarchical map-reduce for very long videos (3+ hours).
    
    Stage 1: Summarize chunks in batches of intermediate_batch_size
    Stage 2: Combine batch summaries into section summaries
    Stage 3: Combine section summaries into final summary
    """
    fast_llm = ChatOpenAI(model=map_model, temperature=0)
    smart_llm = ChatOpenAI(model=combine_model, temperature=0)
    
    map_prompt = PromptTemplate(
        template="Summarize this transcript segment in 3-5 sentences:\n\n{text}\n\nSummary:",
        input_variables=["text"]
    )
    
    batch_combine_prompt = PromptTemplate(
        template="""Combine these segment summaries into a coherent section summary (5-8 sentences):

{text}

Section summary:""",
        input_variables=["text"]
    )
    
    final_combine_prompt = PromptTemplate(
        template="""You are writing the final summary of a YouTube video from section summaries.

Section summaries:
{text}

Write a comprehensive, well-structured summary with:
**Overview**: What this video covers and why it matters
**Key Points**: The 5-7 most important takeaways
**Tools/Techniques Mentioned**: Specific tools, frameworks, or methods discussed
**Conclusion**: Main recommendation or conclusion

Final summary:""",
        input_variables=["text"]
    )
    
    print(f"Stage 1: Summarizing {len(chunks)} chunks in batches of {intermediate_batch_size}...")
    
    # Stage 1: Map each chunk
    chunk_summaries = []
    for chunk in chunks:
        chain = load_summarize_chain(fast_llm, chain_type="stuff", prompt=map_prompt)
        result = chain.invoke({"input_documents": [chunk]})
        chunk_summaries.append(result["output_text"])
    
    # Stage 2: Combine chunks into batches
    print(f"Stage 2: Combining into {len(chunk_summaries) // intermediate_batch_size + 1} sections...")
    
    section_summaries = []
    for i in range(0, len(chunk_summaries), intermediate_batch_size):
        batch = chunk_summaries[i:i + intermediate_batch_size]
        batch_text = "\n\n".join(batch)
        batch_doc = Document(page_content=batch_text)
        
        chain = load_summarize_chain(fast_llm, chain_type="stuff", prompt=batch_combine_prompt)
        result = chain.invoke({"input_documents": [batch_doc]})
        section_summaries.append(result["output_text"])
    
    # Stage 3: Final combine
    print("Stage 3: Generating final summary...")
    final_text = "\n\n".join(section_summaries)
    final_doc = Document(page_content=final_text)
    
    chain = load_summarize_chain(smart_llm, chain_type="stuff", prompt=final_combine_prompt)
    result = chain.invoke({"input_documents": [final_doc]})
    
    return result["output_text"]

import os
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, HttpUrl
from typing import Optional, Literal
import uvicorn
from datetime import datetime
import uuid


app = FastAPI(
    title="YouTube Summarizer API",
    description="Summarize YouTube videos using LangChain and GPT-4o",
    version="1.0.0"
)

# In-memory job store (use Redis in production)
jobs = {}


class SummarizeRequest(BaseModel):
    url: str
    language: str = "en"
    output_format: Literal["standard", "bullets", "structured"] = "structured"
    whisper_model: str = "base"


class SummarizeResponse(BaseModel):
    job_id: str
    status: str
    created_at: str


class JobResult(BaseModel):
    job_id: str
    status: str
    result: Optional[dict] = None
    error: Optional[str] = None


def run_summarization_job(job_id: str, request: SummarizeRequest):
    """Background task that runs the summarization pipeline."""
    try:
        jobs[job_id]["status"] = "processing"
        
        result = summarize_youtube_video(
            url=request.url,
            language=request.language,
            output_format=request.output_format,
            whisper_model=request.whisper_model
        )
        
        jobs[job_id]["status"] = "completed"
        jobs[job_id]["result"] = result
        
    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = str(e)


@app.post("/summarize", response_model=SummarizeResponse)
async def create_summary_job(
    request: SummarizeRequest,
    background_tasks: BackgroundTasks
):
    """
    Submit a YouTube video for summarization.
    Returns a job ID to poll for results.
    """
    job_id = str(uuid.uuid4())[:8]
    
    jobs[job_id] = {
        "status": "queued",
        "created_at": datetime.utcnow().isoformat(),
        "result": None,
        "error": None
    }
    
    background_tasks.add_task(run_summarization_job, job_id, request)
    
    return SummarizeResponse(
        job_id=job_id,
        status="queued",
        created_at=jobs[job_id]["created_at"]
    )


@app.get("/jobs/{job_id}", response_model=JobResult)
async def get_job_status(job_id: str):
    """
    Get the status and result of a summarization job.
    Poll this endpoint until status is 'completed' or 'failed'.
    """
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
    
    job = jobs[job_id]
    
    return JobResult(
        job_id=job_id,
        status=job["status"],
        result=job.get("result"),
        error=job.get("error")
    )


@app.post("/summarize/sync")
async def summarize_sync(request: SummarizeRequest):
    """
    Synchronous summarization endpoint.
    Blocks until complete — use for videos under 30 minutes.
    """
    try:
        result = summarize_youtube_video(
            url=request.url,
            language=request.language,
            output_format=request.output_format,
            whisper_model=request.whisper_model
        )
        return {"status": "success", "data": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health_check():
    return {"status": "ok", "active_jobs": len(jobs)}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

uvicorn app:app --reload --port 8000

# Submit a job
curl -X POST http://localhost:8000/summarize \
  -H "Content-Type: application/json" \
  -d '{"url": "https://www.youtube.com/watch?v=YOUR_ID", "output_format": "structured"}'

# Poll for result
curl http://localhost:8000/jobs/YOUR_JOB_ID

# Cost comparison for a 1-hour video (~8,000 words, 8 chunks)
# gpt-4o map + combine:      ~$0.08 per video
# gpt-4o-mini map + gpt-4o combine: ~$0.02 per video
# gpt-4o-mini both:          ~$0.005 per video

# For high volume (1000 videos/day):
# All gpt-4o:     $80/day
# Mini map + 4o combine: $20/day
# All mini:       $5/day

import hashlib
import json
from pathlib import Path

TRANSCRIPT_CACHE_DIR = Path("./transcript_cache")
TRANSCRIPT_CACHE_DIR.mkdir(exist_ok=True)

def get_cached_transcript(url: str) -> Optional[tuple]:
    """Return cached transcript if available."""
    url_hash = hashlib.md5(url.encode()).hexdigest()[:12]
    cache_path = TRANSCRIPT_CACHE_DIR / f"{url_hash}.json"
    
    if cache_path.exists():
        with open(cache_path) as f:
            data = json.load(f)
        print(f"Transcript cache hit: {url_hash}")
        return data["text"], data["metadata"], data["source"]
    
    return None

def cache_transcript(url: str, text: str, metadata: dict, source: str):
    """Save transcript to cache."""
    url_hash = hashlib.md5(url.encode()).hexdigest()[:12]
    cache_path = TRANSCRIPT_CACHE_DIR / f"{url_hash}.json"
    
    with open(cache_path, "w") as f:
        json.dump({"text": text, "metadata": metadata, "source": source}, f)
    
    print(f"Transcript cached: {url_hash}")

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI


def build_video_qa_system(chunks: list) -> RetrievalQA:
    """
    Build a Q&A system over a video's transcript.
    Lets users ask specific questions about video content.
    """
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = FAISS.from_documents(chunks, embeddings)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
    
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    
    chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )
    
    return chain


# Use it
chunks = chunk_transcript(transcript_text, metadata)
qa_chain = build_video_qa_system(chunks)

# Ask specific questions
response = qa_chain.invoke({"query": "What library did the presenter use for authentication?"})
print("Answer:", response["result"])

if metadata.get("length", 0) < 60:
    return {"error": "Video too short to summarize (under 1 minute)"}

import time
from typing import Callable, Any

def retry_with_backoff(
    func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0
) -> Any:
    """Retry a function with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
            time.sleep(delay)