diff --git a/src/agents/tatlock.py b/src/agents/tatlock.py index b18f657..e26a945 100644 --- a/src/agents/tatlock.py +++ b/src/agents/tatlock.py @@ -5,7 +5,6 @@ This is the production Tatlock agent using PydanticAI with Ollama backend. The agent embodies a witty, capable British butler personality. """ -import logging import secrets from typing import AsyncGenerator, Any from dataclasses import dataclass, field @@ -13,7 +12,7 @@ from dataclasses import dataclass, field from pydantic_ai import Agent, RunContext from src.agents.base import AgentInterface, OutputItem -from src.agents.tools import ( +from src.agents.tatlock_core.tools import ( calculate, get_current_datetime, calculate_time_offset, @@ -21,8 +20,9 @@ from src.agents.tools import ( search_web, ) from src.core.config import config +from src.core.logging_config import get_logger -logger = logging.getLogger(__name__) +logger = get_logger(__name__) @dataclass @@ -115,7 +115,11 @@ class TatlockAgent(AgentInterface): if self._agent is not None: return - logger.info(f"Initializing Tatlock agent with Ollama at {self.ollama_host}, model: {self.model_name}") + logger.info( + "tatlock_agent_initializing", + ollama_host=self.ollama_host, + model=self.model_name, + ) # Import required classes for Ollama configuration from pydantic_ai.models.openai import OpenAIChatModel @@ -410,6 +414,190 @@ class TatlockAgent(AgentInterface): """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 benchmarking + + 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 pydantic_ai.models.openai import OpenAIChatModel + from pydantic_ai.providers.ollama import OllamaProvider + + 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 + clean_host = self.ollama_host.rstrip('/') + base_url = f"{clean_host}/v1" + + ollama_model = OpenAIChatModel( + model_name=self.model_name, + provider=OllamaProvider(base_url=base_url) + ) + + # Create agent with scoped tools + # Tools from household registry are already PydanticAI Tool objects + scoped_agent = Agent( + ollama_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 ModelRequest, ModelResponse, UserPromptPart, TextPart + + pydantic_history = [] + 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 + result = await scoped_agent.run( + enriched_message, + message_history=pydantic_history if pydantic_history else None, + deps=tool_tracker + ) + + 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 pydantic_ai.models.openai import OpenAIChatModel + from pydantic_ai.providers.ollama import OllamaProvider + + 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 + clean_host = self.ollama_host.rstrip('/') + base_url = f"{clean_host}/v1" + + ollama_model = OpenAIChatModel( + model_name=self.model_name, + provider=OllamaProvider(base_url=base_url) + ) + + # Create agent with scoped tools + scoped_agent = Agent( + ollama_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 ModelRequest, ModelResponse, UserPromptPart, TextPart + + pydantic_history = [] + 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)]) + ) + + # Stream with scoped tools and tracker + async with scoped_agent.run_stream( + enriched_message, + message_history=pydantic_history if pydantic_history else None, + deps=tool_tracker + ) as stream: + async for chunk in stream.stream_text(delta=True): + yield chunk + + logger.info("tatlock_stream_complete") + async def get_capabilities(self) -> dict: """Return current capabilities.""" return { diff --git a/src/responses/streaming.py b/src/responses/streaming.py index c09a497..beaadb8 100644 --- a/src/responses/streaming.py +++ b/src/responses/streaming.py @@ -113,6 +113,134 @@ class StreamingCoordinator: 5. Final response event """ + async def stream_response_with_steward( + self, + request: "ResponseRequest" # type: ignore # Forward reference + ) -> AsyncGenerator[StreamEvent, None]: + """ + Stream response with Steward preprocessing (Phase 2 flow). + + Streams in order: + 1. Steward's analysis as reasoning summary + 2. Tatlock's response as output text + + Args: + request: Response request + + Yields: + StreamEvent: Stream of SSE events + """ + from src.responses.service import _calculate_usage, generate_id, _conversation_history + from src.core.preprocessing import preprocess_request + from src.core.tool_tracking import ToolCallTracker + from src.responses.schemas import MessageOutputItem, ReasoningOutputItem, OutputTextContent + from src.agents.tatlock import TatlockAgent + import asyncio + + output_items = [] + + try: + # Get or generate conversation ID + conversation_id = await _conversation_history.get_conversation_id(request) + + # Extract user message and conversation history + user_message = "" + for msg in reversed(request.input): + if msg.get("role") == "user": + user_message = msg.get("content", "") + break + + conversation_history = request.input[:-1] if len(request.input) > 1 else [] + + # Phase 1: Steward preprocessing + enriched = await preprocess_request( + user_message, + conversation_history=conversation_history, + conversation_id=conversation_id, + ) + + # Stream Steward's analysis as reasoning summary + steward_lines = enriched.steward_reasoning.split('\n') + for line in steward_lines: + if line.strip(): + yield ReasoningSummaryDelta(delta=line + "\n") + await asyncio.sleep(0.05) + + yield ReasoningSummaryDone() + + # Add Steward reasoning to output items + reasoning_item = ReasoningOutputItem( + id=f"reasoning_{generate_id()}", + summary=[ + "🎩 Steward's Analysis:", + enriched.steward_reasoning, + ], + status="completed" + ) + output_items.append(reasoning_item) + + # Phase 2: Initialize tool tracker + tracker = ToolCallTracker( + recommended_capabilities=enriched.recommendation.recommended_capabilities, + conversation_id=conversation_id, + ) + + # Phase 3: Stream Tatlock's response with scoped tools + tatlock = TatlockAgent() + tatlock_response_parts = [] + + async for chunk in tatlock.run_with_scoped_tools_stream( + user_message=user_message, + steward_note=enriched.steward_note, + scoped_tools=enriched.scoped_tools, + message_history=conversation_history, + tool_tracker=tracker, + ): + tatlock_response_parts.append(chunk) + yield OutputTextDelta(delta=chunk) + + yield OutputTextDone() + + # Combine response for output item + tatlock_response = "".join(tatlock_response_parts) + + # Add Tatlock message to output items + message_item = MessageOutputItem( + id=f"msg_{generate_id()}", + role="assistant", + content=[OutputTextContent( + type="output_text", + text=tatlock_response, + annotations=[] + )], + status="completed" + ) + output_items.append(message_item) + + # Phase 4: Finalize tool tracking + await tracker.finalize() + + # Calculate usage and build final response + usage = _calculate_usage(request.input, output_items) + + final_response = Response( + id=f"resp_{generate_id()}", + created_at=int(time.time()), + model=request.model, + status="completed", + output=output_items, + usage=usage + ) + + # Track conversation history + await _conversation_history.add_response(conversation_id, final_response) + + yield ResponseDone(response=final_response) + + except Exception as e: + # Stream error event + yield self._create_error_event(e) + async def stream_response( self, request: "ResponseRequest" # type: ignore # Forward reference