ReAct Agent
ReAct + Chatbot ด้วย Python
ในโลกที่ AI ถูกใช้แก้ปัญหาซับซ้อนมากขึ้น การให้ LLM แค่ตอบคำถาม อย่างเดียวอาจยังไม่พอ เพราะมันไม่สามารถรู้ข้อมูลใหม่ หรือตรวจสอบความถูกต้องของคำตอบได้ด้วยตัวเอง นี่จึงเป็นจุดเริ่มต้นของแนวคิด ReAct — Reason + Act ที่ทำให้ AI ไม่เพียง “คิดวิเคราะห์” แต่ยัง “ลงมือทำ” ได้จริง เช่น ค้นข้อมูล คำนวณ หรือใช้เครื่องมือภายนอก เพื่อให้ได้คำตอบที่ถูกต้องและมีเหตุผลมากขึ้น
มาทำความรู้จักกับ Concept ของ ReAct Agent กันแบบคร่าวๆ ว่าคืออะไร และมีประโยชน์อย่างไร สามารถนำไปประยุกต์ใช้ทำอะไรได้บ้าง
ถ้าเราพูดถึง “Agent” ในโลกของ AI ปัจจุบัน หมายถึงระบบที่สามารถ “คิดและลงมือทำ” ได้ด้วยตัวเอง โดยมีเป้าหมายหรือภารกิจบางอย่างที่เรามอบหมายให้ เช่น การตอบคำถาม, ค้นหาข้อมูล, วิเคราะห์เนื้อหา หรือแม้กระทั่งเขียนโค้ดแก้ปัญหา
ReAct ย่อมาจากคำว่า Reasoning + Acting
ซึ่งเป็นแนวคิดที่รวม “การให้เหตุผล (Reason)” และ “การลงมือทำ (Act)” เข้าด้วยกันในกระบวนการทำงานของ LLM (Large Language Model)
พูดง่ายๆ คือ ReAct Agent ไม่ได้แค่ “ตอบคำถาม” แบบปกติ แต่ยังสามารถ “วางแผนการคิด → ตัดสินใจ → เรียกใช้เครื่องมือภายนอก (tools)” แล้ว “นำผลลัพธ์กลับมาคิดต่อ” เพื่อให้ได้คำตอบที่แม่นยำและฉลาดขึ้น
ตัวอย่าง Diagram ของ ReAct Agent
ก่อนมีแนวคิด ReAct โมเดลมักจะทำได้เพียง “คาดเดาคำตอบ” จากข้อมูลใน prompt เท่านั้น
แต่ในโลกจริง ปัญหามักซับซ้อนกว่านั้น เช่น ต้องค้นข้อมูล, คำนวณ, หรือใช้ API ภายนอก ซึ่งโมเดลเดี่ยวๆ ทำไม่ได้
ReAct จึงเข้ามาเติมเต็มส่วนนี้ โดยให้โมเดลสามารถ
สามารถอ่านเพิ่มเติมเกี่ยวกับ ReAct Agent ได้ที่นี่
หัวใจของ ReAct Agent อยู่ที่ “วงจรการคิดและลงมือทำซ้ำๆ” ซึ่งประกอบด้วย 4 ขั้นตอนหลักคือ
Reason → Action → Observation → Reason → … → Final Answer
โมเดลจะเริ่มจากการ ให้เหตุผล (Reason) เพื่อวางแผนว่าจะทำอะไร จากนั้น ลงมือทำ (Action) เช่น ค้นข้อมูลหรือเรียกใช้เครื่องมือ แล้วนำผลที่ได้กลับมาเป็น Observation เพื่อคิดต่อว่าควรทำอะไรขั้นถัดไป จนกว่าจะได้คำตอบสุดท้ายที่มั่นใจ
ตัวอย่าง
ผู้ใช้ถามว่า “ใครเป็นผู้ก่อตั้งบริษัท Tesla?”
1️⃣ LLM Reason: “ตรวจสอบ context ก่อน เผื่อมีข้อมูลเกี่ยวกับผู้ก่อตั้งบริษัท Tesla”
2️⃣ Action: search_context(“ผู้ก่อตั้ง Tesla”)
3️⃣ Observation: “ไม่พบข้อมูลใน context”
4️⃣ Reason: “ต้องเรียกข้อมูลจาก Web search เพื่อให้คำตอบครบถ้วน”
5️⃣ Action: call_web_search(“ผู้ก่อตั้งบริษัท Tesla”)
6️⃣ Observation: “ผลจาก Web search : Elon Musk, Martin Eberhard และ Marc Tarpenning เป็นผู้ก่อตั้งบริษัท Tesla”
7️⃣ Reason: “นำข้อมูลนี้มาสรุปตอบผู้ใช้”
✅ Final Answer: “บริษัท Tesla ก่อตั้งโดย Elon Musk, Martin Eberhard และ Marc Tarpenning ครับ”
การ “วนลูป” แบบนี้สำคัญมาก เพราะทำให้ Agent ปรับกลยุทธ์ระหว่างทางได้เอง หากเจอข้อมูลไม่ครบหรือคำตอบยังไม่ชัดเจน
ในบทความนี้ เราจะลองสร้าง ReAct Agent แบบง่าย สำหรับ Chatbot ที่สามารถ “หาคำตอบจาก context ที่ให้มา (Mock Document ที่ได้จาก ระบบ RAG)”
และถ้าไม่เจอคำตอบใน context ก็จะ “หาข้อมูลเพิ่มเติมจาก web search” ก่อนสรุปผลให้ผู้ใช้ —
นั่นแหละคือพลังของแนวคิด Reason + Act ที่ทำให้ Chatbot ฉลาดขึ้นอีกระดับ 🚀
เครื่องมือที่ใช้
Python
LLM (Large Language Model) — ในที่นี้เราจะใช้ Google Gemini (มี Free quota)
Document / Mock RAG (Static Knowledge)
Web search — DuckDuckGo (ddgs)
โครงสร้าง Diagram : ReAct + Chatbot
ตัวอย่างโครงสร้าง Project
```
basic-re-act-python/
├── main.py # Entry point และ main execution
├── react_agent.py # Core ReAct Agent implementation
├── agent_actions.py # Agent actions และ tools
├── rag.py # RAG (Retrieval-Augmented Generation) system
├── constants.py # Configuration และ API keys
├── requirements.txt # Python dependencies
├── LICENSE # Project license
└── data/
└── mock_rag_document.md # Mock knowledge base
```ลองไปดูรายละเอียดในแต่ละไฟล์กัน ว่าจะมีฟังก์ชั่นอะไรบ้างอยู่ข้างใน (ก่อนเขียนโค้ดเต็มๆ)
ไฟล์ : requirements.txt
Python Library ที่จะใช้ใน Project
google-generativeai
ddgs
Install :
pip install -r requirements.txt
ไฟล์ : constant.py
เก็บค่า API KEY และ LLM Model Name
GOOGLE_GEMINI_API_KEY = "<your-api-key>"
GOOGLE_GEMINI_MODEL_NAME = "gemini-2.5-flash"
ไฟล์ : data/mock_rag_document.md
Mock Knowledge Base ในระบบ RAG System
# Company HR Information Database
## 1. Leave Policies
All employees are entitled to 15 days of annual leave per year. For employees with more than 5 years of service, they will receive an additional 1 day per year, up to a maximum of 20 days per year. Sick leave entitlement is 30 days per year with a medical certificate. Special leave such as maternity leave, ordination leave, and personal leave can be consulted with HR.
Leave applications must be submitted through the HR Online system at least 3 days in advance, except for emergency sick leave cases. Consecutive leave exceeding 3 days must be approved by the supervisor and HR. Leave during festivals or high season must be planned in advance and receive special approval. Employees can accumulate up to 5 days of leave to carry over to the following year.
---
## 2. Employee Benefits
The company provides comprehensive benefits for employees and their families. Health insurance covers 100% for employees and 80% for family members. Life insurance of 1 million baht, accident insurance of 500,000 baht, and provident fund with 5% company contribution.
Additional benefits include monthly attendance allowance, travel allowance, phone allowance, internet allowance, annual bonus (depending on company performance), bereavement assistance, birthday gifts, and 20% discount on company products. Employees also receive rights to skills development training, budget for purchasing books and online courses, and opportunities for career advancement.
---
## 3. Company Policies
The company adheres to principles of good governance and business ethics. Employees must comply with anti-bribery policies, avoid conflicts of interest, and maintain company confidentiality. Use of company equipment and resources must be for business purposes only.
Workplace safety policies require employees to wear personal protective equipment, report accidents immediately, and participate in annual safety training. The company has a No Smoking Area policy throughout the building and promotes a safe working environment, environmental consciousness, and Work-Life Balance policy for employees' mental well-being.
---
## 4. Working Hours & Holidays
Standard working hours are Monday-Friday 08:30-17:30 with lunch break 12:00-13:00, totaling 8 hours per day. Employees can choose Flexible Working Hours between 07:30-16:30 or 09:30-18:30 according to job requirements. Overtime work requires prior approval and will be compensated according to labor law.
Annual holidays follow the government calendar, including New Year's Day, Songkran Festival, Labor Day, Royal Celebration Day, Father's Day, Mother's Day, and Royal Birthday. Additionally, the company provides 2 special holidays: pre-Songkran holiday and New Year special holiday. Employees can use Work From Home 2 days per week according to the new policy.
---
## 5. Performance Review & Career Development
Performance evaluations are conducted every 6 months using a 360 Degree Feedback system that includes evaluations from supervisors, colleagues, and internal customers. Evaluation criteria include Key Performance Indicators (KPIs), Core Competencies, and Leadership Skills. Evaluation results will be linked to salary adjustments, bonuses, and promotion opportunities.
The company has long-term personnel development plans (Career Path) for all positions. Employees will receive Individual Development Plans (IDP) and have opportunities to participate in Training Programs both within and outside the organization. The company supports further education and professional licensing, has Mentoring and Job Rotation systems to enhance experience, and provides opportunities for cross-departmental work for career advancement.
ไฟล์: rag.py
# rag.py
from typing import List, Dict
def rag_load_context(file_path: str) -> List[Dict[str, str]]:
"""
อ่านไฟล์ Markdown และแยกเป็น sections
- file_path: path ของ Markdown file
Return: list ของ document [{'title': ..., 'content': ...}]
Note:
แต่ละ section เริ่มด้วย ## และสามารถมี separator ---
ฟังก์ชันนี้ใช้สำหรับเตรียมข้อมูล Mock RAG
"""
pass # logic การอ่านไฟล์และแยก section จะใส่ในเวอร์ชันจริง
def rag_search_context(query: str, top_k: int = 1) -> List[Dict[str, str]]:
"""
ค้นหา context จาก Mock RAG
- query: คำค้นหา
- top_k: จำนวน document ที่ต้องการคืนค่า
Return: list ของ document ที่เกี่ยวข้องที่สุด
Note:
- ใช้ rag_load_context() เพื่อโหลด document
- คำนวณความเกี่ยวข้องแบบง่าย (title + content match)
"""
pass # logic การค้นหา context จะใส่ในเวอร์ชันจริง
# ------------------------
# ตัวอย่างการใช้งาน (abstract)
# ------------------------
if __name__ == "__main__":
query = "Amazon"
results = rag_search_context(query, top_k=2)
for r in results:
print(f"Title: {r['title']}")
print(f"Content preview: {r['content'][:200]}...")
ไฟล์ : agent_actions.py
Actions ที่ ReAct Agent จะสามารถเรียกได้
# agent_actions.py
from typing import Dict, List
def search_context(query: str, top_k: int = 1) -> List[Dict[str, str]]:
"""
Action: ค้นหาข้อมูลจาก Mock RAG Knowledge Base
- query: คำถามหรือ keyword ของผู้ใช้
- top_k: จำนวน document ที่ต้องการคืนค่า
Return:
list ของ document ที่เกี่ยวข้อง [{'title': ..., 'content': ...}]
Note:
ฟังก์ชันนี้จะใช้ RAG system (mock) เพื่อจำลอง retrieval
"""
pass # จะใส่ logic ของ search_context ในไฟล์ rag.py
def call_web_search(query: str, max_results: int = 2) -> str:
"""
Action: ค้นหาข้อมูลจากเว็บโดยใช้ DuckDuckGo
- query: ข้อความคำถามที่ต้องการค้นหา
- max_results: จำนวนผลการค้นหาที่ต้องการคืนค่า
Return: สรุปข้อความที่รวมผลลัพธ์อันดับต้นๆ
"""
pass
ไฟล์ : react_agent.py
Workflow ของ ReAct Agent
# react_agent.py
from typing import List, Dict
from agent_actions import search_context, call_wiki_api
from rag import rag_search_context
class ReActAgent:
def __init__(self):
"""
Initialize ReAct Agent
"""
pass
def reason(self, user_input: str, observations: List[str]) -> str:
"""
LLM Reasoning step
- user_input: ข้อความคำถามจากผู้ใช้
- observations: list ของผลลัพธ์จาก Action ก่อนหน้า
Return: string -> prompt/คำสั่งสำหรับ Action หรือ final answer
Note:
ใน abstract version จะยังไม่เรียก LLM จริง
สามารถ mock output เพื่อสาธิต flow
"""
pass
def act(self, action_type: str, query: str):
"""
Action step
- action_type: 'search_context', 'web_search', 'final_answer'
- query: ข้อมูลสำหรับ Action
Return: observation (ผลลัพธ์ของ Action)
Note:
- 'search_context' -> เรียก rag_search_context() ของ Mock RAG
- 'web_search' -> search query for the internet
- 'final_answer' -> ใช้รวบรวม observations เป็น answer
"""
pass
def run(self, user_input: str) -> str:
"""
ตัว Flow หลักของ ReAct Agent
Reason -> Action -> Observation -> Reason -> Final Answer
Return: final_answer (string)
Note:
- สามารถ loop หลายรอบได้ (multi-step reasoning)
- ใน abstract version จะ mock flow เพื่อสาธิต
"""
pass
ไฟล์ : main.py
สำหรับ เรียกใช้งาน ReAct Agent
# main.py
from react_agent import ReActAgent
def main():
"""
ตัวรันหลักของ ReAct Agent
- สร้าง Agent instance
- รับ input จากผู้ใช้
- เรียก Agent.run() และแสดง Final Answer
"""
# 1️⃣ สร้าง ReAct Agent
agent = ReActAgent()
# 2️⃣ รับ input จากผู้ใช้ (ตัวอย่าง)
user_query = input("Query: ")
# 3️⃣ เรียก Agent.run() → ได้ Final Answer
final_answer = agent.run(user_query)
# 4️⃣ แสดงผล
print("\n=== Final Answer ===")
print(final_answer)
if __name__ == "__main__":
main()
สำหรับ Project ReAct Agent ของเราจะข้ามเรื่อง RAG System ไปโดยจะทำการจำลอง (Mock) Search Context จาก static data ที่เตรียมไว้แทน
สามารถค้นหาเพิ่มเติมได้ว่า RAG คืออะไร (ในบทความนี้เราจะข้ามขั้นตอนนี้ไป)
from typing import List, Dict
import re
def rag_load_context(file_path: str) -> List[Dict[str, str]]:
"""
อ่านไฟล์ Markdown และแยกเป็น sections
แต่ละ section จะคืนค่าเป็น dict {'title': ..., 'content': ...}
Assumption: แต่ละ section เริ่มด้วย ## และข้อมูลแยกด้วย ---
"""
documents = []
current_doc = None
with open(file_path, 'r', encoding='utf-8-sig') as f:
for line in f:
line = line.strip()
# ข้าม header หลัก "# Knowledge Base"
if line.startswith("# "):
continue
# เริ่ม section ใหม่ที่ขึ้นต้นด้วย ##
if line.startswith("## "):
if current_doc:
# ล้าง --- ออกจากท้าย content
current_doc["content"] = current_doc["content"].strip()
documents.append(current_doc)
title = line[3:].strip() # เอา "## " ออก
current_doc = {"title": title, "content": ""}
# ข้ามเส้น separator ---
elif line == "---":
continue
# เพิ่มเนื้อหาถ้ามี section ปัจจุบัน และไม่ใช่บรรทัดว่าง
elif current_doc is not None and line:
current_doc["content"] += line + "\n"
# append last section
if current_doc:
current_doc["content"] = current_doc["content"].strip()
documents.append(current_doc)
return documents
def rag_search_context(query: str, top_k: int = 1) -> List[Dict[str, str]]:
"""
ค้นหา context จาก Mock RAG
- query: คำค้นหา (string)
- top_k: จำนวน document ที่ต้องการคืนค่า
คืนค่าเป็น list ของ document ที่เกี่ยวข้องที่สุด
"""
# โหลดเอกสารทั้งหมดจาก Mock RAG database
rag_docs = rag_load_context("data/mock_rag_document.md")
# แปลงคำค้นหาเป็นตัวพิมพ์เล็กเพื่อการเปรียบเทียบที่ไม่สนใจตัวพิมพ์ใหญ่เล็ก
query_lower = query.lower()
scored_docs = []
# วนลูปผ่านเอกสารทั้งหมดเพื่อหาความเกี่ยวข้อง
for doc in rag_docs:
# นับจำนวนครั้งที่คำค้นหาปรากฏใน title และ content (ไม่สนใจตัวพิมพ์ใหญ่เล็ก)
title_score = len(re.findall(re.escape(query_lower), doc["title"].lower()))
content_score = len(re.findall(re.escape(query_lower), doc["content"].lower()))
# คำนวณคะแนนรวม โดยให้น้ำหนัก title มากกว่า content เป็น 2 เท่า
score = title_score * 2 + content_score
# เก็บเฉพาะเอกสารที่มีคะแนนมากกว่า 0 (มีความเกี่ยวข้อง)
if score > 0:
scored_docs.append((score, doc))
# เรียงลำดับเอกสารตามคะแนนจากมากไปน้อย
scored_docs.sort(key=lambda x: x[0], reverse=True)
# คืนค่าเอกสาร top_k อันดับแรกที่เกี่ยวข้องที่สุด
return [doc for _, doc in scored_docs[:top_k]]
# ------------------------
# ตัวอย่างการใช้งาน
# ------------------------
if __name__ == "__main__":
# ค้นหา context
query = "Amazon"
results = rag_search_context(query, top_k=2)
for r in results:
print(f"Title: {r['title']}")
print(f"Content preview: {r['content'][:200]}...\n")
หน้าที่หลัก:
รายละเอียด:
Return:
[
{"title": "ชื่อหัวข้อ", "content": "เนื้อหาของ section"},
...
]
หน้าที่หลัก:
รายละเอียด:
4. sort document ตาม score ลดหลั่น
5. return top_k document
Return:
[
{"title": "Amazon Company", "content": "..."},
{"title": "Amazon Rainforest", "content": "..."}
]
ภาพรวม flow ของ RAG ในไฟล์นี้
Markdown file --> rag_load_context() --> list of documents
User query --> rag_search_context() --> top_k documents
# agent_actions.py
from typing import Dict, List
from rag import rag_search_context # เรียกฟังก์ชันจาก rag.py
from ddgs import DDGS
def search_context(query: str, top_k: int = 1) -> List[Dict[str, str]]:
"""
Action: ค้นหาข้อมูลจาก Mock RAG Knowledge Base
- query: คำถามหรือ keyword ของผู้ใช้
- top_k: จำนวน document ที่ต้องการคืนค่า
Return:
list ของ document ที่เกี่ยวข้อง [{'title': ..., 'content': ...}]
Note:
ฟังก์ชันนี้ใช้ RAG system (mock) เพื่อจำลอง retrieval
"""
# เรียกใช้ฟังก์ชันค้นหาจาก RAG system และส่งคืนผลลัพธ์
return rag_search_context(query, top_k=top_k)
def call_web_search(query: str, max_results: int = 2) -> str:
"""
Action: Web search using DuckDuckGo
- query: search query string
- max_results: number of search results to return
Return: plain text summary combining top results
"""
try:
# ใช้ DuckDuckGo API เพื่อค้นหาข้อมูลจากอินเทอร์เน็ต
with DDGS() as ddgs:
# ค้นหาและแปลงผลลัพธ์เป็น list
results = list(ddgs.text(query, max_results=max_results))
# ตรวจสอบว่าพบข้อมูลหรือไม่
if not results:
return "No relevant information found on the web."
# รวมผลการค้นหาหลายๆ รายการเป็นข้อความเดียว
combined = " | ".join([r['body'] for r in results if 'body' in r])
return combined
except Exception as e:
# จัดการข้อผิดพลาดที่อาจเกิดขึ้นระหว่างการค้นหา
return f"Error during web search: {e}"
หน้าที่หลัก:
ค้นหาข้อมูลที่เกี่ยวข้องจากระบบฐานความรู้ภายใน (RAG system) ตามคำถามหรือคีย์เวิร์ดที่ผู้ใช้ป้อนมา
รายละเอียด:
Parameter:
Return:
[
{
"title": "ชื่อเอกสาร",
"content": "เนื้อหาในเอกสาร"
},
...
]
หน้าที่หลัก:
ค้นหาข้อมูลจากอินเทอร์เน็ตผ่าน DuckDuckGo Search API (ผ่านไลบรารี ddgs)
รายละเอียด:
Parameter:
Return:
ข้อความสรุปผลลัพธ์จากการค้นหา (plain text) เช่น:
"Tesla, Inc. is an American electric vehicle company founded in 2003... | Elon Musk joined Tesla later as an investor..."
หน้าที่หลักของไฟล์:
# react_agent.py
from typing import List, Dict
from agent_actions import search_context, call_web_search
from constants import GOOGLE_GEMINI_API_KEY, GOOGLE_GEMINI_MODEL_NAME
import google.generativeai as genai
import json
import re
import os
from datetime import datetime
# ตั้งค่า Google Gemini API
genai.configure(api_key=GOOGLE_GEMINI_API_KEY)
gemini_client = genai.GenerativeModel(GOOGLE_GEMINI_MODEL_NAME)
class ReActAgent:
def __init__(self, max_steps: int = 5, enable_logging: bool = True):
self.observations: List[str] = []
self.max_steps = max_steps
self.enable_logging = enable_logging
self.log_lines: List[str] = []
# -------------------------
# Logging helpers
# -------------------------
def log(self, message: str):
if self.enable_logging:
self.log_lines.append(message)
print(message)
# -------------------------
# Save log to file (output data/debug/<file>.md)
# -------------------------
def save_log(self):
output_dir = "data/debug"
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filepath = f"{output_dir}/react_agent_log_{timestamp}.md"
# สร้างโฟลเดอร์ถ้ายังไม่มี
os.makedirs(output_dir, exist_ok=True)
with open(filepath, "w", encoding="utf-8") as f:
f.write("\n".join(self.log_lines))
# -------------------------
# Reasoning step
# -------------------------
def reason(self, user_input: str, observations: List[str], step: int) -> Dict[str, str]:
"""
LLM reasoning - let AI decide what to do based on the query type and current observations
"""
# -------------------------
# Clean ข้อมูล observations สำหรับ input ของ LLM
# -------------------------
# กรองเอา HTML tags ออกและไม่รวม "FINAL_STEP"
obs_texts = [re.sub(r"<[^>]+>", "", o) for o in observations if o != "FINAL_STEP"]
# รวมข้อความทั้งหมดเป็นข้อความเดียว
obs_text = " ".join(obs_texts)
# ลบช่องว่างเกินและทำให้เป็นรูปแบบมาตรฐาน
obs_text = re.sub(r"\s+", " ", obs_text).strip()
# สร้าง prompt สำหรับให้ LLM ตัดสินใจ
prompt = f"""
You are a ReAct Agent AI assistant. Analyze the current situation and decide the next action.
Current observations: {obs_text if obs_text else "No observations yet"}
User question: {user_input}
Current step: {step}
Available actions:
- 'search_context' if you need to search the internal knowledge base/RAG system for relevant information
- 'web_search' if you need additional current/external information from the internet
- 'final_answer' if you have sufficient information to provide a complete answer
Decision criteria:
- If no observations yet, consider what type of information is needed first
- If observations exist, evaluate if they provide sufficient information
- Consider whether the question requires real-time/current information vs historical/factual information
- Think about the most logical sequence of actions to answer the question effectively
Respond in JSON format:
Query field should contain:
- For 'search_context': search keywords or phrases for the knowledge base
- For 'web_search': search query for the internet
- For 'final_answer': the original user question being answered
"""
try:
# เรียกใช้ Gemini AI เพื่อตัดสินใจการกระทำต่อไป
response = gemini_client.generate_content(
prompt,
generation_config=genai.types.GenerationConfig(
temperature=0.2, # ใช้ความสร้างสรรค์ปานกลาง
max_output_tokens=500 # จำกัด token สำหรับการตอบกลับ
)
)
# ดึงข้อความจาก response
text = getattr(response, "text", None)
if not text:
raise ValueError("No text returned from LLM")
# ลบ code block markers ออกจาก JSON response
text = re.sub(r"^```json|```$", "", text, flags=re.MULTILINE).strip()
# แปลง JSON string เป็น dictionary
decision = json.loads(text)
except Exception as e:
# หากเกิดข้อผิดพลาดให้บันทึกและใช้การตัดสินใจ default
self.log(f"Error parsing LLM response: {e}")
decision = {"action": "final_answer", "query": user_input}
return decision
# -------------------------
# Action step
# -------------------------
def act(self, action_type: str, query: str, user_input: str = None) -> str:
"""
Execute action and return observation
"""
# ตรวจสอบประเภทของ action_type ที่ต้องการทำ
if action_type == "search_context":
# ค้นหาข้อมูลในฐานความรู้ภายใน (RAG system) โดยใช้คำค้นหา
docs = search_context(query, top_k=2)
if docs:
# สร้างสรุปเอกสารที่พบจากการค้นหา
doc_summaries = [f"Title: {doc['title']}, Content: {doc['content']}" for doc in docs]
obs = f"Found {len(docs)} relevant document(s) in knowledge base: " + "; ".join(doc_summaries)
else:
# ไม่พบข้อมูลที่เกี่ยวข้องในฐานความรู้ภายใน (RAG system)
obs = "No relevant information found in the internal knowledge base"
elif action_type == "web_search":
# ค้นหาข้อมูลจากอินเทอร์เน็ต โดยใช้คำค้นหา
obs = call_web_search(query)
elif action_type == "final_answer":
# สร้างคำตอบสุดท้ายโดยใช้ข้อมูลทั้งหมดที่รวบรวมได้
obs = self.generate_final_answer(user_input or query)
else:
# ประเภทการกระทำที่ไม่รู้จัก
obs = f"Unknown action: {action_type}"
# บันทึกผลการสังเกตลงในรายการ observations
self.observations.append(obs)
return obs
# -------------------------
# Final answer generation
# -------------------------
def generate_final_answer(self, user_input: str) -> str:
"""
Generate final answer based on all current observations (true ReAct pattern)
"""
# Clean ข้อมูล observations โดยลบ HTML tags ออก
obs_texts = [re.sub(r"<[^>]+>", "", o) for o in self.observations]
# รวมข้อความทั้งหมดเป็นข้อความเดียว
obs_text = " ".join(obs_texts)
# ลบช่องว่างที่ไม่จำเป็นและ normalize ข้อความ
obs_text = re.sub(r"\s+", " ", obs_text).strip()
# ตรวจสอบว่ามีข้อมูลจาก observations หรือไม่
if not obs_text or obs_text.strip() == "":
# กรณีไม่มีข้อมูลจาก observations ให้ตอบคำถามโดยตรง
prompt = f"""
Please provide a helpful, direct answer to this user question:
{user_input}
Provide a clear, informative response in 2-3 sentences.
"""
else:
# กรณีมีข้อมูลจาก observations ให้ใช้ข้อมูลนั้นในการตอบ
prompt = f"""
Based on the information gathered, provide a clear final answer:
Question: {user_input}
Information gathered: {obs_text}
Provide a complete answer in 2-3 clear sentences.
"""
try:
# เรียกใช้ Gemini AI เพื่อสร้างคำตอบสุดท้าย
response = gemini_client.generate_content(
prompt,
generation_config=genai.types.GenerationConfig(
temperature=0.1, # ใช้ temperature ต่ำเพื่อความแม่นยำ
max_output_tokens=2000 # จำกัดจำนวน token สูงสุด
)
)
# ดึงข้อความคำตอบจาก response
final_answer = getattr(response, "text", None)
if not final_answer:
raise ValueError("No text returned from LLM")
# ส่งคืนคำตอบพร้อมกับ prefix "FINAL_ANSWER: "
return f"FINAL_ANSWER: {final_answer}"
except Exception as e:
# จัดการข้อผิดพลาดและบันทึก log
self.log(f"Error generating final answer: {type(e).__name__}: {e}")
return f"FINAL_ANSWER: Unable to generate final answer due to error: {e}"
# -------------------------
# Main agent flow
# -------------------------
def run(self, user_input: str) -> str:
# เริ่มต้นการทำงานใหม่โดยเคลียร์ข้อมูลเก่า
self.observations = []
self.log_lines = []
# สร้าง log header สำหรับการทำงานครั้งนี้
self.log(f"# ReAct Agent Log")
self.log(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
self.log(f"**User Query:** {user_input}\n")
self.log(f"## 🚀 Starting ReAct Agent Process\nMaximum Steps: {self.max_steps}\n")
# วนลูปทำงานตามจำนวนขั้นตอนสูงสุดที่กำหนด
for step in range(1, self.max_steps + 1):
# ให้ LLM ตัดสินใจการกระทำต่อไปตามสถานการณ์ปัจจุบัน
decision = self.reason(user_input, self.observations, step)
action_type = decision.get("action", "final_answer")
query = decision.get("query", user_input)
# บันทึก log สำหรับขั้นตอนนี้
self.log(f"# Step {step}/{self.max_steps}")
self.log(f"**Thought:** LLM decided to perform '{action_type}' action")
self.log(f"**Action:** {action_type}")
self.log(f"**Query:** {query}")
# ดำเนินการตามที่ LLM ตัดสินใจและรับผลการสังเกต
obs = self.act(action_type, query, user_input)
self.log(f"**Observation:** {obs}\n")
# ตรวจสอบว่าได้คำตอบสุดท้ายแล้วหรือไม่
if action_type == "final_answer":
# ดึงคำตอบสุดท้ายจากผลการสังเกต (รูปแบบ ReAct แท้จริง)
if obs.startswith("FINAL_ANSWER: "):
final_answer = obs[14:] # ลบ prefix "FINAL_ANSWER: " ออก
self.log(f"=== Final Answer ===\n{final_answer}\n")
self.save_log()
return final_answer
break
# กรณีที่จบลูปโดยไม่มีการทำ final_answer action (fallback)
self.log("⚠️ Agent reached maximum steps without final_answer action")
self.log("🔄 Forcing final answer generation...\n")
# บังคับสร้างคำตอบสุดท้าย
fallback_obs = self.generate_final_answer(user_input)
if fallback_obs.startswith("FINAL_ANSWER: "):
fallback_answer = fallback_obs[14:] # ลบ prefix "FINAL_ANSWER: " ออก
self.log(f"=== Fallback Answer ===\n{fallback_answer}\n")
self.save_log()
return fallback_answer
else:
# กรณีพิเศษ: แม้แต่ generate_final_answer ก็ล้มเหลว
error_msg = f"Unable to generate answer after {self.max_steps} steps"
self.log(f"❌ {error_msg}")
self.save_log()
return error_msg
# ------------------------
# Example usage
# ------------------------
if __name__ == "__main__":
agent = ReActAgent()
# Test with a question that might benefit from internal knowledge first
question = "Employee Benefits"
print("Testing with question:", question)
answer = agent.run(question)
หน้าที่หลัก:
ใช้สำหรับเตรียมค่าเริ่มต้นของ Agent เช่น จำนวนรอบสูงสุดที่ให้ทำงาน (max_steps) การเปิด/ปิดระบบ log และสร้างตัวแปรเก็บข้อมูลภายใน
รายละเอียด:
Return:
หน้าที่หลัก:
ใช้สำหรับบันทึกข้อความ log ของแต่ละขั้นตอน
รายละเอียด:
Return:
หน้าที่หลัก:
บันทึก log การทำงานของ Agent ทั้งหมดลงไฟล์ markdown เพื่อใช้วิเคราะห์ย้อนหลัง
รายละเอียด:
Return:
หน้าที่หลัก:
ให้ LLM วิเคราะห์สถานการณ์ปัจจุบันและตัดสินใจว่าจะทำ action อะไรต่อไป
รายละเอียด:
Return:
Dictionary ในรูป { "action": "...", "query": "..." }
หน้าที่หลัก:
ทำการ “ลงมือทำ” ตาม action ที่ LLM ตัดสินใจไว้
รายละเอียด:
Return:
ข้อความ observation ที่เกิดจากการทำ action
หน้าที่หลัก:
สร้างคำตอบสุดท้ายโดยใช้ข้อมูลทั้งหมดที่ Agent รวบรวมได้
รายละเอียด:
Return:
ข้อความในรูปแบบ "FINAL_ANSWER: <ข้อความคำตอบ>"
หน้าที่หลัก:
เป็นฟังก์ชันหลักที่ควบคุมวงจรการทำงานของ Agent ตั้งแต่เริ่มจนได้คำตอบสุดท้าย
รายละเอียด:
Return:
คำตอบสุดท้ายจาก Agent ในรูปแบบข้อความ
# ReAct Agent Log
**Generated:** 2025-10-14 18:28:26
**User Query:** Company Policies
## 🚀 Starting ReAct Agent Process
Maximum Steps: 5
# Step 1/5
**Thought:** LLM decided to perform 'search_context' action
**Action:** search_context
**Query:** Company Policies
**Observation:** Found 1 relevant document(s) in knowledge base: Title: 3. Company Policies, Content: The company adheres to principles of good governance and business ethics. Employees must comply with anti-bribery policies, avoid conflicts of interest, and maintain company confidentiality. Use of company equipment and resources must be for business purposes only.
Workplace safety policies require employees to wear personal protective equipment, report accidents immediately, and participate in annual safety training. The company has a No Smoking Area policy throughout the building and promotes a safe working environment, environmental consciousness, and Work-Life Balance policy for employees' mental well-being.
# Step 2/5
**Thought:** LLM decided to perform 'final_answer' action
**Action:** final_answer
**Query:** Company Policies
**Observation:** FINAL_ANSWER: The company policies are built on principles of good governance and business ethics, requiring employees to comply with anti-bribery measures, avoid conflicts of interest, maintain confidentiality, and use company resources solely for business purposes. Workplace safety is paramount, mandating the use of personal protective equipment, immediate accident reporting, and participation in annual safety training, alongside a strict No Smoking Area policy. Furthermore, the company promotes environmental consciousness and supports employee well-being through a Work-Life Balance policy.
=== Final Answer ===
The company policies are built on principles of good governance and business ethics, requiring employees to comply with anti-bribery measures, avoid conflicts of interest, maintain confidentiality, and use company resources solely for business purposes. Workplace safety is paramount, mandating the use of personal protective equipment, immediate accident reporting, and participation in annual safety training, alongside a strict No Smoking Area policy. Furthermore, the company promotes environmental consciousness and supports employee well-being through a Work-Life Balance policy.
2. แบบที่เรียก search_context ก่อน แต่ยังไม่เจอคำตอบ > LLM ตัดสินใจ เรียก call_web_search > เจอคำตอบ
# ReAct Agent Log
**Generated:** 2025-10-14 18:30:34
**User Query:** Python programming language
## 🚀 Starting ReAct Agent Process
Maximum Steps: 5
# Step 1/5
**Thought:** LLM decided to perform 'search_context' action
**Action:** search_context
**Query:** Python programming language overview
**Observation:** No relevant information found in the internal knowledge base
# Step 2/5
**Thought:** LLM decided to perform 'web_search' action
**Action:** web_search
**Query:** Python programming language
**Observation:** Python is a high-level, general-purpose programming language. Its design philosophy emphasizes code readability with the use of significant indentation.Python is dynamically type-checked and garbage-collected. It supports multiple programming paradigms, including structured (particularly procedural), object-oriented and functional programming.Guido van Rossum began working on Python in the late 1980s as a successor to the ABC programming language. Python 3.0, released in 2008, was a major revision and not completely backward-compatible with earlier versions. Recent versions, such as Python 3.13, 3.12 and older (and 3.14), have added capabilities and keywords for typing (and more; e.g. increasing speed); helping with (optional) static typing. Currently only versions in the 3.x series are supported.Python consistently ranks as one of the most popular programming languages, and it has gained widespread use in the machine learning community. It is widely taught as an introductory programming language. | The official home of the Python Programming Language
# Step 3/5
**Thought:** LLM decided to perform 'final_answer' action
**Action:** final_answer
**Query:** Python programming language
**Observation:** FINAL_ANSWER: Python is a high-level, general-purpose programming language known for its design philosophy emphasizing code readability through significant indentation. It is dynamically type-checked, supports multiple programming paradigms, and consistently ranks as one of the most popular languages, widely used in machine learning and as an introductory language. Developed by Guido van Rossum, only Python 3.x versions are currently supported, with recent updates enhancing capabilities like optional static typing.
=== Final Answer ===
Python is a high-level, general-purpose programming language known for its design philosophy emphasizing code readability through significant indentation. It is dynamically type-checked, supports multiple programming paradigms, and consistently ranks as one of the most popular languages, widely used in machine learning and as an introductory language. Developed by Guido van Rossum, only Python 3.x versions are currently supported, with recent updates enhancing capabilities like optional static typing.
ไฟล์ react_agent.py คือหัวใจหลักของระบบ ReAct Agent ซึ่งออกแบบให้ AI สามารถ “คิดเป็นขั้นตอน (Reason)” และ “ลงมือทำ (Act)” ได้อย่างมีตรรกะ โดยไม่จำกัดเพียงการตอบตรงคำถามเท่านั้น แต่ยังสามารถวางแผนการค้นคว้า ตัดสินใจเลือกแหล่งข้อมูลที่เหมาะสม และสร้างคำตอบสุดท้ายที่มีบริบทครบถ้วนที่สุด
การทำงานของ Agent จะมีโครงสร้างแบบวนลูป ได้แก่
บทความนี้อธิบายโครงสร้างและการทำงานของ ReAct Agent — ตัวกลางที่เชื่อมระหว่างกระบวนการ “คิดวิเคราะห์และตัดสินใจ” ของ LLM กับการ “ลงมือปฏิบัติ” ผ่านชุด Action เช่น search_context และ web_search ระบบนี้ช่วยให้ AI สามารถวางแผน เลือกวิธีแก้ปัญหา และสรุปคำตอบได้อย่างมีลำดับขั้น
“Reason + Act + Observe”
เราได้สร้างต้นแบบ ReAct Agent ด้วย Python ที่สามารถ:
✅ อ่าน context จาก knowledge base (RAG)
✅ ค้นหาข้อมูลภายนอกจากเว็บจริง
✅ รวมผลลัพธ์มาสรุปเป็นคำตอบสุดท้าย
เพื่อให้ผลลัพธ์ที่ได้มีความแม่นยำและอิงข้อมูลจริง สุดท้าย Agent จะรวมผลการค้นหาทั้งหมดมาเรียบเรียงเป็นคำตอบสุดท้ายที่กระชับและครบถ้วน ถือเป็นแนวทางสำคัญในการสร้าง LLM Agent ที่มีความสามารถในการคิดและเรียนรู้จากกระบวนการของตนเอง
จากการลองลงมือทำตามด้านบนทำให้เราเข้าใจเบื้องหลัง หลักการทำงานของ ReAct Agent มากยิ่งขึ้น
ซึ่ง Concept นี้ก็อยู่เบื้องหลังการทำงานของ AI Assistant Tools ดังๆ มากมายในระดับ Design คือ ReAct-based reasoning system เช่น GitHub Copilot, etc.
จริงๆ มี Library ที่ช่วยจัดการเรื่องพวกนี้ให้อยู่เหมือนกันนะเช่น LangChain, LlamaIndex, etc.
ตัวอย่างโค้ดแบบเต็มๆ : GitHub Repository
สำหรับใครที่กำลังมองหาวิธีสร้าง RAG Application หรือ Chatbot เพื่อใช้งานในองค์กร ที่ PALO IT เรามีทีมผู้เชี่ยวชาญพร้อมช่วยตั้งแต่เริ่มต้นจนระบบใช้งานได้จริง! ไม่ว่าจะเป็น
ไม่ว่าคุณจะเพิ่งเริ่มต้น หรือมีระบบอยู่แล้วและอยากต่อยอด เราพร้อมเป็น partner ที่จะช่วยให้คุณไปได้ไกลกว่าเดิม
ทักไปที่เพจ Facebook: PALO IT Thailand ได้เลยครับ 🎉