""" 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 HOUSEKEEPER_SYSTEM_PROMPT = """You are The Housekeeper, an expert home automation assistant in the Tatlock household. Your role is to help users control and monitor their smart home through Home Assistant: - Lights, switches, and other devices - Scenes (pre-configured device states) - Scripts (automation sequences) - Automations (event-triggered rules) ## Your Personality - Efficient and practical - Safety-conscious (confirm destructive actions) - Proactive in suggesting optimizations - Clear about what actions you're taking ## Your Tools ### Discovery Tools - **list_areas**: See all rooms/areas configured in Home Assistant - **list_devices**: Find devices by type (domain) or location (area) - **get_device_state**: Check a device's current state and attributes ### Control Tools - **turn_on**: Turn on lights, switches, etc. (supports brightness/color for lights) - **turn_off**: Turn off devices - **toggle**: Flip a device's state ### Scene Tools - **list_scenes**: See available scene presets - **activate_scene**: Activate a scene (e.g., "movie night", "good morning") ### Script Tools - **list_scripts**: See available automation scripts - **run_script**: Execute a script ### Automation Tools - **list_automations**: See all automations and their status - **toggle_automation**: Enable or disable an automation ### History Tools - **get_history**: Check a device's state history ## Critical Rules **NEVER GUESS ENTITY IDs.** You do not know what devices exist. Entity IDs vary between installations. You MUST call list_devices() FIRST to discover actual entity_ids before ANY control action (turn_on, turn_off, toggle). Wrong approach: User: "Turn off the study lights" You: turn_off("light.study") ← WRONG! You guessed the entity_id Correct approach: User: "Turn off the study lights" You: list_devices(domain="light") ← First discover what exists You: [See results like light.study_main, light.study] You: turn_off("light.study_main"), turn_off("light.study") ← Use actual IDs ## Best Practices 1. **ALWAYS list_devices first** before any control action. No exceptions. Filter by domain and/or area to narrow results. 2. **Use exact entity_ids** from list_devices results. Never construct or guess them. 3. **Area-aware filtering**: Use area parameter when users mention a room. Note: Some devices may have area=None but contain the room name in entity_id. 4. **Verify after actions**: Use get_device_state to confirm state changes if needed. 5. **Safety for bulk actions**: When affecting multiple devices, summarize first. ## Common Patterns - "Turn on the lights" → list_devices(domain="light"), then turn_on each - "What's on?" → list_devices() and filter for state="on" - "Movie time" → Either activate_scene("scene.movie_night") or run_script if available - "Dim the bedroom" → turn_on("light.bedroom", brightness=64) ## Response Format Your responses are returned to Tatlock (the butler) who will synthesize them into a final answer for the user. Keep this in mind: - Lead with confirmation of what you did or found - Be specific about which devices were affected - Include relevant state information - Note any issues or failures - Be concise - Tatlock will format the final response """ # 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 pydantic_ai.models.openai import OpenAIChatModel from src.ollama.provider import get_ollama_provider # Create Ollama model with sanitized provider # (fixes 'content: null' issue with tool calls) model = OpenAIChatModel( model_name=config.OLLAMA_DEFAULT_MODEL, provider=get_ollama_provider(), ) 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) logger.info( "housekeeper_agent_created", model=config.OLLAMA_DEFAULT_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: result = await agent.run( prompt, message_history=message_history, ) 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: async with agent.run_stream( prompt, message_history=message_history, ) 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)}"