Streaming de mesaje și unelte în LangChain și LangGraph în LangChain / LangGraph: sintaxă Python și exemple

stream() și astream() ale unui graf LangGraph (inclusiv agenții creați cu create_agent) emit evenimente după stream_mode: values (starea completă după fiecare pas), updates (doar modificările, cu numele nodului), messages (perechi (token, metadata) de la apelurile LLM), custom (date trimise din noduri cu get_stream_writer()), plus checkpoints, tasks și debug.

Formatul depinde de opțiuni: cu un singur mod primești direct datele, cu o listă de moduri primești tupluri (mod, date), iar cu subgraphs=True se adaugă namespace-ul. Din LangGraph 1.1, version="v2" dă mereu aceeași formă: un dicționar cu type, ns și data. În modul messages, metadata conține langgraph_node, util ca să afișezi doar tokenii unui nod; apelurile de unelte apar progresiv în tool_call_chunks ale fiecărui AIMessageChunk.

Pentru aplicații noi, documentația recomandă event streaming: graph.stream_events(intrare, version="v3") (LangGraph 1.2, LangChain 1.3) întoarce un obiect cu proiecții separate – stream.messages, stream.values, stream.subgraphs, stream.interrupts, stream.output – care se pot citi independent. La agenții LangChain există și proiecția tool_calls.

Concepte cheie și caracteristici

Concept / Parametru	Descriere practică
`stream_mode="values"`	Starea completă după fiecare pas.
`stream_mode="updates"`	Actualizările fiecărui nod, cu numele nodului.
`stream_mode="messages"`	Tupluri `(message_chunk, metadata)` de la apelurile LLM, din orice nod, unealtă sau subgraf.
`stream_mode="custom"`	Date proprii trimise cu `get_stream_writer()` din `langgraph.config`.
`version="v2"`	Format unic de ieșire: `{"type", "ns", "data"}` (LangGraph 1.1+).
`AIMessageChunk.tool_call_chunks`	Fragmentele de nume și argumente ale apelurilor de unelte, pe măsură ce sunt generate.
`stream_events(..., version="v3")`	Event streaming cu proiecții tipizate (`messages`, `values`, `subgraphs`, `output`).

Exemple practice și scenarii reale

Exemplul 1: Tokeni și pași combinați, asincron

async for mod, chunk in app.astream(
    {"messages": [{"role": "user", "content": "Caută știri despre AI și rezumă-le."}]},
    stream_mode=["messages", "updates"],
):
    if mod == "messages":
        msg_chunk, metadata = chunk
        if msg_chunk.text:
            print(msg_chunk.text, end="", flush=True)
    elif mod == "updates":
        for nod in chunk:
            print(f"\n[nodul {nod} a terminat]")

Exemplul 2: Event streaming la un agent: text și unelte

După exemplul din documentația LangChain.

stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "Cum e vremea la Brașov?"}]},
    config=config,
    version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
    if kind == "messages":
        for token in item.text:
            print(token, end="", flush=True)
    elif kind == "tool_calls":
        print(f"\nUnealtă: {item.tool_name}({item.input})")
        print(f"Rezultat: {item.output}")
final = stream.output

Exemplul 3: Tokenii unui singur nod

for msg, metadata in graph.stream(intrare, stream_mode="messages"):
    if msg.text and metadata["langgraph_node"] == "model":
        print(msg.text, end="", flush=True)

Greșeli frecvente și capcane

Să afișezi doar content. Un chunk poate conține numai fragmente de apel de unealtă (tool_call_chunks) sau raționament; folosește .text pentru text și content_blocks pentru restul.
Cod async pe Python 3.10 fără config transmis. Documentația cere, pentru Python sub 3.11, să dai explicit RunnableConfig la ainvoke() din noduri, altfel modul messages nu primește tokenii.
Să filtrezi după nodul "agent" la un agent creat cu create_agent. Nodul modelului se numește "model".

`Streaming de mesaje și unelte în LangChain și LangGraph` în LangChain & LangGraph