Understanding the Agent System

@tool
def my_tool(self) -> dict:
    """A tool that behaves differently during error recovery."""
    if self.execution_state == self.STATE_ERROR_RECOVERY:
        # Be more conservative during error recovery
        return {"status": "skipped", "reason": "In recovery mode"}
    # Normal execution
    return {"status": "success", "data": "..."}

from gaia.agents.base.agent import Agent
from gaia.agents.base.console import AgentConsole

class MinimalAgent(Agent):
    """The simplest possible GAIA agent."""

    def _get_system_prompt(self) -> str:
        # This is what the LLM "believes" about itself
        return "You are a helpful assistant."

    def _create_console(self):
        # AgentConsole provides colorful CLI output
        return AgentConsole()

    def _register_tools(self):
        # No tools yet - agent can only chat
        pass

# Create and use
agent = MinimalAgent()
result = agent.process_query("Hello! What can you help me with?")
print(result["result"])

from gaia.agents.base.agent import Agent
from gaia.agents.base.tools import tool
from gaia.agents.base.console import AgentConsole
from datetime import datetime

class TimeAgent(Agent):
    """Agent that can tell you the current time."""

    def _get_system_prompt(self) -> str:
        return """You are a helpful assistant that can tell the time.
        When users ask about time, use the get_current_time tool."""

    def _create_console(self):
        return AgentConsole()

    def _register_tools(self):
        @tool
        def get_current_time() -> dict:
            """Get the current date and time.

            Use this tool when the user asks:
            - What time is it?
            - What's the date?
            - What day is it?

            Returns:
                Dictionary with time, date, and day of week
            """
            now = datetime.now()
            return {
                "time": now.strftime("%I:%M %p"),
                "date": now.strftime("%B %d, %Y"),
                "day": now.strftime("%A")
            }

# Test it
agent = TimeAgent()
result = agent.process_query("What time is it?")
print(result["result"])  # "It's 2:30 PM on Thursday, January 9, 2025"

from gaia.agents.base.agent import Agent
from gaia.agents.base.tools import tool
from gaia.agents.base.console import AgentConsole
import requests
import os

class WeatherAgent(Agent):
    """Agent that provides real weather information."""

    def __init__(self, **kwargs):
        # Store API key before calling super().__init__
        # (super().__init__ calls _register_tools, which needs api_key)
        self.api_key = os.getenv("WEATHER_API_KEY")
        super().__init__(**kwargs)

    def _get_system_prompt(self) -> str:
        return """You are a weather assistant.

        When users ask about weather:
        1. Use get_weather to fetch current conditions
        2. Present the information in a friendly, conversational way
        3. Include temperature, conditions, and any relevant warnings

        Be helpful and proactive - if someone asks about weather for travel,
        mention if they should bring an umbrella or jacket."""

    def _create_console(self):
        return AgentConsole()

    def _register_tools(self):
        @tool
        def get_weather(city: str, country_code: str = "US") -> dict:
            """Get current weather for a city.

            Args:
                city: Name of the city (e.g., "Seattle", "London")
                country_code: Two-letter country code (default: US)

            Use this tool when users ask about:
            - Current weather conditions
            - Temperature
            - Whether they need an umbrella/jacket

            Returns:
                Dictionary with temperature, conditions, humidity, wind
            """
            try:
                url = f"https://api.openweathermap.org/data/2.5/weather"
                params = {
                    "q": f"{city},{country_code}",
                    "appid": self.api_key,
                    "units": "imperial"
                }
                response = requests.get(url, params=params, timeout=10)
                data = response.json()

                if response.status_code != 200:
                    return {
                        "status": "error",
                        "error": data.get("message", "Unknown error"),
                        "suggestion": "Check the city name spelling"
                    }

                return {
                    "status": "success",
                    "city": city,
                    "temperature_f": round(data["main"]["temp"]),
                    "feels_like_f": round(data["main"]["feels_like"]),
                    "conditions": data["weather"][0]["description"],
                    "humidity": data["main"]["humidity"],
                    "wind_mph": round(data["wind"]["speed"])
                }

            except requests.Timeout:
                return {
                    "status": "error",
                    "error": "Weather service timed out",
                    "suggestion": "Try again in a moment"
                }
            except Exception as e:
                return {
                    "status": "error",
                    "error": str(e),
                    "suggestion": "Check your internet connection"
                }

# Usage
agent = WeatherAgent()
result = agent.process_query("What's the weather like in Seattle?")
print(result["result"])

# Multi-turn conversation works too
result = agent.process_query("How about in Miami?")
print(result["result"])

agent = MyAgent(
    # === LLM Selection ===
    use_claude=False,              # Use Anthropic Claude API
    use_chatgpt=False,             # Removed; True raises migration guidance
    # If both are False, uses local Lemonade Server

    # === Local LLM Settings ===
    base_url="http://localhost:13305/api/v1",  # Lemonade server URL
    model_id="Gemma-4-E4B-it-GGUF",  # Model to use

    # === Cloud LLM Settings ===
    claude_model="claude-sonnet-4-20250514",  # Claude model version
    # API key is read from environment: ANTHROPIC_API_KEY

    # === Agent Behavior ===
    max_steps=20,                  # Max reasoning loop iterations
    streaming=True,                # Stream responses token-by-token
    silent_mode=False,             # Suppress console output

    # === Debugging ===
    debug_prompts=False,           # Print raw prompts to console
    show_prompts=False,            # Show prompts in output
)

[evicted: read_file path=src/gaia/cli.py; 41203 chars; read_tool_output(artifact=output_…, entry=1) fetches it]

Fixed the off-by-one in the pagination helper.

I haven't confirmed this works — pytest didn't pass.

result = agent.process_query("fix the pagination bug")
result["verification"]
# {"state": "partially verified", "tool_calls": 4, "passed": ["ruff"],
#  "failed": ["pytest"], "not_run": [], "unchecked_change": None,
#  "ungrounded": []}

I haven't confirmed this works — I didn't read `toybox/dates.py` this turn.

from gaia.agents.base.verification import strip_verification_scope

answer_only = strip_verification_scope(result["result"])

from typing import Any, Dict, List, Optional

from gaia.agents.base.agent import Agent
from gaia.agents.base.console import SilentConsole
from gaia.agents.base.goal_store import Proposal


class WorkspaceAgent(Agent):
    def _get_system_prompt(self) -> str:
        return "You help keep a project workspace tidy."

    def _create_console(self):
        return SilentConsole()

    def _register_tools(self):
        pass

    def on_first_run(self, context: Optional[Dict[str, Any]]) -> List[Proposal]:
        # Reading a directory is safe -> low risk, but still needs approval.
        return [
            Proposal(
                action="index_workspace",
                rationale="No index found; build one for fast retrieval.",
                action_class="file_read",
                risk="low",
            )
        ]

    def on_heartbeat(self, context: Optional[Dict[str, Any]]) -> List[Proposal]:
        # Deleting files is destructive -> needs explicit approval.
        return [
            Proposal(
                action="archive_stale_logs",
                rationale="12 log files older than 30 days.",
                action_class="file_write",
                risk="high",
            )
        ]


agent = WorkspaceAgent(silent_mode=True, skip_lemonade=True)

for proposal in agent.on_first_run(context=None):
    goal = agent.propose(proposal)          # low risk -> still awaits the user
    print(goal.status)                       # "pending_approval"

for proposal in agent.on_heartbeat(context=None):
    goal = agent.propose(proposal)          # high risk -> awaits the user
    print(goal.status)                       # "pending_approval"

python scripts/dev/validate_document_extraction.py MODEL /path/to/new-results-directory