Retrievers

from exo.retrieval import (
    Retriever,
    VectorRetriever,
    SparseRetriever,
    HybridRetriever,
    AgenticRetriever,
    GraphRetriever,
    QueryRewriter,
    Reranker,
    LLMReranker,
    Triple,
    TripleExtractor,
    retrieve_tool,
    index_tool,
)

class Retriever(abc.ABC)

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> list[RetrievalResult]

class VectorRetriever(Retriever)

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> list[RetrievalResult]

from exo.retrieval import OpenAIEmbeddings, InMemoryVectorStore, VectorRetriever

embeddings = OpenAIEmbeddings(api_key="sk-...")
store = InMemoryVectorStore()
retriever = VectorRetriever(embeddings, store, score_threshold=0.5)

results = await retriever.retrieve("What is Exo?", top_k=3)
for r in results:
    print(f"[{r.score:.3f}] {r.chunk.content[:80]}...")

class SparseRetriever(Retriever)

def index(self, chunks: list[Chunk]) -> None

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> list[RetrievalResult]

from exo.retrieval import SparseRetriever, CharacterChunker, Document

doc = Document(id="doc1", content="Python is a programming language...")
chunks = CharacterChunker(chunk_size=200).chunk(doc)

sparse = SparseRetriever(k1=1.5, b=0.75)
sparse.index(chunks)

results = await sparse.retrieve("programming language", top_k=3)

class HybridRetriever(Retriever)

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> list[RetrievalResult]

from exo.retrieval import (
    VectorRetriever,
    SparseRetriever,
    HybridRetriever,
    OpenAIEmbeddings,
    InMemoryVectorStore,
)

embeddings = OpenAIEmbeddings(api_key="sk-...")
store = InMemoryVectorStore()
vector = VectorRetriever(embeddings, store)

sparse = SparseRetriever()
sparse.index(chunks)  # pre-indexed chunks

hybrid = HybridRetriever(
    vector_retriever=vector,
    sparse_retriever=sparse,
    vector_weight=0.7,  # 70% dense, 30% sparse
)

results = await hybrid.retrieve("What is Exo?", top_k=5)

class AgenticRetriever(Retriever)

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> list[RetrievalResult]

from exo.retrieval import (
    AgenticRetriever,
    QueryRewriter,
    VectorRetriever,
    OpenAIEmbeddings,
    InMemoryVectorStore,
)

embeddings = OpenAIEmbeddings(api_key="sk-...")
store = InMemoryVectorStore()
base = VectorRetriever(embeddings, store)
rewriter = QueryRewriter("openai:gpt-4o")

agentic = AgenticRetriever(
    base_retriever=base,
    rewriter=rewriter,
    model="openai:gpt-4o",
    max_rounds=3,
    sufficiency_threshold=0.7,
)

# May perform up to 3 rounds of retrieve-judge-rewrite
results = await agentic.retrieve("Explain how Exo agents handle tool calls")

class GraphRetriever(Retriever)

async def retrieve(
    self,
    query: str,
    *,
    top_k: int = 5,
    **kwargs: Any,
) -> list[RetrievalResult]

from exo.retrieval import GraphRetriever, TripleExtractor, VectorRetriever

# Extract triples from your chunks first
extractor = TripleExtractor("openai:gpt-4o")
triples = await extractor.extract(chunks)

# Wrap a base retriever with graph expansion
graph = GraphRetriever(
    base_retriever=vector_retriever,
    triples=triples,
    beam_width=3,
    max_hops=2,
)

results = await graph.retrieve("Who created Python?")
for r in results:
    if "graph_hop" in r.metadata:
        print(f"  (expanded at hop {r.metadata['graph_hop']})")

@dataclass(frozen=True)
class Triple

class TripleExtractor(
    model: str,
    *,
    prompt_template: str | None = None,
    **provider_kwargs: Any,
)

async def extract(self, chunks: list[Chunk]) -> list[Triple]

class QueryRewriter(
    model: str,
    *,
    prompt_template: str | None = None,
    **provider_kwargs: Any,
)

async def rewrite(
    self,
    query: str,
    *,
    history: list[str] | None = None,
) -> str

from exo.retrieval import QueryRewriter

rewriter = QueryRewriter("openai:gpt-4o")

# Simple rewrite
better_query = await rewriter.rewrite("How do agents work?")

# With conversation history
better_query = await rewriter.rewrite(
    "How does it handle errors?",
    history=["User asked about Exo agents", "Discussed tool calls"],
)

class Reranker(abc.ABC)

async def rerank(
    self,
    query: str,
    results: list[RetrievalResult],
    *,
    top_k: int = 5,
) -> list[RetrievalResult]

class LLMReranker(Reranker)

from exo.retrieval import LLMReranker

reranker = LLMReranker("openai:gpt-4o")

# Rerank initial retrieval results
reranked = await reranker.rerank(
    "What is Exo?",
    initial_results,
    top_k=3,
)
for r in reranked:
    print(f"[{r.score:.3f}] {r.chunk.content[:80]}...")
    print(f"  (original score: {r.metadata['original_score']:.3f})")

def retrieve_tool(
    retriever: Retriever,
    *,
    name: str = "retrieve",
) -> FunctionTool

def index_tool(
    chunker: Chunker,
    store: VectorStore,
    embeddings: Embeddings,
    *,
    name: str = "index_document",
) -> FunctionTool

from exo import Agent, run
from exo.retrieval import (
    CharacterChunker,
    OpenAIEmbeddings,
    InMemoryVectorStore,
    VectorRetriever,
    retrieve_tool,
    index_tool,
)

embeddings = OpenAIEmbeddings(api_key="sk-...")
store = InMemoryVectorStore()
chunker = CharacterChunker()
retriever = VectorRetriever(embeddings, store)

agent = Agent(
    name="rag_agent",
    model="openai:gpt-4o",
    tools=[
        retrieve_tool(retriever),
        index_tool(chunker, store, embeddings),
    ],
)

result = await run(agent, "Index this document: 'Exo is a multi-agent framework.'")
result = await run(agent, "What is Exo?")

from exo.retrieval import Parser, TextParser, MarkdownParser, JSONParser, PDFParser

class Parser(abc.ABC)

# From the exo-ai monorepo root or packages/exo-retrieval directory
uv sync --extra pdf

from pathlib import Path
from exo.retrieval import MarkdownParser, CharacterChunker

parser = MarkdownParser()
doc = parser.parse(Path("README.md"))

chunker = CharacterChunker(chunk_size=500)
chunks = chunker.chunk(doc)