File Upload with Cache

"""
In this example, we upload a text file to Google and then create a cache.

This greatly saves on tokens during normal prompting.
"""

from pathlib import Path
from time import sleep

import requests
from agno.agent import Agent
from agno.models.google import Gemini
from google import genai
from google.genai.types import UploadFileConfig

# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------

client = genai.Client()

# Download txt file
url = "https://storage.googleapis.com/generativeai-downloads/data/a11.txt"
path_to_txt_file = Path(__file__).parent.joinpath("a11.txt")
if not path_to_txt_file.exists():
    print("Downloading txt file...")
    with path_to_txt_file.open("wb") as wf:
        response = requests.get(url, stream=True)
        for chunk in response.iter_content(chunk_size=32768):
            wf.write(chunk)

# Upload the txt file using the Files API
remote_file_path = Path("a11.txt")
remote_file_name = f"files/{remote_file_path.stem.lower().replace('_', '-')}"

txt_file = None
try:
    txt_file = client.files.get(name=remote_file_name)
    print(f"Txt file exists: {txt_file.uri}")
except Exception:
    pass

if not txt_file:
    print("Uploading txt file...")
    txt_file = client.files.upload(
        file=path_to_txt_file, config=UploadFileConfig(name=remote_file_name)
    )

    # Wait for the file to finish processing
    while txt_file and txt_file.state and txt_file.state.name == "PROCESSING":
        print("Waiting for txt file to be processed.")
        sleep(2)
        txt_file = client.files.get(name=remote_file_name)

    print(f"Txt file processing complete: {txt_file.uri}")

# Create a cache with 5min TTL
cache = client.caches.create(
    model="gemini-3.7-flash",
    config={
        "system_instruction": "You are an expert at analyzing transcripts.",
        "contents": [txt_file],
        "ttl": "300s",
    },
)

# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    agent = Agent(
        model=Gemini(id="gemini-3.7-flash", cached_content=cache.name),
    )
    run_output = agent.run(
        "Find a lighthearted moment from this transcript",  # No need to pass the txt file
    )
    print("Metrics: ", run_output.metrics)

from contextlib import contextmanager
from pathlib import Path
from time import monotonic, sleep


@contextmanager
def ready_file(client, path: Path, existing_name: str | None = None):
    existing_name = existing_name or None
    if existing_name is None and not path.is_file():
        raise FileNotFoundError(path)

    uploaded = (
        client.files.get(name=existing_name)
        if existing_name
        else client.files.upload(file=path)
    )
    owned_name = uploaded.name if existing_name is None else None
    try:
        deadline = monotonic() + 300
        while True:
            state = uploaded.state.name if uploaded.state else None
            if state == "ACTIVE":
                if not uploaded.uri or not uploaded.mime_type:
                    raise RuntimeError("Active file has no URI or MIME type")
                yield uploaded
                return
            if state == "FAILED":
                raise RuntimeError(f"File processing failed: {uploaded.name}")
            if state != "PROCESSING" or not uploaded.name:
                raise RuntimeError(f"Unexpected file state: {state}")
            if monotonic() >= deadline:
                raise TimeoutError("File processing exceeded five minutes")
            sleep(2)
            uploaded = client.files.get(name=uploaded.name)
    finally:
        if owned_name:
            client.files.delete(name=owned_name)

from pathlib import Path

import requests
from agno.agent import Agent
from agno.models.google import Gemini
from google import genai
from google_files import ready_file

path = Path(__file__).parent / "a11.txt"
if not path.exists():
    response = requests.get(
        "https://storage.googleapis.com/generativeai-downloads/data/a11.txt", timeout=60
    )
    response.raise_for_status()
    path.write_bytes(response.content)

client = genai.Client()
with ready_file(client, path) as uploaded:
    cache = client.caches.create(
        model="gemini-3.7-flash",
        config={
            "system_instruction": "You analyze transcripts.",
            "contents": [uploaded],
            "ttl": "300s",
        },
    )
    try:
        if not cache.name:
            raise RuntimeError("Cache creation returned no resource name")
        agent = Agent(model=Gemini(id="gemini-3.7-flash", cached_content=cache.name))
        for prompt in ["Find a lighthearted moment.", "Summarize the main technical issue."]:
            result = agent.run(prompt)
            print(result.content)
            print(result.metrics)
    finally:
        if cache.name:
            client.caches.delete(name=cache.name)