Salta al contenuto principale

Gli agent non sono chat: architetture autonome che fanno più di quanto chiedi

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...

Cosa ne pensi di questo articolo?

Gli agent non sono chat: architetture autonome che fanno più di quanto chiedi
AI EngineeringFoto:Matheus BertelliPexels

In Breve

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...

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.

Prova Questi Strumenti

Strumenti gratuiti correlati a questo argomento.

Commenti

Caricamento commenti...

Accedi per commentare

Discuti questo articolo nella community

Unisciti al gruppo Telegram per discutere con altri sviluppatori, fare domande e condividere le tue esperienze.

Ti è piaciuto questo articolo?

Esplora altri contenuti sul blog o scopri i miei progetti