ronantakizawa/codereview-bench huggingface dataset:
Repository: phidatahq/phidata
File: cookbook/agents/caching/async_cache_model_response.py (Python)

Code before:
"""
Example showing async caching for model responses.

Run this cookbook twice to see the difference in response time.

The first time should take a while to run.
The second time should be instant.
You can also see in the console about the cache hit/miss status
"""

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-4o", cache_response=True)
)


async def main():
    # Should take a while to run the first time, then replay from cache
    response = await agent.arun(
        "Write me a very very short and sweet story about a cat that can talk and solve problems."
    )
    print(response.content)


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


Review comment (performance):
```suggestion
    # Should take a while to run the first time, then reply from cache
```

Code after:
"""
Example showing async caching for model responses.

The first run will take a while to finish.
The second run will hit the cache and be much faster.

You can also see the cache hit announcement in the console logs.
"""

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(model=OpenAIChat(id="gpt-4o", cache_response=True))


async def main():
    # Run the same query twice to demonstrate caching
    for i in range(1, 3):
        print(f"\n{'=' * 60}")
        print(
            f"Run {i}: {'Cache Miss (First Request)' if i == 1 else 'Cache Hit (Cached Response)'}"
        )
        print(f"{'=' * 60}\n")

        response = await agent.arun(
            "Write me a short story about a cat that can talk and solve problems."
        )

        print(response.content)
        print(f"\n Elapsed time: {response.metrics.duration:.3f}s")

        # Small delay between iterations for clarity
        if i == 1:
            await asyncio.sleep(0.5)


if __name__ == "__main__":
    asyncio.run(main())
Repository: phidatahq/phidata
File: cookbook/agents/caching/async_cache_model_response.py (Python)

Code before:
"""
Example showing async caching for model responses.

Run this cookbook twice to see the difference in response time.

The first time should take a while to run.
The second time should be instant.
You can also see in the console about the cache hit/miss status
"""

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-4o", cache_response=True)
)


async def main():
    # Should take a while to run the first time, then replay from cache
    response = await agent.arun(
        "Write me a very very short and sweet story about a cat that can talk and solve problems."
    )
    print(response.content)


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


Review comment (performance):
should we rather have a loop with 2 iterations here to run the same query so that in a single run the cookbook makes the point and shows the cache feature. instead of having to run the cookbook twice? wdyt

Code after:
"""
Example showing async caching for model responses.

The first run will take a while to finish.
The second run will hit the cache and be much faster.

You can also see the cache hit announcement in the console logs.
"""

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(model=OpenAIChat(id="gpt-4o", cache_response=True))


async def main():
    # Run the same query twice to demonstrate caching
    for i in range(1, 3):
        print(f"\n{'=' * 60}")
        print(
            f"Run {i}: {'Cache Miss (First Request)' if i == 1 else 'Cache Hit (Cached Response)'}"
        )
        print(f"{'=' * 60}\n")

        response = await agent.arun(
            "Write me a short story about a cat that can talk and solve problems."
        )

        print(response.content)
        print(f"\n Elapsed time: {response.metrics.duration:.3f}s")

        # Small delay between iterations for clarity
        if i == 1:
            await asyncio.sleep(0.5)


if __name__ == "__main__":
    asyncio.run(main())
Repository: phidatahq/phidata
File: cookbook/agents/caching/async_cache_model_response_stream.py (Python)

Code before:
"""
Example showing async caching for streaming model responses.

Run this cookbook twice to see the difference in response time.

The first time should take a while to run.
The second time should be instant.
"""

import asyncio

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    model=OpenAIChat(id="gpt-4o", cache_response=True)
)

# Should take a while to run the first time, then replay from cache
asyncio.run(
    agent.aprint_response(
        "Write me a very very very short story about a cat that can talk and solve problems.",
        stream=True,
    )
)


Review comment (performance):
```suggestion
# Should take a while to run the first time, then reply from cache
```

Code after:
"""
Example showing async caching for streaming model responses.

The first run will take a while to finish.
The second run will hit the cache and be much faster.

You can also see the cache hit log in the console logs.
"""

import asyncio
import time

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(model=OpenAIChat(id="gpt-4o", cache_response=True))


async def main():
    # Run the same query twice to demonstrate caching
    for i in range(1, 3):
        print(f"\n{'=' * 60}")
        print(
            f"Run {i}: {'Cache Miss (First Request)' if i == 1 else 'Cache Hit (Cached Response)'}"
        )
        print(f"{'=' * 60}\n")

        start_time = time.time()
        await agent.aprint_response(
            "Write me a short story about a cat that can talk and solve problems.",
            stream=True
        )
        elapsed_time = time.time() - start_time

        print()  # New line after streaming
        print(f"\n Elapsed time: {elapsed_time:.3f}s")

        # Small delay between iterations for clarity
        if i == 1:
            await asyncio.sleep(0.5)


if __name__ == "__main__":
    asyncio.run(main())
Repository: phidatahq/phidata
File: libs/agno/scripts/migrate_vectordbs_to_v2.py (Python)

Code before:
"""Use this script to migrate your Agno VectorDBs from v1 to v2

This script adds the new columns introduced in v2 to SQL-based vector databases:
- content_hash: String column for content hash tracking
- content_id: String column for content ID tracking

Supported databases:
- PgVector: Adds columns to existing tables (no data movement needed)
- SingleStore: Adds columns to existing tables (no data movement needed)

- Configure your db_url and table details in the script
- Run the script
"""

from agno.utils.log import log_error, log_info, log_warning

# --- Set these variables before running the script ---

## Your database connection details ##
db_url = ""  # For PgVector, SingleStore

## Postgres Sample ##
# db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"

## SingleStore Sample ##
# db_url = "mysql+pymysql://user:password@host:port/database"

## Configure which vectordbs to migrate ##

Review comment (suggestion):
```suggestion
This script works with PgVector and SingleStore.

This script will update the provided tables to add the two new columns introduced in v2:
```

Code after:
# mypy: disable-error-code=var-annotated
"""Use this script to migrate your Agno VectorDBs from v1 to v2

This script works with PgVector and SingleStore.

This script will update the provided tables to add the two new columns introduced in v2:
- content_hash: String column for content hash tracking
- content_id: String column for content ID tracking

To use the script simply:
- For PGVector, set the `pg_vector_db_url` and `pg_vector_config` variables
- For SingleStore, set the `singlestore_db_url` and `singlestore_config` variables
- Run the script
"""

from agno.utils.log import log_error, log_info, log_warning

# ------------ Setup for PGVector ------------

## Your database connection string
pg_vector_db_url = ""  # Example: "postgresql+psycopg://ai:ai@localhost:5532/ai"

## Configuration of the schema and tables to migrate
pg_vector_config = {
    # "schema": "ai",  # Schema where your tables are located
    # "table_names": ["documents"],  # Tables to migrate
}
# -----------------------------------------
Repository: phidatahq/phidata
File: libs/agno/scripts/migrate_vectordbs_to_v2.py (Python)

Code before:
"""Use this script to migrate your Agno VectorDBs from v1 to v2

This script works with PgVector and SingleStore.

This script will update the provided tables to add the two new columns introduced in v2:
- content_hash: String column for content hash tracking
- content_id: String column for content ID tracking

- Configure your db_url and table details in the script
- Run the script
"""

from agno.utils.log import log_error, log_info, log_warning

# --- Set these variables before running the script ---

## Your database connection details ##
db_url = ""  # For PgVector, SingleStore

## Postgres Sample ##
# db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"

## SingleStore Sample ##
# db_url = "mysql+pymysql://user:password@host:port/database"

## Configure which vectordbs to migrate ##
vectordb_configs = {
    # Uncomment and configure the databases you want to migrate
    # "pgvector": {
    #     "schema": "ai",  # Schema name where your vectordb tables are located
    #     "table_names": ["knowledge_base", "documents"],  # List of table names to migrate
    # },
    # "singlestore": {
    #     "schema": "ai",
    #     "table_names": ["knowledge_base", "documents"],

Review comment (suggestion):
```suggestion
Usage: 
- Configure your db_url and table details in the script
- Run the script
```

Code after:
# mypy: disable-error-code=var-annotated
"""Use this script to migrate your Agno VectorDBs from v1 to v2

This script works with PgVector and SingleStore.

This script will update the provided tables to add the two new columns introduced in v2:
- content_hash: String column for content hash tracking
- content_id: String column for content ID tracking

To use the script simply:
- For PGVector, set the `pg_vector_db_url` and `pg_vector_config` variables
- For SingleStore, set the `singlestore_db_url` and `singlestore_config` variables
- Run the script
"""

from agno.utils.log import log_error, log_info, log_warning

# ------------ Setup for PGVector ------------

## Your database connection string
pg_vector_db_url = ""  # Example: "postgresql+psycopg://ai:ai@localhost:5532/ai"

## Configuration of the schema and tables to migrate
pg_vector_config = {
    # "schema": "ai",  # Schema where your tables are located
    # "table_names": ["documents"],  # Tables to migrate
}
# -----------------------------------------

# ------------ Setup for SingleStore ------------

# Your database connection string
singlestore_db_url = ""  # Example: "mysql+pymysql://user:password@host:port/database"

# Exact configuration of the tables to migrate
Repository: phidatahq/phidata
File: cookbook/knowledge/filters/async_filtering.py (Python)

Code before:
import asyncio
import os

from agno.agent import Agent
from agno.db.sqlite import AsyncSqliteDb
from agno.knowledge.knowledge import Knowledge
from agno.utils.media import (
    SampleDataFileExtension,
    download_knowledge_filters_sample_data,
)
from agno.vectordb.lancedb import LanceDb

# Download all sample CVs and get their paths
downloaded_cv_paths = download_knowledge_filters_sample_data(
    num_files=5, file_extension=SampleDataFileExtension.DOCX
)

# Initialize PostgresDB
if os.path.exists("tmp/knowledge_contents.db"):
    os.remove("tmp/knowledge_contents.db")

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

# Initialize LanceDB
# By default, it stores data in /tmp/lancedb
vector_db = LanceDb(
    table_name="CVs",
    uri="tmp/lancedb",  # You can change this path to store data elsewhere
)

# Step 1: Initialize knowledge base with documents and metadata
# ------------------------------------------------------------------------------
# When initializing the knowledge base, we can attach metadata that will be used for filtering
# This metadata can include user IDs, document types, dates, or any other attributes

knowledge = Knowledge(
    name="Async Filtering",
    vector_db=vector_db,
    contents_db=contents_db,
)


Review comment (suggestion):
```suggestion
# Cleanup Sqlite database if it already exists
```

Code after:
import asyncio
import os

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

# Download all sample CVs and get their paths
downloaded_cv_paths = download_knowledge_filters_sample_data(
    num_files=5, file_extension=SampleDataFileExtension.DOCX
)

# Clean up old databases
if os.path.exists("tmp/knowledge_contents.db"):
    os.remove("tmp/knowledge_contents.db")
db = AsyncSqliteDb(
    db_file="tmp/knowledge_contents.db",
)

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

# Step 1: Initialize knowledge base with documents and metadata
# ------------------------------------------------------------------------------
# When initializing the knowledge base, we can attach metadata that will be used for filtering
# This metadata can include user IDs, document types, dates, or any other attributes

knowledge = Knowledge(
    name="Async Filtering",
    vector_db=vector_db,
    contents_db=db,
)

asyncio.run(
Repository: phidatahq/phidata
File: cookbook/agent_os/interfaces/slack/basic.py (Python)

Code before:
from agno.agent import Agent
from agno.db.sqlite.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.os.app import AgentOS
from agno.os.interfaces.slack import Slack

agent_db = SqliteDb(session_table="agent_sessions", db_file="tmp/persistent_memory.db")

basic_agent = Agent(
    name="Basic Agent",
    model=OpenAIChat(id="gpt-4o"),
    db=agent_db,
    add_history_to_context=True,
    num_history_runs=3,
    add_datetime_to_context=True,
)

# Setup our AgentOS app
agent_os = AgentOS(
    agents=[basic_agent],
    interfaces=[
        Slack(
            agent=basic_agent, reply_to_mentions_only=True
        )  # bot replies only when mentioned
    ],
)
app = agent_os.get_app()


if __name__ == "__main__":
    """Run your AgentOS.

    You can see the configuration and available apps at:
    http://localhost:7777/config

    """
    agent_os.serve(app="basic:app", reload=True)


Review comment (suggestion):
```suggestion
        Slack(
            agent=basic_agent, 
            reply_to_mentions_only=True, # The Agent will react only to messages mentioning it
        ) 
```

Code after:
from agno.agent import Agent
from agno.db.sqlite.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.os.app import AgentOS
from agno.os.interfaces.slack import Slack

agent_db = SqliteDb(session_table="agent_sessions", db_file="tmp/persistent_memory.db")

basic_agent = Agent(
    name="Basic Agent",
    model=OpenAIChat(id="gpt-4o"),
    db=agent_db,
    add_history_to_context=True,
    num_history_runs=3,
    add_datetime_to_context=True,
)

# Setup our AgentOS app
agent_os = AgentOS(
    agents=[basic_agent],
    interfaces=[
        Slack(
            agent=basic_agent, 
            reply_to_mentions_only=True, # The Agent will react only to messages mentioning it
        ) 
    ],
)
app = agent_os.get_app()


if __name__ == "__main__":
    """Run your AgentOS.

    You can see the configuration and available apps at:
    http://localhost:7777/config

    """
    agent_os.serve(app="basic:app", reload=True)
Repository: phidatahq/phidata
File: libs/agno/agno/exceptions.py (Python)

Code before:
    """Exception raised when an output check fails."""

    def __init__(
        self,
        message: str,
        check_trigger: CheckTrigger = CheckTrigger.OUTPUT_NOT_ALLOWED,
        additional_data: Optional[Dict[str, Any]] = None,
    ):
        super().__init__(message)
        self.type = "output_check_error"
        if isinstance(check_trigger, CheckTrigger):
            self.error_id = check_trigger.value
        else:
            self.error_id = str(check_trigger)

        self.message = message
        self.check_trigger = check_trigger
        self.additional_data = additional_data


@dataclass
class RetryableModelProviderError(Exception):
    # Guidance message to retry a model invocation after an error
    retry_guidance_message: Optional[str] = None
    # Error code identifying the type of error
    error_code: Optional[str] = None


Review comment (suggestion):
I'd make this more generic and just say `original_error` or `error_message`. Then in Gemini you make a string that includes that error code they give.

Code after:
    """Exception raised when an output check fails."""

    def __init__(
        self,
        message: str,
        check_trigger: CheckTrigger = CheckTrigger.OUTPUT_NOT_ALLOWED,
        additional_data: Optional[Dict[str, Any]] = None,
    ):
        super().__init__(message)
        self.type = "output_check_error"
        if isinstance(check_trigger, CheckTrigger):
            self.error_id = check_trigger.value
        else:
            self.error_id = str(check_trigger)

        self.message = message
        self.check_trigger = check_trigger
        self.additional_data = additional_data


@dataclass
class RetryableModelProviderError(Exception):
    original_error: Optional[str] = None
    # Guidance message to retry a model invocation after an error
    retry_guidance_message: Optional[str] = None
Repository: phidatahq/phidata
File: libs/agno/agno/knowledge/reader/wikipedia_reader.py (Python)

Code before:
                    name=topic,
                    meta_data={"topic": topic},
                    content=summary,
                )
            ]
        return []

    async def async_read(self, topic: str) -> List[Document]:
        """
        Asynchronously read content from Wikipedia.

        Args:
            topic: The Wikipedia topic to read

        Returns:
            A list of documents containing the Wikipedia summary
        """
        log_debug(f"Async reading Wikipedia topic: {topic}")
        summary = None
        try:
            # Run the synchronous wikipedia API call in a thread pool
            summary = await asyncio.to_thread(wikipedia.summary, topic, auto_suggest=self.auto_suggest)

        except wikipedia.exceptions.PageError:
            summary = None
            log_info("PageError: Page not found.")

        # Only create Document if we successfully got a summary
        if summary:
            return [
                Document(
                    name=topic,
                    meta_data={"topic": topic},
                    content=summary,
                )
            ]
        return []


Review comment (bug):
```suggestion
            log_debug("Wikipedia PageError: Page not found.")
```

Code after:
                    name=topic,
                    meta_data={"topic": topic},
                    content=summary,
                )
            ]
        return []

    async def async_read(self, topic: str) -> List[Document]:
        """
        Asynchronously read content from Wikipedia.

        Args:
            topic: The Wikipedia topic to read

        Returns:
            A list of documents containing the Wikipedia summary
        """
        log_debug(f"Async reading Wikipedia topic: {topic}")
        summary = None
        try:
            # Run the synchronous wikipedia API call in a thread pool
            summary = await asyncio.to_thread(wikipedia.summary, topic, auto_suggest=self.auto_suggest)

        except wikipedia.exceptions.PageError:
            summary = None
            log_info("Wikipedia Error: Page not found.")

        # Only create Document if we successfully got a summary
        if summary:
            return [
                Document(
                    name=topic,
                    meta_data={"topic": topic},
                    content=summary,
                )
            ]
        return []
Repository: phidatahq/phidata
File: cookbook/06_agent_os/advanced_demo/reasoning_model.py (Python)

Code before:
"""
Test script for streaming reasoning content.

This demonstrates the new streaming reasoning feature where reasoning content
is streamed as it arrives instead of all at once.
"""

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.run.agent import RunEvent  # noqa
from agno.os import AgentOS

# Create an agent with reasoning enabled
agent = Agent(
    reasoning_model=Claude(
        id="claude-sonnet-4-5",
        thinking={"type": "enabled", "budget_tokens": 1024},
    ),
    reasoning=True,
    instructions="Think step by step about the problem.",
)

# Setup our AgentOS app
agent_os = AgentOS(
    description="Reasoning model streaming",
    agents=[agent],
)
app = agent_os.get_app()



Review comment (suggestion):
```suggestion
Example showing a reasoning Agent in the AgentOS.

You can interact with the Agent as normally. It will reason before providing a final answer.
You will see its chain of thought live as it is generated.
```

Code after:
"""
Example showing a reasoning Agent in the AgentOS.

You can interact with the Agent as normally. It will reason before providing a final answer.
You will see its chain of thought live as it is generated.
"""

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.run.agent import RunEvent  # noqa
from agno.os import AgentOS

# Create an agent with reasoning enabled
agent = Agent(
    reasoning_model=Claude(
        id="claude-sonnet-4-5",
        thinking={"type": "enabled", "budget_tokens": 1024},
    ),
    reasoning=True,
    instructions="Think step by step about the problem.",
)

# Setup our AgentOS app
agent_os = AgentOS(
    description="Reasoning model streaming",
    agents=[agent],
)
app = agent_os.get_app()