feat: aggiungi ReAct Agent Loop per trasformare chatbot in agente
- Nuovo ReActAgent con loop Pensa->Agisci->Osserva - Usa intent detection esistente per affidabilità - Supporta tutti i tool built-in - Planning e memory integration point - docs/AGENT_ROADMAP.md: gap analysis e roadmap
This commit is contained in:
opencode committed
1 parent
e5ce70c68e
commit
1c7781a68f
2 files changed
+625
No files matched your search
@@ -0,0 +1,398 @@
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
):
|
||||
"""
|
||||
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
|
||||
"""
|
||||
self.llm = llm
|
||||
self.tools = {tool.name: tool for tool in tools}
|
||||
self.max_iterations = max_iterations
|
||||
self.verbose = verbose
|
||||
|
||||
# Planning e memory esistenti
|
||||
self.planner = planner or EnhancedPlanner()
|
||||
self.memory = memory or SimplePersistentMemory()
|
||||
|
||||
# Stato interno
|
||||
self.steps: List[Step] = []
|
||||
self.current_goal: Optional[str] = None
|
||||
self.context: 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) -> Dict[str, Any]:
|
||||
"""
|
||||
Esegue un goal attraverso il loop ReAct.
|
||||
|
||||
Args:
|
||||
goal: Goal da raggiungere
|
||||
|
||||
Returns:
|
||||
Dict con success, result, steps, iterations
|
||||
"""
|
||||
self.current_goal = goal
|
||||
self.context = f"GOAL: {goal}\n"
|
||||
self.steps = []
|
||||
|
||||
if self.verbose:
|
||||
print(f"\n🎯 GOAL: {goal}")
|
||||
print("="*60)
|
||||
|
||||
# Esegui prima azione
|
||||
current_input = goal
|
||||
|
||||
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)
|
||||
return self._build_result(True, result, f"Goal raggiunto in {iteration} step")
|
||||
|
||||
if thought.strip().upper().startswith("FAIL"):
|
||||
reason = thought[len("FAIL:"):].strip()
|
||||
self._add_step(AgentState.FAILED, thought)
|
||||
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, verifica se il goal è raggiunto
|
||||
if action_result.action_name != "respond":
|
||||
# Goal semplice raggiunto - tool ha dato la risposta
|
||||
# Restituisci SUCCESS con il risultato
|
||||
self._add_step(AgentState.DONE, thought, action_result)
|
||||
return self._build_result(
|
||||
True,
|
||||
action_result.result,
|
||||
f"Goal raggiunto in {iteration} step con tool {action_result.action_name}"
|
||||
)
|
||||
|
||||
# 4. EVALUATE - solo per risposte dirette
|
||||
self._add_step(AgentState.EVALUATING, thought, action_result, observation)
|
||||
|
||||
# Max iterations reached
|
||||
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 create_react_agent(
|
||||
llm,
|
||||
tools: Optional[List[Tool]] = None,
|
||||
include_builtin_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
|
||||
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
|
||||
)
|
||||
Reference in new issue
Block a user