> For the complete documentation index, see [llms.txt](https://docs.graphlit.dev/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.graphlit.dev/examples/deep-research/agno.md).

# Agno (Python)

⏱️ **Time**: 30-40 minutes\
🎯 **Level**: Advanced\
💻 **SDK**: Python (Agno framework)

## What You'll Learn

In this tutorial, you'll build a production-ready research agent in Python that:

* ✅ Extracts entities from documents using Graphlit's knowledge graph
* ✅ Performs multi-hop web research (searches for discovered entities)
* ✅ Filters sources **before ingesting** using native reranking
* ✅ Detects convergence automatically (knows when to stop)
* ✅ Synthesizes multi-source reports with citations

**Why Agno**: 5000x faster than LangGraph, 50x less memory. Simple Python functions as tools—no decorators, no complex schemas.

***

## What You'll Build

Same autonomous research agent, Python implementation:

1. **Ingests seed source** - URL or search results
2. **Discovers entities** - From your knowledge graph
3. **Researches each entity** - Exa search, 10 sources per entity
4. **Filters intelligently** - Analyzes 50, ingests only top 8
5. **Detects convergence** - Stops at novelty score <30%
6. **Synthesizes report** - Markdown with citations

**Example**: Wikipedia on "RAG" → 15 entities → 50 sources → 8 filtered → 2000-word report in \~45 seconds

**🔗 Full code**: [GitHub](https://github.com/graphlit/graphlit-samples/tree/main/python/agno-deep-research)

***

## Prerequisites

{% hint style="warning" %}
**Required**:

* Python 3.11+
* [Graphlit account](https://portal.graphlit.dev) + credentials
* [OpenAI API key](https://platform.openai.com/api-keys)
* Package manager: [uv](https://docs.astral.sh/uv/getting-started/installation/) (recommended, `curl -LsSf https://astral.sh/uv/install.sh | sh`) or pip

**Recommended**:

* Complete [Quickstart](/getting-started/quickstart.md) (7 minutes)
* Complete [Knowledge Graph tutorial](/tutorials/knowledge-graph.md) (20 minutes)
  {% endhint %}

***

## Why Agno + Python?

### Agno's Advantages

**Performance:**

* **5000x faster** than LangGraph (\~2-3 microseconds per agent)
* **50x less memory** (\~6.5KB per agent vs 325KB)

**Simplicity:**

* Tools are just Python functions (no decorators!)
* No complex schemas (docstrings = tool descriptions)
* Clean async/await syntax

**Compare Tool Definition:**

**Mastra (TypeScript):**

```typescript
export const myTool = createTool({
  id: 'my-tool',
  description: 'Tool description',
  inputSchema: z.object({ /* Zod schema */ }),
  outputSchema: z.object({ /* Zod schema */ }),
  execute: async ({ context }) => { /* ... */ },
});
```

**Agno (Python):**

```python
async def my_tool(param: str) -> dict:
    """Tool description."""
    # ... implementation
    return {"result": "value"}
```

**460 lines of Python vs 750 lines of TypeScript** for the same functionality.

***

## Why This Matters: What Graphlit Handles

Before we dive into building, understand what Graphlit provides so you don't have to build it:

### Infrastructure (Weeks → Hours)

* ✅ **File parsing** - PDFs, DOCX, audio, video (30+ formats)
* ✅ **Vector database** - Managed Qdrant, auto-scaled
* ✅ **Multi-tenant isolation** - Each user gets isolated environment
* ✅ **GraphQL API** - Auto-generated, authenticated

### Intelligence (Months → API Calls)

* ✅ **Automatic entity extraction** - LLM-powered workflow extracts Person, Organization, Category during ingestion
* ✅ **Knowledge graph** - Built on Schema.org/JSON-LD standard, relationships auto-created
* ✅ **Native reranker** - Fast, accurate relevance scoring (enables our pre-filtering!)
* ✅ **Exa search built-in** - No separate API key needed, semantic web search included
* ✅ **Summary-based RAG** - Scales to 100+ documents via optimized summaries

**Time savings**: Estimated 12-14 weeks of infrastructure development you skip.

**Production proof**: This pattern is used in [Zine](/examples/zine-case-study.md), serving thousands of users with millions of documents.

***

## The Key Innovation: Pre-Ingestion Filtering

Most research implementations blindly ingest everything they find. This creates noise and wastes processing.

**The breakthrough**: Analyze sources **before** fully ingesting them.

Here's the pattern:

1. Quick ingest to temporary collection (lightweight)
2. Use Graphlit's native reranker to score relevance
3. Filter out low-scoring sources (<0.5 relevance)
4. Only fully ingest top 5-8 sources
5. Delete temporary collection

**Why this works**: Graphlit's native reranker is fast enough (\~2 seconds) to analyze 50 sources before deciding which to fully process.

**Result**: Process 8 sources instead of 50. Faster, higher quality, better signal-to-noise.

***

## The 5-Phase Research Algorithm

### Phase 1: Seed Acquisition

Two starting modes:

**URL Mode** - Start from a specific source:

```bash
uv run deep-research --url "https://arxiv.org/abs/2005.11401"
```

Best for: Research papers, documentation, whitepapers

**Search Mode** - Discover seed sources automatically:

```bash
uv run deep-research --search "retrieval augmented generation" --results 5
```

Best for: Open-ended research, new topics

### Phase 2: Entity-Driven Discovery

Instead of keyword-based research, let the knowledge graph drive discovery:

* **Automatic extraction**: Entities extracted during ingestion (no separate step!)
* **Types**: Person, Organization, Category (concepts/technical terms)
* **Ranking**: By occurrence count and semantic importance
* **Selection**: Top 5 become research seeds

**Why entity-driven works**: A RAG paper mentions "vector databases" and "BERT"—those naturally become your next research directions. Mimics human researcher behavior.

### Phase 3: Intelligent Expansion

For each entity:

1. Search Exa for 10 related sources
2. **Pre-filter before ingesting** (the key innovation!)
3. Only ingest top 3-5 highest-quality sources

The filtering workflow:

```
50 sources found via Exa search
  ↓ Quick ingest to temp collection
  ↓ Rerank by relevance (native reranker)
  ↓ Filter (keep score >0.5)
  ↓ Full ingest top 5 only
  ↓ Delete temp collection
8 sources ingested total
```

**Benefit**: Analyze 50, process 8. Significantly faster with better quality.

### Phase 4: Convergence Detection

Automatically detect when research has plateaued:

**Novelty scoring algorithm:**

1. After ingesting new sources, rerank ALL content by relevance to query
2. Check how many recent sources appear in top 10
3. Calculate novelty score: `recent_in_top_10 / total_recent`
4. If score <30%, diminishing returns detected → stop

**Example:**

* Ingested 5 new sources
* Reranked all 25 total sources
* Only 1 new source in top 10
* Novelty: 1/5 = 20% → **Stop researching**

**Why this works**: If new sources don't rank highly vs existing content, they're redundant. Agent stops automatically, no manual intervention.

### Phase 5: Multi-Source Synthesis

Traditional RAG struggles beyond 10-20 documents. We scale to 100+:

**Summary-based RAG approach:**

1. Create conversation scoped to research collection
2. Use `publish_contents()` which operates on optimized summaries
3. LLM synthesizes across all sources simultaneously
4. Citations automatically included

**Python implementation:**

```python
# Create conversation
conversation = await graphlit.client.create_conversation(
    input=ConversationInput(
        name="Research Report",
        collections=[EntityReferenceInput(id=collection_id)],
    )
)

# Generate report using publishContents (summary-based)
response = await graphlit.client.publish_contents(
    publish_type=PublishTypes.MARKDOWN,
    prompt="Synthesize comprehensive report with citations",
    conversation=EntityReferenceInput(id=conversation.create_conversation.id),
)
```

**Why it scales**: Operates on summaries, not full content. Fast, accurate, handles 100+ sources.

***

## Implementation: Step-by-Step

### Step 1: Project Setup (3 min)

**With uv (recommended - faster than pip):**

Install uv if you haven't already:

```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
```

Create project:

```bash
mkdir deep-research && cd deep-research
uv init
uv add agno graphlit-client python-dotenv rich openai
```

**Or with pip:**

```bash
mkdir deep-research && cd deep-research
python -m venv venv
source venv/bin/activate
pip install agno graphlit-client python-dotenv
```

Create `.env`:

```env
# From portal.graphlit.dev
GRAPHLIT_ENVIRONMENT_ID=your_id
GRAPHLIT_ORGANIZATION_ID=your_org
GRAPHLIT_JWT_SECRET=your_secret

# From platform.openai.com
OPENAI_API_KEY=your_key
```

### Step 2: Singleton Graphlit Client (1 min)

**File: `deep_research/graphlit_client.py`**

```python
"""Singleton Graphlit client."""
from graphlit import Graphlit
from dotenv import load_dotenv

# Load environment variables first
load_dotenv()

# One instance, auto-reads env vars
graphlit = Graphlit()
```

**Why singleton**: Same pattern as Mastra—one shared instance, efficient.

**Note**: We load `dotenv` here so environment variables are available when the module is imported.

### Step 3: Build Tools (15 min)

Here's where Agno shines—simple Python functions!

**File: `deep_research/tools.py`**

```python
"""Research tools as simple Python functions.

Agno advantage: No decorators, no schemas - just functions with docstrings!
"""
from .graphlit_client import graphlit
from graphlit_api import *


# Tool 1: Create Workflow
async def create_workflow(name: str) -> dict:
    """Create collection and workflow with automatic entity extraction.
    
    Args:
        name: Name for the research collection
        
    Returns:
        Dict with collection_id and workflow_id
    """
    # Create collection
    coll_response = await graphlit.client.create_collection(
        input=CollectionInput(name=name)
    )
    
    collection_id = coll_response.create_collection.id
    
    # Create workflow with entity extraction
    wf_response = await graphlit.client.upsert_workflow(
        workflow=WorkflowInput(
            name=f"{name} Workflow",
            ingestion=IngestionWorkflowStageInput(
                collections=[EntityReferenceInput(id=collection_id)]
            ),
            extraction=ExtractionWorkflowStageInput(
                jobs=[
                    ExtractionWorkflowJobInput(
                        connector=EntityExtractionConnectorInput(
                            type=EntityExtractionServiceTypes.MODELTEXT,
                            extracted_types=[
                                ObservableTypes.PERSON,
                                ObservableTypes.ORGANIZATION,
                                ObservableTypes.CATEGORY,
                            ],
                            extracted_count=10,
                        )
                    )
                ]
            ),
        )
    )
    
    return {
        "collection_id": collection_id,
        "workflow_id": wf_response.upsert_workflow.id,
    }


# Tool 2: Ingest Document
async def ingest_document(url: str, workflow_id: str, collection_id: str) -> dict:
    """Ingest single document with entity extraction.
    
    Args:
        url: URL to ingest
        workflow_id: Workflow for processing
        collection_id: Collection to add to
        
    Returns:
        Dict with content_id
    """
    response = await graphlit.client.ingest_uri(
        uri=url,
        is_synchronous=True,  # No polling!
        workflow=EntityReferenceInput(id=workflow_id),
        collections=[EntityReferenceInput(id=collection_id)],
    )
    
    return {"content_id": response.ingest_uri.id}
```

**Compare to Mastra:**

* No `createTool()` wrapper
* No Zod schemas
* Docstring = tool description (Agno reads it!)
* Type hints = parameter validation
* Clean async/await

**Tool 3: Pre-Ingestion Filtering** (abbreviated - see [full code](https://github.com/graphlit/graphlit-samples/blob/main/python/agno-deep-research/deep_research/tools.py)):

```python
async def filter_search_results(
    search_results: list[dict],
    query: str,
    max_results: int = 5,
    min_relevance_score: float = 0.5,
) -> dict:
    """Filter search results BEFORE full ingestion.
    
    This is the key innovation - analyze before processing!
    """
    import time
    
    # Create temp resources
    temp_wf = await graphlit.client.upsert_workflow(
        workflow=WorkflowInput(name=f"Temp Filter {int(time.time())}")
    )
    
    temp_coll = await graphlit.client.create_collection(
        input=CollectionInput(name=f"Temp Filter {int(time.time())}")
    )
    
    # Quick ingestion for analysis
    import asyncio
    tasks = [/* parallel ingestion */]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Rerank with native reranker
    reranked = await graphlit.client.retrieve_sources(
        prompt=query,
        filter=ContentCriteriaInput(
            collections=[EntityReferenceInput(id=temp_coll.create_collection.id)]
        ),
        search=SearchStrategyInput(type=SearchTypes.VECTOR),
        rerank=RerankStrategyInput(type=RerankTypes.RERANK),
    )
    
    # Filter and clean up
    # ... (see full code)
    
    return {"filtered_urls": filtered_urls, "reasoning": "..."}
```

**Python advantages**:

* `asyncio.gather()` for parallel operations (cleaner than `Promise.allSettled`)
* List comprehensions for filtering
* Clean exception handling with `return_exceptions=True`

{% hint style="info" %}
**Full tool implementations**: See [GitHub - tools.py](https://github.com/graphlit/graphlit-samples/blob/main/python/agno-deep-research/deep_research/tools.py) for all 9 tools (\~300 lines).
{% endhint %}

### Step 4: Create Agno Agent (2 min)

**File: `deep_research/agent.py`**

```python
"""Deep Research Agent using Agno."""
from agno.agent import Agent
from agno.models.openai import OpenAIChat

from . import tools


research_agent = Agent(
    name="Deep Research Agent",
    
    model=OpenAIChat(id="gpt-4o"),
    
    instructions="""You are an autonomous research agent using semantic memory.

Your workflow:
1. Create workflow + collection with entity extraction
2. Ingest seed URL or search results
3. Extract entities from knowledge graph
4. Select top 5 entities (PERSON, ORGANIZATION, CATEGORY)
5. Search web for each entity (10 results via Exa)
6. Filter search results BEFORE ingesting (use filter_search_results!)
7. Batch ingest only filtered sources
8. Check convergence (use detect_diminishing_returns - stop if <30%)
9. Generate comprehensive report

Always filter before ingesting.""",
    
    # Agno: Just list functions - that's it!
    tools=[
        tools.create_workflow,
        tools.ingest_document,
        tools.ingest_batch,
        tools.extract_entities,
        tools.select_top_entities,
        tools.search_web,
        tools.filter_search_results,
        tools.detect_diminishing_returns,
        tools.generate_report,
    ],
    
    markdown=True,
)
```

**Agno's simplicity**: No `createTool()`, no tool IDs, no schemas. Just functions.

### Step 5: Build CLI (3 min)

**File: `deep_research/main.py`** (abbreviated - see [full code](https://github.com/graphlit/graphlit-samples/blob/main/python/agno-deep-research/deep_research/main.py)):

```python
#!/usr/bin/env python3
import asyncio
import sys
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel

from .agent import research_agent

console = Console()


async def main():
    load_dotenv()
    
    # Parse args (same pattern as Mastra)
    url = None if "--url" not in sys.argv else sys.argv[sys.argv.index("--url") + 1]
    search_query = None if "--search" not in sys.argv else sys.argv[sys.argv.index("--search") + 1]
    
    # Comprehensive validation (env vars, args)
    # ... validation code ...
    
    # Polished header with rich
    console.print()
    console.print(Panel.fit(
        "[bold cyan]Deep Research Agent[/bold cyan]\n"
        "[dim]Powered by Agno + Graphlit[/dim]",
        border_style="cyan"
    ))
    console.print()
    
    console.print(f"[bold]🔍 Research Query:[/bold] '{search_query}'\n")
    console.print("[bold green]🚀 Starting research...[/bold green]\n")
    
    # Run agent with streaming
    await research_agent.aprint_response(prompt, stream=True)
    
    console.print("\n\n[bold green]✅ Research complete![/bold green]\n")


if __name__ == "__main__":
    asyncio.run(main())
```

**Python advantages**:

* `asyncio.run()` handles event loop (simpler than Node.js setup)
* `rich` library for beautiful CLI (like chalk + boxen + ora combined)
* `aprint_response()` streams automatically with tool display
* Clean, readable code

***

## Running Your Agent

**With uv (recommended):**

```bash
uv run deep-research --search "knowledge graphs"
```

**With pip:**

```bash
pip install -e .
deep-research --url "https://en.wikipedia.org/wiki/RAG"
```

**Save to file:**

```bash
uv run deep-research --search "AI agents" > report.md
```

**Cleanup after** (deletes collection, workflow, and content):

```bash
uv run deep-research --search "test query" --cleanup
```

**Note**: Without `--cleanup`, content remains in your Graphlit account for exploration in the portal.

**Alternative commands (all equivalent):**

```bash
python -m deep_research --search "query"  # Works after install
python -m deep_research.main --search "query"  # Always works
```

**Expected output:**

Terminal (progress):

```
┌─────────────────────────────────┐
│ Deep Research Agent             │
│ Powered by Agno + Graphlit     │
└─────────────────────────────────┘

🔍 Research Query: 'knowledge graphs'
   (Starting with top 5 sources)

🚀 Starting research...

[Tool: create_workflow]
✓ Created collection and workflow

[Tool: search_web]  
✓ Found 5 seed sources

[Tool: ingest_batch]
✓ Ingested 5 sources

[Tool: extract_entities]
✓ Extracted 12 entities

[Tool: search_web]
✓ Searched for 5 entities

[Tool: filter_search_results]
✓ Analyzed 50 sources. Kept 8 (relevance >=0.5)

[Tool: ingest_batch]
✓ Ingested 8 filtered sources

[Tool: detect_diminishing_returns]
✓ Novelty: 0.42 - Continue

[Tool: generate_report]

# Research Report: Knowledge Graphs

## Executive Summary
...

✅ Research complete!
```

***

## Production Patterns

### Performance Optimizations

**Parallel operations:**

```python
import asyncio

# Search all entities concurrently
tasks = [search_web(entity["name"]) for entity in entities]
results = await asyncio.gather(*tasks, return_exceptions=True)
```

**Synchronous ingestion:**

```python
# No polling - content ready when call returns
await graphlit.client.ingest_uri(
    uri=url,
    is_synchronous=True,  # Blocks until processed
    workflow=EntityReferenceInput(id=workflow_id),
)
```

**Graceful error handling:**

```python
# Some sources fail? Continue with successful ones
results = await asyncio.gather(*tasks, return_exceptions=True)
successful = [r for r in results if not isinstance(r, Exception)]
```

**Pre-filtering:**

```python
# Analyze 50, ingest only 8
filtered = await filter_search_results(
    search_results=all_results,
    query=query,
    max_results=5,
    min_relevance_score=0.5
)
```

### Typical Session Metrics

**Without filtering:**

* Sources processed: \~50
* Processing time: 2-3 minutes
* Quality: Significant noise

**With filtering:**

* Sources processed: \~8
* Processing time: 30-45 seconds
* Quality: High signal-to-noise ratio

**Agno Performance:**

* Agent startup: <0.003ms (5000x faster than LangGraph)
* Memory usage: \~6.5KB per agent (50x less)
* Report generation: 5-10 seconds

***

## Agno vs Other Frameworks

| Feature             | Agno            | Mastra       | LangGraph       |
| ------------------- | --------------- | ------------ | --------------- |
| **Language**        | Python          | TypeScript   | Python          |
| **Speed**           | 5000x faster    | Fast         | Baseline        |
| **Memory**          | 50x less        | Standard     | Standard        |
| **Tool Definition** | Just functions  | createTool() | @tool decorator |
| **Schema Required** | No (docstrings) | Yes (Zod)    | Yes (Pydantic)  |
| **Code Size**       | \~460 lines     | \~750 lines  | \~800 lines     |
| **Learning Curve**  | Easy            | Medium       | Hard            |

**Choose Agno when:**

* ✅ You prefer Python
* ✅ You want maximum performance
* ✅ You want simpler code
* ✅ You're building high-throughput systems

***

## Next Steps

### Try It Out

```bash
git clone https://github.com/graphlit/graphlit-samples.git
cd graphlit-samples/python/agno-deep-research
uv sync
cp .env.example .env
# Add credentials
uv run python -m deep_research.main --search "your query"
```

### Extend It

**Domain-specific entities:**

Medical research:

```python
extracted_types=[
    ObservableTypes.MEDICALCONDITION,
    ObservableTypes.DRUG,
    ObservableTypes.PERSON,  # Researchers
]
```

Legal research:

```python
extracted_types=[
    ObservableTypes.LEGALCASE,
    ObservableTypes.CONTRACT,
    ObservableTypes.ORGANIZATION,  # Law firms
]
```

Business intelligence:

```python
extracted_types=[
    ObservableTypes.PRODUCT,
    ObservableTypes.EVENT,
    ObservableTypes.ORGANIZATION,  # Companies
]
```

**Multi-pass research:**

* Extract entities from Layer 2 results
* Research 2-3 passes deep
* Configurable depth limits

**Real-time monitoring:**

* Create Exa feeds for discovered entities
* Auto-expand knowledge base daily

**FastAPI server** (Agno built-in!):

```python
from agno.agent import Agent

agent = Agent(tools=[...])
agent.app  # Built-in FastAPI server!
```

### Learn More

**Related Tutorials:**

* [Mastra (TypeScript)](/examples/deep-research/mastra.md) - Same algorithm, TypeScript
* [Knowledge Graph](/tutorials/knowledge-graph.md) - Entity extraction deep-dive

**Production Example:**

* [Zine Case Study](/examples/zine-case-study.md) - Real app serving thousands

**Resources:**

* [Agno Documentation](https://docs.agno.com)
* [Graphlit Documentation](https://docs.graphlit.dev)
* [Discord Community](https://discord.gg/ygFmfjy3Qx)

***

## Summary

You've learned to build a production-ready autonomous research agent in Python:

**Key innovations** (same as Mastra):

1. Pre-ingestion filtering with native reranker
2. Autonomous convergence detection
3. Summary-based RAG for scale
4. Entity-driven discovery

**Agno advantages**:

* 5000x faster execution
* 50x less memory
* Simpler code (460 vs 750 lines)
* No complex schemas
* Clean Python async/await

**Time investment**: 30-40 minutes\
**Value delivered**: Production-ready patterns, weeks of infrastructure eliminated

This approach works for competitive intelligence, market research, technical deep-dives, and any multi-source synthesis.

***

*Complete implementation:* [*GitHub Repository*](https://github.com/graphlit/graphlit-samples/tree/main/python/agno-deep-research)