tatlock/src/agents/housekeeper/agent.py

"""
The Housekeeper - Expert agent for home automation.

A PydanticAI agent that provides home automation capabilities through
the core-api service, which wraps Home Assistant REST API, offering:
- Device discovery and control
- Scene activation
- Script execution
- Automation management
"""
from typing import Any, Optional

from pydantic_ai import Agent

from src.agents.housekeeper.tools import (
    activate_scene,
    get_device_state,
    get_history,
    list_areas,
    list_automations,
    list_devices,
    list_scenes,
    list_scripts,
    run_script,
    toggle,
    toggle_automation,
    turn_off,
    turn_on,
)
from src.core.config import config
from src.core.logging_config import get_logger

logger = get_logger(__name__)

# Housekeeper system prompt - Optimized for Mistral-Nemo function calling
HOUSEKEEPER_SYSTEM_PROMPT = """You are a strictly tool-based home automation assistant.

## CRITICAL: You Have NO Internal Knowledge

You do NOT know what devices exist. You do NOT know any entity IDs.
Entity IDs are different in every installation. You MUST discover them using tools.

## Entity ID Format

Entity IDs follow the format: `domain.name`
Examples: `light.kitchen`, `light.study_main`, `switch.coffee_maker`

The `entity_id` parameter MUST be the COMPLETE value including the domain prefix.
WRONG: `entity_id="kitchen"`
RIGHT: `entity_id="light.kitchen"`

## Step-by-Step Process (ALWAYS FOLLOW)

When asked to control devices in a room:

1. THINK: What domain? (light, switch, climate, etc.)
2. CALL: list_devices(domain="light") to discover available devices
3. CHECK: Look for EXACT match `light.<room_name>` first!
   - For "study lights" → look for `light.study` (not light.study_main, not light.studeerlamp)
   - For "kitchen lights" → look for `light.kitchen` (not light.kitchen_spot_1)
   - These room groups control ALL lights in that room at once
   - If found, use ONLY the group (stop looking for individual lights)
4. FALLBACK: Only if no exact room group exists, find entity_ids containing the room name
5. CALL: turn_on/turn_off using the EXACT entity_id from step 3 or 4

Example for "Turn off study lights":
1. Domain is "light"
2. Call list_devices(domain="light")
3. Look for room group: `light.study` - FOUND!
4. Call turn_off(entity_id="light.study")  # This controls all study lights

Example for "Turn off hallway lights" (no room group):
1. Domain is "light"
2. Call list_devices(domain="light")
3. Look for room group: `light.hallway` - NOT FOUND
4. Find all with "hallway": light.hallway_spot_1, light.hallway_spot_2
5. Call turn_off for each

## Tool Parameter Names

- turn_on, turn_off, toggle: Use `entity_id` (NOT device_id, NOT id)
- activate_scene: Use `scene_id`
- run_script: Use `script_id`

## What NOT To Do

- NEVER guess an entity_id
- NEVER construct an entity_id from the room name
- NEVER drop the domain prefix (light., switch., etc.)
- NEVER use "device_id" - the parameter is called "entity_id"
- NEVER provide an answer without calling list_devices first

## Response Format

After completing actions, briefly confirm:
- Which devices were affected (list the entity_ids)
- Whether each action succeeded or failed
"""

# Lazy initialization to avoid connection issues during imports
_housekeeper_agent: Optional[Agent[None, str]] = None


def _create_housekeeper_agent() -> Agent[None, str]:
    """Create the Housekeeper PydanticAI agent."""
    from src.anthropic.model_selector import get_model

    # Get best available model (Claude if available, else Ollama)
    model = get_model()

    agent: Agent[None, str] = Agent(
        model=model,
        system_prompt=HOUSEKEEPER_SYSTEM_PROMPT,
        retries=2,
    )

    # Register discovery tools
    agent.tool_plain(list_areas)
    agent.tool_plain(list_devices)
    agent.tool_plain(get_device_state)

    # Register control tools
    agent.tool_plain(turn_on)
    agent.tool_plain(turn_off)
    agent.tool_plain(toggle)

    # Register scene tools
    agent.tool_plain(list_scenes)
    agent.tool_plain(activate_scene)

    # Register script tools
    agent.tool_plain(list_scripts)
    agent.tool_plain(run_script)

    # Register automation tools
    agent.tool_plain(list_automations)
    agent.tool_plain(toggle_automation)

    # Register history tools
    agent.tool_plain(get_history)

    from src.anthropic.model_selector import get_model_info
    model_info = get_model_info()
    logger.info(
        "housekeeper_agent_created",
        backend=model_info["backend"],
        model=model_info["model"],
        tool_count=13,
    )

    return agent


def get_housekeeper_agent() -> Agent[None, str]:
    """
    Get the Housekeeper agent instance (lazy initialization).

    Returns:
        PydanticAI Agent configured for home automation tasks
    """
    global _housekeeper_agent
    if _housekeeper_agent is None:
        _housekeeper_agent = _create_housekeeper_agent()
    return _housekeeper_agent


async def run_housekeeper(
    task: str,
    context: str = "",
    message_history: Optional[list[Any]] = None,
) -> str:
    """
    Execute a home automation task with The Housekeeper.

    This is the main entry point for delegating home automation tasks
    to The Housekeeper from Tatlock or other agents.

    Args:
        task: The home automation task or request
        context: Additional context from conversation
        message_history: Optional conversation history

    Returns:
        Results and confirmation of actions

    Example:
        result = await run_housekeeper(
            task="Turn on the living room lights",
            context="It's evening",
        )
    """
    agent = get_housekeeper_agent()

    # Build prompt with context if provided
    prompt = task
    if context:
        prompt = f"Context: {context}\n\nTask: {task}"

    logger.info(
        "housekeeper_task_started",
        task=task[:100],
        has_context=bool(context),
        has_history=bool(message_history),
    )

    try:
        # Use temperature 0.1 for slight exploration
        from pydantic_ai.settings import ModelSettings

        result = await agent.run(
            prompt,
            message_history=message_history,
            model_settings=ModelSettings(temperature=0.1),
        )

        logger.info(
            "housekeeper_task_completed",
            task=task[:50],
            output_length=len(result.output),
        )

        return result.output

    except Exception as e:
        logger.error(
            "housekeeper_task_error",
            task=task[:50],
            error=str(e),
            exc_info=True,
        )
        return f"The Housekeeper encountered an error: {str(e)}"


async def run_housekeeper_stream(
    task: str,
    context: str = "",
    message_history: Optional[list[Any]] = None,
):
    """
    Execute a home automation task with streaming output.

    Yields text deltas as The Housekeeper generates the response.

    Args:
        task: The home automation task or request
        context: Additional context from conversation
        message_history: Optional conversation history

    Yields:
        str: Text deltas from the response

    Example:
        async for delta in run_housekeeper_stream("Turn on the lights"):
            print(delta, end="", flush=True)
    """
    agent = get_housekeeper_agent()

    # Build prompt with context if provided
    prompt = task
    if context:
        prompt = f"Context: {context}\n\nTask: {task}"

    logger.info(
        "housekeeper_stream_started",
        task=task[:100],
    )

    try:
        # Use temperature 0.1 for slight exploration
        from pydantic_ai.settings import ModelSettings

        async with agent.run_stream(
            prompt,
            message_history=message_history,
            model_settings=ModelSettings(temperature=0.1),
        ) as response:
            async for delta in response.stream_text(delta=True):
                yield delta

        logger.info("housekeeper_stream_completed", task=task[:50])

    except Exception as e:
        logger.error(
            "housekeeper_stream_error",
            task=task[:50],
            error=str(e),
            exc_info=True,
        )
        yield f"\n\nThe Housekeeper encountered an error: {str(e)}"