Dựng agent xử lý hồ sơ nhiều bước bằng planner-executor, DAG và vòng tự phản biện

Với ReAct kiểu truyền thống, LLM chỉ lập kế hoạch cho một bài toán con mỗi lần. Gặp tác vụ ba bước thì cách này ổn.

Nhưng với một hồ sơ vay cần trích xuất thông tin, đối chiếu giấy tờ tùy thân, xác minh thu nhập, chấm rủi ro rồi soạn phản hồi, agent sẽ phải hỏi lại model lớn sau từng hành động, và bạn khó nói trước nó sẽ đi theo đường nào.

Kiến trúc plan-and-execute, được LangChain mô tả trong bài về planning agent, chia việc đó làm hai. Planner dùng LLM sinh kế hoạch nhiều bước cho cả tác vụ lớn. Executor nhận câu hỏi của người dùng kèm một bước trong kế hoạch, rồi gọi một hoặc nhiều tool để làm xong bước đó.

Bài này dựng một agent xử lý hồ sơ như thế qua năm bước: planner, kiểm tra DAG, executor có retry, critic, và replan.

Bạn cần chuẩn bị gì?

Bạn cần Python 3, một hàm call_llm(prompt) -> str bọc model đang dùng, và một dict TOOLS ánh xạ tên tool sang hàm Python. Code bên dưới đã được giản lược để dễ đọc: không có xử lý timeout, logging hay validate schema đầy đủ. Hồ sơ vay là ví dụ giả định, bạn thay bằng nghiệp vụ của mình.

Lời khuyên của Anthropic trong “Building effective agents” đáng nhớ trước khi bắt đầu: tìm giải pháp đơn giản nhất có thể và chỉ tăng độ phức tạp khi thật sự cần. Nếu quy trình của khách hàng đã cố định thì một workflow viết tay là đủ. Planner chỉ đáng dùng khi thứ tự bước thay đổi theo từng hồ sơ.

Bước 1: Planner phải trả về dữ liệu, không trả về văn xuôi

Bắt planner trả JSON có trường depends_on. Nhờ trường này, kế hoạch trở thành thứ máy kiểm tra được.

import json

PLAN_PROMPT = """Bạn là planner. Với hồ sơ dưới đây, trả về JSON:
{"steps": [{"id": "...", "task": "...", "tool": "...", "depends_on": ["..."]}]}
Chỉ dùng các tool: extract_fields, check_id, verify_income, score_risk, draft_reply.
Hồ sơ:
"""

def plan(case_text):
    raw = call_llm(PLAN_PROMPT + case_text)
    steps = json.loads(raw)["steps"]
    for s in steps:
        if s["tool"] not in TOOLS:
            raise ValueError(f"Tool không hợp lệ: {s['tool']}")
    return steps

Kiểm tra: in kế hoạch của ba hồ sơ khác nhau. check_id và verify_income đều nên phụ thuộc vào extract_fields nhưng không phụ thuộc lẫn nhau. Nếu planner xếp chúng thành một chuỗi thẳng thì prompt của bạn chưa diễn đạt được rằng hai việc này độc lập.

Bước 2: Đừng tin kế hoạch trước khi kiểm tra nó là DAG

Tài liệu Airflow định nghĩa DAG là mô hình chứa mọi thứ cần để chạy một workflow. Cấu trúc của DAG đến từ việc khai báo phụ thuộc giữa các task. Kế hoạch do LLM sinh ra cũng là một khai báo phụ thuộc như vậy, chỉ khác là nó có thể sai.

def check_refs(steps):
    ids = {s["id"] for s in steps}
    unknown = {d for s in steps for d in s["depends_on"]} - ids
    if unknown:
        raise ValueError(f"Phụ thuộc vào bước không tồn tại: {unknown}")

def topo_order(steps):
    check_refs(steps)
    deps = {s["id"]: set(s["depends_on"]) for s in steps}
    order = []
    while deps:
        ready = [k for k, ds in deps.items() if not ds]
        if not ready:
            raise ValueError(f"Kế hoạch có vòng lặp: {list(deps)}")
        for k in ready:
            order.append(k)
            del deps[k]
        for ds in deps.values():
            ds.difference_update(ready)
    return order

Để thấy code thực sự cho ra gì, thử hình dung planner trả về kế hoạch sau cho một hồ sơ vay mẫu (ví dụ minh họa, không phải output thật của model nào):

{"steps": [
 {"id": "s1", "task": "Trích trường chính", "tool": "extract_fields", "depends_on": []},
 {"id": "s2", "task": "Đối chiếu CCCD", "tool": "check_id", "depends_on": ["s1"]},
 {"id": "s3", "task": "Đối chiếu sao kê", "tool": "verify_income", "depends_on": ["s1"]},
 {"id": "s4", "task": "Chấm rủi ro", "tool": "score_risk", "depends_on": ["s2", "s3"]},
 {"id": "s5", "task": "Soạn thư trả lời", "tool": "draft_reply", "depends_on": ["s4"]}
]}

Gọi hàm trên danh sách steps này:

>>> topo_order(steps)
['s1', 's2', 's3', 's4', 's5']

Vòng while chạy bốn lượt. Lượt đầu chỉ s1 sẵn sàng; lượt hai s2 và s3 cùng sẵn sàng, tức hai bước này có thể chạy song song; lượt ba đến s4, lượt cuối là s5.

Kiểm tra: tự sửa tay kế hoạch trên cho s4 phụ thuộc vào s5. Hàm phải báo Kế hoạch có vòng lặp: ['s4', 's5'] chứ không được treo. Lỗi ở bước này nên quay về planner kèm thông báo lỗi, không để chạy tiếp.

Bước 3: Executor cần retry và luật chạy rõ ràng

Hai ý tưởng của Airflow dùng được ngay ở đây. Một là đặt tham số mặc định như retries một lần cho mọi bước. Hai là trigger rule: mặc định all_success nghĩa là bước chỉ chạy khi mọi bước upstream đều thành công, còn all_done cho bước chạy sau khi upstream kết thúc, kể cả khi upstream lỗi.

DEFAULTS = {"retries": 2}

def run_step(step, case_text, results):
    inputs = {d: results.get(d) for d in step["depends_on"]}
    last_err = None
    for _ in range(DEFAULTS["retries"] + 1):
        try:
            return TOOLS[step["tool"]](case_text, step["task"], inputs)
        except Exception as e:
            last_err = e
    raise last_err

def execute(steps, case_text):
    by_id = {s["id"]: s for s in steps}
    results, status = {}, {}
    for sid in topo_order(steps):
        step = by_id[sid]
        rule = step.get("trigger_rule", "all_success")
        upstream_ok = all(status[d] == "success" for d in step["depends_on"])
        if rule == "all_success" and not upstream_ok:
            status[sid] = "skipped"
            continue
        try:
            results[sid] = run_step(step, case_text, results)
            status[sid] = "success"
        except Exception:
            status[sid] = "failed"
    return results, status

Executor này không gọi model lớn giữa các bước. LangChain coi đó là lợi ích chính của kiến trúc: workflow nhiều bước chạy nhanh hơn vì agent lớn không phải được hỏi sau mỗi hành động.

Kiểm tra: cho verify_income luôn ném lỗi. Với kế hoạch mẫu ở Bước 2, kết quả đúng là s3 failed, còn s4 và s5 bị skipped. Riêng một bước như “ghi nhật ký thẩm định” nên đặt all_done, để dù hồ sơ hỏng ở đâu thì vẫn có dấu vết lại.

Bước 4: Critic là người chấm bài, không phải người viết lại

Theo LangChain, reflection là một chiến lược prompt giúp nâng chất lượng và tỉ lệ thành công của agent. Trong vòng generator/critic, bộ phản biện được prompt đóng vai giáo viên và góp ý mang tính xây dựng cho bản trả lời đầu tiên.

Anthropic gọi mẫu này là evaluator-optimizer: một lần gọi LLM sinh kết quả, một lần gọi khác đánh giá và phản hồi, lặp lại thành vòng.

CRITIC_PROMPT = """Bạn là giáo viên chấm bài khó tính. Đối chiếu bản nháp với bằng chứng.
Checklist: số tiền khớp giấy tờ; không hứa điều chính sách không cho phép;
mọi lý do từ chối đều có bằng chứng. Trả JSON {"pass": true/false, "issues": [...]}.
"""

def reflect(draft, evidence, max_rounds=2):
    verdict = None
    for _ in range(max_rounds):
        payload = json.dumps({"draft": draft, "evidence": evidence}, ensure_ascii=False)
        verdict = json.loads(call_llm(CRITIC_PROMPT + payload))
        if verdict["pass"]:
            return draft, verdict
        draft = call_llm(f"Sửa bản nháp theo góp ý {verdict['issues']}:\n{draft}")
    return draft, verdict

Checklist mới là phần có giá trị. Một critic chỉ được hỏi chung chung “đã tốt chưa” không có tiêu chí nào để bám vào. Ba tiêu chí cụ thể, lấy từ quy định của khách hàng, cho nó thứ để đối chiếu và cho bạn lý do rõ ràng mỗi khi một bản nháp bị trả lại.

Bước 5: Replan, và biết lúc nào phải dừng

LangChain mô tả rằng sau khi thực thi, agent được gọi lại với prompt replan để chọn giữa trả lời luôn và sinh kế hoạch tiếp theo. Vòng lặp này cần có trần.

REPLAN_PROMPT = """Dựa trên trạng thái các bước, trả JSON:
{"action": "finish", "response": "..."} hoặc {"action": "replan", "steps": [...]}
"""

def handle_case(case_text, max_replans=2):
    steps = plan(case_text)
    for _ in range(max_replans + 1):
        results, status = execute(steps, case_text)
        payload = json.dumps({"status": status, "results": results}, ensure_ascii=False, default=str)
        decision = json.loads(call_llm(REPLAN_PROMPT + payload))
        if decision["action"] == "finish":
            return reflect(decision["response"], results)
        steps = decision["steps"]
    return {"action": "escalate_to_human", "status": status}

Nhánh escalate_to_human là lựa chọn thiết kế, không phải chi tiết phụ. Khi làm với khách hàng, một hồ sơ được chuyển cho người thẩm định kèm trạng thái từng bước thì dễ được chấp nhận hơn nhiều so với một câu trả lời tự tin nhưng sai.

Những lỗi hay gặp

Lỗi đầu tiên là để critic chỉ đọc bản nháp mà không thấy bằng chứng gốc. Nếu extract_fields đọc sai số tiền, bản nháp vẫn có thể trông nhất quán, và critic không có gì để đối chiếu.

LangChain lưu ý một hạn chế liên quan của Reflexion: nó chỉ đi theo một quỹ đạo cố định, nên một bước sai có thể ảnh hưởng các quyết định sau đó. Vì vậy hãy đưa bằng chứng gốc vào critic, như trong code ở trên, để lỗi ở bước đầu còn có cơ hội bị bắt.

Lỗi thứ hai là bỏ hẳn ReAct. Theo nhóm tác giả ReAct, việc xen kẽ suy luận với hành động giúp model theo dõi, cập nhật kế hoạch và xử lý ngoại lệ. Trong một bước khó, ví dụ đọc sao kê ngân hàng có định dạng lạ, executor hoàn toàn có thể là một agent ReAct nhỏ chạy bên trong một node của DAG.

Lỗi thứ ba là dùng chung một mức retries cho mọi bước. Bước gửi email cho khách không nên retry giống bước đọc file, vì retry ở đây có thể khiến khách nhận ba email.

Kỹ năng này xuất hiện ở đâu khi bạn đến chỗ khách hàng?

Ngày đầu ở chỗ khách hàng, việc đầu tiên không phải là viết prompt. Hãy ngồi với người thẩm định và vẽ quy trình thật của họ thành DAG: bước nào chờ bước nào, bước nào lỗi thì cả hồ sơ dừng, bước nào lỗi vẫn phải ghi lại. Bản vẽ đó trở thành depends_on và trigger_rule, còn checklist của họ trở thành prompt cho critic.

Khi đọc JD của các vị trí FDE hay AI engineer, hãy để ý những cụm như “multi-step workflows”, “orchestration” hay “evaluation”. Trong CV, đừng chỉ viết “dựng agent bằng LLM”. Hãy viết rõ bạn đã tách planner khỏi executor, kiểm tra kế hoạch bằng DAG, và hệ thống chuyển cho người thật bao nhiêu phần trăm hồ sơ.

Một agent tốt cho hồ sơ nhiều bước không phải agent tự làm được mọi thứ. Đó là agent mà khi hỏng, bạn chỉ ra được nó hỏng ở node nào và vì sao.