AI Agent Tool Calling Framework: So Sánh ReAct vs Plan-and-Execute Chi Tiết

Mở đầu: Khi AI Agent của tôi rơi vào vòng lặp vô hạn

Một ngày đẹp trời, hệ thống AI Agent xử lý đơn hàng của tôi bắt đầu gửi email spam liên tục đến khách hàng. Sau 3 giờ debug căng thẳng, tôi phát hiện nguyên nhân: framework ReAct mặc định của LangChain đang gọi function send_email() tới 47 lần cho một đơn hàng duy nhất, do thiếu cơ chế dừng khi đã hoàn thành mục tiêu. Kịch bản lỗi cụ thể:

Lỗi 1: Infinite loop trong ReAct Agent
ToolLoopError: Maximum iterations (10) exceeded
  at step 4: send_email() returned {"status": "success", "sent": true}
  at step 5: send_email() called again - Agent didn't recognize task completion
  at step 6-10: Repeated failed attempts to find new tools
  Final: TimeoutError after 45 seconds

Bài học đắt giá: **Việc chọn sai framework tool calling có thể khiến chi phí API tăng 500% và làm sập hệ thống production.** Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi so sánh hai framework phổ biến nhất: **ReAct (Reasoning + Acting)** và **Plan-and-Execute**, giúp bạn chọn đúng framework cho từng use case cụ thể.

ReAct Framework: Ưu điểm, nhược điểm và code mẫu

ReAct là gì?

ReAct (Synergizing Reasoning, Acting, and Planning) là pattern kết hợp **reasoning chain** với **action execution** trong từng step. Agent suy nghĩ một bước, hành động một bước, rồi quan sát kết quả trước khi tiếp tục.

Khi nào nên dùng ReAct?

ReAct phù hợp với các tác vụ:

• Tương tác real-time cần phản hồi nhanh
• Tác vụ đơn giản, ít bước (3-5 steps)
• Khi cần debug dễ dàng từng action
• Use case cần human-in-the-loop để approve

Code mẫu ReAct với HolySheep AI

import requests
import json

class ReActAgent:
    def __init__(self, api_key):
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.max_iterations = 15
        self.tools = {
            "search_database": self.search_database,
            "calculate": self.calculate,
            "send_notification": self.send_notification
        }
    
    def think_and_act(self, user_query):
        """ReAct Loop: Think -> Act -> Observe -> Repeat"""
        context = {"history": [], "observations": []}
        
        for iteration in range(self.max_iterations):
            # Bước 1: Gọi LLM để reason + quyết định action
            response = self.call_llm_for_action(user_query, context)
            
            # Bước 2: Parse response thành action
            action = response.get("action")
            params = response.get("parameters", {})
            
            if action == "finish":
                return response.get("result")
            
            if action not in self.tools:
                context["observations"].append(
                    f"Unknown tool: {action}"
                )
                continue
            
            # Bước 3: Execute action và observe kết quả
            try:
                result = self.tools[action](**params)
                context["observations"].append(
                    f"Executed {action} -> {result}"
                )
                context["history"].append({
                    "action": action,
                    "params": params,
                    "result": result
                })
            except Exception as e:
                context["observations"].append(
                    f"Error in {action}: {str(e)}"
                )
        
        return {"status": "max_iterations", "context": context}
    
    def call_llm_for_action(self, query, context):
        """Gọi DeepSeek V3.2 qua HolySheep - chi phí chỉ $0.42/MTok"""
        prompt = f"""Bạn là ReAct Agent. Với query: "{query}"
        
Context:
{json.dumps(context, indent=2)}

Chọn action tiếp theo từ: search_database, calculate, send_notification, finish
Trả về JSON: {{"action": "tên_action", "parameters": {{}}, "reasoning": "giải thích"}}
"""
        
        payload = {
            "model": "deepseek-v3.2",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3,
            "max_tokens": 500
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 401:
            raise Exception("API key không hợp lệ - kiểm tra YOUR_HOLYSHEEP_API_KEY")
        
        result = response.json()
        return json.loads(result["choices"][0]["message"]["content"])
    
    def search_database(self, query, table="orders"):
        # Mock implementation
        return {"found": 3, "records": [...]}
    
    def calculate(self, expression):
        return {"result": eval(expression)}
    
    def send_notification(self, user_id, message):
        # Mock implementation
        return {"sent": True, "user_id": user_id}

Sử dụng
agent = ReActAgent("YOUR_HOLYSHEEP_API_KEY")
result = agent.think_and_act("Tìm đơn hàng #12345 và gửi thông báo cho khách")
print(result)

Plan-and-Execute Framework: Kiến trúc và code mẫu

Plan-and-Execute là gì?

Plan-and-Execute chia quy trình thành **2 giai đoạn rõ ràng**:

1. **Plan Phase**: LLM phân tích toàn bộ task, tạo execution plan với các bước cụ thể
2. **Execute Phase**: Thực thi từng bước theo plan, có thể adjust plan nếu cần

Khi nào nên dùng Plan-and-Execute?

Plan-and-Execute phù hợp với:

• Tác vụ phức tạp, nhiều bước (10+ steps)
• Khi cần tối ưu hóa thứ tự execution
• Multi-agent systems cần phân công công việc
• Long-horizon planning với nhiều dependencies

Code mẫu Plan-and-Execute với HolySheep AI

import requests
import json
from typing import List, Dict, Any

class PlanAndExecuteAgent:
    def __init__(self, api_key):
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.tools = {
            "fetch_data": self.fetch_data,
            "process_data": self.process_data,
            "save_result": self.save_result,
            "send_email": self.send_email
        }
    
    def run(self, task: str) -> Dict[str, Any]:
        """Main entry point cho Plan-and-Execute"""
        
        # === PHASE 1: PLANNING ===
        print("🔍 Phase 1: Tạo Plan...")
        plan = self.create_plan(task)
        print(f"📋 Plan có {len(plan['steps'])} bước")
        
        # === PHASE 2: EXECUTION ===
        print("⚡ Phase 2: Thực thi Plan...")
        execution_results = []
        
        for idx, step in enumerate(plan["steps"]):
            print(f"\n📌 Bước {idx + 1}/{len(plan['steps'])}: {step['action']}")
            
            try:
                # Merge context từ các bước trước
                context = {
                    "task": task,
                    "plan": plan,
                    "previous_results": execution_results
                }
                
                result = self.execute_step(step, context)
                execution_results.append({
                    "step": idx + 1,
                    "action": step["action"],
                    "status": "success",
                    "result": result
                })
                
                # Kiểm tra xem có cần adjust plan không
                if step.get("adjust_plan_if"):
                    if self.should_adjust_plan(result, step["adjust_plan_if"]):
                        print("🔄 Adjusting plan based on intermediate result...")
                        remaining_steps = plan["steps"][idx + 1:]
                        adjustment = self.create_adjusted_plan(
                            result, remaining_steps, context
                        )
                        plan["steps"] = plan["steps"][:idx + 1] + adjustment
                        print(f"📋 Plan updated: {len(plan['steps'])} steps total")
                
            except Exception as e:
                execution_results.append({
                    "step": idx + 1,
                    "action": step["action"],
                    "status": "error",
                    "error": str(e)
                })
                
                if step.get("fallback_action"):
                    print(f"⚠️ Falling back to: {step['fallback_action']}")
                    fallback_result = self.execute_step(
                        {"action": step["fallback_action"], 
                         "parameters": step.get("fallback_params", {})},
                        context
                    )
                    execution_results[-1]["fallback_result"] = fallback_result
        
        return {
            "plan": plan,
            "execution_results": execution_results,
            "final_result": self.aggregate_results(execution_results)
        }
    
    def create_plan(self, task: str) -> Dict[str, Any]:
        """Gọi LLM để tạo execution plan chi tiết"""
        
        prompt = f"""Bạn là Planner Agent. Với task sau:
        
TASK: {task}

Tạo execution plan chi tiết với format JSON:
{{
  "task_summary": "Tóm tắt 1 câu về task",
  "steps": [
    {{
      "step_number": 1,
      "action": "tên_action",
      "parameters": {{"key": "value"}},
      "expected_output": "Kết quả mong đợi",
      "adjust_plan_if": "điều kiện để adjust plan",
      "fallback_action": "action dự phòng nếu lỗi"
    }}
  ],
  "dependencies": "Mô tả phụ thuộc giữa các bước"
}}

Các action khả dụng: {list(self.tools.keys())}
"""
        
        payload = {
            "model": "deepseek-v3.2",  # $0.42/MTok - chi phí thấp nhất
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.2,
            "max_tokens": 1000
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code != 200:
            raise ConnectionError(f"API Error: {response.status_code}")
        
        content = response.json()["choices"][0]["message"]["content"]
        
        # Parse JSON từ response
        try:
            # Handle markdown code blocks if present
            if "```json" in content:
                content = content.split("``json")[1].split("``")[0]
            elif "```" in content:
                content = content.split("``")[1].split("``")[0]
            
            return json.loads(content.strip())
        except json.JSONDecodeError as e:
            raise ValueError(f"Failed to parse plan JSON: {e}")
    
    def execute_step(self, step: Dict, context: Dict) -> Any:
        """Thực thi một bước trong plan"""
        action = step["action"]
        params = step.get("parameters", {})
        
        if action not in self.tools:
            raise ValueError(f"Unknown action: {action}")
        
        # Inject context vào parameters nếu cần
        if "{previous_result}" in str(params):
            last_result = context["previous_results"][-1]["result"] \
                if context["previous_results"] else None
            params = json.loads(
                str(params).replace("{previous_result}", json.dumps(last_result))
            )
        
        return self.tools[action](**params)
    
    def fetch_data(self, source: str, query: str = None):
        """Mock implementation - thay bằng API thực"""
        return {"source": source, "data": [...], "count": 100}
    
    def process_data(self, data: list, operation: str):
        """Mock implementation"""
        if operation == "filter":
            return [d for d in data if d.get("active")]
        return data
    
    def save_result(self, data: Any, destination: str):
        """Mock implementation"""
        return {"saved": True, "location": destination}
    
    def send_email(self, recipient: str, subject: str, body: str):
        """Mock implementation"""
        return {"sent": True, "recipient": recipient}
    
    def should_adjust_plan(self, result: Any, condition: str) -> bool:
        """Kiểm tra điều kiện điều chỉnh plan"""
        # Simplified check - có thể mở rộng với LLM
        return False
    
    def create_adjusted_plan(self, result, remaining_steps, context):
        """Tạo plan điều chỉnh dựa trên kết quả"""
        return remaining_steps
    
    def aggregate_results(self, results: List[Dict]) -> Any:
        """Tổng hợp kết quả từ tất cả các bước"""
        return {"completed_steps": len(results), "results": results}


Sử dụng
agent = PlanAndExecuteAgent("YOUR_HOLYSHEEP_API_KEY")
result = agent.run(
    "Tìm tất cả đơn hàng chưa xử lý, lọc đơn trên 1000$, "
    "tạo báo cáo và gửi email cho manager"
)
print(json.dumps(result, indent=2, default=str))

So Sánh Chi Tiết: ReAct vs Plan-and-Execute

Tiêu chí	ReAct	Plan-and-Execute
Độ phức tạp task	Thấp-Trung bình (3-10 steps)	Cao (10+ steps)
Tốc độ phản hồi	Nhanh, real-time	Chậm hơn do planning phase
Chi phí API calls	1 call/step (nhiều LLM calls)	1 call cho plan + N calls execution
Độ tin cậy	Có thể lặp vô hạn nếu không giới hạn	Cao - plan được review trước
Debugging	Dễ - từng step rõ ràng	Khó hơn - cần track cả plan và execution
Parallel execution	Không hỗ trợ tốt	Hỗ trợ tốt cho independent steps
Error recovery	Xử lý từng bước	Có fallback và adjust plan

Phù hợp với ai?

Nên chọn ReAct khi:

• 🎯 **Chatbot đơn giản**: FAQ bot, order tracking, basic customer support
• 🎯 **Prototyping nhanh**: Cần demo trong 1-2 ngày
• 🎯 **Human-in-the-loop**: Cần người duyệt trước khi action quan trọng
• 🎯 **Single-task focus**: Mỗi lần chỉ làm 1 việc cụ thể
• 🎯 **Tài nguyên hạn chế**: Server có RAM thấp, cần response nhanh

Nên chọn Plan-and-Execute khi:

• 🎯 **Workflow phức tạp**: Báo cáo tài chính, data pipeline, ETL
• 🎯 **Multi-agent systems**: Phân chia task cho nhiều agent
• 🎯 **Batch processing**: Xử lý hàng ngàn records
• 🎯 **Mission-critical tasks**: Cần validate plan trước khi execute
• 🎯 **Cost optimization**: Có thể cache và reuse plan

Giá và ROI: Tính toán chi phí thực tế

Với mỗi task phức tạp xử lý 1000 requests/ngày:

Framework	Model	Giá/MTok	Tok/request (ước tính)	Chi phí/ngày	Chi phí/tháng
ReAct	DeepSeek V3.2	$0.42	2,000	$0.84	$25.20
ReAct	GPT-4.1	$8.00	2,000	$16.00	$480.00
Plan-and-Execute	DeepSeek V3.2	$0.42	3,500 (plan + exec)	$1.47	$44.10
Plan-and-Execute	Claude Sonnet 4.5	$15.00	3,500	$52.50	$1,575.00

💡 Kết luận: Dùng DeepSeek V3.2 qua HolySheep với tỷ giá ¥1=$1 giúp tiết kiệm 85-95% chi phí so với OpenAI hoặc Anthropic. Với team startup, đây là sự chênh lệch giữa $25/tháng và $480/tháng.

Vì sao chọn HolySheep AI cho AI Agent Development

• 💰 Tiết kiệm 85%+: Tỷ giá ¥1=$1, giá DeepSeek V3.2 chỉ $0.42/MTok
• ⚡ Latency <50ms: Response nhanh, phù hợp real-time applications
• 💳 Thanh toán linh hoạt: Hỗ trợ WeChat Pay, Alipay, Visa, Mastercard
• 🎁 Tín dụng miễn phí: Đăng ký nhận credits để test không giới hạn
• 🔧 Tương thích API: Có thể thay thế OpenAI/Anthropic với base_url mới
• 📊 Dashboard rõ ràng: Theo dõi usage và chi phí dễ dàng

Lỗi thường gặp và cách khắc phục

Lỗi 1: ConnectionError - Timeout khi gọi API

# ❌ CÁCH SAI - Không handle timeout
response = requests.post(url, json=payload)  # Có thể treo vĩnh viễn

✅ CÁCH ĐÚNG - Thêm timeout và retry logic
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def call_with_retry(url, payload, headers, max_retries=3):
    session = requests.Session()
    
    # Retry strategy: backoff exponential
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=1,  # 1s, 2s, 4s
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    for attempt in range(max_retries):
        try:
            response = session.post(
                url,
                json=payload,
                headers=headers,
                timeout=(10, 30)  # (connect_timeout, read_timeout)
            )
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.Timeout:
            print(f"⏰ Timeout at attempt {attempt + 1}/{max_retries}")
            if attempt == max_retries - 1:
                raise ConnectionError("All retries exhausted - check network")
                
        except requests.exceptions.ConnectionError as e:
            print(f"🔌 Connection error: {e}")
            time.sleep(2 ** attempt)  # Exponential backoff
            
    raise ConnectionError("Failed after all retries")

Lỗi 2: 401 Unauthorized - Invalid API Key

# ❌ CÁCH SAI - Hardcode key trực tiếp
api_key = "sk-xxxxxxxxxxxxx"  # Security risk!

✅ CÁCH ĐÚNG - Sử dụng environment variable + validation
import os
from dotenv import load_dotenv

load_dotenv()

def get_validated_api_key():
    api_key = os.environ.get("HOLYSHEEP_API_KEY")
    
    if not api_key:
        raise ValueError(
            "HOLYSHEEP_API_KEY not found. "
            "Set it via: export HOLYSHEEP_API_KEY='your_key'"
        )
    
    # Validate key format (HolySheep keys thường bắt đầu với prefix)
    if len(api_key) < 20:
        raise ValueError("Invalid API key format")
    
    return api_key

Validate trước khi sử dụng
BASE_URL = "https://api.holysheep.ai/v1"
HEADERS = {
    "Authorization": f"Bearer {get_validated_api_key()}",
    "Content-Type": "application/json"
}

Test connection
def verify_api_key():
    try:
        response = requests.get(
            f"{BASE_URL}/models",
            headers=HEADERS,
            timeout=10
        )
        if response.status_code == 401:
            raise PermissionError(
                "Invalid API key - please check at "
                "https://www.holysheep.ai/register"
            )
        return True
    except Exception as e:
        print(f"❌ API verification failed: {e}")
        return False

Lỗi 3: ToolLoopError - Infinite loop trong ReAct

# ❌ CÁCH SAI - Không có mechanism dừng
def react_loop_no_limit(query):
    while True:
        action = llm_decide_action(query)
        result = execute(action)
        # Vòng lặp vô hạn!

✅ CÁCH ĐÚNG - Với early termination và completion check
from dataclasses import dataclass
from typing import Optional

@dataclass
class ExecutionState:
    iteration: int
    max_iterations: int
    completed_actions: list
    final_result: Optional[Any] = None

def react_loop_with_guardrails(query, max_iterations=15):
    """ReAct với guardrails chống infinite loop"""
    state = ExecutionState(
        iteration=0,
        max_iterations=max_iterations,
        completed_actions=[]
    )
    
    while state.iteration < state.max_iterations:
        state.iteration += 1
        
        # 1. LLM quyết định action
        decision = llm_decide_action(query, state)
        
        # 2. Kiểm tra explicit finish
        if decision.action == "finish":
            print(f"✅ Task completed at iteration {state.iteration}")
            return decision.result
        
        # 3. Kiểm tra implicit completion (trùng action gần đây)
        if is_duplicate_action(decision.action, state.completed_actions[-3:]):
            print(f"⚠️ Duplicate action detected - checking completion...")
            if check_task_completion(query, state):
                return state.final_result or "Completed"
        
        # 4. Execute action
        result = execute_action(decision.action, decision.params)
        
        # 5. Update state
        state.completed_actions.append(decision.action)
        
        if decision.action in ["save_data", "send_response", "create_report"]:
            state.final_result = result
        
        # 6. Log cho debugging
        print(f"[{state.iteration}/{state.max_iterations}] "
              f"{decision.action} -> {result}")
    
    # Max iterations reached
    print(f"⚠️ Max iterations ({max_iterations}) reached")
    return {
        "status": "max_iterations_exceeded",
        "completed_actions": state.completed_actions,
        "partial_result": state.final_result
    }

def is_duplicate_action(current, recent_actions, threshold=2):
    """Kiểm tra action có bị lặp lại liên tiếp không"""
    return recent_actions.count(current) >= threshold

def check_task_completion(query, state):
    """LLM check xem task đã hoàn thành chưa"""
    # Gọi LLM nhẹ để verify
    return True  # Simplified

Lỗi 4: Memory/Context overflow với long conversations

# ❌ CÁCH SAI - Giữ toàn bộ history
all_messages = []  # Memory leak!

while True:
    response = llm(all_messages)  # Context ngày càng lớn

✅ CÁCH ĐÚNG - Context summarization và truncation
from collections import deque

class ConversationManager:
    def __init__(self, max_messages=20, summarize_after=15):
        self.messages = deque(maxlen=max_messages)
        self.summarize_after = summarize_after
        self.summary = None
    
    def add_message(self, role, content):
        self.messages.append({"role": role, "content": content})
        
        if len(self.messages) >= self.summarize_after:
            self.summarize()
    
    def summarize(self):
        """Tóm tắt conversation cũ để tiết kiệm context"""
        old_messages = list(self.messages)[:-5]  # Giữ 5 message gần nhất
        
        if not old_messages:
            return
        
        prompt = f"""Summarize this conversation briefly:
        {old_messages}
        
        Return: "Summary: [brief summary]" """
        
        # Gọi LLM nhẹ để summarize
        response = call_holysheep({
            "model": "deepseek-v3.2",
            "messages": [{"role": "user", "content": prompt}]
        })
        
        self.summary = response["content"]
        self.messages = deque(
            [{"role": "system", "content": f"Summary: {self.summary}"}] +
            list(self.messages)[-5:],
            maxlen=self.maxlen
        )
    
    def get_context(self):
        """Trả về context cho LLM call"""
        return list(self.messages)
    
    def clear(self):
        self.messages.clear()
        self.summary = None

Kết luận và Khuyến nghị

Qua quá trình thực chiến với nhiều dự án AI Agent, tôi đã rút ra:

1. Chọn ReAct khi bạn cần prototyping nhanh, task đơn giản, hoặc cần human-in-the-loop. Nhược điểm lớn nhất là có thể rơi vào infinite loop nếu không implement guardrails đúng cách.

2. Chọn Plan-and-Execute khi task phức tạp, cần reliability cao, và có budget để đầu tư vào architecture tốt. Chi phí planning phase có thể đáng giá nếu task cần execute nhiều lần.

3. Luôn dùng guardrails: Timeout, max iterations, duplicate detection, và explicit finish conditions là must-have cho production.

4. Tối ưu chi phí: DeepSeek V3.2 qua HolySheep với giá $0.42/MTok là lựa chọn tối ưu nhất cho AI Agent workloads. Tiết kiệm 85%+ so với GPT-4.1 mà chất lượng reasoning vẫn đủ tốt cho hầu hết use cases.

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký