English version: Graph Engineering: Old Structure, New Nodes, and a Working Agent Graph in LangGraph. Phần 9 của series. Phần trước: A2A. Phần sau: Orca.

Cứ vài tháng ngành mình lại đổi tên cho việc mình đang làm. Prompt engineering thành context engineering, rồi thành harness engineering, và tháng 6 có người gọi việc thiết kế vòng lặp tool của một agent là "loop engineering". Tháng 7 đến lượt graph. Peter Steinberger, người làm OpenClaw, hỏi trên X rằng chúng ta "vẫn đang nói về loop hay đã chuyển sang graph rồi?", và chỉ vài giờ sau Hamel Husain đăng một bài tên "Loop Engineering Is Dead. Enter Graph Engineering." Cả hai ít nhất cũng nửa đùa. Một tuần sau, thuật ngữ đã có vài định nghĩa cạnh tranh nhau, một dòng bài giải thích không dứt, và Harrison Chase của LangChain nói ông không chắc nó nghĩa là gì ngoài "về cơ bản là LangGraph".

Quan điểm của tôi thế này. Cái tên là marketing. Cách làm thì cũ: state machine, workflow engine, DAG của Airflow, sơ đồ BPMN đều là đồ thị các bước với chuyển tiếp tường minh. Cái mới, và là lý do kỷ luật cũ giờ càng quan trọng hơn chứ không kém đi, là thứ nằm bên trong các node. Ngày trước một node là một hàm làm đúng những gì code của nó viết. Giờ một node có thể là một agent tự diễn giải nhiệm vụ, và hai lần chạy cùng một node có thể làm hai việc khác nhau. Cấu trúc tường minh bao quanh điều đó không phải thủ tục rườm rà. Đó là cách giữ cho một hệ thống làm từ các mảnh không tất định đủ dễ đoán để đưa lên production.

Graph là gì, khi nói về agent#

Một agent graph có ba phần.

Node làm việc. Node có thể là code thuần (chạy test, gọi API), một lời gọi model (phân loại ticket này), một lời gọi tool, hoặc trọn một agent loop ở phần 1 với tool riêng.

Edge quyết định bước tiếp theo. Edge cố định luôn đi tới cùng một node. Edge có điều kiện nhìn vào state rồi chọn. Edge fan-out khởi động nhiều nhánh song song, và số nhánh có thể quyết định lúc chạy.

State là thứ chảy qua graph: một bản ghi có kiểu mà các node đọc và cập nhật. Quyết định thiết kế quan trọng là các cập nhật đồng thời được gộp thế nào. Nếu ba reviewer song song mỗi người trả về một danh sách lỗi, chúng ghi đè nhau hay nối vào nhau? Bạn quyết định điều đó cho từng field bằng một reducer.

Bài "Building effective agents" của Anthropic vạch một đường giữa workflow, nơi code quyết định đường đi, và agent, nơi model quyết định. Graph là cách bạn chọn vị trí đường đó cho từng bước thay vì một lần cho cả hệ thống. Cấu trúc nào bạn chắc chắn thì đưa vào edge. Phán đoán nào bạn không viết ra được thì để trong node.

Khi nào một vòng lặp là đủ#

Phần lớn những gì tôi dựng vẫn là một vòng lặp. Một agent có tool tốt, chiến lược context tốt và một harness (phần 1 đến 4) xử lý được một dải công việc rộng đến bất ngờ, và mỗi node thêm vào là thêm một prompt phải bảo trì, thêm một chỗ state có thể sai, thêm một thứ phải đánh giá.

Tôi chỉ dùng tới graph khi trace hoặc eval cho thấy một trong những dấu hiệu sau:

• Các việc con độc lập, không nên chia sẻ context. Review bảo mật và review hiệu năng trên cùng một diff nên do hai lời gọi không nhìn thấy suy luận của nhau, vừa vì chất lượng, vừa để sai lầm của nhánh này không lây sang nhánh kia.
• Một bước bắt buộc phải xảy ra, lần nào cũng vậy. Chạy test sau khi sửa. Kiểm tra tuân thủ trước khi gửi. Nếu trông vào việc model tự nhớ, nó sẽ quên đúng vào lần quan trọng. Edge thì không quên.
• Người kiểm chứng không phải người làm. Agent viết code là người chấm code tệ nhất. Hãy dùng một node riêng với prompt riêng, tốt hơn là model riêng, tốt nhất là một phép kiểm tra tất định.
• Một lần dừng chờ con người có thể kéo dài hàng giờ. Cổng phê duyệt cần state bền vững, không phải một process ngồi chờ input().
• Một vòng lặp cần giới hạn cứng. "Thử sửa đi" thì được. "Thử sửa mãi" là một hoá đơn.

Nếu không có dấu hiệu nào, cứ ở lại với vòng lặp.

Một graph review code#

Đây là hình dạng chúng tôi dùng cho review code có agent hỗ trợ, rút gọn để chạy được mà không cần API key. Ba reviewer nhìn một diff song song. Một bước kiểm chứng chỉ giữ lại những lỗi có thật. Nếu có lỗi được xác nhận, một agent sửa diff rồi test chạy; nếu test hỏng thì thử lại, tối đa ba lần. Mọi đường đi đều kết thúc ở một người duyệt việc merge.

       ┌─ review(security) ─┐
START ─┼─ review(perf) ─────┼─► verify ─► fix ─► tests ─► human
       └─ review(correct) ──┘     │        ▲       │        ▲
                                  │        └─fail──┘        │
                                  │         (max 3)         │
                                  └──── nothing confirmed ──┘

Tôi dùng LangGraph vì nó làm mọi thứ ở trên trở nên tường minh và có sẵn durable execution; cùng thiết kế này làm được với Google ADK, Agent Framework của Microsoft, hay Python thuần cộng một bảng state. Cài bằng pip install langgraph.

# review_graph.py — a small agent graph: fan-out reviewers, a verifier, a fixer loop, a human gate
import operator
from typing import Annotated, Literal, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, Send, interrupt


class Finding(TypedDict):
    reviewer: str
    severity: Literal["low", "medium", "high"]
    issue: str


class State(TypedDict, total=False):
    diff: str
    findings: Annotated[list[Finding], operator.add]   # reducer: parallel branches append, never overwrite
    confirmed: list[Finding]
    fix_attempts: int
    tests_passed: bool
    approved: bool


REVIEWERS = ["security", "performance", "correctness"]


# --- nodes -------------------------------------------------------------------
# Each node is either deterministic code or one bounded agent call. Replace the
# bodies marked LLM with your agent loop from part 1; the graph does not change.

def fan_out(state: State) -> list[Send]:
    """One Send per reviewer: they run in parallel, each with only the diff."""
    return [Send("review", {"diff": state["diff"], "reviewer": r}) for r in REVIEWERS]


def review(payload: dict) -> dict:
    # LLM: "You are the {reviewer} reviewer. List concrete problems in this diff as JSON."
    diff, who = payload["diff"], payload["reviewer"]
    found = []
    if who == "security" and "eval(" in diff:
        found.append({"reviewer": who, "severity": "high", "issue": "eval() on request data"})
    if who == "performance" and "for row in rows" in diff:
        found.append({"reviewer": who, "severity": "low", "issue": "N+1 query inside loop"})
    return {"findings": found}


def verify(state: State) -> dict:
    # Deterministic gate (or a reviewer from a *different* model): keep only what can be reproduced.
    confirmed = [f for f in state.get("findings", []) if f["severity"] in ("medium", "high")]
    return {"confirmed": confirmed}


def route_after_verify(state: State) -> Literal["fix", "human_review"]:
    return "fix" if state["confirmed"] else "human_review"


def fix(state: State) -> dict:
    # LLM: coding agent with the confirmed findings as its task list.
    diff = state["diff"].replace("eval(", "json.loads(")
    return {"diff": diff, "fix_attempts": state.get("fix_attempts", 0) + 1}


def run_tests(state: State) -> dict:
    return {"tests_passed": "eval(" not in state["diff"]}


def route_after_tests(state: State) -> Literal["fix", "human_review"]:
    if state["tests_passed"] or state.get("fix_attempts", 0) >= 3:   # hard stop: three tries, then a human
        return "human_review"
    return "fix"


def human_review(state: State) -> dict:
    decision = interrupt({
        "diff": state["diff"],
        "confirmed": state.get("confirmed", []),
        "tests_passed": state.get("tests_passed"),
        "question": "Merge this change?",
    })
    return {"approved": bool(decision.get("approve"))}


# --- wiring ------------------------------------------------------------------
builder = StateGraph(State)
builder.add_node("review", review)
builder.add_node("verify", verify)
builder.add_node("fix", fix)
builder.add_node("run_tests", run_tests)
builder.add_node("human_review", human_review)

builder.add_conditional_edges(START, fan_out, ["review"])
builder.add_edge("review", "verify")                 # join: verify runs once all reviewers finish
builder.add_conditional_edges("verify", route_after_verify)
builder.add_edge("fix", "run_tests")
builder.add_conditional_edges("run_tests", route_after_tests)
builder.add_edge("human_review", END)

graph = builder.compile(checkpointer=InMemorySaver())   # use SqliteSaver/PostgresSaver outside a demo


if __name__ == "__main__":
    config = {"configurable": {"thread_id": "pr-1842"}, "recursion_limit": 25}
    diff = "def handler(req):\n    data = eval(req.body)\n    for row in rows:\n        db.get(row)\n"

    for step in graph.stream({"diff": diff}, config, stream_mode="updates"):
        print(step)

    paused = graph.get_state(config)
    print("\npaused at:", paused.next, "| question:", paused.tasks[0].interrupts[0].value["question"])

    final = graph.invoke(Command(resume={"approve": True}), config)
    print("approved:", final["approved"], "| fix attempts:", final["fix_attempts"])
    print("findings kept:", [f["issue"] for f in final["confirmed"]])

Kết quả, đã cắt bớt một chút:

{'review': {'findings': [{'reviewer': 'security', 'severity': 'high', 'issue': 'eval() on request data'}]}}
{'review': {'findings': [{'reviewer': 'performance', 'severity': 'low', 'issue': 'N+1 query inside loop'}]}}
{'review': {'findings': []}}
{'verify': {'confirmed': [{'reviewer': 'security', 'severity': 'high', 'issue': 'eval() on request data'}]}}
{'fix': {'diff': 'def handler(req):\n    data = json.loads(req.body)\n ...', 'fix_attempts': 1}}
{'run_tests': {'tests_passed': True}}
{'__interrupt__': (Interrupt(value={..., 'question': 'Merge this change?'}, ...),)}

paused at: ('human_review',) | question: Merge this change?
approved: True | fix attempts: 1
findings kept: ['eval() on request data']

Thân các reviewer là stub có chủ đích. Trong production mỗi reviewer là một lời gọi model với prompt riêng cho vai trò đó, còn fix là một coding agent. Phần graph, thứ bài này nói tới, giữ nguyên hoàn toàn.

Năm quyết định trong file đó#

Fan-out được quyết định lúc chạy. fan_out trả về một danh sách Send, mỗi reviewer một cái, và mỗi nhánh chỉ nhận đúng thứ nó cần: diff và vai trò của nó. Không có lỗi mà reviewer khác tìm ra, không có lịch sử hội thoại. Nếu mai này danh sách reviewer phụ thuộc vào ngôn ngữ lập trình có trong diff, fan_out dựng danh sách đó từ diff và không gì khác phải đổi.

Reducer là hợp đồng về đồng thời. findings là Annotated[list[Finding], operator.add], nên ba nhánh song song cùng nối vào một danh sách. Không có reducer, LangGraph từ chối cập nhật với InvalidUpdateError: At key 'findings': Can receive only one value per step, và đó là hành vi đúng. Kiểu "ai ghi sau thắng" một cách im lặng là cách các lỗi tìm được biến mất không dấu vết.

Điểm hợp nhánh chỉ là một edge thường. review → verify chờ tất cả các nhánh review song song xong rồi mới chạy verify một lần. Bạn có một barrier mà không phải viết code đồng bộ nào.

Vòng lặp có hai giới hạn. route_after_tests dừng sau ba lần sửa và đẩy việc cho con người, dù test qua hay không. recursion_limit trong config là lưới an toàn nếu tôi viết routing sai: LangGraph dừng lần chạy sau ngần ấy bước. Mọi chu trình trong agent graph cần một giới hạn bạn chủ động chọn, và một giới hạn thứ hai bạn không cần phải nghĩ tới.

Con người là một node, không phải chuyện tính sau. interrupt() tạm dừng graph, lưu state và trả payload cho bên đang chạy graph. Command(resume=...) chạy tiếp từ đúng chỗ đó, và giá trị truyền vào trở thành giá trị trả về của interrupt(). Vì có thread_id, "đúng chỗ đó" sống sót qua mọi thứ mà checkpointer sống sót được.

Dừng chờ hàng giờ#

Checkpointer trong bộ nhớ dùng cho demo thì ổn, còn cho một lần duyệt phải chờ qua đêm thì vô dụng. Đổi sang SQLite hoặc Postgres (pip install langgraph-checkpoint-sqlite, hoặc langgraph-checkpoint-postgres) là lần dừng sống qua được restart. Script dưới đây bắt đầu một lần review ở process này và tiếp tục nó ở process khác:

# durable.py — same graph, checkpointed to SQLite: pause in one process, resume in another
import sqlite3
import sys

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command

from review_graph import builder

graph = builder.compile(checkpointer=SqliteSaver(sqlite3.connect("checkpoints.db", check_same_thread=False)))
config = {"configurable": {"thread_id": "pr-1842"}, "recursion_limit": 25}

if sys.argv[1] == "start":
    diff = "def handler(req):\n    data = eval(req.body)\n"
    graph.invoke({"diff": diff}, config)
    print("waiting for a human at:", graph.get_state(config).next)
else:  # "approve" or "reject", possibly hours later, from a webhook or a CLI
    final = graph.invoke(Command(resume={"approve": sys.argv[1] == "approve"}), config)
    print("approved:", final["approved"], "| diff now:", final["diff"].splitlines()[1].strip())

$ python durable.py start
waiting for a human at: ('human_review',)
$ python durable.py approve
approved: True | diff now: data = json.loads(req.body)

Ở hệ thống của chúng tôi, phía start được kích hoạt bởi webhook pull request, payload của interrupt được đăng lên PR thành một comment có hai nút bấm, và handler của nút bấm chính là phía approve. Graph không biết gì về những thứ đó. Nó chỉ biết thread id của nó.

Có một quy tắc trong docs của LangGraph đáng trích lại vì nó từng cắn chúng tôi: khi graph chạy tiếp, node bị ngắt chạy lại từ đầu, nên mọi code đứng trước interrupt() sẽ thực thi hai lần. Side effect trước interrupt phải idempotent, hoặc dời nó ra sau, hoặc tách thành node riêng.

Những lỗi tôi gặp đi gặp lại#

Khi review một agent graph người khác dựng, đây là các vấn đề, xếp gần đúng theo mức độ thường gặp.

Phụ thuộc ngầm giữa các nhánh song song. Hai nhánh trông độc lập nhưng cùng ghi vào một file, hoặc một nhánh lặng lẽ giả định nhánh kia đã chạy. Các nhánh song song chỉ nên chia sẻ input, ngoài ra không gì cả. Nếu chúng cần output của nhau thì chúng là tuần tự.

Orchestrator nguyên khối. Một node "supervisor" với prompt bốn mươi dòng quyết định mọi thứ ở mọi bước. Đó là một vòng lặp khoác áo graph, và khó debug hơn cả hai. Nếu logic routing viết được thành code, hãy viết nó thành conditional edge.

Vòng đánh giá không giới hạn. Một critic gửi việc ngược lại cho worker cho tới khi critic hài lòng. Không có giới hạn cứng thì nó chạy tới khi hết tiền, và theo kinh nghiệm của tôi, đường cong chất lượng gần như phẳng sau vòng thứ hai.

Mỏi mắt vì duyệt. Đặt cổng duyệt ở mọi bước là dạy người duyệt bấm approve mà không đọc. Đặt cổng ở các hành động không đảo ngược được và ở kết quả cuối. Các mức rủi ro ở phần 4 quyết định đặt ở đâu.

Retry lặp lại hành động không đảo ngược được. Một node gửi email rồi hỏng ở dòng tiếp theo sẽ gửi email lần nữa khi retry. Cho các tác động không đảo ngược được một idempotency key, hoặc tách chúng thành node riêng ở cuối.

Đưa toàn bộ state cho mọi node. Truyền nguyên state, kể cả transcript của mọi agent trước đó, cho mọi node chính là context engineering thất bại ở quy mô graph (phần 3). Mỗi node nên nhận một lát cắt của state, chỉ gồm những gì nhiệm vụ của nó cần.

Thiết kế thừa. Chín node cho một việc mà một vòng lặp làm ổn. Cái giá không chỉ là token. Node nào cũng cần eval riêng, và một graph không ai đánh giá theo từng node là một graph không ai hiểu.

Đồng thuận không phải kiểm chứng#

Lời phê bình có ích nhất với làn sóng graph engineering là điều này, và nhiều người đã nói ra hồi tháng 7: thêm agent vào thêm ô không làm hệ thống đúng hơn. Ba reviewer chạy cùng một model, cùng context và cùng điểm mù sẽ vui vẻ đồng ý với cùng một câu trả lời sai. Một graph gồm những agent như vậy có thể sản xuất sai lầm được tổ chức gọn gàng ở quy mô lớn, và chính cấu trúc khiến đầu ra trông đáng tin hơn thực tế.

Cách phòng là neo graph vào những thứ không thể bị thuyết phục để đồng ý. Theo thứ tự ưu tiên:

1. Kiểm tra tất định. Test, type checker, linter, validate schema, cổng CI. Trong ví dụ, run_tests là node tôi tin nhất, vì nó không có ý kiến riêng.
2. Một model khác làm người kiểm chứng. Nếu buộc phải để model chấm, hãy dùng model thuộc họ khác với model đã làm việc, với prompt yêu cầu nó tìm vấn đề chứ không phải xác nhận.
3. Bằng chứng từ ngoài hệ thống. Truy xuất đối chiếu nguồn thật, một câu query trên dữ liệu production, so sánh với số liệu tháng trước.
4. Một con người ở cuối, nhìn vào bản tóm tắt ngắn những gì đã được xác nhận và vì sao, không phải một bức tường transcript.

Tôi sẽ bắt đầu thế nào#

Bắt đầu với một vòng lặp và bộ eval ở phần 5. Chỉ thêm node khi một trace hay một eval cho thấy một lỗi cụ thể mà node đó sẽ sửa, và thêm eval cho nó cùng lúc. Node đầu tiên bạn thêm nên là một bước kiểm chứng thật, tốt nhất là tất định. Cho mọi chu trình một điểm dừng cứng ngay từ ngày đầu. Dùng checkpointer bền vững trước khi cổng duyệt đầu tiên lên production, đừng đợi tới lần mất phê duyệt đầu tiên.

Graph là phần hệ thống do bạn thiết kế. Node là phần model ứng biến. Hãy giữ sự ứng biến bên trong những chiếc hộp bạn hiểu.

Ba bài tiếp theo chuyển từ thiết kế hệ agent sang việc chạy nhiều coding agent cùng lúc trên một codebase thật, bắt đầu với Orca.

Tài liệu tham khảo#

• Sydney Runkle và Harrison Chase, 3 Years of Graph Engineering with LangGraph, LangChain blog (tháng 7/2026)
• LangGraph docs, Interrupts (gồm quy tắc về side effect trước interrupt)
• Anthropic, Building effective agents (workflow và agent)
• LangGraph trên GitHub