บทความนี้จะพาทุกท่านไปทำความรู้จักกับ **Plan-and-Execute Agent** ซึ่งเป็นรูปแบบสถาปัตยกรรมที่ช่วยให้ LLM สามารถวางแผนงานที่ซับซ้อนได้อย่างมีประสิทธิภาพ พร้อมแชร์ประสบการณ์จากการสร้างระบบจริงที่ใช้งานใน Production
Plan-and-Execute คืออะไร
Plan-and-Execute เป็นรูปแบบการทำงานของ Agent ที่แยกกระบวนการออกเป็น 2 ขั้นตอนหลัก:
1. **Plan Phase** — วางแผนลำดับขั้นตอนทั้งหมดก่อนลงมือทำ
2. **Execute Phase** — ทำตามแผนที่วางไว้ทีละขั้นตอน
วิธีนี้แตกต่างจาก **ReAct** (Reason + Act) ที่คิดและทำไปพร้อมกันทีละขั้น เหมาะกับงานที่ต้องการความแม่นยำสูงและมีหลายขั้นตอน
เปรียบเทียบ 3 รูปแบบ Agent
| รูปแบบ | ข้อดี | ข้อเสีย | เหมาะกับ |
|--------|-------|---------|----------|
| **ReAct** | ยืดหยุ่น ตอบสนองเร็ว | อาจหลงทางในงานซับซ้อน | งาน QA ง่าย |
| **Plan-and-Execute** | แม่นยำ ควบคุมได้ดี | ใช้เวลามากกว่า | งานหลายขั้นตอน |
| **Tool-use** | เรียก API ได้ตรงจุด | ต้องประกาศ Tool ล่วงหน้า | งานเฉพาะทาง |
การสร้าง Plan-and-Execute Agent ด้วย HolySheep AI
จากการทดสอบในหลายโปรเจกต์ พบว่า **HolySheep AI** มีความหน่วงเฉลี่ย **<50ms** สำหรับการเรียก API ซึ่งทำให้การสร้าง Agent ที่ต้องเรียกหลายรอบทำได้อย่างรวดเร็ว และราคาประหยัดมาก เช่น GPT-4.1 อยู่ที่ **$8/MTok** เทียบกับ OpenAI ที่แพงกว่า 85% [สมัครที่นี่](https://www.holysheep.ai/register)
"""
Plan-and-Execute Agent Implementation
รองรับ Multi-Step Planning พร้อม Tool Calling
"""
import json
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
class AgentState(Enum):
PLANNING = "planning"
EXECUTING = "executing"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Task:
id: str
description: str
status: str = "pending"
result: Any = None
error: Optional[str] = None
@dataclass
class ExecutionResult:
task_id: str
success: bool
output: Any
latency_ms: float
error: Optional[str] = None
class PlanAndExecuteAgent:
"""
Plan-and-Execute Agent พร้อมระบบ Tool Calling
ใช้ HolySheep AI เป็น LLM Backend
"""
def __init__(
self,
api_key: str,
model: str = "gpt-4.1",
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.model = model
self.base_url = base_url
self.tools = {}
self.state = AgentState.PLANNING
self.plan: List[Task] = []
self.execution_history: List[ExecutionResult] = []
def register_tool(self, name: str, func: callable, description: str):
"""ลงทะเบียน Tool สำหรับ Agent ใช้งาน"""
self.tools[name] = {
"function": func,
"description": description
}
def _build_planning_prompt(self, user_request: str) -> str:
"""สร้าง Prompt สำหรับขั้นตอน Planning"""
tools_desc = "\n".join([
f"- {name}: {info['description']}"
for name, info in self.tools.items()
])
return f"""คุณเป็น Agent ที่ต้องวางแผนการทำงาน
คำขอของผู้ใช้: {user_request}
เครื่องมือที่มี:
{tools_desc}
ให้คุณวางแผนเป็นขั้นตอนโดย:
1. แบ่งงานเป็น Task เล็กๆ ที่ทำได้
2. ระบุว่าแต่ละ Task ใช้ Tool ใด
3. คืนค่า JSON ที่มีโครงสร้าง:
{{
"tasks": [
{{"id": "task_1", "description": "...", "tool": "tool_name"}},
...
],
"summary": "สรุปแผนการทำงาน"
}}
"""
def _call_llm(self, prompt: str) -> str:
"""เรียก LLM ผ่าน HolySheep API"""
import urllib.request
import urllib.error
data = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3
}
req = urllib.request.Request(
f"{self.base_url}/chat/completions",
data=json.dumps(data).encode('utf-8'),
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
method="POST"
)
start_time = time.time()
try:
with urllib.request.urlopen(req, timeout=30) as response:
latency = (time.time() - start_time) * 1000
result = json.loads(response.read().decode('utf-8'))
print(f"[DEBUG] LLM Latency: {latency:.2f}ms")
return result['choices'][0]['message']['content']
except Exception as e:
raise RuntimeError(f"LLM call failed: {e}")
def plan_tasks(self, user_request: str) -> List[Task]:
"""วางแผนงานจากคำขอของผู้ใช้"""
print(f"[AGENT] เริ่มวางแผน...")
self.state = AgentState.PLANNING
prompt = self._build_planning_prompt(user_request)
response = self._call_llm(prompt)
# Parse JSON response
try:
# ค้นหา JSON ใน response
start_idx = response.find('{')
end_idx = response.rfind('}') + 1
if start_idx != -1 and end_idx > start_idx:
json_str = response[start_idx:end_idx]
plan_data = json.loads(json_str)
else:
raise ValueError("No JSON found in response")
self.plan = [
Task(id=t['id'], description=t['description'])
for t in plan_data.get('tasks', [])
]
print(f"[AGENT] วางแผนเสร็จ: {len(self.plan)} ขั้นตอน")
return self.plan
except Exception as e:
print(f"[ERROR] Planning failed: {e}")
raise
def execute_plan(self) -> List[ExecutionResult]:
"""ดำเนินการตามแผนที่วางไว้"""
print(f"[AGENT] เริ่มดำเนินการ {len(self.plan)} ขั้นตอน...")
self.state = AgentState.EXECUTING
results = []
for i, task in enumerate(self.plan):
print(f"[AGENT] ขั้นตอนที่ {i+1}/{len(self.plan)}: {task.description}")
start_time = time.time()
try:
# หา Tool ที่ต้องใช้
task.tool_name = self._determine_tool(task.description)
if task.tool_name and task.tool_name in self.tools:
tool_func = self.tools[task.tool_name]['function']
task.result = tool_func(task.description)
task.status = "completed"
else:
# ถ้าไม่มี Tool ให้ใช้ LLM ตอบ
task.result = self._call_llm(task.description)
task.status = "completed"
latency = (time.time() - start_time) * 1000
result = ExecutionResult(
task_id=task.id,
success=True,
output=task.result,
latency_ms=latency
)
except Exception as e:
task.status = "failed"
task.error = str(e)
latency = (time.time() - start_time) * 1000
result = ExecutionResult(
task_id=task.id,
success=False,
output=None,
latency_ms=latency,
error=str(e)
)
results.append(result)
self.execution_history.append(result)
# แสดงผลลัพธ์
status_icon = "✅" if result.success else "❌"
print(f" {status_icon} เสร็จใน {result.latency_ms:.2f}ms")
self.state = AgentState.COMPLETED
return results
def _determine_tool(self, task_desc: str) -> Optional[str]:
"""ตัดสินใจเลือก Tool จากคำอธิบาย Task"""
keywords_map = {
"ค้นหา": "search",
"คำนวณ": "calculator",
"เขียน": "writer",
"แปล": "translator",
"สรุป": "summarizer"
}
for keyword, tool in keywords_map.items():
if keyword in task_desc and tool in self.tools:
return tool
return None
def run(self, user_request: str) -> Dict[str, Any]:
"""รัน Agent แบบ Complete: Plan -> Execute"""
print("="*50)
print(f"[AGENT] เริ่มทำงาน: {user_request[:50]}...")
print("="*50)
total_start = time.time()
# ขั้นตอนที่ 1: วางแผน
self.plan_tasks(user_request)
# ขั้นตอนที่ 2: ดำเนินการ
results = self.execute_plan()
total_time = (time.time() - total_start) * 1000
success_rate = sum(1 for r in results if r.success) / len(results) * 100
avg_latency = sum(r.latency_ms for r in results) / len(results)
print("="*50)
print(f"[AGENT] เสร็จสิ้น!")
print(f" - จำนวน Task: {len(results)}")
print(f" - อัตราความสำเร็จ: {success_rate:.1f}%")
print(f" - เวลารวม: {total_time:.2f}ms")
print(f" - หน่วงเฉลี่ยต่อ Task: {avg_latency:.2f}ms")
return {
"plan": self.plan,
"results": results,
"summary": {
"total_time_ms": total_time,
"success_rate": success_rate,
"avg_latency_ms": avg_latency
}
}
============ ตัวอย่างการใช้งาน ============
def search_tool(query: str) -> str:
"""Tool สำหรับค้นหาข้อมูล"""
return f"ผลการค้นหา '{query}': พบ 5 รายการ"
def calculator_tool(expr: str) -> str:
"""Tool สำหรับคำนวณ"""
try:
result = eval(expr)
return f"ผลลัพธ์: {result}"
except:
return "เกิดข้อผิดพลาดในการคำนวณ"
def translator_tool(text: str) -> str:
"""Tool สำหรับแปลภาษา"""
return f"[TH->EN] {text}"
if __name__ == "__main__":
# สร้าง Agent
agent = PlanAndExecuteAgent(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1"
)
# ลงทะเบียน Tools
agent.register_tool("search", search_tool, "ค้นหาข้อมูลในอินเทอร์เน็ต")
agent.register_tool("calculator", calculator_tool, "คำนวณนิพจน์ทางคณิตศาสตร์")
agent.register_tool("translator", translator_tool, "แปลภาษา")
# รัน Agent
result = agent.run("ค้นหาข้อมูลเกี่ยวกับ AI แล้วแปลเป็นภาษาอังกฤษ แล้วคำนวณความน่าจะเป็น")
การเพิ่มประสิทธิภาพด้วย Streaming และ Parallel Execution
สำหรับงานที่ต้องการความเร็วสูง สามารถใช้ Parallel Execution ได้โดยปรับโค้ดดังนี้:
"""
Plan-and-Execute Agent with Parallel Execution
เพิ่มความเร็วด้วย Concurrent Tasks
"""
import asyncio
import aiohttp
import json
import time
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor
class ParallelPlanExecuteAgent:
"""
Plan-and-Execute Agent รองรับ Parallel Execution
สำหรับ Task ที่ไม่ขึ้นกันกัน
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.executor = ThreadPoolExecutor(max_workers=5)
async def _async_llm_call(self, prompt: str, model: str = "gpt-4.1") -> str:
"""เรียก LLM แบบ Async สำหรับ HolySheep API"""
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3
}
start = time.time()
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers) as resp:
response = await resp.json()
latency = (time.time() - start) * 1000
print(f"[ASYNC] Request completed in {latency:.2f}ms")
return response['choices'][0]['message']['content']
def _sync_tool_call(self, tool_name: str, params: Dict) -> Any:
"""เรียก Tool แบบ Synchronous"""
# Mock Tool Execution
time.sleep(0.1) # จำลองการทำงาน
return f"Result from {tool_name}: {params}"
async def execute_parallel_tasks(
self,
tasks: List[Dict],
dependency_map: Dict[str, List[str]] = None
)
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง