Async Agentic Filtering

Run Async Agentic Filtering.

import asyncio
import os

from agno.agent import Agent
from agno.db.sqlite import AsyncSqliteDb
from agno.knowledge.knowledge import Knowledge
from agno.models.openai import OpenAIChat
from agno.utils.media import (
    SampleDataFileExtension,
    download_knowledge_filters_sample_data,
)
from agno.vectordb.pgvector import PgVector

# Download all sample sales files and get their paths
downloaded_csv_paths = download_knowledge_filters_sample_data(
    num_files=4, file_extension=SampleDataFileExtension.CSV
)

# Clean up old database
if os.path.exists("tmp/knowledge_contents.db"):
    os.remove("tmp/knowledge_contents.db")


db = AsyncSqliteDb(
    db_file="tmp/knowledge_contents.db",
)

# Initialize PgVector
vector_db = PgVector(
    table_name="recipes",
    db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
)

# Step 1: Initialize knowledge base with documents and metadata
# ------------------------------------------------------------------------------

knowledge = Knowledge(
    name="CSV Knowledge Base",
    description="A knowledge base for CSV files",
    vector_db=vector_db,
    contents_db=db,
)

# Load all documents into the vector database
asyncio.run(
    knowledge.ainsert_many(
        [
            {
                "path": downloaded_csv_paths[0],
                "metadata": {
                    "data_type": "sales",
                    "quarter": "Q1",
                    "year": 2024,
                    "region": "north_america",
                    "currency": "USD",
                },
            },
            {
                "path": downloaded_csv_paths[1],
                "metadata": {
                    "data_type": "sales",
                    "year": 2024,
                    "region": "europe",
                    "currency": "EUR",
                },
            },
            {
                "path": downloaded_csv_paths[2],
                "metadata": {
                    "data_type": "survey",
                    "survey_type": "customer_satisfaction",
                    "year": 2024,
                    "target_demographic": "mixed",
                },
            },
            {
                "path": downloaded_csv_paths[3],
                "metadata": {
                    "data_type": "financial",
                    "sector": "technology",
                    "year": 2024,
                    "report_type": "quarterly_earnings",
                },
            },
        ],
        skip_if_exists=True,
    )
)

# Step 2: Query the knowledge base with Agent using filters from query automatically
# -----------------------------------------------------------------------------------

# Enable agentic filtering
agent = Agent(
    model=OpenAIChat("gpt-5.2"),
    knowledge=knowledge,
    db=db,
    search_knowledge=True,
    enable_agentic_knowledge_filters=True,
    debug_mode=True,
    markdown=True,
)

asyncio.run(
    agent.aprint_response(
        "Tell me about revenue performance and top selling products in the region north_america and data_type sales",
        stream=True,
    )
)

Run the Example

# Clone and setup repo
git clone https://github.com/agno-agi/agno.git
cd agno/cookbook/07_knowledge/filters

# Create and activate virtual environment
./scripts/demo_setup.sh
source .venvs/demo/bin/activate

# Optiona: Run PgVector (needs docker)
./cookbook/scripts/run_pgvector.sh

python async_agentic_filtering.py