Compare commits
28
Commits
4fef38b740
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
442c91ee66 | ||
|
|
858b469def | ||
|
|
2e4175f26b | ||
|
|
e72905fd78 | ||
|
|
aec5736394 | ||
|
|
e69a289361 | ||
|
|
b6e8864829 | ||
|
|
39b9288773 | ||
|
|
20233db282 | ||
|
|
c76beb764c | ||
|
|
6cb514ec13 | ||
|
|
5d9bd48013 | ||
|
|
453156148c | ||
|
|
e75a75e16c | ||
|
|
f1d1248d4b | ||
|
|
3197dedd4c | ||
|
|
918b1f3a5b | ||
|
|
c8e7489c26 | ||
|
|
82b70f88d1 | ||
|
|
3a042355b8 | ||
|
|
9fa564e12b | ||
|
|
4a0342dcb7 | ||
|
|
32ed76e5be | ||
|
|
fc60fc1fd0 | ||
|
|
cce8709149 | ||
|
|
f847b92b0a | ||
|
|
1c7781a68f | ||
|
|
e5ce70c68e |
No files matched your search
+20
-4
@@ -37,15 +37,31 @@ langcode config test
|
||||
|
||||
## Modalità di Utilizzo
|
||||
|
||||
### 1. Chat Interattiva (Agent AI)
|
||||
### 1. Chat Interattiva (EnhancedCLI)
|
||||
|
||||
```bash
|
||||
langcode
|
||||
# oppure
|
||||
langcode chat
|
||||
```
|
||||
|
||||
Chatta con l'agent AI in tempo reale.
|
||||
Chatta con l'agent AI in tempo reale con UI moderna:
|
||||
- **Colors**: User (cyan), Agent (green), Error (red)
|
||||
- **History**: Cronologia comandi persistente (↑↓)
|
||||
- **Autocomplete**: Tab per completare comandi
|
||||
- **4 Temi**: default, dark, light, monokai
|
||||
- **Syntax Highlighting**: Code blocks colorati
|
||||
- **Keyboard shortcuts**: Ctrl+A/E, Ctrl+K/U, Ctrl+W
|
||||
|
||||
**Comandi CLI interni:**
|
||||
```
|
||||
/help - Mostra aiuto
|
||||
/status - Stato sistema
|
||||
/clear - Pulisci schermo (Ctrl+L)
|
||||
/theme-* - Cambia tema
|
||||
/timestamps - Toggle timestamps
|
||||
/syntax - Toggle syntax highlighting
|
||||
/history - Mostra cronologia
|
||||
/exit - Esci
|
||||
```
|
||||
|
||||
### 2. Server Web (API + Dashboard)
|
||||
|
||||
|
||||
+293
@@ -0,0 +1,293 @@
|
||||
"""
|
||||
Terminal UI con scrollback usando Curses
|
||||
Scrollback vero con frecce e PgUp/PgDn
|
||||
"""
|
||||
import sys
|
||||
import signal
|
||||
import curses
|
||||
from typing import Optional, Callable
|
||||
|
||||
try:
|
||||
import curses
|
||||
CURSES_AVAILABLE = True
|
||||
except ImportError:
|
||||
CURSES_AVAILABLE = False
|
||||
|
||||
|
||||
class CursesUI:
|
||||
"""
|
||||
Terminal UI con scrollback usando curses.
|
||||
|
||||
Features:
|
||||
- Scrollback completo con PgUp/PgDn
|
||||
- Frecce su/giù per navigare
|
||||
- Resize automatico
|
||||
- Colors (se supportati)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
):
|
||||
"""
|
||||
Inizializza il terminale UI con curses.
|
||||
|
||||
Args:
|
||||
scrollback: Numero di righe nello scrollback buffer
|
||||
prompt: Prompt da mostrare
|
||||
"""
|
||||
if not CURSES_AVAILABLE:
|
||||
raise ImportError("curses non è disponibile su questo sistema")
|
||||
|
||||
self.scrollback = scrollback
|
||||
self.prompt = prompt
|
||||
|
||||
# Buffer per scrollback
|
||||
self.lines: list = []
|
||||
|
||||
# Callback per l'agent
|
||||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||||
|
||||
def _init_colors(self, stdscr):
|
||||
"""Inizializza i colori"""
|
||||
if curses.has_colors():
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
# Colori: 1=cyan, 2=green, 3=yellow, 4=red, 5=magenta, 6=blue
|
||||
curses.init_pair(1, curses.COLOR_CYAN, -1) # Header
|
||||
curses.init_pair(2, curses.COLOR_GREEN, -1) # Agent output
|
||||
curses.init_pair(3, curses.COLOR_YELLOW, -1) # Thinking
|
||||
curses.init_pair(4, curses.COLOR_RED, -1) # Error
|
||||
curses.init_pair(5, curses.COLOR_MAGENTA, -1) # Prompt
|
||||
curses.init_pair(6, curses.COLOR_BLUE, -1) # User input
|
||||
|
||||
def _add_line(self, text: str, color: int = 2):
|
||||
"""Aggiunge una linea al buffer"""
|
||||
self.lines.append((text, color))
|
||||
# Limita la dimensione del buffer
|
||||
if len(self.lines) > self.scrollback:
|
||||
self.lines.pop(0)
|
||||
|
||||
def _draw(self, stdscr):
|
||||
"""Disegna lo schermo"""
|
||||
stdscr.clear()
|
||||
|
||||
# Ottieni dimensioni
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
|
||||
# Header
|
||||
header = "═" * (max_x - 4)
|
||||
stdscr.addstr(0, 0, f"╔{header}╗", curses.color_pair(1) | curses.A_BOLD)
|
||||
title = " LangCode Agent - Curses UI "
|
||||
stdscr.addstr(1, (max_x - len(title)) // 2, title, curses.color_pair(1) | curses.A_BOLD)
|
||||
stdscr.addstr(2, 0, f"╚{header}╝", curses.color_pair(1) | curses.A_BOLD)
|
||||
|
||||
# Calcola area visibile
|
||||
start_row = 4
|
||||
visible_rows = max_y - start_row - 2 # -2 per status e prompt
|
||||
|
||||
# Mostra linee visibili
|
||||
total_lines = len(self.lines)
|
||||
|
||||
# Scroll offset (0 = showing latest)
|
||||
# La scrolling position va da 0 a max(0, total_lines - visible_rows)
|
||||
self.scroll_offset = max(0, total_lines - visible_rows)
|
||||
|
||||
# Disegna le linee visibili
|
||||
for i in range(min(visible_rows, total_lines)):
|
||||
line_idx = self.scroll_offset + i
|
||||
if line_idx < total_lines:
|
||||
text, color = self.lines[line_idx]
|
||||
# Tronca se troppo lunga
|
||||
display_text = text[:max_x - 2] if len(text) > max_x - 2 else text
|
||||
try:
|
||||
stdscr.addstr(start_row + i, 0, display_text, curses.color_pair(color))
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Status bar
|
||||
status_y = max_y - 2
|
||||
scroll_info = f"Linee: {total_lines} | Scroll: {self.scroll_offset} | PgUp/PgDn=Frecce=Scroll | q=quit | Ctrl+C=exit"
|
||||
stdscr.addstr(status_y, 0, "─" * (max_x - len(scroll_info) - 1), curses.A_DIM)
|
||||
stdscr.addstr(status_y, max_x - len(scroll_info) - 1, scroll_info, curses.A_DIM)
|
||||
|
||||
# Prompt
|
||||
stdscr.addstr(max_y - 1, 0, f"\033[1;35m{self.prompt}\033[0m ", curses.color_pair(5) | curses.A_BOLD)
|
||||
|
||||
stdscr.refresh()
|
||||
|
||||
def _handle_input(self, stdscr):
|
||||
"""Gestisce l'input"""
|
||||
curses.curs_set(1) # Mostra cursore
|
||||
|
||||
# Abilita arrow keys e PgUp/PgDn
|
||||
stdscr.keypad(True)
|
||||
|
||||
# Configura timeout per read non-blocking
|
||||
stdscr.timeout(100) # 100ms timeout
|
||||
|
||||
input_buffer = ""
|
||||
while True:
|
||||
self._draw(stdscr)
|
||||
|
||||
try:
|
||||
key = stdscr.getch()
|
||||
except curses.error:
|
||||
continue
|
||||
|
||||
if key == -1:
|
||||
# Timeout - continua
|
||||
continue
|
||||
|
||||
elif key in [curses.KEY_ENTER, 10, 13]:
|
||||
# Enter - processa comando
|
||||
if input_buffer.strip():
|
||||
return input_buffer.strip()
|
||||
input_buffer = ""
|
||||
|
||||
elif key in [curses.KEY_UP]:
|
||||
# Scroll up nel buffer
|
||||
if self.scroll_offset > 0:
|
||||
self.scroll_offset -= 1
|
||||
|
||||
elif key in [curses.KEY_DOWN]:
|
||||
# Scroll down nel buffer
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
visible_rows = max_y - 6
|
||||
if self.scroll_offset < max(0, len(self.lines) - visible_rows):
|
||||
self.scroll_offset += 1
|
||||
|
||||
elif key in [curses.KEY_PPAGE]:
|
||||
# PgUp - scroll su di una pagina
|
||||
max_y, _ = stdscr.getmaxyx()
|
||||
self.scroll_offset = max(0, self.scroll_offset - (max_y - 6))
|
||||
|
||||
elif key in [curses.KEY_NPAGE]:
|
||||
# PgDn - scroll giu di una pagina
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
visible_rows = max_y - 6
|
||||
self.scroll_offset = min(max(0, len(self.lines) - visible_rows),
|
||||
self.scroll_offset + visible_rows)
|
||||
|
||||
elif key == 27: # ESC
|
||||
return "/exit"
|
||||
|
||||
elif key in [3, 26]: # Ctrl+C
|
||||
return "/exit"
|
||||
|
||||
elif key in [curses.KEY_BACKSPACE, 127, 8]:
|
||||
# Backspace
|
||||
input_buffer = input_buffer[:-1]
|
||||
|
||||
elif 32 <= key <= 126:
|
||||
# Carattere stampabile
|
||||
input_buffer += chr(key)
|
||||
|
||||
# Mostra input corrente da qualche parte se necessario
|
||||
|
||||
def start(self, agent_callback: Callable[[str], str]):
|
||||
"""
|
||||
Avvia il terminale interattivo con curses.
|
||||
"""
|
||||
self.agent_callback = agent_callback
|
||||
|
||||
def handle_sigint(signum, frame):
|
||||
sys.exit(0)
|
||||
|
||||
old_handler = signal.signal(signal.SIGINT, handle_sigint)
|
||||
|
||||
try:
|
||||
curses.wrapper(self._main_loop)
|
||||
finally:
|
||||
signal.signal(signal.SIGINT, old_handler)
|
||||
|
||||
def _main_loop(self, stdscr):
|
||||
"""Loop principale curses"""
|
||||
self._init_colors(stdscr)
|
||||
curses.cbreak()
|
||||
curses.noecho()
|
||||
|
||||
# Aggiungi header
|
||||
self._add_line("╔═══════════════════════════════════════════════════════╗", 1)
|
||||
self._add_line("║ LangCode Agent - Curses UI con Scrollback ║", 1)
|
||||
self._add_line("╚═══════════════════════════════════════════════════════╝", 1)
|
||||
self._add_line("", 0)
|
||||
self._add_line("Scrivi un messaggio e premi Enter. Usa PgUp/PgDn per scrollare.", 6)
|
||||
self._add_line("Comandi: /help, /clear, /exit", 6)
|
||||
self._add_line("", 0)
|
||||
|
||||
while True:
|
||||
# Ottieni input
|
||||
stdscr.timeout(-1) # Bloccante per l'input
|
||||
command = self._handle_input(stdscr)
|
||||
|
||||
if command.lower() in ['/exit', '/quit']:
|
||||
break
|
||||
|
||||
if not command:
|
||||
continue
|
||||
|
||||
if command == '/help':
|
||||
self._add_line("╔═══════════════════════════════════════════════════════╗", 1)
|
||||
self._add_line("║ COMANDI ║", 1)
|
||||
self._add_line("╠═══════════════════════════════════════════════════════╣", 1)
|
||||
self._add_line("║ /help - Mostra questo messaggio ║", 1)
|
||||
self._add_line("║ /clear - Pulisci lo schermo ║", 1)
|
||||
self._add_line("║ /exit - Esci dall'agente ║", 1)
|
||||
self._add_line("║ PgUp - Scroll su ║", 1)
|
||||
self._add_line("║ PgDn - Scroll giu ║", 1)
|
||||
self._add_line("║ Frecce - Naviga nel scrollback ║", 1)
|
||||
self._add_line("╚═══════════════════════════════════════════════════════╝", 1)
|
||||
continue
|
||||
|
||||
if command == '/clear':
|
||||
self.lines.clear()
|
||||
continue
|
||||
|
||||
# Mostra comando utente
|
||||
self._add_line("", 0)
|
||||
self._add_line(f"👤 {command}", 6)
|
||||
|
||||
# Chiamata all'agent
|
||||
self._add_line("", 0)
|
||||
self._add_line("🤔 Pensando...", 3)
|
||||
|
||||
try:
|
||||
response = agent_callback(command)
|
||||
|
||||
# Rimuovi "thinking" e mostra risposta
|
||||
if self.lines and "🤔 Pensando..." in self.lines[-1][0]:
|
||||
self.lines.pop()
|
||||
|
||||
# Aggiungi risposta
|
||||
for line in response.split('\n'):
|
||||
self._add_line(f"🤖 {line}", 2)
|
||||
except Exception as e:
|
||||
self._add_line(f"❌ Errore: {str(e)}", 4)
|
||||
|
||||
self._add_line("", 0)
|
||||
|
||||
|
||||
def create_curses_ui(
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
) -> CursesUI:
|
||||
"""Factory function per creare CursesUI"""
|
||||
return CursesUI(
|
||||
scrollback=scrollback,
|
||||
prompt=prompt
|
||||
)
|
||||
|
||||
|
||||
# Test standalone
|
||||
if __name__ == "__main__":
|
||||
def test_callback(text):
|
||||
return f"Ricevuto: {text}\n\nRisposta multi-linea:\n1. Prima linea\n2. Seconda\n3. Terza\n4. Quarta\n5. Quinta"
|
||||
|
||||
print("Avvio CursesUI di test...")
|
||||
ui = create_curses_ui()
|
||||
try:
|
||||
ui.start(test_callback)
|
||||
except KeyboardInterrupt:
|
||||
print("\nUscito")
|
||||
@@ -0,0 +1,271 @@
|
||||
# Da Chatbot ad Agente AI: Gap Analysis e Roadmap
|
||||
|
||||
## Stato Attuale del Progetto
|
||||
|
||||
**LangCode Agent è ora un Agente AI completo** con loop operativo, pianificazione, memoria persistente e UI moderna.
|
||||
|
||||
L'agente può:
|
||||
- Eseguire goal in autonomia attraverso il loop Pensa→Agisci→Osserva
|
||||
- Pianificare e tracciare task multipli
|
||||
- Salvare e recuperare fatti e preferenze
|
||||
- Interagire con l'utente attraverso una UI moderna a colori
|
||||
|
||||
---
|
||||
|
||||
## Chatbot vs Agente: Gap Analysis
|
||||
|
||||
| Caratteristica | Chatbot (Attuale) | Agente AI (Obiettivo) | Stato Attuale |
|
||||
|----------------|-------------------|----------------------|---------------|
|
||||
| **Input/Output** | Turn-by-turn | Goal-oriented | ✅ Goal-oriented |
|
||||
| **Planning** | Nessuno | Scomposizione in task | ✅ Integrato |
|
||||
| **Loop operativo** | Singola risposta | Pensa → Agisci → Osserva → Ripianifica | ✅ Funzionante |
|
||||
| **Memoria** | Solo cronologia sessione | A lungo termine, persistente | ✅ Integrato |
|
||||
| **Autonomia** | Dipende da input utente | Esegue autonomamente | ✅ Parziale |
|
||||
| **Tool use** | Arricchisci risposta | Agisci sull'ambiente | ✅ Funzionante |
|
||||
| **Goal tracking** | Nessuno | Tiene traccia obiettivi | ✅ Integrato nel planner |
|
||||
|
||||
---
|
||||
|
||||
## Componenti Implementati
|
||||
|
||||
### 1. Planning System (`enhancements/planning_enhancement.py`)
|
||||
- **Integrato in**: `ReActAgent`
|
||||
- **Funzionalità**: Task con priorità, dipendenze, progress tracking
|
||||
- **API**: `run()`, `run_plan()`, `get_plan_progress()`
|
||||
|
||||
### 2. Simple Memory (`enhancements/simple_memory.py`)
|
||||
- **Integrato in**: `ReActAgent`
|
||||
- **Funzionalità**: Memorizzazione fatti, preferenze, cronologia
|
||||
- **Tool**: `remember`, `recall`, `preference`, `memory_stats`
|
||||
|
||||
### 3. Enhanced Memory (`enhancements/memory_enhancement.py`)
|
||||
- **Esiste**: Vector store con FAISS, semantic search
|
||||
- **Dipendenza**: Richiede `nomic-embed-text` in Ollama
|
||||
- **Stato**: Parzialmente funzionante
|
||||
|
||||
---
|
||||
|
||||
## Cosa Manca per Diventare Agente
|
||||
|
||||
### ✅ 1. **Agent Loop (COMPLETATO)**
|
||||
Implementato in `enhancements/react_agent.py`:
|
||||
- Loop Pensa → Agisci → Osserva
|
||||
- Massimo 10 iterazioni configurabili
|
||||
- Supporto per tutti i tool built-in
|
||||
|
||||
### ✅ 2. **Goal Decomposition (COMPLETATO)**
|
||||
- `decompose_goal()` per scomposizione automatica
|
||||
- `run_plan()` per esecuzione sequenziale di goal multipli
|
||||
|
||||
### 3. **Self-Reflection**
|
||||
- L'agente deve valutare se il risultato è soddisfacente
|
||||
- Se un tool fallisce, provare approccio alternativo
|
||||
- **Stato**: Da implementare
|
||||
|
||||
### 4. **Enhanced Memory (Vector Store)**
|
||||
- Integrato ma richiede `nomic-embed-text` in Ollama
|
||||
- **Stato**: Dipendenza esterna necessaria
|
||||
|
||||
### 4. **Persistent Memory Integration**
|
||||
- Memoria deve essere доступна in ogni conversazione
|
||||
- L'agente deve poter salvare e recuperare informazioni
|
||||
|
||||
### 5. **Multi-Step Execution Context**
|
||||
- L'agente deve mantenere stato tra diverse chiamate tool
|
||||
- Non ripetere azioni già completate
|
||||
|
||||
---
|
||||
|
||||
## Roadmap di Implementazione
|
||||
|
||||
### ✅ Fase 1: Agent Loop Base (COMPLETATO)
|
||||
**Obiettivo**: Abilitare esecuzione multi-step
|
||||
|
||||
Implementato in `enhancements/react_agent.py`:
|
||||
- ThinkStep: LLM decide prossima azione (via intent detection)
|
||||
- ActStep: Esegui tool
|
||||
- ObserveStep: Leggi risultato
|
||||
- EvaluateStep: Goal raggiunto?
|
||||
- ReplanStep: Ripianifica se fallito
|
||||
|
||||
**Modifiche a `ollama_agent.py`**:
|
||||
- Aggiunto metodo `run()` con loop
|
||||
- Aggiunto parametro `max_iterations`
|
||||
- Integrato `enhanced_planner` per task tracking
|
||||
|
||||
### ✅ Fase 2: Planning Integration (COMPLETATO)
|
||||
**Obiettivo**: Planning system parte dell'agente
|
||||
|
||||
- `run()` crea e traccia task nel planner
|
||||
- `run_plan()` esegue sequenze di goal
|
||||
- `decompose_goal()` scompone goal complessi via LLM
|
||||
- Task tracciati con stato: pending->in_progress->completed/failed
|
||||
|
||||
### ✅ Fase 3: Memory Integration (COMPLETATO)
|
||||
**Obiettivo**: Rendere memoria parte dell'agente
|
||||
|
||||
- Integrato `simple_memory` in `ReActAgent`
|
||||
- Aggiunti tool: `remember`, `recall`, `preference`, `memory_stats`
|
||||
- Memoria persistente tra sessioni (file `.langcode/simple_memory.json`)
|
||||
- Metodi: `remember()`, `recall()`, `get_memory_stats()`
|
||||
- Totale: 10 tool disponibili (6 built-in + 4 memory)
|
||||
|
||||
### ✅ Fase 4: Enhanced CLI (COMPLETATO)
|
||||
**Obiettivo**: Migliorare l'esperienza utente con UI moderna
|
||||
|
||||
- `enhanced_cli.py`: UI moderna con prompt_toolkit
|
||||
- **Colors**: User (cyan), Agent (green), System (yellow), Error (red)
|
||||
- **History**: Cronologia persistente in `.langcode/.cli_history`
|
||||
- **Autocomplete**: Comandi `/help`, `/status`, `/clear`, `/theme`, `/exit`
|
||||
- **4 Temi**: default, dark, light, monokai
|
||||
- **Syntax Highlighting**: Code blocks con pygments
|
||||
- **Keyboard shortcuts**: Ctrl+A/E, Ctrl+K/U, Ctrl+W
|
||||
- **Toggle**: `/timestamps`, `/syntax`
|
||||
- Integrato in `langcode.py chat`
|
||||
|
||||
---
|
||||
|
||||
## Implementazione Proposta: Nuovo `ReActAgent`
|
||||
|
||||
```python
|
||||
# enhancements/react_agent.py (nuovo)
|
||||
|
||||
class ReActAgent:
|
||||
"""
|
||||
Agente con Loop Pensa→Agisci→Osserva→Ripianifica
|
||||
"""
|
||||
|
||||
def __init__(self, llm, tools, max_iterations=10):
|
||||
self.llm = llm
|
||||
self.tools = {t.name: t for t in tools}
|
||||
self.max_iterations = max_iterations
|
||||
self.planner = EnhancedPlanner() # Esistente
|
||||
self.memory = SimplePersistentMemory() # Esistente
|
||||
|
||||
def run(self, goal: str) -> Dict:
|
||||
"""
|
||||
Esegue un goal attraverso il loop ReAct.
|
||||
"""
|
||||
context = f"Goal: {goal}\n\n"
|
||||
|
||||
for i in range(self.max_iterations):
|
||||
# 1. THINK: LLM decide prossima azione
|
||||
thought = self._think(context, goal)
|
||||
|
||||
# 2. ACT: Esegui tool se necessario
|
||||
if thought.get("action"):
|
||||
result = self._act(thought["action"], thought["args"])
|
||||
context += f"\nAzione: {thought['action']}\nRisultato: {result}\n"
|
||||
|
||||
# Salva in memoria se rilevante
|
||||
if thought["action"] in ["read_file", "execute"]:
|
||||
self.memory.add_fact(f"Azione completata: {thought['action']}")
|
||||
|
||||
# 3. OBSERVE & EVALUATE
|
||||
if self._goal_achieved(context, goal):
|
||||
return {"success": True, "result": context}
|
||||
|
||||
return {"success": False, "result": context, "error": "Max iterations"}
|
||||
|
||||
def _think(self, context, goal) -> Dict:
|
||||
"""LLM decide prossima azione basata su contesto"""
|
||||
# Prompt ingegnerizzato per decision-making
|
||||
...
|
||||
|
||||
def _act(self, action_name, args) -> str:
|
||||
"""Esegue azione tramite tool"""
|
||||
if action_name in self.tools:
|
||||
return self.tools[action_name].invoke(args)
|
||||
return f"Tool {action_name} non trovato"
|
||||
|
||||
def _goal_achieved(self, context, goal) -> bool:
|
||||
"""Valuta se goal è stato raggiunto"""
|
||||
...
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Utilizzo Attuale
|
||||
|
||||
### Test Semplice:
|
||||
```bash
|
||||
source venv/bin/activate
|
||||
./venv/bin/python -c "
|
||||
from enhancements.react_agent import ReActAgent
|
||||
from ollama_agent import get_builtin_tools
|
||||
from langchain_ollama import ChatOllama
|
||||
|
||||
llm = ChatOllama(model='qwen3.5:4b', base_url='http://192.168.0.50:11434')
|
||||
agent = ReActAgent(llm=llm, tools=get_builtin_tools())
|
||||
|
||||
result = agent.run('Che OS stai usando?')
|
||||
print(result['result'])
|
||||
"
|
||||
```
|
||||
|
||||
### Test con Piano (multi-goal):
|
||||
```python
|
||||
goals = [
|
||||
'Che OS stai usando?',
|
||||
'Elenca i file',
|
||||
'Leggi requirements.txt'
|
||||
]
|
||||
result = agent.run_plan(goals, 'test_plan')
|
||||
print(f\"Completati: {result['completed']}/{result['total_goals']}\")
|
||||
```
|
||||
|
||||
### Verifica Task nel Planner:
|
||||
```python
|
||||
print(agent.get_plan_progress())
|
||||
# {'total_tasks': 3, 'plans': 1}
|
||||
```
|
||||
|
||||
### Test Memory:
|
||||
```python
|
||||
# Ricorda un fatto
|
||||
agent.remember("L'utente preferisce il tema scuro", "preferenza")
|
||||
|
||||
# Cerca fatti
|
||||
print(agent.recall("tema"))
|
||||
# Trovati 1 ricordi:
|
||||
# - L'utente preferisce il tema scuro (preferenza)
|
||||
|
||||
# Statistiche memoria
|
||||
print(agent.get_memory_stats())
|
||||
# {'total_conversations': 0, 'total_facts': 11, 'total_preferences': 1, ...}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Commit Log
|
||||
|
||||
| Data | Commit | Descrizione |
|
||||
|------|--------|-------------|
|
||||
| 24 Mar 2026 | `e5ce70c` | fix: migliora intent detection e parsing message dict |
|
||||
| 24 Mar 2026 | `1c778a` | feat: aggiungi ReAct Agent Loop |
|
||||
| 24 Mar 2026 | `f847b92` | feat: integra Planning system nel ReActAgent |
|
||||
| 24 Mar 2026 | `fc60fc1` | feat: integra Simple Memory nel ReActAgent |
|
||||
| 24 Mar 2026 | `4a0342d` | feat: aggiungi EnhancedCLI con prompt_toolkit |
|
||||
| 24 Mar 2026 | `9fa564e` | feat: potenzia EnhancedCLI con funzionalità avanzate |
|
||||
|
||||
---
|
||||
|
||||
## Conclusione
|
||||
|
||||
**Il progetto è progredito da chatbot ad agente con UI moderna.**
|
||||
|
||||
Tutte le fasi della roadmap sono completate:
|
||||
- ✅ Agent Loop (Pensa→Agisci→Osserva)
|
||||
- ✅ Planning Integration
|
||||
- ✅ Memory Integration
|
||||
- ✅ Enhanced CLI (UI moderna con colors, history, autocomplete, temi)
|
||||
|
||||
Il progetto è ora un **Agente AI completo** con:
|
||||
- Loop operativo autonomo
|
||||
- Pianificazione task
|
||||
- Memoria persistente
|
||||
- UI moderna e user-friendly
|
||||
|
||||
---
|
||||
|
||||
*Documento aggiornato: 24 Marzo 2026*
|
||||
+734
@@ -0,0 +1,734 @@
|
||||
"""
|
||||
Enhanced CLI for LangCode Agent
|
||||
Provides modern terminal UI with colors, history, autocomplete, and more.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
from typing import Optional, List, Dict, Any, Callable
|
||||
from datetime import datetime
|
||||
|
||||
# prompt_toolkit imports
|
||||
from prompt_toolkit import PromptSession, HTML
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from prompt_toolkit.styles import Style
|
||||
from prompt_toolkit.completion import WordCompleter, NestedCompleter, PathCompleter
|
||||
from prompt_toolkit.history import FileHistory
|
||||
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
|
||||
from prompt_toolkit.formatted_text import FormattedText
|
||||
from prompt_toolkit.cursor_shapes import CursorShape
|
||||
|
||||
# For syntax highlighting
|
||||
try:
|
||||
from pygments.lexers import get_lexer_by_name, get_lexer_for_filename
|
||||
from pygments.token import Token
|
||||
from pygments import highlight as pygments_highlight
|
||||
PYGMENTS_AVAILABLE = True
|
||||
except ImportError:
|
||||
PYGMENTS_AVAILABLE = False
|
||||
get_lexer_by_name = None
|
||||
get_lexer_for_filename = None
|
||||
pygments_highlight = None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# STYLING
|
||||
# =============================================================================
|
||||
|
||||
class Colors:
|
||||
"""ANSI color codes for terminal output"""
|
||||
RESET = "\033[0m"
|
||||
BOLD = "\033[1m"
|
||||
DIM = "\033[2m"
|
||||
ITALIC = "\033[3m"
|
||||
|
||||
BLACK = "\033[30m"
|
||||
RED = "\033[31m"
|
||||
GREEN = "\033[32m"
|
||||
YELLOW = "\033[33m"
|
||||
BLUE = "\033[34m"
|
||||
MAGENTA = "\033[35m"
|
||||
CYAN = "\033[36m"
|
||||
WHITE = "\033[37m"
|
||||
|
||||
BRIGHT_BLACK = "\033[90m"
|
||||
BRIGHT_RED = "\033[91m"
|
||||
BRIGHT_GREEN = "\033[92m"
|
||||
BRIGHT_YELLOW = "\033[93m"
|
||||
BRIGHT_BLUE = "\033[94m"
|
||||
BRIGHT_MAGENTA = "\033[95m"
|
||||
BRIGHT_CYAN = "\033[96m"
|
||||
BRIGHT_WHITE = "\033[97m"
|
||||
|
||||
|
||||
class Theme:
|
||||
"""Color themes for the CLI"""
|
||||
|
||||
@staticmethod
|
||||
def get_style(theme_name: str) -> Style:
|
||||
"""Get prompt_toolkit style from theme name"""
|
||||
themes = {
|
||||
"default": {
|
||||
"user": "ansicyan bold",
|
||||
"agent": "ansigreen",
|
||||
"system": "ansiyellow",
|
||||
"error": "ansired bold",
|
||||
"info": "ansiblue",
|
||||
"muted": "ansiblack",
|
||||
"prompt": "ansimagenta bold",
|
||||
"code": "ansiyellow",
|
||||
"url": "ansiblue underline",
|
||||
},
|
||||
"dark": {
|
||||
"user": "cyan bold",
|
||||
"agent": "lightgreen",
|
||||
"system": "yellow",
|
||||
"error": "lightred bold",
|
||||
"info": "lightblue",
|
||||
"muted": "gray",
|
||||
"prompt": "lightmagenta bold",
|
||||
"code": "yellow",
|
||||
"url": "lightblue underline",
|
||||
},
|
||||
"light": {
|
||||
"user": "blue bold",
|
||||
"agent": "darkgreen",
|
||||
"system": "darkyellow",
|
||||
"error": "darkred bold",
|
||||
"info": "darkblue",
|
||||
"muted": "gray",
|
||||
"prompt": "darkmagenta bold",
|
||||
"code": "darkyellow",
|
||||
"url": "darkblue underline",
|
||||
},
|
||||
"monokai": {
|
||||
"user": "cyan bold",
|
||||
"agent": "green",
|
||||
"system": "yellow",
|
||||
"error": "red bold",
|
||||
"info": "blue",
|
||||
"muted": "gray",
|
||||
"prompt": "magenta bold",
|
||||
"code": "yellow",
|
||||
"url": "cyan underline",
|
||||
},
|
||||
}
|
||||
theme_dict = themes.get(theme_name, themes["default"])
|
||||
return Style.from_dict(theme_dict)
|
||||
|
||||
@staticmethod
|
||||
def get_ansi_theme(theme_name: str) -> Dict[str, str]:
|
||||
"""Get ANSI color theme for non-prompt_toolkit output"""
|
||||
themes = {
|
||||
"default": {
|
||||
"user": f"{Colors.CYAN}{Colors.BOLD}",
|
||||
"agent": Colors.GREEN,
|
||||
"system": Colors.YELLOW,
|
||||
"error": f"{Colors.RED}{Colors.BOLD}",
|
||||
"info": Colors.BLUE,
|
||||
"muted": Colors.BRIGHT_BLACK,
|
||||
"prompt": f"{Colors.MAGENTA}{Colors.BOLD}",
|
||||
"code": Colors.YELLOW,
|
||||
"reset": Colors.RESET,
|
||||
},
|
||||
"dark": {
|
||||
"user": Colors.BRIGHT_CYAN,
|
||||
"agent": Colors.BRIGHT_GREEN,
|
||||
"system": Colors.BRIGHT_YELLOW,
|
||||
"error": f"{Colors.BRIGHT_RED}{Colors.BOLD}",
|
||||
"info": Colors.BRIGHT_BLUE,
|
||||
"muted": Colors.BRIGHT_BLACK,
|
||||
"prompt": Colors.BRIGHT_MAGENTA,
|
||||
"code": Colors.YELLOW,
|
||||
"reset": Colors.RESET,
|
||||
},
|
||||
"light": {
|
||||
"user": Colors.BLUE,
|
||||
"agent": "\033[32m", # darkgreen
|
||||
"system": "\033[33m", # darkyellow
|
||||
"error": "\033[31m", # darkred
|
||||
"info": "\033[34m", # darkblue
|
||||
"muted": "\033[90m", # brightblack/gray
|
||||
"prompt": "\033[35m", # darkmagenta
|
||||
"code": "\033[33m",
|
||||
"reset": Colors.RESET,
|
||||
},
|
||||
"monokai": {
|
||||
"user": Colors.CYAN,
|
||||
"agent": Colors.GREEN,
|
||||
"system": Colors.YELLOW,
|
||||
"error": Colors.RED,
|
||||
"info": Colors.BLUE,
|
||||
"muted": Colors.BRIGHT_BLACK,
|
||||
"prompt": Colors.MAGENTA,
|
||||
"code": Colors.YELLOW,
|
||||
"reset": Colors.RESET,
|
||||
},
|
||||
}
|
||||
return themes.get(theme_name, themes["default"])
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# SYNTAX HIGHLIGHTING
|
||||
# =============================================================================
|
||||
|
||||
class SyntaxHighlighter:
|
||||
"""Simple syntax highlighter for code blocks"""
|
||||
|
||||
# Language aliases
|
||||
LANG_ALIASES = {
|
||||
"py": "python",
|
||||
"js": "javascript",
|
||||
"ts": "typescript",
|
||||
"rb": "ruby",
|
||||
"sh": "bash",
|
||||
"zsh": "bash",
|
||||
"ps": "powershell",
|
||||
"yml": "yaml",
|
||||
"md": "markdown",
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_lexer(cls, lang: str):
|
||||
"""Get pygments lexer for language"""
|
||||
if not PYGMENTS_AVAILABLE:
|
||||
return None
|
||||
|
||||
lang = cls.LANG_ALIASES.get(lang.lower(), lang.lower())
|
||||
try:
|
||||
return get_lexer_by_name(lang)
|
||||
except:
|
||||
try:
|
||||
return get_lexer_for_filename(f"file.{lang}")
|
||||
except:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def highlight(cls, code: str, lang: str = "text") -> str:
|
||||
"""Apply syntax highlighting to code"""
|
||||
if not PYGMENTS_AVAILABLE:
|
||||
return code
|
||||
|
||||
lexer = cls.get_lexer(lang)
|
||||
if not lexer:
|
||||
return code
|
||||
|
||||
from pygments import highlight as pygments_highlight
|
||||
from pygments.formatters import Terminal256Formatter
|
||||
formatter = Terminal256Formatter()
|
||||
|
||||
try:
|
||||
highlighted = pygments_highlight(code, lexer, formatter)
|
||||
return highlighted.rstrip()
|
||||
except:
|
||||
return code
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# MAIN ENHANCED CLI CLASS
|
||||
# =============================================================================
|
||||
|
||||
class EnhancedCLI:
|
||||
"""
|
||||
Enhanced Command Line Interface for LangCode Agent.
|
||||
|
||||
Features:
|
||||
- Colored input/output with themes
|
||||
- Command history (persistent)
|
||||
- Autocomplete for commands and paths
|
||||
- Multi-line input support
|
||||
- Code syntax highlighting
|
||||
- Mouse support for scrolling
|
||||
- Copy/paste support
|
||||
- Signal handling (Ctrl+C, Ctrl+L)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
history_file: str = "./.langcode/.cli_history",
|
||||
theme: str = "default",
|
||||
welcome_message: Optional[str] = None,
|
||||
show_timestamps: bool = True,
|
||||
enable_mouse: bool = True,
|
||||
enable_syntax_highlight: bool = True,
|
||||
terminal_width: Optional[int] = None,
|
||||
):
|
||||
"""
|
||||
Initialize Enhanced CLI.
|
||||
|
||||
Args:
|
||||
history_file: Path to persist command history
|
||||
theme: Color theme (default, dark, light, monokai)
|
||||
welcome_message: Initial message to show
|
||||
show_timestamps: Show timestamps in output
|
||||
enable_mouse: Enable mouse for scrolling
|
||||
enable_syntax_highlight: Highlight code blocks
|
||||
terminal_width: Force terminal width
|
||||
"""
|
||||
self.history_file = history_file
|
||||
self.theme = theme
|
||||
self.show_timestamps = show_timestamps
|
||||
self.enable_mouse = enable_mouse
|
||||
self.enable_syntax_highlight = enable_syntax_highlight and PYGMENTS_AVAILABLE
|
||||
|
||||
# Get terminal size
|
||||
self.terminal_width = terminal_width or shutil.get_terminal_size().columns
|
||||
|
||||
# ANSI theme for non-prompt output
|
||||
self.ansi_theme = Theme.get_ansi_theme(theme)
|
||||
|
||||
# Welcome message
|
||||
self.welcome_message = welcome_message or self._default_welcome()
|
||||
|
||||
# Ensure history directory exists
|
||||
os.makedirs(os.path.dirname(history_file) or ".", exist_ok=True)
|
||||
|
||||
# Command definitions
|
||||
self.commands = {
|
||||
"/help": "Mostra aiuto",
|
||||
"/status": "Stato sistema",
|
||||
"/clear": "Pulisci schermo",
|
||||
"/theme": "Cicla tra i temi",
|
||||
"/history": "Mostra cronologia",
|
||||
"/theme-default": "Tema default",
|
||||
"/theme-dark": "Tema dark",
|
||||
"/theme-light": "Tema light",
|
||||
"/theme-monokai": "Tema monokai",
|
||||
"/timestamps": "Toggle timestamps",
|
||||
"/syntax": "Toggle syntax highlighting",
|
||||
"/width": "Show terminal width",
|
||||
"/exit": "Esci",
|
||||
"/quit": "Esci",
|
||||
}
|
||||
|
||||
# Path completer for commands that need paths
|
||||
self.path_completer = PathCompleter()
|
||||
|
||||
# Create key bindings
|
||||
self.key_bindings = self._create_key_bindings()
|
||||
|
||||
# Create prompt session
|
||||
self.session = self._create_session()
|
||||
|
||||
# Current theme name
|
||||
self.current_theme = theme
|
||||
|
||||
# Statistics
|
||||
self.stats = {
|
||||
"commands_executed": 0,
|
||||
"session_start": datetime.now(),
|
||||
}
|
||||
|
||||
# Callback for Ctrl+L (clear)
|
||||
self._on_clear: Optional[Callable] = None
|
||||
|
||||
def _default_welcome(self) -> str:
|
||||
"""Generate default welcome message"""
|
||||
theme = self.ansi_theme
|
||||
w = self.terminal_width
|
||||
|
||||
# Box drawing
|
||||
top = f"{theme['prompt']}╔{'═' * (w - 4)}╗{theme['reset']}"
|
||||
bottom = f"{theme['prompt']}╚{'═' * (w - 4)}╝{theme['reset']}"
|
||||
|
||||
lines = [
|
||||
f"{theme['prompt']}║{theme['reset']} {theme['agent']}{Colors.BOLD}LangCode Agent - Enhanced CLI{theme['reset']}" + " " * (w - 43) + f"{theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']}" + " " * (w - 2) + f"{theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} {theme['system']}Comandi:{theme['reset']}" + " " * (w - 17) + f"{theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} {theme['user']}/help{theme['reset']} - Mostra aiuto {theme['user']}/clear{theme['reset']} - Pulisci {theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} {theme['user']}/status{theme['reset']} - Stato sistema {theme['user']}/theme{theme['reset']} - Cambia tema {theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} {theme['user']}/exit{theme['reset']} - Esci {theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']}" + " " * (w - 2) + f"{theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} {theme['info']}Funzionalità:{theme['reset']}" + " " * (w - 22) + f"{theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} • History (↑↓) • Autocomplete (Tab) • Colors {theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']} • Syntax highlighting • Mouse scroll • Multiline {theme['prompt']}║{theme['reset']}",
|
||||
f"{theme['prompt']}║{theme['reset']}" + " " * (w - 2) + f"{theme['prompt']}║{theme['reset']}",
|
||||
]
|
||||
|
||||
return "\n".join([top] + lines + [bottom])
|
||||
|
||||
def _create_key_bindings(self) -> KeyBindings:
|
||||
"""Create custom key bindings"""
|
||||
kb = KeyBindings()
|
||||
|
||||
@kb.add("c-c")
|
||||
def _(event):
|
||||
"""Handle Ctrl+C - interrupt or exit"""
|
||||
# If there's text, clear it; otherwise exit
|
||||
if event.app.current_buffer.text:
|
||||
event.app.current_buffer.text = ""
|
||||
else:
|
||||
event.app.exit(exception=KeyboardInterrupt)
|
||||
|
||||
@kb.add("c-l")
|
||||
def _(event):
|
||||
"""Handle Ctrl+L - clear screen"""
|
||||
os.system("clear" if os.name != "nt" else "cls")
|
||||
if self._on_clear:
|
||||
self._on_clear()
|
||||
|
||||
@kb.add("c-a")
|
||||
def _(event):
|
||||
"""Handle Ctrl+A - go to start of line"""
|
||||
buffer = event.app.current_buffer
|
||||
buffer.cursor_position = 0
|
||||
|
||||
@kb.add("c-e")
|
||||
def _(event):
|
||||
"""Handle Ctrl+E - go to end of line"""
|
||||
buffer = event.app.current_buffer
|
||||
buffer.cursor_position = len(buffer.text)
|
||||
|
||||
@kb.add("c-k")
|
||||
def _(event):
|
||||
"""Handle Ctrl+K - kill line to end"""
|
||||
buffer = event.app.current_buffer
|
||||
buffer.text = buffer.text[:buffer.cursor_position]
|
||||
|
||||
@kb.add("c-u")
|
||||
def _(event):
|
||||
"""Handle Ctrl+U - kill line from start"""
|
||||
buffer = event.app.current_buffer
|
||||
buffer.text = buffer.text[buffer.cursor_position:]
|
||||
buffer.cursor_position = 0
|
||||
|
||||
@kb.add("c-w")
|
||||
def _(event):
|
||||
"""Handle Ctrl+W - kill word before cursor"""
|
||||
buffer = event.app.current_buffer
|
||||
text = buffer.text[:buffer.cursor_position]
|
||||
if text and text[-1] == " ":
|
||||
buffer.text = text[:-1] + buffer.text[buffer.cursor_position:]
|
||||
buffer.cursor_position = len(buffer.text) - (len(buffer.text) - buffer.cursor_position + 1)
|
||||
else:
|
||||
words = text.split()
|
||||
if words:
|
||||
remaining = " ".join(words[:-1]) + (" " if len(words) > 1 else "")
|
||||
buffer.text = remaining + buffer.text[buffer.cursor_position:]
|
||||
buffer.cursor_position = len(remaining)
|
||||
|
||||
# Note: Mouse support is enabled via mouse_support parameter in PromptSession
|
||||
# Scroll events are handled automatically by prompt_toolkit
|
||||
|
||||
return kb
|
||||
|
||||
def _create_session(self) -> PromptSession:
|
||||
"""Create prompt session with history and autocomplete"""
|
||||
|
||||
# Create nested completer for commands
|
||||
command_completer = NestedCompleter.from_nested_dict({
|
||||
cmd: None for cmd in self.commands.keys()
|
||||
})
|
||||
|
||||
return PromptSession(
|
||||
history=FileHistory(self.history_file),
|
||||
completer=command_completer,
|
||||
key_bindings=self.key_bindings,
|
||||
auto_suggest=AutoSuggestFromHistory(),
|
||||
multiline=True,
|
||||
mouse_support=self.enable_mouse,
|
||||
)
|
||||
|
||||
def _get_prompt_tokens(self) -> FormattedText:
|
||||
"""Create prompt tokens with colors and optional info"""
|
||||
theme = self.ansi_theme
|
||||
timestamp = datetime.now().strftime("%H:%M") if self.show_timestamps else None
|
||||
|
||||
tokens = []
|
||||
|
||||
# Prompt symbol with theme
|
||||
tokens.append(("class:prompt", f"{theme['prompt']}❯ {theme['reset']}"))
|
||||
tokens.append(("class:user", f"{theme['user']}TUO{theme['reset']}"))
|
||||
|
||||
if timestamp:
|
||||
tokens.append(("class:muted", f" [{timestamp}]"))
|
||||
|
||||
tokens.append(("class:prompt", f" {theme['prompt']}»{theme['reset']} "))
|
||||
|
||||
return FormattedText(tokens)
|
||||
|
||||
def _print_with_timestamp(self, message: str, prefix: str = "") -> None:
|
||||
"""Print message with optional timestamp"""
|
||||
if self.show_timestamps:
|
||||
ts = datetime.now().strftime("%H:%M:%S")
|
||||
print(f"{self.ansi_theme['muted']}[{ts}]{self.ansi_theme['reset']} {prefix}{message}")
|
||||
else:
|
||||
print(f"{prefix}{message}")
|
||||
|
||||
def _print_formatted(self, message: str, level: str = "info") -> None:
|
||||
"""Print formatted message with color and optional timestamp"""
|
||||
theme = self.ansi_theme
|
||||
colors = {
|
||||
"user": theme["user"],
|
||||
"agent": theme["agent"],
|
||||
"system": theme["system"],
|
||||
"error": theme["error"],
|
||||
"info": theme["info"],
|
||||
}
|
||||
color = colors.get(level, theme["reset"])
|
||||
prefix_map = {
|
||||
"user": f"{color}👤{theme['reset']} ",
|
||||
"agent": f"{color}🤖{theme['reset']} ",
|
||||
"system": f"{color}🔧{theme['reset']} ",
|
||||
"error": f"{color}❌{theme['reset']} ",
|
||||
"info": f"{color}ℹ️{theme['reset']} ",
|
||||
}
|
||||
prefix = prefix_map.get(level, "")
|
||||
self._print_with_timestamp(message, prefix)
|
||||
|
||||
def _print_user(self, message: str) -> None:
|
||||
"""Print user message"""
|
||||
self._print_formatted(message, "user")
|
||||
|
||||
def _print_agent(self, message: str) -> None:
|
||||
"""Print agent response with optional code highlighting"""
|
||||
if self.enable_syntax_highlight:
|
||||
message = self._highlight_code_blocks(message)
|
||||
self._print_formatted(message, "agent")
|
||||
|
||||
def _print_system(self, message: str) -> None:
|
||||
"""Print system message"""
|
||||
self._print_formatted(message, "system")
|
||||
|
||||
def _print_error(self, message: str) -> None:
|
||||
"""Print error message"""
|
||||
self._print_formatted(message, "error")
|
||||
|
||||
def _print_info(self, message: str) -> None:
|
||||
"""Print info message"""
|
||||
self._print_formatted(message, "info")
|
||||
|
||||
def _highlight_code_blocks(self, text: str) -> str:
|
||||
"""Apply syntax highlighting to code blocks in text"""
|
||||
if not self.enable_syntax_highlight:
|
||||
return text
|
||||
|
||||
# Pattern: ```language\ncode\n```
|
||||
code_block_pattern = r"```(\w+)?\n(.*?)```"
|
||||
|
||||
def replace_block(match):
|
||||
lang = match.group(1) or "text"
|
||||
code = match.group(2)
|
||||
|
||||
if PYGMENTS_AVAILABLE and lang.lower() not in ["text", "txt", "ansi"]:
|
||||
try:
|
||||
lexer = SyntaxHighlighter.get_lexer(lang)
|
||||
if lexer:
|
||||
from pygments import highlight as pygments_highlight
|
||||
from pygments.formatters import Terminal256Formatter
|
||||
highlighted = pygments_highlight(code, lexer, Terminal256Formatter())
|
||||
return highlighted.rstrip()
|
||||
except:
|
||||
pass
|
||||
|
||||
# No highlighting available, just return code with a header
|
||||
return f"{self.ansi_theme['code']}--- {lang or 'code'} ---\n{code}{self.ansi_theme['reset']}"
|
||||
|
||||
return re.sub(code_block_pattern, replace_block, text, flags=re.DOTALL)
|
||||
|
||||
def _handle_command(self, command: str) -> Optional[bool]:
|
||||
"""Handle special commands (/command)."""
|
||||
cmd_lower = command.lower().strip()
|
||||
|
||||
if cmd_lower == "/help":
|
||||
self._print_system("""
|
||||
╔══════════════════════════════════════════════════════════════════╗
|
||||
║ COMANDI DISPONIBILI ║
|
||||
╠══════════════════════════════════════════════════════════════════╣
|
||||
║ /help - Mostra questo messaggio ║
|
||||
║ /status - Mostra stato del sistema ║
|
||||
║ /clear, Ctrl+L - Pulisci lo schermo ║
|
||||
║ /theme - Cicla tra i temi ║
|
||||
║ /theme-default - Tema default ║
|
||||
║ /theme-dark - Tema scuro ║
|
||||
║ /theme-light - Tema chiaro ║
|
||||
║ /theme-monokai - Tema monokai ║
|
||||
║ /timestamps - Toggle timestamps ║
|
||||
║ /syntax - Toggle syntax highlighting ║
|
||||
║ /width - Mostra larghezza terminale ║
|
||||
║ /history - Mostra cronologia comandi ║
|
||||
║ /exit, Ctrl+C - Esci dall'agente ║
|
||||
╠══════════════════════════════════════════════════════════════════╣
|
||||
║ SHORTCUTS: ║
|
||||
║ Ctrl+A/E - Inizio/fine riga ║
|
||||
║ Ctrl+K/U - Kill riga da cursore/fino a cursore ║
|
||||
║ Ctrl+W - Kill parola precedente ║
|
||||
║ ↑/↓ - History navigation ║
|
||||
║ Tab - Autocomplete ║
|
||||
║ Shift+Tab - Show suggestions ║
|
||||
╚══════════════════════════════════════════════════════════════════╝
|
||||
""")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/status":
|
||||
uptime = datetime.now() - self.stats["session_start"]
|
||||
self._print_info(f"Theme: {self.current_theme}")
|
||||
self._print_info(f"Timestamps: {'On' if self.show_timestamps else 'Off'}")
|
||||
self._print_info(f"Syntax highlighting: {'On' if self.enable_syntax_highlight else 'Off'}")
|
||||
self._print_info(f"Commands executed: {self.stats['commands_executed']}")
|
||||
self._print_info(f"Session uptime: {uptime}")
|
||||
self._print_info(f"Terminal width: {self.terminal_width}")
|
||||
self._print_info(f"History file: {self.history_file}")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/clear":
|
||||
os.system("clear" if os.name != "nt" else "cls")
|
||||
return None
|
||||
|
||||
elif cmd_lower in ["/theme", "/theme-default"]:
|
||||
self.current_theme = "default"
|
||||
self.ansi_theme = Theme.get_ansi_theme("default")
|
||||
self._print_info("Tema cambiato a: default")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/theme-dark":
|
||||
self.current_theme = "dark"
|
||||
self.ansi_theme = Theme.get_ansi_theme("dark")
|
||||
self._print_info("Tema cambiato a: dark")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/theme-light":
|
||||
self.current_theme = "light"
|
||||
self.ansi_theme = Theme.get_ansi_theme("light")
|
||||
self._print_info("Tema cambiato a: light")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/theme-monokai":
|
||||
self.current_theme = "monokai"
|
||||
self.ansi_theme = Theme.get_ansi_theme("monokai")
|
||||
self._print_info("Tema cambiato a: monokai")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/timestamps":
|
||||
self.show_timestamps = not self.show_timestamps
|
||||
self._print_info(f"Timestamps: {'On' if self.show_timestamps else 'Off'}")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/syntax":
|
||||
self.enable_syntax_highlight = not self.enable_syntax_highlight
|
||||
self._print_info(f"Syntax highlighting: {'On' if self.enable_syntax_highlight else 'Off'}")
|
||||
return None
|
||||
|
||||
elif cmd_lower == "/width":
|
||||
self.terminal_width = shutil.get_terminal_size().columns
|
||||
self._print_info(f"Terminal width: {self.terminal_width}")
|
||||
return None
|
||||
|
||||
elif cmd_lower in ["/exit", "/quit"]:
|
||||
return True
|
||||
|
||||
elif cmd_lower == "/history":
|
||||
try:
|
||||
with open(self.history_file, "r") as f:
|
||||
lines = f.readlines()
|
||||
if lines:
|
||||
self._print_system(f"Ultimi {min(20, len(lines))} comandi:")
|
||||
for i, line in enumerate(lines[-20:], 1):
|
||||
print(f" {i:3d}. {line.strip()}")
|
||||
else:
|
||||
self._print_info("Nessuna cronologia")
|
||||
except FileNotFoundError:
|
||||
self._print_info("Nessuna cronologia")
|
||||
return None
|
||||
|
||||
return None # Command not recognized
|
||||
|
||||
def prompt(self) -> str:
|
||||
"""Get input from user with enhanced UI"""
|
||||
try:
|
||||
text = self.session.prompt(
|
||||
self._get_prompt_tokens(),
|
||||
style=Theme.get_style(self.current_theme),
|
||||
)
|
||||
return text
|
||||
except KeyboardInterrupt:
|
||||
return ""
|
||||
except EOFError:
|
||||
return "/exit"
|
||||
|
||||
def run(self, agent_callback: Callable[[str], str]):
|
||||
"""
|
||||
Run the interactive CLI loop.
|
||||
|
||||
Args:
|
||||
agent_callback: Function that takes user input and returns agent response
|
||||
"""
|
||||
import signal
|
||||
import sys
|
||||
|
||||
# Register signal handler for Ctrl+C
|
||||
def handle_sigint(signum, frame):
|
||||
print("\n\n👋 Arrivederci!")
|
||||
sys.exit(0)
|
||||
|
||||
signal.signal(signal.SIGINT, handle_sigint)
|
||||
|
||||
# Print welcome
|
||||
print(self.welcome_message)
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Get user input
|
||||
user_input = self.prompt().strip()
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
||||
# Show what user typed
|
||||
self._print_user(user_input)
|
||||
self.stats["commands_executed"] += 1
|
||||
|
||||
# Check for commands
|
||||
if user_input.startswith("/"):
|
||||
should_exit = self._handle_command(user_input)
|
||||
if should_exit:
|
||||
break
|
||||
continue
|
||||
|
||||
# Call agent and show response
|
||||
try:
|
||||
response = agent_callback(user_input)
|
||||
self._print_agent(response)
|
||||
except KeyboardInterrupt:
|
||||
self._print_system("\nInterruzione... Riprova.")
|
||||
except Exception as e:
|
||||
self._print_error(f"Errore: {str(e)}")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
self._print_system("\n👋 Arrivederci!")
|
||||
break
|
||||
except EOFError:
|
||||
break
|
||||
|
||||
self._print_system("Sessione terminata.")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FACTORY FUNCTION
|
||||
# =============================================================================
|
||||
|
||||
def create_enhanced_cli(
|
||||
history_file: str = "./.langcode/.cli_history",
|
||||
theme: str = "default",
|
||||
**kwargs
|
||||
) -> EnhancedCLI:
|
||||
"""Create an EnhancedCLI instance"""
|
||||
return EnhancedCLI(
|
||||
history_file=history_file,
|
||||
theme=theme,
|
||||
**kwargs
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# STANDALONE TEST
|
||||
# =============================================================================
|
||||
|
||||
if __name__ == "__main__":
|
||||
def test_callback(text):
|
||||
return f"Hai detto: {text}\n\n```python\nprint('Hello!')\n```"
|
||||
|
||||
cli = create_enhanced_cli(theme="monokai")
|
||||
cli.run(test_callback)
|
||||
@@ -0,0 +1,638 @@
|
||||
"""
|
||||
ReAct Agent - Agent Loop Implementation
|
||||
|
||||
Questo modulo implementa un agente con loop Pensa→Agisci→Osserva→Ripianifica.
|
||||
Ispirato al pattern ReAct (Reasoning + Acting) di Yao et al.
|
||||
|
||||
L'agente:
|
||||
1. THINK: Analizza il goal e decide la prossima azione
|
||||
2. ACT: Esegue l'azione tramite tool
|
||||
3. OBSERVE: Legge il risultato
|
||||
4. EVALUATE: Verifica se il goal è raggiunto
|
||||
5. REPLAN: Se fallito, ripianifica
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import List, Dict, Any, Optional, Callable
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
|
||||
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
|
||||
from langchain_core.tools import Tool
|
||||
|
||||
from enhancements.planning_enhancement import EnhancedPlanner, TaskPriority
|
||||
from enhancements.simple_memory import SimplePersistentMemory, create_simple_memory_tools
|
||||
|
||||
|
||||
class AgentState(Enum):
|
||||
"""Stati dell'agente nel loop"""
|
||||
THINKING = "thinking"
|
||||
ACTING = "acting"
|
||||
OBSERVING = "observing"
|
||||
EVALUATING = "evaluating"
|
||||
REPLANNING = "replanning"
|
||||
DONE = "done"
|
||||
FAILED = "failed"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ActionResult:
|
||||
"""Risultato di un'azione"""
|
||||
action_name: str
|
||||
args: Dict[str, Any]
|
||||
result: str
|
||||
success: bool
|
||||
error: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Step:
|
||||
"""Singolo step nel loop dell'agente"""
|
||||
step_number: int
|
||||
state: AgentState
|
||||
thought: str
|
||||
action: Optional[ActionResult] = None
|
||||
observation: Optional[str] = None
|
||||
timestamp: datetime = field(default_factory=datetime.now)
|
||||
|
||||
|
||||
class ReActAgent:
|
||||
"""
|
||||
Agente con loop ReAct (Reasoning + Acting).
|
||||
|
||||
Esegue un goal attraverso iterazioni di:
|
||||
- THINK: Decide cosa fare
|
||||
- ACT: Esegue l'azione
|
||||
- OBSERVE: Legge il risultato
|
||||
- EVALUATE: Verifica se goal raggiunto
|
||||
- REPLAN: Ripianifica se necessario
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
llm,
|
||||
tools: List[Tool],
|
||||
max_iterations: int = 10,
|
||||
planner: Optional[EnhancedPlanner] = None,
|
||||
memory: Optional[SimplePersistentMemory] = None,
|
||||
verbose: bool = False,
|
||||
include_memory_tools: bool = True
|
||||
):
|
||||
"""
|
||||
Inizializza il ReActAgent.
|
||||
|
||||
Args:
|
||||
llm: Modello LLM da usare
|
||||
tools: Lista di tool disponibili
|
||||
max_iterations: Massimo numero di iterazioni
|
||||
planner: Istanza EnhancedPlanner (creato se None)
|
||||
memory: Istanza SimplePersistentMemory (creato se None)
|
||||
verbose: Se True, stampa debug info
|
||||
include_memory_tools: Se True, aggiunge tool memory (remember, recall, preference)
|
||||
"""
|
||||
self.llm = llm
|
||||
self.max_iterations = max_iterations
|
||||
self.verbose = verbose
|
||||
|
||||
# Planning e memory esistenti
|
||||
self.planner = planner or EnhancedPlanner()
|
||||
self.memory = memory or SimplePersistentMemory()
|
||||
|
||||
# Aggiungi memory tools se richiesto
|
||||
if include_memory_tools:
|
||||
memory_tools = create_simple_memory_tools(self.memory)
|
||||
tools = tools + memory_tools
|
||||
|
||||
self.tools = {tool.name: tool for tool in tools}
|
||||
|
||||
# Stato interno
|
||||
self.steps: List[Step] = []
|
||||
self.current_goal: Optional[str] = None
|
||||
self.context: str = ""
|
||||
self.conversation_history: List[Dict[str, str]] = []
|
||||
|
||||
# System prompt per il loop
|
||||
self._system_prompt = self._build_system_prompt()
|
||||
|
||||
def _build_system_prompt(self) -> str:
|
||||
"""Costruisce il system prompt per il reasoning"""
|
||||
tool_descriptions = []
|
||||
for name, tool in self.tools.items():
|
||||
desc = tool.description or "No description"
|
||||
tool_descriptions.append(f"- {name}: {desc}")
|
||||
|
||||
tools_list = "\n".join(tool_descriptions)
|
||||
|
||||
return f"""Sei un AI Agent che ragiona e agisce per raggiungere obiettivi.
|
||||
|
||||
TOOLS DISPONIBILI:
|
||||
{tools_list}
|
||||
|
||||
MODO DI OPERARE:
|
||||
1. ANALIZZA il goal e il contesto attuale
|
||||
2. DECIDI la prossima azione (usa un tool se necessario)
|
||||
3. Se il goal è raggiunto, rispondi con SUCCESS
|
||||
4. Se serve un'azione, rispondi con: TOOL:<nome_tool>|arg1=val1|arg2=val2
|
||||
5. Se il goal non può essere raggiunto, rispondi con: FAIL:<reason>
|
||||
|
||||
ESEMPI:
|
||||
Goal: Elenca i file nella directory
|
||||
→ TOOL:ls|path=.
|
||||
|
||||
Goal: Leggi il file config.txt
|
||||
→ TOOL:read_file|path=config.txt
|
||||
|
||||
Goal: Mostra info sistema
|
||||
→ TOOL:system_info
|
||||
|
||||
Goal: Crea un file prova.txt con "hello"
|
||||
→ TOOL:write_file|path=prova.txt|content=hello
|
||||
|
||||
Goal: Qual è il contenuto del file?
|
||||
(analizza il contesto - il file è già stato letto)
|
||||
→ SUCCESS: Il file contiene XYZ
|
||||
|
||||
RISPOSTA:"""
|
||||
|
||||
def run(self, goal: str, track_in_planner: bool = True) -> Dict[str, Any]:
|
||||
"""
|
||||
Esegue un goal attraverso il loop ReAct.
|
||||
|
||||
Args:
|
||||
goal: Goal da raggiungere
|
||||
track_in_planner: Se True, traccia il task nel planner
|
||||
|
||||
Returns:
|
||||
Dict con success, result, steps, iterations
|
||||
"""
|
||||
self.current_goal = goal
|
||||
self.context = f"GOAL: {goal}\n"
|
||||
self.steps = []
|
||||
|
||||
# Traccia nel planner se richiesto
|
||||
task_id = None
|
||||
if track_in_planner and self.planner:
|
||||
task_id = self.planner.create_task(
|
||||
description=goal,
|
||||
priority=TaskPriority.MEDIUM
|
||||
)
|
||||
self.planner.start_task(task_id)
|
||||
if self.verbose:
|
||||
print(f"📋 Task creato: {task_id}")
|
||||
|
||||
if self.verbose:
|
||||
print(f"\n🎯 GOAL: {goal}")
|
||||
print("="*60)
|
||||
|
||||
# Esegui prima azione
|
||||
current_input = goal
|
||||
success = False
|
||||
final_result = ""
|
||||
|
||||
for iteration in range(1, self.max_iterations + 1):
|
||||
# 1. THINK - analizza input corrente
|
||||
thought = self._think(current_input)
|
||||
|
||||
if self.verbose:
|
||||
print(f"\n📝 Step {iteration} - THINK:")
|
||||
print(f" {thought[:200]}...")
|
||||
|
||||
# Check per terminazione
|
||||
if thought.strip().upper().startswith("SUCCESS"):
|
||||
result = thought[len("SUCCESS"):].strip()
|
||||
self._add_step(AgentState.DONE, thought)
|
||||
success = True
|
||||
final_result = result
|
||||
break
|
||||
|
||||
if thought.strip().upper().startswith("FAIL"):
|
||||
reason = thought[len("FAIL:"):].strip()
|
||||
self._add_step(AgentState.FAILED, thought)
|
||||
final_result = self.context
|
||||
if task_id:
|
||||
self.planner.fail_task(task_id, reason)
|
||||
return self._build_result(False, self.context, reason)
|
||||
|
||||
# 2. ACT - parse e esegui tool
|
||||
action_result = self._act(thought)
|
||||
self._add_step(AgentState.ACTING, thought, action_result)
|
||||
|
||||
if self.verbose:
|
||||
print(f" 🔧 ACTION: {action_result.action_name}")
|
||||
print(f" 📤 RESULT: {action_result.result[:100]}...")
|
||||
|
||||
# 3. OBSERVE
|
||||
observation = f"[{action_result.action_name}] {action_result.result}"
|
||||
self.context += f"\n{observation}"
|
||||
self._add_step(AgentState.OBSERVING, thought, action_result, observation)
|
||||
|
||||
# Se l'azione era un tool, goal semplice raggiunto
|
||||
if action_result.action_name != "respond":
|
||||
success = True
|
||||
final_result = action_result.result
|
||||
break
|
||||
|
||||
# 4. EVALUATE - solo per risposte dirette
|
||||
self._add_step(AgentState.EVALUATING, thought, action_result, observation)
|
||||
current_input = f"Risultato: {action_result.result}\n\nGoal: {goal}\nGoal raggiunto?"
|
||||
|
||||
# Aggiorna planner
|
||||
if task_id:
|
||||
if success:
|
||||
self.planner.complete_task(task_id, final_result)
|
||||
if self.verbose:
|
||||
print(f"✅ Task {task_id} completato")
|
||||
else:
|
||||
self.planner.fail_task(task_id, "Max iterations reached")
|
||||
|
||||
if success:
|
||||
return self._build_result(
|
||||
True,
|
||||
final_result,
|
||||
f"Goal raggiunto in {iteration} step"
|
||||
)
|
||||
else:
|
||||
return self._build_result(
|
||||
False,
|
||||
self.context,
|
||||
f"Raggiunto massimo numero di iterazioni ({self.max_iterations})"
|
||||
)
|
||||
|
||||
def _think(self, goal: str) -> str:
|
||||
"""LLM decide la prossima azione basata su contesto e goal"""
|
||||
|
||||
# Usa intent detection esistente (già testata e funzionante)
|
||||
from ollama_agent import OllamaToolAgent
|
||||
|
||||
# Crea un agent temporaneo per detect
|
||||
temp_agent = OllamaToolAgent(llm=self.llm, tools=list(self.tools.values()))
|
||||
detected = temp_agent._detect_intent(goal)
|
||||
|
||||
if detected:
|
||||
# Formatta come TOOL:...|
|
||||
args_str = "|".join(f"{k}={v}" for k, v in detected.arguments.items())
|
||||
result = f"TOOL:{detected.name}"
|
||||
if args_str:
|
||||
result += f"|{args_str}"
|
||||
if self.verbose:
|
||||
print(f" [DEBUG] Detected intent: {detected.name}, args: {detected.arguments}")
|
||||
return result
|
||||
|
||||
# Se non c'è intent detection, chiedi al LLM
|
||||
tool_names = ", ".join(self.tools.keys())
|
||||
user_prompt = f"Goal: {goal}\nTool: {tool_names}\nRispondi con TOOL:nome|arg=val oppure SUCCESS:risposta"
|
||||
|
||||
try:
|
||||
messages = [
|
||||
SystemMessage(content=self._system_prompt),
|
||||
HumanMessage(content=user_prompt)
|
||||
]
|
||||
response = self.llm.invoke(messages)
|
||||
if hasattr(response, 'content'):
|
||||
return response.content.strip()
|
||||
return str(response).strip()
|
||||
except Exception as e:
|
||||
return f"FAIL: Errore LLM: {str(e)}"
|
||||
|
||||
def _act(self, thought: str) -> ActionResult:
|
||||
"""Esegue l'azione decisa da _think"""
|
||||
# Parse TOOL:toolname|arg1=val1|...
|
||||
match = re.match(r'^TOOL:(\w+)(?:\|(.+))?$', thought.strip())
|
||||
|
||||
if not match:
|
||||
# Non è un'azione tool valida, ritorna come thought
|
||||
return ActionResult(
|
||||
action_name="respond",
|
||||
args={},
|
||||
result=thought,
|
||||
success=True
|
||||
)
|
||||
|
||||
tool_name = match.group(1)
|
||||
args_str = match.group(2) or "" if match.group(2) is not None else ""
|
||||
|
||||
# Parse args
|
||||
args = {}
|
||||
if args_str:
|
||||
for arg_part in args_str.split("|"):
|
||||
if '=' in arg_part:
|
||||
key, val = arg_part.split('=', 1)
|
||||
args[key.strip()] = val.strip()
|
||||
|
||||
# Esegui tool
|
||||
if tool_name not in self.tools:
|
||||
return ActionResult(
|
||||
action_name=tool_name,
|
||||
args=args,
|
||||
result=f"Tool '{tool_name}' non trovato",
|
||||
success=False,
|
||||
error=f"Unknown tool: {tool_name}"
|
||||
)
|
||||
|
||||
try:
|
||||
tool = self.tools[tool_name]
|
||||
if args:
|
||||
result = tool.invoke(args)
|
||||
else:
|
||||
result = tool.invoke({})
|
||||
|
||||
result_str = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
|
||||
|
||||
return ActionResult(
|
||||
action_name=tool_name,
|
||||
args=args,
|
||||
result=result_str,
|
||||
success=True
|
||||
)
|
||||
except Exception as e:
|
||||
return ActionResult(
|
||||
action_name=tool_name,
|
||||
args=args,
|
||||
result=f"Errore nell'esecuzione: {str(e)}",
|
||||
success=False,
|
||||
error=str(e)
|
||||
)
|
||||
|
||||
def _add_step(
|
||||
self,
|
||||
state: AgentState,
|
||||
thought: str,
|
||||
action: Optional[ActionResult] = None,
|
||||
observation: Optional[str] = None
|
||||
):
|
||||
"""Aggiunge uno step alla cronologia"""
|
||||
step = Step(
|
||||
step_number=len(self.steps) + 1,
|
||||
state=state,
|
||||
thought=thought,
|
||||
action=action,
|
||||
observation=observation
|
||||
)
|
||||
self.steps.append(step)
|
||||
|
||||
def _build_result(self, success: bool, result: str, error: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Costruisce il risultato finale"""
|
||||
return {
|
||||
"success": success,
|
||||
"result": result,
|
||||
"error": error,
|
||||
"iterations": len(self.steps),
|
||||
"steps": [
|
||||
{
|
||||
"step": s.step_number,
|
||||
"state": s.state.value,
|
||||
"thought": s.thought[:200] if s.thought else None,
|
||||
"action": s.action.action_name if s.action else None,
|
||||
"action_result": s.action.result[:100] if s.action and s.action.result else None,
|
||||
"observation": s.observation[:200] if s.observation else None
|
||||
}
|
||||
for s in self.steps
|
||||
]
|
||||
}
|
||||
|
||||
def get_plan_progress(self) -> Dict[str, Any]:
|
||||
"""Restituisce statistiche sullo stato del planner"""
|
||||
return {
|
||||
"total_tasks": len(self.planner.tasks),
|
||||
"plans": len(self.planner.plans)
|
||||
}
|
||||
|
||||
def run_plan(self, goals: List[str], plan_name: str = "multi_step_plan") -> Dict[str, Any]:
|
||||
"""
|
||||
Esegue un piano con più goal in sequenza.
|
||||
|
||||
Args:
|
||||
goals: Lista di goal da eseguire in ordine
|
||||
plan_name: Nome del piano
|
||||
|
||||
Returns:
|
||||
Dict con risultati per ogni goal e progresso complessivo
|
||||
"""
|
||||
# Crea piano nel planner
|
||||
plan_id = self.planner.create_plan(plan_name, goals)
|
||||
|
||||
if self.verbose:
|
||||
print(f"\n📋 PIANO: {plan_name}")
|
||||
print(f" Goal: {len(goals)}")
|
||||
print("="*60)
|
||||
|
||||
results = []
|
||||
completed = 0
|
||||
failed = 0
|
||||
|
||||
for i, goal in enumerate(goals):
|
||||
if self.verbose:
|
||||
print(f"\n--- Goal {i+1}/{len(goals)} ---")
|
||||
|
||||
result = self.run(goal, track_in_planner=True)
|
||||
results.append({
|
||||
"goal": goal,
|
||||
"success": result["success"],
|
||||
"result": result["result"][:200] if result["result"] else None,
|
||||
"error": result.get("error")
|
||||
})
|
||||
|
||||
if result["success"]:
|
||||
completed += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
# Risultato complessivo
|
||||
overall_success = failed == 0
|
||||
|
||||
if self.verbose:
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📊 PIANO COMPLETATO")
|
||||
print(f" Totale: {len(goals)}")
|
||||
print(f" Completati: {completed}")
|
||||
print(f" Falliti: {failed}")
|
||||
|
||||
return {
|
||||
"success": overall_success,
|
||||
"plan_id": plan_id,
|
||||
"total_goals": len(goals),
|
||||
"completed": completed,
|
||||
"failed": failed,
|
||||
"results": results,
|
||||
"plan_progress": self.planner.get_plan_progress(plan_id)
|
||||
}
|
||||
|
||||
def decompose_goal(self, complex_goal: str) -> List[str]:
|
||||
"""
|
||||
Decompone un goal complesso in sotto-goal usando LLM.
|
||||
|
||||
Args:
|
||||
complex_goal: Goal complesso da decomporre
|
||||
|
||||
Returns:
|
||||
Lista di sotto-goal
|
||||
"""
|
||||
prompt = f"""Scomponi questo goal in passi sequenziali semplici.
|
||||
|
||||
Goal complesso: {complex_goal}
|
||||
|
||||
Elenca i passi necessari, uno per riga. Ogni passo deve essere un'azione concreta.
|
||||
Non usare numeri o bullet, solo testo puro.
|
||||
|
||||
ESEMPIO:
|
||||
Goal: Prepara report vendite mensile
|
||||
Output:
|
||||
Leggi i dati di vendita dal database
|
||||
Analizza i dati e calcola totali
|
||||
Genera il report in formato markdown
|
||||
Salva il report come report_mese.md"""
|
||||
|
||||
try:
|
||||
response = self.llm.invoke([HumanMessage(content=prompt)])
|
||||
content = response.content if hasattr(response, 'content') else str(response)
|
||||
|
||||
# Parse output - una linea per passo
|
||||
lines = [l.strip() for l in content.split('\n') if l.strip()]
|
||||
# Filtra linee vuote e possibili header
|
||||
steps = [l for l in lines if len(l) > 10 and not l.startswith('#')]
|
||||
|
||||
if self.verbose:
|
||||
print(f"\n🔬 Goal decomposed into {len(steps)} steps")
|
||||
|
||||
return steps
|
||||
except Exception as e:
|
||||
if self.verbose:
|
||||
print(f"⚠️ Decomposition failed: {e}")
|
||||
# Fallback: ritorna goal originale
|
||||
return [complex_goal]
|
||||
|
||||
def get_relevant_facts(self, query: str) -> List[str]:
|
||||
"""
|
||||
Cerca fatti rilevanti nella memoria.
|
||||
|
||||
Args:
|
||||
query: Query di ricerca
|
||||
|
||||
Returns:
|
||||
Lista di fatti rilevanti
|
||||
"""
|
||||
facts = self.memory.search_facts(query)
|
||||
return [f"{f['fact']} ({f['category']})" for f in facts]
|
||||
|
||||
def save_conversation(self, user_input: str, agent_response: str) -> None:
|
||||
"""
|
||||
Salva una conversazione nella memoria.
|
||||
|
||||
Args:
|
||||
user_input: Input dell'utente
|
||||
agent_response: Risposta dell'agente
|
||||
"""
|
||||
self.memory.add_conversation(user_input, agent_response)
|
||||
|
||||
def remember(self, fact: str, category: str = "general") -> str:
|
||||
"""
|
||||
Ricorda un fatto.
|
||||
|
||||
Args:
|
||||
fact: Il fatto da ricordare
|
||||
category: Categoria del fatto
|
||||
|
||||
Returns:
|
||||
Messaggio di conferma
|
||||
"""
|
||||
self.memory.add_fact(fact, category)
|
||||
return f"Ricordato: {fact}"
|
||||
|
||||
def recall(self, query: str) -> str:
|
||||
"""
|
||||
Cerca fatti nella memoria.
|
||||
|
||||
Args:
|
||||
query: Query di ricerca
|
||||
|
||||
Returns:
|
||||
Risultati della ricerca
|
||||
"""
|
||||
facts = self.memory.search_facts(query)
|
||||
if not facts:
|
||||
return f"Nessun ricordo trovato per: {query}"
|
||||
|
||||
result = f"Trovati {len(facts)} ricordi:\n"
|
||||
for f in facts[:5]:
|
||||
result += f"- {f['fact']} ({f['category']})\n"
|
||||
return result
|
||||
|
||||
def get_memory_stats(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Restituisce statistiche della memoria.
|
||||
|
||||
Returns:
|
||||
Statistiche memoria
|
||||
"""
|
||||
return self.memory.get_memory_stats()
|
||||
|
||||
def add_preference(self, key: str, value: Any) -> str:
|
||||
"""
|
||||
Imposta una preferenza.
|
||||
|
||||
Args:
|
||||
key: Chiave preferenza
|
||||
value: Valore preferenza
|
||||
|
||||
Returns:
|
||||
Messaggio di conferma
|
||||
"""
|
||||
self.memory.set_preference(key, value)
|
||||
return f"Preferenza impostata: {key} = {value}"
|
||||
|
||||
def get_preference(self, key: str) -> str:
|
||||
"""
|
||||
Recupera una preferenza.
|
||||
|
||||
Args:
|
||||
key: Chiave preferenza
|
||||
|
||||
Returns:
|
||||
Valore della preferenza
|
||||
"""
|
||||
val = self.memory.get_preference(key)
|
||||
if val is None:
|
||||
return f"Preferenza '{key}' non trovata"
|
||||
return f"{key} = {val}"
|
||||
|
||||
|
||||
def create_react_agent(
|
||||
llm,
|
||||
tools: Optional[List[Tool]] = None,
|
||||
include_builtin_tools: bool = True,
|
||||
include_memory_tools: bool = True,
|
||||
max_iterations: int = 10,
|
||||
verbose: bool = False
|
||||
) -> ReActAgent:
|
||||
"""
|
||||
Factory function per creare un ReActAgent.
|
||||
|
||||
Args:
|
||||
llm: Modello LLM
|
||||
tools: Lista tool personalizzati
|
||||
include_builtin_tools: Se True, aggiunge tool built-in
|
||||
include_memory_tools: Se True, aggiunge tool memory
|
||||
max_iterations: Massimo iterazioni
|
||||
verbose: Debug output
|
||||
|
||||
Returns:
|
||||
ReActAgent instance
|
||||
"""
|
||||
if tools is None:
|
||||
tools = []
|
||||
|
||||
# Import built-in tools
|
||||
if include_builtin_tools:
|
||||
from ollama_agent import get_builtin_tools
|
||||
builtin = get_builtin_tools()
|
||||
tools = builtin + tools
|
||||
|
||||
return ReActAgent(
|
||||
llm=llm,
|
||||
tools=tools,
|
||||
max_iterations=max_iterations,
|
||||
verbose=verbose,
|
||||
include_memory_tools=include_memory_tools
|
||||
)
|
||||
+189
-17
@@ -29,11 +29,44 @@ 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:
|
||||
@@ -53,34 +86,171 @@ def cmd_chat(args):
|
||||
print(" /exit - Esci dall'agente")
|
||||
print("\n💡 Per chattare, scrivi semplicemente il tuo messaggio.")
|
||||
|
||||
while True:
|
||||
try:
|
||||
user_input = input("\n👤 You: ").strip()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\n👋 Arrivederci!")
|
||||
break
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
||||
# 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):
|
||||
continue
|
||||
|
||||
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:
|
||||
response_text = result["response"]
|
||||
return result["response"]
|
||||
else:
|
||||
ai_message = result["messages"][-1]
|
||||
response_text = ai_message.content if hasattr(ai_message, 'content') else str(ai_message)
|
||||
|
||||
print(f"\n🤖 LangCode: {response_text}")
|
||||
return ai_message.content if hasattr(ai_message, 'content') else str(ai_message)
|
||||
except Exception as e:
|
||||
print(f"\n❌ Errore: {str(e)}")
|
||||
return f"Errore: {str(e)}"
|
||||
|
||||
# Default: CLI semplice, EnhancedCLI/TerminalUI con flag
|
||||
use_enhanced = getattr(args, 'enhanced', False)
|
||||
use_terminal = getattr(args, 'terminal', False)
|
||||
|
||||
if use_terminal:
|
||||
try:
|
||||
from terminal_ui import create_terminal_ui
|
||||
term = create_terminal_ui(
|
||||
scrollback=10000,
|
||||
prompt="❯ "
|
||||
)
|
||||
term.start(agent_callback)
|
||||
return
|
||||
except ImportError:
|
||||
print("⚠️ pyte non installato. Esegui: pip install pyte")
|
||||
print("📝 Uso CLI semplice...")
|
||||
except Exception as e:
|
||||
print(f"⚠️ TerminalUI errore: {e}")
|
||||
print("📝 Uso CLI semplice...")
|
||||
|
||||
if use_enhanced:
|
||||
import errno
|
||||
import fcntl
|
||||
import select
|
||||
import threading
|
||||
|
||||
def timeout_handler():
|
||||
print("\n⚠️ EnhancedCLI timeout - uso CLI semplice")
|
||||
|
||||
try:
|
||||
from enhanced_cli import create_enhanced_cli
|
||||
cli = create_enhanced_cli(
|
||||
history_file="./.langcode/.cli_history",
|
||||
theme="default"
|
||||
)
|
||||
|
||||
# Run with timeout detection
|
||||
timeout_seconds = 30
|
||||
timer = threading.Timer(timeout_seconds, timeout_handler)
|
||||
timer.daemon = True
|
||||
timer.start()
|
||||
|
||||
cli.run(agent_callback)
|
||||
timer.cancel()
|
||||
return
|
||||
except (KeyboardInterrupt, EOFError, OSError) as e:
|
||||
# EnhancedCLI hang - fall back to simple
|
||||
print(f"⚠️ EnhancedCLI bloccato: {e}")
|
||||
print("📝 Uso CLI semplice...")
|
||||
except Exception as e:
|
||||
print(f"⚠️ EnhancedCLI errore: {e}")
|
||||
print("📝 Uso CLI semplice...")
|
||||
|
||||
# CLI semplice con input()
|
||||
# Reset terminal state dopo eventuale EnhancedCLI fallito
|
||||
import subprocess
|
||||
try:
|
||||
subprocess.run(["stty", "sane"], check=False, capture_output=True)
|
||||
except:
|
||||
pass
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print(" CLI SEMPLICE - digita /exit per uscire")
|
||||
print("=" * 60 + "\n")
|
||||
|
||||
while True:
|
||||
try:
|
||||
sys.stdout.flush()
|
||||
sys.stdout.write("❯ ")
|
||||
sys.stdout.flush()
|
||||
user_input = sys.stdin.readline().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)
|
||||
|
||||
# Output con scrollback via less
|
||||
lines = clean_response.split('\n')
|
||||
if len(lines) > 10:
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
output_file = f".langcode/output_{timestamp}.txt"
|
||||
|
||||
# Save to file
|
||||
os.makedirs(".langcode", exist_ok=True)
|
||||
with open(output_file, "w") as f:
|
||||
f.write(clean_response)
|
||||
|
||||
print(f"🤖 Output lungo salvato in: {output_file}")
|
||||
print("💡 Visualizza con: less {0} (usa frecce per scrollare, q per uscire)\n".format(output_file))
|
||||
|
||||
# Try to show with less
|
||||
import subprocess
|
||||
try:
|
||||
subprocess.run(["less", "-R", output_file], check=False)
|
||||
except FileNotFoundError:
|
||||
# less not available, just cat first part
|
||||
print("🤖 " + "\n🤖 ".join(lines[:20]))
|
||||
if len(lines) > 20:
|
||||
print(f"\n... (altre {len(lines)-20} righe nel file {output_file})")
|
||||
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)}")
|
||||
@@ -457,6 +627,8 @@ Esempi:
|
||||
|
||||
# === CHAT (default) ===
|
||||
chat_parser = subparsers.add_parser("chat", help="Avvia l'agent in modalità chat")
|
||||
chat_parser.add_argument("--enhanced", "-e", action="store_true", help="Usa EnhancedCLI (sperimentale)")
|
||||
chat_parser.add_argument("--terminal", "-t", action="store_true", help="Usa TerminalUI con scrollback (pyte)")
|
||||
chat_parser.set_defaults(func=cmd_chat)
|
||||
|
||||
# === SERVER ===
|
||||
|
||||
+31
-14
@@ -342,12 +342,13 @@ For other questions, just answer normally without tools."""
|
||||
file_path = path_match.group(1) if path_match else None
|
||||
|
||||
# Execute command patterns - check FIRST for shell commands with flags
|
||||
# Shell commands like "ls -lha", "git status" should use execute
|
||||
if ('ls' in text_lower or 'git' in text_lower or 'cat' in text_lower or
|
||||
'grep' in text_lower or 'find' in text_lower or 'pwd' in text_lower or
|
||||
'cd ' in text_lower or 'mkdir' in text_lower or 'rm ' in text_lower or
|
||||
'cp ' in text_lower or 'mv ' in text_lower or 'chmod' in text_lower) and ' ' in text:
|
||||
return ToolCall(id='intent', name='execute', arguments={'command': text.strip()})
|
||||
# Shell commands like "ls -lha", "git status", "pwd", "echo" should use execute
|
||||
shell_commands = ['ls', 'git', 'cat', 'grep', 'find', 'pwd', 'echo', 'cd',
|
||||
'mkdir', 'rm', 'cp', 'mv', 'chmod', 'head', 'tail', 'sort', 'uniq']
|
||||
# Check for command alone or with arguments
|
||||
for cmd in shell_commands:
|
||||
if text_lower.strip() == cmd or text_lower.startswith(cmd + ' '):
|
||||
return ToolCall(id='intent', name='execute', arguments={'command': text.strip()})
|
||||
|
||||
# System info patterns (more comprehensive)
|
||||
system_patterns = [
|
||||
@@ -400,17 +401,29 @@ For other questions, just answer normally without tools."""
|
||||
has_write_keyword = any(k in text_lower for k in write_keywords)
|
||||
|
||||
if has_write_keyword and file_path:
|
||||
# Extract content from quoted text
|
||||
content_match = re.search(r'["\']([^"\']+)["\']', text)
|
||||
# Extract content - try multiple patterns
|
||||
content = None
|
||||
|
||||
# Pattern 1: "con dentro 'testo'" o "con 'testo'"
|
||||
content_match = re.search(r'con\s+(?:dentro\s+)?["\']([^"\']+)["\']', text, re.IGNORECASE)
|
||||
if content_match:
|
||||
content = content_match.group(1)
|
||||
else:
|
||||
# Try to extract content between "con dentro" and quotes
|
||||
content_match = re.search(r'con dentro ["\']([^"\']+)["\']', text)
|
||||
|
||||
# Pattern 2: testo tra virgolette anywhere
|
||||
if not content:
|
||||
content_match = re.search(r'["\']([^"\']+)["\']', text)
|
||||
if content_match:
|
||||
content = content_match.group(1)
|
||||
else:
|
||||
content = "Content not specified"
|
||||
|
||||
# Pattern 3: contenuto dopo "con" fino alla fine (senza virgolette)
|
||||
if not content:
|
||||
# Match: con <resto della frase senza virgolette iniziali/finali>
|
||||
after_con = re.search(r'con\s+["\']?(.+?)["\']?\s*$', text, re.IGNORECASE | re.DOTALL)
|
||||
if after_con:
|
||||
content = after_con.group(1).strip()
|
||||
|
||||
if not content:
|
||||
content = "Content not specified"
|
||||
|
||||
return ToolCall(id='intent', name='write_file', arguments={'path': file_path, 'content': content})
|
||||
|
||||
@@ -433,7 +446,11 @@ For other questions, just answer normally without tools."""
|
||||
"""
|
||||
# Estrai il messaggio utente
|
||||
user_message = input_dict.get("messages", [])[-1]
|
||||
if hasattr(user_message, 'content'):
|
||||
|
||||
# Gestisci diversi formati di messaggio
|
||||
if isinstance(user_message, dict):
|
||||
user_content = user_message.get("content", str(user_message))
|
||||
elif hasattr(user_message, 'content'):
|
||||
user_content = user_message.content
|
||||
else:
|
||||
user_content = str(user_message)
|
||||
|
||||
+224
@@ -0,0 +1,224 @@
|
||||
"""
|
||||
Terminal UI con scrollback semplice
|
||||
Usa input() standard + visualizzazione scrollback
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import signal
|
||||
from typing import Optional, Callable
|
||||
|
||||
|
||||
class TerminalUI:
|
||||
"""
|
||||
Terminal UI con scrollback.
|
||||
Uso semplice: input() per input, history buffer per scrollback.
|
||||
"""
|
||||
|
||||
def __init__(self, scrollback: int = 10000, prompt: str = "❯ "):
|
||||
self.scrollback = scrollback
|
||||
self.prompt = prompt
|
||||
|
||||
# History buffer
|
||||
self.history: list = []
|
||||
|
||||
# Callback
|
||||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||||
|
||||
def _print(self, text: str):
|
||||
"""Stampa con colori base"""
|
||||
# Semplice - solo reset e bold
|
||||
if text.startswith("👤"):
|
||||
print(f"\033[1;36m{text}\033[0m")
|
||||
elif text.startswith("🤖"):
|
||||
print(f"\033[1;32m{text}\033[0m")
|
||||
elif text.startswith("🤔"):
|
||||
print(f"\033[1;33m{text}\033[0m")
|
||||
elif text.startswith("❌"):
|
||||
print(f"\033[1;31m{text}\033[0m")
|
||||
else:
|
||||
print(text)
|
||||
|
||||
def _wrap_line(self, text: str, width: int = 80) -> list:
|
||||
"""Wrap linea a width"""
|
||||
lines = []
|
||||
while len(text) > width:
|
||||
lines.append(text[:width])
|
||||
text = text[width:]
|
||||
if text:
|
||||
lines.append(text)
|
||||
return lines
|
||||
|
||||
def _add_history(self, text: str):
|
||||
"""Aggiunge a history"""
|
||||
self.history.append(text)
|
||||
while len(self.history) > self.scrollback:
|
||||
self.history.pop(0)
|
||||
|
||||
def _show_history(self, lines: int = None):
|
||||
"""Mostra ultime N linee di history"""
|
||||
if lines is None:
|
||||
# Determina righe visibili
|
||||
try:
|
||||
size = os.get_terminal_size()
|
||||
lines = size.lines - 5 # spazio per header, status, prompt
|
||||
except:
|
||||
lines = 20
|
||||
|
||||
# Mostra ultime righe
|
||||
start = max(0, len(self.history) - lines)
|
||||
for line in self.history[start:]:
|
||||
self._print(line)
|
||||
|
||||
def _process(self, cmd: str):
|
||||
"""Processa comando"""
|
||||
if not cmd.strip():
|
||||
return
|
||||
|
||||
self._add_history(f"👤 {cmd}")
|
||||
|
||||
# Thinking
|
||||
print("\033[1;33m🤔 Pensando...\033[0m")
|
||||
|
||||
try:
|
||||
resp = self.agent_callback(cmd)
|
||||
|
||||
# Rimuovi thinking (print sopra con risposta)
|
||||
print("\033[2A\033[K", end="") # move up 2, clear line
|
||||
|
||||
# Stampa risposta
|
||||
for line in resp.split('\n'):
|
||||
if line.strip():
|
||||
self._add_history(f"🤖 {line}")
|
||||
self._print(f"🤖 {line}")
|
||||
|
||||
except Exception as e:
|
||||
print("\033[2A\033[K", end="")
|
||||
err_msg = f"❌ {str(e)}"
|
||||
self._add_history(err_msg)
|
||||
self._print(err_msg)
|
||||
|
||||
def _scroll_help(self):
|
||||
"""Mostra aiuto scroll"""
|
||||
print("\n" + "=" * 60)
|
||||
print("SCROLLBACK - Premi q per uscire, frecce per navigare")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
# Paginator semplice
|
||||
page_size = 30
|
||||
start = 0
|
||||
total = len(self.history)
|
||||
|
||||
while start < total:
|
||||
# Mostra pagina
|
||||
for i in range(start, min(start + page_size, total)):
|
||||
self._print(self.history[i])
|
||||
|
||||
start += page_size
|
||||
|
||||
if start < total:
|
||||
print()
|
||||
print(f"[{start}/{total} righe] - Premi ENTER per continuare, q per uscire")
|
||||
try:
|
||||
choice = input().strip()
|
||||
if choice.lower() == 'q':
|
||||
break
|
||||
except:
|
||||
break
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("FINE SCROLLBACK")
|
||||
print("=" * 60)
|
||||
|
||||
def start(self, agent_callback: Callable[[str], str]):
|
||||
"""Avvia UI"""
|
||||
self.agent_callback = agent_callback
|
||||
|
||||
# Signal handler
|
||||
def sigint_handler(signum, frame):
|
||||
print("\n\033[1;33m👋 Arrivederci!\033[0m\n")
|
||||
sys.exit(0)
|
||||
|
||||
signal.signal(signal.SIGINT, sigint_handler)
|
||||
|
||||
# Header
|
||||
print()
|
||||
print("\033[1;36m" + "=" * 50 + "\033[0m")
|
||||
print("\033[1;36m=\033[0m" + " " * 15 + "\033[1;33mLangCode Agent\033[0m" + " " * 15 + "\033[1;36m=\033[0m")
|
||||
print("\033[1;36m" + "=" * 50 + "\033[0m")
|
||||
print()
|
||||
print("digita /help per aiuto, /exit per uscire")
|
||||
print()
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Input
|
||||
user_input = input(f"\033[1;35m{self.prompt}\033[0m ").strip()
|
||||
except EOFError:
|
||||
break
|
||||
except KeyboardInterrupt:
|
||||
print("\n\033[1;33m👋 Arrivederci!\033[0m")
|
||||
break
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
|
||||
# Comandi speciali
|
||||
if user_input.lower() == '/exit':
|
||||
print("\033[1;33m👋 Arrivederci!\033[0m")
|
||||
break
|
||||
|
||||
elif user_input.lower() == '/quit':
|
||||
print("\033[1;33m👋 Arrivederci!\033[0m")
|
||||
break
|
||||
|
||||
elif user_input.lower() == '/help':
|
||||
print()
|
||||
print("\033[1;36m" + "=" * 40 + "\033[0m")
|
||||
print("COMANDI:")
|
||||
print(" /help - Questo messaggio")
|
||||
print(" /clear - Pulisci schermo")
|
||||
print(" /history - Mostra cronologia")
|
||||
print(" /scroll - Entra in scrollback")
|
||||
print(" /exit, /quit - Esci")
|
||||
print()
|
||||
print("SCROLLBACK:")
|
||||
print(" /scroll - Mostra tutto lo scrollback")
|
||||
print(" Frecce su/giù - Naviga cronologia (in some terminals)")
|
||||
print()
|
||||
print("\033[1;36m" + "=" * 40 + "\033[0m")
|
||||
print()
|
||||
continue
|
||||
|
||||
elif user_input.lower() == '/clear':
|
||||
os.system('clear' if os.name != 'nt' else 'cls')
|
||||
continue
|
||||
|
||||
elif user_input.lower() == '/history':
|
||||
print()
|
||||
self._show_history(50)
|
||||
print()
|
||||
continue
|
||||
|
||||
elif user_input.lower() == '/scroll':
|
||||
self._scroll_help()
|
||||
continue
|
||||
|
||||
# Processa comando normale
|
||||
self._process(user_input)
|
||||
print()
|
||||
|
||||
|
||||
def create_terminal_ui(scrollback: int = 10000, prompt: str = "❯ ") -> TerminalUI:
|
||||
return TerminalUI(scrollback=scrollback, prompt=prompt)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
def test(text):
|
||||
lines = [f"Ricevuto: {text}"]
|
||||
lines += [f"Linea {i}: contenuto di esempio" for i in range(1, 31)]
|
||||
return "\n".join(lines)
|
||||
|
||||
ui = create_terminal_ui()
|
||||
ui.start(test)
|
||||
Reference in new issue
Block a user