""" Steward agent schemas. Defines the structured output models for Steward's request analysis and capability recommendations. """ from typing import Any, Literal, Optional from pydantic import BaseModel, Field class ConversationContext(BaseModel): """ Contextual information extracted from conversation history. The Steward analyzes the full conversation to identify references to previous topics, helping the Butler maintain context. """ has_previous_context: bool = Field( description="Whether the current request references previous conversation turns" ) relevant_turns: list[int] = Field( default_factory=list, description="0-indexed turn numbers that are relevant to the current request" ) context_summary: str = Field( default="", description="Brief summary of relevant context for the Butler" ) class StewardRecommendation(BaseModel): """ Structured recommendation from Steward's request analysis. This is the output format for the Steward agent, providing: - Which household capabilities are needed - Why those capabilities were chosen - Complexity assessment - Conversation context - Missing capabilities (if any) """ recommended_capabilities: list[str] = Field( description="List of household member names to include (e.g., ['tatlock_core'])" ) reasoning: str = Field( description="Explanation of why these capabilities were recommended" ) estimated_complexity: Literal["simple", "moderate", "complex"] = Field( description="Complexity assessment: simple (1 tool), moderate (2-3 tools), complex (multiple tools/steps)" ) conversation_context: ConversationContext = Field( description="Contextual information from conversation history" ) missing_capabilities: Optional[str] = Field( default=None, description="Description of capabilities that would be helpful but aren't available" ) memory_context: dict[str, Any] = Field( default_factory=dict, description="Pre-fetched user context from memory (profile, preferences)" ) enriched_query: str = Field( default="", description="User query with auto-filled context (location, timezone) when not specified" ) def format_for_butler(self) -> str: """ Format recommendation as a note for the Butler. Returns: Formatted string suitable for prepending to user request """ lines = [] # Header lines.append("📋 Steward's Analysis") lines.append("=" * 40) # Complexity lines.append(f"Complexity: {self.estimated_complexity.upper()}") # Recommended capabilities if self.recommended_capabilities: caps = ", ".join(self.recommended_capabilities) lines.append(f"Recommended tools: {caps}") else: lines.append("Recommended tools: None (conversational response)") # Context summary if self.conversation_context.has_previous_context: lines.append(f"Context: {self.conversation_context.context_summary}") # Missing capabilities warning if self.missing_capabilities: lines.append(f"⚠️ Missing: {self.missing_capabilities}") # Memory context (user profile and preferences) if self.memory_context: profile = self.memory_context.get("profile", {}) preferences = self.memory_context.get("preferences", {}) if profile or preferences: lines.append("-" * 40) lines.append("User Context:") if profile: for key, value in profile.items(): lines.append(f" • {key}: {value}") if preferences: prefs_str = ", ".join(f"{k}={v}" for k, v in preferences.items()) lines.append(f" • preferences: {prefs_str}") # Add delegation instructions when expert agents are recommended delegation_agents = [c for c in self.recommended_capabilities if c in ("biographer", "librarian")] if delegation_agents: lines.append("-" * 40) lines.append("DELEGATION REQUIRED:") for agent in delegation_agents: lines.append(f' Call: delegate_to_{agent}(task="[user request]")') lines.append(f' Or output: [DELEGATE:{agent}] task="[user request]"') lines.append("=" * 40) return "\n".join(lines)