Tôi từng build một agent tìm kiếm và tổng hợp thông tin từ nhiều nguồn. Agent dùng ReAct. Mỗi bước nó nghĩ, quyết định gọi tool, đọc kết quả, nghĩ tiếp. Sau 12 bước, nó tổng hợp xong.

Vấn đề là task đó tốn 47 giây và khoảng 28.000 token. Tôi nhìn vào trace và thấy LLM được gọi 13 lần. Trong đó, 12 lần đầu mỗi lần nó chỉ làm một việc: quyết định gọi tool nào tiếp theo. Chỉ 1 lần cuối mới thật sự tổng hợp nội dung.

Tôi nghĩ: 12 lần gọi LLM chỉ để “nghĩ xem bước tiếp theo làm gì” khi mà toàn bộ task hoàn toàn có thể plan được từ đầu, đó là chi phí không cần thiết.

Plan-and-Execute là câu trả lời cho bài toán đó.

ReAct nhìn lại: một bước trước, không hơn

Bài trước, ReAct: thought, action, observation cycle, mô tả vòng lặp: Thought, Action, Observation. LLM được gọi sau mỗi observation để quyết định bước tiếp theo.

Cơ chế này hoạt động tốt khi task unpredictable: không biết trước observation nào trả về gì, không biết cần bao nhiêu bước, không biết bước sau phụ thuộc bước trước như thế nào.

Nhưng khi task có thể plan được: “tìm thông tin từ 5 nguồn, merge, viết summary 500 từ”, thì việc gọi LLM 12 lần để ra 12 quyết định đơn lẻ là lãng phí. LLM hoàn toàn có thể nhìn task đó một lần và ra ngay plan đầy đủ.

ReAct: LLM quyết định 1 bước sau mỗi observation. Plan-and-Execute: LLM quyết định N bước một lần, executor chạy từng bước một.

Anatomy của Plan-and-Execute

Hệ thống gồm hai thành phần rõ ràng.

Planner: nhận task, gọi LLM một lần, nhận về một plan là danh sách các bước có thứ tự. Planner không chạy gì cả. Nó chỉ nghĩ.

Executor: nhận plan, chạy từng bước. Mỗi bước có thể gọi tool, gọi API, gọi LLM nhỏ hơn. Executor không cần biết task gốc là gì. Nó chỉ biết bước hiện tại và cần làm gì.

Task
 │
 ▼
┌──────────┐
│  Planner  │  (1 LLM call)
└──────────┘
      │
      │  Plan: [step1, step2, step3, ...]
      ▼
┌──────────────────────────────────────┐
│              Executor                │
│  step1 → result1                     │
│  step2 (dùng result1) → result2      │
│  step3 (dùng result2) → result3      │
│  ...                                 │
└──────────────────────────────────────┘
      │
      ▼
   Final result

Điểm mấu chốt: planner chạy một lần, executor chạy N lần nhưng mỗi lần là một bước nhỏ, thường không cần LLM lớn.

Format của plan

Plan tốt nhất là một list có thứ tự với thông tin phụ thuộc rõ ràng. Khi gọi planner, bạn muốn nhận về cái gì đó như:

{
  "steps": [
    {
      "id": 1,
      "description": "Tìm kiếm thông tin về topic A từ nguồn X",
      "tool": "search",
      "args": {"query": "topic A", "source": "X"},
      "depends_on": []
    },
    {
      "id": 2,
      "description": "Tìm kiếm thông tin về topic A từ nguồn Y",
      "tool": "search",
      "args": {"query": "topic A", "source": "Y"},
      "depends_on": []
    },
    {
      "id": 3,
      "description": "Merge kết quả từ bước 1 và 2, viết summary 500 từ",
      "tool": "synthesize",
      "args": {"max_words": 500},
      "depends_on": [1, 2]
    }
  ]
}

Trường depends_on quan trọng: bước 1 và 2 không phụ thuộc nhau nên có thể chạy song song. Bước 3 cần kết quả của 1 và 2. Một executor thông minh sẽ dùng thông tin này để chạy parallel khi có thể.

Không phải lúc nào cũng cần format phức tạp như vậy. Với task đơn giản, list text thuần cũng đủ:

plan = [
    "Tìm kiếm topic A từ nguồn X",
    "Tìm kiếm topic A từ nguồn Y",
    "Tổng hợp hai kết quả trên thành 500 từ"
]

Mức phức tạp của plan format nên tương xứng với mức phức tạp của task.

Code: Planner-Executor với Anthropic SDK

Đây là implementation tối giản dùng claude-sonnet-4-6:

import json
import os
from anthropic import Anthropic
from typing import Any

client = Anthropic()

TOOLS = [
    {
        "name": "search",
        "description": "Search for information on a topic",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "source": {"type": "string", "enum": ["web", "docs", "internal"]}
            },
            "required": ["query"]
        }
    },
    {
        "name": "synthesize",
        "description": "Synthesize multiple texts into a summary",
        "input_schema": {
            "type": "object",
            "properties": {
                "texts": {"type": "array", "items": {"type": "string"}},
                "max_words": {"type": "integer"}
            },
            "required": ["texts"]
        }
    }
]


def execute_tool(name: str, args: dict) -> str:
    """Stub: thay bằng implementation thật."""
    if name == "search":
        return f"[Kết quả search '{args['query']}' từ {args.get('source', 'web')}]"
    if name == "synthesize":
        combined = " ".join(args["texts"])
        limit = args.get("max_words", 300)
        return f"[Summary ({limit} từ): {combined[:200]}...]"
    return f"Unknown tool: {name}"


def planner(task: str) -> list[dict]:
    """Gọi LLM một lần, trả về list of steps."""
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=(
            "Bạn là planner. Nhận task, trả về plan dạng JSON list.\n"
            "Mỗi step có: id (int), description (str), tool (str), args (dict), depends_on (list[int]).\n"
            "CHỈ trả về JSON, không thêm gì khác."
        ),
        messages=[{"role": "user", "content": f"Task: {task}"}],
    )
    raw = resp.content[0].text.strip()
    # Xử lý trường hợp LLM bọc trong ```json ... ```
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip()
    return json.loads(raw)


def executor(steps: list[dict]) -> dict[int, Any]:
    """Chạy từng step theo thứ tự, trả về map step_id -> result."""
    results: dict[int, Any] = {}
    # Sắp xếp theo id để đảm bảo dependencies chạy trước
    for step in sorted(steps, key=lambda s: s["id"]):
        step_id = step["id"]
        deps = step.get("depends_on", [])
        # Inject kết quả của các bước phụ thuộc vào args
        args = dict(step.get("args", {}))
        if deps:
            dep_results = [str(results[d]) for d in deps if d in results]
            # Nếu tool là synthesize, truyền dep results vào texts
            if step["tool"] == "synthesize":
                args["texts"] = dep_results
        result = execute_tool(step["tool"], args)
        results[step_id] = result
        print(f"  Step {step_id}: {step['description'][:50]}... -> OK")
    return results


def plan_and_execute(task: str) -> str:
    print(f"[Planner] Đang lên plan cho: {task}")
    steps = planner(task)
    print(f"[Planner] Plan: {len(steps)} bước")

    print("[Executor] Bắt đầu chạy...")
    results = executor(steps)

    # Kết quả cuối là output của bước cuối cùng
    last_id = max(results.keys())
    return str(results[last_id])


if __name__ == "__main__":
    task = "Tìm hiểu về Plan-and-Execute pattern trong AI agents từ 2 nguồn, tổng hợp thành 300 từ"
    output = plan_and_execute(task)
    print(f"\n[Kết quả]\n{output}")

Chạy ra trace như sau:

[Planner] Đang lên plan cho: Tìm hiểu về ...
[Planner] Plan: 3 bước
[Executor] Bắt đầu chạy...
  Step 1: Tìm kiếm Plan-and-Execute từ nguồn web... -> OK
  Step 2: Tìm kiếm Plan-and-Execute từ nguồn docs... -> OK
  Step 3: Tổng hợp kết quả bước 1 và 2... -> OK

[Kết quả]
[Summary (300 từ): ...]

Tổng số lần gọi LLM lớn: 1. Với ReAct cho cùng task này, con số đó là 4 đến 6.

Replan: khi thực tế không khớp plan

Đây là phần thú vị. Plan được tạo ra dựa trên thông tin tại thời điểm planner chạy. Nhưng trong quá trình execution, có thể xảy ra:

  • Tool trả về lỗi (API timeout, resource not found)
  • Kết quả của một bước không đủ để tiếp tục bước sau
  • Bước đang chạy phát hiện thêm thông tin khiến plan cũ không còn phù hợp

Đây là lúc cần replan: dừng execution, gọi planner lại với thêm context từ các bước đã chạy, lấy plan mới.

def executor_with_replan(steps: list[dict], task: str, max_replans: int = 2) -> dict[int, Any]:
    results: dict[int, Any] = {}
    replan_count = 0

    current_steps = sorted(steps, key=lambda s: s["id"])

    for step in current_steps:
        step_id = step["id"]
        args = dict(step.get("args", {}))
        deps = step.get("depends_on", [])

        try:
            if deps and step["tool"] == "synthesize":
                dep_results = [str(results[d]) for d in deps if d in results]
                args["texts"] = dep_results

            result = execute_tool(step["tool"], args)
            results[step_id] = result

        except Exception as err:
            print(f"  Step {step_id} failed: {err}")
            if replan_count >= max_replans:
                raise RuntimeError(f"Max replans exceeded at step {step_id}") from err

            # Gọi planner lại với context đã có
            context = f"Task: {task}\nCác bước đã hoàn thành: {json.dumps(results, ensure_ascii=False)}\nLỗi tại bước {step_id}: {err}\nHãy lên lại plan cho phần còn lại."
            new_steps_raw = planner(context)
            # Gán lại id để không xung đột với bước đã chạy
            offset = max(results.keys(), default=0)
            for s in new_steps_raw:
                s["id"] += offset
                s["depends_on"] = [d + offset for d in s.get("depends_on", [])]
            current_steps = sorted(new_steps_raw, key=lambda s: s["id"])
            replan_count += 1
            print(f"  Replanned: {len(current_steps)} bước mới")
            break  # Vòng for reset với current_steps mới

    return results

Replan làm tăng số lần gọi LLM nhưng vẫn ít hơn ReAct thuần trong nhiều trường hợp, vì chỉ replan khi có sự cố chứ không phải sau mỗi bước.

So sánh ReAct và Plan-and-Execute

Tiêu chíReActPlan-and-Execute
Số lần gọi LLM lớnN (mỗi bước 1 lần)1 (planner) + replan nếu có
Chi phí tokenCao nếu N nhiềuThấp hơn khi task predictable
Khả năng thích nghiCao: phản ứng theo từng observationThấp hơn: plan cứng cho đến khi replan
Phù hợp vớiTask dynamic, không đoán trước đượcTask có cấu trúc rõ, nhiều bước độc lập
DebugKhó: thought chain dàiDễ hơn: plan rõ ràng, trace từng step
LatencyCao (N lần LLM nối tiếp)Thấp hơn (1 lần LLM + parallel steps)

Không có cái nào “tốt hơn” tuyệt đối. Chọn dựa trên bản chất của task.

Ví dụ thực tế:

  • Dùng ReAct: agent debug lỗi trong hệ thống lạ, không biết trước cần kiểm tra gì. Mỗi observation có thể dẫn đến hướng khác nhau.
  • Dùng Plan-and-Execute: agent viết báo cáo tuần từ nhiều nguồn dữ liệu có sẵn. Task rõ, bước rõ, có thể plan trước.
  • Kết hợp: executor của Plan-and-Execute dùng ReAct cho các bước phức tạp. Planner quyết định “cần làm gì”, mỗi step là một mini-agent ReAct quyết định “làm như thế nào”.

LangGraph và Plan-and-Execute

LangGraph (được so sánh đầy đủ trong bài 19) có sẵn pattern này trong thư viện. Cấu trúc graph điển hình:

from langgraph.graph import StateGraph, END

# State chứa task, plan hiện tại, results từng bước
class AgentState(TypedDict):
    task: str
    plan: list[dict]
    current_step: int
    results: dict
    final_output: str

# Node: planner
def plan_node(state: AgentState) -> AgentState:
    steps = planner(state["task"])
    return {**state, "plan": steps, "current_step": 0}

# Node: execute một step
def execute_node(state: AgentState) -> AgentState:
    step = state["plan"][state["current_step"]]
    result = execute_tool(step["tool"], step.get("args", {}))
    new_results = {**state["results"], step["id"]: result}
    return {**state, "results": new_results, "current_step": state["current_step"] + 1}

# Edge condition: còn step không?
def should_continue(state: AgentState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "done"
    return "execute"

graph = StateGraph(AgentState)
graph.add_node("plan", plan_node)
graph.add_node("execute", execute_node)
graph.add_edge("plan", "execute")
graph.add_conditional_edges("execute", should_continue, {"execute": "execute", "done": END})
graph.set_entry_point("plan")

app = graph.compile()

LangGraph quản lý state và control flow. Code trên không tự replan. Để thêm replan, bạn thêm một node replan và edge điều kiện từ execute khi tool throw exception.

Lý do dùng framework như LangGraph: state management, checkpoint, streaming, và visualization của graph. Lý do không dùng (hoặc build từ đầu trước): framework che mất control flow, khó debug khi graph phức tạp. Series này recommend: hiểu cách tự build trước, dùng framework khi bạn thấy mình đang tự viết lại những gì framework đã có.

Pitfall: plan stale khi state thay đổi mid-execution

Đây là fail mode tôi thấy nhiều nhất với Plan-and-Execute.

Scenario: agent lên plan vào lúc 9 giờ sáng. Plan có bước “đọc file config.json để lấy database URL”. Trong khi executor đang chạy các bước trước, một người khác sửa file đó. Bước “đọc config” chạy, lấy về URL mới. Nhưng các bước sau trong plan giả định URL cũ, dẫn đến kết quả sai hoặc lỗi.

Plan được tạo ra dựa trên snapshot của world state tại thời điểm lên plan. World state có thể thay đổi.

Cách giảm thiểu:

  1. Giữ plan ngắn: plan 3-5 bước ít bị stale hơn plan 15 bước. Execution nhanh hơn thì window để state thay đổi hẹp hơn.
  2. Validate state trước execution: trước khi executor chạy bước đầu tiên, check các điều kiện tiên quyết: file tồn tại không, API endpoint live không, resource available không.
  3. Replan proactively: không chỉ replan khi lỗi, mà replan khi một step trả về kết quả “khác đáng kể so với dự kiến”. Cần định nghĩa “khác đáng kể” theo từng domain.
  4. Tránh plan hardcode runtime values: thay vì plan ghi “đọc URL từ config.json và dùng URL đó để connect”, plan chỉ ghi “connect to database”. Giá trị cụ thể được lấy lúc execute, không lúc plan.

Fail mode thứ hai: plan quá cụ thể cho inputs chưa có. Planner đôi khi hallucinate values trong args vì nó đang plan cho future state, không có data thực. Ví dụ:

{
  "id": 2,
  "description": "Phân tích kết quả từ bước 1",
  "tool": "analyze",
  "args": {"data": "[PLACEHOLDER FROM STEP 1]"}
}

Nếu planner không biết cách handle dynamic data, nó có thể điền vào args một giá trị hallucinated hoặc một placeholder không hợp lệ. Executor nhận được args sai và fail ngay từ đầu.

Fix: thiết kế plan format để args của step N+1 không cần biết output cụ thể của step N tại lúc plan. Thay vào đó, executor tự inject kết quả vào args khi chạy, dựa trên depends_on.

Cheatsheet

Khái niệmMô tả
PlannerGọi LLM một lần, ra plan đầy đủ
ExecutorChạy từng step của plan, không cần LLM lớn
ReplanGọi lại planner khi step fail hoặc kết quả lạ
depends_onKhai báo phụ thuộc để chạy parallel khi có thể
Plan stalePlan dựa trên state cũ, world đã thay đổi mid-execution
Dùng ReAct khiDùng Plan-and-Execute khi
Task dynamic, không đoán trướcTask có cấu trúc rõ, nhiều bước độc lập
Mỗi observation quyết định bước tiếpBiết trước cần làm gì từ đầu
N bước nhỏ, không song song đượcNhiều bước có thể chạy song song
Debug không quan trọng bằng flexibilityCần trace rõ ràng để debug và audit
PitfallFix
Plan stale khi state thay đổiValidate state trước execute, plan ngắn thôi
Planner hallucinate argsĐể executor inject dynamic values, không hardcode trong plan
Replan vô hạnGiới hạn max_replans, raise error sau đó
Plan quá chi tiếtPlan ở mức “what”, để executor quyết định “how”

Lời kết

Plan-and-Execute giải quyết một bài toán cụ thể: khi task đủ predictable để plan trước, việc gọi LLM sau mỗi bước là chi phí không cần thiết. Tách planning khỏi execution giúp giảm latency, giảm token, và tạo ra trace dễ debug hơn.

Pattern này không thay thế ReAct. Nó bổ sung. Nhiều agent production kết hợp cả hai: Plan-and-Execute ở level cao (planner quyết định các giai đoạn lớn), ReAct ở level thấp (mỗi giai đoạn là một mini-agent tự điều hướng).

Bài tiếp theo, Tree of Thoughts và tree search cho agent, đẩy planning lên một bước xa hơn: thay vì một plan tuyến tính, agent tạo ra nhiều nhánh kế hoạch song song, đánh giá từng nhánh, và chọn nhánh tốt nhất. Phức tạp hơn Plan-and-Execute nhiều, nhưng giải quyết được class task mà cả ReAct lẫn P&E đều không làm tốt.