0Pricing
AI Engineering Academy · Lezione

Creare pipeline multi-agente con LangGraph

Utilizzi LangGraph per definire nodi degli agenti, archi condizionali e stato condiviso in un grafo orientato, abilitando workflow multi-agente complessi con loop, diramazioni e checkpoint con intervento umano.

Creare pipeline multi-agente con LangGraph è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Che cos'è LangGraph?

LangGraph è una libreria basata su LangChain che consente di definire flussi di lavoro multi-agente come grafi orientati. I nodi del grafo rappresentano agenti o passaggi di elaborazione, gli archi rappresentano il flusso dello stato tra di essi e gli archi condizionali consentono un instradamento dinamico in base allo stato corrente. LangGraph gestisce il motore di esecuzione, la persistenza dello stato e i checkpoint con intervento umano.

# pip install langgraph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Define your shared state schema
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # append-only list
    research: str
    draft: str
    status: str

Definizione dello stato in LangGraph

Ogni flusso di lavoro LangGraph opera su un oggetto di stato condiviso che tutti i nodi possono leggere e modificare. Lo stato è un dizionario tipizzato (TypedDict) che trasporta i dati attraverso il grafo. Quando un nodo viene eseguito, riceve lo stato corrente, svolge il proprio lavoro e restituisce un dizionario di aggiornamenti da unire allo stato. Questo modello basato su uno stato condiviso consente agli agenti di comunicare senza un accoppiamento diretto.

from typing import TypedDict

class ResearchState(TypedDict):
    query: str           # input from user
    research_notes: str  # filled by researcher node
    written_draft: str   # filled by writer node
    review_feedback: str # filled by reviewer node
    final_output: str    # filled by synthesizer node
    iteration_count: int # tracks how many revision loops occurred

# Each node returns a PARTIAL update - only the keys it modifies
def researcher_node(state: ResearchState) -> dict:
    notes = do_research(state['query'])
    return {'research_notes': notes}  # only update this key

Creazione dei nodi agente

In LangGraph, ogni agente è una funzione nodo che riceve lo stato corrente, esegue la chiamata LLM e l'esecuzione degli strumenti, quindi restituisce un aggiornamento dello stato. I nodi sono funzioni pure: non memorizzano uno stato interno. Tutto lo stato risiede nell'oggetto di stato condiviso del grafo, rendendo il flusso di lavoro facile da ispezionare, riprendere ed eseguire il debug.

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model='gpt-4o')

def researcher_node(state: ResearchState) -> dict:
    messages = [
        SystemMessage(content='You are a research specialist. Find accurate information.'),
        HumanMessage(content=f'Research this topic: {state["query"]}')
    ]
    response = llm.invoke(messages)
    return {'research_notes': response.content}

def writer_node(state: ResearchState) -> dict:
    messages = [
        SystemMessage(content='You are a technical writer. Write clear, engaging prose.'),
        HumanMessage(content=f'Write a draft using these notes:\n{state["research_notes"]}')
    ]
    response = llm.invoke(messages)
    return {'written_draft': response.content}

Creazione del grafo con StateGraph

Dopo aver definito le funzioni nodo, le collega con StateGraph. Aggiunge i nodi con graph.add_node(), aggiunge gli archi tra di essi con graph.add_edge(), imposta il punto di ingresso con graph.set_entry_point() e compila il grafo in un eseguibile con graph.compile(). Il grafo compilato è un runnable che accetta uno stato iniziale e restituisce lo stato finale.

from langgraph.graph import StateGraph, END

# Build the graph
workflow = StateGraph(ResearchState)

# Add nodes
workflow.add_node('researcher', researcher_node)
workflow.add_node('writer', writer_node)
workflow.add_node('reviewer', reviewer_node)

# Add edges (sequential pipeline)
workflow.set_entry_point('researcher')
workflow.add_edge('researcher', 'writer')
workflow.add_edge('writer', 'reviewer')
workflow.add_edge('reviewer', END)

# Compile into an executable
app = workflow.compile()

# Run it
result = app.invoke({'query': 'What is RAG?', 'iteration_count': 0})
print(result['written_draft'])

Archi condizionali per l'instradamento dinamico

Gli archi condizionali consentono al grafo di instradare l'esecuzione verso nodi diversi in base allo stato corrente. Invece di un arco fisso, si fornisce una funzione di instradamento che esamina lo stato e restituisce il nome del nodo successivo. Questo consente cicli di revisione, controlli di qualità, passaggi di approvazione umana e diramazioni basate sul contenuto degli output degli agenti.

def should_revise(state: ResearchState) -> str:
    '''Router function: returns the name of the next node.'''
    if state['iteration_count'] >= 3:
        return 'finalize'  # Too many revisions - accept as is
    if 'insufficient' in state.get('review_feedback', '').lower():
        return 'researcher'  # Need more research
    if 'rewrite' in state.get('review_feedback', '').lower():
        return 'writer'  # Needs rewriting
    return 'finalize'  # Looks good

# Add conditional edge from reviewer
workflow.add_conditional_edges(
    'reviewer',                    # from node
    should_revise,                 # routing function
    {
        'researcher': 'researcher', # route name -> node name
        'writer': 'writer',
        'finalize': 'finalizer'
    }
)

Cicli e iterazione in LangGraph

LangGraph supporta nativamente i cicli: un agente può essere eseguito nuovamente più volte. Questo è essenziale per i cicli di revisione, i pattern di nuovo tentativo in caso di errore e il perfezionamento iterativo. Includa sempre nello stato una condizione di terminazione del ciclo (ad esempio un contatore delle iterazioni o una soglia per il punteggio di qualità) e la faccia rispettare dal router degli archi condizionali, per evitare cicli infiniti.

def researcher_node(state: ResearchState) -> dict:
    notes = do_research(state['query'])
    return {
        'research_notes': notes,
        'iteration_count': state['iteration_count'] + 1  # always increment
    }

def should_continue_research(state: ResearchState) -> str:
    # Terminate loop after 3 iterations regardless of quality
    if state['iteration_count'] >= 3:
        return END
    # Continue if research is incomplete
    if len(state.get('research_notes', '')) < 500:
        return 'researcher'  # loop back
    return 'writer'  # proceed to next stage

Esecuzione parallela dei nodi

LangGraph supporta i rami paralleli utilizzando RunnableParallel all'interno di un nodo oppure distribuendo l'esecuzione su più nodi. Quando è necessario eseguire simultaneamente sottoattività indipendenti, è possibile creare una diramazione da un nodo verso più nodi paralleli, quindi un nodo di ricongiungimento che attende il completamento di tutti e unisce i risultati nello stato condiviso.

# Fan-out: one node triggers multiple parallel ones
workflow.add_edge('planner', 'researcher_a')
workflow.add_edge('planner', 'researcher_b')
workflow.add_edge('planner', 'researcher_c')

# Fan-in: aggregator waits for all three
workflow.add_edge('researcher_a', 'aggregator')
workflow.add_edge('researcher_b', 'aggregator')
workflow.add_edge('researcher_c', 'aggregator')
workflow.add_edge('aggregator', 'writer')

# Aggregator merges parallel results
def aggregator_node(state: ResearchState) -> dict:
    combined = state.get('notes_a', '') + '\n' + state.get('notes_b', '') + '\n' + state.get('notes_c', '')
    return {'research_notes': combined}

Checkpointing e intervento umano

LangGraph supporta il checkpointing tramite l'integrazione con un backend di checkpoint (SQLite, Redis o PostgreSQL). Con il checkpointing abilitato, lo stato del grafo viene reso persistente dopo l'esecuzione di ogni nodo. Ciò consente di mettere in pausa, ispezionare e riprendere i flussi di lavoro di lunga durata. Inoltre, abilita i pattern con intervento umano, nei quali il grafo si mette in pausa su un nodo specifico e attende che una persona approvi o fornisca input prima di proseguire.

from langgraph.checkpoint.sqlite import SqliteSaver

# Use SQLite for persistent checkpoints
checkpointer = SqliteSaver.from_conn_string(':memory:')
app = workflow.compile(checkpointer=checkpointer, interrupt_before=['human_review'])

# First run - pauses at human_review node
thread = {'configurable': {'thread_id': 'my-workflow-1'}}
result = app.invoke({'query': 'Analyze competitors'}, config=thread)
# result.next == 'human_review' -- waiting for human input

# Human provides feedback and resumes
app.update_state(thread, {'review_feedback': 'Good research, proceed with writing'})
final = app.invoke(None, config=thread)  # resume from checkpoint

Streaming degli output di LangGraph

LangGraph supporta una modalità di streaming che emette aggiornamenti intermedi dello stato al completamento di ogni nodo, invece di attendere la fine dell'intero flusso di lavoro. È utile per le pipeline lunghe in cui si desidera mostrare all'utente i progressi parziali. Utilizzi app.stream() per ottenere un iteratore di snapshot dello stato di ogni passaggio.

# Stream intermediate results as each node completes
for event in app.stream({'query': 'What is RAG?'}):
    for node_name, node_output in event.items():
        print(f'Node completed: {node_name}')
        if 'research_notes' in node_output:
            print('Research done:', node_output['research_notes'][:100])
        if 'written_draft' in node_output:
            print('Draft done:', node_output['written_draft'][:100])

Visualizzazione del grafo

LangGraph può disegnare il grafo come diagramma Mermaid, uno strumento prezioso per comprendere flussi di lavoro complessi con molti nodi e archi condizionali. Chiami app.get_graph().draw_mermaid_png() per ottenere un'immagine PNG oppure app.get_graph().draw_mermaid() per ottenere la sintassi Mermaid da incollare in qualsiasi renderer Mermaid.

# Visualize the workflow graph
graph_image = app.get_graph().draw_mermaid_png()
with open('workflow.png', 'wb') as f:
    f.write(graph_image)

# Or print Mermaid syntax
print(app.get_graph().draw_mermaid())
# Outputs:
# graph TD
#    __start__ --> researcher
#    researcher --> writer
#    writer --> reviewer
#    reviewer -->|Good| finalize
#    reviewer -->|Needs work| researcher

Esempio di agente LangGraph end-to-end

Mettendo insieme tutti gli elementi: una pipeline multi-agente LangGraph completa definisce uno stato TypedDict, crea funzioni nodo per ogni agente, le collega con StateGraph, aggiunge archi condizionali per i cicli di revisione, compila il grafo con un checkpointer e lo richiama con uno stato iniziale. Si ottiene così un flusso di lavoro multi-agente solido, osservabile e riprendibile, con un minimo di codice accessorio.

Verifica rapida

Verifichi la Sua comprensione delle pipeline multi-agente LangGraph illustrate in questa lezione.

Riepilogo della lezione

In questa lezione ha imparato che: LangGraph rappresenta i flussi di lavoro multi-agente come grafi orientati, in cui i nodi sono agenti e gli archi rappresentano il flusso; gli archi condizionali consentono l'instradamento dinamico, i cicli e i cicli di revisione in base allo stato corrente; e il checkpointing permette di mettere in pausa e riprendere i flussi di lavoro di lunga durata con l'approvazione tramite intervento umano. Nella prossima lezione esploreremo la memoria condivisa e la comunicazione tra agenti.

Domande Frequenti

La lezione «Creare pipeline multi-agente con LangGraph» è gratuita?

Sì — il testo completo di «Creare pipeline multi-agente con LangGraph» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Creare pipeline multi-agente con LangGraph»?

Utilizzi LangGraph per definire nodi degli agenti, archi condizionali e stato condiviso in un grafo orientato, abilitando workflow multi-agente complessi con loop, diramazioni e checkpoint con interv… Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.

Quanto tempo richiede la lezione «Creare pipeline multi-agente con LangGraph»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Perché gli agenti singoli arrivano a un limite
  2. Pattern orchestratore-sottoagente
  3. Creare pipeline multi-agente con LangGraph
  4. Memoria condivisa e comunicazione tra agenti
← Torna a AI Engineering Academy