Gli agent non sono chat: architetture autonome che fanno più di quanto chiedi
Un agent vero è un LLM che usa tool, mantiene stato, fallisce, e rincorre l'obiettivo. Se scrivi "ciao" e risponde "ciao", non è un agent - è un chatbot con un'API costosa. La distinzione sembra banale finché non devi scegliere l'architettura giusta per un sistema che deve fare davvero qualcosa: prenotare un viaggio, scrivere e testare codice, analizzare 50 documenti e produrre un report strutturato.
Nel 2025, il termine "agent AI" è diventato marketing catch-all: ogni chatbot con un tool allegato si auto-proclama agente autonomo. Questa serie nasce proprio per tagliare il rumore. In questo primo episodio stabiliamo le definizioni rigorose, esploriamo i quattro pattern architetturali fondamentali con implementazioni Python reali, guardiamo i case study di Devin, Cursor Composer e Claude Code, e mettiamo in chiaro il cost profile di un agent rispetto a una semplice chiamata one-shot. Perché gli agent sono potenti, ma costano - e ignorare quanto costano è il modo più rapido per bruciare budget in produzione.
Cosa Imparerai
- La definizione tecnica rigorosa di agent LLM: autonomia, persistenza, tool use
- Differenza concreta tra agent, chatbot e sistema RAG
- Implementazione del loop ReAct da zero in Python (nessun framework)
- Pattern Plan-Execute per task long-horizon con esecuzione parallela
- Reflexion: auto-critica verbale senza trainer esterno
- Tree of Thoughts: quando vale il costo computazionale
- Persistenza dello stato agent su Redis tra sessioni diverse
- Stima del costo reale agent vs one-shot e come controllarlo
Prerequisiti
- Python 3.11+ e familiarità con chiamate a API LLM (OpenAI-compatible o Ollama)
- Conoscenza di REST API e pattern async/await
- Nozioni base di prompt engineering (system prompt, few-shot)
Sezione 1 - Definizione rigorosa: cosa rende un sistema un "agent"
La letteratura accademica e ingegneristica convergono su tre proprietà necessarie e sufficienti per classificare un sistema come agent LLM. Un sistema che ne manca anche solo una è qualcosa di diverso - e trattarlo come agent porta a errori architetturali costosi.
La prima proprietà è l'autonomia decisionale: l'agent decide i propri passi successivi senza che l'utente specifichi ogni azione. Riceve un obiettivo ad alto livello ("analizza questi contratti e segnala le clausole rischiose") e pianifica da solo la sequenza di operazioni necessarie. Un chatbot classico - anche con RAG - non decide nulla: risponde al prompt e si ferma.
La seconda proprietà è la persistenza dello stato: l'agent mantiene memoria dell'obiettivo corrente, dei passi già eseguiti, degli errori incontrati, e del contesto accumulato attraverso più iterazioni o addirittura sessioni separate. Questa proprietà è ciò che distingue un agent da un pipeline one-shot: non è il numero di chiamate LLM, ma il fatto che ogni chiamata sia informata da ciò che è successo prima.
La terza proprietà è il tool use: l'agent può modificare il mondo esterno. Legge file, fa API call, scrive nel database, esegue codice, naviga pagine web. Senza tool, un LLM che "ragiona" in loop è solo un chain-of-thought glorificato - elabora testo, non cambia stato. Il tool use trasforma il modello da oracolo a attore.
| Proprietà | Chatbot | RAG System | Agent LLM | Esempio concreto (Agent) |
|---|---|---|---|---|
| Autonomia | No - risponde e si ferma | No - recupera e genera | Sì - decide i next step | Sceglie da solo di cercare documentazione prima di scrivere codice |
| Persistenza | No - stateless per request | Parziale - context window | Sì - stato tra iterazioni e sessioni | Ricorda che la build è fallita 3 volte e cambia strategia |
| Tool Use | No - genera testo | Parziale - solo retrieval | Sì - modifica stato esterno | Esegue i test, legge l'output, corregge il codice |
Il confronto in codice è il modo più diretto per vedere la differenza. Ecco il pattern che distingue un chatbot da un agent nella sua forma minima:
# Snippet 1 - Chatbot vs Agent: confronto architetturale minimo
import json
from typing import Callable
# --- CHATBOT: risponde e si ferma ---
def chatbot(user_message: str, llm_call: Callable) -> str:
"""
Un chatbot classico: 1 input → 1 output → fine.
Non ha stato, non usa tool, non itera.
"""
prompt = f"Rispondi a questa domanda: {user_message}"
return llm_call(prompt)
# --- AGENT: loop con tool, stato, obiettivo persistente ---
class SimpleAgent:
"""
Agent minimale con le 3 proprietà fondamentali:
- autonomia: decide i next step nel loop
- persistenza: mantiene history e stato obiettivo
- tool use: chiama tool_registry per modificare il mondo
"""
def __init__(self, llm_call: Callable, tool_registry: dict):
self.llm_call = llm_call
self.tools = tool_registry
# PERSISTENZA: stato tra iterazioni
self.history: list[dict] = []
self.current_goal: str = ""
self.max_iterations = 10
def run(self, goal: str) -> str:
self.current_goal = goal
self.history = []
iterations = 0
while iterations < self.max_iterations:
iterations += 1
# AUTONOMIA: l'LLM decide cosa fare dopo
action = self._decide_next_action()
if action["type"] == "final_answer":
return action["content"]
# TOOL USE: esegue azione nel mondo reale
observation = self._execute_tool(action)
self.history.append({
"action": action,
"observation": observation
})
return "Limite iterazioni raggiunto senza risposta definitiva."
def _decide_next_action(self) -> dict:
context = json.dumps(self.history[-3:]) # ultimi 3 step
prompt = f"""
Obiettivo: {self.current_goal}
Storia recente: {context}
Tool disponibili: {list(self.tools.keys())}
Decidi il prossimo passo. Rispondi in JSON:
{{"type": "tool_call"|"final_answer", "tool": "nome_tool", "args": {{}}, "content": "..."}}
"""
response = self.llm_call(prompt)
return json.loads(response)
def _execute_tool(self, action: dict) -> str:
tool_name = action.get("tool")
if tool_name not in self.tools:
return f"Errore: tool '{tool_name}' non trovato"
tool_fn = self.tools[tool_name]
return str(tool_fn(**action.get("args", {})))
Nota la differenza strutturale: il chatbot è una funzione, l'agent è un sistema con stato interno, un loop di controllo, e la capacità di invocare azioni nel mondo esterno. Anche in questa forma minimale, l'agent può bloccarsi (loop infinito), spendere troppo (context che cresce), o perdere il filo dell'obiettivo - problemi che il chatbot non ha semplicemente perché non persiste.
Sezione 2 - Quattro pattern architetturali fondamentali
Esistono quattro pattern architetturali che coprono la quasi totalità degli agent LLM in produzione. Non sono mutualmente esclusivi: sistemi complessi li combinano. Ma capire ogni pattern in isolamento è prerequisito per combinarli correttamente.
| Pattern | Meccanismo | Costo relativo | Latenza | Use case ideale |
|---|---|---|---|---|
| ReAct | Thought → Action → Observation loop | Medio (context cresce linearmente) | Alta (sequenziale) | Task esplorativi, debugging, ricerca |
| Plan-Execute | Planner genera DAG, Executor parallelizza | Basso per step (modelli piccoli) | Bassa (parallelismo) | Task strutturati, pipeline multi-step |
| Reflexion | Actor + Evaluator + auto-critica verbale | Alto (2-3x token extra) | Alta (cicli aggiuntivi) | Task ad alta precisione, output verificabile |
| Tree of Thoughts | Esplora N branch, beam search su pensieri | Molto alto (N chiamate parallele) | Media (parallelismo branch) | Puzzle, ottimizzazione, planning formale |
Sezione 2a - ReAct: il pattern fondamentale
ReAct (Reasoning + Acting), introdotto da Yao et al. nel 2022, è il pattern più diffuso in produzione. L'LLM alterna pensiero esplicito (Thought) e azione (Action), osserva il risultato (Observation), e decide se ha raggiunto l'obiettivo o deve continuare. La semplicità del loop nasconde la sua potenza: il pensiero esplicito riduce le allucinazioni perché costringe il modello a ragionare passo per passo prima di agire.
# Snippet 2 - ReAct loop implementation
import re
from typing import Callable
REACT_SYSTEM_PROMPT = """Sei un agent che ragiona passo per passo prima di agire.
Formato obbligatorio per ogni turno:
Thought: [ragionamento su cosa fare e perché]
Action: [nome_tool]
Action Input: [parametri JSON per il tool]
Quando hai la risposta finale:
Thought: [conclusione]
Final Answer: [risposta completa all'utente]
Tool disponibili:
{tools_description}
"""
def react_agent(
goal: str,
tools: dict[str, Callable],
llm_call: Callable,
max_steps: int = 15
) -> str:
"""
Implementazione ReAct pura senza framework.
Ogni step: Thought → Action → Observation.
Il context cresce ad ogni iterazione - monitor i token.
"""
# Costruzione descrizione tool per il prompt
tools_desc = "\n".join(
f"- {name}: {fn.__doc__ or 'nessuna descrizione'}"
for name, fn in tools.items()
)
system = REACT_SYSTEM_PROMPT.format(tools_description=tools_desc)
# Conversazione accumulata (il "nastro" del ReAct)
messages = [
{"role": "system", "content": system},
{"role": "user", "content": f"Obiettivo: {goal}"}
]
for step in range(max_steps):
response = llm_call(messages)
messages.append({"role": "assistant", "content": response})
# Parser dell'output strutturato
if "Final Answer:" in response:
return response.split("Final Answer:")[-1].strip()
# Estrai Action e Action Input
action_match = re.search(r"Action: (.+)", response)
input_match = re.search(r"Action Input: (.+)", response, re.DOTALL)
if not action_match:
# L'LLM ha ragionato senza agire - forziamo un'azione
messages.append({
"role": "user",
"content": "Osservazione: Nessuna azione eseguita. Specifica un'azione o dai la risposta finale."
})
continue
tool_name = action_match.group(1).strip()
tool_input_raw = input_match.group(1).strip() if input_match else "{}"
# Esecuzione tool con gestione errori strutturata
try:
import json
tool_args = json.loads(tool_input_raw)
if tool_name not in tools:
observation = f"Errore: tool '{tool_name}' non esiste. Tool disponibili: {list(tools.keys())}"
else:
observation = str(tools[tool_name](**tool_args))
except json.JSONDecodeError:
observation = f"Errore: Action Input non è JSON valido: {tool_input_raw}"
except Exception as exc:
observation = f"Errore esecuzione tool: {exc}"
messages.append({
"role": "user",
"content": f"Observation: {observation}"
})
return "Limite di step raggiunto senza risposta definitiva."
# Esempio di tool per il ReAct
def cerca_documentazione(query: str) -> str:
"""Cerca nella documentazione tecnica interna. Args: query (str)."""
# In produzione: RAG call al ml-service
return f"Risultati ricerca per '{query}': [doc1, doc2, doc3]"
def esegui_codice(code: str) -> str:
"""Esegue uno snippet Python in sandbox sicura. Args: code (str)."""
# In produzione: subprocess con timeout e seccomp
return f"Output: [risultato esecuzione]"
Sezione 2b - Plan-Execute: separare strategia e tattica
ReAct soffre su task long-horizon: il context cresce, l'obiettivo originale si "diluisce" nella storia accumulata, e il modello tende a perdere il filo. Plan-Execute risolve questo separando nettamente due fasi. Un Planner (modello grande, eseguito una volta) genera un piano strutturato - idealmente un DAG di subtask con dipendenze esplicite. Gli Executor (modelli più piccoli, eseguiti in parallelo) lavorano su ciascun subtask in isolamento, con un context pulito e limitato.
Nel portfolio ml-service, questo è esattamente il pattern usato per la chat multi-intent: il Planner Ollama analizza la richiesta e la scompone in intent (ricerca RAG, generazione codice, spiegazione concetto), gli Executor gestiscono ciascuno in parallelo, e un Aggregator combina i risultati in una risposta coerente. La latenza percepita scende del 40-60% rispetto al ReAct sequenziale.
# Snippet 3 - Plan-Execute pattern con esecuzione parallela
import asyncio
import json
from dataclasses import dataclass
from typing import Callable, Awaitable
@dataclass(frozen=True)
class Subtask:
"""Unità atomica di lavoro generata dal Planner. Immutabile."""
id: str
description: str
tool: str
args: dict
depends_on: list[str] # IDs subtask che devono completarsi prima
@dataclass(frozen=True)
class SubtaskResult:
"""Risultato di un singolo subtask. Immutabile."""
subtask_id: str
output: str
success: bool
error: str | None = None
async def plan_execute_agent(
goal: str,
tools: dict[str, Callable],
llm_planner: Callable,
llm_executor: Callable,
) -> str:
"""
Pattern Plan-Execute:
1. Planner (LLM grande) genera piano strutturato una sola volta
2. Executor (LLM piccolo o logica deterministica) esegue in parallelo
3. Aggregator combina i risultati
Vantaggi: latenza bassa (parallelismo), costo contenuto (modelli piccoli per executor),
context pulito per ogni subtask (no context rot).
"""
# FASE 1: Pianificazione (unica chiamata al modello grande)
plan_prompt = f"""
Analizza questo obiettivo e crea un piano di esecuzione strutturato.
Obiettivo: {goal}
Tool disponibili: {list(tools.keys())}
Rispondi con JSON:
{{
"plan_summary": "descrizione breve del piano",
"subtasks": [
{{
"id": "step_1",
"description": "cosa fa questo step",
"tool": "nome_tool",
"args": {{}},
"depends_on": []
}}
]
}}
"""
plan_raw = llm_planner(plan_prompt)
plan = json.loads(plan_raw)
subtasks = [Subtask(**s) for s in plan["subtasks"]]
# FASE 2: Esecuzione parallela rispettando le dipendenze
results: dict[str, SubtaskResult] = {}
async def execute_subtask(subtask: Subtask) -> SubtaskResult:
# Attendi dipendenze
for dep_id in subtask.depends_on:
while dep_id not in results:
await asyncio.sleep(0.1)
if not results[dep_id].success:
return SubtaskResult(
subtask_id=subtask.id,
output="",
success=False,
error=f"Dipendenza {dep_id} fallita"
)
# Esecuzione con context limitato (solo questo subtask)
try:
tool_fn = tools[subtask.tool]
output = str(tool_fn(**subtask.args))
return SubtaskResult(subtask_id=subtask.id, output=output, success=True)
except Exception as exc:
return SubtaskResult(subtask_id=subtask.id, output="", success=False, error=str(exc))
# Lancia tutti i subtask (asyncio gestisce le dipendenze tramite attesa attiva)
tasks = [execute_subtask(st) for st in subtasks]
completed = await asyncio.gather(*tasks)
for result in completed:
results[result.subtask_id] = result
# FASE 3: Aggregazione
results_summary = json.dumps({
r.subtask_id: {"output": r.output, "success": r.success}
for r in completed
})
aggregate_prompt = f"""
Obiettivo originale: {goal}
Piano eseguito: {plan['plan_summary']}
Risultati: {results_summary}
Sintetizza i risultati in una risposta coerente e completa per l'utente.
"""
return llm_executor(aggregate_prompt)
Sezione 2c - Reflexion: auto-critica verbale
Reflexion (Shinn et al., 2023) introduce un terzo ruolo nel loop: l'Evaluator. Dopo che l'Actor ha prodotto una risposta, l'Evaluator la valuta e produce una reflection verbale - non un gradiente, non un reward numerico, ma una critica in linguaggio naturale che viene aggiunta al context del ciclo successivo. Su benchmark come HotpotQA e AlfWorld, Reflexion migliora l'accuracy del 20-40% rispetto al ReAct base, al costo di 2-3x più token per task.
Il punto critico è il rischio di degeneration-of-thought: se la reflection è superficiale o l'Evaluator non ha criteri chiari, l'agent può ripetere lo stesso errore in loop aggiungendo "riflessioni" vuote. La soluzione è dare all'Evaluator criteri di valutazione espliciti e strutturati nel prompt.
# Snippet 4 - Reflexion: self-critique con criteri espliciti
from dataclasses import dataclass
@dataclass(frozen=True)
class ReflexionState:
"""Stato immutabile del ciclo Reflexion."""
attempt: int
actor_output: str
evaluation_score: float # 0.0 - 1.0
reflection: str
passed: bool
def reflexion_agent(
task: str,
evaluation_criteria: list[str],
llm_actor: callable,
llm_evaluator: callable,
max_attempts: int = 3,
success_threshold: float = 0.85
) -> tuple[str, list[ReflexionState]]:
"""
Implementazione Reflexion:
- Actor genera risposta
- Evaluator valuta su criteri espliciti (no ambiguità)
- Reflection verbale viene aggiunta al context successivo
- Si ferma quando score >= threshold o esaurisce i tentativi
Ritorna: (risposta_finale, storia_stati)
"""
history: list[ReflexionState] = []
accumulated_reflections: list[str] = []
for attempt in range(1, max_attempts + 1):
# ACTOR: genera risposta con le reflection precedenti come contesto
reflection_context = ""
if accumulated_reflections:
reflection_context = "\n\nReflection dai tentativi precedenti:\n" + \
"\n".join(f"- Tentativo {i+1}: {r}" for i, r in enumerate(accumulated_reflections))
actor_prompt = f"""
Task: {task}
{reflection_context}
Genera la risposta migliore possibile, tenendo conto delle reflection precedenti.
"""
actor_output = llm_actor(actor_prompt)
# EVALUATOR: valuta su criteri espliciti con score numerico
criteria_text = "\n".join(f"- {c}" for c in evaluation_criteria)
eval_prompt = f"""
Task originale: {task}
Risposta generata: {actor_output}
Valuta la risposta su questi criteri:
{criteria_text}
Rispondi in JSON:
{{
"score": 0.0, // valore tra 0.0 e 1.0
"criteria_scores": {{}}, // score per ogni criterio
"passed": false,
"reflection": "Cosa migliorare specificatamente e come farlo nel prossimo tentativo"
}}
"""
eval_raw = llm_evaluator(eval_prompt)
import json
evaluation = json.loads(eval_raw)
state = ReflexionState(
attempt=attempt,
actor_output=actor_output,
evaluation_score=evaluation["score"],
reflection=evaluation["reflection"],
passed=evaluation["passed"] or evaluation["score"] >= success_threshold
)
history.append(state)
if state.passed:
return actor_output, history
# Accumula reflection per il prossimo ciclo
accumulated_reflections.append(evaluation["reflection"])
# Ritorna il miglior tentativo se nessuno ha passato la soglia
best = max(history, key=lambda s: s.evaluation_score)
return best.actor_output, history
Sezione 2d - Tree of Thoughts: esplorazione branch parallela
Tree of Thoughts (Yao et al., 2023) estende il chain-of-thought generando multiple "chain" di ragionamento in parallelo, valutandole con un Evaluator, e selezionando i branch più promettenti tramite beam search o best-first. È la scelta giusta per problemi dove esiste una soluzione ottima oggettiva - puzzle logici, ottimizzazione, planning formale - e dove il costo computazionale di esplorare N branch è giustificato dalla criticità dell'output.
In produzione, ToT è raramente usato nella sua forma pura: il costo di N chiamate LLM parallele per ogni livello dell'albero è proibitivo per task interattivi. Varianti più pragmatiche limitano la branching factor a 2-3 e la profondità a 2-3 livelli, ottenendo il 60-70% del beneficio a un costo 4-9x invece di 10-100x.
# Snippet 5 - Tree of Thoughts: implementazione pragmatica (branching=2, depth=2)
import asyncio
from dataclasses import dataclass, field
@dataclass
class ThoughtNode:
"""Nodo nell'albero dei pensieri."""
thought: str
depth: int
score: float = 0.0
children: list["ThoughtNode"] = field(default_factory=list)
is_solution: bool = False
async def tree_of_thoughts(
problem: str,
llm_generate: callable, # genera N branch di pensiero
llm_evaluate: callable, # valuta ogni branch con score
branching_factor: int = 2,
max_depth: int = 2,
beam_width: int = 2
) -> str:
"""
Tree of Thoughts semplificato:
1. Genera branching_factor pensieri per ogni nodo
2. Valuta ogni pensiero con score 0-10
3. Mantieni solo beam_width migliori a ogni livello (beam search)
4. Fermati quando trovi soluzione o raggiungi max_depth
Costo: O(branching_factor^max_depth) chiamate LLM
Con branching=2, depth=2: 6 chiamate generate + 4 valutazioni = 10 totali
"""
root = ThoughtNode(thought=problem, depth=0, score=10.0)
current_beam = [root]
for depth in range(1, max_depth + 1):
next_candidates = []
for parent_node in current_beam:
# Genera N pensieri figli in parallelo
generate_tasks = [
_generate_thought(problem, parent_node.thought, depth, llm_generate)
for _ in range(branching_factor)
]
thoughts = await asyncio.gather(*generate_tasks)
for thought_text in thoughts:
child = ThoughtNode(thought=thought_text, depth=depth)
parent_node.children.append(child)
next_candidates.append(child)
# Valuta tutti i candidati in parallelo
eval_tasks = [
_evaluate_thought(problem, node.thought, llm_evaluate)
for node in next_candidates
]
scores = await asyncio.gather(*eval_tasks)
for node, score_data in zip(next_candidates, scores):
node.score = score_data["score"]
node.is_solution = score_data.get("is_solution", False)
# Controlla soluzioni trovate
solutions = [n for n in next_candidates if n.is_solution]
if solutions:
best_solution = max(solutions, key=lambda n: n.score)
return best_solution.thought
# Beam search: mantieni solo i migliori
current_beam = sorted(next_candidates, key=lambda n: n.score, reverse=True)[:beam_width]
# Ritorna il pensiero con score più alto come approssimazione
all_nodes = []
_collect_nodes(root, all_nodes)
best = max(all_nodes, key=lambda n: n.score)
return best.thought
async def _generate_thought(problem: str, context: str, depth: int, llm: callable) -> str:
prompt = f"Problema: {problem}\nContesto corrente: {context}\nStep {depth}: proponi un approccio di soluzione specifico."
return llm(prompt)
async def _evaluate_thought(problem: str, thought: str, llm: callable) -> dict:
import json
prompt = f"Problema: {problem}\nPensiero: {thought}\nValuta con JSON: {{\"score\": 0-10, \"is_solution\": bool, \"rationale\": \"...\"}}"
return json.loads(llm(prompt))
def _collect_nodes(node: ThoughtNode, result: list) -> None:
result.append(node)
for child in node.children:
_collect_nodes(child, result)
Sezione 3 - Persistenza dello stato: Redis come working memory
Un agent senza persistenza muore e rinasce ignorante ad ogni sessione. La persistenza dello stato tra sessioni è il salto che trasforma un agent da demo a sistema produzione. La scelta della strategia dipende da quanto stato serve, per quanto tempo, e con quale granularità di recupero.
Per sessioni utente di durata ore o giorni, Redis è la soluzione più pragmatica: veloce, semplice, con TTL nativo per gestire l'expire automatico dello stato. Il pattern è serializzare l'intero stato dell'agent (history, obiettivo corrente, risultati parziali) in JSON e salvarlo con una chiave per session_id. Per persistenza a lungo termine (settimane, mesi), si affianca Postgres.
# Snippet 6 - Agent state persistence su Redis
import json
import redis
from dataclasses import dataclass, asdict
from typing import Any
@dataclass
class AgentState:
"""
Stato completo dell'agent, serializzabile in JSON.
Immutabile: crea nuova istanza per ogni aggiornamento.
"""
session_id: str
goal: str
current_step: int
history: list[dict]
tool_results: dict[str, Any]
status: str # "running" | "completed" | "failed" | "paused"
token_count: int
created_at: str
updated_at: str
class AgentStateManager:
"""
Gestione persistenza stato agent su Redis.
Sessioni TTL 24h di default, rinnovate ad ogni update.
"""
SESSION_TTL_SECONDS = 86_400 # 24 ore
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url, decode_responses=True)
def _key(self, session_id: str) -> str:
return f"agent:state:{session_id}"
def save(self, state: AgentState) -> None:
"""Salva stato agent su Redis con TTL rinnovato."""
key = self._key(state.session_id)
serialized = json.dumps(asdict(state))
# SETEX: set + expire atomico
self.redis.setex(key, self.SESSION_TTL_SECONDS, serialized)
def load(self, session_id: str) -> AgentState | None:
"""Carica stato agent. Ritorna None se scaduto o inesistente."""
key = self._key(session_id)
raw = self.redis.get(key)
if raw is None:
return None
data = json.loads(raw)
return AgentState(**data)
def update_step(self, session_id: str, new_step: int, new_action: dict, token_delta: int) -> AgentState | None:
"""
Aggiorna step corrente: carica, crea nuova istanza (immutabile), salva.
Ritorna None se la sessione non esiste più.
"""
from datetime import datetime, timezone
current = self.load(session_id)
if current is None:
return None
# Crea nuova istanza - mai mutare quella esistente
updated = AgentState(
session_id=current.session_id,
goal=current.goal,
current_step=new_step,
history=[*current.history, new_action], # spread: no mutation
tool_results=current.tool_results,
status=current.status,
token_count=current.token_count + token_delta,
created_at=current.created_at,
updated_at=datetime.now(timezone.utc).isoformat()
)
self.save(updated)
return updated
def pause(self, session_id: str) -> bool:
"""Marca sessione come paused. Utile per kill switch manuale."""
current = self.load(session_id)
if current is None:
return False
paused = AgentState(**{**asdict(current), "status": "paused"})
self.save(paused)
return True
Sezione 4 - Case Study: Devin, Cursor Composer, Claude Code
I tre sistemi di coding agent più diffusi in produzione nel 2026 illustrano come i pattern teorici si traducono in prodotti reali - e quanto diversi siano nelle scelte architetturali.
Devin (Cognition) è l'agent più autonomo del terzetto: gira in un ambiente cloud sandboxed con IDE, browser, terminale e shell propri. Ricevi un task (es. "implementa questa feature e apri una PR"), Devin pianifica, scrive, testa, corregge, e sottomette il PR senza intervento umano. Il pattern architetturale è Plan-Execute con loop ReAct per i subtask: un Planner genera il piano di sviluppo, gli step di implementazione seguono il ciclo write → test → observe → fix. Il costo è elevato: un task complesso può consumare 400-1000 ACU (Agent Compute Units). Al tier Standard ($60/mese per 130K crediti), un task da 500 ACU vale circa $0.23 - ma i task reali durano ore.
Cursor Composer è un agent human-in-the-loop: non agisce autonomamente oltre la proposta di modifiche, aspettando approvazione esplicita prima di ogni azione significativa. Il pattern è ReAct con checkpoint obbligatori. Questo riduce il rischio di modifiche catastrofiche ma alza la latenza percepita. Con credit-based pricing introdotto a metà 2025, il costo per utenti intensivi può superare i $200/mese.
Claude Code (Anthropic) è un agent CLI che gira nel tuo ambiente locale, con accesso diretto al filesystem, git, e terminale. Usa il pattern Plan-Execute con Reflexion implicita: dopo ogni errore, rianalizza l'output e adatta il piano. In test indipendenti, Claude Code usa 5.5x meno token di Cursor per task identici, rendendolo più efficiente per uso intensivo. La sua architettura di tool use è aperta (MCP - Model Context Protocol) e permette di aggiungere tool custom senza modificare il core.
Sezione 5 - Cost profile: agent vs one-shot (la verità sui token)
Il costo è la variabile che determina se un agent è sostenibile in produzione. L'errore più comune è misurare il costo su task demo (semplici, pochi step) e poi sorprendersi quando il task reale costa 50x di più.
# Snippet 7 - Stima costo agent vs one-shot con budget per step
from dataclasses import dataclass
@dataclass(frozen=True)
class CostEstimate:
"""Stima costo immutabile per tipo di sistema."""
system_type: str
input_tokens: int
output_tokens: int
cost_per_m_input: float # € per milione di token input
cost_per_m_output: float
iterations: int
@property
def total_cost_eur(self) -> float:
total_input = self.input_tokens * self.iterations
total_output = self.output_tokens * self.iterations
return (
(total_input / 1_000_000) * self.cost_per_m_input +
(total_output / 1_000_000) * self.cost_per_m_output
)
def __str__(self) -> str:
return (
f"{self.system_type}: "
f"{self.iterations} iter × "
f"({self.input_tokens} in + {self.output_tokens} out) token = "
f"€{self.total_cost_eur:.4f}"
)
# Prezzi indicativi llama3.1:8b su Ollama VPS (costo margine energia ~€0.002/h)
# Per cloud: sostituire con prezzi provider specifico
COSTO_PER_M_INPUT = 0.00 # Ollama self-hosted: costo infrastruttura fisso
COSTO_PER_M_OUTPUT = 0.00
# Confronto architetturale (modello cloud ipotetico a $3/MTok input)
CLOUD_INPUT_PRICE = 3.00 # €/MTok
CLOUD_OUTPUT_PRICE = 15.00 # €/MTok
esempi = [
CostEstimate("Chatbot one-shot", 500, 200, CLOUD_INPUT_PRICE, CLOUD_OUTPUT_PRICE, 1),
CostEstimate("RAG one-shot", 2_000, 500, CLOUD_INPUT_PRICE, CLOUD_OUTPUT_PRICE, 1),
CostEstimate("Agent ReAct (5 step)", 1_500, 400, CLOUD_INPUT_PRICE, CLOUD_OUTPUT_PRICE, 5),
CostEstimate("Agent ReAct (15 step)", 3_000, 600, CLOUD_INPUT_PRICE, CLOUD_OUTPUT_PRICE, 15),
CostEstimate("Agent Plan-Execute (10 subtask)", 800, 300, CLOUD_INPUT_PRICE, CLOUD_OUTPUT_PRICE, 10),
CostEstimate("Agent Reflexion (5 cicli)", 2_000, 800, CLOUD_INPUT_PRICE, CLOUD_OUTPUT_PRICE, 5),
]
# Output analisi costi
for e in esempi:
print(e)
# Risultato tipico con modello cloud:
# Chatbot one-shot: 1 iter × (500 in + 200 out) token = €0.0046
# RAG one-shot: 1 iter × (2000 in + 500 out) token = €0.0138
# Agent ReAct (5 step): 5 iter × (1500 in + 400 out) token = €0.0525
# Agent ReAct (15 step): 15 iter × (3000 in + 600 out) token = €0.2700
# Agent Plan-Execute (10 subtask): 10 iter × (800 in + 300 out) token = €0.0690
# Agent Reflexion (5 cicli): 5 iter × (2000 in + 800 out) token = €0.0900
# CONCLUSIONE: un agent ReAct a 15 step costa ~60x un chatbot one-shot.
# Con Ollama self-hosted, il costo marginale di token è zero -
# ma il costo compute (GPU/CPU VPS) è comunque reale in latenza.
Il confronto numerico è chiaro: un agent ReAct che itera 15 volte con context crescente costa circa 60x un chatbot one-shot con modelli cloud. Con Ollama self-hosted il costo monetario per token è zero, ma il costo reale si manifesta in latenza e saturazione della CPU/GPU del VPS. Per questa ragione, il portfolio ml-service usa Plan-Execute dove possibile (subtask brevi, context pulito) e ReAct solo per i task esplorativi dove l'imprevedibilità dei passi è genuina.
Il budget per step è il controllo critico in produzione: imposta max_iterations
conservativamente, monitora il token count ad ogni step, e implementa un hard stop che
ritorna una risposta parziale invece di loopare fino all'esaurimento del budget. Il prossimo
episodio tratta esattamente questo: come tracciare ogni step con span granulari per sapere,
in 5 secondi, quale iterazione ha bruciato il budget.
FAQ
Qual è la differenza concreta tra un agent LLM e un chatbot?
Un chatbot risponde a un singolo input e si ferma: stateless, senza obiettivi persistenti, senza tool. Un agent LLM mantiene un obiettivo attraverso multiple iterazioni, decide autonomamente i passi successivi, usa tool per modificare il mondo esterno (file, API, database), e persiste lo stato tra le iterazioni. La differenza non è il numero di chiamate LLM - è la presenza del loop di controllo con stato.
Cos'è il pattern ReAct e quando usarlo?
ReAct (Reasoning + Acting) alterna pensiero esplicito e azione: l'LLM scrive un Thought (cosa sta ragionando), esegue un'Action (tool call), osserva l'Observation (risultato del tool), e decide se continuare o dare la risposta finale. È il pattern più adatto a task esplorativi - debugging, ricerca, analisi di documenti - dove i passi non sono prevedibili a priori.
Quando usare Plan-Execute invece di ReAct?
Plan-Execute è preferibile quando il task ha struttura prevedibile e i subtask sono parallelizzabili. Un Planner (modello grande, eseguito una volta) genera il piano; gli Executor (modelli più piccoli) lavorano in parallelo su context isolati. Riduce latenza del 40-60% e costo per step rispetto a ReAct puro. Limite: il piano iniziale deve essere accurato - un Planner che sbaglia la decomposizione manda tutto offline.
Reflexion migliora davvero le performance?
Sì, ma con costo in token (2-3x) e rischio di degeneration-of-thought se i criteri di valutazione nell'Evaluator sono vaghi. Su benchmark HotpotQA e AlfWorld migliora l'accuracy del 20-40%. In produzione, limitare a 2-3 cicli massimi e dare all'Evaluator criteri strutturati ed espliciti - non "valuta se è buona" ma "valuta se risponde alla domanda, cita fonti, è concisa".
Quanto costa far girare un agent in produzione?
Con modelli cloud a ~$3/MTok input, un agent ReAct a 15 step costa circa €0.27 per
run - 60x rispetto a un chatbot one-shot. Con Ollama self-hosted il costo monetario
per token è nullo, ma la latenza aumenta (3-8 secondi per chiamata su llama3.1:8b
su VPS senza GPU dedicata). Il controllo critico è max_iterations hard
combinato con monitoring del token count per step.
Come si persiste lo stato di un agent tra sessioni?
Redis con TTL è la soluzione più pragmatica per sessioni di durata ore/giorni: serializza l'intero stato in JSON, chiave per session_id, TTL 24h rinnovabile. Per stato a lungo termine (settimane, mesi), affianca Postgres. Per compliance GDPR (right-to-be-forgotten), mantieni l'indice session_id → user_id su Postgres in modo da poter cancellare tutti i dati di un utente anche se distribuiti su Redis.
Tree of Thoughts è utile in produzione o solo in ricerca?
ToT è computazionalmente costoso nella sua forma pura. In produzione, varianti pragmatiche con branching factor 2-3 e profondità 2-3 ottengono il 60-70% del beneficio a costo 4-9x invece di 10-100x. È utile solo per task con risposta oggettivamente valutabile (puzzle, ottimizzazione, planning formale). Per task conversazionali o creativi, ReAct o Reflexion offrono rapporto qualità/costo migliore.
Conclusione
Un agent LLM non è un chatbot più intelligente: è un sistema di controllo con proprietà architetturali specifiche - autonomia decisionale, persistenza dello stato, tool use. Ignorare questa distinzione porta a scegliere il pattern sbagliato, spendere 60x di più del necessario, e costruire sistemi che sembrano agenti ma si comportano da pipeline batch costose.
I quattro pattern coprono lo spazio dei problemi: ReAct per l'esplorazione, Plan-Execute per la struttura e il parallelismo, Reflexion per la precisione su output verificabili, Tree of Thoughts per i problemi di ottimizzazione con risposta oggettiva. Nel portfolio ml-service, Plan-Execute con ReAct per i subtask esplorativi è il pattern che ha ridotto la latenza percepita del 45% mantenendo la qualità delle risposte.
Prossimo episodio: observability e come non perdersi in un loop infinito. Quando un agent fallisce dopo 47 tool call, senza tracing granulare non sai a quale step è andato storto. Vedremo come strutturare span OpenTelemetry per avere root-cause in 5 secondi, non 40 minuti - EP02: Observability agent: tracing span-per-thought.
Se ti interessa come questi pattern di AI engineering si traducono in workflow quotidiano su un codebase reale, ho documentato il mio setup (skill design + planning fasato + sub-agent orchestration) in Claude Code Workflow: Superpowers + GSD framework.

Commenti
Caricamento commenti...
Accedi per commentare