exo.search

git clone https://github.com/Midsphere-AI/exo-ai.git && cd exo-ai
uv sync

uv sync --extra server       # FastAPI server
uv sync --extra embeddings   # Embedding-based reranking (Gemini, Vertex AI, OpenAI)
uv sync --extra pdf          # PDF content extraction

import exo.search

# High-level search (most common)
from exo.search import search, search_with_details, stream

# Configuration
from exo.search import SearchConfig, ResearchMode

# Streaming events
from exo.search import PipelineEvent, SearchResponse

# Multi-turn conversation
from exo.search import ConversationManager

exo.search
  __init__.py          Public API: search(), search_with_details(), stream()
  config.py            SearchConfig, context budget computation
  types.py             All Pydantic models and enums
  pipeline.py          Orchestration: run_search_pipeline(), stream_search_pipeline()
  conversation.py      ConversationManager, format_chat_history()
  focus_modes.py       Focus mode to source-type mapping
  server.py            FastAPI server with /search, /stream, /chat endpoints
  __main__.py          CLI: python -m exo.search
  agents/
    classifier.py      Query classification (skip/academic/social/sequential/complexity)
    researcher.py      Iterative tool-calling research loop
    query_planner.py   Adaptive hybrid research with sub-question decomposition
    deep_researcher.py Sequential multi-step research for complex queries (DAG-based)
    writer.py          Citation-aware answer generation with claim-first writing
    suggestion_generator.py  Follow-up suggestion generation
  tools/
    searxng.py         web_search(), academic_search(), social_search() via Serper/SearXNG
    web_fetcher.py     Page scraping, content enrichment, Jina Reader integration
    embeddings.py      Semantic reranking via cosine similarity (Gemini/Vertex/OpenAI)
    citation_verifier.py  Three-phase citation verification (keyword + LLM spot-check)
    contradiction_detector.py  Cross-source contradiction detection
    confidence.py      Heuristic confidence scoring (citation rate, authority, richness, coverage)
    researcher_tools.py  Agent tools: done() signal, reasoning_preamble()
    jina.py            Jina Reader/Search API integration
    serper.py          Serper (Google Search) API client
    compute.py         Shared computation utilities

import asyncio
from exo.search import search

async def main():
    answer = await search("What is quantum computing?")
    print(answer)

asyncio.run(main())

import asyncio
from exo.search import search_with_details

async def main():
    result = await search_with_details(
        "What are the latest advances in CRISPR gene editing?",
        mode="quality",
    )

    print(result.answer)

    print("\nSources:")
    for i, source in enumerate(result.sources, 1):
        print(f"  [{i}] {source.title} ({source.url})")

    print(f"\nConfidence: {result.confidence:.0%}")

    if result.verification:
        v = result.verification
        print(f"Citations: {v.verified}/{v.total_citations} verified, {v.removed} removed")

    if result.contradictions and result.contradictions.has_contradictions:
        print(f"Contradictions found: {len(result.contradictions.contradictions)}")

    print("\nSuggested follow-ups:")
    for s in result.suggestions:
        print(f"  - {s}")

asyncio.run(main())

import asyncio
from exo.search import stream, PipelineEvent, SearchResponse
from exo.types import TextEvent

async def main():
    async for event in stream("latest AI news", mode="balanced"):
        if isinstance(event, PipelineEvent):
            print(f"[{event.stage}] {event.status}")
        elif isinstance(event, TextEvent):
            print(event.text, end="", flush=True)
        elif isinstance(event, SearchResponse):
            print(f"\n\n{len(event.sources)} sources cited")

asyncio.run(main())

import asyncio
from exo.search import search_with_details, ConversationManager

async def main():
    conv = ConversationManager(max_turns=10)

    # First query
    result = await search_with_details(
        "What is transformer architecture?",
        chat_history=conv.turns,
    )
    conv.add_turn("What is transformer architecture?", result.answer)
    print(result.answer[:200])

    # Follow-up (classifier resolves "it" from history)
    result = await search_with_details(
        "How does it compare to RNNs?",
        chat_history=conv.turns,
    )
    conv.add_turn("How does it compare to RNNs?", result.answer)
    print(result.answer[:200])

asyncio.run(main())

from exo.search import search, SearchConfig

config = SearchConfig(
    model="anthropic:claude-sonnet-4",
    fast_model="openai:gpt-4o-mini",
    max_results=15,
    max_iterations=10,
    max_writer_sources=20,
    llm_verification=True,
    max_revision_rounds=3,
    serper_api_key="sk-...",       # Google Search via Serper
    jina_api_key="jina_...",       # Full-page content extraction
)

answer = await search("complex multi-step question", mode="quality", config=config)

# Single query
python -m exo.search "What is quantum computing?"

# Quality mode
python -m exo.search --quality quality "CRISPR gene editing"

# Streaming output
python -m exo.search --stream "latest AI news"

# Interactive chat
python -m exo.search --chat

# Start FastAPI server
python -m exo.search --serve --port 8000

# With verbose logging
python -m exo.search -v "test query"

class SearchResponse(BaseModel):
    answer: str                                    # Cited answer text with [N] markers
    sources: list[Source]                           # Cited sources (title, url, content)
    suggestions: list[str]                          # Follow-up question suggestions
    query: str                                     # Effective query (after rewriting)
    mode: str                                      # Research mode used
    verification: CitationVerification | None      # Citation verification stats
    contradictions: ContradictionReport | None     # Cross-source contradictions
    confidence: float | None                       # 0-1 confidence score
    confidence_breakdown: dict | None              # Component-level breakdown

class PipelineEvent(BaseModel):
    type: str = "pipeline"
    stage: str     # "classifier", "researcher", "deep_research", "enrichment",
                   # "writer", "revision", "suggestions"
    status: str    # "started", "completed"
    message: str   # Optional detail message

class CitationVerification(BaseModel):
    total_citations: int     # Total [N] markers found
    verified: int            # Citations confirmed by source content
    removed: int             # Citations removed as unsupported
    flagged: int             # Citations flagged but kept
    llm_verified: int        # Citations verified via LLM spot-check
    revision_count: int      # Number of revision rounds performed

class Classification(BaseModel):
    skip_search: bool                    # True for greetings, math, etc.
    personal_search: bool                # User-specific queries
    academic_search: bool                # Scholarly content needed
    discussion_search: bool              # Social/forum content needed
    show_weather_widget: bool            # Weather intent detected
    show_stock_widget: bool              # Stock/finance intent detected
    show_calculation_widget: bool        # Calculation intent detected
    requires_sequential_research: bool   # Multi-step research needed
    estimated_complexity: str            # "simple", "moderate", "complex"

class ResearchMode(StrEnum):
    SPEED = "speed"
    BALANCED = "balanced"
    QUALITY = "quality"
    DEEP = "deep"