Files
langcode_agent/langcode.py
T
opencode f1d1248d4b fix: paginazione output con stile 'more'
- Output mostrato in blocchi da 15 righe
- Premi ENTER per continuare, 'q' per uscire
- SepARATOR visivo tra pagine
2026-03-24 15:37:38 +01:00

673 lines
24 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
LangCode Agent - Unified CLI Entry Point
Usage:
langcode → avvia l'agent (chat interattiva)
langcode chat → avvia l'agent
langcode server [--port N] → avvia web server
langcode tui → apre TUI configurazione
langcode config list → lista configurazioni
langcode config add → aggiungi configurazione
langcode config remove NAME → rimuovi configurazione
langcode config use NAME → attiva configurazione
langcode config active → mostra configurazione attiva
langcode config test [NAME] → testa connessione
langcode config export → esporta configurazioni
langcode config import FILE → importa configurazioni
langcode config search Q → cerca configurazioni
langcode analytics → mostra analytics
langcode dashboard → mostra dashboard performance
langcode version → mostra versione
"""
import sys
import os
import argparse
# Add current directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
VERSION = "1.0.0"
def strip_ansi(text: str) -> str:
"""Rimuove codici ANSI escape sequences per terminali semplici"""
import re
ansi_pattern = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
return ansi_pattern.sub('', text)
def cmd_chat(args):
"""Avvia l'agent in modalità chat interattiva"""
import signal
import sys
import threading
import time
from main import create_langcode_agent, process_command
from langchain_core.messages import HumanMessage
# Signal handler for Ctrl+C
def handle_sigint(signum, frame):
print("\n\n👋 Arrivederci!")
sys.exit(0)
# Register signal handler
signal.signal(signal.SIGINT, handle_sigint)
# Thread-safe flag for thinking indicator
thinking_active = threading.Event()
thinking_active.set()
def thinking_indicator():
"""Mostra indicatore di pensiero mentre aspetta"""
dots = [" ", ". ", ".. ", "..."]
idx = 0
while thinking_active.is_set():
print(f"\r🤖 Pensando{dots[idx % 4]} ", end="", flush=True)
idx += 1
time.sleep(0.5)
print("\r" + " " * 20 + "\r", end="") # Clear line
print("🚀 Inizializzo LangCode Agent...")
try:
agent = create_langcode_agent()
print("✅ Agent inizializzato con successo!")
print("\n🔧 Strumenti disponibili:")
print(" • Provider tools: list models, test connections, set defaults")
print(" • Proxmox tools: manage VMs, snapshots, resources")
print(" • Standard tools: planning, filesystem, shell, subagents")
print("\n💬 Puoi interagire con l'agent.")
print("📌 Comandi speciali (prefissati con /):")
print(" /help - Mostra i comandi disponibili")
print(" /status - Mostra lo stato del sistema")
print(" /config - Gestisci le configurazioni")
print(" /exit - Esci dall'agente")
print("\n💡 Per chattare, scrivi semplicemente il tuo messaggio.")
# Callback per processare i messaggi con l'agente
def agent_callback(user_input: str) -> str:
# Check if it's a command (starts with /)
if process_command(user_input):
return "Comando eseguito."
# Otherwise, treat it as a chat message
try:
result = agent.invoke({"messages": [HumanMessage(content=user_input)]})
# Extract response - support both simple agent (messages[-1]) and tool agent (response)
if "response" in result:
return result["response"]
else:
ai_message = result["messages"][-1]
return ai_message.content if hasattr(ai_message, 'content') else str(ai_message)
except Exception as e:
return f"Errore: {str(e)}"
# Prova EnhancedCLI, altrimenti usa CLI semplice
use_enhanced = not args.simple
if use_enhanced:
try:
from enhanced_cli import create_enhanced_cli
cli = create_enhanced_cli(
history_file="./.langcode/.cli_history",
theme="default"
)
cli.run(agent_callback)
return
except Exception as e:
print(f"⚠️ EnhancedCLI non disponibile: {e}")
print("📝 Uso CLI semplice...")
# CLI semplice con input()
print("\n" + "=" * 60)
print(" CLI SEMPLICE - digita /exit per uscire")
print("=" * 60 + "\n")
while True:
try:
user_input = input("❯ ").strip()
if not user_input:
continue
if user_input.lower() in ["/exit", "/quit"]:
print("👋 Arrivederci!")
break
if user_input.startswith("/"):
if user_input == "/help":
print("""
Comandi disponibili:
/help - Mostra questo messaggio
/status - Mostra stato del sistema
/exit - Esci
""")
continue
elif user_input == "/status":
print("Agent: in esecuzione")
continue
else:
print(f"Comando sconosciuto: {user_input}")
continue
# Mostra input utente
print(f"👤 {user_input}\n")
# Chiamata all'agent CON indicatore di pensiero
thinking_active.set()
thinker = threading.Thread(target=thinking_indicator, daemon=True)
thinker.start()
try:
response = agent_callback(user_input)
finally:
thinking_active.clear()
thinker.join(timeout=1)
# Strip ANSI codes per terminali semplici
clean_response = strip_ansi(response)
# Paginazione con 'more' style per output lunghi
lines = clean_response.split('\n')
if len(lines) > 15:
for i in range(0, len(lines), 15):
chunk = '\n'.join(lines[i:i+15])
print(f"🤖 {chunk}")
if i + 15 < len(lines):
print("\n" + "─" * 40)
print("Premi ENTER per continuare, 'q' per uscire: ", end="")
try:
user_char = input()
if user_char.lower() == 'q':
break
except:
break
print()
else:
print(f"🤖 {clean_response}\n")
except KeyboardInterrupt:
print("\n👋 Arrivederci!")
break
except EOFError:
break
except Exception as e:
print(f"❌ Inizializzazione fallita: {str(e)}")
sys.exit(1)
def cmd_server(args):
"""Avvia il web server"""
import uvicorn
from api_server import app
host = args.host or "0.0.0.0"
port = args.port or 8000
print(f"🌐 Avvio LangCode Web Server su http://{host}:{port}")
print(f"📊 Dashboard: http://{host}:{port}/monitoring")
print(f"🛒 Marketplace: http://{host}:{port}/marketplace")
print(f"📡 API Docs: http://{host}:{port}/docs")
print("Premi Ctrl+C per fermare il server\n")
uvicorn.run(app, host=host, port=port)
def cmd_tui(args):
"""Apre la TUI per gestione configurazioni"""
from config_tui import main as tui_main
tui_main()
def cmd_config_list(args):
"""Lista tutte le configurazioni"""
from config_cli import print_configurations
from config_manager import config_manager
configs = config_manager.get_configurations()
if args.search:
from config_cli import filter_configurations
configs = filter_configurations(configs, args.search)
print_configurations(configs, title=f"Configurazioni trovate per '{args.search}'")
else:
print_configurations(configs)
def cmd_config_add(args):
"""Aggiunge una nuova configurazione"""
from config_manager import config_manager
if args.name and args.provider:
# Modalità non-interattiva
config_data = {"provider": args.provider}
if args.model:
config_data["model_name"] = args.model
else:
# Default model per provider
defaults = {
"openai": "gpt-4o",
"ollama": "gemma2:9b",
"openrouter": "anthropic/claude-3.5-sonnet",
"anthropic": "claude-3-5-sonnet-20241022",
"google": "gemini-1.5-pro"
}
config_data["model_name"] = defaults.get(args.provider, "gpt-4o")
if args.api_key:
config_data["api_key"] = args.api_key
if args.base_url:
config_data["base_url"] = args.base_url
try:
config_manager.add_configuration(args.name, config_data)
print(f"✅ Configurazione '{args.name}' aggiunta con successo!")
if args.activate:
config_manager.set_active_configuration(args.name)
print(f"✅ Configurazione '{args.name}' attivata!")
except Exception as e:
print(f"❌ Errore: {e}")
sys.exit(1)
else:
# Modalità interattiva
from config_cli import add_configuration
add_configuration()
def cmd_config_remove(args):
"""Rimuove una configurazione"""
from config_manager import config_manager
name = args.name
if not name:
name = input("Nome configurazione da rimuovere: ").strip()
if not name:
print("❌ Nome richiesto.")
sys.exit(1)
try:
config_manager.remove_configuration(name)
print(f"✅ Configurazione '{name}' rimossa!")
except Exception as e:
print(f"❌ Errore: {e}")
sys.exit(1)
def cmd_config_use(args):
"""Attiva una configurazione"""
from config_manager import config_manager
try:
config_manager.set_active_configuration(args.name)
print(f"✅ Configurazione '{args.name}' attivata!")
except Exception as e:
print(f"❌ Errore: {e}")
sys.exit(1)
def cmd_config_active(args):
"""Mostra la configurazione attiva"""
from config_manager import config_manager
active = config_manager.get_active_configuration()
if active:
config = config_manager.get_configuration(active)
print(f"\n📌 Configurazione attiva: {active}")
print(f" Provider: {config.get('provider', 'unknown')}")
print(f" Model: {config.get('model_name', 'unknown')}")
if config.get('base_url'):
print(f" Base URL: {config.get('base_url')}")
else:
print("⚠️ Nessuna configurazione attiva.")
def cmd_config_test(args):
"""Testa una configurazione"""
from config_manager import config_manager
name = args.name or config_manager.get_active_configuration()
if not name:
print("❌ Nessuna configurazione specificata o attiva.")
sys.exit(1)
config = config_manager.get_configuration(name)
if not config:
print(f"❌ Configurazione '{name}' non trovata.")
sys.exit(1)
print(f"🔍 Test connessione per '{name}'...")
provider = config.get("provider", "")
try:
if provider == "ollama":
import requests
base_url = config.get("base_url", "http://localhost:11434")
resp = requests.get(f"{base_url}/api/tags", timeout=10)
if resp.status_code == 200:
models = resp.json().get("models", [])
print(f"✅ Ollama connesso! {len(models)} modelli disponibili.")
for m in models[:5]:
print(f" • {m['name']}")
else:
print(f"❌ Errore: HTTP {resp.status_code}")
elif provider in ("openai", "openrouter", "anthropic"):
api_key = config.get("api_key", "")
if not api_key:
print("❌ API key non configurata.")
sys.exit(1)
print("✅ API key presente (test completo richiede una chiamata API)")
elif provider == "google":
api_key = config.get("api_key", "")
if not api_key:
print("❌ API key non configurata.")
sys.exit(1)
print("✅ API key presente (test completo richiede una chiamata API)")
else:
print(f"⚠️ Test non supportato per provider '{provider}'")
except Exception as e:
print(f"❌ Errore connessione: {e}")
sys.exit(1)
def cmd_config_export(args):
"""Esporta configurazioni"""
from config_cli import export_configurations
if args.format and args.output:
import json
import yaml
from pathlib import Path
from config_manager import config_manager
configs = config_manager.get_configurations()
if not configs:
print("❌ Nessuna configurazione da esportare.")
return
try:
if args.format == 'json':
export_file = Path(f"{args.output}.json")
with open(export_file, 'w') as f:
json.dump(configs, f, indent=2)
else:
export_file = Path(f"{args.output}.yaml")
with open(export_file, 'w') as f:
yaml.dump(configs, f, default_flow_style=False)
print(f"✅ Configurazioni esportate in {export_file}")
except Exception as e:
print(f"❌ Errore: {e}")
sys.exit(1)
else:
export_configurations()
def cmd_config_import(args):
"""Importa configurazioni"""
if args.file:
import json
import yaml
from pathlib import Path
from config_manager import config_manager
file_path = Path(args.file)
if not file_path.exists():
print(f"❌ File '{args.file}' non trovato.")
sys.exit(1)
try:
with open(file_path, 'r') as f:
if args.file.endswith('.json'):
imported = json.load(f)
elif args.file.endswith(('.yaml', '.yml')):
imported = yaml.safe_load(f)
else:
print("❌ Formato non supportato. Usa .json o .yaml")
sys.exit(1)
if not isinstance(imported, dict):
print("❌ Formato configurazione non valido.")
sys.exit(1)
count = 0
for name, config in imported.items():
try:
config_manager.add_configuration(name, config)
count += 1
print(f"✅ Importata: {name}")
except Exception as e:
print(f"⚠️ '{name}': {e}")
print(f"\n✅ {count} configurazioni importate.")
except Exception as e:
print(f"❌ Errore: {e}")
sys.exit(1)
else:
from config_cli import import_configurations
import_configurations()
def cmd_config_search(args):
"""Cerca configurazioni"""
from config_cli import print_configurations, filter_configurations
from config_manager import config_manager
configs = config_manager.get_configurations()
filtered = filter_configurations(configs, args.query)
print_configurations(filtered, title=f"Risultati per '{args.query}'")
def cmd_analytics(args):
"""Mostra analytics"""
from analytics import AnalyticsEngine
engine = AnalyticsEngine()
if args.report:
report = engine.get_detailed_report()
print(report)
elif args.clear:
engine.clear_old_data()
print("✅ Dati analytics cancellati.")
else:
# Mostra riepilogo rapido
summary = engine.get_usage_summary()
insights = engine.get_insights()
print("\n📊 LangCode Analytics")
print("=" * 40)
if summary.get("total_conversations", 0) > 0:
print(f"\n💬 Conversazioni totali: {summary['total_conversations']}")
print(f"📝 Messaggi totali: {summary.get('total_messages', 0)}")
print(f"⏱️ Tempo medio risposta: {summary.get('avg_response_time', 0):.2f}s")
if summary.get("provider_usage"):
print("\n🔌 Utilizzo provider:")
for provider, count in summary["provider_usage"].items():
print(f" • {provider}: {count} volte")
if summary.get("tool_usage"):
print("\n🔧 Utilizzo strumenti:")
for tool, count in sorted(summary["tool_usage"].items(), key=lambda x: -x[1])[:5]:
print(f" • {tool}: {count} volte")
else:
print("\n📭 Nessun dato analytics ancora.")
print(" Usa l'agent per generare dati!")
if insights:
print("\n💡 Insights:")
for insight in insights[:3]:
print(f" • {insight}")
print(f"\n📁 Dati salvati in: {engine.data_dir}")
print(" Usa 'langcode analytics --report' per report completo")
def cmd_dashboard(args):
"""Mostra dashboard performance"""
from dashboard import PerformanceDashboard
dashboard = PerformanceDashboard()
if args.clear:
dashboard.clear_old_data()
print("✅ Dati performance cancellati.")
else:
report = dashboard.get_summary()
print("\n📈 LangCode Performance Dashboard")
print("=" * 40)
if report.get("total_requests", 0) > 0:
print(f"\n📊 Richieste totali: {report['total_requests']}")
print(f"⏱️ Tempo medio risposta: {report.get('avg_response_time', 0):.2f}s")
print(f"✅ Tasso successo: {report.get('success_rate', 0):.1f}%")
if report.get("slowest_operations"):
print("\n🐌 Operazioni più lente:")
for op in report["slowest_operations"][:3]:
print(f" • {op['name']}: {op['time']:.2f}s")
if report.get("memory_usage"):
mem = report["memory_usage"]
print(f"\n💾 Memoria: {mem.get('current', 0):.1f}MB (picco: {mem.get('peak', 0):.1f}MB)")
else:
print("\n📭 Nessun dato performance ancora.")
print(f"\n📁 Dati salvati in: {dashboard.data_dir}")
def cmd_version(args):
"""Mostra versione"""
print(f"LangCode Agent v{VERSION}")
print("Agent harness basato su LangChain Deep Agents")
print("https://github.com/truetype/langcode_agent")
def main():
parser = argparse.ArgumentParser(
prog="langcode",
description="🚀 LangCode Agent - AI Agent harness configurabile",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Esempi:
langcode Avvia l'agent (chat interattiva)
langcode server --port 3000 Avvia web server su porta 3000
langcode config list Lista tutte le configurazioni
langcode config add -n ollama -p ollama -m gemma2:9b
langcode config use ollama Attiva configurazione 'ollama'
langcode config test Testa configurazione attiva
langcode analytics Mostra riepilogo analytics
langcode analytics --report Genera report completo
"""
)
parser.add_argument("--version", "-v", action="version", version=f"langcode {VERSION}")
subparsers = parser.add_subparsers(dest="command", help="Comandi disponibili")
# === CHAT (default) ===
chat_parser = subparsers.add_parser("chat", help="Avvia l'agent in modalità chat")
chat_parser.add_argument("--simple", "-s", action="store_true", help="Usa CLI semplice invece di EnhancedCLI")
chat_parser.set_defaults(func=cmd_chat)
# === SERVER ===
server_parser = subparsers.add_parser("server", help="Avvia il web server")
server_parser.add_argument("--port", "-p", type=int, default=8000, help="Porta del server (default: 8000)")
server_parser.add_argument("--host", "-H", default="0.0.0.0", help="Host del server (default: 0.0.0.0)")
server_parser.set_defaults(func=cmd_server)
# === TUI ===
tui_parser = subparsers.add_parser("tui", help="Apre TUI per gestione configurazioni")
tui_parser.set_defaults(func=cmd_tui)
# === CONFIG (nested subcommands) ===
config_parser = subparsers.add_parser("config", help="Gestione configurazioni")
config_subparsers = config_parser.add_subparsers(dest="config_command", help="Sottocomandi config")
# config list
cl = config_subparsers.add_parser("list", help="Lista configurazioni")
cl.add_argument("--search", "-s", help="Filtra per termine di ricerca")
cl.set_defaults(func=cmd_config_list)
# config add
ca = config_subparsers.add_parser("add", help="Aggiungi configurazione")
ca.add_argument("--name", "-n", help="Nome configurazione")
ca.add_argument("--provider", "-p", choices=["openai", "ollama", "openrouter", "anthropic", "google"], help="Provider")
ca.add_argument("--model", "-m", help="Nome modello")
ca.add_argument("--api-key", "-k", help="API key")
ca.add_argument("--base-url", "-u", help="Base URL")
ca.add_argument("--activate", "-a", action="store_true", help="Attiva dopo la creazione")
ca.set_defaults(func=cmd_config_add)
# config remove
cr = config_subparsers.add_parser("remove", help="Rimuovi configurazione")
cr.add_argument("name", help="Nome configurazione da rimuovere")
cr.set_defaults(func=cmd_config_remove)
# config use
cu = config_subparsers.add_parser("use", help="Attiva configurazione")
cu.add_argument("name", help="Nome configurazione da attivare")
cu.set_defaults(func=cmd_config_use)
# config active
cv = config_subparsers.add_parser("active", help="Mostra configurazione attiva")
cv.set_defaults(func=cmd_config_active)
# config test
ct = config_subparsers.add_parser("test", help="Testa connessione configurazione")
ct.add_argument("name", nargs="?", help="Nome configurazione (default: attiva)")
ct.set_defaults(func=cmd_config_test)
# config export
ce = config_subparsers.add_parser("export", help="Esporta configurazioni")
ce.add_argument("--format", "-f", choices=["json", "yaml"], help="Formato export")
ce.add_argument("--output", "-o", help="Nome file output (senza estensione)")
ce.set_defaults(func=cmd_config_export)
# config import
ci = config_subparsers.add_parser("import", help="Importa configurazioni")
ci.add_argument("file", nargs="?", help="File da importare (.json o .yaml)")
ci.set_defaults(func=cmd_config_import)
# config search
cs = config_subparsers.add_parser("search", help="Cerca configurazioni")
cs.add_argument("query", help="Termine di ricerca")
cs.set_defaults(func=cmd_config_search)
# === ANALYTICS ===
analytics_parser = subparsers.add_parser("analytics", help="Mostra analytics")
analytics_parser.add_argument("--report", "-r", action="store_true", help="Genera report completo")
analytics_parser.add_argument("--clear", "-c", action="store_true", help="Cancella dati analytics")
analytics_parser.set_defaults(func=cmd_analytics)
# === DASHBOARD ===
dashboard_parser = subparsers.add_parser("dashboard", help="Mostra dashboard performance")
dashboard_parser.add_argument("--clear", "-c", action="store_true", help="Cancella dati performance")
dashboard_parser.set_defaults(func=cmd_dashboard)
# === VERSION ===
version_parser = subparsers.add_parser("version", help="Mostra versione")
version_parser.set_defaults(func=cmd_version)
# Parse arguments
args = parser.parse_args()
# Se nessun comando, avvia chat (default)
if not args.command:
args.func = cmd_chat
args.func(args)
return
# Esegui il comando
if hasattr(args, 'func'):
args.func(args)
else:
parser.print_help()
if __name__ == "__main__":
main()