Compare commits

...

5 Commits

Author SHA1 Message Date
b6063a5037 docs(research): Add warm session provisioning README
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 58s
Part of #327. Documentation for warm session provisioning research.
2026-04-14 00:52:41 +00:00
4e48baf9f9 test(research): Add warm session provisioning test script
Part of #327. Test script for session analysis, template creation, and warm session bootstrapping.
2026-04-14 00:52:05 +00:00
7cce4ada0f feat(cli): Add warm session provisioning commands
Part of #327. Adds `hermes warm` command for session analysis and template management.
2026-04-14 00:51:17 +00:00
5298c4fda8 feat(research): Add warm session provisioning module
Part of #327. Implements session profiling, template extraction, and warm session bootstrapping.
2026-04-14 00:49:42 +00:00
800cd0c6c9 docs(research): Add warm session provisioning research document
Part of #327. Research framework for pre-proficient agent sessions.
2026-04-14 00:48:27 +00:00
5 changed files with 1256 additions and 0 deletions

View File

@@ -0,0 +1,143 @@
# Warm Session Provisioning
**Research Implementation for Issue #327**
## Overview
This module implements warm session provisioning - creating pre-proficient agent sessions that start with marathon-level reliability by pre-seeding with successful patterns.
## Key Findings
- Marathon sessions (100+ messages) have **5.7%** error rate vs **9.0%** for mid-length sessions
- Agents improve with experience within a session
- Patterns can be extracted and transferred to new sessions
## Components
### 1. Session Profiler (`SessionProfiler`)
Analyzes existing sessions to extract proficiency patterns:
- Tool call success rates
- User interaction patterns
- Error recovery strategies
- Context markers
### 2. Template Manager (`TemplateManager`)
Manages session templates:
- Save/load templates
- Template versioning
- Template validation
### 3. Warm Session Bootstrapper (`WarmSessionBootstrapper`)
Creates warm sessions from templates:
- Injects proficiency patterns
- Creates synthetic tool call examples
- Establishes context from templates
## Usage
### CLI Commands
```bash
# Analyze a session for proficiency patterns
hermes warm analyze SESSION_ID
# Create a template from a successful session
hermes warm create-template SESSION_ID --name "My Warm Template"
# List available templates
hermes warm list-templates
# Test warm session creation
hermes warm test TEMPLATE_ID "Test message"
```
### Python API
```python
from tools.warm_session import SessionProfiler, TemplateManager, WarmSessionBootstrapper
from hermes_state import SessionDB
# Analyze a session
session_db = SessionDB()
profiler = SessionProfiler(session_db)
analysis = profiler.analyze_session("session_id")
# Create a template
template = create_template_from_session("session_id", session_db)
# Create a warm session
manager = TemplateManager()
bootstrapper = WarmSessionBootstrapper(manager)
messages = bootstrapper.create_warm_session_messages(template, "User message")
```
## Research Questions
1. **What makes marathon sessions more reliable?**
- Context richness
- User guidance patterns
- Agent adaptation
2. **Can we pre-seed sessions with successful tool-call examples?**
- Yes, via synthetic tool calls in template
3. **Does compression preserve proficiency?**
- To be tested with different compression strategies
4. **Minimum viable warm-up sequence?**
- Research in progress
## A/B Testing Framework
Compare warm vs. cold sessions:
- Same tasks, different starting conditions
- Measure: error rate, completion time, user satisfaction
- Statistical significance testing
## Expected Outcomes
### Short-term
- Session profiling system operational
- Initial pattern extraction
- Basic template storage
### Medium-term
- Warm session prototype
- Initial A/B test results
- Paper outline
### Long-term
- Production-ready warm session provisioning
- Published research paper
- Open-source template sharing
## Files
- `tools/warm_session.py` - Main implementation
- `docs/research/warm-session-provisioning.md` - Research document
- `tests/test_warm_session.py` - Test suite
- `README.md` - This file
## Next Steps
1. ✅ Create research framework
2. ✅ Implement session profiling
3. ⏳ Analyze 5-10 marathon sessions
4. ⏳ Extract proficiency patterns
5. ⏳ Build template storage system
6. ⏳ Implement warm session bootstrapping
7. ⏳ Run A/B tests
8. ⏳ Write paper
## Contributing
This is research code. Contributions welcome:
- Additional proficiency metrics
- Better pattern extraction
- A/B testing improvements
- Documentation
## References
- Empirical Audit 2026-04-12, Finding 4
- Issue #327

View File

@@ -0,0 +1,148 @@
# Warm Session Provisioning: Pre-Proficient Agent Sessions
**Research Document**
**Issue:** #327
**Date:** April 2026
**Status:** Research & Prototype
## Executive Summary
Empirical analysis reveals a counterintuitive finding: marathon sessions (100+ messages) exhibit **lower** per-tool error rates (5.7%) than mid-length sessions (9.0% at 51-100 messages). This suggests agents improve with experience within a session, learning user patterns and establishing successful tool-call conventions.
This research explores whether we can pre-seed sessions with proficiency patterns, effectively creating "warm" sessions that start at marathon-level reliability.
## Key Findings from Empirical Audit
### 1. Session Length vs. Error Rate
- **0-50 messages:** 7.2% error rate
- **51-100 messages:** 9.0% error rate (peak)
- **100+ messages:** 5.7% error rate (lowest)
### 2. Hypothesis: Context Richness Drives Proficiency
Marathon sessions develop:
- **User-specific patterns:** How the user phrases requests
- **Tool-call conventions:** Successful argument formats
- **Error recovery patterns:** How to handle failures
- **Context anchoring:** Shared reference points
### 3. Research Questions
1. What specific context elements drive proficiency?
2. Can we extract and transfer these patterns?
3. Does compression preserve proficiency or reset it?
4. What's the minimum viable warm-up sequence?
## Technical Approach
### Phase 1: Analysis (Current)
Analyze existing marathon sessions to identify proficiency markers:
- Tool-call success patterns
- User interaction conventions
- Error recovery sequences
- Context window utilization
### Phase 2: Template Extraction
Extract successful patterns from marathon sessions:
```python
# Conceptual template structure
session_template = {
"tool_patterns": [
{"tool": "terminal", "success_pattern": "..."},
{"tool": "file_operations", "conventions": "..."}
],
"user_patterns": {
"request_style": "direct",
"feedback_style": "terse"
},
"recovery_patterns": [
{"error": "FileNotFound", "recovery": "..."}
]
}
```
### Phase 3: Warm Session Creation
Implement session seeding mechanism:
1. Run 10-20 diverse tasks to build context
2. Compress the session (preserving proficiency markers)
3. Save as warm template
4. New sessions start from template
### Phase 4: A/B Testing
Compare warm vs. cold sessions:
- Same tasks, different starting conditions
- Measure: error rate, time to completion, user satisfaction
- Statistical significance testing
## Implementation Plan
### 1. Session Profiling System
- Add telemetry to track proficiency markers
- Identify what makes marathon sessions successful
- Extract transferable patterns
### 2. Template Management
- Save/load session templates
- Version control for templates
- Template validation
### 3. Warm Session Bootstrapping
- Inject template context at session start
- Preserve tool-call conventions
- Maintain user pattern awareness
### 4. Research Infrastructure
- A/B testing framework
- Statistical analysis tools
- Visualization of results
## Expected Outcomes
### Short-term (Weeks 1-2)
- Session profiling system operational
- Initial pattern extraction from 5-10 marathon sessions
- Basic template storage
### Medium-term (Weeks 3-4)
- Warm session prototype
- Initial A/B test results
- Paper outline
### Long-term (Month 2+)
- Production-ready warm session provisioning
- Published research paper
- Open-source template sharing
## Paper-Worthy Contributions
1. **Empirical finding:** Session proficiency increases with length
2. **Novel approach:** Pre-seeding sessions with proficiency patterns
3. **Production system:** Warm session provisioning infrastructure
4. **Open dataset:** Session proficiency markers and templates
## Risks and Mitigations
### Technical Risks
- **Compression resets proficiency:** Test different compression strategies
- **Patterns don't transfer:** Start with similar task domains
- **Template bloat:** Keep templates minimal, focus on high-impact patterns
### Research Risks
- **Statistical insignificance:** Run sufficient A/B test iterations
- **Confounding variables:** Control for task difficulty, user expertise
- **Publication bias:** Report all results, including negative findings
## Next Steps
1. ✅ Create research framework (this document)
2. ⏳ Implement session profiling telemetry
3. ⏳ Analyze 5-10 marathon sessions
4. ⏳ Extract proficiency patterns
5. ⏳ Build template storage system
6. ⏳ Implement warm session bootstrapping
7. ⏳ Run A/B tests
8. ⏳ Write paper
## References
- Empirical Audit 2026-04-12, Finding 4
- Session compression research (trajectory_compressor.py)
- Tool-call error analysis (existing telemetry)

View File

@@ -5258,6 +5258,35 @@ For more help on a command:
sessions_parser.set_defaults(func=cmd_sessions)
# Warm session command (research #327)
warm_parser = subparsers.add_parser(
"warm",
help="Warm session provisioning (research)",
description="Pre-proficient agent sessions with extracted patterns"
)
warm_subparsers = warm_parser.add_subparsers(dest="warm_command")
# Warm analyze command
warm_analyze = warm_subparsers.add_parser("analyze", help="Analyze a session for proficiency patterns")
warm_analyze.add_argument("session_id", help="Session ID to analyze")
# Warm create-template command
warm_create = warm_subparsers.add_parser("create-template", help="Create warm template from session")
warm_create.add_argument("session_id", help="Session ID to create template from")
warm_create.add_argument("--name", "-n", help="Template name")
# Warm list-templates command
warm_subparsers.add_parser("list-templates", help="List available warm templates")
# Warm test command
warm_test = warm_subparsers.add_parser("test", help="Test warm session creation")
warm_test.add_argument("template_id", help="Template ID to test")
warm_test.add_argument("message", help="Test message")
warm_parser.set_defaults(func=cmd_warm)
# =========================================================================
# insights command
# =========================================================================
@@ -5598,3 +5627,47 @@ Examples:
if __name__ == "__main__":
main()
def cmd_warm(args):
"""Handle warm session provisioning commands."""
from hermes_cli.colors import Colors, color
subcmd = getattr(args, 'warm_command', None)
if subcmd is None:
print(color("Warm Session Provisioning (Research #327)", Colors.CYAN))
print("\nCommands:")
print(" hermes warm analyze SESSION_ID - Analyze session proficiency")
print(" hermes warm create-template SESSION_ID - Create warm template")
print(" hermes warm list-templates - List available templates")
print(" hermes warm test TEMPLATE_ID MESSAGE - Test warm session")
return 0
# Import warm session module
try:
from tools.warm_session import warm_session_command
# Convert args to list for the module
args_list = []
if subcmd == "analyze":
args_list = ["analyze", args.session_id]
elif subcmd == "create-template":
args_list = ["create-template", args.session_id]
if hasattr(args, 'name') and args.name:
args_list.extend(["--name", args.name])
elif subcmd == "list-templates":
args_list = ["list-templates"]
elif subcmd == "test":
args_list = ["test", args.template_id, args.message]
return warm_session_command(args_list)
except ImportError as e:
print(color(f"Error: Cannot import warm_session module: {e}", Colors.RED))
print("Make sure tools/warm_session.py exists")
return 1
except Exception as e:
print(color(f"Error: {e}", Colors.RED))
return 1

218
tests/test_warm_session.py Normal file
View File

@@ -0,0 +1,218 @@
#!/usr/bin/env python3
"""
Test script for warm session provisioning.
This script demonstrates the warm session provisioning system
by analyzing sessions, creating templates, and testing warm sessions.
Issue: #327
"""
import sys
import os
from pathlib import Path
# Add the tools directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
def test_session_analysis():
"""Test session analysis functionality."""
print("=== Testing Session Analysis ===\n")
try:
from tools.warm_session import SessionProfiler
from hermes_state import SessionDB
session_db = SessionDB()
profiler = SessionProfiler(session_db)
# Get a session to analyze
sessions = session_db.get_messages.__self__.execute_write(
"SELECT id FROM sessions ORDER BY started_at DESC LIMIT 1"
)
if not sessions:
print("No sessions found in database.")
return False
session_id = sessions[0][0]
print(f"Analyzing session: {session_id}\n")
analysis = profiler.analyze_session(session_id)
if "error" in analysis:
print(f"Analysis error: {analysis['error']}")
return False
print(f"Message count: {analysis['message_count']}")
print(f"Proficiency score: {analysis['proficiency_score']:.2f}")
print("\nTool call analysis:")
for tool, stats in analysis.get("tool_calls", {}).items():
print(f" {tool}: {stats.get('success_rate', 0):.0%} success")
return True
except Exception as e:
print(f"Test failed: {e}")
return False
def test_template_creation():
"""Test template creation from session."""
print("\n=== Testing Template Creation ===\n")
try:
from tools.warm_session import create_template_from_session, TemplateManager
from hermes_state import SessionDB
session_db = SessionDB()
# Get a session
sessions = session_db.get_messages.__self__.execute_write(
"SELECT id FROM sessions ORDER BY started_at DESC LIMIT 1"
)
if not sessions:
print("No sessions found.")
return False
session_id = sessions[0][0]
print(f"Creating template from session: {session_id}\n")
template = create_template_from_session(
session_id,
session_db,
name="Test Template"
)
if not template:
print("Failed to create template.")
return False
print(f"Template ID: {template.template_id}")
print(f"Name: {template.name}")
print(f"Tool patterns: {len(template.tool_patterns)}")
# Save the template
manager = TemplateManager()
path = manager.save_template(template)
print(f"Saved to: {path}")
return True
except Exception as e:
print(f"Test failed: {e}")
return False
def test_warm_session():
"""Test warm session creation."""
print("\n=== Testing Warm Session Creation ===\n")
help:from tools.warm_session import TemplateManager, WarmSessionBootstrapper, SessionTemplate, ToolCallPattern, UserPattern
# Create a mock template for testing
template = SessionTemplate(
template_id="test_warm_001",
name="Test Warm Session",
description="Test template for warm session provisioning",
tool_patterns=[
ToolCallPattern(
tool_name="terminal",
success_rate=0.95,
common_args={"command": "ls -la"},
context_requirements=[]
),
ToolCallPattern(
tool_name="file_operations",
success_rate=0.90,
common_args={"operation": "read", "path": "README.md"},
context_requirements=[]
)
],
user_patterns=UserPattern(
request_style="direct",
feedback_style="terse",
common_domains=["coding", "file_management"],
preferred_tools=["terminal", "file_operations"]
),
recovery_patterns=[],
context_markers=["README.md", "src/", "tests/"],
created_at="2026-04-13T00:00:00",
source_session_id=None
)
# Create bootstrapper
manager = TemplateManager()
bootstrapper = WarmSessionBootstrapper(manager)
# Test message
test_message = "Help me understand the project structure"
print(f"Creating warm session with message: {test_message}\n")
messages = bootstrapper.create_warm_session_messages(template, test_message)
print(f"Generated {len(messages)} messages:")
for i, msg in enumerate(messages):
role = msg.get("role", "unknown")
content = msg.get("content", "")
if role == "system":
print(f"\n[System Context] ({len(content)} chars)")
print(content[:100] + "..." if len(content) > 100 else content)
elif role == "user":
print(f"\n[User]: {content}")
elif role == "assistant":
print(f"\n[Assistant]: {content}")
if msg.get("tool_calls"):
for tc in msg["tool_calls"]:
func = tc.get("function", {})
print(f" Tool Call: {func.get('name')}({func.get('arguments')})")
elif role == "tool":
print(f" [Tool Result]: {content[:50]}...")
return True
def main():
"""Run all tests."""
print("Warm Session Provisioning Test Suite")
print("=" * 50)
tests = [
("Session Analysis", test_session_analysis),
("Template Creation", test_template_creation),
("Warm Session Creation", test_warm_session)
]
results = []
for name, test_func in tests:
print(f"\nRunning: {name}")
try:
result = test_func()
results.append((name, result))
print(f"Result: {'PASS' if result else 'FAIL'}")
except Exception as e:
print(f"Error: {e}")
results.append((name, False))
print("\n" + "=" * 50)
print("Test Results:")
passed = sum(1 for _, result in results if result)
total = len(results)
for name, result in results:
status = "✓ PASS" if result else "✗ FAIL"
print(f" {status}: {name}")
print(f"\nPassed: {passed}/{total}")
return 0 if passed == total else 1
if __name__ == "__main__":
sys.exit(main())

674
tools/warm_session.py Normal file
View File

@@ -0,0 +1,674 @@
"""
Warm Session Provisioning: Pre-Proficient Agent Sessions
This module implements the research framework for creating warm sessions
that start with marathon-level proficiency by pre-seeding with successful
patterns extracted from long-running sessions.
Issue: #327
"""
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
import hashlib
logger = logging.getLogger(__name__)
@dataclass
class ToolCallPattern:
"""Pattern for successful tool calls."""
tool_name: str
success_rate: float
common_args: Dict[str, Any]
error_recovery: Optional[str] = None
context_requirements: List[str] = None
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@dataclass
class UserPattern:
"""User interaction patterns."""
request_style: str # "direct", "detailed", "conversational"
feedback_style: str # "terse", "detailed", "corrections_only"
common_domains: List[str] # ["coding", "research", "file_management"]
preferred_tools: List[str]
@dataclass
class RecoveryPattern:
"""Error recovery patterns."""
error_type: str
error_message_pattern: str
recovery_strategy: str
success_rate: float
@dataclass
class SessionTemplate:
"""Template for warm sessions."""
template_id: str
name: str
description: str
tool_patterns: List[ToolCallPattern]
user_patterns: UserPattern
recovery_patterns: List[RecoveryPattern]
context_markers: List[str] # Key context elements to preserve
created_at: str
source_session_id: Optional[str] = None
version: str = "1.0"
def to_dict(self) -> Dict[str, Any]:
return {
"template_id": self.template_id,
"name": self.name,
"description": self.description,
"tool_patterns": [p.to_dict() for p in self.tool_patterns],
"user_patterns": asdict(self.user_patterns),
"recovery_patterns": [asdict(r) for r in self.recovery_patterns],
"context_markers": self.context_markers,
"created_at": self.created_at,
"source_session_id": self.source_session_id,
"version": self.version
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'SessionTemplate':
"""Create template from dictionary."""
tool_patterns = [
ToolCallPattern(**p) for p in data.get("tool_patterns", [])
]
user_patterns = UserPattern(**data.get("user_patterns", {}))
recovery_patterns = [
RecoveryPattern(**r) for r in data.get("recovery_patterns", [])
]
return cls(
template_id=data["template_id"],
name=data["name"],
description=data["description"],
tool_patterns=tool_patterns,
user_patterns=user_patterns,
recovery_patterns=recovery_patterns,
context_markers=data.get("context_markers", []),
created_at=data.get("created_at", datetime.now().isoformat()),
source_session_id=data.get("source_session_id"),
version=data.get("version", "1.0")
)
def generate_id(self) -> str:
"""Generate deterministic ID from content."""
content = json.dumps(self.to_dict(), sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]
class SessionProfiler:
"""Analyze sessions to extract proficiency patterns."""
def __init__(self, session_db=None):
self.session_db = session_db
def analyze_session(self, session_id: str) -> Dict[str, Any]:
"""
Analyze a session for proficiency markers.
Returns:
Dict with proficiency analysis including:
- tool_success_rates: Success rate per tool
- error_patterns: Common error patterns
- user_patterns: User interaction style
- context_markers: Key context elements
"""
if not self.session_db:
return {"error": "No session database available"}
try:
messages = self.session_db.get_messages(session_id)
if not messages:
return {"error": "No messages found"}
analysis = {
"session_id": session_id,
"message_count": len(messages),
"tool_calls": self._analyze_tool_calls(messages),
"error_patterns": self._analyze_errors(messages),
"user_patterns": self._analyze_user_patterns(messages),
"context_markers": self._extract_context_markers(messages),
"proficiency_score": self._calculate_proficiency(messages)
}
return analysis
except Exception as e:
logger.error(f"Session analysis failed: {e}")
return {"error": str(e)}
def _analyze_tool_calls(self, messages: List[Dict]) -> Dict[str, Any]:
"""Analyze tool call patterns."""
tool_stats = {}
for msg in messages:
if msg.get("role") == "assistant" and msg.get("tool_calls"):
for tool_call in msg["tool_calls"]:
tool_name = tool_call.get("function", {}).get("name", "unknown")
if tool_name not in tool_stats:
tool_stats[tool_name] = {"calls": 0, "successes": 0, "errors": 0}
tool_stats[tool_name]["calls"] += 1
# Check for tool results in subsequent messages
for i, msg in enumerate(messages):
if msg.get("role") == "tool":
# Find the corresponding tool call
for j in range(i-1, -1, -1):
if messages[j].get("role") == "assistant" and messages[j].get("tool_calls"):
for tool_call in messages[j]["tool_calls"]:
tool_name = tool_call.get("function", {}).get("name", "unknown")
if tool_name in tool_stats:
# Check if result indicates success or error
content = msg.get("content", "")
if "error" in content.lower() or "failed" in content.lower():
tool_stats[tool_name]["errors"] += 1
else:
tool_stats[tool_name]["successes"] += 1
break
# Calculate success rates
for tool_name, stats in tool_stats.items():
total = stats["successes"] + stats["errors"]
stats["success_rate"] = stats["successes"] / total if total > 0 else 0
return tool_stats
def _analyze_errors(self, messages: List[Dict]) -> List[Dict[str, Any]]:
"""Analyze error patterns."""
errors = []
for msg in messages:
if msg.get("role") == "tool":
content = msg.get("content", "")
if "error" in content.lower() or "failed" in content.lower():
errors.append({
"content": content[:200],
"timestamp": msg.get("timestamp"),
"tool_call_id": msg.get("tool_call_id")
})
return errors
def _analyze_user_patterns(self, messages: List[Dict]) -> Dict[str, Any]:
"""Analyze user interaction patterns."""
user_messages = [m for m in messages if m.get("role") == "user"]
if not user_messages:
return {}
# Analyze message lengths
lengths = [len(m.get("content", "")) for m in user_messages]
avg_length = sum(lengths) / len(lengths)
# Analyze question patterns
questions = sum(1 for m in user_messages if "?" in m.get("content", ""))
question_ratio = questions / len(user_messages)
# Analyze command patterns
commands = sum(1 for m in user_messages if m.get("content", "").startswith(("/")))
command_ratio = commands / len(user_messages)
return {
"message_count": len(user_messages),
"avg_message_length": avg_length,
"question_ratio": question_ratio,
"command_ratio": command_ratio,
"style": self._classify_user_style(avg_length, question_ratio, command_ratio)
}
def _classify_user_style(self, avg_length: float, question_ratio: float, command_ratio: float) -> str:
"""Classify user interaction style."""
if command_ratio > 0.3:
return "command_driven"
elif question_ratio > 0.5:
return "conversational"
elif avg_length < 50:
return "terse"
elif avg_length > 200:
return "detailed"
else:
return "balanced"
def _extract_context_markers(self, messages: List[Dict]) -> List[str]:
"""Extract key context markers from session."""
markers = []
# Look for file paths, URLs, code snippets
for msg in messages:
content = msg.get("content", "")
# File paths
import re
file_paths = re.findall(r'[/.\][\w/.-]+\.\w+', content)
markers.extend(file_paths[:3]) # Limit to 3 per message
# URLs
urls = re.findall(r'https?://[^\s]+', content)
markers.extend(urls[:2])
# Code snippets (simple heuristic)
if "```" in content:
markers.append("code_block")
# Deduplicate and limit
return list(set(markers))[:20]
def _calculate_proficiency(self, messages: List[Dict]) -> float:
"""
Calculate session proficiency score (0-1).
Based on:
- Tool call success rate
- Error recovery rate
- Context establishment
"""
if len(messages) < 10:
return 0.0
# Simple proficiency calculation
tool_calls = sum(1 for m in messages if m.get("role") == "assistant" and m.get("tool_calls"))
tool_results = sum(1 for m in messages if m.get("role") == "tool")
if tool_calls == 0:
return 0.5 # No tool calls, neutral score
# Ratio of successful tool results
successful_results = sum(
1 for m in messages
if m.get("role") == "tool" and
"error" not in m.get("content", "").lower() and
"failed" not in m.get("content", "").lower()
)
success_rate = successful_results / tool_results if tool_results > 0 else 0
# Factor in session length (marathon sessions get bonus)
length_factor = min(1.0, len(messages) / 100) # Cap at 100 messages
return (success_rate * 0.7) + (length_factor * 0.3)
class TemplateManager:
"""Manage session templates."""
def __init__(self, template_dir: Path = None):
self.template_dir = template_dir or Path.home() / ".hermes" / "session_templates"
self.template_dir.mkdir(parents=True, exist_ok=True)
def save_template(self, template: SessionTemplate) -> Path:
"""Save a session template."""
template_path = self.template_dir / f"{template.template_id}.json"
with open(template_path, 'w') as f:
json.dump(template.to_dict(), f, indent=2)
logger.info(f"Saved template {template.template_id} to {template_path}")
return template_path
def load_template(self, template_id: str) -> Optional[SessionTemplate]:
"""Load a session template."""
template_path = self.template_dir / f"{template_id}.json"
if not template_path.exists():
logger.warning(f"Template {template_id} not found")
return None
try:
with open(template_path, 'r') as f:
data = json.load(f)
return SessionTemplate.from_dict(data)
except Exception as e:
logger.error(f"Failed to load template {template_id}: {e}")
return None
def list_templates(self) -> List[Dict[str, Any]]:
"""List all available templates."""
templates = []
for template_path in self.template_dir.glob("*.json"):
try:
with open(template_path, 'r') as f:
data = json.load(f)
templates.append({
"template_id": data.get("template_id"),
"name": data.get("name"),
"description": data.get("description"),
"created_at": data.get("created_at"),
"source_session_id": data.get("source_session_id")
})
except Exception as e:
logger.warning(f"Failed to read template {template_path}: {e}")
return templates
def delete_template(self, template_id: str) -> bool:
"""Delete a session template."""
template_path = self.template_dir / f"{template_id}.json"
if template_path.exists():
template_path.unlink()
logger.info(f"Deleted template {template_id}")
return True
return False
class WarmSessionBootstrapper:
"""Bootstrap warm sessions from templates."""
def __init__(self, template_manager: TemplateManager = None):
self.template_manager = template_manager or TemplateManager()
def create_warm_session_messages(
self,
template: SessionTemplate,
user_message: str
) -> List[Dict[str, Any]]:
"""
Create initial messages for a warm session.
This injects context from the template to give the agent
a "memory" of successful patterns.
"""
messages = []
# Add system context about the template
template_context = f"""You are starting a warm session with pre-loaded proficiency patterns.
Template: {template.name}
Description: {template.description}
User Style: {template.user_patterns.request_style}
Preferred Tools: {', '.join(template.user_patterns.preferred_tools)}
Successful Tool Patterns:
{self._format_tool_patterns(template.tool_patterns)}
Error Recovery Patterns:
{self._format_recovery_patterns(template.recovery_patterns)}
Use these patterns to provide more reliable responses from the start."""
messages.append({
"role": "system",
"content": template_context
})
# Add synthetic successful tool calls to establish patterns
synthetic_messages = self._create_synthetic_tool_calls(template)
messages.extend(synthetic_messages)
# Add the actual user message
messages.append({
"role": "user",
"content": user_message
})
return messages
def _format_tool_patterns(self, patterns: List[ToolCallPattern]) -> str:
"""Format tool patterns for context."""
if not patterns:
return "No specific patterns recorded."
lines = []
for p in patterns[:5]: # Limit to top 5
lines.append(f"- {p.tool_name}: {p.success_rate:.0%} success, common args: {p.common_args}")
return "\n".join(lines)
def _format_recovery_patterns(self, patterns: List[RecoveryPattern]) -> str:
"""Format recovery patterns for context."""
if not patterns:
return "No recovery patterns recorded."
lines = []
for p in patterns[:3]: # Limit to top 3
lines.append(f"- {p.error_type}: {p.recovery_strategy} ({p.success_rate:.0%} success)")
return "\n".join(lines)
def _create_synthetic_tool_calls(self, template: SessionTemplate) -> List[Dict[str, Any]]:
"""
Create synthetic tool call examples from template.
This gives the agent examples of successful tool usage
without actually executing them.
"""
messages = []
# Create one synthetic successful tool call per pattern
for i, pattern in enumerate(template.tool_patterns[:3]):
# Create a synthetic user request
user_msg = {
"role": "user",
"content": f"[Synthetic example {i+1}] Show me how to use {pattern.tool_name}"
}
# Create a synthetic assistant response with tool call
assistant_msg = {
"role": "assistant",
"content": f"I'll demonstrate using {pattern.tool_name}.",
"tool_calls": [{
"id": f"synthetic_{i}",
"type": "function",
"function": {
"name": pattern.tool_name,
"arguments": json.dumps(pattern.common_args)
}
}]
}
# Create a synthetic successful result
tool_msg = {
"role": "tool",
"tool_call_id": f"synthetic_{i}",
"content": "Success: Operation completed successfully."
}
messages.extend([user_msg, assistant_msg, tool_msg])
return messages
# CLI Integration
def create_template_from_session(session_id: str, session_db, name: str = None) -> Optional[SessionTemplate]:
"""Create a template from an existing session."""
profiler = SessionProfiler(session_db)
analysis = profiler.analyze_session(session_id)
if "error" in analysis:
logger.error(f"Cannot create template: {analysis['error']}")
return None
# Convert analysis to template
tool_patterns = []
for tool_name, stats in analysis.get("tool_calls", {}).items():
if stats.get("success_rate", 0) > 0.7: # Only include successful patterns
tool_patterns.append(ToolCallPattern(
tool_name=tool_name,
success_rate=stats["success_rate"],
common_args={}, # Would need more analysis
context_requirements=[]
))
user_patterns_data = analysis.get("user_patterns", {})
user_patterns = UserPattern(
request_style=user_patterns_data.get("style", "balanced"),
feedback_style="balanced",
common_domains=[],
preferred_tools=list(analysis.get("tool_calls", {}).keys())[:3]
)
template = SessionTemplate(
template_id=f"warm_{session_id[:8]}",
name=name or f"Warm session from {session_id[:8]}",
description=f"Auto-generated from session {session_id}",
tool_patterns=tool_patterns,
user_patterns=user_patterns,
recovery_patterns=[],
context_markers=analysis.get("context_markers", []),
created_at=datetime.now().isoformat(),
source_session_id=session_id
)
return template
def warm_session_command(args):
"""CLI command for warm session management."""
import argparse
parser = argparse.ArgumentParser(description="Warm session provisioning")
subparsers = parser.add_subparsers(dest="command")
# Analyze command
analyze_parser = subparsers.add_parser("analyze", help="Analyze a session")
analyze_parser.add_argument("session_id", help="Session ID to analyze")
# Create template command
create_parser = subparsers.add_parser("create-template", help="Create template from session")
create_parser.add_argument("session_id", help="Session ID to create template from")
create_parser.add_argument("--name", "-n", help="Template name")
# List templates command
subparsers.add_parser("list-templates", help="List available templates")
# Test warm session command
test_parser = subparsers.add_parser("test", help="Test warm session creation")
test_parser.add_argument("template_id", help="Template ID to test")
test_parser.add_argument("message", help="Test message")
# Parse args
parsed = parser.parse_args(args)
if not parsed.command:
parser.print_help()
return 1
# Import session DB
try:
from hermes_state import SessionDB
session_db = SessionDB()
except ImportError:
print("Error: Cannot import SessionDB")
return 1
if parsed.command == "analyze":
profiler = SessionProfiler(session_db)
analysis = profiler.analyze_session(parsed.session_id)
print(f"\n=== Session Analysis: {parsed.session_id} ===\n")
if "error" in analysis:
print(f"Error: {analysis['error']}")
return 1
print(f"Messages: {analysis['message_count']}")
print(f"Proficiency Score: {analysis['proficiency_score']:.2f}")
print("\nTool Call Analysis:")
for tool, stats in analysis.get("tool_calls", {}).items():
print(f" {tool}: {stats.get('success_rate', 0):.0%} success ({stats.get('calls', 0)} calls)")
print("\nUser Patterns:")
user = analysis.get("user_patterns", {})
print(f" Style: {user.get('style', 'unknown')}")
print(f" Avg Length: {user.get('avg_message_length', 0):.0f} chars")
print("\nContext Markers:")
for marker in analysis.get("context_markers", [])[:5]:
print(f" {marker}")
return 0
elif parsed.command == "create-template":
template = create_template_from_session(
parsed.session_id,
session_db,
name=parsed.name
)
if template:
manager = TemplateManager()
path = manager.save_template(template)
print(f"Created template: {template.template_id}")
print(f"Saved to: {path}")
return 0
else:
print("Failed to create template")
return 1
elif parsed.command == "list-templates":
manager = TemplateManager()
templates = manager.list_templates()
if not templates:
print("No templates found.")
return 0
print("\n=== Available Templates ===\n")
for t in templates:
print(f"ID: {t['template_id']}")
print(f"Name: {t['name']}")
print(f"Description: {t['description']}")
print(f"Created: {t['created_at']}")
if t.get('source_session_id'):
print(f"Source: {t['source_session_id']}")
print()
return 0
elif parsed.command == "test":
manager = TemplateManager()
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.create_warm_session_messages(template, parsed.message)
print(f"\n=== Warm Session Test: {parsed.template_id} ===\n")
print(f"Generated {len(messages)} messages:")
for i, msg in enumerate(messages):
role = msg.get("role", "unknown")
content = msg.get("content", "")
if role == "system":
print(f"\n[System Context]")
print(content[:200] + "..." if len(content) > 200 else content)
elif role == "user":
print(f"\n[User]: {content}")
elif role == "assistant":
print(f"\n[Assistant]: {content}")
if msg.get("tool_calls"):
for tc in msg["tool_calls"]:
func = tc.get("function", {})
print(f" Tool Call: {func.get('name')}({func.get('arguments')})")
elif role == "tool":
print(f" [Tool Result]: {content[:100]}...")
return 0
return 1
if __name__ == "__main__":
import sys
sys.exit(warm_session_command(sys.argv[1:]))