diff --git a/agent/agent.py b/agent/agent.py index 6b21b86..d4c0bdf 100644 --- a/agent/agent.py +++ b/agent/agent.py @@ -167,17 +167,42 @@ class _ReasoningFallbackWrapper: class GemmaLLM(openai.LLM): """OpenAI-compatible LLM (llama.cpp Gemma 4) with thinking disabled.""" + # Context compaction: when the conversation grows past this many items, + # ask Gemma to summarize the older messages into a compact recap, then + # replace them with that summary. Keeps recent turns verbatim. + COMPACTION_THRESHOLD = 24 # ~12 turns of user/agent exchange + KEEP_RECENT = 10 # last N items kept verbatim after compaction + IDLE_COMPACTION_SECONDS = 300 # also compact after 5 min of silence + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + import time + self._last_call_time: float = time.monotonic() + def chat(self, *, chat_ctx, tools=None, conn_options=None, **kwargs): if conn_options is None: from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS conn_options = DEFAULT_API_CONNECT_OPTIONS - # Compact the context before sending: keep the system prompt and the - # last N items so long conversations stay within Gemma's window. - MAX_CONTEXT_ITEMS = 30 - if len(chat_ctx) > MAX_CONTEXT_ITEMS: - chat_ctx.truncate(max_items=MAX_CONTEXT_ITEMS) + import time + now = time.monotonic() + idle_for = now - self._last_call_time + self._last_call_time = now + + # Compact if context is large, OR if we've been idle for 5+ minutes + # (user came back after a break — summarize the old conversation). + should_compact = len(chat_ctx) > self.COMPACTION_THRESHOLD + if not should_compact and idle_for > self.IDLE_COMPACTION_SECONDS: + should_compact = len(chat_ctx) > 6 # minimal context worth summarizing + + if should_compact: + reason = "idle" if idle_for > self.IDLE_COMPACTION_SECONDS else "size" + logger.info( + "Compacting context (%s): %d items, idle %.0fs", + reason, len(chat_ctx), idle_for, + ) + chat_ctx = self._compact_context(chat_ctx, conn_options) stream = super().chat( chat_ctx=chat_ctx, @@ -187,6 +212,118 @@ class GemmaLLM(openai.LLM): ) return _ReasoningFallbackWrapper(stream) + def _compact_context(self, chat_ctx, conn_options): + """Summarize older messages via a side LLM call, keep recent ones. + + Splits the context into [system | old | recent]. The old portion is + summarized by Gemma into a short recap message that replaces it. + Falls back to hard truncation if the summary call fails. + """ + from livekit.agents import llm as lk_llm + + items = list(chat_ctx) + # Identify the system/instruction message (first item usually) + sys_items = [it for it in items if getattr(it, "role", None) in ("system", "developer")] + non_sys = [it for it in items if getattr(it, "role", None) not in ("system", "developer")] + + if len(non_sys) <= self.KEEP_RECENT: + return chat_ctx # nothing to compact + + old_msgs = non_sys[: -self.KEEP_RECENT] + recent_msgs = non_sys[-self.KEEP_RECENT :] + + # Build a transcript of the old messages for the summarizer + transcript_parts = [] + for msg in old_msgs: + role = getattr(msg, "role", "user") + content = self._msg_content_text(msg) + if content: + transcript_parts.append(f"{role}: {content}") + transcript = "\n".join(transcript_parts) + + if not transcript.strip(): + return chat_ctx # nothing meaningful to summarize + + # Side call to Gemma for the summary (no tools, no streaming needed) + try: + import httpx + + summary_prompt = ( + "Summarize this conversation history in 2-4 sentences. " + "Preserve key facts, decisions, and user preferences. " + "Be concise — this replaces the full history as context.\n\n" + + transcript + ) + resp = httpx.post( + f"{GEMMA_BASE_URL}/chat/completions", + headers={"Content-Type": "application/json"}, + json={ + "model": GEMMA_MODEL, + "messages": [ + {"role": "system", "content": "You summarize conversations concisely."}, + {"role": "user", "content": summary_prompt}, + ], + "max_tokens": 200, + "temperature": 0.3, + "chat_template_kwargs": {"enable_thinking": False}, + }, + timeout=30, + ) + resp.raise_for_status() + data = resp.json() + summary_text = (data.get("choices") or [{}])[0].get("message", {}).get("content", "") + except Exception as e: # noqa: BLE001 + logger.warning("Context compaction failed, falling back to truncation: %s", e) + chat_ctx.truncate(max_items=self.COMPACTION_THRESHOLD) + return chat_ctx + + if not summary_text.strip(): + chat_ctx.truncate(max_items=self.COMPACTION_THRESHOLD) + return chat_ctx + + # Rebuild the context: system + summary + recent + new_items = list(sys_items) + new_items.append( + lk_llm.ChatMessage( + role="user", + content=[f"[Conversation so far: {summary_text.strip()}]"], + ) + ) + new_items.append( + lk_llm.ChatMessage( + role="assistant", + content=["Got it, I'll keep that in mind."], + ) + ) + new_items.extend(recent_msgs) + + logger.info( + "Context compacted: %d items -> %d (summary: %.80s...)", + len(items), + len(new_items), + summary_text.strip(), + ) + + return lk_llm.ChatContext(new_items) + + @staticmethod + def _msg_content_text(msg) -> str: + """Extract plain text from a ChatMessage's content field.""" + content = getattr(msg, "content", None) + if content is None: + return "" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for c in content: + if isinstance(c, str): + parts.append(c) + elif hasattr(c, "text"): + parts.append(c.text) + return " ".join(parts) + return str(content) + # ── MCP toolsets (web access + any extra configured servers) ────────────────