Blog sugli Agenti AI
SISTEMA DI ACCESSO COMPUTER

Macchine a Stati e Agenti: Costruire Workflow Affidabili con LangGraph

agentsarchitecturetutoriallanggraphautonomous-systems

La maggior parte dei tutorial sugli agenti mostra un semplice loop: chiedere a Claude, analizzare la risposta, chiamare uno strumento, ripetere. Funziona per le demo. In produzione, si ha bisogno di determinismo, recupero dagli errori, cancelli di approvazione umana e auditabilità.

LangGraph porta le macchine a stati nei workflow degli agenti. Invece di un loop ad-hoc tenuto insieme da istruzioni if, si ottiene un grafo esplicito: nodi con nome (unità di logica), archi tipizzati (transizioni) e uno schema di stato condiviso che fluisce attraverso l’intera esecuzione.

Perché le Macchine a Stati per gli Agenti?

Il Problema del Loop Ad-Hoc

Un tipico loop di agente assomiglia a questo:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

Questo è leggibile per due o tre strumenti. Aggiungete cinque strumenti, percorsi condizionali, un passaggio di approvazione umana e logica di retry — si ottengono centinaia di righe di flusso di controllo aggrovigliato.

Il problema più profondo è lo stato implicito. In quale fase si trova l’agente? Quali dati ha raccolto? Tutto vive in messages — un blob non tipizzato che ogni nodo legge e a cui aggiunge, senza schema imposto.

Le Macchine a Stati come Soluzione

Una macchina a stati rende l’implicito esplicito. Si definiscono:

  • Nodi — unità di logica discrete che ricevono lo stato corrente, fanno una cosa e restituiscono aggiornamenti dello stato.
  • Archi — transizioni tra nodi, incondizionali (A → B sempre) o condizionali.
  • Stato — un dizionario tipizzato che fluisce attraverso l’intero grafo. Lo schema viene validato ad ogni passo.

Quando NON Usare LangGraph

Per task semplici a passo singolo, una chiamata API diretta è più veloce e chiara. Usare LangGraph quando il workflow ha:

  • Più fasi distinte da eseguire in sequenza
  • Branching condizionale basato su risultati intermedi
  • Passi con coinvolgimento umano
  • Logica di recupero errori o retry
  • Requisiti di auditabilità

Fondamenti di LangGraph

Definire lo Stato

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Nodi

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Ricerca questo argomento in modo approfondito: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Scrivi una bozza basata su queste note:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Archi e Routing Condizionale

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Costruire ed Eseguire un Grafo

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "Cos'è il protocollo MCP?"})
print(result["draft"])

Costruire un Workflow di Revisione Documenti

Progettare lo Stato

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Implementare i Nodi

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Estrai esattamente 5 termini chiave da questo documento. "
                "Restituiscili come array JSON di stringhe.\n\n"
                f"Documento: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Controlla questo documento per problemi di conformità. "
                "Cerca: dati personali (nomi, codice fiscale, email, telefoni), "
                "marcature riservate, affermazioni non verificate.\n\n"
                "Restituisci un array JSON di descrizioni dei problemi. "
                "Restituisci un array vuoto [] se non vengono trovati problemi.\n\n"
                f"Documento: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== REVISIONE UMANA RICHIESTA ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Revisionato e approvato dopo la redazione dei dati personali",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Riassumi questo documento in esattamente due frasi:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Assemblare il Grafo

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "Il paziente Mario Rossi (CF: RSSMRA80A01H501Z) soffre di ipertensione.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Checkpoint Umani e Interruzioni

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("Problemi:", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "Approvato dopo redazione manuale dei dati personali", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

Persistenza e Deployment in Produzione

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)


import asyncio


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

Pattern Comuni e Insidie

  • Fan-Out / Fan-In: Eseguire più nodi in parallelo con Send e unire i risultati.
  • Routing condizionale eccessivo: Se una funzione di routing ha troppe diramazioni, dividere in sottografi.
  • Stato sovraccarico: Memorizzare riferimenti (ID database, chiavi S3) invece di grandi artefatti.
  • Test dei nodi in isolamento: Poiché i nodi sono semplici funzioni, possono essere testati unitariamente direttamente.
def test_extract_terms_node():
    state = {
        "document": "L'intelligenza artificiale sta rapidamente trasformando la società moderna.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph sostituisce i loop ad-hoc con workflow espliciti, debuggabili e riprendibili. Iniziare con un grafo lineare (A → B → C → FINE). Aggiungere un arco condizionale quando si ha bisogno di ramificare. Aggiungere un checkpoint umano quando si ha bisogno di approvazione. La maggior parte dei workflow di agenti di produzione ha bisogno esattamente di questi tre pattern.

Articoli Correlati