init: Kira — AI body double with Honcho memory

Full voice pipeline (Whisper STT -> DeepSeek LLM -> OpenAI TTS),
animated SVG avatar (Live2D-ready), girly-pop UI, lofi music,
timer/notes/pets/wardrobe widgets, 10 background scenes with
particle effects, Honcho cross-session memory.
This commit is contained in:
2026-06-04 10:51:38 -04:00
commit 97424cb98f
47 changed files with 5691 additions and 0 deletions
+183
View File
@@ -0,0 +1,183 @@
"""Honcho memory service for Kira.
Integrates Honcho persistent memory into Kira's conversation pipeline:
- User context retrieval before LLM calls
- Message storage after each exchange
- Cross-session memory for personalized responses
"""
import logging
from honcho import Honcho
from honcho.peer import Peer
from honcho.session import Session
from config import settings
logger = logging.getLogger("kira.memory")
class KiraMemory:
"""Manages Honcho memory for Kira conversations."""
def __init__(self):
self._honcho: Honcho | None = None
self._user_peer: Peer | None = None
self._kira_peer: Peer | None = None
self._session: Session | None = None
self._initialized = False
def init(self) -> bool:
"""Initialize Honcho connection. Returns False if not configured."""
api_key = settings.honcho_api_key
base_url = settings.honcho_base_url
if not api_key:
logger.warning("HONCHO_API_KEY not set — memory disabled")
return False
if not base_url:
self._honcho = Honcho(
api_key=api_key,
workspace_id="kira",
environment="production",
)
else:
self._honcho = Honcho(
api_key=api_key,
base_url=base_url,
workspace_id="kira",
)
logger.info(f"Honcho connected to workspace 'kira'")
self._initialized = True
return True
@property
def enabled(self) -> bool:
return self._initialized and self._honcho is not None
def ensure_peers(self, user_id: str = "default-user") -> None:
"""Get or create Honcho peers for the user and Kira."""
if not self.enabled:
return
self._user_peer = self._honcho.peer(user_id)
self._kira_peer = self._honcho.peer("kira")
logger.info(f"Peers ready: user={user_id}, kira")
def ensure_session(self, session_id: str) -> None:
"""Get or create a Honcho session for this conversation."""
if not self.enabled:
return
self._session = self._honcho.session(session_id)
# Add peers to session if not already members
if self._user_peer and self._kira_peer:
self._session.add_peers([self._user_peer, self._kira_peer])
logger.info(f"Session ready: {session_id}")
def get_user_context(self) -> str:
"""Query Honcho for context about the user.
Returns a string summary of what Honcho knows about the user,
to inject into the LLM system prompt. Empty string if no context.
"""
if not self.enabled or not self._user_peer:
return ""
try:
# Query Honcho's dialectic reasoning about the user
context = self._user_peer.chat(
"What should Kira know about this user? "
"Summarize their preferences, current projects, mood, "
"and any important context in 2-3 sentences."
)
if context:
return f"\n[Memory: {context}]"
return ""
except Exception as e:
logger.warning(f"Failed to get user context: {e}")
return ""
def get_kira_context(self) -> str:
"""Get what the user knows about Kira (relationship context)."""
if not self.enabled or not self._user_peer:
return ""
try:
context = self._user_peer.chat(
"What is the user's relationship with Kira? "
"How do they feel about their focus sessions? "
"Summarize in 1-2 sentences.",
target="kira",
)
if context:
return f"\n[Kira Context: {context}]"
return ""
except Exception as e:
logger.warning(f"Failed to get relationship context: {e}")
return ""
def build_system_prompt_suffix(self) -> str:
"""Build a context suffix to append to Kira's system prompt."""
if not self.enabled:
return ""
user_ctx = self.get_user_context()
kira_ctx = self.get_kira_context()
parts = [s for s in [user_ctx, kira_ctx] if s]
if not parts:
return ""
return "\n\n---\n### What Kira remembers:" + "".join(parts)
def store_messages(
self,
user_message: str,
kira_message: str,
) -> None:
"""Store a conversation exchange in Honcho."""
if not self.enabled or not self._session:
return
try:
messages = []
if self._user_peer:
messages.append(
self._user_peer.message(user_message)
)
if self._kira_peer:
messages.append(
self._kira_peer.message(kira_message)
)
if messages:
self._session.add_messages(messages)
logger.debug("Stored conversation exchange in Honcho")
except Exception as e:
logger.warning(f"Failed to store messages: {e}")
def store_user_message(self, text: str) -> None:
"""Store a single user message."""
if not self.enabled or not self._session or not self._user_peer:
return
try:
self._session.add_messages([self._user_peer.message(text)])
except Exception as e:
logger.warning(f"Failed to store user message: {e}")
def store_kira_message(self, text: str) -> None:
"""Store a single Kira message."""
if not self.enabled or not self._session or not self._kira_peer:
return
try:
self._session.add_messages([self._kira_peer.message(text)])
except Exception as e:
logger.warning(f"Failed to store Kira message: {e}")
# Singleton instance for the app
kira_memory = KiraMemory()