ReAct Agent 模式详解与 Python 实现 — Từ Lý Thuyết Đến Thực Chiến

Kết luận trước: Nếu bạn đang tìm cách triển khai ReAct Agent với chi phí thấp nhất, đăng ký HolySheep AI là lựa chọn tối ưu — tỷ giá ¥1=$1 giúp tiết kiệm 85%+ so với API chính thức, độ trễ dưới 50ms, và hỗ trợ thanh toán qua WeChat/Alipay ngay lập tức.

ReAct Agent là gì và Tại sao cần triển khai ngay?

ReAct (Reasoning + Acting) là pattern kết hợp tư duy phản biện (Reasoning) với hành động thực tế (Acting) trong cùng một vòng lặp. Thay vì LLM đơn thuần trả lời, ReAct Agent sẽ:

• Thought — Suy nghĩ về bước tiếp theo cần làm
• Action — Thực thi hành động (gọi tool, truy vấn database, search web)
• Observation — Quan sát kết quả và quyết định bước tiếp theo

Bảng so sánh HolySheep AI với API chính thức và đối thủ

Từ bảng so sánh có thể thấy: HolySheep AI là lựa chọn duy nhất hỗ trợ thanh toán WeChat/Alipay không cần thẻ quốc tế, đồng thời giá cả cạnh tranh nhất thị trường với độ trễ thấp nhất.

Cài đặt môi trường và Cấu hình

# Cài đặt thư viện cần thiết
pip install openai python-dotenv requests aiohttp

Tạo file .env trong thư mục project
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
EOF

Kiểm tra kết nối nhanh
python -c "
import os
from openai import OpenAI
client = OpenAI(
    api_key=os.getenv('HOLYSHEEP_API_KEY'),
    base_url=os.getenv('HOLYSHEEP_BASE_URL')
)
models = client.models.list()
print('Kết nối thành công! Các model khả dụng:')
for m in models.data[:5]:
    print(f'  - {m.id}')
"

Triển khai ReAct Agent Core — Mã nguồn hoàn chỉnh

import os
import json
from openai import OpenAI
from dataclasses import dataclass, field
from typing import List, Dict, Callable, Optional
from enum import Enum

=== CẤU HÌNH HOLYSHEEP ===
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # KHÔNG dùng api.openai.com
)

class AgentAction(Enum):
    ANSWER = "answer"
    SEARCH = "search"
    CALCULATE = "calculate"
    QUERY_DB = "query_database"
    FINISH = "finish"

@dataclass
class Tool:
    name: str
    description: str
    function: Callable

@dataclass
class ReActStep:
    step_number: int
    thought: str
    action: str
    action_input: str
    observation: str
    is_final: bool = False

class ReActAgent:
    """
    ReAct Agent Implementation
    Pattern: Thought -> Action -> Observation -> Next Step
    """
    
    def __init__(self, model: str = "gpt-4.1", max_steps: int = 10):
        self.model = model
        self.max_steps = max_steps
        self.tools: Dict[str, Tool] = {}
        self.conversation_history: List[Dict] = []
        
    def register_tool(self, name: str, description: str, func: Callable):
        """Đăng ký tool cho agent sử dụng"""
        self.tools[name] = Tool(name=name, description=description, function=func)
        print(f"✓ Tool '{name}' đã được đăng ký")
    
    def _build_system_prompt(self) -> str:
        """Xây dựng system prompt với định nghĩa tools"""
        tools_def = []
        for name, tool in self.tools.items():
            tools_def.append(f"- {tool.name}: {tool.description}")
        
        tools_section = "\n".join(tools_def) if tools_def else "Không có tool nào"
        
        return f"""Bạn là một ReAct Agent thông minh.

Quy trình suy nghĩ (MỖI BƯỚC):
1. THOUGHT: Phân tích tình huống hiện tại và quyết định hành động
2. ACTION: Chọn một trong: {', '.join(self.tools.keys())} hoặc 'finish'
3. OBSERVATION: Kết quả từ action sẽ được cung cấp ở bước tiếp theo

Tools khả dụng:
{tools_section}

Luật quan trọng:
- Trả lời theo định dạng JSON: {{"thought": "...", "action": "...", "action_input": "..."}}
- Nếu đã có đủ thông tin, action = "finish" với câu trả lời trong action_input
- LUÔN suy nghĩ trước khi hành động"""

    def _parse_llm_response(self, response: str) -> Dict:
        """Parse response từ LLM thành dict"""
        try:
            # Thử parse JSON trực tiếp
            return json.loads(response)
        except:
            # Fallback: tìm JSON trong response
            import re
            json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL)
            if json_match:
                return json.loads(json_match.group())
            return {"thought": response, "action": "finish", "action_input": response}
    
    def _execute_tool(self, action: str, action_input: str) -> str:
        """Thực thi tool và trả về kết quả"""
        if action == "finish":
            return f"[FINAL ANSWER] {action_input}"
        
        if action in self.tools:
            try:
                result = self.tools[action].function(action_input)
                return str(result)
            except Exception as e:
                return f"[ERROR] Tool execution failed: {str(e)}"
        
        return f"[ERROR] Unknown action: {action}"
    
    def run(self, question: str, verbose: bool = True) -> str:
        """
        Chạy ReAct Agent để trả lời câu hỏi
        
        Args:
            question: Câu hỏi đầu vào
            verbose: In chi tiết từng bước
        
        Returns:
            Câu trả lời cuối cùng
        """
        if verbose:
            print(f"\n{'='*60}")
            print(f"🎯 Câu hỏi: {question}")
            print(f"{'='*60}\n")
        
        context = question
        steps_history = []
        
        for step in range(1, self.max_steps + 1):
            # Gọi LLM để quyết định action
            messages = [
                {"role": "system", "content": self._build_system_prompt()},
                {"role": "user", "content": f"Câu hỏi: {question}\n\nNgữ cảnh từ các bước trước:\n{context}"}
            ]
            
            response = client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.3,
                max_tokens=500
            )
            
            llm_output = response.choices[0].message.content
            parsed = self._parse_llm_response(llm_output)
            
            thought = parsed.get("thought", "")
            action = parsed.get("action", "finish")
            action_input = parsed.get("action_input", "")
            
            if verbose:
                print(f"📌 Bước {step}:")
                print(f"   Thought: {thought[:100]}..." if len(thought) > 100 else f"   Thought: {thought}")
                print(f"   Action: {action}")
            
            # Thực thi action
            observation = self._execute_tool(action, action_input)
            
            if verbose:
                obs_display = observation[:150] + "..." if len(observation) > 150 else observation
                print(f"   Observation: {obs_display}\n")
            
            # Kiểm tra nếu là bước cuối
            if action == "finish":
                if verbose:
                    print(f"{'='*60}")
                    print(f"✅ Kết quả: {observation.replace('[FINAL ANSWER] ', '')}")
                    print(f"{'='*60}")
                return observation.replace("[FINAL ANSWER] ", "")
            
            # Cập nhật context cho bước tiếp theo
            context += f"\n\n[Bước {step}]\nThought: {thought}\nAction: {action}({action_input})\nObservation: {observation}"
        
        return "Đã đạt max_steps, không có kết luận."


=== DEMO SỬ DỤNG ===
if __name__ == "__main__":
    # Khởi tạo agent
    agent = ReActAgent(model="gpt-4.1", max_steps=8)
    
    # Đăng ký các tools
    def search_wikipedia(query: str) -> str:
        """Tool tìm kiếm Wikipedia (mock)"""
        db = {
            "ai": "AI - Trí tuệ nhân tạo, ngành khoa học máy tính...",
            "python": "Python - Ngôn ngữ lập trình bậc cao...",
            "react": "ReAct - Pattern kết hợp Reasoning và Acting..."
        }
        return db.get(query.lower(), f"Không tìm thấy kết quả cho: {query}")
    
    def calculate(expression: str) -> str:
        """Tool tính toán"""
        try:
            result = eval(expression)
            return f"{expression} = {result}"
        except:
            return f"Lỗi tính toán: {expression}"
    
    def get_weather(city: str) -> str:
        """Tool lấy thời tiết (mock)"""
        weather_db = {
            "hanoi": "25°C, nắng",
            "hcm": "32°C, mưa rào",
            "tokyo": "18°C, âm u"
        }
        return weather_db.get(city.lower(), f"Không có dữ liệu thời tiết cho: {city}")
    
    # Đăng ký tools
    agent.register_tool("search", "Tìm kiếm thông tin trên Wikipedia", search_wikipedia)
    agent.register_tool("calculate", "Tính toán biểu thức toán học", calculate)
    agent.register_tool("weather", "Lấy thông tin thời tiết của thành phố", get_weather)
    
    # Chạy demo
    print("🚀 Demo ReAct Agent với HolySheep AI\n")
    
    question = "Thời tiết ở Hanoi như thế nào? Và tính 15 + 27 bằng bao nhiêu?"
    result = agent.run(question)

Triển khai Multi-Agent ReAct System

import asyncio
import aiohttp
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import json

=== MULTI-AGENT ORCHESTRATOR ===
@dataclass
class AgentMessage:
    sender: str
    receiver: str
    content: Any
    timestamp: datetime = None
    msg_type: str = "request"
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now()

class MultiAgentReActSystem:
    """
    Hệ thống Multi-Agent với pattern ReAct
    - Planner Agent: Phân tích và lập kế hoạch
    - Research Agent: Tìm kiếm thông tin
    - Execution Agent: Thực thi tác vụ
    - Validator Agent: Kiểm tra kết quả
    """
    
    def __init__(self):
        self.agents: Dict[str, Any] = {}
        self.message_queue: List[AgentMessage] = []
        
    def register_agent(self, name: str, agent_class):
        """Đăng ký agent vào hệ thống"""
        self.agents[name] = agent_class
        print(f"✓ Agent '{name}' đã tham gia hệ thống")
    
    async def _send_message(self, sender: str, receiver: str, content: Any, msg_type: str = "request"):
        """Gửi message giữa các agents"""
        msg = AgentMessage(sender=sender, receiver=receiver, content=content, msg_type=msg_type)
        self.message_queue.append(msg)
        return msg
    
    async def _broadcast(self, sender: str, content: Any):
        """Broadcast message tới tất cả agents"""
        tasks = []
        for agent_name in self.agents:
            if agent_name != sender:
                tasks.append(self._send_message(sender, agent_name, content))
        await asyncio.gather(*tasks)
    
    async def run_task(self, task: str) -> Dict[str, Any]:
        """
        Chạy task thông qua multi-agent coordination
        
        Workflow:
        1. Planner phân tích task
        2. Research tìm thông tin cần thiết
        3. Execution thực hiện
        4. Validator kiểm tra
        """
        print(f"\n{'#'*60}")
        print(f"🎯 TASK: {task}")
        print(f"{'#'*60}")
        
        # Bước 1: Planning
        planning_result = await self._planner_phase(task)
        
        # Bước 2: Research (nếu cần)
        research_result = await self._research_phase(planning_result)
        
        # Bước 3: Execution
        execution_result = await self._execution_phase(research_result)
        
        # Bước 4: Validation
        validation_result = await self._validation_phase(execution_result)
        
        return {
            "task": task,
            "planning": planning_result,
            "research": research_result,
            "execution": execution_result,
            "validation": validation_result,
            "final_output": validation_result.get("output"),
            "success": validation_result.get("is_valid", False)
        }
    
    async def _planner_phase(self, task: str) -> Dict:
        """Planner phân tích và lập kế hoạch"""
        print("\n📋 [PLANNER] Đang phân tích task...")
        
        # Gọi LLM qua HolySheep
        response = client.chat.completions.create(
            model="gpt-4.1",
            messages=[
                {"role": "system", "content": "Bạn là Planner Agent. Phân tích task và đưa ra kế hoạch thực hiện."},
                {"role": "user", "content": f"Phân tích task sau và đề xuất các bước thực hiện:\n{task}"}
            ]
        )
        
        plan = response.choices[0].message.content
        
        await self._broadcast("planner", {"type": "plan_created", "content": plan})
        
        return {"plan": plan, "steps": self._extract_steps(plan)}
    
    async def _research_phase(self, planning_result: Dict) -> Dict:
        """Research agent tìm kiếm thông tin cần thiết"""
        print("🔍 [RESEARCHER] Đang tìm kiếm thông tin...")
        
        steps = planning_result.get("steps", [])
        research_results = []
        
        for step in steps[:3]:  # Giới hạn 3 bước đầu
            response = client.chat.completions.create(
                model="gpt-4.1",
                messages=[
                    {"role": "system", "content": "Bạn là Research Agent. Tìm kiếm và tổng hợp thông tin liên quan."},
                    {"role": "user", "content": f"Tìm hiểu về: {step}"}
                ]
            )
            research_results.append({
                "topic": step,
                "findings": response.choices[0].message.content
            })
        
        await self._broadcast("researcher", {"type": "research_complete", "count": len(research_results)})
        
        return {"research_results": research_results}
    
    async def _execution_phase(self, research_result: Dict) -> Dict:
        """Execution agent thực hiện task"""
        print("⚙️ [EXECUTOR] Đang thực thi...")
        
        research = research_result.get("research_results", [])
        
        # Tổng hợp findings
        combined = "\n".join([r.get("findings", "") for r in research])
        
        response = client.chat.completions.create(
            model="gpt-4.1",
            messages=[
                {"role": "system", "content": "Bạn là Execution Agent. Tổng hợp thông tin và đưa ra kết quả thực thi."},
                {"role": "user", "content": f"Dựa trên các nghiên cứu sau, hãy t
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
AI Scientist: Cuộc cách mạng trong nghiên cứu khoa học tự độ
Hướng Dẫn Toàn Diện API AI Cho Nhà Phát Triển Nhật Bản: Than
AI API 错误日志分析：ELK Stack 集成教程