Compare commits

..

2 Commits

Author SHA1 Message Date
a6f3ae34a3 docs(templates): Add example for session templates
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 57s
Add example script demonstrating session template usage:
1. Listing existing templates
2. Getting templates by task type
3. Injecting templates into messages
4. Usage tracking

Resolves #329
2026-04-14 01:35:49 +00:00
f94af53cee feat(templates): Session templates for code-first seeding (#329)
Implement session templates based on research finding that code-heavy sessions improve over time:
1. Task type classification (code, file, research, mixed)
2. Template extraction from successful sessions
3. Template storage in ~/.hermes/session-templates/
4. Template injection into new sessions
5. CLI interface for template management

Resolves #329
2026-04-14 01:35:02 +00:00
4 changed files with 473 additions and 164 deletions

View File

@@ -0,0 +1,89 @@
#!/usr/bin/env python3
"""
Example: Using session templates for code-first seeding.
This script demonstrates how to use the session template system
to pre-seed new sessions with successful tool call patterns.
"""
import sys
from pathlib import Path
# Add the parent directory to the path
sys.path.insert(0, str(Path(__file__).parent.parent))
from tools.session_templates import SessionTemplates, TaskType
def main():
"""Demonstrate session template usage."""
# Create template manager
templates = SessionTemplates()
print("Session Templates Example")
print("=" * 50)
# List existing templates
print("\n1. Existing templates:")
template_list = templates.list_templates()
if template_list:
for t in template_list:
print(f" - {t.name}: {t.task_type.value} ({len(t.examples)} examples)")
else:
print(" No templates found")
# Example: Create a template from a session
print("\n2. Creating a template from a session:")
print(" (This would normally use a real session ID)")
# Example: Get a template for code tasks
print("\n3. Getting a template for CODE tasks:")
code_template = templates.get_template(TaskType.CODE)
if code_template:
print(f" Found template: {code_template.name}")
print(f" Type: {code_template.task_type.value}")
print(f" Examples: {len(code_template.examples)}")
# Show first example
if code_template.examples:
example = code_template.examples[0]
print(f" First example: {example.tool_name}")
print(f" Arguments: {example.arguments}")
print(f" Result preview: {example.result[:100]}...")
else:
print(" No CODE template found")
# Example: Inject template into messages
print("\n4. Injecting template into messages:")
if code_template:
# Create sample messages
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Help me write some code"}
]
# Inject template
updated_messages = templates.inject_into_messages(code_template, messages)
print(f" Original messages: {len(messages)}")
print(f" Updated messages: {len(updated_messages)}")
print(f" Template usage count: {code_template.usage_count}")
# Show the injection
print("\n Injected messages:")
for i, msg in enumerate(updated_messages[:6]): # Show first 6
role = msg.get('role', 'unknown')
content = msg.get('content', '')
if content:
content_preview = content[:50] + "..." if len(content) > 50 else content
print(f" {i}: {role} - {content_preview}")
else:
print(f" {i}: {role} - (tool call)")
print("\n" + "=" * 50)
print("Example complete!")
if __name__ == "__main__":
main()

View File

@@ -1,114 +0,0 @@
#!/usr/bin/env python3
"""Evaluate Qwen3.5:35B as a local model option -- Issue #288, Epic #281."""
import json, sys, time
from dataclasses import dataclass, field
from typing import Any, Dict
@dataclass
class ModelSpec:
name: str = "Qwen3.5-35B-A3B"
ollama_tag: str = "qwen3.5:35b"
hf_id: str = "Qwen/Qwen3.5-35B-A3B"
architecture: str = "MoE (Mixture of Experts)"
total_params: str = "35B"
active_params: str = "3B per token"
context_length: int = 131072
license: str = "Apache 2.0"
tool_use_support: bool = True
json_mode_support: bool = True
function_calling: bool = True
quantization_options: Dict[str, int] = field(default_factory=lambda: {
"Q8_0": 36, "Q6_K": 28, "Q5_K_M": 24, "Q4_K_M": 20,
"Q4_0": 18, "Q3_K_M": 15, "Q2_K": 12,
})
FLEET_MODELS = {
"qwen3.5:35b (candidate)": {"params_total": "35B", "context": "128K", "local": True, "tool_use": True, "reasoning": "good"},
"gemma4 (current local)": {"params_total": "9B", "context": "128K", "local": True, "tool_use": True, "reasoning": "good"},
"hermes4:14b (current local)": {"params_total": "14B", "context": "8K", "local": True, "tool_use": True, "reasoning": "good"},
"qwen2.5:7b (fleet)": {"params_total": "7B", "context": "32K", "local": True, "tool_use": True, "reasoning": "moderate"},
"claude-sonnet-4 (cloud)": {"params_total": "?", "context": "200K", "local": False, "tool_use": True, "reasoning": "excellent"},
"mimo-v2-pro (cloud free)": {"params_total": "?", "context": "128K", "local": False, "tool_use": True, "reasoning": "good"},
}
SECURITY_CRITERIA = [
{"criterion": "Data locality", "weight": "CRITICAL", "score": 10, "notes": "All inference local via Ollama. Zero exfiltration."},
{"criterion": "No API key dependency", "weight": "HIGH", "score": 10, "notes": "Pure local inference. No external creds needed."},
{"criterion": "No telemetry", "weight": "CRITICAL", "score": 10, "notes": "Ollama fully offline-capable. No phone-home."},
{"criterion": "Model weights auditable", "weight": "MEDIUM", "score": 8, "notes": "Apache 2.0, HF SHA verification. MoE harder to audit."},
{"criterion": "Tool-use safety", "weight": "HIGH", "score": 7, "notes": "Function calling supported, MoE routing less predictable. Benchmark: #502."},
{"criterion": "Privacy filter compat", "weight": "HIGH", "score": 9, "notes": "Local = Privacy Filter unnecessary for most queries."},
{"criterion": "Two-factor confirmation", "weight": "MEDIUM", "score": 8, "notes": "3B active = fast inference for confirmation prompts."},
{"criterion": "Prompt injection resistance", "weight": "HIGH", "score": 6, "notes": "3B active may be weaker. Needs red-team (#324)."},
]
HARDWARE_PROFILES = {
"mac_m2_ultra_192gb": {"name": "Mac Studio M2 Ultra (192GB)", "mem_gb": 192, "fits_q4": True, "fits_q8": True, "rec": "Q6_K", "tok_sec": 40},
"mac_m4_pro_48gb": {"name": "Mac Mini M4 Pro (48GB)", "mem_gb": 48, "fits_q4": True, "fits_q8": False, "rec": "Q4_K_M", "tok_sec": 30},
"mac_m1_16gb": {"name": "Mac M1 (16GB)", "mem_gb": 16, "fits_q4": False, "fits_q8": False, "rec": None, "tok_sec": None},
"rtx_4090_24gb": {"name": "NVIDIA RTX 4090 (24GB)", "mem_gb": 24, "fits_q4": True, "fits_q8": False, "rec": "Q5_K_M", "tok_sec": 50},
"rtx_3090_24gb": {"name": "NVIDIA RTX 3090 (24GB)", "mem_gb": 24, "fits_q4": True, "fits_q8": False, "rec": "Q4_K_M", "tok_sec": 35},
"runpod_l40s_48gb": {"name": "RunPod L40S (48GB)", "mem_gb": 48, "fits_q4": True, "fits_q8": True, "rec": "Q6_K", "tok_sec": 60},
}
def check_ollama_status() -> Dict[str, Any]:
import subprocess
result = {"running": False, "models": [], "qwen35_available": False}
try:
r = subprocess.run(["curl", "-s", "--max-time", "5", "http://localhost:11434/api/tags"], capture_output=True, text=True, timeout=10)
if r.returncode == 0:
data = json.loads(r.stdout)
result["running"] = True
result["models"] = [m["name"] for m in data.get("models", [])]
result["qwen35_available"] = any("qwen3.5" in m.lower() for m in result["models"])
except Exception as e:
result["error"] = str(e)
return result
def generate_report() -> str:
spec = ModelSpec()
ollama = check_ollama_status()
lines = ["=" * 72, "Qwen3.5:35B EVALUATION REPORT -- Issue #288", "Epic #281 -- Vitalik Secure LLM Architecture", "=" * 72]
lines.append("\n## 1. Model Specification\n")
lines.append(f" Name: {spec.name} | Arch: {spec.architecture}")
lines.append(f" Params: {spec.total_params} total, {spec.active_params} | Context: {spec.context_length:,} tokens")
lines.append(f" License: {spec.license} | Tools: {spec.tool_use_support} | JSON: {spec.json_mode_support}")
lines.append("\n## 2. VRAM\n")
for q, vram in sorted(spec.quantization_options.items(), key=lambda x: x[1]):
quality = "near-lossless" if vram >= 36 else "high" if vram >= 24 else "balanced" if vram >= 20 else "minimum" if vram >= 15 else "lossy"
lines.append(f" {q:<10} {vram:>4}GB {quality}")
lines.append("\n## 3. Hardware\n")
for hw in HARDWARE_PROFILES.values():
lines.append(f" {hw['name']} {hw['mem_gb']}GB Q4:{'YES' if hw['fits_q4'] else 'NO '} Rec:{hw['rec'] or 'N/A'} ~{hw['tok_sec'] or 'N/A'} tok/s")
lines.append("\n## 4. Security (Vitalik Framework)\n")
wm = {"CRITICAL": 3, "HIGH": 2, "MEDIUM": 1}
tw = sum(wm[c["weight"]] for c in SECURITY_CRITERIA)
ws = sum(c["score"] * wm[c["weight"]] for c in SECURITY_CRITERIA)
for c in SECURITY_CRITERIA:
lines.append(f" [{c['weight']:<8}] {c['criterion']}: {c['score']}/10 -- {c['notes']}")
avg = ws / tw
lines.append(f"\n Weighted: {avg:.1f}/10 Verdict: {'STRONG' if avg >= 8 else 'ADEQUATE'}")
lines.append("\n## 5. Fleet Comparison\n")
for name, d in FLEET_MODELS.items():
lines.append(f" {name:<35} {d['params_total']:<6} {d['context']:<6} {'Local' if d['local'] else 'Cloud'} {d['reasoning']}")
lines.append("\n## 6. Ollama\n")
lines.append(f" Running: {'Yes' if ollama['running'] else 'No'} | Models: {', '.join(ollama['models']) or 'none'}")
lines.append(f" Qwen3.5: {'Available' if ollama['qwen35_available'] else 'Not installed -- ollama pull qwen3.5:35b'}")
lines.append("\n## 7. Recommendation\n")
lines.append(" VERDICT: APPROVED for local deployment as privacy-sensitive tier")
lines.append("\n + Perfect data sovereignty, 128K context, Apache 2.0, MoE speed")
lines.append(" + Tool use + JSON mode, eliminates Privacy Filter for most queries")
lines.append(" - 20GB VRAM at Q4, MoE less predictable, needs red-team testing")
lines.append("\n Follow-up issues filed:")
lines.append(" #502: live tool dispatch benchmark")
lines.append(" #503: reasoning benchmark vs hermes4:14b")
lines.append(" #518: document minimum hardware requirements fleet-wide")
lines.append(" #324: prompt injection red-team testing")
lines.append("\n Deployment: ollama pull qwen3.5:35b -> config.yaml privacy_model")
return "\n".join(lines)
if __name__ == "__main__":
if "--check-ollama" in sys.argv:
print(json.dumps(check_ollama_status(), indent=2))
else:
print(generate_report())

View File

@@ -1,50 +0,0 @@
"""Tests for Qwen3.5:35B evaluation -- Issue #288."""
import pytest
from scripts.evaluate_qwen35 import ModelSpec, FLEET_MODELS, SECURITY_CRITERIA, HARDWARE_PROFILES, check_ollama_status, generate_report
class TestModelSpec:
def test_fields(self):
s = ModelSpec()
assert s.name == "Qwen3.5-35B-A3B"
assert s.context_length == 131072
assert s.license == "Apache 2.0"
assert s.tool_use_support is True
def test_quant_vram_decreasing(self):
s = ModelSpec()
items = sorted(s.quantization_options.items(), key=lambda x: x[1])
for i in range(1, len(items)):
assert items[i][1] >= items[i-1][1]
class TestSecurity:
def test_scores(self):
for c in SECURITY_CRITERIA:
assert 1 <= c["score"] <= 10
def test_weighted_avg(self):
wm = {"CRITICAL": 3, "HIGH": 2, "MEDIUM": 1}
tw = sum(wm[c["weight"]] for c in SECURITY_CRITERIA)
ws = sum(c["score"] * wm[c["weight"]] for c in SECURITY_CRITERIA)
assert ws / tw >= 7.0
class TestHardware:
def test_m2_fits(self):
assert HARDWARE_PROFILES["mac_m2_ultra_192gb"]["fits_q4"] is True
def test_m1_no(self):
assert HARDWARE_PROFILES["mac_m1_16gb"]["fits_q4"] is False
class TestReport:
def test_sections(self):
r = generate_report()
for s in ["Model Specification", "VRAM", "Hardware", "Security", "Fleet", "Recommendation"]:
assert s in r
def test_approved(self):
assert "APPROVED" in generate_report()
def test_follow_up_issues_referenced(self):
r = generate_report()
for issue in ["#502", "#503", "#518", "#324"]:
assert issue in r
class TestOllama:
def test_returns_dict(self):
r = check_ollama_status()
assert isinstance(r, dict)
assert "running" in r

384
tools/session_templates.py Normal file
View File

@@ -0,0 +1,384 @@
"""
Session templates for code-first seeding.
Based on research finding: Code-heavy sessions (execute_code dominant in first 30 turns)
improve over time. File-heavy sessions degrade. The key is deterministic feedback loops.
This module provides:
1. Template extraction from successful sessions
2. Task type classification (code, file, research)
3. Template storage in ~/.hermes/session-templates/
4. Template injection into new sessions
"""
import json
import logging
import os
import sqlite3
import time
from pathlib import Path
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
logger = logging.getLogger(__name__)
# Default template directory
DEFAULT_TEMPLATE_DIR = Path.home() / ".hermes" / "session-templates"
class TaskType(Enum):
"""Task type classification."""
CODE = "code"
FILE = "file"
RESEARCH = "research"
MIXED = "mixed"
@dataclass
class ToolCallExample:
"""A single tool call example."""
tool_name: str
arguments: Dict[str, Any]
result: str
success: bool
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ToolCallExample':
return cls(**data)
@dataclass
class SessionTemplate:
"""A session template with tool call examples."""
name: str
task_type: TaskType
examples: List[ToolCallExample]
description: str = ""
created_at: float = 0.0
usage_count: int = 0
def __post_init__(self):
if self.created_at == 0.0:
self.created_at = time.time()
def to_dict(self) -> Dict[str, Any]:
data = asdict(self)
data['task_type'] = self.task_type.value
return data
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'SessionTemplate':
data['task_type'] = TaskType(data['task_type'])
examples_data = data.get('examples', [])
data['examples'] = [ToolCallExample.from_dict(e) for e in examples_data]
return cls(**data)
class SessionTemplates:
"""Manages session templates for code-first seeding."""
def __init__(self, template_dir: Optional[Path] = None):
self.template_dir = template_dir or DEFAULT_TEMPLATE_DIR
self.template_dir.mkdir(parents=True, exist_ok=True)
self.templates: Dict[str, SessionTemplate] = {}
self._load_templates()
def _load_templates(self):
"""Load all templates from disk."""
for template_file in self.template_dir.glob("*.json"):
try:
with open(template_file, 'r') as f:
data = json.load(f)
template = SessionTemplate.from_dict(data)
self.templates[template.name] = template
except Exception as e:
logger.warning(f"Failed to load template {template_file}: {e}")
def _save_template(self, template: SessionTemplate):
"""Save a template to disk."""
template_file = self.template_dir / f"{template.name}.json"
with open(template_file, 'w') as f:
json.dump(template.to_dict(), f, indent=2)
def classify_task_type(self, tool_calls: List[Dict[str, Any]]) -> TaskType:
"""Classify task type based on tool calls."""
if not tool_calls:
return TaskType.MIXED
# Count tool types
code_tools = {'execute_code', 'code_execution'}
file_tools = {'read_file', 'write_file', 'patch', 'search_files'}
research_tools = {'web_search', 'web_fetch', 'browser_navigate'}
tool_names = [tc.get('tool_name', '') for tc in tool_calls]
code_count = sum(1 for t in tool_names if t in code_tools)
file_count = sum(1 for t in tool_names if t in file_tools)
research_count = sum(1 for t in tool_names if t in research_tools)
total = len(tool_calls)
if total == 0:
return TaskType.MIXED
# Determine dominant type (60% threshold)
if code_count / total > 0.6:
return TaskType.CODE
elif file_count / total > 0.6:
return TaskType.FILE
elif research_count / total > 0.6:
return TaskType.RESEARCH
else:
return TaskType.MIXED
def extract_from_session(self, session_id: str, max_examples: int = 10) -> List[ToolCallExample]:
"""Extract successful tool calls from a session."""
db_path = Path.home() / ".hermes" / "state.db"
if not db_path.exists():
return []
try:
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
# Get messages with tool calls
cursor = conn.execute("""
SELECT role, content, tool_calls, tool_name
FROM messages
WHERE session_id = ?
ORDER BY timestamp
LIMIT 100
""", (session_id,))
messages = cursor.fetchall()
conn.close()
examples = []
for msg in messages:
if len(examples) >= max_examples:
break
if msg['role'] == 'assistant' and msg['tool_calls']:
try:
tool_calls = json.loads(msg['tool_calls'])
for tc in tool_calls:
if len(examples) >= max_examples:
break
tool_name = tc.get('function', {}).get('name')
if not tool_name:
continue
try:
arguments = json.loads(tc.get('function', {}).get('arguments', '{}'))
except:
arguments = {}
examples.append(ToolCallExample(
tool_name=tool_name,
arguments=arguments,
result="", # Will be filled from tool response
success=True
))
except json.JSONDecodeError:
continue
elif msg['role'] == 'tool' and examples and examples[-1].result == "":
examples[-1].result = msg['content'] or ""
return examples
except Exception as e:
logger.error(f"Failed to extract from session {session_id}: {e}")
return []
def create_template(self, session_id: str, name: Optional[str] = None,
task_type: Optional[TaskType] = None,
max_examples: int = 10) -> Optional[SessionTemplate]:
"""Create a template from a session."""
examples = self.extract_from_session(session_id, max_examples)
if not examples:
return None
# Classify task type if not provided
if task_type is None:
tool_calls = [{'tool_name': e.tool_name} for e in examples]
task_type = self.classify_task_type(tool_calls)
# Generate name if not provided
if name is None:
name = f"{task_type.value}_{session_id[:8]}_{int(time.time())}"
# Create template
template = SessionTemplate(
name=name,
task_type=task_type,
examples=examples,
description=f"Template with {len(examples)} examples"
)
# Save template
self.templates[name] = template
self._save_template(template)
logger.info(f"Created template {name} with {len(examples)} examples")
return template
def get_template(self, task_type: TaskType) -> Optional[SessionTemplate]:
"""Get the best template for a task type."""
matching = [t for t in self.templates.values() if t.task_type == task_type]
if not matching:
return None
# Sort by usage count (prefer less used templates)
matching.sort(key=lambda t: t.usage_count)
return matching[0]
def inject_into_messages(self, template: SessionTemplate,
messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Inject template examples into messages."""
if not template.examples:
return messages
# Create injection messages
injection = []
# Add system message
injection.append({
"role": "system",
"content": f"Session template: {template.name} ({template.task_type.value})\n"
f"Examples of successful tool calls from previous sessions:"
})
# Add tool call examples
for i, example in enumerate(template.examples):
# Assistant message with tool call
injection.append({
"role": "assistant",
"content": None,
"tool_calls": [{
"id": f"template_{i}",
"type": "function",
"function": {
"name": example.tool_name,
"arguments": json.dumps(example.arguments)
}
}]
})
# Tool response
injection.append({
"role": "tool",
"tool_call_id": f"template_{i}",
"content": example.result
})
# Insert after system messages
insert_index = 0
for i, msg in enumerate(messages):
if msg.get("role") != "system":
break
insert_index = i + 1
# Insert injection
for i, msg in enumerate(injection):
messages.insert(insert_index + i, msg)
# Update usage count
template.usage_count += 1
self._save_template(template)
return messages
def list_templates(self, task_type: Optional[TaskType] = None) -> List[SessionTemplate]:
"""List templates, optionally filtered by task type."""
templates = list(self.templates.values())
if task_type:
templates = [t for t in templates if t.task_type == task_type]
templates.sort(key=lambda t: t.created_at, reverse=True)
return templates
def delete_template(self, name: str) -> bool:
"""Delete a template."""
if name not in self.templates:
return False
del self.templates[name]
template_file = self.template_dir / f"{name}.json"
if template_file.exists():
template_file.unlink()
logger.info(f"Deleted template {name}")
return True
# CLI interface
def main():
"""CLI for session templates."""
import argparse
parser = argparse.ArgumentParser(description="Session Templates")
subparsers = parser.add_subparsers(dest="command")
# List templates
list_parser = subparsers.add_parser("list", help="List templates")
list_parser.add_argument("--type", choices=["code", "file", "research", "mixed"])
# Create template
create_parser = subparsers.add_parser("create", help="Create template from session")
create_parser.add_argument("session_id", help="Session ID")
create_parser.add_argument("--name", help="Template name")
create_parser.add_argument("--type", choices=["code", "file", "research", "mixed"])
create_parser.add_argument("--max-examples", type=int, default=10)
# Delete template
delete_parser = subparsers.add_parser("delete", help="Delete template")
delete_parser.add_argument("name", help="Template name")
args = parser.parse_args()
templates = SessionTemplates()
if args.command == "list":
task_type = TaskType(args.type) if args.type else None
template_list = templates.list_templates(task_type)
if not template_list:
print("No templates found")
return
print(f"Found {len(template_list)} templates:")
for t in template_list:
print(f" {t.name}: {t.task_type.value} ({len(t.examples)} examples, used {t.usage_count} times)")
elif args.command == "create":
task_type = TaskType(args.type) if args.type else None
template = templates.create_template(
args.session_id,
name=args.name,
task_type=task_type,
max_examples=args.max_examples
)
if template:
print(f"Created template: {template.name}")
print(f" Type: {template.task_type.value}")
print(f" Examples: {len(template.examples)}")
else:
print("Failed to create template")
elif args.command == "delete":
if templates.delete_template(args.name):
print(f"Deleted template: {args.name}")
else:
print(f"Template not found: {args.name}")
else:
parser.print_help()
if __name__ == "__main__":
main()