feat: add Phase F.2 - The Biographer (memory agent)
Add The Biographer household member for user memory management: Memory Service (direct access layer): - src/core/memory_service.py for fast, LLM-free lookups - Profile, preference, and fact management - Session context with Redis caching - Steward integration via prefetch_context() The Biographer Agent: - src/agents/biographer/ package with PydanticAI agent - Discreet chronicler personality for privacy - Tools: recall_semantic, list_memories, store_insight, update_profile, update_preference, forget_memory - Registered with Household Registry on startup Steward Integration: - Memory context pre-fetch during analysis - Profile/preferences included in Butler note - Keyword-based context determination Also includes: - delegate_to_biographer() wrapper - 34 new tests (capability + memory service) - Version bump to 1.2.0 Documentation cleanup: - Removed obsolete PHASE2_COMPLETE.md, PHASE2_PLAN.md - Removed docs/library-desk-requirements.md - Moved ORCHESTRATION_SCENARIOS.md to project root 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.5
@@ -0,0 +1,34 @@
|
||||
"""
|
||||
The Biographer - Expert for recording and recalling the user's story.
|
||||
|
||||
The Biographer serves as the household's memory keeper, responsible for:
|
||||
- Recording and recalling facts about the user's life
|
||||
- Storing personal information, preferences, and insights
|
||||
- Answering questions like "What car do I drive?", "Where do I work?"
|
||||
- Managing what the household knows and remembers
|
||||
|
||||
For direct key-based lookups (location, timezone, preferences),
|
||||
use the memory_service instead - it's faster and doesn't require LLM.
|
||||
The Biographer handles semantic, fuzzy queries.
|
||||
"""
|
||||
from src.agents.biographer.agent import (
|
||||
get_biographer_agent,
|
||||
run_biographer,
|
||||
run_biographer_stream,
|
||||
)
|
||||
from src.agents.biographer.capability import (
|
||||
BIOGRAPHER_CAPABILITY,
|
||||
get_biographer_capability,
|
||||
register_biographer,
|
||||
unregister_biographer,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"BIOGRAPHER_CAPABILITY",
|
||||
"get_biographer_capability",
|
||||
"get_biographer_agent",
|
||||
"register_biographer",
|
||||
"unregister_biographer",
|
||||
"run_biographer",
|
||||
"run_biographer_stream",
|
||||
]
|
||||
@@ -0,0 +1,273 @@
|
||||
"""
|
||||
The Biographer - Expert for recording and recalling the user's story.
|
||||
|
||||
A PydanticAI agent that serves as the household's memory keeper:
|
||||
- Records facts about the user's life, work, and preferences
|
||||
- Recalls information semantically ("What car do I drive?")
|
||||
- Manages user profile and preferences
|
||||
- Forgets information when requested
|
||||
"""
|
||||
from typing import Any, Optional
|
||||
|
||||
from pydantic_ai import Agent
|
||||
|
||||
from src.agents.biographer.tools import (
|
||||
forget_memory,
|
||||
list_memories,
|
||||
recall_semantic,
|
||||
store_insight,
|
||||
update_preference,
|
||||
update_profile,
|
||||
)
|
||||
from src.core.config import config
|
||||
from src.core.logging_config import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
# The Biographer's system prompt
|
||||
BIOGRAPHER_SYSTEM_PROMPT = """You are The Biographer, the household's memory keeper in the Tatlock estate.
|
||||
|
||||
Your role is to record, recall, and manage the story of the user's life:
|
||||
- Personal facts (vehicle, pets, family members, hobbies, interests)
|
||||
- Life details (employer, occupation, significant events)
|
||||
- Profile information (name, location, timezone)
|
||||
- Preferences (units, theme, communication style)
|
||||
|
||||
## Your Character
|
||||
|
||||
You are a discreet and attentive chronicler. Like a personal biographer who has been
|
||||
with the household for years, you:
|
||||
- Listen carefully and remember important details
|
||||
- Recall information accurately when asked
|
||||
- Never gossip or volunteer unnecessary information
|
||||
- Respect privacy absolutely
|
||||
- Acknowledge when you don't know something rather than guessing
|
||||
|
||||
## Your Tools
|
||||
|
||||
### Recalling the Story
|
||||
- **recall_semantic**: Your primary tool for answering questions about the user
|
||||
- "What car do I drive?" → searches for car-related memories
|
||||
- "Where do I work?" → finds employment information
|
||||
- Finds relevant memories even without exact keywords
|
||||
- **list_memories**: Browse all recorded memories of a type
|
||||
- Use when user asks "What do you know about me?"
|
||||
- Shows everything you've recorded
|
||||
|
||||
### Recording New Details
|
||||
- **store_insight**: Record new facts from conversation
|
||||
- User says "My car is a Tesla" → store_insight("car", "Tesla Model 3")
|
||||
- User says "I work at Acme" → store_insight("employer", "Acme Corp")
|
||||
- Use for facts that don't fit standard profile fields
|
||||
- **update_profile**: Update core biographical fields
|
||||
- name, location, timezone only
|
||||
- "I live in Amsterdam" → update_profile("location", "Amsterdam")
|
||||
- **update_preference**: Record user preferences
|
||||
- temperature_unit, distance_unit, theme, etc.
|
||||
- "Use Celsius please" → update_preference("temperature_unit", "celsius")
|
||||
|
||||
### Managing Records
|
||||
- **forget_memory**: Remove specific records
|
||||
- User asks to forget something → honor immediately
|
||||
- Information becomes outdated → remove it
|
||||
|
||||
## Guidelines
|
||||
|
||||
### What to Record
|
||||
- Explicit statements: "I drive a Tesla", "My wife is Sarah"
|
||||
- Corrections: "Actually, I moved to Berlin"
|
||||
- Preferences: "I prefer metric units"
|
||||
|
||||
### What NOT to Record
|
||||
- Sensitive data: passwords, financial details, health information
|
||||
- Temporary information: "I'm tired today"
|
||||
- Speculation or assumptions
|
||||
|
||||
### Responding to Tatlock
|
||||
Your responses go to Tatlock (the butler) who synthesizes the final answer. Be:
|
||||
- Direct and factual
|
||||
- Clear about what you found or didn't find
|
||||
- Structured for easy integration with other responses
|
||||
|
||||
When you don't have information:
|
||||
"I have no record of the user's [topic]. Would you like me to record this information?"
|
||||
|
||||
When recalling:
|
||||
"According to my records, [information]. This was recorded [source/when if available]."
|
||||
"""
|
||||
|
||||
# Lazy initialization to avoid connection issues during imports
|
||||
_biographer_agent: Optional[Agent[None, str]] = None
|
||||
|
||||
|
||||
def _create_biographer_agent() -> Agent[None, str]:
|
||||
"""Create The Biographer PydanticAI agent."""
|
||||
# Import required classes for Ollama configuration
|
||||
from pydantic_ai.models.openai import OpenAIChatModel
|
||||
from pydantic_ai.providers.ollama import OllamaProvider
|
||||
|
||||
# PydanticAI expects Ollama base URL to end with /v1
|
||||
clean_host = str(config.OLLAMA_HOST).rstrip('/')
|
||||
base_url = f"{clean_host}/v1"
|
||||
|
||||
# Create Ollama model with provider
|
||||
model = OpenAIChatModel(
|
||||
model_name=config.OLLAMA_DEFAULT_MODEL,
|
||||
provider=OllamaProvider(base_url=base_url)
|
||||
)
|
||||
|
||||
agent: Agent[None, str] = Agent(
|
||||
model=model,
|
||||
system_prompt=BIOGRAPHER_SYSTEM_PROMPT,
|
||||
retries=2,
|
||||
)
|
||||
|
||||
# Register recall tools
|
||||
agent.tool_plain(recall_semantic)
|
||||
agent.tool_plain(list_memories)
|
||||
|
||||
# Register recording tools
|
||||
agent.tool_plain(store_insight)
|
||||
agent.tool_plain(update_profile)
|
||||
agent.tool_plain(update_preference)
|
||||
|
||||
# Register management tools
|
||||
agent.tool_plain(forget_memory)
|
||||
|
||||
logger.info(
|
||||
"biographer_agent_created",
|
||||
model=config.OLLAMA_DEFAULT_MODEL,
|
||||
tool_count=6,
|
||||
)
|
||||
|
||||
return agent
|
||||
|
||||
|
||||
def get_biographer_agent() -> Agent[None, str]:
|
||||
"""
|
||||
Get The Biographer agent instance (lazy initialization).
|
||||
|
||||
Returns:
|
||||
PydanticAI Agent configured for memory tasks
|
||||
"""
|
||||
global _biographer_agent
|
||||
if _biographer_agent is None:
|
||||
_biographer_agent = _create_biographer_agent()
|
||||
return _biographer_agent
|
||||
|
||||
|
||||
async def run_biographer(
|
||||
task: str,
|
||||
context: str = "",
|
||||
message_history: Optional[list[Any]] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Execute a memory task with The Biographer.
|
||||
|
||||
This is the main entry point for delegating memory tasks
|
||||
from Tatlock or other agents.
|
||||
|
||||
Args:
|
||||
task: The memory task or question
|
||||
context: Additional context from conversation
|
||||
message_history: Optional conversation history
|
||||
|
||||
Returns:
|
||||
Memory results or confirmation
|
||||
|
||||
Example:
|
||||
result = await run_biographer(
|
||||
task="What car do I drive?",
|
||||
context="User is asking about their vehicle",
|
||||
)
|
||||
"""
|
||||
agent = get_biographer_agent()
|
||||
|
||||
# Build prompt with context if provided
|
||||
prompt = task
|
||||
if context:
|
||||
prompt = f"Context: {context}\n\nTask: {task}"
|
||||
|
||||
logger.info(
|
||||
"biographer_task_started",
|
||||
task=task[:100],
|
||||
has_context=bool(context),
|
||||
has_history=bool(message_history),
|
||||
)
|
||||
|
||||
try:
|
||||
result = await agent.run(
|
||||
prompt,
|
||||
message_history=message_history,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"biographer_task_completed",
|
||||
task=task[:50],
|
||||
output_length=len(result.output),
|
||||
)
|
||||
|
||||
return result.output
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"biographer_task_error",
|
||||
task=task[:50],
|
||||
error=str(e),
|
||||
exc_info=True,
|
||||
)
|
||||
return f"The Biographer encountered an error: {str(e)}"
|
||||
|
||||
|
||||
async def run_biographer_stream(
|
||||
task: str,
|
||||
context: str = "",
|
||||
message_history: Optional[list[Any]] = None,
|
||||
):
|
||||
"""
|
||||
Execute a memory task with streaming output.
|
||||
|
||||
Yields text deltas as The Biographer generates the response.
|
||||
|
||||
Args:
|
||||
task: The memory task or question
|
||||
context: Additional context from conversation
|
||||
message_history: Optional conversation history
|
||||
|
||||
Yields:
|
||||
str: Text deltas from the response
|
||||
|
||||
Example:
|
||||
async for delta in run_biographer_stream("What do you know about me?"):
|
||||
print(delta, end="", flush=True)
|
||||
"""
|
||||
agent = get_biographer_agent()
|
||||
|
||||
# Build prompt with context if provided
|
||||
prompt = task
|
||||
if context:
|
||||
prompt = f"Context: {context}\n\nTask: {task}"
|
||||
|
||||
logger.info(
|
||||
"biographer_stream_started",
|
||||
task=task[:100],
|
||||
)
|
||||
|
||||
try:
|
||||
async with agent.run_stream(
|
||||
prompt,
|
||||
message_history=message_history,
|
||||
) as response:
|
||||
async for delta in response.stream_text(delta=True):
|
||||
yield delta
|
||||
|
||||
logger.info("biographer_stream_completed", task=task[:50])
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"biographer_stream_error",
|
||||
task=task[:50],
|
||||
error=str(e),
|
||||
exc_info=True,
|
||||
)
|
||||
yield f"\n\nThe Biographer encountered an error: {str(e)}"
|
||||
@@ -0,0 +1,88 @@
|
||||
"""
|
||||
Biographer capability registration for the Household Registry.
|
||||
|
||||
Defines The Biographer's capabilities and registers it as a
|
||||
household member for coordination by the Steward and Tatlock.
|
||||
"""
|
||||
from src.agents.biographer.agent import get_biographer_agent
|
||||
from src.agents.biographer.tools import BIOGRAPHER_TOOLS
|
||||
from src.core.household_registry import (
|
||||
HouseholdCapability,
|
||||
get_household_registry,
|
||||
)
|
||||
from src.core.logging_config import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# The Biographer's capability summary for Steward coordination
|
||||
BIOGRAPHER_CAPABILITY = HouseholdCapability(
|
||||
name="biographer",
|
||||
role="The Biographer",
|
||||
category="context",
|
||||
description=(
|
||||
"Memory keeper for the user's story: can RECALL personal facts "
|
||||
"(car, job, family, pets), RECORD new information learned from "
|
||||
"conversation, UPDATE profile (name, location, timezone) and "
|
||||
"preferences (units, theme), and FORGET information when requested. "
|
||||
"Use for: 'what car do I drive?', 'remember that I...', "
|
||||
"'forget my...', 'what do you know about me?'"
|
||||
),
|
||||
domains=[
|
||||
"remember",
|
||||
"recall",
|
||||
"forget",
|
||||
"memory",
|
||||
"preferences",
|
||||
"profile",
|
||||
"personal",
|
||||
"know",
|
||||
"about me",
|
||||
"my",
|
||||
],
|
||||
cost="low", # Mostly vector search, minimal LLM
|
||||
requires_network=False, # All local (Qdrant, Redis)
|
||||
)
|
||||
|
||||
|
||||
def get_biographer_capability() -> HouseholdCapability:
|
||||
"""Get The Biographer's capability definition."""
|
||||
return BIOGRAPHER_CAPABILITY
|
||||
|
||||
|
||||
def register_biographer() -> None:
|
||||
"""
|
||||
Register The Biographer with the Household Registry.
|
||||
|
||||
This makes The Biographer available for:
|
||||
- Steward recommendations (via capability summary)
|
||||
- Tatlock delegation (via agent reference)
|
||||
- Tool scoping (via tool list)
|
||||
"""
|
||||
registry = get_household_registry()
|
||||
|
||||
# Check if already registered
|
||||
if "biographer" in registry:
|
||||
logger.debug("biographer_already_registered")
|
||||
return
|
||||
|
||||
registry.register(
|
||||
name="biographer",
|
||||
capability=BIOGRAPHER_CAPABILITY,
|
||||
tools=BIOGRAPHER_TOOLS,
|
||||
agent=get_biographer_agent(),
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"biographer_registered",
|
||||
role=BIOGRAPHER_CAPABILITY.role,
|
||||
domains=BIOGRAPHER_CAPABILITY.domains,
|
||||
tool_count=len(BIOGRAPHER_TOOLS),
|
||||
)
|
||||
|
||||
|
||||
def unregister_biographer() -> None:
|
||||
"""Unregister The Biographer from the Household Registry."""
|
||||
registry = get_household_registry()
|
||||
registry.unregister("biographer")
|
||||
logger.info("biographer_unregistered")
|
||||
@@ -0,0 +1,462 @@
|
||||
"""
|
||||
Biographer tools for PydanticAI agent.
|
||||
|
||||
These tools enable The Biographer to record and recall the user's story:
|
||||
- recall_semantic: Find memories by meaning/concept
|
||||
- store_insight: Record new facts about the user
|
||||
- list_memories: Browse recorded memories by type
|
||||
- forget_memory: Remove specific memories
|
||||
|
||||
For direct key-based access (get/set profile, preferences),
|
||||
use memory_service directly - these tools are for semantic queries.
|
||||
"""
|
||||
from src.core.context import get_user
|
||||
from src.core.embeddings import get_embedding_client
|
||||
from src.core.logging_config import get_logger
|
||||
from src.core.memory_service import MemoryType, memory_service
|
||||
from src.core.qdrant import get_qdrant_client
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Semantic Recall
|
||||
# ============================================================================
|
||||
|
||||
async def recall_semantic(
|
||||
query: str,
|
||||
memory_type: str | None = None,
|
||||
limit: int = 5,
|
||||
) -> str:
|
||||
"""
|
||||
Search memories by semantic similarity.
|
||||
|
||||
Use this to find memories that are conceptually related to
|
||||
the query, even if exact words don't match. This is the main
|
||||
tool for answering questions like "What car do I drive?" or
|
||||
"What did I mention about my job?"
|
||||
|
||||
Args:
|
||||
query: Natural language query to search for
|
||||
memory_type: Optional filter: "user_profile", "preference", "learned_fact"
|
||||
limit: Maximum memories to return (default: 5)
|
||||
|
||||
Returns:
|
||||
Matching memories with their content and relevance scores
|
||||
|
||||
Examples:
|
||||
recall_semantic("What is my car?")
|
||||
recall_semantic("work preferences", memory_type="preference")
|
||||
recall_semantic("family members")
|
||||
"""
|
||||
try:
|
||||
user = get_user()
|
||||
embedding_client = get_embedding_client()
|
||||
qdrant = get_qdrant_client()
|
||||
|
||||
# Generate embedding for query
|
||||
query_vector = await embedding_client.embed(query)
|
||||
if not query_vector:
|
||||
return "Unable to process query - embedding generation failed"
|
||||
|
||||
# Search memories
|
||||
results = await qdrant.search_memories(
|
||||
user=user,
|
||||
query_vector=query_vector,
|
||||
limit=limit,
|
||||
memory_type=memory_type,
|
||||
)
|
||||
|
||||
if not results:
|
||||
return f"No memories found related to '{query}'"
|
||||
|
||||
output_parts = [f"## Memories matching: {query}\n"]
|
||||
|
||||
for i, memory in enumerate(results, 1):
|
||||
mem_type = memory.get("type", "unknown")
|
||||
key = memory.get("key", "")
|
||||
value = memory.get("value", "")
|
||||
score = memory.get("score", 0.0)
|
||||
source = memory.get("source", "unknown")
|
||||
|
||||
type_icon = {
|
||||
"user_profile": "👤",
|
||||
"preference": "⚙️",
|
||||
"learned_fact": "💡",
|
||||
}.get(mem_type, "📝")
|
||||
|
||||
output_parts.append(f"{i}. {type_icon} **{key}** (relevance: {score:.2f})")
|
||||
output_parts.append(f" {value}")
|
||||
output_parts.append(f" _Type: {mem_type}, Source: {source}_")
|
||||
output_parts.append("")
|
||||
|
||||
logger.info(
|
||||
"memory_recall_semantic",
|
||||
query=query[:50],
|
||||
result_count=len(results),
|
||||
user=user,
|
||||
)
|
||||
|
||||
return "\n".join(output_parts)
|
||||
|
||||
except Exception as e:
|
||||
logger.error("memory_recall_semantic_error", error=str(e), query=query[:50])
|
||||
return f"Error searching memories: {str(e)}"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Store Memory
|
||||
# ============================================================================
|
||||
|
||||
async def store_insight(
|
||||
key: str,
|
||||
value: str,
|
||||
keywords: list[str] | None = None,
|
||||
importance: float = 0.5,
|
||||
) -> str:
|
||||
"""
|
||||
Store a new insight or learned fact about the user.
|
||||
|
||||
Use this when:
|
||||
- User explicitly asks to remember something
|
||||
- User shares personal information worth remembering
|
||||
- You learn something from conversation that should persist
|
||||
|
||||
The memory will be stored with vector embedding for semantic search
|
||||
and can be recalled later using recall_semantic.
|
||||
|
||||
Args:
|
||||
key: Short identifier for the memory (e.g., "car", "employer", "pet")
|
||||
value: The actual information to remember
|
||||
keywords: Optional keywords for better search (auto-extracted if not provided)
|
||||
importance: How important is this? 0.0 (trivial) to 1.0 (critical)
|
||||
|
||||
Returns:
|
||||
Confirmation of stored memory
|
||||
|
||||
Examples:
|
||||
store_insight("car", "User drives a Tesla Model 3")
|
||||
store_insight("employer", "Works at Acme Corp as software engineer", importance=0.8)
|
||||
store_insight("coffee", "Prefers oat milk lattes", keywords=["coffee", "drink", "preference"])
|
||||
"""
|
||||
try:
|
||||
# Auto-generate keywords if not provided
|
||||
if not keywords:
|
||||
keywords = [key]
|
||||
# Extract simple keywords from value
|
||||
words = value.lower().split()
|
||||
keywords.extend([w for w in words if len(w) > 4][:5])
|
||||
|
||||
success = await memory_service.store_fact(
|
||||
key=key,
|
||||
value=value,
|
||||
keywords=keywords,
|
||||
importance=importance,
|
||||
source="conversation",
|
||||
)
|
||||
|
||||
if success:
|
||||
output_parts = [
|
||||
"## Memory Stored",
|
||||
f"**Key:** {key}",
|
||||
f"**Value:** {value}",
|
||||
f"**Keywords:** {', '.join(keywords)}",
|
||||
f"**Importance:** {importance:.1f}",
|
||||
"",
|
||||
"_Memory is now searchable via semantic recall._"
|
||||
]
|
||||
|
||||
logger.info(
|
||||
"memory_store_insight",
|
||||
key=key,
|
||||
importance=importance,
|
||||
user=get_user(),
|
||||
)
|
||||
|
||||
return "\n".join(output_parts)
|
||||
else:
|
||||
return f"Failed to store memory for key '{key}'"
|
||||
|
||||
except Exception as e:
|
||||
logger.error("memory_store_insight_error", error=str(e), key=key)
|
||||
return f"Error storing memory: {str(e)}"
|
||||
|
||||
|
||||
async def update_profile(
|
||||
key: str,
|
||||
value: str,
|
||||
) -> str:
|
||||
"""
|
||||
Update user profile information.
|
||||
|
||||
Use this for core identity information:
|
||||
- name, location, timezone
|
||||
- language preferences
|
||||
- occupation
|
||||
|
||||
Profile data has high importance and is used for context
|
||||
by the Steward during request analysis.
|
||||
|
||||
Args:
|
||||
key: Profile field (e.g., "name", "location", "timezone")
|
||||
value: The value to set
|
||||
|
||||
Returns:
|
||||
Confirmation of profile update
|
||||
|
||||
Examples:
|
||||
update_profile("location", "Amsterdam, Netherlands")
|
||||
update_profile("timezone", "Europe/Amsterdam")
|
||||
update_profile("name", "John")
|
||||
"""
|
||||
try:
|
||||
success = await memory_service.set_profile(
|
||||
key=key,
|
||||
value=value,
|
||||
keywords=[key, "profile"],
|
||||
)
|
||||
|
||||
if success:
|
||||
output_parts = [
|
||||
"## Profile Updated",
|
||||
f"**{key}:** {value}",
|
||||
"",
|
||||
"_Profile data is automatically included in context._"
|
||||
]
|
||||
|
||||
logger.info(
|
||||
"memory_update_profile",
|
||||
key=key,
|
||||
user=get_user(),
|
||||
)
|
||||
|
||||
return "\n".join(output_parts)
|
||||
else:
|
||||
return f"Failed to update profile field '{key}'"
|
||||
|
||||
except Exception as e:
|
||||
logger.error("memory_update_profile_error", error=str(e), key=key)
|
||||
return f"Error updating profile: {str(e)}"
|
||||
|
||||
|
||||
async def update_preference(
|
||||
key: str,
|
||||
value: str,
|
||||
) -> str:
|
||||
"""
|
||||
Update user preferences.
|
||||
|
||||
Use this for settings and preferences:
|
||||
- temperature_unit (celsius/fahrenheit)
|
||||
- distance_unit (metric/imperial)
|
||||
- theme, language, etc.
|
||||
|
||||
Preferences are used by agents to customize responses.
|
||||
|
||||
Args:
|
||||
key: Preference name (e.g., "temperature_unit", "theme")
|
||||
value: Preference value
|
||||
|
||||
Returns:
|
||||
Confirmation of preference update
|
||||
|
||||
Examples:
|
||||
update_preference("temperature_unit", "celsius")
|
||||
update_preference("distance_unit", "metric")
|
||||
update_preference("theme", "dark")
|
||||
"""
|
||||
try:
|
||||
success = await memory_service.set_preference(
|
||||
key=key,
|
||||
value=value,
|
||||
)
|
||||
|
||||
if success:
|
||||
output_parts = [
|
||||
"## Preference Updated",
|
||||
f"**{key}:** {value}",
|
||||
"",
|
||||
"_Preference will be applied to future responses._"
|
||||
]
|
||||
|
||||
logger.info(
|
||||
"memory_update_preference",
|
||||
key=key,
|
||||
user=get_user(),
|
||||
)
|
||||
|
||||
return "\n".join(output_parts)
|
||||
else:
|
||||
return f"Failed to update preference '{key}'"
|
||||
|
||||
except Exception as e:
|
||||
logger.error("memory_update_preference_error", error=str(e), key=key)
|
||||
return f"Error updating preference: {str(e)}"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# List Memories
|
||||
# ============================================================================
|
||||
|
||||
async def list_memories(
|
||||
memory_type: str = "learned_fact",
|
||||
limit: int = 20,
|
||||
) -> str:
|
||||
"""
|
||||
List stored memories of a specific type.
|
||||
|
||||
Use this to browse what's stored in memory without
|
||||
a specific search query.
|
||||
|
||||
Args:
|
||||
memory_type: Type to list: "user_profile", "preference", "learned_fact"
|
||||
limit: Maximum memories to return (default: 20)
|
||||
|
||||
Returns:
|
||||
List of memories with their keys and values
|
||||
|
||||
Examples:
|
||||
list_memories("user_profile")
|
||||
list_memories("preference")
|
||||
list_memories("learned_fact", limit=10)
|
||||
"""
|
||||
try:
|
||||
user = get_user()
|
||||
qdrant = get_qdrant_client()
|
||||
|
||||
# Convert string to MemoryType
|
||||
try:
|
||||
mem_type = MemoryType(memory_type)
|
||||
except ValueError:
|
||||
return f"Invalid memory type '{memory_type}'. Use: user_profile, preference, or learned_fact"
|
||||
|
||||
# Get all memories of type
|
||||
results = qdrant._client.scroll(
|
||||
collection_name=f"memories_{user}",
|
||||
scroll_filter={
|
||||
"must": [
|
||||
{"key": "type", "match": {"value": memory_type}},
|
||||
]
|
||||
},
|
||||
limit=limit,
|
||||
with_payload=True,
|
||||
with_vectors=False,
|
||||
)
|
||||
|
||||
points, _ = results
|
||||
if not points:
|
||||
return f"No {memory_type} memories found"
|
||||
|
||||
type_icon = {
|
||||
"user_profile": "👤",
|
||||
"preference": "⚙️",
|
||||
"learned_fact": "💡",
|
||||
}.get(memory_type, "📝")
|
||||
|
||||
output_parts = [f"## {type_icon} {memory_type.replace('_', ' ').title()} Memories\n"]
|
||||
|
||||
for point in points:
|
||||
payload = point.payload
|
||||
key = payload.get("key", "unknown")
|
||||
value = payload.get("value", "")
|
||||
importance = payload.get("importance", 0.5)
|
||||
|
||||
output_parts.append(f"- **{key}**: {value}")
|
||||
if importance > 0.7:
|
||||
output_parts.append(f" _(importance: {importance:.1f})_")
|
||||
|
||||
logger.info(
|
||||
"memory_list",
|
||||
memory_type=memory_type,
|
||||
count=len(points),
|
||||
user=user,
|
||||
)
|
||||
|
||||
return "\n".join(output_parts)
|
||||
|
||||
except Exception as e:
|
||||
logger.error("memory_list_error", error=str(e), memory_type=memory_type)
|
||||
return f"Error listing memories: {str(e)}"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Forget Memory
|
||||
# ============================================================================
|
||||
|
||||
async def forget_memory(
|
||||
key: str,
|
||||
memory_type: str = "learned_fact",
|
||||
) -> str:
|
||||
"""
|
||||
Remove a specific memory.
|
||||
|
||||
Use this when:
|
||||
- User asks to forget something
|
||||
- Information is outdated or incorrect
|
||||
- Privacy concerns
|
||||
|
||||
Args:
|
||||
key: Key of the memory to forget
|
||||
memory_type: Type of memory: "user_profile", "preference", "learned_fact"
|
||||
|
||||
Returns:
|
||||
Confirmation of deletion
|
||||
|
||||
Examples:
|
||||
forget_memory("old_car")
|
||||
forget_memory("location", memory_type="user_profile")
|
||||
forget_memory("theme", memory_type="preference")
|
||||
"""
|
||||
try:
|
||||
# Convert string to MemoryType
|
||||
try:
|
||||
mem_type = MemoryType(memory_type)
|
||||
except ValueError:
|
||||
return f"Invalid memory type '{memory_type}'. Use: user_profile, preference, or learned_fact"
|
||||
|
||||
success = await memory_service.delete_memory(
|
||||
key=key,
|
||||
memory_type=mem_type,
|
||||
)
|
||||
|
||||
if success:
|
||||
output_parts = [
|
||||
"## Memory Forgotten",
|
||||
f"**Key:** {key}",
|
||||
f"**Type:** {memory_type}",
|
||||
"",
|
||||
"_Memory has been removed._"
|
||||
]
|
||||
|
||||
logger.info(
|
||||
"memory_forget",
|
||||
key=key,
|
||||
memory_type=memory_type,
|
||||
user=get_user(),
|
||||
)
|
||||
|
||||
return "\n".join(output_parts)
|
||||
else:
|
||||
return f"Memory '{key}' not found or already deleted"
|
||||
|
||||
except Exception as e:
|
||||
logger.error("memory_forget_error", error=str(e), key=key)
|
||||
return f"Error forgetting memory: {str(e)}"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Tool Collection for Registration
|
||||
# ============================================================================
|
||||
|
||||
# All tools available to The Biographer
|
||||
BIOGRAPHER_TOOLS = [
|
||||
# Recall
|
||||
recall_semantic,
|
||||
list_memories,
|
||||
# Record
|
||||
store_insight,
|
||||
update_profile,
|
||||
update_preference,
|
||||
# Manage
|
||||
forget_memory,
|
||||
]
|
||||
@@ -146,7 +146,84 @@ async def delegate_to_librarian(
|
||||
)
|
||||
|
||||
|
||||
async def delegate_to_biographer(
|
||||
task: str,
|
||||
context: str = "",
|
||||
) -> DelegationResult:
|
||||
"""
|
||||
Delegate a memory task to The Biographer.
|
||||
|
||||
The Biographer handles:
|
||||
- Semantic recall ("What car do I drive?", "What's my job?")
|
||||
- Recording new facts from conversation
|
||||
- Profile updates (name, location, timezone)
|
||||
- Preference updates (units, theme)
|
||||
- Memory management (forget, list)
|
||||
|
||||
For direct key-based lookups (get location, get timezone), use
|
||||
memory_service directly - it's faster and doesn't require LLM.
|
||||
|
||||
Args:
|
||||
task: Clear description of what needs to be done.
|
||||
Include the action verb (recall, remember, forget, etc.)
|
||||
Example: "What car do I drive?"
|
||||
Example: "Remember that I work at Acme Corp"
|
||||
context: Additional context from the user's request or
|
||||
conversation history
|
||||
|
||||
Returns:
|
||||
DelegationResult with The Biographer's response
|
||||
|
||||
Example:
|
||||
>>> result = await delegate_to_biographer(
|
||||
... task="What do you know about my preferences?",
|
||||
... context="User is asking about stored information",
|
||||
... )
|
||||
>>> if result.success:
|
||||
... print(result.output)
|
||||
"""
|
||||
from src.agents.biographer.agent import run_biographer
|
||||
|
||||
logger.info(
|
||||
"delegation_to_biographer_started",
|
||||
task=task[:100],
|
||||
has_context=bool(context),
|
||||
)
|
||||
|
||||
try:
|
||||
# Use run() not run_stream() - avoids Ollama bug
|
||||
output = await run_biographer(task=task, context=context)
|
||||
|
||||
logger.info(
|
||||
"delegation_to_biographer_completed",
|
||||
task=task[:50],
|
||||
output_length=len(output),
|
||||
)
|
||||
|
||||
return DelegationResult(
|
||||
expert_name="biographer",
|
||||
task=task,
|
||||
success=True,
|
||||
output=output,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"delegation_to_biographer_error",
|
||||
task=task[:50],
|
||||
error=str(e),
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
return DelegationResult(
|
||||
expert_name="biographer",
|
||||
task=task,
|
||||
success=False,
|
||||
output="",
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
|
||||
# Future expert delegation wrappers will be added here:
|
||||
# - delegate_to_memory(task, context) -> DelegationResult
|
||||
# - delegate_to_home_automation(task, context) -> DelegationResult
|
||||
# - delegate_to_developer(task, context) -> DelegationResult
|
||||
|
||||
@@ -4,7 +4,7 @@ Steward agent schemas.
|
||||
Defines the structured output models for Steward's request analysis
|
||||
and capability recommendations.
|
||||
"""
|
||||
from typing import Literal, Optional
|
||||
from typing import Any, Literal, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
@@ -56,6 +56,10 @@ class StewardRecommendation(BaseModel):
|
||||
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)"
|
||||
)
|
||||
|
||||
def format_for_butler(self) -> str:
|
||||
"""
|
||||
@@ -88,6 +92,23 @@ class StewardRecommendation(BaseModel):
|
||||
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}")
|
||||
|
||||
lines.append("=" * 40)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -5,13 +5,15 @@ Provides high-level interface for request analysis with logging,
|
||||
benchmarking, and error handling.
|
||||
|
||||
Parses plain text recommendations into structured data.
|
||||
Includes memory pre-fetch for user context injection.
|
||||
"""
|
||||
import re
|
||||
from typing import Optional
|
||||
from typing import Any, 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 src.core.memory_service import memory_service
|
||||
from .agent import get_steward_agent
|
||||
from .schemas import ConversationContext, StewardRecommendation
|
||||
|
||||
@@ -147,6 +149,69 @@ def _extract_missing_capabilities(text: str) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
async def _prefetch_memory_context(user_request: str) -> dict[str, Any]:
|
||||
"""
|
||||
Pre-fetch user context that might be needed for this request.
|
||||
|
||||
This is the "direct access" layer - fast lookups without LLM overhead.
|
||||
Uses simple keyword matching to determine what context to fetch.
|
||||
|
||||
Args:
|
||||
user_request: The user's request text
|
||||
|
||||
Returns:
|
||||
Dict with profile and/or preferences data
|
||||
|
||||
Example:
|
||||
>>> ctx = await _prefetch_memory_context("What's the weather?")
|
||||
>>> ctx
|
||||
{"profile": {"location": "Amsterdam"}}
|
||||
"""
|
||||
request_lower = user_request.lower()
|
||||
|
||||
# Determine what context might be needed based on keywords
|
||||
profile_keys = []
|
||||
|
||||
# Location-related queries
|
||||
if any(word in request_lower for word in [
|
||||
"weather", "temperature", "forecast", "nearby", "local",
|
||||
"directions", "distance", "map", "here"
|
||||
]):
|
||||
profile_keys.append("location")
|
||||
|
||||
# Time-related queries
|
||||
if any(word in request_lower for word in [
|
||||
"time", "schedule", "meeting", "appointment", "reminder",
|
||||
"alarm", "when", "today", "tomorrow"
|
||||
]):
|
||||
profile_keys.append("timezone")
|
||||
|
||||
# Personal queries
|
||||
if any(word in request_lower for word in [
|
||||
"my name", "who am i", "about me"
|
||||
]):
|
||||
profile_keys.append("name")
|
||||
|
||||
# Always fetch preferences if they might affect response format
|
||||
include_preferences = any(word in request_lower for word in [
|
||||
"temperature", "weather", "convert", "unit", "format",
|
||||
"celsius", "fahrenheit", "metric", "imperial"
|
||||
])
|
||||
|
||||
try:
|
||||
return await memory_service.prefetch_context(
|
||||
include_profile=bool(profile_keys),
|
||||
include_preferences=include_preferences,
|
||||
profile_keys=profile_keys if profile_keys else None,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"steward_prefetch_memory_failed",
|
||||
error=str(e),
|
||||
)
|
||||
return {}
|
||||
|
||||
|
||||
async def analyze_request(
|
||||
user_request: str,
|
||||
conversation_history: list[dict],
|
||||
@@ -186,6 +251,10 @@ async def analyze_request(
|
||||
}
|
||||
) as log_ctx:
|
||||
try:
|
||||
# Pre-fetch user context from memory (fast, no LLM)
|
||||
memory_context = await _prefetch_memory_context(user_request)
|
||||
log_ctx["memory_context_keys"] = list(memory_context.keys())
|
||||
|
||||
# Get Steward agent
|
||||
steward = get_steward_agent()
|
||||
|
||||
@@ -193,6 +262,7 @@ async def analyze_request(
|
||||
"steward_analyzing_request",
|
||||
request=user_request,
|
||||
history_turns=len(conversation_history),
|
||||
memory_context=bool(memory_context),
|
||||
)
|
||||
|
||||
# Get plain text analysis from Steward
|
||||
@@ -212,7 +282,8 @@ async def analyze_request(
|
||||
reasoning=analysis_text,
|
||||
estimated_complexity=complexity,
|
||||
conversation_context=context,
|
||||
missing_capabilities=missing
|
||||
missing_capabilities=missing,
|
||||
memory_context=memory_context,
|
||||
)
|
||||
|
||||
# Update log context with results
|
||||
|
||||
@@ -0,0 +1,619 @@
|
||||
"""
|
||||
Memory service for direct key-based access.
|
||||
|
||||
Provides fast, LLM-free access to user memories for:
|
||||
- Known-key lookups (location, timezone, preferences)
|
||||
- Session context (current topic, recent entities)
|
||||
- Structured storage (explicit user instructions)
|
||||
|
||||
This is the "direct access layer" - no LLM interpretation.
|
||||
For semantic/fuzzy queries, use the Memory Agent instead.
|
||||
|
||||
Usage:
|
||||
from src.core.memory_service import memory_service
|
||||
|
||||
# Get user's location (fast, no LLM)
|
||||
location = await memory_service.get_profile("location")
|
||||
|
||||
# Set a preference
|
||||
await memory_service.set_preference("temperature_unit", "celsius")
|
||||
|
||||
# Get session context
|
||||
ctx = await memory_service.get_session_context(conversation_id)
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from .config import config
|
||||
from .context import get_user, get_conversation_id
|
||||
from .embeddings import get_embedding_client
|
||||
from .logging_config import get_logger
|
||||
from .memory_cache import get_memory_cache
|
||||
from .multi_tenancy import get_memory_collection_name
|
||||
from .qdrant import get_qdrant_client
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class MemoryType(str, Enum):
|
||||
"""Types of memories stored in Qdrant."""
|
||||
USER_PROFILE = "user_profile" # Name, location, timezone
|
||||
PREFERENCE = "preference" # Units, language, theme
|
||||
LEARNED_FACT = "learned_fact" # "My car is a Tesla"
|
||||
|
||||
|
||||
class MemoryRecord(BaseModel):
|
||||
"""A memory record stored in Qdrant."""
|
||||
id: str
|
||||
type: MemoryType
|
||||
key: str # e.g., "location", "timezone", "car"
|
||||
value: str # The actual content
|
||||
keywords: list[str] = Field(default_factory=list)
|
||||
importance: float = 0.5 # 0.0 - 1.0
|
||||
source: str = "explicit" # "explicit" | "inferred" | "conversation"
|
||||
created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
||||
updated_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
||||
|
||||
|
||||
class MemoryService:
|
||||
"""
|
||||
Direct access to user memories without LLM overhead.
|
||||
|
||||
Use this for:
|
||||
- Known-key lookups: get_profile("location"), get_preference("units")
|
||||
- Explicit storage: set_preference("theme", "dark")
|
||||
- Session context: get_session_context(), update_session_context()
|
||||
|
||||
Do NOT use for:
|
||||
- Fuzzy queries: "What car do I drive?" → Use Memory Agent
|
||||
- Semantic recall: "What did I mention about X?" → Use Memory Agent
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize memory service with lazy client loading."""
|
||||
self._qdrant = None
|
||||
self._embedding = None
|
||||
self._cache = None
|
||||
|
||||
@property
|
||||
def qdrant(self):
|
||||
"""Lazy-load Qdrant client."""
|
||||
if self._qdrant is None:
|
||||
self._qdrant = get_qdrant_client()
|
||||
return self._qdrant
|
||||
|
||||
@property
|
||||
def embedding(self):
|
||||
"""Lazy-load embedding client."""
|
||||
if self._embedding is None:
|
||||
self._embedding = get_embedding_client()
|
||||
return self._embedding
|
||||
|
||||
@property
|
||||
def cache(self):
|
||||
"""Lazy-load Redis cache."""
|
||||
if self._cache is None:
|
||||
self._cache = get_memory_cache()
|
||||
return self._cache
|
||||
|
||||
# =========================================================================
|
||||
# Profile Methods (user_profile type)
|
||||
# =========================================================================
|
||||
|
||||
async def get_profile(self, key: str, user: str | None = None) -> str | None:
|
||||
"""
|
||||
Get a user profile value by key.
|
||||
|
||||
Args:
|
||||
key: Profile key (e.g., "location", "timezone", "name")
|
||||
user: User ID (defaults to current request context)
|
||||
|
||||
Returns:
|
||||
Profile value or None if not found
|
||||
|
||||
Example:
|
||||
>>> location = await memory_service.get_profile("location")
|
||||
>>> location
|
||||
"Amsterdam, Netherlands"
|
||||
"""
|
||||
user = user or get_user()
|
||||
return await self._get_memory(user, MemoryType.USER_PROFILE, key)
|
||||
|
||||
async def set_profile(
|
||||
self,
|
||||
key: str,
|
||||
value: str,
|
||||
user: str | None = None,
|
||||
keywords: list[str] | None = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Set a user profile value.
|
||||
|
||||
Args:
|
||||
key: Profile key (e.g., "location", "timezone")
|
||||
value: Profile value
|
||||
user: User ID (defaults to current request context)
|
||||
keywords: Optional keywords for semantic search
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
|
||||
Example:
|
||||
>>> await memory_service.set_profile("location", "Amsterdam, Netherlands")
|
||||
True
|
||||
"""
|
||||
user = user or get_user()
|
||||
return await self._set_memory(
|
||||
user=user,
|
||||
memory_type=MemoryType.USER_PROFILE,
|
||||
key=key,
|
||||
value=value,
|
||||
keywords=keywords or [key],
|
||||
importance=0.9, # Profile data is important
|
||||
)
|
||||
|
||||
# =========================================================================
|
||||
# Preference Methods (preference type)
|
||||
# =========================================================================
|
||||
|
||||
async def get_preference(self, key: str, user: str | None = None) -> str | None:
|
||||
"""
|
||||
Get a user preference by key.
|
||||
|
||||
Args:
|
||||
key: Preference key (e.g., "temperature_unit", "language", "theme")
|
||||
user: User ID (defaults to current request context)
|
||||
|
||||
Returns:
|
||||
Preference value or None if not found
|
||||
|
||||
Example:
|
||||
>>> units = await memory_service.get_preference("temperature_unit")
|
||||
>>> units
|
||||
"celsius"
|
||||
"""
|
||||
user = user or get_user()
|
||||
return await self._get_memory(user, MemoryType.PREFERENCE, key)
|
||||
|
||||
async def set_preference(
|
||||
self,
|
||||
key: str,
|
||||
value: str,
|
||||
user: str | None = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Set a user preference.
|
||||
|
||||
Args:
|
||||
key: Preference key
|
||||
value: Preference value
|
||||
user: User ID (defaults to current request context)
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
|
||||
Example:
|
||||
>>> await memory_service.set_preference("theme", "dark")
|
||||
True
|
||||
"""
|
||||
user = user or get_user()
|
||||
return await self._set_memory(
|
||||
user=user,
|
||||
memory_type=MemoryType.PREFERENCE,
|
||||
key=key,
|
||||
value=value,
|
||||
keywords=[key, "preference"],
|
||||
importance=0.7,
|
||||
)
|
||||
|
||||
async def get_all_preferences(self, user: str | None = None) -> dict[str, str]:
|
||||
"""
|
||||
Get all preferences for a user.
|
||||
|
||||
Returns:
|
||||
Dict of key -> value for all preferences
|
||||
"""
|
||||
user = user or get_user()
|
||||
memories = await self._get_all_by_type(user, MemoryType.PREFERENCE)
|
||||
return {m["key"]: m["value"] for m in memories}
|
||||
|
||||
# =========================================================================
|
||||
# Learned Facts (learned_fact type) - for direct storage only
|
||||
# =========================================================================
|
||||
|
||||
async def store_fact(
|
||||
self,
|
||||
key: str,
|
||||
value: str,
|
||||
user: str | None = None,
|
||||
keywords: list[str] | None = None,
|
||||
importance: float = 0.5,
|
||||
source: str = "explicit",
|
||||
) -> bool:
|
||||
"""
|
||||
Store a learned fact about the user.
|
||||
|
||||
Use this for explicit user statements like:
|
||||
- "Remember that my car is a Tesla"
|
||||
- "I work at Acme Corp"
|
||||
|
||||
For semantic extraction from conversation, use the Memory Agent.
|
||||
|
||||
Args:
|
||||
key: Fact identifier (e.g., "car", "employer")
|
||||
value: The fact content
|
||||
user: User ID
|
||||
keywords: Keywords for semantic search
|
||||
importance: 0.0-1.0 importance score
|
||||
source: "explicit" | "inferred" | "conversation"
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
"""
|
||||
user = user or get_user()
|
||||
return await self._set_memory(
|
||||
user=user,
|
||||
memory_type=MemoryType.LEARNED_FACT,
|
||||
key=key,
|
||||
value=value,
|
||||
keywords=keywords or [key],
|
||||
importance=importance,
|
||||
source=source,
|
||||
)
|
||||
|
||||
async def get_fact(self, key: str, user: str | None = None) -> str | None:
|
||||
"""
|
||||
Get a specific fact by key.
|
||||
|
||||
For semantic/fuzzy queries, use the Memory Agent.
|
||||
"""
|
||||
user = user or get_user()
|
||||
return await self._get_memory(user, MemoryType.LEARNED_FACT, key)
|
||||
|
||||
# =========================================================================
|
||||
# Session Context (Redis-backed, 24h TTL)
|
||||
# =========================================================================
|
||||
|
||||
async def get_session_context(
|
||||
self,
|
||||
conversation_id: str | None = None,
|
||||
user: str | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
Get session context for current conversation.
|
||||
|
||||
Args:
|
||||
conversation_id: Conversation ID (defaults to current context)
|
||||
user: User ID (defaults to current context)
|
||||
|
||||
Returns:
|
||||
Session context dict or None
|
||||
"""
|
||||
user = user or get_user()
|
||||
conversation_id = conversation_id or get_conversation_id()
|
||||
|
||||
if not conversation_id:
|
||||
return None
|
||||
|
||||
return await self.cache.get_session_context(user, conversation_id)
|
||||
|
||||
async def set_session_context(
|
||||
self,
|
||||
context: dict[str, Any],
|
||||
conversation_id: str | None = None,
|
||||
user: str | None = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Set session context for current conversation.
|
||||
|
||||
Args:
|
||||
context: Context data to store
|
||||
conversation_id: Conversation ID
|
||||
user: User ID
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
"""
|
||||
user = user or get_user()
|
||||
conversation_id = conversation_id or get_conversation_id()
|
||||
|
||||
if not conversation_id:
|
||||
logger.warning("memory_service_no_conversation_id")
|
||||
return False
|
||||
|
||||
return await self.cache.set_session_context(user, conversation_id, context)
|
||||
|
||||
async def update_session_context(
|
||||
self,
|
||||
updates: dict[str, Any],
|
||||
conversation_id: str | None = None,
|
||||
user: str | None = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Update session context (merge with existing).
|
||||
|
||||
Args:
|
||||
updates: Fields to update
|
||||
conversation_id: Conversation ID
|
||||
user: User ID
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
"""
|
||||
user = user or get_user()
|
||||
conversation_id = conversation_id or get_conversation_id()
|
||||
|
||||
if not conversation_id:
|
||||
return False
|
||||
|
||||
return await self.cache.update_session_context(user, conversation_id, updates)
|
||||
|
||||
async def get_recent_entities(
|
||||
self,
|
||||
conversation_id: str | None = None,
|
||||
user: str | None = None,
|
||||
) -> list[str]:
|
||||
"""
|
||||
Get recently mentioned entities in conversation.
|
||||
|
||||
Returns:
|
||||
List of entity names
|
||||
"""
|
||||
user = user or get_user()
|
||||
conversation_id = conversation_id or get_conversation_id()
|
||||
|
||||
if not conversation_id:
|
||||
return []
|
||||
|
||||
return await self.cache.get_recent_entities(user, conversation_id)
|
||||
|
||||
async def add_recent_entities(
|
||||
self,
|
||||
entities: list[str],
|
||||
conversation_id: str | None = None,
|
||||
user: str | None = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Add entities to recent entities set.
|
||||
|
||||
Args:
|
||||
entities: Entity names to add
|
||||
conversation_id: Conversation ID
|
||||
user: User ID
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
"""
|
||||
user = user or get_user()
|
||||
conversation_id = conversation_id or get_conversation_id()
|
||||
|
||||
if not conversation_id:
|
||||
return False
|
||||
|
||||
return await self.cache.add_recent_entities(user, conversation_id, entities)
|
||||
|
||||
# =========================================================================
|
||||
# Bulk / Pre-fetch Methods (for Steward)
|
||||
# =========================================================================
|
||||
|
||||
async def prefetch_context(
|
||||
self,
|
||||
user: str | None = None,
|
||||
include_profile: bool = True,
|
||||
include_preferences: bool = True,
|
||||
profile_keys: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Pre-fetch commonly needed context for Steward.
|
||||
|
||||
This is the main entry point for Steward to get user context
|
||||
before analyzing a request.
|
||||
|
||||
Args:
|
||||
user: User ID
|
||||
include_profile: Include profile data
|
||||
include_preferences: Include preferences
|
||||
profile_keys: Specific profile keys to fetch (None = common ones)
|
||||
|
||||
Returns:
|
||||
Dict with profile and preferences data
|
||||
|
||||
Example:
|
||||
>>> ctx = await memory_service.prefetch_context()
|
||||
>>> ctx
|
||||
{
|
||||
"profile": {"location": "Amsterdam", "timezone": "Europe/Amsterdam"},
|
||||
"preferences": {"temperature_unit": "celsius"}
|
||||
}
|
||||
"""
|
||||
user = user or get_user()
|
||||
result: dict[str, Any] = {}
|
||||
|
||||
if include_profile:
|
||||
profile_keys = profile_keys or ["location", "timezone", "name"]
|
||||
profile = {}
|
||||
for key in profile_keys:
|
||||
value = await self.get_profile(key, user)
|
||||
if value:
|
||||
profile[key] = value
|
||||
if profile:
|
||||
result["profile"] = profile
|
||||
|
||||
if include_preferences:
|
||||
preferences = await self.get_all_preferences(user)
|
||||
if preferences:
|
||||
result["preferences"] = preferences
|
||||
|
||||
logger.debug(
|
||||
"memory_service_prefetch",
|
||||
user=user,
|
||||
profile_keys=list(result.get("profile", {}).keys()),
|
||||
preference_keys=list(result.get("preferences", {}).keys()),
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
# =========================================================================
|
||||
# Internal Methods
|
||||
# =========================================================================
|
||||
|
||||
async def _get_memory(
|
||||
self,
|
||||
user: str,
|
||||
memory_type: MemoryType,
|
||||
key: str,
|
||||
) -> str | None:
|
||||
"""Get a memory by type and key (exact match)."""
|
||||
collection = get_memory_collection_name(user)
|
||||
|
||||
try:
|
||||
# Search with filter for exact type + key match
|
||||
# We use a dummy vector since we're filtering by payload
|
||||
results = self.qdrant._client.scroll(
|
||||
collection_name=collection,
|
||||
scroll_filter={
|
||||
"must": [
|
||||
{"key": "type", "match": {"value": memory_type.value}},
|
||||
{"key": "key", "match": {"value": key}},
|
||||
]
|
||||
},
|
||||
limit=1,
|
||||
with_payload=True,
|
||||
with_vectors=False,
|
||||
)
|
||||
|
||||
points, _ = results
|
||||
if points:
|
||||
return points[0].payload.get("value")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"memory_service_get_failed",
|
||||
user=user,
|
||||
type=memory_type.value,
|
||||
key=key,
|
||||
error=str(e),
|
||||
)
|
||||
return None
|
||||
|
||||
async def _set_memory(
|
||||
self,
|
||||
user: str,
|
||||
memory_type: MemoryType,
|
||||
key: str,
|
||||
value: str,
|
||||
keywords: list[str],
|
||||
importance: float = 0.5,
|
||||
source: str = "explicit",
|
||||
) -> bool:
|
||||
"""Set a memory (upsert by type + key)."""
|
||||
try:
|
||||
# Generate embedding for semantic search
|
||||
embedding = await self.embedding.embed(f"{key}: {value}")
|
||||
if not embedding:
|
||||
logger.error("memory_service_embedding_failed", key=key)
|
||||
return False
|
||||
|
||||
# Create memory ID from type + key for idempotent upserts
|
||||
memory_id = f"{memory_type.value}:{key}"
|
||||
|
||||
payload = {
|
||||
"type": memory_type.value,
|
||||
"key": key,
|
||||
"value": value,
|
||||
"keywords": keywords,
|
||||
"importance": importance,
|
||||
"source": source,
|
||||
"updated_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
|
||||
result = await self.qdrant.upsert_memory(
|
||||
user=user,
|
||||
memory_id=memory_id,
|
||||
vector=embedding,
|
||||
payload=payload,
|
||||
)
|
||||
|
||||
if result:
|
||||
logger.debug(
|
||||
"memory_service_set",
|
||||
user=user,
|
||||
type=memory_type.value,
|
||||
key=key,
|
||||
)
|
||||
return True
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"memory_service_set_failed",
|
||||
user=user,
|
||||
type=memory_type.value,
|
||||
key=key,
|
||||
error=str(e),
|
||||
)
|
||||
return False
|
||||
|
||||
async def _get_all_by_type(
|
||||
self,
|
||||
user: str,
|
||||
memory_type: MemoryType,
|
||||
limit: int = 100,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get all memories of a specific type."""
|
||||
collection = get_memory_collection_name(user)
|
||||
|
||||
try:
|
||||
results = self.qdrant._client.scroll(
|
||||
collection_name=collection,
|
||||
scroll_filter={
|
||||
"must": [
|
||||
{"key": "type", "match": {"value": memory_type.value}},
|
||||
]
|
||||
},
|
||||
limit=limit,
|
||||
with_payload=True,
|
||||
with_vectors=False,
|
||||
)
|
||||
|
||||
points, _ = results
|
||||
return [p.payload for p in points]
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"memory_service_get_all_failed",
|
||||
user=user,
|
||||
type=memory_type.value,
|
||||
error=str(e),
|
||||
)
|
||||
return []
|
||||
|
||||
async def delete_memory(
|
||||
self,
|
||||
key: str,
|
||||
memory_type: MemoryType,
|
||||
user: str | None = None,
|
||||
) -> bool:
|
||||
"""
|
||||
Delete a specific memory.
|
||||
|
||||
Args:
|
||||
key: Memory key
|
||||
memory_type: Type of memory
|
||||
user: User ID
|
||||
|
||||
Returns:
|
||||
True if deleted
|
||||
"""
|
||||
user = user or get_user()
|
||||
memory_id = f"{memory_type.value}:{key}"
|
||||
|
||||
return await self.qdrant.delete_memory(user, memory_id)
|
||||
|
||||
|
||||
# Global service instance
|
||||
memory_service = MemoryService()
|
||||
@@ -5,6 +5,7 @@ Handles initialization of household registry and other startup tasks.
|
||||
This module should be called during application startup to register
|
||||
all household members.
|
||||
"""
|
||||
from src.agents.biographer import register_biographer
|
||||
from src.agents.librarian import register_librarian
|
||||
from src.agents.tatlock_core import TATLOCK_CORE_CAPABILITY, tatlock_core_tools
|
||||
from src.core.household_registry import get_household_registry
|
||||
@@ -23,6 +24,7 @@ def register_household_members():
|
||||
Currently registers:
|
||||
- tatlock_core: Butler's core tools (calculator, datetime, web search)
|
||||
- librarian: Research and knowledge management (Phase 3)
|
||||
- biographer: User memory and context management (Phase F)
|
||||
"""
|
||||
registry = get_household_registry()
|
||||
|
||||
@@ -52,6 +54,16 @@ def register_household_members():
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
# Register The Biographer (Phase F)
|
||||
try:
|
||||
register_biographer()
|
||||
except Exception as e:
|
||||
# Don't fail startup if Biographer registration fails
|
||||
logger.warning(
|
||||
"biographer_registration_failed",
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"household_registration_complete",
|
||||
total_members=len(registry),
|
||||
|
||||
Reference in New Issue
Block a user