LangGraph: subgrafuri (arhitecturi modulare) în LangChain & LangGraph

# stări cu chei comune: subgraful compilat devine nod
builder.add_node("cercetare", subgraf_compilat)

# stări diferite: subgraful se apelează dintr-o funcție-nod
def apel_subgraf(state: MainState):
    rez = subgraf_compilat.invoke({"topic": state["intrebare"]})
    return {"raspuns": rez["raport"]}

Un subgraf este un graf compilat folosit ca nod în alt graf. Documentația îl recomandă pentru sisteme multi-agent, pentru reutilizarea unui set de noduri în mai multe grafuri și pentru echipe care lucrează separat pe părți ale aplicației, cât timp interfața subgrafului (schemele de intrare și ieșire) rămâne aceeași.

Comunicarea dintre părinte și subgraf are două forme. Dacă au chei de stare comune (de exemplu messages), dai subgraful compilat direct lui add_node: el citește și scrie aceleași canale ca părintele. Dacă schemele sunt diferite (fără chei comune), apelezi subgraful dintr-o funcție-nod care transformă starea părintelui în intrarea subgrafului și rezultatul înapoi în starea părintelui.

Persistența se alege la compile()-ul subgrafului. Cu checkpointer=None (implicit), fiecare apel pornește de la zero, dar moștenește checkpointer-ul părintelui pentru interrupt și execuție durabilă. Cu checkpointer=True, starea se acumulează între apeluri pe același fir. Cu checkpointer=False, nu există checkpoint-uri. Părintele trebuie compilat cu un checkpointer pentru ca aceste funcții să meargă; starea subgrafului se vede cu graph.get_state(config, subgraphs=True).

Concepte cheie și caracteristici

Concept / Parametru	Descriere practică
add_node("nume", subgraf)	Subgraful ca nod, când părintele și subgraful au chei de stare comune.
subgraf.invoke(...) dintr-un nod	Pentru scheme diferite; nodul face traducerea stărilor.
compile(checkpointer=None)	Implicit: fiecare apel pornește de la zero, moștenește checkpointer-ul părintelui.
compile(checkpointer=True)	Stare acumulată pe fir; nu suportă apeluri paralele ale aceluiași subgraf.
compile(checkpointer=False)	Fără checkpoint-uri: fără interrupt și fără execuție durabilă.
get_state(config, subgraphs=True)	Include starea subgrafurilor în StateSnapshot.
stream_events(..., version="v3")	Proiecția stream.subgraphs arată fiecare execuție de subgraf cu graph_name și path.

Exemple practice și scenarii reale

Exemplul 1: Subgraf cu chei comune, adăugat direct ca nod

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    topic: str
    raport: str

def cauta(state: State):
    return {"raport": f"Date găsite despre {state['topic']}"}

sub = StateGraph(State)
sub.add_node("cauta", cauta)
sub.add_edge(START, "cauta")
sub.add_edge("cauta", END)
cercetare = sub.compile()

parinte = StateGraph(State)
parinte.add_node("cercetare", cercetare)
parinte.add_edge(START, "cercetare")
parinte.add_edge("cercetare", END)
app = parinte.compile()

print(app.invoke({"topic": "LangGraph"})["raport"])

Exemplul 2: Scheme diferite: subgraful apelat dintr-un nod

class MainState(TypedDict):
    intrebare: str
    raspuns: str

def modul_cercetare(state: MainState):
    rez = cercetare.invoke({"topic": state["intrebare"]})
    return {"raspuns": rez["raport"]}

main = StateGraph(MainState)
main.add_node("cercetare", modul_cercetare)
main.add_edge(START, "cercetare")
main.add_edge("cercetare", END)
app = main.compile()

Greșeli frecvente și capcane

• Să adaugi direct ca nod un subgraf care nu are nicio cheie comună cu părintele. Nu primește date de la părinte; în acest caz apelează-l dintr-o funcție-nod.
• checkpointer=True pe un subagent pe care modelul îl poate apela de mai multe ori în paralel. Apelurile scriu în același namespace și intră în conflict.
• Părinte compilat fără checkpointer. Interrupt-urile, inspectarea stării și memoria pe fir ale subgrafului nu funcționează.

Noțiuni și module înrudite

• langgraph-stategraph
• langgraph-time-travel
• langchain-streaming-tools
• agent-middleware