Bài 29: Conditional edge — branching logic

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

• ✅ Hiểu conditional edge khác edge thường ở điểm nào
• ✅ Dùng được add_conditional_edges với và không có path_map
• ✅ Xây fan-out (chạy nhiều node song song) và fan-in (gom kết quả)
• ✅ Dùng Send API để dispatch task động với input state riêng cho từng worker
• ✅ Nhận biết và tránh các runtime error phổ biến của conditional edge

Trong LangGraph có hai loại edge:

• Edge thường (add_edge): destination cố định tại build-time — luôn đi từ node A sang node B.
• Conditional edge (add_conditional_edges): destination được quyết định bởi một router function tại runtime — nhận state, trả về tên node tiếp theo.

Router function là Python thuần:

def route(state: State) -> str:
    if state["score"] > 0.8:
        return "approve"
    return "reject"

LangGraph gọi function này sau khi node nguồn chạy xong và dùng return value để xác định node tiếp theo. Vì vậy logic phân nhánh có thể phức tạp tuỳ ý, miễn là return value hợp lệ.

Conditional edge cũng có thể trả về list tên node để khởi chạy nhiều node song song — đây là cơ chế fan-out sẽ nói ở phần 6.

builder.add_conditional_edges(
    source="classify",      # node trước conditional
    path=route,             # router function: State -> str
    path_map={              # map return value -> node name
        "search_node": "search_node",
        "code_node":   "code_node",
        "default_node": "default_node",
    },
)

Ba tham số chính:

• source: tên node mà conditional edge xuất phát.
• path: router function nhận state, trả str (hoặc list[str] cho fan-out).
• path_map (tùy chọn): dict ánh xạ return value → tên node thực. Nếu return value đã là tên node chính xác, có thể bỏ qua tham số này.

Khi nào bỏ path_map

Nếu router function trả về đúng tên node (hoặc constant END), không cần path_map:

def route(state: State) -> str:
    return "worker_a"  # đúng tên node

builder.add_conditional_edges("dispatcher", route)
# path_map không cần thiết

path_map hữu ích khi muốn tách rời logic return value và tên node thực — ví dụ router trả "yes"/"no" nhưng node tên là "approve_node"/"reject_node".

Chatbot nhận tin nhắn, phân loại intent (search / code / chat), rồi route sang node chuyên biệt:

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI

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

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    intent: str

# --- Nodes ---

def classify(state: State) -> dict:
    last = state["messages"][-1].content
    prompt = f"Phân loại intent (search/code/chat): {last}\nChỉ trả 1 từ."
    intent = llm.invoke(prompt).content.strip().lower()
    return {"intent": intent}

def search_node(state: State) -> dict:
    # xử lý search query...
    return {"messages": [{"role": "assistant", "content": "Kết quả search..."}]}

def code_node(state: State) -> dict:
    # sinh code...
    return {"messages": [{"role": "assistant", "content": "```python\n# code...\n```"}]}

def chat_node(state: State) -> dict:
    # chat thông thường...
    return {"messages": [{"role": "assistant", "content": "Câu trả lời..."}]}

# --- Router ---

def route_intent(state: State) -> str:
    intent = state["intent"]
    if intent in ("search", "code"):
        return intent
    return "chat"

# --- Build graph ---

builder = StateGraph(State)
builder.add_node("classify",    classify)
builder.add_node("search",      search_node)
builder.add_node("code",        code_node)
builder.add_node("chat",        chat_node)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_intent,
    {"search": "search", "code": "code", "chat": "chat"},
)
builder.add_edge("search", END)
builder.add_edge("code",   END)
builder.add_edge("chat",   END)

graph = builder.compile()

Luồng thực thi khi user hỏi "Viết function sort list Python":

1. classify gọi LLM, ghi intent = "code" vào state.
2. Router đọc state["intent"], trả "code".
3. LangGraph route sang code_node, bỏ qua search và chat.
4. code_node → END.

Lưu ý chi phí: node classify dùng 1 LLM call riêng. Khi throughput cao, cân nhắc rule-based pre-filter (keyword matching) trước để giảm số lần gọi LLM.

Pattern agent loop kinh điển: agent gọi LLM, nếu LLM trả về tool call thì route sang tools, ngược lại kết thúc:

from langgraph.graph import END

def should_continue(state: State) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END   # constant đặc biệt — kết thúc graph

builder.add_conditional_edges("agent", should_continue)
# Không có path_map — LangGraph dùng return value trực tiếp
# "tools" → node tên "tools", END → kết thúc

Khi bỏ path_map, LangGraph dùng return value của router làm node name. END là constant từ langgraph.graph, không phải string "END" — LangGraph xử lý đặc biệt để dừng graph.

Graph có cycle: agent → (conditional) → tools → agent → .... LangGraph hỗ trợ cycle natively — đây là điểm khác biệt so với DAG-only framework.

Router trả về list tên node thay vì string đơn — LangGraph chạy tất cả node trong list song song:

def fan_out(state: State) -> list[str]:
    return ["worker_a", "worker_b", "worker_c"]

builder.add_conditional_edges(
    "dispatcher",
    fan_out,
    ["worker_a", "worker_b", "worker_c"],  # path_map dạng list thay vì dict
)

Khi dùng path_map là list, LangGraph hiểu đây là tập hợp node hợp lệ mà router có thể trả về. Router vẫn trả list[str].

Ba worker chạy đồng thời. Output của mỗi worker được merge vào state qua reducer. Reducer mặc định của add_messages append danh sách — nếu key không có reducer, giá trị từ worker cuối chạy xong sẽ ghi đè (last-write-wins), có thể mất dữ liệu.

Cần reducer khi fan-out: với key được nhiều worker cùng update, phải khai báo reducer tường minh:

from operator import add
from typing import Annotated

class State(TypedDict):
    results: Annotated[list[str], add]  # reducer: nối list, không ghi đè
    query: str

Fan-in không cần API đặc biệt. Chỉ cần thêm edge thường từ mỗi worker về node aggregator:

builder.add_edge("worker_a", "aggregator")
builder.add_edge("worker_b", "aggregator")
builder.add_edge("worker_c", "aggregator")

LangGraph tự động chờ tất cả worker hoàn thành trước khi chạy aggregator. State lúc aggregator nhận được đã chứa update được merge từ cả 3 worker.

Node aggregator xử lý như node thông thường:

def aggregator(state: State) -> dict:
    # state["results"] đã chứa output từ cả worker_a, b, c
    combined = "\n\n".join(state["results"])
    return {"final_answer": combined}

Ví dụ đầy đủ fan-out + fan-in với 3 worker:

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

class State(TypedDict):
    query: str
    results: Annotated[list[str], add]  # accumulate từ workers
    final_answer: str

def dispatcher(state: State) -> dict:
    return {}  # chỉ pass-through, fan-out do conditional edge

def worker_a(state: State) -> dict:
    return {"results": [f"Worker A: kết quả cho '{state['query']}'"]}

def worker_b(state: State) -> dict:
    return {"results": [f"Worker B: kết quả cho '{state['query']}'"]}

def worker_c(state: State) -> dict:
    return {"results": [f"Worker C: kết quả cho '{state['query']}'"]}

def aggregator(state: State) -> dict:
    combined = "\n".join(state["results"])
    return {"final_answer": combined}

def fan_out(state: State) -> list[str]:
    return ["worker_a", "worker_b", "worker_c"]

builder = StateGraph(State)
builder.add_node("dispatcher", dispatcher)
builder.add_node("worker_a",   worker_a)
builder.add_node("worker_b",   worker_b)
builder.add_node("worker_c",   worker_c)
builder.add_node("aggregator", aggregator)

builder.add_edge(START, "dispatcher")
builder.add_conditional_edges("dispatcher", fan_out, ["worker_a", "worker_b", "worker_c"])
builder.add_edge("worker_a", "aggregator")
builder.add_edge("worker_b", "aggregator")
builder.add_edge("worker_c", "aggregator")
builder.add_edge("aggregator", END)

graph = builder.compile()
result = graph.invoke({"query": "climate change", "results": [], "final_answer": ""})
print(result["final_answer"])

Map-reduce dùng fan-out + fan-in để xử lý tập item song song:

1. Node splitter chia task lớn thành N item, ghi vào state.
2. Conditional edge fan-out sang N worker (hoặc dùng Send API — xem phần 9).
3. Mỗi worker xử lý 1 item, ghi kết quả vào state qua reducer.
4. Node reducer gom kết quả, tạo output cuối.

Use case điển hình: summarize 50 PDF song song rồi gộp final summary.

# Minh họa cấu trúc (worker tĩnh — số lượng cố định)
#
# START → splitter → [worker_0, worker_1, ..., worker_N] → reducer → END
#
# Giới hạn: số worker phải biết trước lúc build graph.
# Nếu N động, dùng Send API (phần 9).

Với N cố định nhỏ, fan-out list trả về đủ. Với N không biết trước (ví dụ số file PDF upload bởi user), cần Send.

Send (LangGraph 0.1+) mạnh hơn list return: mỗi worker nhận một input state riêng thay vì toàn bộ state chung:

from langgraph.constants import Send

def fan_out_dynamic(state: State) -> list[Send]:
    # Tạo 1 Send per item — số lượng không cần biết trước
    return [
        Send("worker", {"item": item, "context": state["context"]})
        for item in state["items"]
    ]

builder.add_conditional_edges(
    "dispatcher",
    fan_out_dynamic,
    ["worker"],  # danh sách node hợp lệ
)

Send("worker", payload) — tham số:

• Tham số 1: tên node đích.
• Tham số 2: dict state riêng truyền vào node đó.

Ví dụ thực tế — summarize nhiều file PDF song song:

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

class State(TypedDict):
    file_paths: list[str]
    summaries: Annotated[list[str], add]
    final_summary: str

class WorkerInput(TypedDict):
    file_path: str

def load_files(state: State) -> dict:
    # Giả sử file_paths đã có trong state khi invoke
    return {}

def summarize_file(worker_input: WorkerInput) -> dict:
    path = worker_input["file_path"]
    # Đọc và summarize file...
    summary = f"Summary của {path}"
    return {"summaries": [summary]}

def merge_summaries(state: State) -> dict:
    combined = "\n\n".join(state["summaries"])
    # Gọi LLM để tổng hợp...
    return {"final_summary": f"Final: {combined}"}

def dispatch_files(state: State) -> list[Send]:
    return [Send("summarize_file", {"file_path": p}) for p in state["file_paths"]]

builder = StateGraph(State)
builder.add_node("load_files",      load_files)
builder.add_node("summarize_file",  summarize_file)
builder.add_node("merge_summaries", merge_summaries)

builder.add_edge(START, "load_files")
builder.add_conditional_edges("load_files", dispatch_files, ["summarize_file"])
builder.add_edge("summarize_file", "merge_summaries")
builder.add_edge("merge_summaries", END)

graph = builder.compile()
result = graph.invoke({
    "file_paths": ["doc1.pdf", "doc2.pdf", "doc3.pdf"],
    "summaries": [],
    "final_summary": "",
})

Điểm khác biệt Send vs list return:

• List return: tất cả worker nhận cùng global state — cần tự tách item từ state bên trong worker.
• Send: mỗi worker nhận payload riêng — worker không cần biết toàn bộ state, dễ reuse hơn.

Lồng conditional edges

Có thể đặt nhiều cấp conditional edge trong một graph:

# Cấp 1: A → (conditional1) → B hoặc C
# Cấp 2: B → (conditional2) → D hoặc E

builder.add_conditional_edges("A", router1, {"B": "B", "C": "C"})
builder.add_conditional_edges("B", router2, {"D": "D", "E": "E"})
builder.add_edge("C", END)
builder.add_edge("D", END)
builder.add_edge("E", END)

Giữ số cấp lồng nhau ở mức 2-3. Khi vượt 3 cấp, debug runtime rất khó — mỗi nhánh thêm 1 chiều tổ hợp path cần kiểm tra.

Subgraph — đóng gói phức tạp

Khi logic một nhánh đủ phức tạp, đóng gói nó thành subgraph rồi nhúng làm node trong graph chính:

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

# --- Subgraph ---

class SubState(TypedDict):
    query: str
    result: str

def sub_node_a(state: SubState) -> dict:
    return {"result": f"processed: {state['query']}"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("sub_node_a", sub_node_a)
sub_builder.add_edge(START, "sub_node_a")
sub_builder.add_edge("sub_node_a", END)
subgraph = sub_builder.compile()

# --- Main graph ---

class MainState(TypedDict):
    query: str
    result: str
    route: str

def router(state: MainState) -> str:
    return state["route"]

main_builder = StateGraph(MainState)
main_builder.add_node("subprocess", subgraph)  # subgraph là Runnable
main_builder.add_node("other_node", lambda s: {"result": "other"})

main_builder.add_edge(START, "router_node")
# ... thêm router_node và conditional edge ...

Subgraph hoạt động như một Runnable thông thường — nhận dict input, trả dict output. State mapping giữa MainState và SubState cần xử lý bằng tay nếu key khác nhau (wrap subgraph trong một function node để map).

LangGraph cung cấp ASCII visualization để kiểm tra cấu trúc graph sau khi compile:

print(graph.get_graph().draw_ascii())

Output mẫu cho intent routing graph:

        +-----------+
        | __start__ |
        +-----------+
               *
               *
               *
         +----------+
         | classify |
         +----------+
        ...          ...
       .               .
      .                 .
+--------+    +------+    +------+
| search |    | code |    | chat |
+--------+    +------+    +------+
      .                 .
       .               .
        ...          ...
         +---------+
         | __end__ |
         +---------+

Conditional edge hiển thị dạng dotted line trong ASCII. LangGraph Studio (UI riêng) hiển thị runtime path đã đi qua bằng màu khác nhau — hữu ích khi debug luồng phức tạp.

Nếu muốn PNG/SVG (cần cài thêm pygraphviz hoặc Pillow):

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))
# hoặc
print(graph.get_graph().draw_mermaid())

1. Return value không khớp path_map

Router trả "Search" (hoa) nhưng path_map chỉ có "search" (thường) → runtime error "no matching destination".

# Sai
def route(state):
    return state["intent"].strip()  # LLM có thể trả "Search" hoặc "search"

# Đúng — normalize trước khi return
def route(state):
    return state["intent"].strip().lower()

2. Quên END cho nhánh kết thúc

Nhánh không có edge về END → graph không biết dừng, treo vô hạn (hoặc raise error tùy version).

# Thiếu
builder.add_conditional_edges("classify", route, {"search": "search"})
# Khi route trả "other" → không có destination → crash

# Đúng — luôn cover all branches
builder.add_conditional_edges(
    "classify", route,
    {"search": "search", "other": END},
)

3. Fan-out trả list rỗng

Khi router trả [], không node nào được chạy. Graph dừng nhưng state không có kết quả từ worker — aggregator nhận state thiếu dữ liệu. Cần xử lý trường hợp input rỗng trước khi fan-out.

def dispatch(state):
    if not state["items"]:
        return [Send("empty_handler", {})]  # route sang node xử lý rỗng
    return [Send("worker", {"item": x}) for x in state["items"]]

4. Race condition khi fan-out thiếu reducer

Hai worker cùng update key "result" (string, không có reducer) → giá trị của worker nào chạy xong sau ghi đè giá trị worker kia. Luôn khai báo reducer cho key mà nhiều worker cùng viết.

5. Router function throw exception

Exception trong router → graph crash ngay. Router nên wrap try/except và trả về nhánh fallback:

def safe_router(state):
    try:
        intent = state["intent"]
        return intent if intent in VALID_INTENTS else "fallback"
    except Exception:
        return "fallback"

6. Dùng LLM bên trong router

Router chạy sau mỗi node — nếu gọi LLM bên trong router thêm 1 API call mỗi bước. Tốt hơn: tách classify thành node riêng (như ví dụ 1), router chỉ đọc state.

Bài 30: Human-in-the-loop — pause để con người duyệt