Simple Team Memory Performance Evaluation

Benchmark memory growth across 5 async runs of a weather team with persistent memory and history in Postgres.

Demonstrates team response performance with memory enabled.

Before running, apply the stream-consumption correction below. The original function creates an async iterator but never consumes it, so it does not execute a Team generation. The corrected benchmark completes five coordinator runs with memory extraction enabled.

PerformanceEval runs the callable separately for each enabled metric: warm-ups first, then runtime measurements, then memory measurements. num_iterations applies to each metric, and the default is 10 additional warm-up calls. Model retries, tools, delegation, and memory extraction can add provider requests beyond the callable count.

Memory measurements use Python’s tracemalloc; they do not measure process RSS, GPU memory, database-server memory, or remote model memory. Record dependency versions, database state, and model settings when comparing results.

team_response_with_memory_simple.py
"""
Simple Team Memory Performance Evaluation
=========================================

Demonstrates team response performance with memory enabled.
"""

import asyncio
import random

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.eval.performance import PerformanceEval
from agno.models.openai import OpenAIChat
from agno.team.team import Team

# ---------------------------------------------------------------------------
# Create Sample Inputs
# ---------------------------------------------------------------------------
cities = [
    "New York",
    "Los Angeles",
    "Chicago",
    "Houston",
    "Miami",
    "San Francisco",
    "Seattle",
    "Boston",
    "Washington D.C.",
    "Atlanta",
    "Denver",
    "Las Vegas",
]

# ---------------------------------------------------------------------------
# Create Database
# ---------------------------------------------------------------------------
db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=db_url)


# ---------------------------------------------------------------------------
# Create Tool
# ---------------------------------------------------------------------------
def get_weather(city: str) -> str:
    return f"The weather in {city} is sunny."


# ---------------------------------------------------------------------------
# Create Team
# ---------------------------------------------------------------------------
weather_agent = Agent(
    id="weather_agent",
    model=OpenAIChat(id="gpt-5.2"),
    role="Weather Agent",
    description="You are a helpful assistant that can answer questions about the weather.",
    instructions="Be concise, reply with one sentence.",
    tools=[get_weather],
    db=db,
    update_memory_on_run=True,
    add_history_to_context=True,
)

team = Team(
    members=[weather_agent],
    model=OpenAIChat(id="gpt-5.2"),
    instructions="Be concise, reply with one sentence.",
    db=db,
    markdown=True,
    update_memory_on_run=True,
    add_history_to_context=True,
)


# ---------------------------------------------------------------------------
# Create Benchmark Function
# ---------------------------------------------------------------------------
async def run_team():
    random_city = random.choice(cities)
    _ = team.arun(
        input=f"I love {random_city}! What weather can I expect in {random_city}?",
        stream=True,
        stream_events=True,
    )

    return "Successfully ran team"


# ---------------------------------------------------------------------------
# Create Evaluation
# ---------------------------------------------------------------------------
team_response_with_memory_impact = PerformanceEval(
    name="Team Memory Impact",
    func=run_team,
    num_iterations=5,
    warmup_runs=0,
    measure_runtime=False,
    memory_growth_tracking=True,
)

# ---------------------------------------------------------------------------
# Run Evaluation
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    asyncio.run(
        team_response_with_memory_impact.arun(print_results=True, print_summary=True)
    )

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U agno "psycopg[binary]" openai sqlalchemy

Export your OpenAI API key

export OPENAI_API_KEY="your_openai_api_key_here"

Run PgVector

docker run -d \
  -e POSTGRES_DB=ai \
  -e POSTGRES_USER=ai \
  -e POSTGRES_PASSWORD=ai \
  -e PGDATA=/var/lib/postgresql \
  -v pgvolume:/var/lib/postgresql \
  -p 5532:5432 \
  --name pgvector \
  agnohq/pgvector:18

Consume the Team stream

Replace the complete run_team function with:

run_team replacement
async def run_team():
    random_city = random.choice(cities)
    async for _event in team.arun(
        input=f"I love {random_city}! What weather can I expect in {random_city}?",
        stream=True,
        stream_events=True,
    ):
        pass
    return "Successfully ran team"

An async stream must be iterated to completion; awaiting the iterator itself does not run it.

Enable allocation diagnostics

Add debug_mode=True to team_response_with_memory_impact = PerformanceEval(...) to print allocation comparisons and top allocations. memory_growth_tracking=True alone collects snapshots without printing those details. Debug logging adds work to the benchmark; snapshot growth by itself does not establish a leak.

Run the example

Save the code above as team_response_with_memory_simple.py, then run:

python team_response_with_memory_simple.py

Full source: cookbook/09_evals/performance/team_response_with_memory_simple.py