AI-Driven Automation: costruire sistemi di workflow con AI Agent

Per anni l'automazione dei processi software ha significato script bash, cron job e pipeline ETL gestite a mano. Con l'avvento dei Large Language Model capaci di ragionare, pianificare ed eseguire strumenti arbitrari, il paradigma è cambiato radicalmente. Gli AI agent non sono semplici chatbot: sono unità di esecuzione autonome che ricevono un obiettivo, decompongono il problema in sotto-task, invocano API e strumenti, osservano i risultati e iterano fino al completamento. Il mercato degli AI agent è proiettato a raggiungere 7,4 miliardi di dollari entro fine 2025 — e la domanda che ogni team di ingegneria deve rispondersi non è 'se' adottarli, ma 'come' farlo in modo affidabile in produzione.

Cos'è un AI agent e quando usarlo

Un AI agent è un sistema che combina un LLM con memoria, strumenti (tool) e un loop di esecuzione. Il modello osserva lo stato corrente, decide quale azione intraprendere, esegue l'azione tramite un tool call, osserva il risultato e ripete finché il goal non è raggiunto o una condizione di uscita viene soddisfatta. Questo pattern — spesso chiamato ReAct (Reasoning + Acting) — è il cuore di quasi tutti i framework di orchestrazione moderni. Gli agent sono la scelta giusta quando il problema richiede decisioni condizionali dinamiche, l'integrazione con sistemi eterogenei, gestione dello stato persistente tra più passi e capacità di recovery autonomo dagli errori.

LangGraph: workflow agentici come grafi stateful

LangGraph, sviluppato dal team LangChain, è il framework di riferimento per chi vuole il massimo controllo sul comportamento degli agent in produzione. Il concetto centrale è semplice: il workflow è un grafo diretto in cui ogni nodo rappresenta un'azione — una chiamata LLM, un tool call, una funzione Python — e ogni arco rappresenta una transizione condizionale. Lo stato del grafo, un oggetto Python tipizzato, viene passato da nodo a nodo e persiste tra le esecuzioni. Questo rende LangGraph particolarmente adatto a workflow che devono essere interrotti, ispezionati e ripresi — fondamentale per i pattern human-in-the-loop richiesti in contesti enterprise. LangGraph ha raggiunto la versione 1.0 stabile in ottobre 2025, con un impegno formale alla backward compatibility fino alla 2.0, ed è oggi in uso da parte di Klarna, Replit ed Elastic.

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
import operator

# Definizione dello stato condiviso tra i nodi del grafo
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    research_results: list[str]

# Nodo: orchestratore che decide il passo successivo
def orchestrator_node(state: AgentState) -> AgentState:
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    llm_with_tools = llm.bind_tools([search_tool, summarize_tool])
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# Nodo: esecutore degli strumenti (tool calls)
tool_node = ToolNode(tools=[search_tool, summarize_tool])

# Funzione di routing condizionale
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# Costruzione del grafo
workflow = StateGraph(AgentState)
workflow.add_node("orchestrator", orchestrator_node)
workflow.add_node("tools", tool_node)

workflow.set_entry_point("orchestrator")
workflow.add_conditional_edges(
    "orchestrator",
    should_continue,
    {"tools": "tools", END: END},
)
workflow.add_edge("tools", "orchestrator")

# Compilazione con checkpointer per persistenza dello stato
app = workflow.compile(checkpointer=MemorySaver())

# Esecuzione con thread_id per conversazioni multi-turn
config = {"configurable": {"thread_id": "research-session-001"}}
result = app.invoke(
    {"messages": [HumanMessage(content="Ricerca le ultime novità su LangGraph v1.0")]},
    config=config,
)

n8n: automazione visuale con AI integrata

Se LangGraph è lo strumento del software engineer che vuole controllo totale, n8n è lo strumento del team che vuole velocità di iterazione. n8n è una piattaforma open-source di workflow automation con oltre 500 integrazioni native: database, API REST, email, Slack, Notion, Airtable e decine di altri servizi. Dal 2024 n8n ha integrato nodi LangChain nativi, permettendo di costruire pipeline AI senza scrivere codice. Il pattern tipico è usare n8n come orchestratore di sistema — gestisce trigger, data routing e integrazioni di terze parti — mentre gli AI agent vengono invocati come nodi specializzati per il ragionamento.

• Trigger multipli: webhook HTTP, polling schedulato, event bus, email in arrivo, form submission.
• Nodi AI nativi: OpenAI Chat, LangChain Agent, Vector Store Retrieval, Document Loader, Embeddings.
• Branching e looping: logica condizionale e iterazione su array di dati senza codice.
• Error handling visuale: definizione di fallback e retry direttamente nel grafo del workflow.
• Self-hosted o cloud: deploy su infrastruttura propria per mantenere il controllo dei dati sensibili.

Temporal: workflow durevoli per agent long-running

Un loop agentivo semplice funziona bene per task che si completano in pochi secondi. Ma i workflow enterprise raramente sono così semplici: un processo di onboarding cliente può durare giorni, un'analisi di compliance può richiedere approvazioni umane intermedie, un agente di procurement deve gestire fallimenti di API esterne con retry intelligenti. Temporal è stato progettato esattamente per questi casi: un orchestratore di workflow che persiste lo stato su disco, gestisce retry con backoff esponenziale, supporta il pattern Saga per le compensazioni transazionali e garantisce l'esecuzione exactly-once anche in presenza di crash del processo.

from temporalio import workflow, activity
from temporalio.client import Client
from temporalio.common import RetryPolicy
from temporalio.worker import Worker
from datetime import timedelta
import asyncio

@activity.defn
async def create_user_account(user_data: dict) -> str:
    """Crea account nel CRM — retry automatico su errori transitori."""
    response = await crm_client.create_user(user_data)
    return response["user_id"]

@activity.defn
async def send_welcome_email(user_id: str, email: str) -> bool:
    return await email_service.send_welcome(user_id, email)

@activity.defn
async def run_ai_enrichment_agent(user_id: str) -> dict:
    """Invoca un LangGraph agent per arricchire il profilo utente."""
    return await langgraph_app.ainvoke({"user_id": user_id, "task": "enrich_profile"})

@workflow.defn
class UserOnboardingWorkflow:
    @workflow.run
    async def run(self, user_data: dict) -> dict:
        # Step 1: crea account con policy di retry esplicita
        user_id = await workflow.execute_activity(
            create_user_account,
            user_data,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        # Step 2: email di benvenuto (non blocca il flusso)
        await workflow.execute_activity(
            send_welcome_email,
            args=[user_id, user_data["email"]],
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Step 3: arricchimento AI (può richiedere minuti, stato persistito)
        enrichment = await workflow.execute_activity(
            run_ai_enrichment_agent,
            user_id,
            start_to_close_timeout=timedelta(minutes=5),
        )

        return {"user_id": user_id, "enrichment": enrichment, "status": "completed"}

async def main():
    client = await Client.connect("localhost:7233")
    async with Worker(
        client,
        task_queue="onboarding-queue",
        workflows=[UserOnboardingWorkflow],
        activities=[create_user_account, send_welcome_email, run_ai_enrichment_agent],
    ):
        await asyncio.Future()

Pattern architetturali per sistemi multi-agent

Quando un singolo agent non è sufficiente — perché il problema è troppo ampio, richiede competenze specializzate diverse, o deve essere parallelizzato — si passa ai sistemi multi-agent. Esistono tre pattern fondamentali: nel pattern Supervisor, un agente orchestratore analizza il task entrante e lo smista a worker agent specializzati, raccogliendo e sintetizzando i risultati. Nel pattern Network, gli agent comunicano tra loro attraverso uno stato condiviso, aggiungendo osservazioni e decisioni senza un orchestratore centrale. Nel pattern Hierarchical, i team di agent vengono organizzati come sotto-grafi annidati in LangGraph, un approccio più strutturato adatto a domini molto complessi.

Osservabilità e debugging in produzione

I sistemi agentici sono difficili da debuggare perché il percorso di esecuzione è determinato dinamicamente dal modello, non da codice deterministico. Il tracing distribuito applicato agli LLM, attraverso strumenti come LangSmith, Langfuse o Phoenix, permette di visualizzare ogni singolo passo di esecuzione: quale nodo è stato eseguito, quale prompt è stato inviato, quale risposta è arrivata, quale tool è stato chiamato con quali argomenti e quanto tempo ha impiegato. In produzione, è fondamentale tracciare il costo per esecuzione in token, la latenza end-to-end per workflow, il tasso di fallimento per nodo e le deviazioni dal comportamento atteso.

1. 1Instrumenta ogni nodo del grafo con span OpenTelemetry per il tracing distribuito.
2. 2Registra input e output di ogni tool call in un audit log immutabile per il debugging post-mortem.
3. 3Imposta alert su latenza P95 per workflow e su errori di tool call superiori al 5%.
4. 4Usa LangSmith o Langfuse per la visualizzazione delle trace LLM in staging e produzione.
5. 5Implementa un circuit breaker per le chiamate ai modelli esterni: se il provider è degradato, fallisci velocemente invece di accumulare timeout costosi.

Considerazioni su costo e latenza

Un errore comune nei proof-of-concept è ottimizzare per la qualità dell'output ignorando il costo di produzione. Un workflow multi-agent che chiama GPT-4o ad ogni nodo può costare decine di centesimi per esecuzione. La strategia corretta è usare modelli più piccoli e veloci — GPT-4o-mini, Gemini Flash, Claude Haiku — per routing, classificazione e task strutturati, riservando i modelli frontier ai nodi di ragionamento complesso. Il caching delle risposte LLM per prompt identici, supportato nativamente da Anthropic e OpenAI a livello API, può ridurre i costi fino al 90% per workflow ad alta ripetizione.

Conclusioni

I sistemi di workflow AI-driven non sono più sperimentali: sono architetture di produzione con pattern consolidati, strumenti maturi e una community attiva. LangGraph fornisce il controllo granulare necessario per agent stateful e complessi; n8n abbassa la barriera di accesso per team misti; Temporal risolve i problemi di durabilità e fault tolerance che i framework LLM-native ignorano. La chiave per passare dal prototipo alla produzione è trattare gli agent come sistemi distribuiti: con tracing, circuit breaker, costi monitorati e un'architettura che permette di evolvere il grafo di esecuzione senza riscrivere tutto da zero.