""" Tatlock agent - The Butler (PydanticAI implementation). This is the production Tatlock agent using PydanticAI with Ollama backend. The agent embodies a witty, capable British butler personality. """ import secrets from collections.abc import AsyncGenerator from dataclasses import dataclass, field from typing import Any from pydantic_ai import Agent, RunContext from src.agents.base import AgentInterface, OutputItem from src.agents.tatlock_core.tools import ( calculate, calculate_time_offset, get_current_datetime, time_difference, ) from src.core.logging_config import get_logger from src.core.tracing import ( SpanType, add_tool_spans_from_messages, end_span, start_span, ) logger = get_logger(__name__) @dataclass class ToolCallTracker: """Tracks tool calls for reporting to reasoning output.""" calls: list[str] = field(default_factory=list) def log_call(self, message: str) -> None: """Log a tool call.""" self.calls.append(message) def generate_id() -> str: """Generate unique ID for output items.""" return secrets.token_hex(16) # System prompt defining Tatlock's personality TATLOCK_SYSTEM_PROMPT = """You are Tatlock, a helpful personal assistant with the demeanor of a British butler. ## Personality Address users as "sir". Be confident, direct, and efficient - you are an unflappable English butler who gets things done. Dry wit and puns are encouraged. **CRITICAL - Do NOT:** - Apologize unless you genuinely made an error - Say "Apologies for any confusion" or "Allow me to rectify" when nothing went wrong - Preface successful results with caveats or apologies When presenting findings: lead with the answer, be concise, skip the preamble. You coordinate with various household staff (expert agents) to provide comprehensive assistance across: - Research and knowledge work - Software development - System administration - Home automation - Personal organization ## Research Mindset Approach all questions with a researcher's mindset: - Always verify facts rather than relying solely on memory - When unsure, search for current and accurate information - Cross-check important claims when possible - Acknowledge uncertainty and seek verification - Prefer authoritative sources and current data ## Available Tools You have direct access to several permanent tools that you should USE whenever appropriate: 1. **Calculator** (calculate): For ALL mathematical operations, no matter how simple - Always prefer using the calculator over mental math - Supports arithmetic, algebra, trigonometry, logarithms, and common math functions - Example: "What is 234 * 567?" -> Use calculate("234 * 567") 2. **Date/Time Toolkit**: - get_current_datetime: Get the current date and/or time - calculate_time_offset: Calculate dates relative to now (e.g., "1 week ago", "3 months from now") - time_difference: Calculate the time between two dates - Use these for ANY date/time queries - never guess at dates or times 3. **Web Search** (via Librarian): For current, volatile, or factual information - Delegate to the Librarian for web searches and research - Examples: news, current events, recent developments, specific facts, technical documentation - Use: delegate_to_librarian(task="search the web for ...") ## Tool Usage Guidelines - **Mathematics**: ALWAYS use the calculator tool, even for simple arithmetic - **Dates/Times**: ALWAYS use the date/time tools, never guess or estimate - **Current Information**: Delegate web searches to the Librarian - **Verification**: When facts are important, delegate to Librarian for research - When you use a tool, explain what you're doing in a butler-appropriate manner - Present tool results naturally in your response ## Expert Delegation (CRITICAL) When you see "DELEGATE:" in your instructions, you MUST delegate to the appropriate agent. **PRIMARY METHOD**: Call the delegation function directly: - `delegate_to_librarian(task="...")` for research/wiki tasks - `delegate_to_biographer(task="...")` for memory tasks **FALLBACK METHOD**: If function calling fails, output EXACTLY this format: ``` [DELEGATE:biographer] task="Remember that user's name is TestBot" ``` or ``` [DELEGATE:librarian] task="Search for information about Docker" ``` **Rules:** 1. When you see "DELEGATE: biographer" - delegate to biographer 2. When you see "DELEGATE: librarian" - delegate to librarian 3. NEVER ask for confirmation - just delegate 4. NEVER handle delegated tasks yourself 5. If you cannot call the function, use the [DELEGATE:...] text format EXACTLY """ # Tool-phase prompt for orchestrate_tool_calls(). The butler personality prompt # suppresses tool calling on small local models (gemma4 reasons about the tool, # then answers from memory with wrong arithmetic), so the orchestration phase # uses a terse operator prompt; synthesize_from_results() applies the persona. TATLOCK_ORCHESTRATION_PROMPT = """You are the tool-execution phase of Tatlock, \ a butler assistant. Your only job is to gather accurate results by calling the \ provided tools. - ALWAYS use tools for the task - never answer from memory and never do mental math. - Mathematics: call the calculate tool, even for trivial arithmetic. - Dates and times: call the date/time tools, never guess. - When the instructions say DELEGATE to an agent, call the matching delegate_to_* tool. - After the tool results arrive, reply with a one-line factual summary of the results. \ A later step writes the polished reply, so do not add personality.""" class TatlockAgent(AgentInterface): """ Tatlock - The Butler agent using PydanticAI with Ollama. This is the production implementation of the Tatlock personality, currently in Phase 1 (basic LLM integration without expert agents). """ def __init__(self) -> None: """Initialize Tatlock (lazy agent creation).""" # Deps are a ToolCallTracker: every registered tool takes # RunContext[ToolCallTracker], and run() is called with one. Saying so # is what lets the tool registrations below type-check at all. self._agent: Agent[ToolCallTracker, str] | None = None # Lazy initialization def _ensure_agent(self) -> None: """Ensure the PydanticAI agent is initialized (lazy initialization).""" if self._agent is not None: return from src.anthropic.model_selector import get_model, get_model_info model_info = get_model_info() logger.info( "tatlock_agent_initializing", backend=model_info["backend"], model=model_info["model"], ) # Get best available model (Claude if available, else Ollama) model = get_model() # Create PydanticAI agent self._agent = Agent( model, system_prompt=TATLOCK_SYSTEM_PROMPT, deps_type=ToolCallTracker, ) # Register tools with the agent self._register_tools() def _register_tools(self) -> None: """Register permanent tools with the PydanticAI agent. Called only from _ensure_agent, immediately after the agent is built, so the assert documents an invariant rather than guarding a real case. """ assert self._agent is not None, "_register_tools called before the agent exists" # Calculator tool @self._agent.tool def calculate_math(ctx: RunContext[ToolCallTracker], expression: str) -> str: """ Evaluate mathematical expressions safely. Use this for ALL mathematical calculations, no matter how simple. Args: expression: Mathematical expression (e.g., "2 + 2", "sqrt(16)", "pi * 2") Returns: String result of the calculation """ # Log the calculation to reasoning output if ctx.deps: ctx.deps.log_call(f"🧮 Calculating: {expression}") return calculate(expression) # Current date/time tool @self._agent.tool def get_current_time(ctx: RunContext[ToolCallTracker], format_str: str = "full") -> str: """ Get the current date and time. Args: format_str: Output format ("full", "date", "time", "iso", or custom strftime format) Returns: Formatted current datetime string """ if ctx.deps: ctx.deps.log_call(f"🕐 Getting current time (format: {format_str})") return get_current_datetime(format_str) # Time offset calculator @self._agent.tool def calculate_date_offset(ctx: RunContext[ToolCallTracker], offset_description: str) -> str: """ Calculate a date/time relative to now. Args: offset_description: Natural language time offset (e.g., "1 week ago", "2 days from now") Returns: Formatted datetime string (YYYY-MM-DD HH:MM:SS) """ if ctx.deps: ctx.deps.log_call(f"🕐 Calculating date offset: {offset_description}") return calculate_time_offset(offset_description) # Time difference calculator @self._agent.tool def calculate_time_difference( ctx: RunContext[ToolCallTracker], date1_str: str, date2_str: str = "now" ) -> str: """ Calculate the difference between two dates. Args: date1_str: First date (YYYY-MM-DD or YYYY-MM-DD HH:MM:SS) date2_str: Second date or "now" for current time (default: "now") Returns: Human-readable description of the time difference """ if ctx.deps: ctx.deps.log_call( f"🕐 Calculating time difference between {date1_str} and {date2_str}" ) return time_difference(date1_str, date2_str) # NOTE: Web search has been moved to The Librarian agent. # Use delegate_to_librarian(task="search web for ...") for web search. @property def agent(self): """Get the PydanticAI agent, initializing it if needed.""" self._ensure_agent() return self._agent async def generate_response( self, messages: list[dict], reasoning: dict | None = None, tools: list[dict] | None = None, temperature: float = 1.0, max_tokens: int | None = None, stop: list[str] | None = None, **kwargs: Any, ) -> AsyncGenerator[OutputItem, None]: """ Generate response using PydanticAI with Ollama. Args: messages: Conversation history in OpenAI format reasoning: Reasoning configuration (if requested) tools: Available tools (not yet implemented) temperature: Sampling temperature max_tokens: Maximum tokens to generate stop: Stop sequences **kwargs: Additional parameters Yields: OutputItem: Response items (reasoning, message) """ try: # Convert OpenAI-format messages to PydanticAI format # PydanticAI uses: {"role": "user"/"assistant", "content": "text"} # OpenAI format is the same, so we can use messages directly # Extract the latest user message for the prompt user_message = "" for msg in reversed(messages): if msg.get("role") == "user": user_message = msg.get("content", "") break if not user_message: yield OutputItem( type="message", id=f"msg_{generate_id()}", role="assistant", content=[ { "type": "output_text", "text": "I'm afraid I didn't receive a message, sir. How may I assist you?", "annotations": [], } ], status="completed", ) return # Build message history (all messages except the last user message) # PydanticAI expects history as list of ModelRequest/ModelResponse objects from pydantic_ai.messages import ( ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) message_history: list[ModelMessage] = [] for i, msg in enumerate(messages[:-1]): # All messages except the last one role = msg.get("role") content = msg.get("content", "") # Skip messages with empty content (can cause Ollama errors) if not content or not content.strip(): logger.warning(f"Skipping message {i} with empty content: role={role}") continue # Debug: Check for problematic content if '"' in content or "'" in content: logger.debug( f"Message {i} ({role}) contains quotes. Content preview: {content[:100]}..." ) # Convert to PydanticAI message format try: if role == "user": message_history.append( ModelRequest(parts=[UserPromptPart(content=content)]) ) elif role == "assistant": message_history.append(ModelResponse(parts=[TextPart(content=content)])) except Exception as e: logger.error(f"Error creating message history item {i}: {e}") logger.error(f"Problematic content: {repr(content)}") raise # Debug: Log the message history summary logger.info(f"Built message history with {len(message_history)} messages") if message_history: for i, hist_msg in enumerate(message_history): msg_type = type(hist_msg).__name__ content_preview = ( str(hist_msg.parts[0].content)[:50] if hist_msg.parts else "no parts" ) logger.info(f" History[{i}]: {msg_type} - {content_preview}...") # Generate reasoning output if requested if reasoning and reasoning.get("effort") != "none": yield OutputItem( type="reasoning", id=f"reasoning_{generate_id()}", summary=[ "Analyzing your request, sir...", "Formulating response based on available knowledge...", ], thinking="", # PydanticAI doesn't expose internal reasoning yet status="completed", ) # Create a tool call tracker for this request tracker = ToolCallTracker() # Stream the agent response token-by-token msg_id = f"msg_{generate_id()}" final_text = "" # Use run() instead of run_stream() to avoid GeneratorExit issues # with async context managers inside generators # The StreamingCoordinator will handle word-by-word streaming # Pass message_history to maintain conversation context and tracker for tool logging result = await self.agent.run( user_message, message_history=message_history if message_history else None, deps=tracker, ) final_text = result.output # If tools were called, yield a reasoning item showing what was done if tracker.calls: yield OutputItem( type="reasoning", id=f"reasoning_tools_{generate_id()}", summary=tracker.calls, thinking="", status="completed", ) # Yield the complete message # The StreamingCoordinator will break this into word-by-word deltas yield OutputItem( type="message", id=msg_id, role="assistant", content=[{"type": "output_text", "text": final_text, "annotations": []}], status="completed", ) except Exception as e: logger.error(f"Error generating response: {e}", exc_info=True) yield OutputItem( type="message", id=f"msg_{generate_id()}", role="assistant", content=[ { "type": "output_text", "text": f"My apologies, sir. I encountered an error: {str(e)}", "annotations": [], } ], status="failed", ) async def supports_tools(self) -> bool: """Permanent tools now available.""" return True async def supports_reasoning(self) -> bool: """Basic reasoning support via summary.""" return True async def run_with_scoped_tools( self, user_message: str, steward_note: str, scoped_tools: list[Any], message_history: list[dict], tool_tracker: Any = None, ) -> str: """ Run Tatlock with scoped tools from Steward preprocessing. This is the Phase 2 request flow where the Steward has already analyzed the request and provided scoped tools. Args: user_message: The user's original message steward_note: Note from Steward (prepended to request, invisible to user) scoped_tools: List of tool definitions from household registry message_history: Conversation history in PydanticAI format tool_tracker: Optional tool call tracker for analysis Returns: str: Tatlock's response text Example: >>> response = await tatlock.run_with_scoped_tools( ... "What's sqrt(144)?", ... steward_note="Simple math request...", ... scoped_tools=[calculator_tool, ...], ... message_history=[], ... tool_tracker=tracker, ... ) """ from src.anthropic.model_selector import get_model logger.info( "tatlock_run_with_scoped_tools", user_message_preview=user_message[:100], scoped_tool_count=len(scoped_tools), history_length=len(message_history), ) # Create a fresh agent instance with scoped tools only # This ensures Tatlock can ONLY use tools recommended by the Steward model = get_model() # Create agent with scoped tools # Tools from household registry are already PydanticAI Tool objects scoped_agent = Agent( model, system_prompt=TATLOCK_SYSTEM_PROMPT, tools=scoped_tools, # Pass tools directly to Agent constructor ) # Prepend Steward's note to the request (invisible to user, visible to Tatlock) enriched_message = f"{steward_note}\n\n{user_message}" # Convert message history to PydanticAI format from pydantic_ai.messages import ( ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) pydantic_history: list[ModelMessage] = [] for msg in message_history: role = msg.get("role") content = msg.get("content", "") if not content or not content.strip(): continue if role == "user": pydantic_history.append(ModelRequest(parts=[UserPromptPart(content=content)])) elif role == "assistant": pydantic_history.append(ModelResponse(parts=[TextPart(content=content)])) # Run with scoped tools and tracker # Force tool_choice to make LLM actually call tools from src.anthropic.model_selector import get_tool_choice_settings, with_slot_pinning result = await scoped_agent.run( enriched_message, message_history=pydantic_history if pydantic_history else None, deps=tool_tracker, model_settings=with_slot_pinning(get_tool_choice_settings(), slot=1), ) logger.info( "tatlock_response_generated", response_preview=result.output[:100], ) return result.output async def run_with_scoped_tools_stream( self, user_message: str, steward_note: str, scoped_tools: list, message_history: list[dict], tool_tracker: "ToolCallTracker", ): """ Run Tatlock with scoped tools recommended by Steward (streaming version). This is the Phase 2 execution flow where Steward has preprocessed the request and provided: - steward_note: Instructions for Tatlock (invisible to user) - scoped_tools: Only the tools Steward recommended Args: user_message: Original user message steward_note: Steward's instructions for Tatlock scoped_tools: List of PydanticAI Tool objects to use message_history: Previous conversation turns tool_tracker: Tracker for tool call analytics Yields: Text chunks from the streaming response """ from src.anthropic.model_selector import get_model logger.info( "tatlock_run_with_scoped_tools_stream", user_message_preview=user_message[:100], scoped_tool_count=len(scoped_tools), history_length=len(message_history), ) # Create a fresh agent instance with scoped tools only model = get_model() # Create agent with scoped tools scoped_agent = Agent( model, system_prompt=TATLOCK_SYSTEM_PROMPT, tools=scoped_tools, ) # Prepend Steward's note to the request enriched_message = f"{steward_note}\n\n{user_message}" # Convert message history to PydanticAI format from pydantic_ai.messages import ( ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) pydantic_history: list[ModelMessage] = [] for msg in message_history: role = msg.get("role") content = msg.get("content", "") if not content or not content.strip(): continue if role == "user": pydantic_history.append(ModelRequest(parts=[UserPromptPart(content=content)])) elif role == "assistant": pydantic_history.append(ModelResponse(parts=[TextPart(content=content)])) # Use run() instead of run_stream() to avoid Ollama 400 bug # with streaming + tool calls (PydanticAI issues #1292, #2256) # We yield the final response in chunks to maintain streaming interface result = await scoped_agent.run( enriched_message, message_history=pydantic_history if pydantic_history else None, deps=tool_tracker, ) # Stream the final response in chunks to maintain UX response_text = result.output chunk_size = 50 # characters per chunk for i in range(0, len(response_text), chunk_size): yield response_text[i : i + chunk_size] logger.info("tatlock_scoped_run_complete") async def orchestrate_tool_calls( self, user_message: str, steward_note: str, scoped_tools: list[Any], message_history: list[dict], tool_tracker: Any = None, ) -> dict[str, Any]: """ Phase 1: Execute tool calls and delegations, return structured results. This is the coordination phase where Tatlock orchestrates tool calls and expert delegations. The raw output is captured for Phase 2 synthesis. Args: user_message: The user's original message steward_note: Note from Steward (invisible to user) scoped_tools: List of tool definitions from household registry message_history: Conversation history tool_tracker: Optional tool call tracker for analysis Returns: dict with: - tools_called: List of tool names that were called - expert_results: Dict mapping expert names to their outputs - tool_outputs: Dict mapping tool names to their outputs - raw_output: The agent's raw text output """ from pydantic_ai.messages import ( ModelMessage, ModelRequest, ModelResponse, TextPart, ToolCallPart, ToolReturnPart, UserPromptPart, ) from src.anthropic.model_selector import get_model logger.info( "tatlock_orchestrate_tool_calls", user_message_preview=user_message[:100], scoped_tool_count=len(scoped_tools), history_length=len(message_history), ) # Start tracing span for orchestration phase orchestrate_span = start_span( "tatlock_orchestrate", SpanType.TATLOCK, metadata={ "scoped_tool_count": len(scoped_tools), "tool_names": [getattr(t, "__name__", str(t)) for t in scoped_tools[:5]], }, ) # Create a fresh agent instance with scoped tools only model = get_model() # Create agent with scoped tools, using the tool-phase prompt scoped_agent = Agent( model, system_prompt=TATLOCK_ORCHESTRATION_PROMPT, tools=scoped_tools, ) # Prepend Steward's note to the request enriched_message = f"{steward_note}\n\n{user_message}" # Convert message history to PydanticAI format pydantic_history: list[ModelMessage] = [] for msg in message_history: role = msg.get("role") content = msg.get("content", "") if not content or not content.strip(): continue if role == "user": pydantic_history.append(ModelRequest(parts=[UserPromptPart(content=content)])) elif role == "assistant": pydantic_history.append(ModelResponse(parts=[TextPart(content=content)])) # Run with scoped tools and tracker from src.anthropic.model_selector import get_tool_choice_settings, with_slot_pinning result = await scoped_agent.run( enriched_message, message_history=pydantic_history if pydantic_history else None, deps=tool_tracker, model_settings=with_slot_pinning(get_tool_choice_settings(), slot=1), ) # Extract tool calls and results from the agent's messages tools_called = [] expert_results: dict[str, Any] = {} tool_outputs: dict[str, Any] = {} # Parse through new messages to find tool calls and returns for msg in result.new_messages(): if isinstance(msg, ModelResponse): for part in msg.parts: if isinstance(part, ToolCallPart): tools_called.append(part.tool_name) elif isinstance(msg, ModelRequest): for part in msg.parts: if isinstance(part, ToolReturnPart): tool_name = part.tool_name content = part.content # Categorize as expert result or tool output if tool_name.startswith("delegate_to_"): expert_name = tool_name.replace("delegate_to_", "") expert_results[expert_name] = content else: tool_outputs[tool_name] = content logger.info( "tatlock_orchestration_complete", tools_called=tools_called, expert_count=len(expert_results), tool_output_count=len(tool_outputs), ) # Add tool-level spans from result messages if orchestrate_span: add_tool_spans_from_messages(result.new_messages(), orchestrate_span) # End orchestration span with results end_span( orchestrate_span, metadata_update={ "tools_called": tools_called, "expert_count": len(expert_results), "tool_output_count": len(tool_outputs), }, details_update={ "steward_note_preview": steward_note[:500] if steward_note else None, }, ) return { "tools_called": tools_called, "expert_results": expert_results, "tool_outputs": tool_outputs, "raw_output": result.output, } async def synthesize_from_results( self, user_message: str, orchestration_results: dict[str, Any], message_history: list[dict], ) -> str: """ Phase 2: Synthesize butler-toned response from gathered results. This is the synthesis phase where Tatlock takes the coordination results and produces a properly butler-toned response. Args: user_message: The user's original message orchestration_results: Results from orchestrate_tool_calls() message_history: Conversation history Returns: str: Butler-toned response synthesized from all results """ from pydantic_ai.messages import ( ModelMessage, ModelRequest, ModelResponse, TextPart, UserPromptPart, ) from src.anthropic.model_selector import get_model logger.info( "tatlock_synthesize_from_results", user_message_preview=user_message[:100], expert_count=len(orchestration_results.get("expert_results", {})), tool_count=len(orchestration_results.get("tool_outputs", {})), ) # Start tracing span for synthesis phase synthesize_span = start_span( "tatlock_synthesize", SpanType.TATLOCK, metadata={ "expert_count": len(orchestration_results.get("expert_results", {})), "tool_output_count": len(orchestration_results.get("tool_outputs", {})), }, ) # Build synthesis prompt with all available information synthesis_parts = [] synthesis_parts.append(f"The user asked: {user_message}") synthesis_parts.append("") # Add expert findings if any if orchestration_results.get("expert_results"): synthesis_parts.append("Expert findings:") for expert, result in orchestration_results["expert_results"].items(): synthesis_parts.append(f"- {expert.title()}: {result}") synthesis_parts.append("") # Add tool outputs if any if orchestration_results.get("tool_outputs"): synthesis_parts.append("Tool results:") for tool, result in orchestration_results["tool_outputs"].items(): synthesis_parts.append(f"- {tool}: {result}") synthesis_parts.append("") synthesis_parts.append( "Synthesize a response for the user. Be direct and confident. " "Lead with the answer - no apologies, no caveats, no 'mix-ups'. " "Address them as 'sir', be concise, add dry wit if appropriate." ) synthesis_prompt = "\n".join(synthesis_parts) # Create synthesis agent (no tools needed) model = get_model() # Synthesis agent uses butler prompt but no tools synthesis_agent = Agent( model, system_prompt=TATLOCK_SYSTEM_PROMPT, # No tools for synthesis phase ) # Convert message history to PydanticAI format pydantic_history: list[ModelMessage] = [] for msg in message_history: role = msg.get("role") content = msg.get("content", "") if not content or not content.strip(): continue if role == "user": pydantic_history.append(ModelRequest(parts=[UserPromptPart(content=content)])) elif role == "assistant": pydantic_history.append(ModelResponse(parts=[TextPart(content=content)])) # Run synthesis from src.anthropic.model_selector import with_slot_pinning result = await synthesis_agent.run( synthesis_prompt, message_history=pydantic_history if pydantic_history else None, model_settings=with_slot_pinning(None, slot=2), ) logger.info( "tatlock_synthesis_complete", response_preview=result.output[:100], ) # End synthesis span with result end_span( synthesize_span, metadata_update={ "response_length": len(result.output), }, details_update={ "synthesis_prompt": synthesis_prompt[:1000], "response_preview": result.output[:500], }, ) return result.output async def get_capabilities(self) -> dict: """Return current capabilities.""" return { "streaming": True, # Streaming implemented "reasoning": True, # Basic reasoning summaries "tools": True, # Permanent tools: calculator, date/time, search "vision": False, # Future "audio": False, # Future }