Bài 3 — Dựng Orchestrator: bộ điều phối luồng làm việc

Orchestrator là "quản lý dự án" của đội agent

Các agent tự chúng không biết ai làm trước, khi nào trả lại việc, khi nào dừng. Orchestrator là đoạn code (không phải LLM) điều phối: gọi đúng agent, truyền đúng dữ liệu, quyết định lặp hay dừng dựa trên trạng thái trả về. Đây là bộ não quy trình, và vì là code thuần nên nó xác định, dễ test, dễ log.

Luồng cơ bản

plan = Planner(yeu_cau)
for task in plan.tasks (theo depends_on):
    diff = Coder(task)
    review = Reviewer(diff, task)
    while review.verdict == "changes_requested":
        diff = Coder(task, feedback=review.issues)
        review = Reviewer(diff, task)
    test = Tester(diff, task.acceptance)
    if not test.passed:
        quay lại Coder với log lỗi
    commit(diff)

Pseudocode có kiểm soát vòng lặp

Điểm sống còn: luôn có giới hạn số vòng lặp để tránh hai agent "đá qua đá lại" mãi mãi.

def run_task(task, agents, max_rounds=3):
    feedback = None
    for round_i in range(max_rounds):
        diff = agents.coder(task, feedback)
        review = agents.reviewer(diff, task)
        if review["verdict"] == "approved":
            test = agents.tester(diff, task["acceptance"])
            if test["passed"]:
                return {"status": "done", "diff": diff}
            feedback = test["failures"]
        else:
            feedback = review["issues"]
    return {"status": "escalate", "last_diff": diff, "reason": feedback}

Khi vượt max_rounds, task được escalate (đẩy lên cho người hoặc một agent cấp cao hơn) thay vì lặp vô tận.

Các bước dựng orchestrator

• Viết hàm gọi từng agent trả về dict/JSON đã parse (không phải chuỗi thô).
• Sắp xếp task theo depends_on (topological sort) để không chạy task chưa đủ điều kiện.
• Bọc mỗi lượt gọi trong try/except: nếu agent trả JSON hỏng, thử lại 1 lần rồi mới báo lỗi.
• Đặt max_rounds cho vòng coder↔reviewer và coder↔tester.
• Ghi log mọi lượt: ai gọi, input tóm tắt, output trạng thái, số token.
• Định nghĩa điều kiện dừng rõ ràng: done, escalate, hoặc failed.

Trạng thái là vua

Orchestrator ra quyết định chỉ dựa trên trường trạng thái (verdict, passed), không đọc hiểu văn xuôi. Điều này khiến logic điều phối tách hoàn toàn khỏi nội dung ngôn ngữ, nên bạn có thể unit-test orchestrator bằng cách giả lập (mock) output của agent.

def test_orchestrator_retries_on_changes():
    reviewer = fake_reviewer(["changes_requested", "approved"])
    tester = fake_tester([True])
    result = run_task(sample_task, Agents(coder, reviewer, tester))
    assert result["status"] == "done"

Khung tư duy: "Bốn quyết định của orchestrator"

• Gọi ai tiếp theo? (dựa trên trạng thái hiện tại)
• Truyền gì? (feedback tối thiểu đủ để agent hành động)
• Lặp hay dừng? (đối chiếu max_rounds)
• Khi bí thì đẩy đi đâu? (escalate cho ai)

Sai lầm thường gặp

• Không giới hạn vòng lặp: coder và reviewer bất đồng vĩnh viễn, đốt token vô tận.
• Để LLM làm luôn orchestrator: mất tính xác định, khó debug, dễ trôi khỏi quy trình.
• Bỏ qua depends_on: chạy task phụ thuộc trước khi task nền hoàn thành.
• Không log token và trạng thái: khi hỏng không biết truy vết ở đâu.
• Không xử lý JSON hỏng: một lần LLM trả sai format là cả pipeline sập.