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

> Một agent cứ làm xong một bước lại phải hỏi model lớn bước tiếp theo sẽ chậm và khó kiểm soát; tách việc lập kế hoạch, thứ tự chạy và khâu phản biện ra riêng sẽ cho bạn một hệ thống đo được, sửa được và giải thích được với khách hàng.

Bản gốc: https://fdetimes.net/vi/bach-khoa/planner-executor-va-self-critique/

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.

```python
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.

```python
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):

```json
{"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.

```python
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.

```python
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.

```python
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.

**Điểm mấu chốt:** Kế hoạch do LLM sinh ra chỉ là một bản đề xuất; nó thành chương trình khi đã qua kiểm tra DAG, có retry, có trigger rule và có trần số vòng.

## 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.

**Thử ngay tuần này:**

- Lấy một quy trình nhiều bước có thật ở công ty bạn, vẽ nó thành DAG trên giấy, rồi bắt planner sinh lại đúng DAG đó từ mô tả bằng lời.
- Viết hàm topo_order trong bài và thử với ba kế hoạch hỏng: phụ thuộc vào bước không tồn tại, hai bước phụ thuộc vòng vào nhau, và một tool không có trong danh sách.
- Chạy 10 hồ sơ mẫu, ghi lại số vòng critic và số lần replan của từng hồ sơ, rồi chọn ngưỡng dừng dựa trên số liệu đó.

## Nguồn

- [Plan-and-Execute Agents (LangChain)](https://www.langchain.com/blog/planning-agents)

- [Dags (Airflow Documentation)](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html)

- [Reflection Agents (LangChain)](https://www.langchain.com/blog/reflection-agents)

- [ReAct: Synergizing Reasoning and Acting in Language Models](https://react-lm.github.io/)

- [Building effective agents (Anthropic)](https://www.anthropic.com/engineering/building-effective-agents)