jpmschweitzer/tatlock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: implement Phase 2 two-tier architecture with Steward

Browse Source 
Add comprehensive two-tier architecture where Steward analyzes requests
and Tatlock executes with scoped tools. Includes full infrastructure for
request preprocessing, tool tracking, benchmarking, and streaming.

**Added:**
- Steward agent for request analysis and capability recommendation
- Household Registry for centralized capability management
- Request preprocessing pipeline (Steward → Tatlock flow)
- Tool usage tracking and benchmarking system
- Streaming transparency (Steward reasoning visible in streams)
- Structured logging with operation timing
- Redis benchmark storage with 30-day expiry
- Benchmark analysis CLI tools

**Infrastructure:**
- src/agents/steward/ - Steward agent implementation
- src/agents/tatlock_core/ - Tatlock capability domain
- src/core/preprocessing.py - Request preprocessing pipeline
- src/core/tool_tracking.py - Tool call tracking
- src/core/benchmarks.py - Benchmark recording system
- src/core/household_registry.py - Capability registry
- src/core/startup.py - Application startup coordination
- src/core/logging_config.py - Structured logging setup

**Integration:**
- Responses API uses Steward for Tatlock requests
- Chat Completions wraps Responses API for OpenAI compatibility
- Streaming coordinator supports Steward + Tatlock flow
- Tool scoping per request based on Steward recommendations

**Testing:**
- Integration tests for Steward-Tatlock flow
- Benchmark and registry unit tests
- Steward streaming tests

See PHASE2_PLAN.md and PHASE2_COMPLETE.md for detailed documentation.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-07 15:39:20 +01:00
co-authored by Claude Sonnet 4.5
parent 2577730546
commit 6eed5f4d13
34 changed files with 6362 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/steward/__init__.py
+18
View File
@@ -0,0 +1,18 @@
"""
Steward agent package.
The Steward analyzes incoming requests and recommends relevant household
capabilities, creating a two-tier architecture with the Butler.
"""
from .agent import StewardAgent, get_steward_agent
from .schemas import ConversationContext, StewardRecommendation
from .service import analyze_request, format_steward_note
__all__ = [
"StewardAgent",
"get_steward_agent",
"ConversationContext",
"StewardRecommendation",
"analyze_request",
"format_steward_note",
]
src/agents/steward/agent.py
+165
View File
@@ -0,0 +1,165 @@
"""
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 with Ollama models.
"""
import httpx
from typing import Optional
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.
{history_text}
USER QUERY: {query}
GUIDELINES:
- Be conservative - only recommend truly necessary capabilities
- Simple greetings/chat → no capabilities needed (conversational response only)
- Math/calculations → tatlock_core
- Web searches → tatlock_core
- Time/date queries → tatlock_core
- If conversation history is relevant, note which previous turns matter
- Assess complexity: simple (1 tool), moderate (2-3 tools), complex (multiple steps)
- If capabilities are missing, mention what would be needed
RESPOND WITH 2-3 SENTENCES:
1. Which capabilities (if any) are needed and why
2. Complexity assessment (simple/moderate/complex)
3. Any conversation context or missing capabilities
Use capability names in your response (e.g., "tatlock_core for calculations").
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 with Ollama models.
"""
def __init__(self):
"""Initialize Steward with Ollama model (same as Tatlock for VRAM efficiency)."""
self.ollama_host = str(config.OLLAMA_HOST).rstrip('/')
self.model_name = config.OLLAMA_DEFAULT_MODEL
self.timeout = 30.0 # 30 second timeout for analysis
logger.info(
"steward_agent_created",
ollama_host=self.ollama_host,
model=self.model_name,
timeout=self.timeout,
)
async def analyze(
self,
query: str,
conversation_history: Optional[list[dict]] = None
) -> str:
"""
Analyze query and return plain text recommendation.
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)
logger.debug("steward_calling_ollama", query_preview=query[:100])
# Call Ollama API directly (more reliable than PydanticAI for plain text)
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.post(
f"{self.ollama_host}/api/generate",
json={
"model": self.model_name,
"prompt": prompt,
"stream": False,
"options": {
"temperature": 0.3, # Lower = more consistent
"top_p": 0.9
}
}
)
response.raise_for_status()
result = response.json()
analysis_text = result["response"].strip()
logger.debug(
"steward_analysis_received",
text_preview=analysis_text[:150]
)
return analysis_text
# 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
src/agents/steward/schemas.py
+93
View File
@@ -0,0 +1,93 @@
"""
Steward agent schemas.
Defines the structured output models for Steward's request analysis
and capability recommendations.
"""
from typing import 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"
)
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}")
lines.append("=" * 40)
return "\n".join(lines)
src/agents/steward/service.py
+282
View File
@@ -0,0 +1,282 @@
"""
Steward service layer.
Provides high-level interface for request analysis with logging,
benchmarking, and error handling.
Parses plain text recommendations into structured data.
"""
import re
from typing import Optional
from src.core.benchmarks import PerformanceBenchmark, get_benchmark_store
from src.core.household_registry import get_household_registry
from src.core.logging_config import get_logger, log_operation
from .agent import get_steward_agent
from .schemas import ConversationContext, StewardRecommendation
logger = get_logger(__name__)
def _extract_capabilities(text: str) -> list[str]:
"""
Extract capability names from Steward's text response.
Uses keyword matching to find mentioned capabilities.
Args:
text: Steward's plain text analysis
Returns:
List of capability names (e.g., ['tatlock_core'])
"""
text_lower = text.lower()
registry = get_household_registry()
capabilities = registry.get_all_capabilities()
found_caps = []
for cap in capabilities:
# Check if capability name is mentioned
if cap.name.lower() in text_lower:
found_caps.append(cap.name)
continue
# Check if any domains are mentioned
for domain in cap.domains:
if domain.lower() in text_lower:
found_caps.append(cap.name)
break
return found_caps
def _extract_complexity(text: str) -> str:
"""
Extract complexity assessment from text.
Args:
text: Steward's plain text analysis
Returns:
One of: "simple", "moderate", "complex"
"""
text_lower = text.lower()
if "complex" in text_lower:
return "complex"
elif "moderate" in text_lower:
return "moderate"
else:
return "simple" # Default to simple
def _extract_conversation_context(
text: str,
conversation_history: list[dict]
) -> ConversationContext:
"""
Extract conversation context analysis from text.
Args:
text: Steward's plain text analysis
conversation_history: Previous conversation turns
Returns:
ConversationContext with relevant turn analysis
"""
text_lower = text.lower()
# Check if conversation history is referenced
has_context = bool(conversation_history) and any([
"previous" in text_lower,
"earlier" in text_lower,
"context" in text_lower,
"turn" in text_lower,
"history" in text_lower,
])
# Extract turn numbers if mentioned (e.g., "turn 0", "turn 1")
relevant_turns = []
turn_pattern = r"turn\s+(\d+)"
matches = re.findall(turn_pattern, text_lower)
relevant_turns = [int(m) for m in matches]
# Create summary from relevant portion of text
context_summary = ""
if has_context:
# Extract sentence(s) mentioning context
sentences = text.split('.')
context_sentences = [s for s in sentences if any(
word in s.lower() for word in ["previous", "earlier", "context", "history"]
)]
if context_sentences:
context_summary = context_sentences[0].strip()
return ConversationContext(
has_previous_context=has_context,
relevant_turns=relevant_turns,
context_summary=context_summary
)
def _extract_missing_capabilities(text: str) -> Optional[str]:
"""
Extract missing capability notes from text.
Args:
text: Steward's plain text analysis
Returns:
Description of missing capabilities, or None
"""
text_lower = text.lower()
# Look for indicators of missing capabilities
if any(word in text_lower for word in [
"missing", "unavailable", "not available", "don't have", "doesn't have"
]):
# Find the sentence mentioning missing capabilities
sentences = text.split('.')
for sentence in sentences:
if any(word in sentence.lower() for word in [
"missing", "unavailable", "not available"
]):
return sentence.strip()
return None
async def analyze_request(
user_request: str,
conversation_history: list[dict],
conversation_id: Optional[str] = None,
) -> StewardRecommendation:
"""
Analyze user request with full conversation context.
This is the main entry point for Steward analysis. It:
1. Calls the Steward agent with full conversation history
2. Logs the operation with timing
3. Records performance benchmarks to Redis
4. Returns structured recommendations
Args:
user_request: The current user message to analyze
conversation_history: Full conversation history (all previous turns)
conversation_id: Optional conversation ID for tracking
Returns:
StewardRecommendation with capability recommendations and context analysis
Example:
>>> recommendation = await analyze_request(
... "What's sqrt(144)?",
... conversation_history=[],
... )
>>> print(recommendation.recommended_capabilities)
['tatlock_core']
"""
async with log_operation(
"steward_analysis",
{
"request_preview": user_request[:100],
"conversation_id": conversation_id,
"history_length": len(conversation_history),
}
) as log_ctx:
try:
# Get Steward agent
steward = get_steward_agent()
logger.debug(
"steward_analyzing_request",
request=user_request,
history_turns=len(conversation_history),
)
# Get plain text analysis from Steward
analysis_text = await steward.analyze(
user_request,
conversation_history=conversation_history
)
# Parse plain text into structured recommendation
capabilities = _extract_capabilities(analysis_text)
complexity = _extract_complexity(analysis_text)
context = _extract_conversation_context(analysis_text, conversation_history)
missing = _extract_missing_capabilities(analysis_text)
recommendation = StewardRecommendation(
recommended_capabilities=capabilities,
reasoning=analysis_text,
estimated_complexity=complexity,
conversation_context=context,
missing_capabilities=missing
)
# Update log context with results
log_ctx["recommendation_count"] = len(recommendation.recommended_capabilities)
log_ctx["complexity"] = recommendation.estimated_complexity
log_ctx["has_context"] = recommendation.conversation_context.has_previous_context
log_ctx["missing_capabilities"] = recommendation.missing_capabilities is not None
logger.info(
"steward_analysis_complete",
recommended=recommendation.recommended_capabilities,
complexity=recommendation.estimated_complexity,
reasoning=analysis_text[:200], # First 200 chars
)
# Record performance benchmark
if log_ctx.get("duration_seconds"):
benchmark = PerformanceBenchmark(
operation="steward_analysis",
duration_seconds=log_ctx["duration_seconds"],
success=True,
recommendation_count=len(recommendation.recommended_capabilities),
confidence=None, # Could add confidence scoring in future
conversation_id=conversation_id,
metadata={
"complexity": recommendation.estimated_complexity,
"has_context": recommendation.conversation_context.has_previous_context,
"missing_capabilities": recommendation.missing_capabilities is not None,
},
)
await get_benchmark_store().record(benchmark)
return recommendation
except Exception as e:
logger.error(
"steward_analysis_failed",
error=str(e),
error_type=type(e).__name__,
exc_info=True,
)
raise
async def format_steward_note(recommendation: StewardRecommendation) -> str:
"""
Format Steward's recommendation as a note for the Butler.
This creates a structured message that will be prepended to the user's
request when sent to Tatlock, providing context and guidance.
Args:
recommendation: Steward's analysis and recommendations
Returns:
Formatted note string for the Butler
Example:
>>> note = await format_steward_note(recommendation)
>>> print(note)
📋 Steward's Analysis
========================================
Complexity: SIMPLE
Recommended tools: tatlock_core
========================================
"""
return recommendation.format_for_butler()