Deep Research Agent Search Pattern

import requests, os

# Three-service pattern: Serper -> Jina -> E2B

def deep_research_three_service(question: str) -> dict:
    """Search-Read-Compute pipeline using three APIs."""

    # Step 1: Search (Serper) - find relevant URLs
    search_resp = requests.post(
        "https://google.serper.dev/search",
        headers={"X-API-KEY": os.environ["SERPER_API_KEY"]},
        json={"q": question, "num": 5},
        timeout=10,
    )
    urls = [r["link"] for r in search_resp.json().get("organic", [])[:5]]

    # Step 2: Read (Jina) - extract content from each URL
    pages = []
    for url in urls:
        read_resp = requests.get(
            f"https://r.jina.ai/{url}",
            headers={"Accept": "application/json"},
            timeout=15,
        )
        if read_resp.status_code == 200:
            pages.append({
                "url": url,
                "content": read_resp.json().get("content", "")[:3000],
            })

    # Step 3: Compute (E2B or local) - analyze extracted data
    # This step typically sends pages to an LLM for synthesis
    # or runs code in a sandbox for data analysis

    return {
        "question": question,
        "sources": len(pages),
        "pages": pages,
    }

# Cost per research query:
# Serper: $0.001 + Jina: ~$0 (free tier) + E2B: ~$0.01 = ~$0.011/query
# But: 3 API keys, 3 failure points, 3 rate limits to manage

import requests, os

def deep_research_single_api(question: str) -> dict:
    """Research pipeline using one API for structured search."""

    resp = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": os.environ["SCAVIO_API_KEY"]},
        json={"query": question, "platform": "google", "country_code": "us"},
        timeout=15,
    )
    data = resp.json()

    research = {
        "question": question,
        "direct_answer": None,
        "sources": [],
        "related_questions": [],
    }

    # AI Overview often contains the synthesized answer already
    ao = data.get("ai_overview")
    if ao:
        research["direct_answer"] = ao.get("text", "")
        research["sources"].extend([
            {"url": s.get("link", ""), "title": s.get("title", "")}
            for s in ao.get("sources", [])
        ])

    # Organic results provide supporting evidence
    for r in data.get("organic_results", [])[:5]:
        research["sources"].append({
            "url": r.get("link", ""),
            "title": r.get("title", ""),
            "snippet": r.get("snippet", ""),
        })

    # PAA gives related angles to explore
    for paa in data.get("people_also_ask", []):
        research["related_questions"].append(paa.get("question", ""))

    return research

# Cost: $0.005/query (1 Scavio credit)
# 1 API key, 1 failure point, 1 rate limit
# Tradeoff: no full page content, only SERP-extracted data

decision_matrix = {
    "single_api_wins": [
        "Quick fact-checking (answer is in SERP snippets)",
        "Market research (pricing, feature comparisons from snippets)",
        "Trend monitoring (SERP feature changes over time)",
        "Multi-platform research (Google + Reddit + YouTube in one call)",
        "Agent workflows where latency matters (1 call vs 6+)",
    ],
    "three_service_wins": [
        "Deep technical research (need full documentation pages)",
        "Legal/regulatory research (need complete document text)",
        "Academic research (need full paper content)",
        "Content extraction for RAG (building a knowledge base)",
        "Competitive analysis requiring full page scraping",
    ],
}

# Hybrid: use single API for initial triage,
# then three-service for deep dives on promising leads

def hybrid_research(question: str, deep_dive_threshold: int = 3) -> dict:
    """Triage with single API, deep-dive on top results if needed."""

    # Quick triage: single API call
    triage = deep_research_single_api(question)

    # If we got a direct answer from AI Overview, done
    if triage["direct_answer"]:
        triage["method"] = "ai_overview"
        return triage

    # If snippets are sufficient (short factual queries), done
    if len(triage["sources"]) >= 3:
        snippet_text = " ".join(s.get("snippet", "") for s in triage["sources"])
        if len(snippet_text) > 500:
            triage["method"] = "snippets"
            return triage

    # Otherwise, deep-dive on top URLs using Jina reader
    for source in triage["sources"][:deep_dive_threshold]:
        try:
            read_resp = requests.get(
                f"https://r.jina.ai/{source['url']}",
                headers={"Accept": "application/json"},
                timeout=15,
            )
            if read_resp.status_code == 200:
                source["full_content"] = read_resp.json().get("content", "")[:3000]
        except Exception:
            source["full_content"] = None

    triage["method"] = "deep_dive"
    return triage

# Average cost: ~$0.01/query (most resolve at triage stage)
# vs. $0.011/query for always-deep-dive