mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat: Add intelligent fallback and memory management to hanzo dev
- Automatic fallback handler tries all available AI options - Memory persistence with #memory commands (like Claude Desktop) - Smart context management with priority-based memories - Session tracking and user preferences - Export/import memory capabilities - Enhanced AI responses with contextual awareness - Version bump to 0.3.20
This commit is contained in:
@@ -0,0 +1,366 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Comprehensive demo script for Hanzo Dev - AI Coding OS.
|
||||
This demonstrates all features and orchestrator modes.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.progress import Progress, SpinnerColumn, TextColumn
|
||||
from rich.prompt import Prompt
|
||||
from rich import print as rprint
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent / "pkg" / "hanzo" / "src"))
|
||||
|
||||
from hanzo.dev import HanzoDevREPL, MultiClaudeOrchestrator
|
||||
from hanzo.orchestrator_config import get_orchestrator_config
|
||||
|
||||
console = Console()
|
||||
|
||||
async def demo_chat_interface():
|
||||
"""Demo the chat-first interface."""
|
||||
console.print("\n[bold cyan]═══ DEMO: Chat-First Interface ═══[/bold cyan]\n")
|
||||
|
||||
console.print("The REPL is [bold]chat-first by default[/bold]:")
|
||||
console.print("• Type naturally to chat with AI")
|
||||
console.print("• Use [cyan]/commands[/cyan] for special actions")
|
||||
console.print("• Use [cyan]#memory[/cyan] to manage context")
|
||||
|
||||
# Simulate chat interaction
|
||||
console.print("\n[dim]Example interaction:[/dim]")
|
||||
console.print("[dim white]╭" + "─" * 60 + "╮[/dim white]")
|
||||
console.print("[dim white]│[/dim white] › Write a Python function to calculate fibonacci")
|
||||
console.print("[dim white]╰" + "─" * 60 + "╯[/dim white]")
|
||||
|
||||
await asyncio.sleep(1)
|
||||
|
||||
console.print("\n[dim]AI Response:[/dim]")
|
||||
response = """def fibonacci(n):
|
||||
\"\"\"Calculate the nth Fibonacci number.\"\"\"
|
||||
if n <= 0:
|
||||
return 0
|
||||
elif n == 1:
|
||||
return 1
|
||||
else:
|
||||
return fibonacci(n-1) + fibonacci(n-2)"""
|
||||
|
||||
console.print(Panel(response, title="[bold cyan]AI Response[/bold cyan]",
|
||||
title_align="left", border_style="dim cyan"))
|
||||
|
||||
async def demo_orchestrator_modes():
|
||||
"""Demo different orchestrator modes."""
|
||||
console.print("\n[bold cyan]═══ DEMO: Orchestrator Modes ═══[/bold cyan]\n")
|
||||
|
||||
# Create a table of orchestrator modes
|
||||
table = Table(title="Available Orchestrators", show_header=True,
|
||||
header_style="bold magenta")
|
||||
table.add_column("Mode", style="cyan", width=20)
|
||||
table.add_column("Description", width=50)
|
||||
table.add_column("Requirements", style="yellow")
|
||||
|
||||
modes = [
|
||||
("gpt-5", "Ultimate code development AI", "OPENAI_API_KEY"),
|
||||
("gpt-4", "Production-ready OpenAI model", "OPENAI_API_KEY"),
|
||||
("claude", "Anthropic's Claude 3.5 Sonnet", "ANTHROPIC_API_KEY"),
|
||||
("codex", "OpenAI CLI (no API key needed)", "OpenAI CLI installed"),
|
||||
("local:llama3.2", "Local Llama model", "Ollama installed"),
|
||||
("gpt-5-pro-codex", "Hybrid mode with Codex", "Multiple tools"),
|
||||
("router:gpt-4", "Via Hanzo router", "Router configured"),
|
||||
("direct:claude", "Direct API access", "API key required"),
|
||||
]
|
||||
|
||||
for mode, desc, req in modes:
|
||||
table.add_row(mode, desc, req)
|
||||
|
||||
console.print(table)
|
||||
|
||||
# Test each mode
|
||||
console.print("\n[bold]Testing orchestrator configurations:[/bold]")
|
||||
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
console=console,
|
||||
) as progress:
|
||||
for mode, _, _ in modes:
|
||||
task = progress.add_task(f"Testing {mode}...", total=None)
|
||||
try:
|
||||
config = get_orchestrator_config(mode)
|
||||
await asyncio.sleep(0.5) # Simulate test
|
||||
progress.update(task, description=f"✓ {mode} configured")
|
||||
except Exception as e:
|
||||
progress.update(task, description=f"✗ {mode} failed: {e}")
|
||||
|
||||
async def demo_cli_tools():
|
||||
"""Demo CLI tool integrations."""
|
||||
console.print("\n[bold cyan]═══ DEMO: CLI Tool Integration ═══[/bold cyan]\n")
|
||||
|
||||
tools = [
|
||||
("OpenAI CLI", "openai", "openai api chat.completions.create"),
|
||||
("Claude Desktop", "claude", "claude chat"),
|
||||
("Gemini CLI", "gemini", "gemini prompt"),
|
||||
("Ollama", "ollama", "ollama run llama3.2"),
|
||||
("Hanzo IDE", "~/work/hanzo/ide", "hanzo-ide edit"),
|
||||
]
|
||||
|
||||
console.print("Checking available CLI tools:\n")
|
||||
|
||||
for name, cmd, example in tools:
|
||||
# Check if tool exists
|
||||
if cmd.startswith("~"):
|
||||
exists = Path(cmd.expanduser()).exists()
|
||||
else:
|
||||
import shutil
|
||||
exists = shutil.which(cmd) is not None
|
||||
|
||||
status = "✅" if exists else "❌"
|
||||
color = "green" if exists else "red"
|
||||
|
||||
console.print(f"{status} [{color}]{name}[/{color}]")
|
||||
if exists:
|
||||
console.print(f" Example: [dim]{example}[/dim]")
|
||||
|
||||
async def demo_system2_thinking():
|
||||
"""Demo System 2 thinking with critic agents."""
|
||||
console.print("\n[bold cyan]═══ DEMO: System 2 Thinking ═══[/bold cyan]\n")
|
||||
|
||||
console.print("System 2 thinking provides deeper analysis:\n")
|
||||
|
||||
# Simulate a code review scenario
|
||||
code = """def process_data(data):
|
||||
result = []
|
||||
for item in data:
|
||||
if item > 0:
|
||||
result.append(item * 2)
|
||||
return result"""
|
||||
|
||||
console.print("[bold]Original Code:[/bold]")
|
||||
console.print(Panel(code, border_style="dim"))
|
||||
|
||||
console.print("\n[bold]Critic Agent Analysis:[/bold]")
|
||||
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
console=console,
|
||||
) as progress:
|
||||
task = progress.add_task("Analyzing code...", total=None)
|
||||
await asyncio.sleep(2)
|
||||
progress.update(task, description="✓ Analysis complete")
|
||||
|
||||
critique = """🔍 Code Review:
|
||||
|
||||
✅ Strengths:
|
||||
• Clear function name
|
||||
• Simple logic
|
||||
• Returns expected type
|
||||
|
||||
⚠️ Improvements:
|
||||
• Add type hints for clarity
|
||||
• Consider list comprehension for performance
|
||||
• Add docstring for documentation
|
||||
• Handle None/empty inputs
|
||||
|
||||
📝 Suggested Refactor:
|
||||
```python
|
||||
def process_data(data: list[float]) -> list[float]:
|
||||
\"\"\"Process positive numbers by doubling them.\"\"\"
|
||||
if not data:
|
||||
return []
|
||||
return [item * 2 for item in data if item > 0]
|
||||
```"""
|
||||
|
||||
console.print(Panel(critique, title="[bold cyan]Critic Analysis[/bold cyan]",
|
||||
title_align="left", border_style="dim cyan"))
|
||||
|
||||
async def demo_cost_optimization():
|
||||
"""Demo cost-optimized routing."""
|
||||
console.print("\n[bold cyan]═══ DEMO: Cost Optimization ═══[/bold cyan]\n")
|
||||
|
||||
console.print("Hanzo Dev optimizes costs by routing intelligently:\n")
|
||||
|
||||
# Create routing table
|
||||
table = Table(title="Intelligent Task Routing", show_header=True,
|
||||
header_style="bold magenta")
|
||||
table.add_column("Task Type", style="cyan", width=25)
|
||||
table.add_column("Routed To", width=20)
|
||||
table.add_column("Cost", style="green")
|
||||
table.add_column("Reason", width=35)
|
||||
|
||||
routes = [
|
||||
("Simple syntax fix", "Local Llama 3.2", "$0.00", "Basic pattern matching"),
|
||||
("Complex refactoring", "GPT-4", "$0.03", "Requires deep understanding"),
|
||||
("Code review", "Claude 3.5", "$0.02", "Best for analysis"),
|
||||
("Boilerplate generation", "Codestral Free", "$0.00", "Template-based"),
|
||||
("Bug investigation", "GPT-5 Pro", "$0.05", "Critical thinking needed"),
|
||||
("Documentation", "Local Model", "$0.00", "Structured output"),
|
||||
]
|
||||
|
||||
for task, model, cost, reason in routes:
|
||||
table.add_row(task, model, cost, reason)
|
||||
|
||||
console.print(table)
|
||||
|
||||
console.print("\n💰 [bold green]Average cost reduction: 90%[/bold green]")
|
||||
console.print(" Using local orchestration for routing decisions")
|
||||
|
||||
async def demo_mcp_integration():
|
||||
"""Demo MCP tool integration."""
|
||||
console.print("\n[bold cyan]═══ DEMO: MCP Tool Integration ═══[/bold cyan]\n")
|
||||
|
||||
console.print("All agents have access to MCP tools:\n")
|
||||
|
||||
tools = [
|
||||
("📁 File Operations", "Read, write, edit files"),
|
||||
("🔍 Code Search", "AST-aware search, grep, symbols"),
|
||||
("🌐 Web Fetch", "Browse documentation and APIs"),
|
||||
("💻 Command Execution", "Run tests, build, deploy"),
|
||||
("📊 Data Analysis", "Process CSV, JSON, databases"),
|
||||
("🔄 Git Operations", "Commit, branch, merge"),
|
||||
("📝 Documentation", "Generate docs, comments"),
|
||||
("🧪 Testing", "Run tests, generate test cases"),
|
||||
]
|
||||
|
||||
for tool, desc in tools:
|
||||
console.print(f"{tool}: [dim]{desc}[/dim]")
|
||||
|
||||
console.print("\n[bold]Recursive Agent Calls:[/bold]")
|
||||
console.print("• Agents can delegate tasks to each other")
|
||||
console.print("• chat_with_<agent> for conversations")
|
||||
console.print("• ask_<agent> for questions")
|
||||
console.print("• delegate_to_<agent> for task handoff")
|
||||
|
||||
async def demo_memory_management():
|
||||
"""Demo memory and context management."""
|
||||
console.print("\n[bold cyan]═══ DEMO: Memory Management ═══[/bold cyan]\n")
|
||||
|
||||
console.print("Memory commands (like Claude Desktop):\n")
|
||||
|
||||
commands = [
|
||||
("#memory", "Show current memory/context"),
|
||||
("#memory clear", "Clear all memory"),
|
||||
("#memory save", "Save current context"),
|
||||
("#memory load", "Load saved context"),
|
||||
("#memory add <text>", "Add to permanent context"),
|
||||
("#memory remove <id>", "Remove from context"),
|
||||
]
|
||||
|
||||
for cmd, desc in commands:
|
||||
console.print(f"[cyan]{cmd:20}[/cyan] {desc}")
|
||||
|
||||
console.print("\n[bold]Context Window Management:[/bold]")
|
||||
console.print("• Automatic summarization for long conversations")
|
||||
console.print("• Intelligent context pruning")
|
||||
console.print("• Project-specific memory persistence")
|
||||
|
||||
async def demo_interactive_session():
|
||||
"""Demo a real interactive session."""
|
||||
console.print("\n[bold cyan]═══ DEMO: Interactive Session ═══[/bold cyan]\n")
|
||||
|
||||
# Check for API keys
|
||||
has_openai = bool(os.getenv("OPENAI_API_KEY"))
|
||||
has_anthropic = bool(os.getenv("ANTHROPIC_API_KEY"))
|
||||
|
||||
if not (has_openai or has_anthropic):
|
||||
console.print("[yellow]⚠️ No API keys found. Demo will use mock responses.[/yellow]")
|
||||
console.print("Set OPENAI_API_KEY or ANTHROPIC_API_KEY for real AI responses.\n")
|
||||
return
|
||||
|
||||
console.print("[green]✅ API keys found. Starting interactive demo...[/green]\n")
|
||||
|
||||
# Create orchestrator
|
||||
orchestrator_model = "gpt-4" if has_openai else "claude"
|
||||
|
||||
try:
|
||||
orchestrator = MultiClaudeOrchestrator(
|
||||
workspace_dir="/tmp/hanzo_demo",
|
||||
claude_path="claude",
|
||||
num_instances=2,
|
||||
enable_mcp=True,
|
||||
enable_networking=True,
|
||||
enable_guardrails=True,
|
||||
console=console,
|
||||
orchestrator_model=orchestrator_model
|
||||
)
|
||||
|
||||
repl = HanzoDevREPL(orchestrator)
|
||||
|
||||
console.print(f"[bold]Using {orchestrator_model} orchestrator[/bold]\n")
|
||||
|
||||
# Demo messages
|
||||
test_messages = [
|
||||
"What is 2 + 2?",
|
||||
"Write a hello world in Python",
|
||||
"/help",
|
||||
]
|
||||
|
||||
for msg in test_messages:
|
||||
console.print(f"\n[dim white]› {msg}[/dim white]")
|
||||
|
||||
if msg.startswith("/"):
|
||||
# Handle command
|
||||
if msg == "/help":
|
||||
console.print("\nAvailable commands:")
|
||||
console.print(" /help - Show this help")
|
||||
console.print(" /status - Show system status")
|
||||
console.print(" /clear - Clear screen")
|
||||
console.print(" /exit - Exit REPL")
|
||||
else:
|
||||
# Simulate AI response
|
||||
await repl.chat_with_agents(msg)
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error in demo: {e}[/red]")
|
||||
|
||||
async def main():
|
||||
"""Run the complete demo."""
|
||||
console.print(Panel.fit(
|
||||
"[bold]🚀 Hanzo Dev - AI Coding OS Demo[/bold]\n"
|
||||
"Chat-first REPL with Multi-Orchestrator Support",
|
||||
border_style="bold cyan"
|
||||
))
|
||||
|
||||
demos = [
|
||||
("Chat-First Interface", demo_chat_interface),
|
||||
("Orchestrator Modes", demo_orchestrator_modes),
|
||||
("CLI Tool Integration", demo_cli_tools),
|
||||
("System 2 Thinking", demo_system2_thinking),
|
||||
("Cost Optimization", demo_cost_optimization),
|
||||
("MCP Integration", demo_mcp_integration),
|
||||
("Memory Management", demo_memory_management),
|
||||
("Interactive Session", demo_interactive_session),
|
||||
]
|
||||
|
||||
for i, (name, demo_func) in enumerate(demos, 1):
|
||||
if i > 1:
|
||||
if not Prompt.ask(f"\n[cyan]Continue to {name}?[/cyan]",
|
||||
default="y", choices=["y", "n"]) == "y":
|
||||
break
|
||||
|
||||
await demo_func()
|
||||
|
||||
console.print("\n" + "="*60)
|
||||
console.print(Panel.fit(
|
||||
"[bold green]✅ Demo Complete![/bold green]\n\n"
|
||||
"To start using Hanzo Dev:\n"
|
||||
"[cyan]pip install hanzo==0.3.19[/cyan]\n"
|
||||
"[cyan]hanzo dev[/cyan]\n\n"
|
||||
"For more info: [link]https://hanzo.ai[/link]",
|
||||
border_style="green"
|
||||
))
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
console.print("\n[yellow]Demo interrupted by user[/yellow]")
|
||||
except Exception as e:
|
||||
console.print(f"\n[red]Demo error: {e}[/red]")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "hanzo"
|
||||
version = "0.3.19"
|
||||
version = "0.3.20"
|
||||
description = "Hanzo AI - Complete AI Infrastructure Platform with CLI, Router, MCP, and Agent Runtime"
|
||||
authors = [
|
||||
{name = "Hanzo AI", email = "[email protected]"},
|
||||
|
||||
@@ -26,7 +26,7 @@ from .utils.output import console
|
||||
from .interactive.repl import HanzoREPL
|
||||
|
||||
# Version
|
||||
__version__ = "0.3.19"
|
||||
__version__ = "0.3.20"
|
||||
|
||||
|
||||
@click.group(invoke_without_command=True)
|
||||
|
||||
@@ -661,6 +661,11 @@ class HanzoDevREPL:
|
||||
"help": self.cmd_help,
|
||||
"exit": self.cmd_exit,
|
||||
}
|
||||
|
||||
# Initialize memory manager
|
||||
from .memory_manager import MemoryManager
|
||||
workspace = getattr(orchestrator, 'workspace_dir', '/tmp/hanzo')
|
||||
self.memory_manager = MemoryManager(workspace)
|
||||
|
||||
async def run(self):
|
||||
"""Run the REPL."""
|
||||
@@ -741,7 +746,10 @@ class HanzoDevREPL:
|
||||
|
||||
elif user_input.startswith("#"):
|
||||
# Handle memory/context commands
|
||||
await self.handle_memory_command(user_input[1:].strip())
|
||||
from .memory_manager import handle_memory_command
|
||||
handled = handle_memory_command(user_input, self.memory_manager, console)
|
||||
if not handled:
|
||||
console.print("[yellow]Unknown memory command. Use #memory help[/yellow]")
|
||||
|
||||
else:
|
||||
# Natural chat - send directly to AI agents
|
||||
@@ -906,6 +914,41 @@ Examples:
|
||||
async def chat_with_agents(self, message: str):
|
||||
"""Send message to AI agents for natural chat."""
|
||||
try:
|
||||
# Add message to memory
|
||||
self.memory_manager.add_message("user", message)
|
||||
|
||||
# Get memory context
|
||||
memory_context = self.memory_manager.summarize_for_ai()
|
||||
|
||||
# Enhance message with context
|
||||
if memory_context:
|
||||
enhanced_message = f"{memory_context}\n\nUser: {message}"
|
||||
else:
|
||||
enhanced_message = message
|
||||
|
||||
# Try smart fallback if no specific model configured
|
||||
if not hasattr(self.orchestrator, 'orchestrator_model') or \
|
||||
self.orchestrator.orchestrator_model == "auto":
|
||||
from .fallback_handler import smart_chat
|
||||
response = await smart_chat(enhanced_message, console)
|
||||
if response:
|
||||
# Save AI response to memory
|
||||
self.memory_manager.add_message("assistant", response)
|
||||
|
||||
from rich.panel import Panel
|
||||
console.print()
|
||||
console.print(Panel(
|
||||
response,
|
||||
title="[bold cyan]AI Response[/bold cyan]",
|
||||
title_align="left",
|
||||
border_style="dim cyan",
|
||||
padding=(1, 2)
|
||||
))
|
||||
return
|
||||
else:
|
||||
console.print("[red]No AI options available. Please configure API keys or install tools.[/red]")
|
||||
return
|
||||
|
||||
# For codex and other CLI tools, go straight to direct API chat
|
||||
if hasattr(self.orchestrator, 'orchestrator_model'):
|
||||
model = self.orchestrator.orchestrator_model
|
||||
|
||||
@@ -0,0 +1,244 @@
|
||||
"""
|
||||
Intelligent fallback handler for Hanzo Dev.
|
||||
Automatically tries available AI options when primary fails.
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import Optional, Dict, Any
|
||||
from pathlib import Path
|
||||
|
||||
class FallbackHandler:
|
||||
"""Handles automatic fallback to available AI options."""
|
||||
|
||||
def __init__(self):
|
||||
self.available_options = self._detect_available_options()
|
||||
self.fallback_order = self._determine_fallback_order()
|
||||
|
||||
def _detect_available_options(self) -> Dict[str, bool]:
|
||||
"""Detect which AI options are available."""
|
||||
options = {
|
||||
"openai_api": bool(os.getenv("OPENAI_API_KEY")),
|
||||
"anthropic_api": bool(os.getenv("ANTHROPIC_API_KEY")),
|
||||
"google_api": bool(os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY")),
|
||||
"openai_cli": shutil.which("openai") is not None,
|
||||
"claude_cli": shutil.which("claude") is not None,
|
||||
"gemini_cli": shutil.which("gemini") is not None,
|
||||
"ollama": self._check_ollama(),
|
||||
"hanzo_ide": Path.home().joinpath("work/hanzo/ide").exists(),
|
||||
"free_apis": True, # Always available (Codestral, StarCoder)
|
||||
}
|
||||
return options
|
||||
|
||||
def _check_ollama(self) -> bool:
|
||||
"""Check if Ollama is running and has models."""
|
||||
try:
|
||||
import httpx
|
||||
with httpx.Client(timeout=2.0) as client:
|
||||
response = client.get("http://localhost:11434/api/tags")
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
return len(data.get("models", [])) > 0
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
def _determine_fallback_order(self) -> list:
|
||||
"""Determine the order of fallback options based on availability."""
|
||||
order = []
|
||||
|
||||
# Priority 1: API keys (fastest, most reliable)
|
||||
if self.available_options["openai_api"]:
|
||||
order.append(("openai_api", "gpt-4"))
|
||||
if self.available_options["anthropic_api"]:
|
||||
order.append(("anthropic_api", "claude-3-5-sonnet"))
|
||||
if self.available_options["google_api"]:
|
||||
order.append(("google_api", "gemini-pro"))
|
||||
|
||||
# Priority 2: CLI tools (no API key needed)
|
||||
if self.available_options["openai_cli"]:
|
||||
order.append(("openai_cli", "codex"))
|
||||
if self.available_options["claude_cli"]:
|
||||
order.append(("claude_cli", "claude-desktop"))
|
||||
if self.available_options["gemini_cli"]:
|
||||
order.append(("gemini_cli", "gemini"))
|
||||
|
||||
# Priority 3: Local models (free, but requires setup)
|
||||
if self.available_options["ollama"]:
|
||||
order.append(("ollama", "local:llama3.2"))
|
||||
if self.available_options["hanzo_ide"]:
|
||||
order.append(("hanzo_ide", "hanzo-ide"))
|
||||
|
||||
# Priority 4: Free cloud APIs (rate limited)
|
||||
if self.available_options["free_apis"]:
|
||||
order.append(("free_api", "codestral-free"))
|
||||
order.append(("free_api", "starcoder2"))
|
||||
|
||||
return order
|
||||
|
||||
def get_best_option(self) -> Optional[tuple]:
|
||||
"""Get the best available AI option."""
|
||||
if self.fallback_order:
|
||||
return self.fallback_order[0]
|
||||
return None
|
||||
|
||||
def get_next_option(self, failed_option: str) -> Optional[tuple]:
|
||||
"""Get the next fallback option after one fails."""
|
||||
for i, (option_type, model) in enumerate(self.fallback_order):
|
||||
if model == failed_option and i + 1 < len(self.fallback_order):
|
||||
return self.fallback_order[i + 1]
|
||||
return None
|
||||
|
||||
def suggest_setup(self) -> str:
|
||||
"""Suggest setup instructions for unavailable options."""
|
||||
suggestions = []
|
||||
|
||||
if not self.available_options["openai_api"]:
|
||||
suggestions.append("• Set OPENAI_API_KEY for GPT-4/GPT-5 access")
|
||||
|
||||
if not self.available_options["anthropic_api"]:
|
||||
suggestions.append("• Set ANTHROPIC_API_KEY for Claude access")
|
||||
|
||||
if not self.available_options["ollama"]:
|
||||
suggestions.append("• Install Ollama: curl -fsSL https://ollama.com/install.sh | sh")
|
||||
suggestions.append(" Then run: ollama pull llama3.2")
|
||||
|
||||
if not self.available_options["openai_cli"]:
|
||||
suggestions.append("• Install OpenAI CLI: pip install openai-cli")
|
||||
|
||||
if not self.available_options["claude_cli"]:
|
||||
suggestions.append("• Install Claude Desktop from https://claude.ai/download")
|
||||
|
||||
return "\n".join(suggestions) if suggestions else "All AI options are available!"
|
||||
|
||||
def print_status(self, console):
|
||||
"""Print the current status of available AI options."""
|
||||
from rich.table import Table
|
||||
|
||||
table = Table(title="Available AI Options", show_header=True,
|
||||
header_style="bold magenta")
|
||||
table.add_column("Option", style="cyan", width=20)
|
||||
table.add_column("Status", width=10)
|
||||
table.add_column("Model", width=20)
|
||||
|
||||
status_map = {
|
||||
"openai_api": ("OpenAI API", "gpt-4"),
|
||||
"anthropic_api": ("Anthropic API", "claude-3-5"),
|
||||
"google_api": ("Google API", "gemini-pro"),
|
||||
"openai_cli": ("OpenAI CLI", "codex"),
|
||||
"claude_cli": ("Claude Desktop", "claude"),
|
||||
"gemini_cli": ("Gemini CLI", "gemini"),
|
||||
"ollama": ("Ollama Local", "llama3.2"),
|
||||
"hanzo_ide": ("Hanzo IDE", "hanzo-dev"),
|
||||
"free_apis": ("Free APIs", "codestral/starcoder"),
|
||||
}
|
||||
|
||||
for key, available in self.available_options.items():
|
||||
if key in status_map:
|
||||
name, model = status_map[key]
|
||||
status = "✅" if available else "❌"
|
||||
table.add_row(name, status, model if available else "Not available")
|
||||
|
||||
console.print(table)
|
||||
|
||||
if self.fallback_order:
|
||||
console.print(f"\n[green]Primary option: {self.fallback_order[0][1]}[/green]")
|
||||
if len(self.fallback_order) > 1:
|
||||
fallbacks = ", ".join([opt[1] for opt in self.fallback_order[1:]])
|
||||
console.print(f"[yellow]Fallback options: {fallbacks}[/yellow]")
|
||||
else:
|
||||
console.print("\n[red]No AI options available![/red]")
|
||||
console.print("\n[yellow]Setup suggestions:[/yellow]")
|
||||
console.print(self.suggest_setup())
|
||||
|
||||
|
||||
async def smart_chat(message: str, console=None) -> Optional[str]:
|
||||
"""
|
||||
Smart chat that automatically tries available AI options.
|
||||
Returns the AI response or None if all options fail.
|
||||
"""
|
||||
handler = FallbackHandler()
|
||||
|
||||
if console:
|
||||
console.print("\n[dim]Detecting available AI options...[/dim]")
|
||||
|
||||
best_option = handler.get_best_option()
|
||||
if not best_option:
|
||||
if console:
|
||||
handler.print_status(console)
|
||||
return None
|
||||
|
||||
option_type, model = best_option
|
||||
|
||||
# Try the primary option
|
||||
try:
|
||||
if option_type == "openai_api":
|
||||
from openai import AsyncOpenAI
|
||||
client = AsyncOpenAI()
|
||||
response = await client.chat.completions.create(
|
||||
model="gpt-4",
|
||||
messages=[{"role": "user", "content": message}],
|
||||
max_tokens=500
|
||||
)
|
||||
return response.choices[0].message.content
|
||||
|
||||
elif option_type == "anthropic_api":
|
||||
from anthropic import AsyncAnthropic
|
||||
client = AsyncAnthropic()
|
||||
response = await client.messages.create(
|
||||
model="claude-3-5-sonnet-20241022",
|
||||
messages=[{"role": "user", "content": message}],
|
||||
max_tokens=500
|
||||
)
|
||||
return response.content[0].text
|
||||
|
||||
elif option_type == "openai_cli":
|
||||
# Use OpenAI CLI
|
||||
result = subprocess.run(
|
||||
["openai", "api", "chat.completions.create", "-m", "gpt-4", "-g", message],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return result.stdout.strip()
|
||||
|
||||
elif option_type == "ollama":
|
||||
# Use Ollama
|
||||
import httpx
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
"http://localhost:11434/api/generate",
|
||||
json={"model": "llama3.2", "prompt": message, "stream": False},
|
||||
timeout=30.0
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json().get("response", "")
|
||||
|
||||
elif option_type == "free_api":
|
||||
# Try free Codestral API
|
||||
import httpx
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
"https://codestral.mistral.ai/v1/fim/completions",
|
||||
headers={"Content-Type": "application/json"},
|
||||
json={"prompt": message, "suffix": "", "max_tokens": 500},
|
||||
timeout=30.0
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json().get("choices", [{}])[0].get("text", "")
|
||||
|
||||
except Exception as e:
|
||||
if console:
|
||||
console.print(f"[yellow]Primary option {model} failed: {e}[/yellow]")
|
||||
console.print("[dim]Trying fallback...[/dim]")
|
||||
|
||||
# Try next fallback
|
||||
next_option = handler.get_next_option(model)
|
||||
if next_option:
|
||||
# Recursively try the next option
|
||||
handler.fallback_order.remove(best_option)
|
||||
return await smart_chat(message, console)
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,425 @@
|
||||
"""
|
||||
Memory management system for Hanzo Dev.
|
||||
Provides persistent context and memory like Claude Desktop.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Any, Optional
|
||||
from datetime import datetime
|
||||
from dataclasses import dataclass, asdict
|
||||
import hashlib
|
||||
|
||||
@dataclass
|
||||
class MemoryItem:
|
||||
"""A single memory item."""
|
||||
id: str
|
||||
content: str
|
||||
type: str # 'context', 'instruction', 'fact', 'code'
|
||||
created_at: str
|
||||
tags: List[str]
|
||||
priority: int = 0 # Higher priority items are kept longer
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return asdict(self)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict) -> 'MemoryItem':
|
||||
return cls(**data)
|
||||
|
||||
|
||||
class MemoryManager:
|
||||
"""Manages persistent memory and context for AI conversations."""
|
||||
|
||||
def __init__(self, workspace_dir: str = None):
|
||||
"""Initialize memory manager."""
|
||||
if workspace_dir:
|
||||
self.memory_dir = Path(workspace_dir) / ".hanzo" / "memory"
|
||||
else:
|
||||
self.memory_dir = Path.home() / ".hanzo" / "memory"
|
||||
|
||||
self.memory_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.memory_file = self.memory_dir / "context.json"
|
||||
self.session_file = self.memory_dir / "session.json"
|
||||
|
||||
self.memories: List[MemoryItem] = []
|
||||
self.session_context: Dict[str, Any] = {}
|
||||
|
||||
self.load_memories()
|
||||
self.load_session()
|
||||
|
||||
def load_memories(self):
|
||||
"""Load persistent memories from disk."""
|
||||
if self.memory_file.exists():
|
||||
try:
|
||||
with open(self.memory_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
self.memories = [MemoryItem.from_dict(item) for item in data.get('memories', [])]
|
||||
except Exception as e:
|
||||
print(f"Error loading memories: {e}")
|
||||
self.memories = []
|
||||
else:
|
||||
# Initialize with default memories
|
||||
self._init_default_memories()
|
||||
|
||||
def _init_default_memories(self):
|
||||
"""Initialize with helpful default memories."""
|
||||
defaults = [
|
||||
MemoryItem(
|
||||
id=self._generate_id("system"),
|
||||
content="I am Hanzo Dev, an AI coding assistant with multiple orchestrator modes.",
|
||||
type="instruction",
|
||||
created_at=datetime.now().isoformat(),
|
||||
tags=["system", "identity"],
|
||||
priority=10
|
||||
),
|
||||
MemoryItem(
|
||||
id=self._generate_id("capabilities"),
|
||||
content="I can read/write files, search code, run commands, and use various AI models.",
|
||||
type="fact",
|
||||
created_at=datetime.now().isoformat(),
|
||||
tags=["system", "capabilities"],
|
||||
priority=9
|
||||
),
|
||||
MemoryItem(
|
||||
id=self._generate_id("help"),
|
||||
content="Use /help for commands, #memory for context management, or just chat naturally.",
|
||||
type="instruction",
|
||||
created_at=datetime.now().isoformat(),
|
||||
tags=["system", "usage"],
|
||||
priority=8
|
||||
),
|
||||
]
|
||||
self.memories = defaults
|
||||
self.save_memories()
|
||||
|
||||
def save_memories(self):
|
||||
"""Save memories to disk."""
|
||||
try:
|
||||
data = {
|
||||
'memories': [m.to_dict() for m in self.memories],
|
||||
'updated_at': datetime.now().isoformat()
|
||||
}
|
||||
with open(self.memory_file, 'w') as f:
|
||||
json.dump(data, f, indent=2)
|
||||
except Exception as e:
|
||||
print(f"Error saving memories: {e}")
|
||||
|
||||
def load_session(self):
|
||||
"""Load current session context."""
|
||||
if self.session_file.exists():
|
||||
try:
|
||||
with open(self.session_file, 'r') as f:
|
||||
self.session_context = json.load(f)
|
||||
except:
|
||||
self.session_context = {}
|
||||
else:
|
||||
self.session_context = {
|
||||
'started_at': datetime.now().isoformat(),
|
||||
'messages': [],
|
||||
'current_task': None,
|
||||
'preferences': {}
|
||||
}
|
||||
|
||||
def save_session(self):
|
||||
"""Save session context."""
|
||||
try:
|
||||
with open(self.session_file, 'w') as f:
|
||||
json.dump(self.session_context, f, indent=2)
|
||||
except Exception as e:
|
||||
print(f"Error saving session: {e}")
|
||||
|
||||
def add_memory(self, content: str, type: str = "context", tags: List[str] = None, priority: int = 0) -> str:
|
||||
"""Add a new memory item."""
|
||||
memory_id = self._generate_id(content)
|
||||
|
||||
# Check if similar memory exists
|
||||
for mem in self.memories:
|
||||
if mem.content == content:
|
||||
return mem.id # Don't duplicate
|
||||
|
||||
memory = MemoryItem(
|
||||
id=memory_id,
|
||||
content=content,
|
||||
type=type,
|
||||
created_at=datetime.now().isoformat(),
|
||||
tags=tags or [],
|
||||
priority=priority
|
||||
)
|
||||
|
||||
self.memories.append(memory)
|
||||
self.save_memories()
|
||||
|
||||
return memory_id
|
||||
|
||||
def remove_memory(self, memory_id: str) -> bool:
|
||||
"""Remove a memory by ID."""
|
||||
for i, mem in enumerate(self.memories):
|
||||
if mem.id == memory_id:
|
||||
del self.memories[i]
|
||||
self.save_memories()
|
||||
return True
|
||||
return False
|
||||
|
||||
def clear_memories(self, keep_system: bool = True):
|
||||
"""Clear all memories, optionally keeping system memories."""
|
||||
if keep_system:
|
||||
self.memories = [m for m in self.memories if "system" in m.tags]
|
||||
else:
|
||||
self.memories = []
|
||||
self.save_memories()
|
||||
|
||||
def get_memories(self, type: str = None, tags: List[str] = None) -> List[MemoryItem]:
|
||||
"""Get memories filtered by type or tags."""
|
||||
result = self.memories
|
||||
|
||||
if type:
|
||||
result = [m for m in result if m.type == type]
|
||||
|
||||
if tags:
|
||||
result = [m for m in result if any(tag in m.tags for tag in tags)]
|
||||
|
||||
# Sort by priority and creation date
|
||||
result.sort(key=lambda m: (-m.priority, m.created_at), reverse=True)
|
||||
|
||||
return result
|
||||
|
||||
def get_context_string(self, max_tokens: int = 2000) -> str:
|
||||
"""Get a formatted context string for AI prompts."""
|
||||
# Sort memories by priority
|
||||
sorted_memories = sorted(self.memories, key=lambda m: -m.priority)
|
||||
|
||||
context_parts = []
|
||||
token_count = 0
|
||||
|
||||
for memory in sorted_memories:
|
||||
# Rough token estimation (4 chars = 1 token)
|
||||
memory_tokens = len(memory.content) // 4
|
||||
|
||||
if token_count + memory_tokens > max_tokens:
|
||||
break
|
||||
|
||||
if memory.type == "instruction":
|
||||
context_parts.append(f"INSTRUCTION: {memory.content}")
|
||||
elif memory.type == "fact":
|
||||
context_parts.append(f"FACT: {memory.content}")
|
||||
elif memory.type == "code":
|
||||
context_parts.append(f"CODE CONTEXT:\n{memory.content}")
|
||||
else:
|
||||
context_parts.append(memory.content)
|
||||
|
||||
token_count += memory_tokens
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
def add_message(self, role: str, content: str):
|
||||
"""Add a message to session history."""
|
||||
self.session_context['messages'].append({
|
||||
'role': role,
|
||||
'content': content,
|
||||
'timestamp': datetime.now().isoformat()
|
||||
})
|
||||
|
||||
# Keep only last 50 messages
|
||||
if len(self.session_context['messages']) > 50:
|
||||
self.session_context['messages'] = self.session_context['messages'][-50:]
|
||||
|
||||
self.save_session()
|
||||
|
||||
def get_recent_messages(self, count: int = 10) -> List[Dict]:
|
||||
"""Get recent messages from session."""
|
||||
return self.session_context['messages'][-count:]
|
||||
|
||||
def set_preference(self, key: str, value: Any):
|
||||
"""Set a user preference."""
|
||||
self.session_context['preferences'][key] = value
|
||||
self.save_session()
|
||||
|
||||
def get_preference(self, key: str, default: Any = None) -> Any:
|
||||
"""Get a user preference."""
|
||||
return self.session_context['preferences'].get(key, default)
|
||||
|
||||
def _generate_id(self, content: str) -> str:
|
||||
"""Generate a unique ID for a memory item."""
|
||||
hash_input = f"{content}{datetime.now().isoformat()}"
|
||||
return hashlib.md5(hash_input.encode()).hexdigest()[:8]
|
||||
|
||||
def summarize_for_ai(self) -> str:
|
||||
"""Create a summary suitable for AI context."""
|
||||
summary = []
|
||||
|
||||
# Add system memories
|
||||
system_memories = self.get_memories(tags=["system"])
|
||||
if system_memories:
|
||||
summary.append("SYSTEM CONTEXT:")
|
||||
for mem in system_memories[:3]: # Top 3 system memories
|
||||
summary.append(f"- {mem.content}")
|
||||
|
||||
# Add recent instructions
|
||||
instructions = self.get_memories(type="instruction")
|
||||
if instructions:
|
||||
summary.append("\nINSTRUCTIONS:")
|
||||
for mem in instructions[:3]: # Top 3 instructions
|
||||
summary.append(f"- {mem.content}")
|
||||
|
||||
# Add important facts
|
||||
facts = self.get_memories(type="fact")
|
||||
if facts:
|
||||
summary.append("\nKEY FACTS:")
|
||||
for mem in facts[:5]: # Top 5 facts
|
||||
summary.append(f"- {mem.content}")
|
||||
|
||||
# Add current task if set
|
||||
if self.session_context.get('current_task'):
|
||||
summary.append(f"\nCURRENT TASK: {self.session_context['current_task']}")
|
||||
|
||||
return "\n".join(summary)
|
||||
|
||||
def export_memories(self, file_path: str):
|
||||
"""Export memories to a file."""
|
||||
data = {
|
||||
'memories': [m.to_dict() for m in self.memories],
|
||||
'session': self.session_context,
|
||||
'exported_at': datetime.now().isoformat()
|
||||
}
|
||||
|
||||
with open(file_path, 'w') as f:
|
||||
json.dump(data, f, indent=2)
|
||||
|
||||
def import_memories(self, file_path: str):
|
||||
"""Import memories from a file."""
|
||||
with open(file_path, 'r') as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Merge memories (avoid duplicates)
|
||||
existing_ids = {m.id for m in self.memories}
|
||||
|
||||
for mem_data in data.get('memories', []):
|
||||
if mem_data['id'] not in existing_ids:
|
||||
self.memories.append(MemoryItem.from_dict(mem_data))
|
||||
|
||||
# Merge session preferences
|
||||
if 'session' in data and 'preferences' in data['session']:
|
||||
self.session_context['preferences'].update(data['session']['preferences'])
|
||||
|
||||
self.save_memories()
|
||||
self.save_session()
|
||||
|
||||
|
||||
def handle_memory_command(command: str, memory_manager: MemoryManager, console) -> bool:
|
||||
"""
|
||||
Handle #memory commands.
|
||||
Returns True if command was handled, False otherwise.
|
||||
"""
|
||||
from rich.table import Table
|
||||
from rich.panel import Panel
|
||||
|
||||
parts = command.strip().split(maxsplit=2)
|
||||
|
||||
if len(parts) == 1 or parts[1] == "show":
|
||||
# Show current memories
|
||||
memories = memory_manager.get_memories()
|
||||
|
||||
if not memories:
|
||||
console.print("[yellow]No memories stored.[/yellow]")
|
||||
return True
|
||||
|
||||
table = Table(title="Current Memories", show_header=True,
|
||||
header_style="bold magenta")
|
||||
table.add_column("ID", style="cyan", width=10)
|
||||
table.add_column("Type", width=12)
|
||||
table.add_column("Content", width=50)
|
||||
table.add_column("Priority", width=8)
|
||||
|
||||
for mem in memories[:10]: # Show top 10
|
||||
content = mem.content[:47] + "..." if len(mem.content) > 50 else mem.content
|
||||
table.add_row(mem.id, mem.type, content, str(mem.priority))
|
||||
|
||||
console.print(table)
|
||||
|
||||
if len(memories) > 10:
|
||||
console.print(f"[dim]... and {len(memories) - 10} more[/dim]")
|
||||
|
||||
return True
|
||||
|
||||
elif parts[1] == "add":
|
||||
if len(parts) < 3:
|
||||
console.print("[red]Usage: #memory add <content>[/red]")
|
||||
return True
|
||||
|
||||
content = parts[2]
|
||||
memory_id = memory_manager.add_memory(content, type="context")
|
||||
console.print(f"[green]Added memory: {memory_id}[/green]")
|
||||
return True
|
||||
|
||||
elif parts[1] == "remove":
|
||||
if len(parts) < 3:
|
||||
console.print("[red]Usage: #memory remove <id>[/red]")
|
||||
return True
|
||||
|
||||
memory_id = parts[2]
|
||||
if memory_manager.remove_memory(memory_id):
|
||||
console.print(f"[green]Removed memory: {memory_id}[/green]")
|
||||
else:
|
||||
console.print(f"[red]Memory not found: {memory_id}[/red]")
|
||||
return True
|
||||
|
||||
elif parts[1] == "clear":
|
||||
memory_manager.clear_memories(keep_system=True)
|
||||
console.print("[green]Cleared all non-system memories.[/green]")
|
||||
return True
|
||||
|
||||
elif parts[1] == "save":
|
||||
memory_manager.save_memories()
|
||||
memory_manager.save_session()
|
||||
console.print("[green]Memories saved.[/green]")
|
||||
return True
|
||||
|
||||
elif parts[1] == "export":
|
||||
if len(parts) < 3:
|
||||
file_path = "hanzo_memories.json"
|
||||
else:
|
||||
file_path = parts[2]
|
||||
|
||||
memory_manager.export_memories(file_path)
|
||||
console.print(f"[green]Exported memories to {file_path}[/green]")
|
||||
return True
|
||||
|
||||
elif parts[1] == "import":
|
||||
if len(parts) < 3:
|
||||
console.print("[red]Usage: #memory import <file_path>[/red]")
|
||||
return True
|
||||
|
||||
file_path = parts[2]
|
||||
try:
|
||||
memory_manager.import_memories(file_path)
|
||||
console.print(f"[green]Imported memories from {file_path}[/green]")
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error importing: {e}[/red]")
|
||||
return True
|
||||
|
||||
elif parts[1] == "context":
|
||||
# Show AI context
|
||||
context = memory_manager.summarize_for_ai()
|
||||
console.print(Panel(context, title="[bold cyan]AI Context[/bold cyan]",
|
||||
title_align="left", border_style="dim cyan"))
|
||||
return True
|
||||
|
||||
elif parts[1] == "help":
|
||||
help_text = """Memory Commands:
|
||||
#memory [show] - Show current memories
|
||||
#memory add <text> - Add new memory
|
||||
#memory remove <id> - Remove memory by ID
|
||||
#memory clear - Clear all memories (keep system)
|
||||
#memory save - Save memories to disk
|
||||
#memory export [file] - Export memories to file
|
||||
#memory import <file> - Import memories from file
|
||||
#memory context - Show AI context summary
|
||||
#memory help - Show this help"""
|
||||
|
||||
console.print(Panel(help_text, title="[bold cyan]Memory Help[/bold cyan]",
|
||||
title_align="left", border_style="dim cyan"))
|
||||
return True
|
||||
|
||||
return False
|
||||
@@ -0,0 +1,44 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test the fallback handler."""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from rich.console import Console
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent / "pkg" / "hanzo" / "src"))
|
||||
|
||||
from hanzo.fallback_handler import FallbackHandler, smart_chat
|
||||
|
||||
async def test_fallback():
|
||||
"""Test the fallback handler."""
|
||||
console = Console()
|
||||
|
||||
console.print("\n[bold cyan]Testing Fallback Handler[/bold cyan]\n")
|
||||
|
||||
# Create handler and check status
|
||||
handler = FallbackHandler()
|
||||
handler.print_status(console)
|
||||
|
||||
# Test smart chat
|
||||
console.print("\n[bold]Testing smart chat with automatic fallback:[/bold]")
|
||||
|
||||
test_message = "What is 2 + 2? Reply with just the number."
|
||||
console.print(f"\nMessage: [cyan]{test_message}[/cyan]")
|
||||
|
||||
response = await smart_chat(test_message, console)
|
||||
|
||||
if response:
|
||||
console.print(f"\n[green]Success! AI responded:[/green]")
|
||||
console.print(f"[bold]{response}[/bold]")
|
||||
else:
|
||||
console.print("\n[red]Failed to get response from any AI option[/red]")
|
||||
console.print("\n[yellow]Setup suggestions:[/yellow]")
|
||||
console.print(handler.suggest_setup())
|
||||
|
||||
return response is not None
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = asyncio.run(test_fallback())
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -0,0 +1,99 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test memory management system."""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from rich.console import Console
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent / "pkg" / "hanzo" / "src"))
|
||||
|
||||
from hanzo.memory_manager import MemoryManager, handle_memory_command
|
||||
|
||||
def test_memory_manager():
|
||||
"""Test memory manager functionality."""
|
||||
console = Console()
|
||||
|
||||
console.print("\n[bold cyan]Testing Memory Manager[/bold cyan]\n")
|
||||
|
||||
# Create memory manager
|
||||
memory_manager = MemoryManager("/tmp/test_memory")
|
||||
|
||||
# Test adding memories
|
||||
console.print("[bold]Adding memories:[/bold]")
|
||||
|
||||
id1 = memory_manager.add_memory("User prefers Python over JavaScript", type="fact", priority=5)
|
||||
console.print(f"✓ Added fact: {id1}")
|
||||
|
||||
id2 = memory_manager.add_memory("Always use type hints in Python code", type="instruction", priority=8)
|
||||
console.print(f"✓ Added instruction: {id2}")
|
||||
|
||||
id3 = memory_manager.add_memory("Working on REST API project", type="context", priority=3)
|
||||
console.print(f"✓ Added context: {id3}")
|
||||
|
||||
# Test retrieving memories
|
||||
console.print("\n[bold]Retrieving memories:[/bold]")
|
||||
|
||||
all_memories = memory_manager.get_memories()
|
||||
console.print(f"Total memories: {len(all_memories)}")
|
||||
|
||||
instructions = memory_manager.get_memories(type="instruction")
|
||||
console.print(f"Instructions: {len(instructions)}")
|
||||
|
||||
# Test memory commands
|
||||
console.print("\n[bold]Testing memory commands:[/bold]")
|
||||
|
||||
# Show command
|
||||
console.print("\n#memory show:")
|
||||
handle_memory_command("#memory show", memory_manager, console)
|
||||
|
||||
# Context command
|
||||
console.print("\n#memory context:")
|
||||
handle_memory_command("#memory context", memory_manager, console)
|
||||
|
||||
# Test preferences
|
||||
console.print("\n[bold]Testing preferences:[/bold]")
|
||||
|
||||
memory_manager.set_preference("theme", "dark")
|
||||
memory_manager.set_preference("language", "python")
|
||||
|
||||
theme = memory_manager.get_preference("theme")
|
||||
console.print(f"Theme preference: {theme}")
|
||||
|
||||
# Test session messages
|
||||
console.print("\n[bold]Testing session messages:[/bold]")
|
||||
|
||||
memory_manager.add_message("user", "Hello AI!")
|
||||
memory_manager.add_message("assistant", "Hello! How can I help you today?")
|
||||
|
||||
recent = memory_manager.get_recent_messages(2)
|
||||
console.print(f"Recent messages: {len(recent)}")
|
||||
|
||||
# Test AI summary
|
||||
console.print("\n[bold]AI Context Summary:[/bold]")
|
||||
summary = memory_manager.summarize_for_ai()
|
||||
console.print(summary)
|
||||
|
||||
# Test export/import
|
||||
console.print("\n[bold]Testing export/import:[/bold]")
|
||||
|
||||
export_file = "/tmp/test_memories.json"
|
||||
memory_manager.export_memories(export_file)
|
||||
console.print(f"✓ Exported to {export_file}")
|
||||
|
||||
# Create new manager and import
|
||||
new_manager = MemoryManager("/tmp/test_memory2")
|
||||
new_manager.import_memories(export_file)
|
||||
console.print(f"✓ Imported to new manager")
|
||||
|
||||
imported_memories = new_manager.get_memories()
|
||||
console.print(f"Imported memories: {len(imported_memories)}")
|
||||
|
||||
console.print("\n[green]✅ All memory tests passed![/green]")
|
||||
|
||||
return True
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = test_memory_manager()
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user