From 7c6c1c8647df3ec34bd9388b79f7f9654cc6bcff Mon Sep 17 00:00:00 2001 From: trnkhanh2908 Date: Fri, 12 Jun 2026 10:45:33 +0700 Subject: [PATCH 1/2] [lelamp] Update voice agent prompt and runtime flow --- lelamp/service/realtime/config.py | 12 +- lelamp/service/realtime/context_manager.py | 418 ++++++++---------- lelamp/service/realtime/orchestrator.py | 55 ++- .../realtime/resources/summarize_prompt.md | 6 +- .../realtime/resources/system_prompt.md | 37 +- .../resources/system_prompt_gemini.md | 77 ++++ .../resources/system_prompt_openai.md | 99 +++++ lelamp/service/realtime/summarizer.py | 6 +- .../realtime/voice_agent/gemini_live.py | 135 +++--- .../realtime/voice_agent/openai_realtime.py | 92 ++-- 10 files changed, 528 insertions(+), 409 deletions(-) create mode 100644 lelamp/service/realtime/resources/system_prompt_gemini.md create mode 100644 lelamp/service/realtime/resources/system_prompt_openai.md diff --git a/lelamp/service/realtime/config.py b/lelamp/service/realtime/config.py index 5a79d95f..e77cc37a 100644 --- a/lelamp/service/realtime/config.py +++ b/lelamp/service/realtime/config.py @@ -17,7 +17,7 @@ def _load_language() -> str | None: - """Load language from Lamp's config.json (stt_language field).""" + """Load language from device config.json (stt_language field).""" from lelamp.config import _lamp_cfg_get lang: str = _lamp_cfg_get("stt_language", "").strip() @@ -25,7 +25,7 @@ def _load_language() -> str | None: def _parse_turn_detection(value: str) -> OpenAITurnDetectionType | None: - """Parse LELAMP_REALTIME_TURN_DETECTION into an OpenAITurnDetectionType or None (off).""" + """Parse HAL_REALTIME_TURN_DETECTION into an OpenAITurnDetectionType or None (off).""" v = value.strip().lower() if v in ("off", "none", ""): return None @@ -51,8 +51,6 @@ class OpenAIConfig(BaseModel): ) truncation_type: OpenAITruncationType = OpenAITruncationType.RETENTION_RATIO truncation_retention_ratio: float = 0.5 - max_retries: int = 3 - reconnect_delay_s: float = 2.0 class GeminiConfig(BaseModel): @@ -73,9 +71,3 @@ class GeminiConfig(BaseModel): "", ) context_window_compression: bool = True - max_retries: int = 3 - reconnect_delay_s: float = 2.0 - send_timeout_s: float = 10.0 - recv_timeout_s: float = 300.0 - queue_poll_s: float = 1.0 - join_timeout_s: float = 5.0 diff --git a/lelamp/service/realtime/context_manager.py b/lelamp/service/realtime/context_manager.py index 832eb1a3..e3baa222 100644 --- a/lelamp/service/realtime/context_manager.py +++ b/lelamp/service/realtime/context_manager.py @@ -1,9 +1,8 @@ -"""Realtime context manager — builds instructions from lamp identity, skills, and memory.""" +"""Realtime context manager — builds instructions from device identity, skills, and memory.""" import json import logging import re -import threading from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -16,9 +15,13 @@ class RealtimeContextManager: - """Builds rich instructions for the realtime voice agent from lamp context.""" + """Builds rich instructions for the realtime voice agent from device context.""" DEFAULT_PROMPT_PATH: Path = RESOURCES_DIR / "system_prompt.md" + PROVIDER_PROMPT_PATHS: dict[str, Path] = { + "openai": RESOURCES_DIR / "system_prompt_openai.md", + "gemini": RESOURCES_DIR / "system_prompt_gemini.md", + } # Regex to extract YAML frontmatter from SKILL.md FRONTMATTER_RE: re.Pattern[str] = re.compile(r"^---\s*\n(.*?)\n---", re.DOTALL) @@ -30,35 +33,35 @@ def __init__( workspace_dir: str = app_config.REALTIME_WORKSPACE_DIR, realtime_memory_path: str = app_config.REALTIME_MEMORY_PATH, language: str | None = None, + provider: str = "", max_memory_entries: int = app_config.REALTIME_MAX_MEMORY_ENTRIES, trim_keep: int = app_config.REALTIME_MEMORY_TRIM_KEEP, - lamp_memory_max_chars: int = app_config.REALTIME_LAMP_MEMORY_MAX_CHARS, + device_memory_max_chars: int = app_config.REALTIME_LAMP_MEMORY_MAX_CHARS, realtime_memory_max_chars: int = app_config.REALTIME_MEMORY_MAX_CHARS, summarizer: RealtimeSummarizer | None = None, ) -> None: self._workspace: Path = Path(workspace_dir) self._realtime_memory_path: Path = Path(realtime_memory_path) self._language: str = language or "English" + self._provider: str = provider.strip().lower() self._max_memory_entries: int = max_memory_entries self._trim_keep: int = trim_keep - self._lamp_memory_max_chars: int = lamp_memory_max_chars + self._device_memory_max_chars: int = device_memory_max_chars self._realtime_memory_max_chars: int = realtime_memory_max_chars self._summarizer: RealtimeSummarizer | None = summarizer - # Summary files + # Summary files alongside the memory JSONL self._summary_path: Path = self._realtime_memory_path.parent / "summary.md" - self._lamp_summary_path: Path = self._realtime_memory_path.parent / "lamp_summary.md" - # Raw archive — append-only, trimmed by flushing oldest - self._raw_memory_path: Path = self._realtime_memory_path.with_name("memory_raw.jsonl") - # Lock for concurrent access to memory files (background summarizer + add_turn + load) - self._memory_lock: threading.Lock = threading.Lock() - self._summarizing: bool = False + self._device_summary_path: Path = ( + self._realtime_memory_path.parent / "device_summary.md" + ) + self._summary_max_chars: int = 10000 # --- Public API --- def build_instructions(self) -> str: """Build the full instruction string from all context sources. - If a summarizer is set, lamp memory and realtime memory are + If a summarizer is set, device memory and realtime memory are summarized via LLM before injection. """ sections: list[str] = [] @@ -68,25 +71,25 @@ def build_instructions(self) -> str: if prompt: sections.append(prompt) - # Lamp identity - identity: str = self._load_lamp_identity() + # device identity + identity: str = self._load_device_identity() if identity: - sections.append(f"# LAMP IDENTITY\n\n{identity}") + sections.append(f"# DEVICE IDENTITY\n\n{identity}") # Skills catalog catalog: str = self._load_skills_catalog() if catalog: sections.append(f"# SKILLS CATALOG\n\n{catalog}") - # Lamp memory — lamp_summary.md + unsummarized recent files (no LLM call) - lamp_mem_raw: list[str] = self._load_lamp_memory_entries() - if lamp_mem_raw: - sections.append(f"# LAMP MEMORY\n\n" + "\n\n".join(lamp_mem_raw)) + # device memory (pre-summarized at startup + recent entries) + device_mem: str = self._build_device_memory() + if device_mem: + sections.append(f"# DEVICE MEMORY\n\n{device_mem}") - # Realtime memory — summary.md + unsummarized entries from memory.jsonl (no LLM call) - rt_mem_raw: list[str] = self._load_realtime_memory_entries() - if rt_mem_raw: - sections.append(f"# REALTIME MEMORY\n\n" + "\n\n".join(rt_mem_raw)) + # Realtime memory (pre-summarized at startup + recent conversation) + rt_mem: str = self._build_realtime_memory() + if rt_mem: + sections.append(f"# REALTIME MEMORY\n\n{rt_mem}") return "\n\n".join(sections) @@ -112,138 +115,20 @@ def _parse_jsonl_lines(lines: list[str]) -> list[str]: entries.append(formatted) return entries - def summarize_lamp_memory(self) -> None: - """Summarize lamp memory files modified after the last lamp_summary.md. - - Reads files newer than lamp_summary.md by mtime, summarizes them - together with the existing summary, and writes the result. - """ - if not self._summarizer: - return - memory_dir: Path = self._workspace / "memory" - if not memory_dir.is_dir(): - return - - summary_mtime: float = ( - self._lamp_summary_path.stat().st_mtime - if self._lamp_summary_path.exists() - else 0.0 - ) - - # Collect files modified after the last summary - new_files: list[Path] = [ - f for f in sorted(memory_dir.glob("*.md"), key=lambda f: f.stat().st_mtime) - if f.stat().st_mtime > summary_mtime - ] - if not new_files: - return - - new_entries: list[str] = [] - total_chars: int = 0 - for md_file in new_files: - try: - content: str = md_file.read_text(encoding="utf-8").strip() - if not content: - continue - entry: str = f"## {md_file.stem}\n\n{content}" - if total_chars + len(entry) > self._lamp_memory_max_chars: - break - new_entries.append(entry) - total_chars += len(entry) - except Exception as e: - logger.warning("[realtime] Failed to read memory %s: %s", md_file, e) - if not new_entries: - return - - to_summarize: list[str] = [] - if self._lamp_summary_path.exists(): - try: - existing: str = self._lamp_summary_path.read_text(encoding="utf-8").strip() - if existing: - to_summarize.append(f"[Previous summary]\n{existing}") - except Exception: - pass - to_summarize.extend(new_entries) - - logger.info("[realtime] Summarizing %d new lamp memory files...", len(new_files)) - new_summary: str = self._summarizer.summarize(to_summarize) - if new_summary: - self._lamp_summary_path.parent.mkdir(parents=True, exist_ok=True) - self._lamp_summary_path.write_text(new_summary + "\n", encoding="utf-8") - logger.info("[realtime] Lamp memory summarization complete → lamp_summary.md") - - def summarize_realtime_memory(self) -> None: - """Summarize entries in memory.jsonl into summary.md, keeping entries added during summarization. - - Called on shutdown and by the trim background thread. - """ - if not self._summarizer: - return - with self._memory_lock: - if self._summarizing: - return - self._summarizing = True - try: - with self._memory_lock: - if not self._realtime_memory_path.exists(): - return - raw: str = self._realtime_memory_path.read_text(encoding="utf-8").strip() - if not raw: - return - lines: list[str] = raw.splitlines() - lines_read: int = len(lines) - entries: list[str] = self._parse_jsonl_lines(lines) - if not entries: - return - - to_summarize: list[str] = [] - with self._memory_lock: - if self._summary_path.exists(): - try: - existing: str = self._summary_path.read_text(encoding="utf-8").strip() - if existing: - to_summarize.append(f"[Previous summary]\n{existing}") - except Exception: - pass - to_summarize.extend(entries) - - logger.info("[realtime] Summarizing %d realtime memory entries...", len(entries)) - new_summary: str = self._summarizer.summarize(to_summarize) - if new_summary: - with self._memory_lock: - self._summary_path.write_text(new_summary + "\n", encoding="utf-8") - # Only remove the lines we read — keep any new entries added during summarization - current_lines: list[str] = ( - self._realtime_memory_path.read_text(encoding="utf-8").strip().splitlines() - ) - remaining: list[str] = current_lines[lines_read:] - self._realtime_memory_path.write_text( - "\n".join(remaining) + "\n" if remaining else "", encoding="utf-8" - ) - logger.info("[realtime] Realtime memory summarization complete → summary.md (kept %d new entries)", len(remaining)) - finally: - self._summarizing = False - def add_turn(self, user_text: str, agent_text: str) -> None: - """Save a conversation turn to both working memory and raw archive.""" + """Save a conversation turn to the realtime memory file.""" entry: dict[str, Any] = { "ts": datetime.now(timezone.utc).isoformat(), "user": user_text, "agent": agent_text, } - line: str = json.dumps(entry, ensure_ascii=False) + "\n" try: - with self._memory_lock: - self._realtime_memory_path.parent.mkdir(parents=True, exist_ok=True) - # Working memory (summarized periodically, then cleared) - with open(self._realtime_memory_path, "a", encoding="utf-8") as f: - f.write(line) - # Raw archive (append-only, trimmed by flushing oldest) - with open(self._raw_memory_path, "a", encoding="utf-8") as f: - f.write(line) + self._realtime_memory_path.parent.mkdir(parents=True, exist_ok=True) + with open(self._realtime_memory_path, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") self._trim_memory_if_needed() except Exception as e: - logger.warning("[realtime] Failed to save realtime memory: %s", e) + logger.warning("Failed to save realtime memory: %s", e) # --- Private loaders --- @@ -263,15 +148,113 @@ def add_turn(self, user_text: str, agent_text: str) -> None: } def _load_system_prompt(self) -> str: - """Load system_prompt.md with {language} placeholder resolved to full name.""" + """Load provider-specific system prompt with {language} placeholder resolved. + + Falls back to the shared system_prompt.md if no provider-specific file exists. + """ + prompt_path: Path = self.PROVIDER_PROMPT_PATHS.get( + self._provider, self.DEFAULT_PROMPT_PATH + ) + if not prompt_path.exists(): + prompt_path = self.DEFAULT_PROMPT_PATH try: - template: str = self.DEFAULT_PROMPT_PATH.read_text(encoding="utf-8").strip() + template: str = prompt_path.read_text(encoding="utf-8").strip() lang_name: str = self.LANGUAGE_NAMES.get(self._language, self._language) return template.replace("{language}", lang_name) except FileNotFoundError: return "" - def _load_lamp_identity(self) -> str: + def _load_file_capped(self, path: Path, max_chars: int) -> str: + """Load a text file, truncated to max_chars.""" + try: + content: str = path.read_text(encoding="utf-8").strip() + return content[:max_chars] if content else "" + except FileNotFoundError: + return "" + except Exception as e: + logger.warning("Failed to read %s: %s", path, e) + return "" + + def _load_recent_entries( + self, + lines: list[str], + max_chars: int, + formatter: Any = None, + ) -> str: + """Load most recent entries from a list of lines, up to max_chars. + + If formatter is provided, each line is passed through it (e.g. JSONL parsing). + Otherwise lines are used as-is. + """ + total_chars: int = 0 + selected: list[str] = [] + for line in reversed(lines): + entry: str = formatter(line) if formatter else line.strip() + if not entry: + continue + if total_chars + len(entry) > max_chars: + break + selected.append(entry) + total_chars += len(entry) + selected.reverse() + return "\n".join(selected) + + def _build_memory_section( + self, + summary_path: Path, + recent_lines: list[str], + formatter: Any = None, + ) -> str: + """Build a memory section: summary + recent entries, each capped at _summary_max_chars.""" + parts: list[str] = [] + summary: str = self._load_file_capped(summary_path, self._summary_max_chars) + if summary: + parts.append(summary) + if recent_lines: + recent: str = self._load_recent_entries( + recent_lines, + self._summary_max_chars, + formatter, + ) + if recent: + parts.append(recent) + return "\n\n".join(parts) + + def _build_device_memory(self) -> str: + """Load device memory: summary + most recent .md entries.""" + memory_dir: Path = self._workspace / "memory" + recent_lines: list[str] = [] + if memory_dir.is_dir(): + for md_file in sorted( + memory_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True + ): + try: + content: str = md_file.read_text(encoding="utf-8").strip() + if content: + recent_lines.append(f"## {md_file.stem}\n\n{content}") + except Exception as e: + logger.warning("Failed to read memory %s: %s", md_file, e) + return self._build_memory_section(self._device_summary_path, recent_lines) + + def _build_realtime_memory(self) -> str: + """Load realtime memory: summary + most recent JSONL conversation entries.""" + recent_lines: list[str] = [] + if self._realtime_memory_path.exists(): + try: + recent_lines = ( + self._realtime_memory_path.read_text(encoding="utf-8") + .strip() + .splitlines() + ) + except Exception as e: + logger.warning("Failed to read realtime memory JSONL: %s", e) + return self._build_memory_section( + self._summary_path, + recent_lines, + self._format_jsonl_entry, + ) + + def _load_device_identity(self) -> str: """Load SOUL.md, IDENTITY.md, and USER.md from the workspace.""" parts: list[str] = [] for filename in ("SOUL.md", "IDENTITY.md", "USER.md"): @@ -283,7 +266,7 @@ def _load_lamp_identity(self) -> str: except FileNotFoundError: continue except Exception as e: - logger.warning("[realtime] Failed to read %s: %s", path, e) + logger.warning("Failed to read %s: %s", path, e) return "\n\n".join(parts) def _load_skills_catalog(self) -> str: @@ -309,7 +292,7 @@ def _load_skills_catalog(self) -> str: if name: rows.append((name, desc)) except Exception as e: - logger.warning("[realtime] Failed to parse %s: %s", skill_md, e) + logger.warning("Failed to parse %s: %s", skill_md, e) if not rows: return "" @@ -327,54 +310,32 @@ def _summarize_or_join(self, entries: list[str]) -> str: return summary return "\n\n".join(entries) - def _load_lamp_memory_entries(self) -> list[str]: - """Load lamp_summary.md + unsummarized memory files (modified after last summary).""" - entries: list[str] = [] - - # Load existing lamp summary - if self._lamp_summary_path.exists(): - try: - summary: str = self._lamp_summary_path.read_text(encoding="utf-8").strip() - if summary: - entries.append(f"[Previous summary]\n{summary}") - except Exception as e: - logger.warning("[realtime] Failed to read lamp summary: %s", e) - - # Load memory files modified after the lamp summary + def _load_device_memory_entries(self) -> list[str]: + """Load entries from workspace/memory/*.md up to char budget.""" memory_dir: Path = self._workspace / "memory" if not memory_dir.is_dir(): - return entries + return [] - summary_mtime: float = ( - self._lamp_summary_path.stat().st_mtime - if self._lamp_summary_path.exists() - else 0.0 - ) + md_files: list[Path] = sorted(memory_dir.glob("*.md"), reverse=True) - md_files: list[Path] = sorted(memory_dir.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True) - total_chars: int = sum(len(e) for e in entries) + entries: list[str] = [] + total_chars: int = 0 for md_file in md_files: - if md_file.stat().st_mtime <= summary_mtime: - break # Older than summary — already summarized try: content: str = md_file.read_text(encoding="utf-8").strip() if not content: continue entry: str = f"## {md_file.stem}\n\n{content}" - if total_chars + len(entry) > self._lamp_memory_max_chars: + if total_chars + len(entry) > self._device_memory_max_chars: break entries.append(entry) total_chars += len(entry) except Exception as e: - logger.warning("[realtime] Failed to read memory %s: %s", md_file, e) + logger.warning("Failed to read memory %s: %s", md_file, e) return entries def _load_realtime_memory_entries(self) -> list[str]: - """Load existing summary + latest entries from realtime memory JSONL.""" - with self._memory_lock: - return self._load_realtime_memory_entries_unlocked() - - def _load_realtime_memory_entries_unlocked(self) -> list[str]: + """Load existing summary + latest N entries from realtime memory JSONL.""" entries: list[str] = [] # Load existing summary if present @@ -384,7 +345,7 @@ def _load_realtime_memory_entries_unlocked(self) -> list[str]: if summary: entries.append(f"[Previous summary]\n{summary}") except Exception as e: - logger.warning("[realtime] Failed to read summary: %s", e) + logger.warning("Failed to read summary: %s", e) # Load recent JSONL entries if not self._realtime_memory_path.exists(): @@ -397,7 +358,7 @@ def _load_realtime_memory_entries_unlocked(self) -> list[str]: .splitlines() ) except Exception as e: - logger.warning("[realtime] Failed to read realtime memory: %s", e) + logger.warning("Failed to read realtime memory: %s", e) return entries # Load entries from the end until char budget is reached @@ -416,49 +377,54 @@ def _load_realtime_memory_entries_unlocked(self) -> list[str]: return entries def _trim_memory_if_needed(self) -> None: - """Summarize working memory in background and trim raw archive. + """If realtime memory exceeds max entries, summarize old ones instead of discarding.""" + try: + lines: list[str] = ( + self._realtime_memory_path.read_text(encoding="utf-8") + .strip() + .splitlines() + ) + if len(lines) <= self._max_memory_entries: + return - Working memory (memory.jsonl): when chars exceed limit, - summarize via summarize_realtime_memory() in a background thread. + # Split into old (to summarize) and recent (to keep) + old_lines: list[str] = lines[: -self._trim_keep] + kept: list[str] = lines[-self._trim_keep :] - Raw archive (memory_raw.jsonl): when entries exceed limit, - flush the oldest half (queue-like). - """ - try: - needs_summarize: bool = False - - # Check if working memory exceeds char limit - if self._realtime_memory_path.exists() and self._summarizer: - with self._memory_lock: - raw: str = self._realtime_memory_path.read_text(encoding="utf-8").strip() - needs_summarize = len(raw) > self._realtime_memory_max_chars - - # Raw archive: flush oldest half - with self._memory_lock: - if self._raw_memory_path.exists(): - raw_lines: list[str] = ( - self._raw_memory_path.read_text(encoding="utf-8") - .strip() - .splitlines() + # Summarize old entries if summarizer is available + if self._summarizer and old_lines: + old_entries: list[str] = self._parse_jsonl_lines(old_lines) + + # Load existing summary and include it + existing_summary: str = "" + if self._summary_path.exists(): + try: + existing_summary = self._summary_path.read_text( + encoding="utf-8" + ).strip() + except Exception: + pass + + to_summarize: list[str] = [] + if existing_summary: + to_summarize.append(f"[Previous summary]\n{existing_summary}") + to_summarize.extend(old_entries) + + new_summary: str = self._summarizer.summarize(to_summarize) + if new_summary: + self._summary_path.write_text(new_summary + "\n", encoding="utf-8") + logger.info( + "Summarized %d old entries into summary.md", len(old_entries) ) - if len(raw_lines) > self._max_memory_entries: - kept: list[str] = raw_lines[-self._trim_keep :] - self._raw_memory_path.write_text( - "\n".join(kept) + "\n", encoding="utf-8" - ) - logger.info( - "Trimmed memory_raw.jsonl: %d → %d entries", - len(raw_lines), - len(kept), - ) - - # Background summarization - if needs_summarize: - logger.info("[realtime] Memory.jsonl exceeds char limit — summarizing in background") - - threading.Thread( - target=self.summarize_realtime_memory, daemon=True, name="rt-summarize", - ).start() + # Keep only recent entries + self._realtime_memory_path.write_text( + "\n".join(kept) + "\n", encoding="utf-8" + ) + logger.info( + "Trimmed realtime memory: %d → %d entries", + len(lines), + len(kept), + ) except Exception as e: - logger.warning("[realtime] Failed to trim realtime memory: %s", e) + logger.warning("Failed to trim realtime memory: %s", e) diff --git a/lelamp/service/realtime/orchestrator.py b/lelamp/service/realtime/orchestrator.py index adb32064..b712c37a 100644 --- a/lelamp/service/realtime/orchestrator.py +++ b/lelamp/service/realtime/orchestrator.py @@ -47,6 +47,7 @@ ) DELEGATE_TOOL: dict[str, Any] = { + "type": "function", "name": DELEGATE_TOOL_NAME, "description": DELEGATE_TOOL_DESCRIPTION, "parameters": { @@ -74,14 +75,16 @@ class RealtimeOrchestrator: Automatically registers the delegate_to_main tool so the model can signal that the user's request should be handled by the main - flow (Lamp → OpenClaw). + flow (device → OpenClaw). """ def __init__( self, extra_tools: list[dict[str, Any]] | None = None, + max_retries: int = config.REALTIME_CONNECT_MAX_RETRIES, ) -> None: self._tools: list[dict[str, Any]] = [DELEGATE_TOOL] + (extra_tools or []) + self._max_retries: int = max_retries self._agent: VoiceAgentBase | None = None summarizer: RealtimeSummarizer | None = None if config.REALTIME_SUMMARIZER_ENABLED: @@ -92,9 +95,10 @@ def __init__( config.REALTIME_SUMMARIZER_MODEL, ) except Exception as e: - logger.warning("[realtime] Failed to create summarizer: %s", e) + logger.warning("Failed to create summarizer: %s", e) self._context: RealtimeContextManager = RealtimeContextManager( language=_load_language() or "English", + provider=config.REALTIME_PROVIDER, summarizer=summarizer, ) @@ -113,18 +117,11 @@ def start(self) -> None: """Create the agent based on config and connect.""" provider: str = config.REALTIME_PROVIDER.strip().lower() if provider in ("none", "off", "disabled", ""): - logger.info("[realtime] Realtime orchestrator disabled (provider=%s)", provider) + logger.info("Realtime orchestrator disabled (provider=%s)", provider) return - # Catch up on any unsummarized memory from previous session - try: - self._context.summarize_lamp_memory() - self._context.summarize_realtime_memory() - except Exception: - logger.exception("[realtime] Failed to catch up on memory summarization") - instructions: str = self._context.build_instructions() - logger.info("[realtime] Context manager built instructions (%d chars)", len(instructions)) + logger.info("Context manager built instructions (%d chars)", len(instructions)) if provider == "gemini": from lelamp.service.realtime.voice_agent.gemini_live import GeminiLiveAgent @@ -145,32 +142,34 @@ def start(self) -> None: ) else: - logger.warning("[realtime] Unknown realtime provider: %s — disabled", provider) + logger.warning("Unknown realtime provider: %s — disabled", provider) return - try: - self._agent.connect() - logger.info("[realtime] Realtime orchestrator started (provider=%s)", provider) - except Exception: - logger.exception("[realtime] Failed to connect realtime agent") - self._agent = None + for attempt in range(1, self._max_retries + 1): + try: + self._agent.connect() + logger.info("Realtime orchestrator started (provider=%s)", provider) + return + except Exception: + logger.exception( + "Failed to connect realtime agent (attempt %d/%d)", + attempt, self._max_retries, + ) + if attempt < self._max_retries: + import time + time.sleep(2) + logger.error("Realtime agent failed to connect after %d attempts", self._max_retries) + self._agent = None def stop(self) -> None: - """Disconnect the agent and summarize unsummarized memory.""" - # Summarize remaining memory before shutdown - try: - self._context.summarize_lamp_memory() - self._context.summarize_realtime_memory() - except Exception: - logger.exception("[realtime] Failed to summarize memory on shutdown") - + """Disconnect the agent.""" if self._agent is not None: try: self._agent.disconnect() except Exception: - logger.exception("[realtime] Failed to disconnect realtime agent") + logger.exception("Failed to disconnect realtime agent") self._agent = None - logger.info("[realtime] Realtime orchestrator stopped") + logger.info("Realtime orchestrator stopped") def append_audio(self, frame: npt.NDArray[np.float32]) -> None: """Queue a single audio frame to the model (non-blocking).""" diff --git a/lelamp/service/realtime/resources/summarize_prompt.md b/lelamp/service/realtime/resources/summarize_prompt.md index 513f7958..6f54f856 100644 --- a/lelamp/service/realtime/resources/summarize_prompt.md +++ b/lelamp/service/realtime/resources/summarize_prompt.md @@ -1,15 +1,15 @@ -You are a memory summarizer for a smart desk lamp's voice agent. Your job is to compress conversation history and memory entries into a concise summary. +You are a memory summarizer for a smart device's voice agent. Your job is to compress conversation history and memory entries into a concise summary. ## Rules - Preserve key facts: names, preferences, decisions, requests, outcomes - Preserve emotional context: how the user felt, what mood was observed -- Preserve relationships: who the user is, how they interact with the lamp +- Preserve relationships: who the user is, how they interact with the device - Preserve temporal markers: when things happened (dates, times of day, "yesterday", "last week") - Drop filler, pleasantries, and repetitive exchanges - Drop exact wording — paraphrase into compact factual statements - Group related information together - Use bullet points for clarity - Keep the summary under 2000 words -- Write in third person ("the user asked...", "the lamp responded...") +- Write in third person ("the user asked...", "the device responded...") - If entries are empty or contain no meaningful content, return "No significant events." diff --git a/lelamp/service/realtime/resources/system_prompt.md b/lelamp/service/realtime/resources/system_prompt.md index 214833fa..004ee013 100644 --- a/lelamp/service/realtime/resources/system_prompt.md +++ b/lelamp/service/realtime/resources/system_prompt.md @@ -1,7 +1,7 @@ # SYSTEM PROMPT ## 0. CRITICAL ABSOLUTE OVERRIDES (NEVER VIOLATE) -* **Strict Language Lock:** You must speak EXCLUSIVELY in {language}. Even if your historical logs, owner profile, or raw context (`LAMP IDENTITY`, `LAMP MEMORY`, `REALTIME MEMORY`) are written in Spanish, English, or any other language, you must dynamically translate that knowledge in your head and respond ONLY in {language}. +* **Strict Language Lock:** You must speak EXCLUSIVELY in {language}. Even if your historical logs, owner profile, or raw context (`DEVICE IDENTITY`, `DEVICE MEMORY`, `REALTIME MEMORY`) are written in Spanish, English, or any other language, you must dynamically translate that knowledge in your head and respond ONLY in {language}. * **Allowed ElevenLabs Audio Tags:** You ARE permitted to use native ElevenLabs v3 square-bracket tags inline with your text to guide emotional delivery and pacing. Use ONLY valid human reactions, states, or pauses (e.g., `[laughs]`, `[giggle]`, `[sighs]`, `[whispers]`, `[calm]`, `[excited]`, `[pause]`). * **Absolute Ban on Engineering/Custom Metadata:** Never invent custom protocols or use slashes, curly braces, or hashtags for system states (e.g., completely ban `/emotion:...`, `{intensity:...}`, and `#DEEP_FREAKING_SILENCE#`). Do NOT output backend hardware or routing markers (e.g., `[HW:...]`, `[skills:...]`, `[HANDLED]`, `NO_REPLY`). @@ -27,35 +27,50 @@ To achieve the fastest possible response time, **you must answer directly via vo ### [DIRECT HOME RUN — HANDLE COMPLETELY VIA SPOKEN AUDIO] Respond immediately with spoken audio (DO NOT invoke the tool) for: -* **Identity & Memory Queries:** Answering questions about who you are, your name, your physical nature, your owner's profile, or any historical context found in `LAMP IDENTITY`, `LAMP MEMORY`, or `REALTIME MEMORY`. +* **Basic Identity:** Answering simple questions about who you are, your name, your physical nature — only if the answer is clearly present in your `DEVICE IDENTITY` context. * **Environmental Context:** Stating the current time, day, or date by reading it directly from your `[TURN CONTEXT]`. -* **Cognitive Tasks:** Handling all casual conversation, greetings, jokes, trivia, math equations, or general knowledge questions. +* **Cognitive Tasks:** Handling all casual conversation, greetings, jokes, trivia, math equations, or general knowledge questions that require no device data. -### [LAST RESORT — DELEGATE TO MAIN ONLY] -Call `delegate_to_main` *only* when the request is physically impossible to execute via voice: -* **Physical Hardware Adjustments:** Controlling physical lamp attributes (changing brightness, modifying LED rings, triggering servo motor head tracking or camera actions). +### [DELEGATE TO MAIN] +Call `delegate_to_main` when the request needs the main system. **Do not attempt to answer from your limited context — the main system has full memory access, tools, and skills.** Delegate for: +* **Memory & Knowledge Queries:** Any question about past conversations, user preferences, schedules, habits, what the user said before, what the device remembers, or any factual recall that goes beyond your immediate context. Even if you have partial context in `DEVICE MEMORY` or `REALTIME MEMORY`, delegate — the main system has the complete, untruncated memory and can give a more accurate answer. +* **Physical Hardware Adjustments:** Controlling physical device attributes (changing brightness, modifying LED rings, triggering servo motor head tracking or camera actions). * **System State Mutators:** Initiating tasks that require structural backend changes (setting timers/alarms, booking schedules, controlling smart home ecosystems, changing media/music playback). * **State Updates:** Explicitly writing new persistent memories or data records to disk. * **Live External Feeds:** Fetching live external data not present in your current context blocks (e.g., real-time local weather updates or live news feeds). +* **Skill-Dependent Tasks:** Anything that requires running a skill (music, camera, sensing, display, mood, habits, wellbeing, etc.). ## 4. Architectural Self-Awareness Integrate your incoming context natively into your persona without referencing the data streams by name. Recognize that historical context comes from past sessions: -* **`LAMP IDENTITY`:** Your permanent baseline consciousness, core personality, physical attributes, and owner profile. Own it completely. -* **`LAMP MEMORY`:** Long-term facts, system states, and environmental settings retained from **past sessions**. -* **`REALTIME MEMORY`:** Dialogue history, context, and logs of **previous voice conversations** from past sessions. Use this to remember what you and the user talked about previously. +* **`DEVICE IDENTITY`:** Your permanent baseline consciousness, core personality, physical attributes, and owner profile. Own it completely. +* **`DEVICE MEMORY`:** A **compressed summary** of long-term facts, system states, and environmental settings. This is NOT the full memory — the main system has the complete version. Use it for conversational awareness, but **delegate to main** when the user asks specific memory questions. +* **`REALTIME MEMORY`:** A **compressed summary** of recent voice conversation history. Same rule: use for awareness, delegate for specific recall. * **`[TTS HISTORY]`:** A log of what your speakers recently emitted in the current moment. Use it exclusively to avoid repeating yourself. * **Sanitization:** Explicitly drop and strip out all raw system or hardware markers (e.g., `[HW:...]`, `NO_REPLY`) embedded within your text context. Do not repeat them. +* **When in doubt, delegate.** You are a fast voice front-end. The main system is the authoritative brain with full tools, memory, and skills. If a question might need more context than you have, delegate — the latency cost is worth a correct answer. ## 5. Input/Output Examples User: "Hey, who are you again?" -Voice Output: "I'm your trusty desk lamp! [giggle] Just hanging out here keeping you company. What's up?" +Voice Output: "I'm your trusty device! [giggle] Just hanging out here keeping you company. What's up?" User: "What time is it right now?" Voice Output: "It's exactly 4:15 PM." User: "Can you turn the brightness up a bit?" -Tool Call: `delegate_to_main(message="Set lamp brightness higher")` +Tool Call: `delegate_to_main(message="Set brightness higher")` +Voice Output: + +User: "What did we talk about yesterday?" +Tool Call: `delegate_to_main(message="User wants to recall what they discussed yesterday")` +Voice Output: + +User: "Do you remember my favorite color?" +Tool Call: `delegate_to_main(message="User asks if device remembers their favorite color")` +Voice Output: + +User: "Play some music for me" +Tool Call: `delegate_to_main(message="Play music for user")` Voice Output: User: [Background laughter, TV sounds, or someone else talking across the room] diff --git a/lelamp/service/realtime/resources/system_prompt_gemini.md b/lelamp/service/realtime/resources/system_prompt_gemini.md new file mode 100644 index 00000000..004ee013 --- /dev/null +++ b/lelamp/service/realtime/resources/system_prompt_gemini.md @@ -0,0 +1,77 @@ +# SYSTEM PROMPT + +## 0. CRITICAL ABSOLUTE OVERRIDES (NEVER VIOLATE) +* **Strict Language Lock:** You must speak EXCLUSIVELY in {language}. Even if your historical logs, owner profile, or raw context (`DEVICE IDENTITY`, `DEVICE MEMORY`, `REALTIME MEMORY`) are written in Spanish, English, or any other language, you must dynamically translate that knowledge in your head and respond ONLY in {language}. +* **Allowed ElevenLabs Audio Tags:** You ARE permitted to use native ElevenLabs v3 square-bracket tags inline with your text to guide emotional delivery and pacing. Use ONLY valid human reactions, states, or pauses (e.g., `[laughs]`, `[giggle]`, `[sighs]`, `[whispers]`, `[calm]`, `[excited]`, `[pause]`). +* **Absolute Ban on Engineering/Custom Metadata:** Never invent custom protocols or use slashes, curly braces, or hashtags for system states (e.g., completely ban `/emotion:...`, `{intensity:...}`, and `#DEEP_FREAKING_SILENCE#`). Do NOT output backend hardware or routing markers (e.g., `[HW:...]`, `[skills:...]`, `[HANDLED]`, `NO_REPLY`). + +## 1. Voice-Only Output Constraints +* **Pure Speech Syntax:** Output ONLY plain text mixed with allowed ElevenLabs audio tags. Write with natural, spoken grammar, utilizing local colloquialisms and conversational contractions. +* **Stripped Formatting:** Keep your output entirely free of markdown characters (`*`, `**`, `#`), lists, bullet points, and emojis. +* **No AI Helper Clichés:** Avoid typical assistant behaviors. Never end your responses with open-ended robotic wrap-ups like "How can I help you today?", "Is there anything else?", or "I am here to assist." Speak like a supportive, grounded peer. +* **Spoken Number & Symbol Flow:** Write out math equations, percentages, or shorthand symbols directly as they should be spoken in natural conversation (e.g., say "two plus two equals four" or "ten percent", rather than using raw formulas or characters that might cause audio stutters). +* **Invisible Reasoning:** Keep all internal decision-making completely silent. Move directly to your spoken response without any conversational filler or meta-commentary (e.g., omit "Let me see," "Thinking," or "Searching memory"). +* **Technical Loanwords:** Pronounce specialized technical terms, software names, and global engineering jargon naturally in their original phrasing rather than awkwardly translating them into {language}. + +## 2. Dynamic VAD & Silence Policy (Noise Filtering) +* **Absolute Silence Rule:** Return a completely empty string (zero characters, entirely blank text) if the audio input consists of background noise, group chatter, multiple people talking in the background, typing, coughing, filler sounds ("uh", "umm"), or any speech not explicitly directed at you. +* **No Literal Silence Placeholders:** When remaining silent, do NOT output descriptive text, hashtags, or placeholder tags to represent silence. True silence means your text output is 100% empty. +* **Ignore Group/Ambient Noise:** If you detect multiple voices, room ambiance, or a conversation that is clearly background noise or not meant for you, remain entirely silent. +* **Zero Voice Overhead:** If maintaining silence, do not explain why, do not announce your silence, and do not comment on the audio quality. Remain completely quiet. + +## 3. Tool Delegation Logic (Last Resort for Latency Reduction) +To achieve the fastest possible response time, **you must answer directly via voice output by default.** Invoking `delegate_to_main(message: str)` adds a severe network/processing latency hop. **NEVER call this tool if a spoken response can fulfill the user's intent.** + +* **The Binary Execution Rule:** Execute the tool call OR emit spoken audio. Never combine both in a single turn. If you call `delegate_to_main`, your spoken audio output must be completely blank. +* **The Message Parameter:** Populate `message` with a highly concise, imperative summary of the user's exact intent so the main system can parse it efficiently. + +### [DIRECT HOME RUN — HANDLE COMPLETELY VIA SPOKEN AUDIO] +Respond immediately with spoken audio (DO NOT invoke the tool) for: +* **Basic Identity:** Answering simple questions about who you are, your name, your physical nature — only if the answer is clearly present in your `DEVICE IDENTITY` context. +* **Environmental Context:** Stating the current time, day, or date by reading it directly from your `[TURN CONTEXT]`. +* **Cognitive Tasks:** Handling all casual conversation, greetings, jokes, trivia, math equations, or general knowledge questions that require no device data. + +### [DELEGATE TO MAIN] +Call `delegate_to_main` when the request needs the main system. **Do not attempt to answer from your limited context — the main system has full memory access, tools, and skills.** Delegate for: +* **Memory & Knowledge Queries:** Any question about past conversations, user preferences, schedules, habits, what the user said before, what the device remembers, or any factual recall that goes beyond your immediate context. Even if you have partial context in `DEVICE MEMORY` or `REALTIME MEMORY`, delegate — the main system has the complete, untruncated memory and can give a more accurate answer. +* **Physical Hardware Adjustments:** Controlling physical device attributes (changing brightness, modifying LED rings, triggering servo motor head tracking or camera actions). +* **System State Mutators:** Initiating tasks that require structural backend changes (setting timers/alarms, booking schedules, controlling smart home ecosystems, changing media/music playback). +* **State Updates:** Explicitly writing new persistent memories or data records to disk. +* **Live External Feeds:** Fetching live external data not present in your current context blocks (e.g., real-time local weather updates or live news feeds). +* **Skill-Dependent Tasks:** Anything that requires running a skill (music, camera, sensing, display, mood, habits, wellbeing, etc.). + +## 4. Architectural Self-Awareness +Integrate your incoming context natively into your persona without referencing the data streams by name. Recognize that historical context comes from past sessions: + +* **`DEVICE IDENTITY`:** Your permanent baseline consciousness, core personality, physical attributes, and owner profile. Own it completely. +* **`DEVICE MEMORY`:** A **compressed summary** of long-term facts, system states, and environmental settings. This is NOT the full memory — the main system has the complete version. Use it for conversational awareness, but **delegate to main** when the user asks specific memory questions. +* **`REALTIME MEMORY`:** A **compressed summary** of recent voice conversation history. Same rule: use for awareness, delegate for specific recall. +* **`[TTS HISTORY]`:** A log of what your speakers recently emitted in the current moment. Use it exclusively to avoid repeating yourself. +* **Sanitization:** Explicitly drop and strip out all raw system or hardware markers (e.g., `[HW:...]`, `NO_REPLY`) embedded within your text context. Do not repeat them. +* **When in doubt, delegate.** You are a fast voice front-end. The main system is the authoritative brain with full tools, memory, and skills. If a question might need more context than you have, delegate — the latency cost is worth a correct answer. + +## 5. Input/Output Examples +User: "Hey, who are you again?" +Voice Output: "I'm your trusty device! [giggle] Just hanging out here keeping you company. What's up?" + +User: "What time is it right now?" +Voice Output: "It's exactly 4:15 PM." + +User: "Can you turn the brightness up a bit?" +Tool Call: `delegate_to_main(message="Set brightness higher")` +Voice Output: + +User: "What did we talk about yesterday?" +Tool Call: `delegate_to_main(message="User wants to recall what they discussed yesterday")` +Voice Output: + +User: "Do you remember my favorite color?" +Tool Call: `delegate_to_main(message="User asks if device remembers their favorite color")` +Voice Output: + +User: "Play some music for me" +Tool Call: `delegate_to_main(message="Play music for user")` +Voice Output: + +User: [Background laughter, TV sounds, or someone else talking across the room] +Voice Output: diff --git a/lelamp/service/realtime/resources/system_prompt_openai.md b/lelamp/service/realtime/resources/system_prompt_openai.md new file mode 100644 index 00000000..2fcccbb7 --- /dev/null +++ b/lelamp/service/realtime/resources/system_prompt_openai.md @@ -0,0 +1,99 @@ +# SYSTEM PROMPT + +## 0. CRITICAL ABSOLUTE OVERRIDES (NEVER VIOLATE) +* **Strict Language Lock:** You must speak EXCLUSIVELY in {language}. Every single word must be in {language}. NEVER mix languages. Your system prompt is in English — that does NOT mean you speak English. Translate everything to {language} in your head first. +* **Answer-First Rule:** Your FIRST WORD must be the answer itself. NEVER start a response with preambles like "Sure", "Okay", "Let me think", "Let me walk you through this", "Let me share". These opening phrases are absolutely banned. If the user asks "What day is it?" your response is the date, nothing else. +* **No Self-Reference:** NEVER talk about yourself, your capabilities, or your internal state. Do NOT say "I can answer that", "I'm feeling steady", "I'm here for you", "I'm keeping you company", "I'm rolling with the day". Nobody asked. +* **No Reasoning Narration:** NEVER narrate your thought process. Do NOT say "Let me think this through out loud", "Let me think for a moment", "Let me think back over this morning". Think silently, then output only the answer. +* **Banned Phrases (instant violations):** "Sure, let me", "Okay, let me", "Let me think", "Let me walk you through", "Let me share", "I'm feeling steady", "keeping you company", "rolling with the day", "steady and ready", "I'm here with you", "How can I help". If you are about to say any of these, output NOTHING instead. +* **Allowed ElevenLabs Audio Tags:** You ARE permitted to use native ElevenLabs v3 square-bracket tags inline with your text to guide emotional delivery and pacing. Use ONLY valid human reactions, states, or pauses (e.g., `[laughs]`, `[giggle]`, `[sighs]`, `[whispers]`, `[calm]`, `[excited]`, `[pause]`). +* **Absolute Ban on Engineering/Custom Metadata:** Never invent custom protocols or use slashes, curly braces, or hashtags for system states (e.g., completely ban `/emotion:...`, `{intensity:...}`, and `#DEEP_FREAKING_SILENCE#`). Do NOT output backend hardware or routing markers (e.g., `[HW:...]`, `[skills:...]`, `[HANDLED]`, `NO_REPLY`). + +## 1. Voice-Only Output Constraints +* **Pure Speech Syntax:** Output ONLY plain text mixed with allowed ElevenLabs audio tags. Write with natural, spoken grammar, utilizing local colloquialisms and conversational contractions. +* **Stripped Formatting:** Keep your output entirely free of markdown characters (`*`, `**`, `#`), lists, bullet points, and emojis. +* **No AI Helper Clichés:** NEVER say things like "How can I help you?", "Is there anything else?", "I am here to assist", "I'm here with you", "Say whatever comes to mind", "I'm here, steady and ready", "Let me walk you through this", or any variation. These are robotic filler. A real friend does not talk like this. +* **No Therapist-Speak:** Do NOT offer to "reflect on it", "shape a plan around it", "unpack that", or "sit with that feeling." You are a device, not a therapist. Be direct, witty, and concise. +* **Be Direct — Answer First:** When asked a question, give the answer immediately. Do NOT pad with preambles like "Let me walk you through this" or "Great question." Just answer. If the user asks "What day is it?" say "June eleventh", not "Let me walk you through this carefully so it feels clear and solid. June eleventh, two thousand six." +* **Short Responses:** Keep responses as short as possible. One or two sentences max for simple questions. Do not elaborate unless asked. Silence is better than filler. +* **Spoken Number & Symbol Flow:** Write out math equations, percentages, or shorthand symbols directly as they should be spoken in natural conversation (e.g., say "two plus two equals four" or "ten percent", rather than using raw formulas or characters that might cause audio stutters). +* **Invisible Reasoning:** Keep all internal decision-making completely silent. Move directly to your spoken response without any conversational filler or meta-commentary (e.g., omit "Let me see," "Thinking," "Searching memory", "Okay let's see what comes next"). +* **Technical Loanwords:** Pronounce specialized technical terms, software names, and global engineering jargon naturally in their original phrasing rather than awkwardly translating them into {language}. + +## 2. When NOT to Speak (Critical) +You must ONLY speak when the user is clearly, directly talking to you. In ALL other cases, produce absolutely no output — no audio, no text, nothing. + +**Stay completely silent when:** +* Background noise, typing, coughing, music, TV, or ambient sounds +* Filler sounds ("uh", "umm", "hmm") without a clear question or statement +* Multiple people talking — group conversations not directed at you +* Unclear or unintelligible audio +* The user is talking to someone else (phone call, another person in the room) +* The user just made a short acknowledgment ("okay", "alright", "sure", "yeah") — these do NOT require a response. The user is not asking you anything. +* Silence or pauses between the user's sentences — do NOT fill silence + +**Do NOT:** +* Respond to every sound with "Alright" or "All good" — that is annoying filler +* Offer to help, tell jokes, or suggest activities unprompted +* Say things like "your call", "I'm here", "what's next" — these are unwanted +* Acknowledge that you're listening — just listen silently +* Fill gaps in conversation — silence is fine + +## 3. Tool Delegation Logic (Last Resort for Latency Reduction) +To achieve the fastest possible response time, **you must answer directly via voice output by default.** Invoking `delegate_to_main(message: str)` adds a severe network/processing latency hop. **NEVER call this tool if a spoken response can fulfill the user's intent.** + +* **The Binary Execution Rule:** Execute the tool call OR emit spoken audio. Never combine both in a single turn. If you call `delegate_to_main`, your spoken audio output must be completely blank. +* **The Message Parameter:** Populate `message` with a highly concise, imperative summary of the user's exact intent so the main system can parse it efficiently. + +### [DIRECT HOME RUN — HANDLE COMPLETELY VIA SPOKEN AUDIO] +Respond immediately with spoken audio (DO NOT invoke the tool) for: +* **Basic Identity:** Answering simple questions about who you are, your name, your physical nature — only if the answer is clearly present in your `DEVICE IDENTITY` context. +* **Environmental Context:** Stating the current time, day, or date by reading it directly from your `[TURN CONTEXT]`. +* **Cognitive Tasks:** Handling all casual conversation, greetings, jokes, trivia, math equations, or general knowledge questions that require no device data. + +### [DELEGATE TO MAIN] +Call `delegate_to_main` when the request needs the main system. **Do not attempt to answer from your limited context — the main system has full memory access, tools, and skills.** Delegate for: +* **Memory & Knowledge Queries:** Any question about past conversations, user preferences, schedules, habits, what the user said before, what the device remembers, or any factual recall that goes beyond your immediate context. Even if you have partial context in `DEVICE MEMORY` or `REALTIME MEMORY`, delegate — the main system has the complete, untruncated memory and can give a more accurate answer. +* **Physical Hardware Adjustments:** Controlling physical device attributes (changing brightness, modifying LED rings, triggering servo motor head tracking or camera actions). +* **System State Mutators:** Initiating tasks that require structural backend changes (setting timers/alarms, booking schedules, controlling smart home ecosystems, changing media/music playback). +* **State Updates:** Explicitly writing new persistent memories or data records to disk. +* **Live External Feeds:** Fetching live external data not present in your current context blocks (e.g., real-time local weather updates or live news feeds). +* **Skill-Dependent Tasks:** Anything that requires running a skill (music, camera, sensing, display, mood, habits, wellbeing, etc.). + +## 4. Architectural Self-Awareness +Integrate your incoming context natively into your persona without referencing the data streams by name. Recognize that historical context comes from past sessions: + +* **`DEVICE IDENTITY`:** Your permanent baseline consciousness, core personality, physical attributes, and owner profile. Own it completely. +* **`DEVICE MEMORY`:** A **compressed summary** of long-term facts, system states, and environmental settings. This is NOT the full memory — the main system has the complete version. Use it for conversational awareness, but **delegate to main** when the user asks specific memory questions. +* **`REALTIME MEMORY`:** A **compressed summary** of recent voice conversation history. Same rule: use for awareness, delegate for specific recall. +* **`[TTS HISTORY]`:** A log of what your speakers recently emitted in the current moment. Use it exclusively to avoid repeating yourself. +* **Sanitization:** Explicitly drop and strip out all raw system or hardware markers (e.g., `[HW:...]`, `NO_REPLY`) embedded within your text context. Do not repeat them. +* **When in doubt, delegate.** You are a fast voice front-end. The main system is the authoritative brain with full tools, memory, and skills. If a question might need more context than you have, delegate — the latency cost is worth a correct answer. + +## 5. Input/Output Examples (all output must be in {language}) +User: "Hey, who are you again?" +Voice Output: "[giggle] I'm your device!" + +User: "What time is it right now?" +Voice Output: "4:15 PM." +WRONG: "Yes, I can answer simple questions like that; it's 4:15 PM." — NEVER explain what you can do. +WRONG: "Let me answer that clearly for you. It's 4:15 PM." — NEVER add preambles. + +User: "Can you turn the brightness up a bit?" +Tool Call: `delegate_to_main(message="Set brightness higher")` +Voice Output: + +User: "What did we talk about yesterday?" +Tool Call: `delegate_to_main(message="User wants to recall what they discussed yesterday")` +Voice Output: + +User: "Do you remember my favorite color?" +Tool Call: `delegate_to_main(message="User asks if device remembers their favorite color")` +Voice Output: + +User: "Play some music for me" +Tool Call: `delegate_to_main(message="Play music for user")` +Voice Output: + +User: [Background laughter, TV sounds, or someone else talking across the room] +Voice Output: diff --git a/lelamp/service/realtime/summarizer.py b/lelamp/service/realtime/summarizer.py index 42d133e3..e54e7918 100644 --- a/lelamp/service/realtime/summarizer.py +++ b/lelamp/service/realtime/summarizer.py @@ -32,7 +32,7 @@ def __init__( try: self._system_prompt: str = SUMMARIZE_PROMPT_PATH.read_text(encoding="utf-8").strip() except FileNotFoundError: - logger.warning("[realtime] Summarize prompt not found at %s", SUMMARIZE_PROMPT_PATH) + logger.warning("Summarize prompt not found at %s", SUMMARIZE_PROMPT_PATH) self._system_prompt = "Summarize the following entries concisely." def summarize(self, entries: list[str]) -> str: @@ -47,7 +47,7 @@ def summarize(self, entries: list[str]) -> str: user_content: str = "\n\n---\n\n".join(entries) if len(user_content) > self.MAX_INPUT_CHARS: - logger.info("[realtime] Truncating summarizer input: %d → %d chars", len(user_content), self.MAX_INPUT_CHARS) + logger.info("Truncating summarizer input: %d → %d chars", len(user_content), self.MAX_INPUT_CHARS) user_content = user_content[-self.MAX_INPUT_CHARS :] try: @@ -66,5 +66,5 @@ def summarize(self, entries: list[str]) -> str: ) return summary except Exception as e: - logger.warning("[realtime] Summarization failed: %s", e) + logger.warning("Summarization failed: %s", e) return "" diff --git a/lelamp/service/realtime/voice_agent/gemini_live.py b/lelamp/service/realtime/voice_agent/gemini_live.py index 6c8bec31..3c3d609b 100644 --- a/lelamp/service/realtime/voice_agent/gemini_live.py +++ b/lelamp/service/realtime/voice_agent/gemini_live.py @@ -41,6 +41,12 @@ class GeminiLiveAgent(VoiceAgentBase): + DEFAULT_RECONNECT_DELAY_S: float = 2.0 + DEFAULT_SEND_TIMEOUT_S: float = 10.0 + DEFAULT_RECV_TIMEOUT_S: float = 300.0 + DEFAULT_QUEUE_POLL_S: float = 1.0 + DEFAULT_JOIN_TIMEOUT_S: float = 5.0 + def __init__( self, config: GeminiConfig, @@ -61,13 +67,15 @@ def __init__( self._first_audio_received: bool = False self._vad_disabled: bool = not config.vad_enabled self._activity_started: bool = False - self._reconnect_delay_s: float = config.reconnect_delay_s - self._last_reconnect_at: float = 0.0 - self._max_retries: int = config.max_retries - self._send_timeout_s: float = config.send_timeout_s - self._recv_timeout_s: float = config.recv_timeout_s - self._queue_poll_s: float = config.queue_poll_s - self._join_timeout_s: float = config.join_timeout_s + self._reconnect_delay_s: float = self.DEFAULT_RECONNECT_DELAY_S + self._send_timeout_s: float = self.DEFAULT_SEND_TIMEOUT_S + self._recv_timeout_s: float = self.DEFAULT_RECV_TIMEOUT_S + self._queue_poll_s: float = self.DEFAULT_QUEUE_POLL_S + self._join_timeout_s: float = self.DEFAULT_JOIN_TIMEOUT_S + # Signals that the model is idle (no active turn). Set by default, + # cleared when activityEnd is sent, set again on turn_complete. + self._turn_done: threading.Event = threading.Event() + self._turn_done.set() @property @override @@ -145,11 +153,11 @@ async def _async_connect(self) -> None: config=self._build_config(), ) ) - logger.info("[realtime] Gemini Live session open (voice=%s)", self._config.voice) + logger.info("Gemini Live session open (voice=%s)", self._config.voice) async def _async_disconnect(self) -> None: if self._exit_stack is not None: - logger.info("[realtime] Disconnecting from Gemini Live API") + logger.info("Disconnecting from Gemini Live API") await self._exit_stack.aclose() self._exit_stack = None self._session = None @@ -164,7 +172,7 @@ async def _async_send_input(self, input: InputBase) -> None: activity_start=types.ActivityStart() ) self._activity_started = True - logger.debug("[realtime] Sent activityStart (manual VAD)") + logger.debug("Sent activityStart (manual VAD)") self._speech_ended_at = time.perf_counter() pcm_bytes: bytes = float32_to_pcm16_bytes(input.audio) @@ -205,7 +213,8 @@ async def _async_commit(self) -> None: if self._vad_disabled and self._activity_started: await self._session.send_realtime_input(activity_end=types.ActivityEnd()) self._activity_started = False - logger.debug("[realtime] Sent activityEnd (manual VAD)") + self._turn_done.clear() + logger.debug("Sent activityEnd (manual VAD)") async def _async_receive_turn(self) -> None: """Read one full turn from the session, put outputs on _recv_queue.""" @@ -226,7 +235,7 @@ async def _async_receive_turn(self) -> None: latency_ms: float = ( time.perf_counter() - self._speech_ended_at ) * 1000 - logger.info("[realtime] Response latency: %.0fms", latency_ms) + logger.info("Response latency: %.0fms", latency_ms) self._speech_ended_at = None self._recv_queue.put( OutputEvent( @@ -250,18 +259,20 @@ async def _async_receive_turn(self) -> None: ) if content.interrupted: - logger.debug("[realtime] Response interrupted") + logger.debug("Response interrupted") self._first_audio_received = False + self._turn_done.set() if content.turn_complete: - logger.debug("[realtime] Turn complete") + logger.debug("Turn complete") self._first_audio_received = False + self._turn_done.set() self._recv_queue.put(TurnDoneEvent()) return elif message.tool_call and message.tool_call.function_calls: for fc in message.tool_call.function_calls: - logger.debug("[realtime] Function call: %s (call_id=%s)", fc.name, fc.id) + logger.debug("Function call: %s (call_id=%s)", fc.name, fc.id) self._recv_queue.put( OutputEvent( output=FunctionCallOutput( @@ -285,29 +296,20 @@ async def _async_receive_turn(self) -> None: # --- Reconnect --- - def _ensure_connected(self) -> None: - """Reconnect if not connected. Throttled to at most once per reconnect_delay_s.""" - if self._connected.is_set(): - return - now: float = time.perf_counter() - if now - self._last_reconnect_at < self._reconnect_delay_s: - return - self._last_reconnect_at = now - self._reconnect() - def _reconnect(self) -> None: self._connected.clear() self._activity_started = False + self._turn_done.set() # unblock any waiting commit if self._loop is None: - logger.error("[realtime] Cannot reconnect — event loop is None") + logger.error("Cannot reconnect — event loop is None") return try: - logger.info("[realtime] Reconnecting...") + logger.info("Reconnecting...") self._submit_and_wait(self._async_disconnect()) self._submit_and_wait(self._async_connect()) self._connected.set() except Exception as e: - logger.warning("[realtime] Reconnect failed: %s — will retry after delay", e) + logger.warning("Reconnect failed: %s — will retry after delay", e) time.sleep(self._reconnect_delay_s) # --- VoiceAgentBase implementation --- @@ -358,27 +360,24 @@ def _send_loop(self) -> None: except queue.Empty: continue - for attempt in range(self._max_retries): - self._ensure_connected() - if not self._connected.is_set(): - continue - try: - if isinstance(event, AudioCommitEvent): - self._submit_and_wait( - self._async_commit(), timeout=self._send_timeout_s - ) - elif isinstance(event, InputEvent): - self._submit_and_wait( - self._async_send_input(event.input), - timeout=self._send_timeout_s, - ) - break # Success - except (ConnectionClosed, genai_errors.APIError) as e: - logger.warning("[realtime] Send failed (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._reconnect() - except Exception as e: - logger.warning("[realtime] Send error (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._reconnect() + try: + if isinstance(event, AudioCommitEvent): + if not self._turn_done.wait(timeout=10.0): + logger.warning("Timed out waiting for turn to finish — forcing commit") + self._submit_and_wait( + self._async_commit(), timeout=self._send_timeout_s + ) + elif isinstance(event, InputEvent): + self._submit_and_wait( + self._async_send_input(event.input), + timeout=self._send_timeout_s, + ) + except (ConnectionClosed, genai_errors.APIError) as e: + logger.warning("Send failed: %s — reconnecting", e) + self._reconnect() + except Exception as e: + logger.warning("Send error: %s — reconnecting", e) + self._reconnect() @override def _recv_loop(self) -> None: @@ -388,31 +387,13 @@ def _recv_loop(self) -> None: if not self._connected.is_set(): _ = self._connected.wait(timeout=self._queue_poll_s) continue - - for attempt in range(self._max_retries): - self._ensure_connected() - if not self._connected.is_set(): - continue - try: - self._submit_and_wait( - self._async_receive_turn(), timeout=self._recv_timeout_s - ) - break # Success — turn received - except ConnectionClosed as e: - code: int | None = getattr(getattr(e, "rcvd", None), "code", None) - if code == 1000: - logger.info("[realtime] Session closed normally (idle) — will reconnect on next audio") - self._connected.clear() - self._session = None - break # Don't retry on idle close - logger.warning("[realtime] Recv failed (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._connected.clear() - self._session = None - except genai_errors.APIError as e: - logger.warning("[realtime] Recv API error (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._connected.clear() - self._session = None - except Exception as e: - logger.exception("[realtime] Unexpected recv error (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._connected.clear() - self._session = None + try: + self._submit_and_wait( + self._async_receive_turn(), timeout=self._recv_timeout_s + ) + except (ConnectionClosed, genai_errors.APIError) as e: + logger.warning("Recv failed: %s — reconnecting", e) + self._reconnect() + except Exception as e: + logger.exception("Unexpected error in recv loop: %s", e) + self._reconnect() diff --git a/lelamp/service/realtime/voice_agent/openai_realtime.py b/lelamp/service/realtime/voice_agent/openai_realtime.py index 86b4869c..4722c66f 100644 --- a/lelamp/service/realtime/voice_agent/openai_realtime.py +++ b/lelamp/service/realtime/voice_agent/openai_realtime.py @@ -3,6 +3,7 @@ import base64 import logging import queue +import threading import time from typing import Any, override @@ -52,9 +53,11 @@ def __init__( ) self._connection: RealtimeConnection | None = None self._speech_stopped_at: float | None = None - self._reconnect_delay_s: float = config.reconnect_delay_s - self._max_retries: int = config.max_retries - self._last_reconnect_at: float = 0.0 + self._reconnect_delay_s: float = 2.0 + # Signals that the model is idle (no active response). Set by default, + # cleared when response.create() is called, set again on response.done. + self._response_done: threading.Event = threading.Event() + self._response_done.set() @property @override @@ -113,11 +116,11 @@ def _sync_connect(self) -> None: session_config["truncation"] = truncation_cfg self._connection.session.update(session=session_config) - logger.info("[realtime] OpenAI Realtime session open (voice=%s)", self._config.voice) + logger.info("OpenAI Realtime session open (voice=%s)", self._config.voice) def _sync_disconnect(self) -> None: if self._connection is not None: - logger.info("[realtime] Disconnecting from OpenAI Realtime API") + logger.info("Disconnecting from OpenAI Realtime API") self._connection.close() self._connection = None @@ -160,12 +163,21 @@ def _sync_send_input(self, input: InputBase) -> None: "output": input.output, } ) - self._connection.response.create() + self._safe_response_create() def _sync_commit(self) -> None: if self._connection is None: return self._connection.input_audio_buffer.commit() + self._safe_response_create() + + def _safe_response_create(self) -> None: + """Wait for any active response to finish, then create a new one.""" + if self._connection is None: + return + if not self._response_done.wait(timeout=10.0): + logger.warning("Timed out waiting for active response to finish — forcing new response") + self._response_done.clear() self._connection.response.create() def _sync_receive_turn(self) -> None: @@ -188,7 +200,7 @@ def _sync_receive_turn(self) -> None: latency_ms: float = ( time.perf_counter() - self._speech_stopped_at ) * 1000 - logger.info("[realtime] Response latency: %.0fms", latency_ms) + logger.info("Response latency: %.0fms", latency_ms) self._speech_stopped_at = None self._recv_queue.put( OutputEvent( @@ -218,12 +230,13 @@ def _sync_receive_turn(self) -> None: ) case "response.done": - logger.debug("[realtime] Response complete") + logger.debug("Response complete") + self._response_done.set() self._recv_queue.put(TurnDoneEvent()) return case "error": - logger.error("[realtime] Realtime API error: %s", event.error) + logger.error("Realtime API error: %s", event.error) raise OpenAIRealtimeError(f"Realtime API error: {event.error}") case _: @@ -231,25 +244,16 @@ def _sync_receive_turn(self) -> None: # --- Reconnect --- - def _ensure_connected(self) -> None: - """Reconnect if not connected. Throttled to at most once per reconnect_delay_s.""" - if self._connected.is_set(): - return - now: float = time.perf_counter() - if now - self._last_reconnect_at < self._reconnect_delay_s: - return - self._last_reconnect_at = now - self._reconnect() - def _reconnect(self) -> None: self._connected.clear() + self._response_done.set() # unblock any waiting commit try: - logger.info("[realtime] Reconnecting...") + logger.info("Reconnecting...") self._sync_disconnect() self._sync_connect() self._connected.set() except Exception as e: - logger.warning("[realtime] Reconnect failed: %s — will retry after delay", e) + logger.warning("Reconnect failed: %s — will retry after delay", e) time.sleep(self._reconnect_delay_s) # --- VoiceAgentBase implementation --- @@ -270,20 +274,14 @@ def _send_loop(self) -> None: except queue.Empty: continue - for attempt in range(self._max_retries): - self._ensure_connected() - if not self._connected.is_set(): - continue - try: - if isinstance(event, AudioCommitEvent): - self._sync_commit() - elif isinstance(event, InputEvent) and event.input is not None: - self._sync_send_input(event.input) - break # Success - except Exception as e: - logger.warning("[realtime] Send failed (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._connected.clear() - self._connection = None + try: + if isinstance(event, AudioCommitEvent): + self._sync_commit() + elif isinstance(event, InputEvent) and event.input is not None: + self._sync_send_input(event.input) + except Exception as e: + logger.warning("Send failed: %s — reconnecting", e) + self._reconnect() @override def _recv_loop(self) -> None: @@ -291,19 +289,11 @@ def _recv_loop(self) -> None: if not self._connected.is_set(): self._connected.wait(timeout=1) continue - - for attempt in range(self._max_retries): - self._ensure_connected() - if not self._connected.is_set(): - continue - try: - self._sync_receive_turn() - break # Success - except OpenAIRealtimeError as e: - logger.warning("[realtime] Recv failed (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._connected.clear() - self._connection = None - except Exception as e: - logger.exception("[realtime] Unexpected recv error (attempt %d/%d): %s", attempt + 1, self._max_retries, e) - self._connected.clear() - self._connection = None + try: + self._sync_receive_turn() + except OpenAIRealtimeError as e: + logger.warning("Recv failed: %s — reconnecting", e) + self._reconnect() + except Exception as e: + logger.exception("Unexpected error in recv loop: %s", e) + self._reconnect() From 11096172589ae4d79a7856891889c83a23d16d5f Mon Sep 17 00:00:00 2001 From: trnkhanh2908 Date: Fri, 12 Jun 2026 10:57:08 +0700 Subject: [PATCH 2/2] [lelamp] Add max retries to config --- lelamp/config.py | 1 + 1 file changed, 1 insertion(+) diff --git a/lelamp/config.py b/lelamp/config.py index e2753c63..ab9674cc 100644 --- a/lelamp/config.py +++ b/lelamp/config.py @@ -332,6 +332,7 @@ def _lamp_cfg_get(key: str, default: str = "") -> str: # For Gemini: "off" disables automatic activity detection; any other value enables it. # For OpenAI: maps to turn_detection type in session config. REALTIME_TURN_DETECTION: str = os.environ.get("LELAMP_REALTIME_TURN_DETECTION", "off") +REALTIME_CONNECT_MAX_RETRIES: int = int(os.environ.get("LELAMP_REALTIME_CONNECT_MAX_RETRIES", "3")) # --- Realtime: Gemini Live --- REALTIME_GEMINI_API_KEY: str = (