Stare și reducer-e
Nodurile returnează actualizări parțiale, niciodată toată starea. O cheie fără reducer e suprascrisă; o cheie adnotată cu un reducer e combinată. add_messages adaugă mesajele noi la final și îl înlocuiește pe cel cu același id; așa editezi un mesaj.
import operator
from typing import Annotated, TypedDict
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import add_messages
class State(TypedDict):
question: str # no reducer: a new value overwrites the old one
log: Annotated[list[str], operator.add] # reducer: lists are concatenated
messages: Annotated[list, add_messages] # append, or replace by message id
print(operator.add(["start"], ["agent"]))
old = [HumanMessage("Hi", id="1"), AIMessage("Draft", id="2")]
print([m.content for m in add_messages(old, [AIMessage("Final", id="2")])])
print([m.content for m in add_messages(old, [AIMessage("More")])])
['start', 'agent']
['Hi', 'Final']
['Hi', 'Draft', 'More']
MessagesState este starea gata făcută, doar cu messages și add_messages; moștenește din ea ca să adaugi chei.
Noduri, muchii și muchii condiționate
- Un nod este o funcție (sincronă sau asincronă) care primește starea și returnează un dict de actualizări.
add_edge(a, b) merge mereu de la a la b. START și END marchează intrarea și ieșirea.
add_conditional_edges(a, router) apelează router(state) și merge unde indică valoarea returnată. Dă o listă sau un dict de destinații, ca graful să poată fi desenat.
- Un nod poate returna și
Command(goto="b", update={...}), ca să actualizeze starea și să aleagă ruta în același loc.
- O muchie înapoi spre un nod anterior formează un ciclu. Bucla asta e diferența dintre un graf și un chain.
Cicluri și limita de recursivitate
Fiecare super-pas se numără la recursion_limit. Când o rulare atinge limita, LangGraph aruncă GraphRecursionError în loc să ruleze la nesfârșit. O setezi per apel, în config.
from typing import TypedDict
from langgraph.errors import GraphRecursionError
from langgraph.graph import START, StateGraph
class State(TypedDict):
n: int
builder = StateGraph(State)
builder.add_node("loop", lambda s: {"n": s["n"] + 1})
builder.add_edge(START, "loop")
builder.add_edge("loop", "loop") # a cycle with no exit
graph = builder.compile()
try:
graph.invoke({"n": 0}, {"recursion_limit": 5})
except GraphRecursionError as e:
print(type(e).__name__, str(e).splitlines()[0])
GraphRecursionError Recursion limit of 5 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
Versiunile mai vechi aveau implicit 25 de pași. În langgraph 1.2 valoarea implicită este 10.007 (LANGGRAPH_DEFAULT_RECURSION_LIMIT), iar create_agent setează 9.999, așa că pune-ți propria limită pentru agenții care pot intra în buclă.
Checkpointere, thread-uri și time travel
Compilezi cu un checkpointer și fiecare super-pas este salvat ca checkpoint sub thread_id din config. Asta îți dă memorie de conversație, rulări reluabile după un crash, interrupt-uri și time travel: te întorci la un checkpoint vechi, îl modifici și rulezi mai departe pe o ramură nouă.
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
class State(TypedDict):
n: int
builder = StateGraph(State)
builder.add_node("double", lambda s: {"n": s["n"] * 2})
builder.add_node("inc", lambda s: {"n": s["n"] + 1})
builder.add_edge(START, "double")
builder.add_edge("double", "inc")
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "t"}}
print(graph.invoke({"n": 5}, config))
history = list(graph.get_state_history(config)) # newest first
before_inc = next(c for c in history if c.next == ("inc",))
print(before_inc.values)
fork = graph.update_state(before_inc.config, {"n": 100})
print(graph.invoke(None, fork))
{'n': 11}
{'n': 10}
{'n': 101}
Folosește InMemorySaver în teste și un saver pe bază de date în producție, de exemplu PostgresSaver din langgraph-checkpoint-postgres.
Interrupt-uri
interrupt(value) într-un nod pune rularea pe pauză și returnează value apelantului, sub __interrupt__. Are nevoie de un checkpointer.
- Reiei cu
graph.invoke(Command(resume=answer), config) pe același thread. Nodul repornește de la prima linie, iar interrupt() returnează answer.
- Din cauza repornirii, pune efectele secundare după apelul
interrupt() sau fă-le idempotente.
interrupt_before și interrupt_after, setate la compilare, pun pauză în jurul unor noduri întregi; sunt utile mai ales la depanare.
Send: map-reduce și ramuri paralele
Când numărul de ramuri se știe abia la rulare, returnează dintr-o muchie condiționată o listă de Send(node, input). Fiecare rulează nodul cu propriul input, în paralel, în același super-pas, iar un reducer colectează rezultatele.
import operator
from typing import Annotated, TypedDict
from langgraph.graph import START, StateGraph
from langgraph.types import Send
class State(TypedDict):
topics: list[str]
summaries: Annotated[list[str], operator.add]
def fan_out(state: State):
return [Send("summarise", {"topic": t}) for t in state["topics"]]
def summarise(item: dict):
return {"summaries": [f"summary of {item['topic']}"]}
builder = StateGraph(State)
builder.add_node("summarise", summarise)
builder.add_conditional_edges(START, fan_out, ["summarise"])
graph = builder.compile()
print(graph.invoke({"topics": ["cats", "dogs", "owls"]}))
{'topics': ['cats', 'dogs', 'owls'], 'summaries': ['summary of cats', 'summary of dogs', 'summary of owls']}
Pentru un set fix de ramuri nu ai nevoie de Send: adaugi mai multe muchii care pleacă din același nod, iar destinațiile rulează în paralel.
Subgrafuri
Un graf compilat este un Runnable, deci poate fi nod în alt graf. Dacă au chei de stare comune, îl dai direct lui add_node. Dacă schemele diferă, îl apelezi dintr-o funcție-nod și mapezi starea la intrare și la ieșire. Cu subgrafuri construiești sisteme multi-agent din agenți mai mici, testabili.
from typing import TypedDict
from langgraph.graph import START, StateGraph
class State(TypedDict):
text: str
def clean(state: State):
return {"text": state["text"].strip()}
inner = StateGraph(State)
inner.add_node("clean", clean)
inner.add_edge(START, "clean")
cleaner = inner.compile()
outer = StateGraph(State)
outer.add_node("cleaner", cleaner) # a compiled graph is a node
outer.add_node("shout", lambda s: {"text": s["text"].upper()})
outer.add_edge(START, "cleaner")
outer.add_edge("cleaner", "shout")
print(outer.compile().invoke({"text": " hi "}))
{'text': 'HI'}
Memorie pe termen lung: Store
Un checkpointer ține minte un singur thread. Un Store păstrează documente JSON în namespace-uri, partajate între thread-uri: preferințele utilizatorului, fapte învățate. Nodurile ajung la el, ca și la context-ul rulării, prin argumentul Runtime.
from dataclasses import dataclass
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
def remember(state: MessagesState, runtime: Runtime[Context]):
ns = ("users", runtime.context.user_id)
runtime.store.put(ns, "prefs", {"language": "Romanian"})
return {}
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node("remember", remember)
builder.add_edge(START, "remember")
store = InMemoryStore()
graph = builder.compile(store=store)
graph.invoke({"messages": []}, context=Context(user_id="ana"))
print(store.get(("users", "ana"), "prefs").value)
{'language': 'Romanian'}
Cu un index de embedding-uri configurat, store.search(namespace, query=...) găsește amintiri după sens.