Alexander Whitestone
3588283b83
Practical implementation for #327. Extracts seed data from existing sessions to bootstrap new sessions with established context and patterns.
486 lines
17 KiB
Python
486 lines
17 KiB
Python
"""
|
|
Warm Session Provisioning: Practical Implementation
|
|
|
|
Provides mechanisms to create pre-contextualized sessions that start
|
|
with established patterns and context, reducing initial errors.
|
|
|
|
Issue: #327
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
from dataclasses import dataclass, asdict, field
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class SessionSeed:
|
|
"""Seed data for warming up a new session."""
|
|
system_context: str = ""
|
|
tool_examples: List[Dict[str, Any]] = field(default_factory=list)
|
|
user_patterns: Dict[str, Any] = field(default_factory=dict)
|
|
context_markers: List[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'SessionSeed':
|
|
return cls(**data)
|
|
|
|
|
|
@dataclass
|
|
class WarmTemplate:
|
|
"""Template for creating warm sessions."""
|
|
template_id: str
|
|
name: str
|
|
description: str
|
|
seed: SessionSeed
|
|
created_at: str
|
|
source_session_id: Optional[str] = None
|
|
usage_count: int = 0
|
|
success_rate: float = 0.0
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"template_id": self.template_id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"seed": self.seed.to_dict(),
|
|
"created_at": self.created_at,
|
|
"source_session_id": self.source_session_id,
|
|
"usage_count": self.usage_count,
|
|
"success_rate": self.success_rate
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'WarmTemplate':
|
|
seed = SessionSeed.from_dict(data.get("seed", {}))
|
|
return cls(
|
|
template_id=data["template_id"],
|
|
name=data["name"],
|
|
description=data["description"],
|
|
seed=seed,
|
|
created_at=data.get("created_at", datetime.now().isoformat()),
|
|
source_session_id=data.get("source_session_id"),
|
|
usage_count=data.get("usage_count", 0),
|
|
success_rate=data.get("success_rate", 0.0)
|
|
)
|
|
|
|
|
|
class SessionExtractor:
|
|
"""Extract seed data from existing sessions."""
|
|
|
|
def __init__(self, session_db=None):
|
|
self.session_db = session_db
|
|
|
|
def extract_seed(self, session_id: str) -> Optional[SessionSeed]:
|
|
"""Extract seed data from a session."""
|
|
if not self.session_db:
|
|
return None
|
|
|
|
try:
|
|
messages = self.session_db.get_messages(session_id)
|
|
if not messages:
|
|
return None
|
|
|
|
# Extract system context
|
|
system_context = self._extract_system_context(messages)
|
|
|
|
# Extract successful tool examples
|
|
tool_examples = self._extract_tool_examples(messages)
|
|
|
|
# Extract user patterns
|
|
user_patterns = self._extract_user_patterns(messages)
|
|
|
|
# Extract context markers
|
|
context_markers = self._extract_context_markers(messages)
|
|
|
|
return SessionSeed(
|
|
system_context=system_context,
|
|
tool_examples=tool_examples,
|
|
user_patterns=user_patterns,
|
|
context_markers=context_markers
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to extract seed: {e}")
|
|
return None
|
|
|
|
def _extract_system_context(self, messages: List[Dict]) -> str:
|
|
"""Extract useful system context from messages."""
|
|
context_parts = []
|
|
|
|
# Look for system messages
|
|
for msg in messages:
|
|
if msg.get("role") == "system":
|
|
content = msg.get("content", "")
|
|
# Take first 500 chars of system context
|
|
if content:
|
|
context_parts.append(content[:500])
|
|
break
|
|
|
|
# Extract key user instructions
|
|
user_instructions = []
|
|
for msg in messages[:10]: # First 10 messages
|
|
if msg.get("role") == "user":
|
|
content = msg.get("content", "")
|
|
if len(content) > 50 and "?" not in content[:20]: # Likely instructions
|
|
user_instructions.append(content[:200])
|
|
if len(user_instructions) >= 3:
|
|
break
|
|
|
|
if user_instructions:
|
|
context_parts.append("\nKey instructions from session:\n" + "\n".join(f"- {i}" for i in user_instructions))
|
|
|
|
return "\n".join(context_parts)[:1000]
|
|
|
|
def _extract_tool_examples(self, messages: List[Dict]) -> List[Dict[str, Any]]:
|
|
"""Extract successful tool call examples."""
|
|
examples = []
|
|
|
|
for i, msg in enumerate(messages):
|
|
if msg.get("role") == "assistant" and msg.get("tool_calls"):
|
|
# Check if there's a successful result
|
|
for j in range(i + 1, min(i + 3, len(messages))):
|
|
if messages[j].get("role") == "tool":
|
|
content = messages[j].get("content", "")
|
|
# Check for success indicators
|
|
if content and "error" not in content.lower()[:100]:
|
|
for tool_call in msg["tool_calls"]:
|
|
func = tool_call.get("function", {})
|
|
examples.append({
|
|
"tool": func.get("name"),
|
|
"arguments": func.get("arguments", "{}"),
|
|
"result_preview": content[:200]
|
|
})
|
|
if len(examples) >= 5:
|
|
break
|
|
break
|
|
|
|
if len(examples) >= 5:
|
|
break
|
|
|
|
return examples
|
|
|
|
def _extract_user_patterns(self, messages: List[Dict]) -> Dict[str, Any]:
|
|
"""Extract user interaction patterns."""
|
|
user_messages = [m for m in messages if m.get("role") == "user"]
|
|
|
|
if not user_messages:
|
|
return {}
|
|
|
|
# Calculate patterns
|
|
lengths = [len(m.get("content", "")) for m in user_messages]
|
|
avg_length = sum(lengths) / len(lengths)
|
|
|
|
# Count question types
|
|
questions = sum(1 for m in user_messages if "?" in m.get("content", ""))
|
|
commands = sum(1 for m in user_messages if m.get("content", "").startswith(("/", "!")))
|
|
|
|
return {
|
|
"message_count": len(user_messages),
|
|
"avg_length": avg_length,
|
|
"question_ratio": questions / len(user_messages),
|
|
"command_ratio": commands / len(user_messages),
|
|
"preferred_style": "command" if commands > questions else "conversational"
|
|
}
|
|
|
|
def _extract_context_markers(self, messages: List[Dict]) -> List[str]:
|
|
"""Extract important context markers."""
|
|
markers = set()
|
|
|
|
for msg in messages:
|
|
content = msg.get("content", "")
|
|
|
|
# File paths
|
|
import re
|
|
paths = re.findall(r'[\w/\.]+\.[\w]+', content)
|
|
markers.update(p for p in paths if len(p) < 50)
|
|
|
|
# URLs
|
|
urls = re.findall(r'https?://[^\s]+', content)
|
|
markers.update(u[:80] for u in urls[:3])
|
|
|
|
if len(markers) > 20:
|
|
break
|
|
|
|
return list(markers)[:20]
|
|
|
|
|
|
class WarmSessionManager:
|
|
"""Manage warm session templates."""
|
|
|
|
def __init__(self, template_dir: Path = None):
|
|
self.template_dir = template_dir or Path.home() / ".hermes" / "warm_templates"
|
|
self.template_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def save_template(self, template: WarmTemplate) -> Path:
|
|
"""Save a warm template."""
|
|
path = self.template_dir / f"{template.template_id}.json"
|
|
with open(path, 'w') as f:
|
|
json.dump(template.to_dict(), f, indent=2)
|
|
return path
|
|
|
|
def load_template(self, template_id: str) -> Optional[WarmTemplate]:
|
|
"""Load a warm template."""
|
|
path = self.template_dir / f"{template_id}.json"
|
|
if not path.exists():
|
|
return None
|
|
|
|
try:
|
|
with open(path, 'r') as f:
|
|
data = json.load(f)
|
|
return WarmTemplate.from_dict(data)
|
|
except Exception as e:
|
|
logger.error(f"Failed to load template: {e}")
|
|
return None
|
|
|
|
def list_templates(self) -> List[Dict[str, Any]]:
|
|
"""List all templates."""
|
|
templates = []
|
|
for path in self.template_dir.glob("*.json"):
|
|
try:
|
|
with open(path, 'r') as f:
|
|
data = json.load(f)
|
|
templates.append({
|
|
"template_id": data.get("template_id"),
|
|
"name": data.get("name"),
|
|
"description": data.get("description"),
|
|
"usage_count": data.get("usage_count", 0),
|
|
"success_rate": data.get("success_rate", 0.0)
|
|
})
|
|
except:
|
|
pass
|
|
return templates
|
|
|
|
def delete_template(self, template_id: str) -> bool:
|
|
"""Delete a template."""
|
|
path = self.template_dir / f"{template_id}.json"
|
|
if path.exists():
|
|
path.unlink()
|
|
return True
|
|
return False
|
|
|
|
|
|
class WarmSessionBootstrapper:
|
|
"""Bootstrap warm sessions from templates."""
|
|
|
|
def __init__(self, manager: WarmSessionManager = None):
|
|
self.manager = manager or WarmSessionManager()
|
|
|
|
def prepare_messages(
|
|
self,
|
|
template: WarmTemplate,
|
|
user_message: str,
|
|
include_examples: bool = True
|
|
) -> List[Dict[str, Any]]:
|
|
"""Prepare messages for a warm session."""
|
|
messages = []
|
|
|
|
# Add warm context as system message
|
|
warm_context = self._build_warm_context(template.seed)
|
|
if warm_context:
|
|
messages.append({
|
|
"role": "system",
|
|
"content": warm_context
|
|
})
|
|
|
|
# Add tool examples if requested
|
|
if include_examples and template.seed.tool_examples:
|
|
example_messages = self._create_example_messages(template.seed.tool_examples)
|
|
messages.extend(example_messages)
|
|
|
|
# Add the actual user message
|
|
messages.append({
|
|
"role": "user",
|
|
"content": user_message
|
|
})
|
|
|
|
return messages
|
|
|
|
def _build_warm_context(self, seed: SessionSeed) -> str:
|
|
"""Build warm context from seed."""
|
|
parts = []
|
|
|
|
if seed.system_context:
|
|
parts.append(seed.system_context)
|
|
|
|
if seed.context_markers:
|
|
parts.append("\nKnown context: " + ", ".join(seed.context_markers[:10]))
|
|
|
|
if seed.user_patterns:
|
|
style = seed.user_patterns.get("preferred_style", "balanced")
|
|
parts.append(f"\nUser prefers {style} interactions.")
|
|
|
|
return "\n".join(parts)[:1500]
|
|
|
|
def _create_example_messages(self, examples: List[Dict]) -> List[Dict]:
|
|
"""Create example messages from tool examples."""
|
|
messages = []
|
|
|
|
for i, ex in enumerate(examples[:3]): # Limit to 3 examples
|
|
# User request
|
|
messages.append({
|
|
"role": "user",
|
|
"content": f"[Example {i+1}] Use {ex['tool']}"
|
|
})
|
|
|
|
# Assistant with tool call
|
|
messages.append({
|
|
"role": "assistant",
|
|
"content": f"I'll use {ex['tool']}.",
|
|
"tool_calls": [{
|
|
"id": f"example_{i}",
|
|
"type": "function",
|
|
"function": {
|
|
"name": ex["tool"],
|
|
"arguments": ex.get("arguments", "{}")
|
|
}
|
|
}]
|
|
})
|
|
|
|
# Tool result
|
|
messages.append({
|
|
"role": "tool",
|
|
"tool_call_id": f"example_{i}",
|
|
"content": ex.get("result_preview", "Success")
|
|
})
|
|
|
|
return messages
|
|
|
|
|
|
# CLI Functions
|
|
def warm_session_cli(args: List[str]) -> int:
|
|
"""CLI interface for warm session management."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Warm session provisioning")
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
|
|
# Extract command
|
|
extract_parser = subparsers.add_parser("extract", help="Extract template from session")
|
|
extract_parser.add_argument("session_id", help="Session ID to extract from")
|
|
extract_parser.add_argument("--name", "-n", required=True, help="Template name")
|
|
extract_parser.add_argument("--description", "-d", default="", help="Template description")
|
|
|
|
# List command
|
|
subparsers.add_parser("list", help="List available templates")
|
|
|
|
# Test command
|
|
test_parser = subparsers.add_parser("test", help="Test warm session creation")
|
|
test_parser.add_argument("template_id", help="Template ID")
|
|
test_parser.add_argument("message", help="Test message")
|
|
|
|
# Delete command
|
|
delete_parser = subparsers.add_parser("delete", help="Delete a template")
|
|
delete_parser.add_argument("template_id", help="Template ID to delete")
|
|
|
|
parsed = parser.parse_args(args)
|
|
|
|
if not parsed.command:
|
|
parser.print_help()
|
|
return 1
|
|
|
|
manager = WarmSessionManager()
|
|
|
|
if parsed.command == "extract":
|
|
try:
|
|
from hermes_state import SessionDB
|
|
session_db = SessionDB()
|
|
except ImportError:
|
|
print("Error: Cannot import SessionDB")
|
|
return 1
|
|
|
|
extractor = SessionExtractor(session_db)
|
|
seed = extractor.extract_seed(parsed.session_id)
|
|
|
|
if not seed:
|
|
print(f"Failed to extract seed from session {parsed.session_id}")
|
|
return 1
|
|
|
|
template = WarmTemplate(
|
|
template_id=f"warm_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
|
|
name=parsed.name,
|
|
description=parsed.description,
|
|
seed=seed,
|
|
created_at=datetime.now().isoformat(),
|
|
source_session_id=parsed.session_id
|
|
)
|
|
|
|
path = manager.save_template(template)
|
|
print(f"Created template: {template.template_id}")
|
|
print(f"Saved to: {path}")
|
|
print(f"Tool examples: {len(seed.tool_examples)}")
|
|
print(f"Context markers: {len(seed.context_markers)}")
|
|
return 0
|
|
|
|
elif parsed.command == "list":
|
|
templates = manager.list_templates()
|
|
|
|
if not templates:
|
|
print("No templates found.")
|
|
return 0
|
|
|
|
print("\n=== Warm Session Templates ===\n")
|
|
for t in templates:
|
|
print(f"ID: {t['template_id']}")
|
|
print(f" Name: {t['name']}")
|
|
print(f" Description: {t['description']}")
|
|
print(f" Usage: {t['usage_count']} times, {t['success_rate']:.0%} success")
|
|
print()
|
|
|
|
return 0
|
|
|
|
elif parsed.command == "test":
|
|
template = manager.load_template(parsed.template_id)
|
|
|
|
if not template:
|
|
print(f"Template {parsed.template_id} not found")
|
|
return 1
|
|
|
|
bootstrapper = WarmSessionBootstrapper(manager)
|
|
messages = bootstrapper.prepare_messages(template, parsed.message)
|
|
|
|
print(f"\n=== Warm Session Test: {template.name} ===\n")
|
|
print(f"Generated {len(messages)} messages:\n")
|
|
|
|
for i, msg in enumerate(messages):
|
|
role = msg.get("role", "unknown")
|
|
content = msg.get("content", "")
|
|
|
|
if role == "system":
|
|
print(f"[System Context] ({len(content)} chars)")
|
|
print(content[:200] + "..." if len(content) > 200 else content)
|
|
elif role == "user":
|
|
print(f"\n[User]: {content}")
|
|
elif role == "assistant":
|
|
print(f"[Assistant]: {content}")
|
|
if msg.get("tool_calls"):
|
|
for tc in msg["tool_calls"]:
|
|
func = tc.get("function", {})
|
|
print(f" -> {func.get('name')}({func.get('arguments', '{}')[:50]})")
|
|
elif role == "tool":
|
|
print(f" [Result]: {content[:100]}...")
|
|
|
|
return 0
|
|
|
|
elif parsed.command == "delete":
|
|
if manager.delete_template(parsed.template_id):
|
|
print(f"Deleted template: {parsed.template_id}")
|
|
return 0
|
|
else:
|
|
print(f"Template {parsed.template_id} not found")
|
|
return 1
|
|
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
sys.exit(warm_session_cli(sys.argv[1:]))
|