""" Steward agent - First-tier request analyzer. The Steward analyzes incoming requests, identifies relevant household capabilities, and provides focused recommendations to Tatlock (the Butler). This creates a two-tier architecture that prevents cognitive overload. Uses plain text output (not JSON) for reliability. Supports both Claude (preferred) and Ollama (fallback) backends via direct API calls. """ import httpx from typing import Optional from src.anthropic.model_selector import is_claude_available, get_model_info from src.core.config import config from src.core.household_registry import get_household_registry from src.core.logging_config import get_logger logger = get_logger(__name__) # System prompt for plain text recommendations def build_steward_prompt(query: str, conversation_history: list[dict]) -> str: """Build the steward's analysis prompt with query and conversation history.""" # Get available capabilities from registry registry = get_household_registry() capabilities = registry.get_all_capabilities() cap_list = [] for cap in capabilities: cap_list.append( f"• {cap.name} - {cap.description} (domains: {', '.join(cap.domains)})" ) capabilities_text = "\n".join(cap_list) # Format conversation history if present history_text = "" if conversation_history: history_lines = [] for i, msg in enumerate(conversation_history): role = msg.get("role", "unknown") content = msg.get("content", "")[:100] # Truncate long messages history_lines.append(f"{i}. {role}: {content}") history_text = "\n\nCONVERSATION HISTORY:\n" + "\n".join(history_lines) return f"""You are the Steward of the household, advising the Butler (Tatlock) on which capabilities to use. AVAILABLE HOUSEHOLD CAPABILITIES: {capabilities_text} YOUR TASK: Analyze the user's query and recommend which capabilities are needed, with specific delegation instructions. {history_text} USER QUERY: {query} GUIDELINES: - Be conservative - only recommend truly necessary capabilities - Simple greetings/chat → no capabilities needed (conversational response only) - Questions about prior conversation ("what did I say", "what we discussed") → no capabilities (Tatlock has full history) - Math/calculations → tatlock_core - Time/date queries → tatlock_core - PERSONAL MEMORY queries → biographer to recall (ALWAYS use for questions about the user themselves): - "where do I live", "what's my location", "my address" → biographer to recall location - "what's my name", "who am I" → biographer to recall name - "what car do I drive", "my vehicle" → biographer to recall car - "what do you know about me", "what have I told you" → biographer to recall or list_memories - "remember that I...", "store that..." → biographer to store_insight - "forget my...", "delete..." → biographer to forget_memory - "my timezone", "my preferences" → biographer to recall preferences - Web searches, weather, news, current information → librarian with search_web - Read a URL or article → librarian with read_url - Wiki creation ("create a page about X", "add X to wiki") → librarian with smart_create - Wiki updates ("update the page", "add to dossier") → librarian with update - Research queries about TOPICS (not about the user) → librarian with hybrid_search - In-depth research, knowledge synthesis, document lookup → librarian with hybrid_search - If conversation history is relevant, note which previous turns matter - Assess complexity: simple (1 tool), moderate (2-3 tools), complex (multiple steps) RESPOND IN THIS FORMAT: DELEGATE: [capability name] to [action] [specific task] REASON: [why this capability handles the request] COMPLEXITY: [simple/moderate/complex] CONTEXT: [any relevant conversation context, or "none"] EXAMPLES: - "DELEGATE: biographer to recall the user's location" (for "where do I live?") - "DELEGATE: biographer to recall the user's car" (for "what car do I drive?") - "DELEGATE: biographer to list_memories about the user" (for "what do you know about me?") - "DELEGATE: biographer to store_insight about user's pet" (for "remember that I have a dog named Max") - "DELEGATE: librarian to search_web for tomorrow's weather forecast" - "DELEGATE: librarian to create a wiki page about CI/CD pipelines" - "DELEGATE: librarian to hybrid_search for information about Docker networking" - "DELEGATE: librarian to read_url https://example.com/article" - "DELEGATE: tatlock_core to calculate the result" - "DELEGATE: none (conversational response only)" Be specific about what Tatlock should delegate - include the action verb (create, update, search, etc.). Plain text only - no JSON, no special formatting.""" class StewardAgent: """ The Steward - Request analyzer and capability coordinator. Analyzes requests with full conversation context and recommends which household capabilities the Butler should use. Uses plain text output for reliability. Supports both Claude (preferred) and Ollama (fallback) backends via direct API calls. """ def __init__(self): """Initialize Steward with backend selection based on availability.""" # Ollama config (fallback) self.ollama_host = str(config.OLLAMA_HOST).rstrip('/') self.ollama_model = config.OLLAMA_DEFAULT_MODEL # Claude config (preferred) self.claude_model = config.ANTHROPIC_MODEL self._anthropic_client = None # Determine which backend to use self._use_claude = config.PREFER_CLOUD_BACKEND and is_claude_available() self.timeout = 30.0 # 30 second timeout for analysis model_info = get_model_info() logger.info( "steward_agent_created", backend=model_info["backend"], model=model_info["model"], timeout=self.timeout, ) def _get_anthropic_client(self): """Get or create Anthropic client (lazy initialization).""" if self._anthropic_client is None: from anthropic import AsyncAnthropic self._anthropic_client = AsyncAnthropic(api_key=config.ANTHROPIC_API_KEY) return self._anthropic_client async def _call_claude(self, system_prompt: str, user_message: str) -> str: """Call Claude API directly for plain text generation.""" client = self._get_anthropic_client() response = await client.messages.create( model=self.claude_model, max_tokens=1024, system=system_prompt, messages=[{"role": "user", "content": user_message}], temperature=0.3, # Lower = more consistent ) return response.content[0].text.strip() async def _call_ollama(self, prompt: str) -> str: """Call Ollama API directly for plain text generation.""" async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.post( f"{self.ollama_host}/api/generate", json={ "model": self.ollama_model, "prompt": prompt, "stream": False, "options": { "temperature": 0.3, # Lower = more consistent "top_p": 0.9 } } ) response.raise_for_status() result = response.json() return result["response"].strip() async def analyze( self, query: str, conversation_history: Optional[list[dict]] = None ) -> str: """ Analyze query and return plain text recommendation. Uses Claude if available, falls back to Ollama. Args: query: User's query to analyze conversation_history: Previous conversation turns Returns: Plain text analysis from Steward Example: >>> text = await steward.analyze("What's 2 + 2?") >>> print(text) "This requires tatlock_core for mathematical calculations. Complexity: simple." """ history = conversation_history or [] prompt = build_steward_prompt(query, history) backend = "claude" if self._use_claude else "ollama" logger.debug( "steward_calling_llm", backend=backend, query_preview=query[:100], ) try: if self._use_claude: # For Claude, split into system + user message # The prompt contains both, but Claude prefers explicit system analysis_text = await self._call_claude( system_prompt="You are the Steward of the household, advising the Butler (Tatlock) on which capabilities to use. Be concise and specific.", user_message=prompt, ) else: analysis_text = await self._call_ollama(prompt) logger.debug( "steward_analysis_received", backend=backend, text_preview=analysis_text[:150], ) return analysis_text except Exception as e: # If Claude fails, try Ollama as fallback if self._use_claude: logger.warning( "steward_claude_fallback", error=str(e), ) analysis_text = await self._call_ollama(prompt) logger.debug( "steward_analysis_received", backend="ollama_fallback", text_preview=analysis_text[:150], ) return analysis_text raise # Global Steward instance _steward_agent = None def get_steward_agent() -> StewardAgent: """ Get the global Steward agent instance. Returns: StewardAgent instance """ global _steward_agent if _steward_agent is None: _steward_agent = StewardAgent() return _steward_agent