Bài 27: State, Node, Edge — 3 khái niệm cốt lõi

Sau bài này bạn sẽ:

• ✅ Định nghĩa State bằng TypedDict, Pydantic BaseModel hoặc dataclass
• ✅ Hiểu semantics overwrite mặc định và cách dùng Annotated để gắn Reducer
• ✅ Phân biệt operator.add reducer và add_messages reducer
• ✅ Dùng MessagesState shortcut cho 90% agent chatbot
• ✅ Viết Node function với signature đúng, bao gồm async variant
• ✅ Thêm edge thường, edge từ START / đến END, sơ lược conditional edge
• ✅ Compile graph, invoke, stream và đọc output từng node
• ✅ Nhận biết và tránh các pitfall phổ biến về State và graph structure

Trong LangGraph, State là dict duy nhất di chuyển qua toàn bộ graph. Mỗi node nhận State hiện tại, làm việc với nó, rồi trả về một dict chứa các key cần cập nhật (partial update). LangGraph merge partial update đó vào State trước khi chuyển sang node tiếp theo.

Cách định nghĩa State phổ biến nhất là TypedDict (Python 3.8+):

from typing import TypedDict

class AgentState(TypedDict):
    messages: list[str]
    counter: int
    result: str

LangGraph 0.2.x hỗ trợ cả 3 loại sau cho State schema:

Ví dụ với Pydantic BaseModel (có default value đúng chuẩn):

from pydantic import BaseModel, Field

class AgentState(BaseModel):
    messages: list[str] = Field(default_factory=list)
    counter: int = 0
    result: str = ""

Khi khởi tạo graph với StateGraph(AgentState), LangGraph dùng schema này để validate partial update từ mỗi node và để serialize/deserialize state khi checkpoint.

Khi node trả về dict, LangGraph merge vào State hiện tại theo quy tắc overwrite: value mới ghi đè value cũ trên cùng key.

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

class State(TypedDict):
    counter: int
    label: str

def increment(state: State) -> dict:
    # Chỉ trả key cần thay đổi — không cần trả toàn bộ State
    return {"counter": state["counter"] + 1}

def tag(state: State) -> dict:
    return {"label": f"step_{state['counter']}"}

builder = StateGraph(State)
builder.add_node("inc", increment)
builder.add_node("tag", tag)
builder.add_edge(START, "inc")
builder.add_edge("inc", "tag")
builder.add_edge("tag", END)

graph = builder.compile()
result = graph.invoke({"counter": 0, "label": ""})
print(result)
# {"counter": 1, "label": "step_1"}

Điểm quan trọng:

• Node increment chỉ trả {"counter": 1} — key label không bị ảnh hưởng, giữ nguyên giá trị "" từ trước.
• Node tag đọc state["counter"] đã là 1 (sau khi increment chạy xong).
• Nếu node trả về key không tồn tại trong State schema: LangGraph bỏ qua (warning trong log, không raise exception). Đây là nguồn gốc bug khó debug.
• Node trả None hoặc {} đều hợp lệ — State không thay đổi sau node đó.

Overwrite mặc định phù hợp với các field đơn (số, string, bool). Nhưng với list như message history, overwrite sẽ xóa mất dữ liệu cũ. Đây là lúc cần Reducer.

Reducer là function nhận (old_value, new_value) và trả về merged_value. Gắn Reducer vào field bằng Annotated:

from typing import Annotated, TypedDict
from operator import add

class AgentState(TypedDict):
    # field "messages" dùng reducer "add" (= nối list)
    messages: Annotated[list[str], add]
    # field "counter" vẫn dùng overwrite mặc định
    counter: int

Với Annotated[list[str], add]:

• add ở đây là operator.add — khi áp dụng lên list Python sẽ nối (concatenate), tương đương old + new.
• Node trả {"messages": ["tin nhắn mới"]} → LangGraph gọi add(old_messages, ["tin nhắn mới"]) → kết quả là danh sách cũ + tin nhắn mới, không mất lịch sử.

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    items: Annotated[list[str], add]

def step_a(state: State) -> dict:
    return {"items": ["a1", "a2"]}

def step_b(state: State) -> dict:
    return {"items": ["b1"]}

builder = StateGraph(State)
builder.add_node("a", step_a)
builder.add_node("b", step_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)

graph = builder.compile()
result = graph.invoke({"items": []})
print(result["items"])
# ["a1", "a2", "b1"]  — concat, không phải overwrite

Bạn cũng có thể viết reducer tùy chỉnh:

def keep_latest_5(old: list, new: list) -> list:
    """Giữ tối đa 5 phần tử gần nhất."""
    combined = old + new
    return combined[-5:]

class State(TypedDict):
    history: Annotated[list[str], keep_latest_5]

Reducer nhận đúng 2 tham số positional (old_value, new_value) và phải trả về value cùng type với field. LangGraph gọi reducer mỗi khi có node trả về key tương ứng.

operator.add nối list đơn giản, nhưng với BaseMessage của LangChain, cần logic thông minh hơn: nếu message mới có cùng id với message cũ, phải update thay vì append (tránh trùng lặp khi retry). LangGraph cung cấp sẵn add_messages cho use case này:

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Hành vi của add_messages:

• Message mới có id chưa có trong list → append.
• Message mới có id đã tồn tại → replace message cũ bằng message mới (update in place).
• Hỗ trợ cả list[BaseMessage] lẫn BaseMessage đơn lẻ ở phía new value.

from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def respond(state: AgentState) -> dict:
    last_msg = state["messages"][-1]
    reply = AIMessage(content=f"Bạn hỏi: {last_msg.content}")
    return {"messages": [reply]}  # add_messages sẽ append vào list

builder = StateGraph(AgentState)
builder.add_node("respond", respond)
builder.add_edge(START, "respond")
builder.add_edge("respond", END)

graph = builder.compile()
result = graph.invoke({"messages": [HumanMessage("Xin chào")]})

for msg in result["messages"]:
    print(f"{msg.__class__.__name__}: {msg.content}")
# HumanMessage: Xin chào
# AIMessage: Bạn hỏi: Xin chào

add_messages là lựa chọn mặc định cho 90% agent chatbot có conversation history. Dùng operator.add chỉ khi list chứa non-message objects.

LangGraph cung cấp MessagesState — một TypedDict đã có sẵn field messages: Annotated[list[BaseMessage], add_messages]. Không cần tự khai báo lại.

from langgraph.graph import MessagesState

# MessagesState tương đương:
# class MessagesState(TypedDict):
#     messages: Annotated[list[BaseMessage], add_messages]

# Dùng trực tiếp
builder = StateGraph(MessagesState)

# Hoặc mở rộng với thêm field
class MyState(MessagesState):
    user_id: str
    language: str
    tool_output: str

builder = StateGraph(MyState)

Điểm lưu ý khi mở rộng MessagesState:

• Field thêm vào sẽ dùng overwrite mặc định, trừ khi bạn annotate Reducer riêng.
• Pydantic BaseModel không kế thừa được MessagesState (MessagesState là TypedDict). Nếu cần Pydantic, tự khai báo lại field messages với add_messages.
• MessagesState được import từ langgraph.graph, không phải langgraph.graph.message.

Node trong LangGraph là function Python bình thường với signature cố định:

def node_name(state: State) -> dict:
    # đọc từ state
    # làm việc gì đó
    # trả về partial update
    return {"key": new_value}

Một số đặc điểm quan trọng:

• Tham số: Luôn nhận đúng 1 tham số là State (dict hoặc TypedDict/BaseModel tương ứng). LangGraph gọi node với state snapshot tại thời điểm node chạy.
• Return value: dict chứa subset các key muốn cập nhật. Trả None hoặc {} cũng hợp lệ (state không đổi).
• Async: Node async được — dùng graph.ainvoke() / graph.astream() ở phía caller để chạy đúng.
• Không có global state: Node chỉ nên dùng dữ liệu từ state và parameter được inject qua constructor (nếu dùng class-based node). Avoid shared mutable global variables.

from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Node sync
def call_model(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Node async
async def call_model_async(state: AgentState) -> dict:
    response = await llm.ainvoke(state["messages"])
    return {"messages": [response]}

# Node không làm gì nếu điều kiện không thỏa
def maybe_log(state: AgentState) -> dict:
    if state.get("debug"):
        print(f"[debug] messages count: {len(state['messages'])}")
    return {}  # không cập nhật state

# Đăng ký node với graph
builder.add_node("call_model", call_model)
builder.add_node("log", maybe_log)

Quy ước đặt tên node: snake_case, ngắn, mô tả rõ hành động (call_model, execute_tool, format_output). Tên node xuất hiện trong stream output và trace log, nên tránh tên chung chung như step1.

Một node có thể làm bất kỳ thứ gì trong Python — gọi LLM, query DB, gọi REST API, transform data, validate input. LangGraph không ràng buộc implementation bên trong node.

Edge xác định node nào chạy tiếp theo sau khi node hiện tại kết thúc. LangGraph có 3 loại edge chính; bài này trình bày 2 loại đầu, conditional edge sẽ đào sâu ở bài 29.

Edge thường

Luôn đi từ node A sang node B, không phụ thuộc vào state:

builder.add_edge("node_a", "node_b")
# Sau khi node_a chạy xong, luôn chạy node_b

START và END

START và END là 2 node đặc biệt của LangGraph — không cần đăng ký bằng add_node:

• START — entry point của graph. Phải có ít nhất một edge add_edge(START, "tên_node"). Thiếu → compile error.
• END — kết thúc graph. Node cuối cùng phải có edge đến END (trực tiếp hoặc qua conditional edge). Thiếu → graph treo.

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("step1", fn1)
builder.add_node("step2", fn2)
builder.add_node("step3", fn3)

builder.add_edge(START, "step1")   # entry: luôn bắt đầu từ step1
builder.add_edge("step1", "step2") # step1 → step2
builder.add_edge("step2", "step3") # step2 → step3
builder.add_edge("step3", END)     # step3 là node cuối

Một số điểm cần biết:

• Một node có thể nhận edge từ nhiều node khác nhau (fan-in) — node đó chạy sau bất kỳ node nào trong số đó hoàn thành.
• Một node có thể có edge đến nhiều node khác nhau (fan-out tĩnh) — LangGraph chạy chúng song song nếu không có dependency. Song song thực sự chỉ xảy ra trong ainvoke/astream với coroutine.
• builder.set_entry_point("step1") là alias của add_edge(START, "step1") — vẫn được hỗ trợ nhưng add_edge(START, ...) rõ ràng hơn.
• builder.set_finish_point("step3") là alias của add_edge("step3", END).

Khi cần đi node khác nhau tùy theo state (ví dụ: nếu LLM muốn gọi tool thì đi node tools, nếu không thì kết thúc), dùng conditional edge. Bài 29 sẽ đào sâu; ở đây chỉ trình bày cú pháp để hoàn chỉnh bức tranh 3 building block.

from langgraph.graph import END

# 1. Viết routing function — nhận state, trả string tên node tiếp theo
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    # Nếu LLM muốn gọi tool → đi node "tools"
    if last_message.tool_calls:
        return "tools"
    # Ngược lại → kết thúc graph
    return "end"

# 2. Đăng ký conditional edge
builder.add_conditional_edges(
    "agent",          # từ node nào
    should_continue,  # routing function
    {
        "tools": "execute_tools",  # "tools" → node tên "execute_tools"
        "end": END,                # "end" → kết thúc graph
    }
)

Routing function phải trả về string — một trong các key trong dict mapping ở tham số thứ 3. Nếu trả về string không có trong dict → runtime error.

Pattern chuẩn của agent loop: START → agent → (conditional) → tools → agent → (conditional) → END. Bài 28 sẽ build agent đầy đủ theo pattern này.

Sau khi thêm đủ node và edge, gọi builder.compile() để chuyển graph definition thành đối tượng CompiledStateGraph có thể chạy được:

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)
builder.add_node("greet", greet_fn)
builder.add_node("process", process_fn)
builder.add_edge(START, "greet")
builder.add_edge("greet", "process")
builder.add_edge("process", END)

graph = builder.compile()
# graph là CompiledStateGraph — cũng implement Runnable interface

compile() thực hiện validation:

• Kiểm tra mọi node được đăng ký đều reachable từ START.
• Kiểm tra có ít nhất một path dẫn đến END.
• Kiểm tra node được tham chiếu trong edge đã được add_node.
• Nếu validation fail → raise ValueError với message mô tả lỗi cụ thể.

compile() nhận thêm tham số tùy chọn:

from langgraph.checkpoint.memory import MemorySaver

# Thêm checkpointer để lưu state sau mỗi node (cần cho Human-in-the-loop, bài 30)
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Interrupt trước/sau node chỉ định (dùng trong testing và Human-in-the-loop)
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_tools"],  # pause trước khi chạy node này
)

Graph đã compile là một Runnable — implement đầy đủ invoke, ainvoke, stream, astream, batch giống mọi component LangChain. Điều này cho phép embed graph vào LCEL chain nếu cần.

invoke — chạy graph đến END, trả về State cuối cùng:

from langchain_core.messages import HumanMessage

initial_state = {"messages": [HumanMessage("Xin chào")]}
result = graph.invoke(initial_state)
# result là dict chứa toàn bộ State sau khi graph kết thúc
print(result["messages"])

stream — yield từng chunk sau mỗi node, không đợi END:

for chunk in graph.stream(initial_state):
    # chunk là dict: {"tên_node": partial_state_update}
    node_name, state_update = next(iter(chunk.items()))
    print(f"[{node_name}]", state_update)

# Ví dụ output:
# [greet] {"messages": [AIMessage("Chào bạn!")]}
# [process] {"result": "done"}

Stream mode mặc định (mode="values") trả state đầy đủ sau mỗi node, không phải chỉ partial update. Để lấy chỉ update:

for chunk in graph.stream(initial_state, stream_mode="updates"):
    # chunk chứa partial update của node vừa chạy
    print(chunk)

astream_events — stream chi tiết từng event (bao gồm cả LLM token, tool calls) với LangChain event schema:

import asyncio

async def run():
    async for event in graph.astream_events(initial_state, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            # LLM đang yield token
            chunk = event["data"]["chunk"]
            print(chunk.content, end="", flush=True)
        elif kind == "on_chain_end":
            # node vừa kết thúc
            print(f"\n[node done: {event['name']}]")

asyncio.run(run())

astream_events version "v2" là format ổn định từ LangChain v0.2+. Dùng khi cần streaming token-level cho UI chat.

Ví dụ hoàn chỉnh không cần LLM — graph đếm từ trong text, dùng cả Reducer và edge thường:

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END

# ── State ──────────────────────────────────────────────────
class State(TypedDict):
    text: str
    # Dùng reducer "add" để mỗi node có thể append vào list log
    log: Annotated[list[str], add]
    words: Annotated[list[str], add]
    word_count: int

# ── Nodes ──────────────────────────────────────────────────
def split_words(state: State) -> dict:
    words = state["text"].split()
    return {
        "words": words,
        "log": [f"split_words: tách được {len(words)} từ"],
    }

def count(state: State) -> dict:
    n = len(state["words"])
    return {
        "word_count": n,
        "log": [f"count: word_count = {n}"],
    }

def format_result(state: State) -> dict:
    summary = f"Có {state['word_count']} từ trong văn bản"
    return {
        "text": summary,
        "log": [f"format_result: done"],
    }

# ── Build graph ────────────────────────────────────────────
builder = StateGraph(State)
builder.add_node("split", split_words)
builder.add_node("count", count)
builder.add_node("format", format_result)

builder.add_edge(START, "split")
builder.add_edge("split", "count")
builder.add_edge("count", "format")
builder.add_edge("format", END)

graph = builder.compile()

# ── Run ────────────────────────────────────────────────────
result = graph.invoke({
    "text": "Hello world from LangGraph",
    "log": [],
    "words": [],
    "word_count": 0,
})

print(result["text"])
# → "Có 4 từ trong văn bản"

print(result["log"])
# → ["split_words: tách được 4 từ",
#    "count: word_count = 4",
#    "format_result: done"]
# log được concat bởi reducer "add", không bị overwrite

Stream từng bước để thấy thứ tự chạy:

initial = {"text": "Hello world from LangGraph", "log": [], "words": [], "word_count": 0}

for chunk in graph.stream(initial, stream_mode="updates"):
    node_name, update = next(iter(chunk.items()))
    print(f"── {node_name} ──")
    for k, v in update.items():
        print(f"  {k}: {v}")

# Output:
# ── split ──
#   words: ['Hello', 'world', 'from', 'LangGraph']
#   log: ['split_words: tách được 4 từ']
# ── count ──
#   word_count: 4
#   log: ['count: word_count = 4']
# ── format ──
#   text: Có 4 từ trong văn bản
#   log: ['format_result: done']

LangGraph có built-in visualization qua Mermaid và ASCII, hữu ích khi debug graph phức tạp:

# ASCII — dùng trong terminal, không cần dependency thêm
print(graph.get_graph().draw_ascii())

# Output ví dụ:
#         +-----------+
#         | __start__ |
#         +-----------+
#               *
#               *
#           +-------+
#           | split |
#           +-------+
#               *
#               *
#           +-------+
#           | count |
#           +-------+
#               *
#               *
#           +--------+
#           | format |
#           +--------+
#               *
#               *
#         +---------+
#         | __end__ |
#         +---------+

# Mermaid PNG — dùng trong Jupyter Notebook
# Cần cài: pip install grandalf  (hoặc pygraphviz)
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

# Hoặc lấy source Mermaid để dán vào mermaid.live
print(graph.get_graph().draw_mermaid())

LangGraph Studio (desktop app) cũng visualize real-time khi graph đang chạy, kèm state inspector tại mỗi node. Studio cần LangSmith account và chạy local với langgraph dev CLI.

1. Quên annotate Reducer cho list — mất message history

# Sai — mỗi node overwrite messages, mất lịch sử trước
class State(TypedDict):
    messages: list[BaseMessage]  # không có Reducer

# Đúng
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Triệu chứng: graph invoke thành công nhưng result["messages"] chỉ có message của node cuối cùng.

2. Node trả key không có trong State schema

class State(TypedDict):
    counter: int

def node(state: State) -> dict:
    return {"count": state["counter"] + 1}  # sai key: "count" thay vì "counter"
    # LangGraph 0.2.x: silently ignored, counter không tăng

LangGraph 0.2.x bỏ qua key không hợp lệ, chỉ ghi warning vào log. Kết quả state không thay đổi mà không có exception → bug khó phát hiện. Luôn match tên key chính xác với TypedDict.

3. Quên add_edge(START, ...) — compile error

builder.add_node("my_node", fn)
# Thiếu: builder.add_edge(START, "my_node")
graph = builder.compile()
# ValueError: Graph must have an entrypoint...

4. Quên edge đến END — graph treo

builder.add_node("last_node", fn)
builder.add_edge("prev_node", "last_node")
# Thiếu: builder.add_edge("last_node", END)
# graph.invoke() sẽ chạy mãi (hoặc raise GraphRecursionError sau max_iterations)

graph.invoke() mặc định có recursion_limit=25. Nếu graph không đến được END trong 25 bước, raise GraphRecursionError.

5. Modify state in-place thay vì trả dict mới

# Sai — modify trực tiếp state dict
def bad_node(state: State) -> dict:
    state["counter"] += 1  # KHÔNG làm thế này
    return {}

# Đúng — tạo value mới, trả về dict
def good_node(state: State) -> dict:
    return {"counter": state["counter"] + 1}

LangGraph dựa trên immutability để tính diff giữa state versions và để checkpoint đúng. Modify in-place dẫn đến checkpoint sai, và nếu dùng MemorySaver, resume từ checkpoint sẽ có state không nhất quán.

6. TypedDict default value không hoạt động như mong đợi

class State(TypedDict):
    counter: int = 0  # Python 3.10 trở xuống: không có hiệu lực khi invoke

# Luôn truyền đủ initial state khi invoke
result = graph.invoke({"counter": 0})  # tường minh

Python 3.11+ có TypedDict với Required/NotRequired. Trước 3.11, tất cả key của TypedDict đều required khi tạo dict instance. Nếu cần default value, dùng Pydantic BaseModel.

7. Nhầm operator.add cho non-list — TypeError

class State(TypedDict):
    score: Annotated[int, add]  # Sai — "add" ở đây = cộng số

# Khi node trả {"score": 5}, reducer gọi add(old_score, 5)
# = old_score + 5 — cộng dồn, không overwrite
# Có thể là ý định đúng hoặc sai tùy use case, nhưng dễ nhầm

Với số nguyên / float, add reducer = cộng dồn (accumulate), không phải overwrite. Chỉ gắn add nếu bạn thực sự muốn cộng dồn. Với số cần overwrite, không annotate Reducer.

• State = dict duy nhất di chuyển qua graph. Định nghĩa bằng TypedDict, Pydantic BaseModel hoặc dataclass.
• Node trả partial update; LangGraph merge vào State. Mặc định: overwrite.
• Reducer = function (old, new) → merged, gắn vào field qua Annotated[type, reducer_fn].
• operator.add concat list; add_messages concat và deduplicate BaseMessage theo id.
• MessagesState shortcut đã có messages: Annotated[list[BaseMessage], add_messages] built-in.
• Node = function def node(state) -> dict. Async được. Không modify state in-place.
• Edge thường: add_edge(A, B) — A → B không điều kiện. START/END bắt buộc.
• Conditional edge: routing function trả string → map sang tên node tiếp theo. Bài 29 đào sâu.
• builder.compile() validate graph và trả CompiledStateGraph (Runnable). Thêm checkpointer=MemorySaver() để bật persistence.
• graph.invoke() trả State cuối; graph.stream() yield chunk sau mỗi node; graph.astream_events() stream token-level.

Bài 28: Xây Agent đầu tiên với LangGraph — dùng 3 building block vừa học để build ReAct agent thực sự: LLM + tool calling + vòng lặp agent/tools có conditional edge.