langgraph & stateful multi-agent

Chuỗi bước chạy một chiều chỉ là pipeline. Muốn có Agent thật thì nó phải biết lặp lại và tự sửa sai.

Nếu bạn từng dùng LangChain hồi đầu thì chắc quen với Chain: một chuỗi các bước chạy một chiều từ đầu đến cuối, dạng đồ thị có hướng không chu trình (DAG - Directed Acyclic Graph):

Prompt ──► LLM ──► OutputParser ──► Kết quả

Mô hình này chạy ổn cho các tác vụ ngôn ngữ đơn giản: tóm tắt văn bản, trích xuất thực thể, hoặc Q&A một lượt.

Nhưng với Agentic Workflows, khi phần mềm phải tự làm những việc phức tạp như viết code, sửa lỗi, nghiên cứu thị trường, kiểm tra tính toán, thì chuỗi DAG tuyến tính không đủ. Giải quyết vấn đề ngoài đời vốn là vòng lặp (loops), thử sai (trial & error) và quay lui (backtracking):

1. Agent viết code → 2. Chạy thử test → 3. Test fail → 4. Quay lại bước 1 để sửa → 5. Lặp lại cho đến khi pass hoặc cần con người can thiệp.

Để giải quyết chuyện này, team LangChain làm ra LangGraph, một framework đưa tư duy của Máy trạng thái hữu hạn (Finite State Machine) vào việc điều phối AI Agent đơn lẻ lẫn Multi-Agent Systems. (Bên dưới, runtime của LangGraph chạy theo mô hình Pregel của Google: mỗi "super-step" chạy các node đang được kích hoạt, gom kết quả vào state rồi mới sang bước kế.)

Mình sẽ nói về kiến trúc của LangGraph, vì sao State Machine hợp với việc điều phối AI, rồi tới cách dựng một hệ thống Multi-Agent.

# tại sao dag không đủ? state machine vào cuộc

Trong khoa học máy tính cổ điển, khi cần điều khiển một quy trình có nhiều điều kiện chuyển và chu trình lặp (ví dụ: giao thức mạng TCP, quy trình thanh toán đơn hàng e-commerce), không ai dùng script chạy thẳng từ trên xuống dưới cả. Người ta dùng State Machine.

LangGraph định nghĩa một Agent dựa trên 3 thành phần:

1. State (Trạng thái): cấu trúc dữ liệu trung tâm (Data Schema) giữ toàn bộ ngữ cảnh hiện tại: lịch sử tin nhắn, danh sách file đã tạo, kết quả test, các flag...
2. Nodes: các hàm Python/TypeScript đại diện cho từng hành động. Mỗi Node nhận State hiện tại, xử lý (gọi LLM, query database, gọi API) rồi trả về bản cập nhật cho State.
3. Edges: định nghĩa luồng đi từ Node này sang Node khác. Riêng Conditional Edges (cạnh điều kiện) cho phép rẽ nhánh dựa trên kết quả trong State.

# mấy tính năng đáng chú ý của langgraph

# cycles (vòng lặp)

DAG cấm chu trình (cycles), còn LangGraph cho phép luồng quay lại chính nó hoặc các bước trước đó. Đây là nền tảng để làm mẫu Reflection & Self-Correction:

# human-in-the-loop & breakpoints (tạm dừng an toàn)

Với các nghiệp vụ như chuyển tiền, xóa database, gửi email hàng loạt, không nên để AI tự làm 100%. Cần người duyệt trước khi chạy hành động rủi ro.

• LangGraph cho phép dừng graph trước/sau một Node (static breakpoint qua interrupt_before/interrupt_after, hợp để debug), hoặc gọi hàm interrupt() ngay trong Node. Cách thứ hai là cách tài liệu hiện tại khuyên dùng cho human-in-the-loop.
• Khi dừng, graph lưu State hiện tại qua checkpointer (bắt buộc phải cấu hình) rồi trả quyền về cho ứng dụng. Việc báo cho người duyệt (gửi Slack, email, hiện nút trên UI) là phần ứng dụng tự làm, LangGraph không tự gửi.
• Người duyệt có thể Approve, Reject, hoặc sửa trực tiếp State (update_state), sau đó ứng dụng gọi lại graph với cùng thread_id (kèm Command(resume=...) nếu dùng interrupt()) để chạy tiếp.

# time-travel (tua lại để gỡ lỗi)

LangGraph lưu lịch sử trạng thái theo từng bước (Checkpointing), nên bạn có thể:

• Tua lại (rewind) hệ thống về thời điểm trước khi có quyết định sai: get_state_history(config) trả danh sách checkpoint, mỗi cái có checkpoint_id.
• Đổi một nhánh suy luận và cho Agent chạy tiếp từ điểm đó, khỏi chạy lại từ đầu: gọi graph với checkpoint_id cũ (sửa State trước bằng update_state nếu muốn), LangGraph tạo một nhánh mới, lịch sử cũ vẫn giữ. Với hệ thống AI phức tạp thì debug kiểu này tiện hơn hẳn. Lưu ý là các node chạy lại sẽ gọi LLM và tool lại thật, nên tool có side effect (gửi mail, trừ tiền) cần idempotent.

# các mẫu thiết kế phối hợp đa agent (multi-agent patterns)

Khi bài toán quá lớn, nhồi hết vào một Agent duy nhất (Single Monolithic Agent) sẽ làm context window bị loãng và LLM mất tập trung. Cách xử lý là chia nhỏ ra thành một đội Agent chuyên biệt (Specialized Agents).

Hai kiểu phối hợp hay gặp nhất với LangGraph (ngoài ra còn kiểu network/swarm, các agent tự chuyển việc cho nhau qua "handoff" mà không cần trưởng nhóm):

# supervisor pattern (mô hình trưởng nhóm điều phối)

Một Supervisor Agent đóng vai "Project Manager": nhận yêu cầu từ người dùng rồi quyết định giao việc tiếp theo cho Agent chuyên môn nào:

# hierarchical teams (đội nhóm phân cấp)

Các nhóm Agent nhỏ tự phối hợp nội bộ trước, rồi mới báo kết quả lên cấp cao hơn. Giống cách phòng ban trong công ty phần mềm chạy, ví dụ nhóm Dev + Test làm việc với nhau dưới sự quản lý của Tech Lead.

# ví dụ triển khai stateful graph với code

Đây là khung cơ bản của một workflow tự sửa lỗi trong LangGraph:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
import operator
 
# 1. Định nghĩa State Schema trung tâm
class AgentState(TypedDict):
    task: str
    code: str
    test_output: str
    is_passed: bool  # Key nào node trả về cũng phải khai báo ở đây, không thì bị bỏ qua
    iteration: Annotated[int, operator.add] # Reducer: cộng dồn giá trị node trả về thay vì ghi đè
 
# 2. Định nghĩa các Nodes xử lý
def write_code_node(state: AgentState):
    prompt = f"Task: {state['task']}\nPrevious error: {state.get('test_output', '')}"
    # Gọi LLM sinh code
    generated_code = call_llm_code_generator(prompt)
    return {"code": generated_code, "iteration": 1}
 
def test_code_node(state: AgentState):
    # Thực thi test trong Docker Sandbox an toàn
    passed, output = run_unit_tests(state["code"])
    return {"test_output": output, "is_passed": passed}
 
def human_review_node(state: AgentState):
    # Thường gọi interrupt() ở đây để chờ người duyệt (cần checkpointer)
    return {}
 
# 3. Định nghĩa Conditional Edge
def should_continue(state: AgentState):
    if state.get("is_passed", False):
        return END # Đạt yêu cầu, kết thúc!
    if state["iteration"] >= 3:
        return "human_review" # Thử lại quá 3 lần -> Chuyển người
    return "write_code" # Quay lại bước viết code để sửa!
 
# 4. Ghép nối Graph
workflow = StateGraph(AgentState)
workflow.add_node("write_code", write_code_node)
workflow.add_node("test_code", test_code_node)
workflow.add_node("human_review", human_review_node)
 
workflow.add_edge(START, "write_code")
workflow.add_edge("write_code", "test_code")
workflow.add_conditional_edges("test_code", should_continue, ["write_code", "human_review", END])
workflow.add_edge("human_review", END)
 
app = workflow.compile()
result = app.invoke({"task": "Viết hàm parse ngày tháng", "iteration": 0})

Vài chỗ trong đoạn code trên đáng nói kỹ hơn:

• Node không sửa State trực tiếp mà trả về một dict "bản cập nhật". Mặc định LangGraph ghi đè từng key; key nào gắn reducer (như iteration với operator.add) thì giá trị mới được cộng vào giá trị cũ. Lịch sử tin nhắn thường dùng reducer add_messages để nối thêm thay vì thay thế.
• Node trả về key không có trong AgentState thì LangGraph lặng lẽ bỏ qua, không báo lỗi. Nếu quên khai báo is_passed, should_continue sẽ không bao giờ thấy test pass.
• Mọi đích mà conditional edge có thể trả về (ở đây là human_review) phải là node đã add_node. Truyền thêm danh sách đích làm tham số thứ ba giúp graph biết trước các nhánh để vẽ sơ đồ và kiểm tra.
• Vòng lặp cần điểm dừng do mình đặt (iteration >= 3). LangGraph cũng có recursion_limit (số super-step tối đa, vượt thì ném GraphRecursionError), nhưng giá trị mặc định đã đổi qua các phiên bản (bản cũ là 25, bản 1.2 là hơn 10.000), nên đừng trông vào nó, hãy truyền rõ trong config: app.invoke(input, {"recursion_limit": 20}).

# tổng kết

Hướng đi của AI đang dịch dần từ ý tưởng "một model AGI vạn năng lo hết mọi việc" sang hệ thống phần mềm có cấu trúc: nhiều Agent chuyên môn phối hợp với nhau, chạy dưới sự kiểm soát của state machine.

LangGraph không chỉ để làm chatbot. Với checkpointer, interrupt và vòng lặp có kiểm soát, nó cho bạn đủ đồ nghề để dựng workflow tự động hoá chạy được trên Production. Muốn đi tiếp thì nên đọc phần persistence và interrupts trong tài liệu chính thức.

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.