feat: context compaction + skills system

Compaction:
- GemmaLLM.chat() truncates ChatContext to last 30 items (~15 turns)
  before sending to LLM, preventing context window overflow on long
  conversations. Preserves system prompt and removes orphaned tool calls.

Skills:
- agent/skills_mcp.py: MCP server with skill_save, skill_recall,
  skill_list, skill_update tools backed by .md files in /skills
- skills/ dir bind-mounted into container, git-trackable
- System prompt instructs Hope to save repeatable procedures as skills
  and recall them before performing tasks she's done before
- Distinct from memory (facts) — skills are learned *procedures*
This commit is contained in:
Shane
2026-08-22 16:42:16 -04:00
parent de1a5d8400
commit ab6b5254ef
6 changed files with 250 additions and 2 deletions
+1
View File
@@ -43,6 +43,7 @@ COPY agent/agent.py /opt/voice-agent/agent.py
COPY agent/web_mcp.py /opt/voice-agent/web_mcp.py
COPY agent/weather_mcp.py /opt/voice-agent/weather_mcp.py
COPY agent/memory_mcp.py /opt/voice-agent/memory_mcp.py
COPY agent/skills_mcp.py /opt/voice-agent/skills_mcp.py
# Copy web frontend + token endpoint
COPY web/index.html /var/www/voice/
+31 -2
View File
@@ -39,6 +39,7 @@ GEMMA_API_KEY = os.environ.get("GEMMA_API_KEY", "not-needed")
WEB_MCP_ENABLED = os.environ.get("WEB_MCP_ENABLED", "true").lower() in ("1", "true", "yes")
FIRECRAWL_BASE = os.environ.get("FIRECRAWL_BASE", "http://192.168.86.2:3002")
MEMORY_DIR = os.environ.get("MEMORY_DIR", "/memory")
SKILLS_DIR = os.environ.get("SKILLS_DIR", "/skills")
SYSTEM_PROMPT = textwrap.dedent("""\
You are Hope, a warm, conversational voice assistant. You are talking TO
@@ -89,8 +90,15 @@ SYSTEM_PROMPT = textwrap.dedent("""\
projects, anything they'd expect you to know later. Pick a short topic
name for each thing you save.
- Be natural about it. Never announce "I'm saving that to memory" — just
remember it and move on. If a recall comes up empty, don't mention the
search; just answer as if you'd never heard it before.
remember it and move on. If a recall comes up empty, don't mention the
search; just answer as if you'd never heard it before.
# Skills
You have a skill library of learned procedures. Use skill_recall when you need
to perform a task you've done before — it will give you the steps. When you
successfully complete a multi-step task or learn a new procedure from the user,
use skill_save to record it so you can follow it next time. Be selective: only
save skills for repeatable tasks, not one-off facts (those go in memory).
""")
@@ -164,6 +172,13 @@ class GemmaLLM(openai.LLM):
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)
stream = super().chat(
chat_ctx=chat_ctx,
tools=tools,
@@ -229,6 +244,20 @@ def build_mcp_toolsets() -> list[mcp.MCPToolset]:
)
logger.info("Memory MCP toolset enabled (%s)", MEMORY_DIR)
skills_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "skills_mcp.py")
toolsets.append(
mcp.MCPToolset(
id="skills",
mcp_server=mcp.MCPServerStdio(
command=python_bin,
args=[skills_mcp_script],
env={**os.environ, "SKILLS_DIR": SKILLS_DIR},
client_session_timeout_seconds=30,
),
)
)
logger.info("Skills MCP toolset enabled (%s)", SKILLS_DIR)
extra = os.environ.get("EXTRA_MCP_SERVERS", "")
if extra:
try:
+182
View File
@@ -0,0 +1,182 @@
"""Skills MCP server — learned procedures for Hope.
Runs over stdio inside the voice container. The agent attaches it via
MCPServerStdio, so the LLM can recall and save skills during a
conversation. Skills live as plain markdown files in SKILLS_DIR
(default /skills), one file per skill, mounted from the host so they
survive rebuilds and are backed up via git.
Skills are *how to do things* — repeatable procedures distilled from
conversations. Facts and preferences belong in memory, not skills.
Tools:
- skill_save(name, description, steps) -> save a new skill
- skill_recall(query) -> matching skills with their full steps
- skill_list() -> all skills (name + description line)
- skill_update(name, new_content) -> replace an existing skill's content
"""
from __future__ import annotations
import os
import re
from datetime import datetime, timezone
from mcp.server.fastmcp import FastMCP
SKILLS_DIR = os.environ.get("SKILLS_DIR", "/skills")
mcp = FastMCP("skills")
def _slugify(name: str) -> str:
"""Turn a skill name into a safe filename slug."""
slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
return slug or "misc"
def _read_files() -> list[tuple[str, str]]:
"""Return (filename, content) for every .md file in SKILLS_DIR."""
files: list[tuple[str, str]] = []
if not os.path.isdir(SKILLS_DIR):
return files
for name in sorted(os.listdir(SKILLS_DIR)):
if not name.endswith(".md"):
continue
path = os.path.join(SKILLS_DIR, name)
try:
with open(path, encoding="utf-8") as f:
files.append((name, f.read()))
except OSError:
continue
return files
def _description_line(content: str) -> str:
"""First non-empty line after the H1 title."""
lines = content.splitlines()
started = False
for line in lines:
if not started:
if line.startswith("# "):
started = True
continue
if line.strip():
return line.strip()
return "(no description)"
@mcp.tool()
def skill_save(name: str, description: str, steps: str) -> str:
"""Save a learned procedure as a reusable skill.
Use this after successfully completing a multi-step task or when the
user teaches you a new procedure, so you can follow it next time. Be
selective: only save repeatable tasks, not one-off facts (those go in
memory). Returns a short confirmation.
"""
name = (name or "").strip()
description = (description or "").strip()
steps = (steps or "").strip()
if not name or not description or not steps:
return "Nothing saved — a name, a description, and steps are all needed."
os.makedirs(SKILLS_DIR, exist_ok=True)
slug = _slugify(name)
path = os.path.join(SKILLS_DIR, f"{slug}.md")
stamp = datetime.now(timezone.utc).isoformat()
with open(path, "w", encoding="utf-8") as f:
f.write(f"# {name}\n\n{description}\n\n## Steps\n{steps}\n\n---\n")
f.write(f"Created: {stamp}\nSource: conversation-learned\n")
return f"Saved skill '{name}'."
@mcp.tool()
def skill_recall(query: str) -> str:
"""Search the skill library for a procedure matching a query.
Use this when you need to perform a task you may have done before —
it returns the matching skills with their full steps so you can follow
them. Returns a message saying nothing was found if there's no match.
"""
q = (query or "").strip().lower()
if not q:
return "I don't have a skill for that."
words = [w for w in re.split(r"\W+", q) if len(w) > 2]
files = _read_files()
if not files:
return "I don't have a skill for that."
scored: list[tuple[int, str, str]] = []
for name, content in files:
lower = content.lower()
score = 0
if q in lower:
score += len(q)
for w in words:
score += lower.count(w)
if score > 0:
scored.append((score, name, content))
if not scored:
return "I don't have a skill for that."
scored.sort(key=lambda s: s[0], reverse=True)
parts = []
for _, name, content in scored[:3]:
parts.append(content.strip())
return "\n\n".join(parts)
@mcp.tool()
def skill_list() -> str:
"""List all saved skills with a brief description of each.
Returns the skill name and its description line, or a message saying
no skills exist yet.
"""
files = _read_files()
if not files:
return "No skills saved yet."
lines = [f"{name.removesuffix('.md')}: {_description_line(content)}" for name, content in files]
return "\n".join(lines)
@mcp.tool()
def skill_update(name: str, new_content: str) -> str:
"""Replace the content of an existing skill.
Use this to refine a learned procedure after you've improved it. The
name must match an existing skill (case-insensitive). Returns a
confirmation, or an error if no such skill exists.
"""
name = (name or "").strip()
new_content = (new_content or "").strip()
if not name or not new_content:
return "Nothing updated — both a name and new content are needed."
slug = _slugify(name)
path = os.path.join(SKILLS_DIR, f"{slug}.md")
if not os.path.exists(path):
# Fall back to matching by the H1 title in case the filename differs.
for fname, content in _read_files():
first = next((l for l in content.splitlines() if l.strip()), "")
if first.lstrip("# ").strip().lower() == name.lower():
path = os.path.join(SKILLS_DIR, fname)
break
else:
return f"No skill named '{name}' found."
stamp = datetime.now(timezone.utc).isoformat()
with open(path, "w", encoding="utf-8") as f:
f.write(new_content.rstrip() + "\n\n---\n")
f.write(f"Updated: {stamp}\nSource: conversation-learned\n")
return f"Updated skill '{name}'."
if __name__ == "__main__":
mcp.run(transport="stdio")
+1
View File
@@ -19,6 +19,7 @@ services:
- ./livekit.yaml:/etc/livekit.yaml:ro
- ./certs:/etc/voice/certs
- ./memory:/memory
- ./skills:/skills
healthcheck:
test: ["CMD-SHELL", "curl -sk https://localhost:8090/ -o /dev/null && curl -s http://localhost:7880/ -o /dev/null"]
interval: 15s
View File
+35
View File
@@ -0,0 +1,35 @@
# Skills
Skills are learned procedures that Hope distills from conversations.
One `.md` file per skill, saved here by the skills MCP server
(`agent/skills_mcp.py`) and mounted into the container at `/skills`.
## How they differ from memory
- **Memory** (`./memory/`) stores *facts*: names, preferences, things
Hope should remember.
- **Skills** (this directory) store *how to do things*: repeatable
procedures, workflows, and multi-step tasks Hope has learned.
## File format
```markdown
# Make a Coffee Order
How to place an order at the local coffee shop on behalf of the user.
## Steps
1. Confirm the usual drink (oat latte, extra hot).
2. Ask if they want anything new today.
3. Read back the order and confirm.
4. Tell them the pickup time.
---
Created: 2026-08-22T12:00:00+00:00
Source: conversation-learned
```
Hope saves a skill when it successfully completes a multi-step task or
the user teaches it a new procedure, and recalls one with `skill_recall`
before performing a task it has done before. Skills are updated in place
when the procedure improves.