การออกแบบและพัฒนาระบบ AI Agent ด้วย Role และ RAG โดยใช้ AI Qwen API

การออกแบบและพัฒนาระบบ AI Agent ด้วย Role และ RAG

บทนำ

ในยุคที่เทคโนโลยี AI ก้าวหน้าอย่างรวดเร็ว การออกแบบระบบ AI Agent ที่สามารถตอบคำถามและแก้ปัญหาได้อย่างมีประสิทธิภาพเป็นสิ่งสำคัญ บทความนี้จะพาคุณไปรู้จักกับแนวคิดการออกแบบระบบ AI Agent โดยใช้ Role และ RAG (Retrieval-Augmented Generation) เพื่อให้ระบบสามารถทำงานได้อย่างชาญฉลาดและตอบสนองความต้องการของผู้ใช้ได้ดียิ่งขึ้น

1. Role: บทบาทของ AI Agent

Role คือบทบาทหรือหน้าที่ของ AI Agent ในการตอบคำถามหรือทำงานต่าง ๆ การกำหนด Role ให้ชัดเจนช่วยให้ระบบตัดสินใจได้ดีขึ้นว่าจะตอบคำถามอย่างไร หรือควรใช้ฟังก์ชันใด

ตัวอย่าง Role

  • ผู้ช่วยทั่วไป: ตอบคำถามทั่วไป เช่น การทักทายหรือคำถามพื้นฐาน
  • ผู้เชี่ยวชาญด้านการขาย: ตอบคำถามเกี่ยวกับยอดขายหรือรายงาน
  • ผู้เชี่ยวชาญด้านการผลิต: ตอบคำถามเกี่ยวกับวัตถุดิบหรือต้นทุนการผลิต
  • ผู้ดูแลลูกค้า: จัดการปัญหาหรือคำร้องเรียนจากลูกค้า

วิธีการใช้งาน Role

  1. ใช้เงื่อนไข (Condition) เพื่อตรวจสอบประเภทของคำถาม
  2. กำหนด Role ที่เหมาะสมให้กับคำถามแต่ละประเภท
  3. เรียกใช้ฟังก์ชันเฉพาะทางตาม Role ที่กำหนด

2. RAG: การค้นหาและสร้างคำตอบ

RAG (Retrieval-Augmented Generation) เป็นเทคนิคที่ผสมผสานระหว่างการค้นหาข้อมูล (Retrieval) และการสร้างข้อความ (Generation) เพื่อให้ระบบสามารถตอบคำถามได้อย่างแม่นยำและเป็นประโยชน์

ขั้นตอนการทำงานของ RAG

  1. ค้นหาข้อมูล: ค้นหาข้อมูลที่เกี่ยวข้องจากแหล่งข้อมูล (เช่น ฐานข้อมูลหรือเอกสาร)
  2. สร้างคำตอบ: นำข้อมูลที่ค้นหาได้มาสร้างเป็นคำตอบที่เหมาะสม

ตัวอย่างการใช้งาน RAG

  • บริการลูกค้า: ใช้ RAG เพื่อค้นหาข้อมูลเกี่ยวกับปัญหาลูกค้าและสร้างคำตอบที่เหมาะสม
  • รายงานยอดขาย: ใช้ RAG เพื่อดึงข้อมูลยอดขายและสร้างรายงาน

3. การออกแบบระบบ AI Agent

การออกแบบระบบ AI Agent ที่ดีควรมีองค์ประกอบดังนี้:

3.1 การแบ่งส่วนการทำงาน

  • แยกการทำงานออกเป็นส่วน ๆ เช่น การตอบคำถามทั่วไป การจัดการปัญหาลูกค้า การคำนวณต้นทุนการผลิต
  • ใช้ Role เพื่อกำหนดหน้าที่ของแต่ละส่วน

3.2 การใช้งาน RAG เฉพาะส่วน

  • ใช้ RAG เฉพาะส่วนที่ต้องการความแม่นยำสูง เช่น บริการลูกค้า
  • สร้างไฟล์ Markdown เพื่อเก็บข้อมูลเฉพาะทาง (เช่น service_rag.md)

3.3 การโหลดข้อมูลจากไฟล์

อ่านข้อมูลจากไฟล์ Markdown เพื่อให้ระบบสามารถปรับปรุงข้อมูลได้โดยไม่ต้องแก้ไขโค้ด

# ข้อมูลบริการลูกค้า (Service RAG)

## 1. สินค้าไม่ตรงตามสั่ง
- **คำถามที่เกี่ยวข้อง**:
 - สินค้าไม่ตรงตามสั่ง
 - ได้รับสินค้าผิด
- **คำตอบ**: เราขออภัยในความไม่สะดวก กรุณาติดต่อฝ่ายบริการลูกค้าเพื่อจัดส่งสินค้าใหม่

## 2. สินค้าชำรุด
- **คำถามที่เกี่ยวข้อง**:
 - สินค้าชำรุด
 - สินค้าเสียหาย
- **คำตอบ**: เราจะส่งทีมงานไปตรวจสอบและแก้ไขปัญหาโดยเร็วที่สุด

## 3. บริการล่าช้า
- **คำถามที่เกี่ยวข้อง**:
 - บริการล่าช้า
 - ส่งของช้า
- **คำตอบ**: เราขออภัยในความล่าช้า และจะเร่งดำเนินการให้เร็วที่สุด

## 4. อื่น ๆ
- **คำถามที่เกี่ยวข้อง**:
 - ปัญหาอื่น ๆ
- **คำตอบ**: ขอบคุณสำหรับการแจ้งปัญหา เราจะดำเนินการแก้ไขโดยเร็วที่สุด

Python Code :


from qwen_agent import Agent
import re

class MyAgent(Agent):
    def __init__(self, api_key, role_file):
        super().__init__(api_key=api_key)
        self.role = "ผู้ช่วยทั่วไป"  # กำหนด Role เริ่มต้น
        self.roles = self.load_roles(role_file)  # โหลด Role จากไฟล์

    def _run(self, *args, **kwargs):
        return "Running the agent"

    def load_roles(self, role_file):
        # อ่าน Role จากไฟล์
        with open(role_file, "r", encoding="utf-8") as file:
            roles = file.read()
        return roles

    def chat_with_functions(self, prompt, functions):
        # กำหนด Role ตามประเภทคำถาม
        self.determine_role(prompt)

        # หากเป็นคำถามทั่วไป
        if self.role == "ผู้ช่วยทั่วไป":
            return self.handle_general_question(prompt)

        # หากเป็นคำถามเฉพาะทาง
        else:
            return self.handle_specialized_question(prompt, functions)

    def determine_role(self, prompt):
        # ตรวจสอบและกำหนด Role ตามคำถาม
        if re.search(r"(รายงานยอดขาย|ขอยอดขายทั้งหมด|ยอดขาย)", prompt):
            self.role = "ผู้เชี่ยวชาญด้านการขาย"
        elif re.search(r"(วัตถุดิบ|จำนวน)", prompt):
            self.role = "ผู้เชี่ยวชาญด้านการผลิต"
        elif re.search(r"(ปัญหา|สินค้าชำรุด|บริการล่าช้า)", prompt):
            self.role = "ผู้ดูแลลูกค้า"
        else:
            self.role = "ผู้ช่วยทั่วไป"

    def handle_general_question(self, prompt):
        # ตอบคำถามทั่วไป
        if "สวัสดี" in prompt:
            return "สวัสดีครับ! มีอะไรให้ช่วยเหลือไหมครับ?"
        elif "สบายดีไหม" in prompt:
            return "ฉันเป็น AI ครับ แต่ขอบคุณที่ถามนะครับ!"
        else:
            return "มีอะไรให้ช่วยเหลือไหมครับ?"

    def handle_specialized_question(self, prompt, functions):
        # ใช้ RAG เพื่อค้นหาข้อมูลและสร้างคำตอบ
        if self.role == "ผู้เชี่ยวชาญด้านการขาย":
            return self.retrieve_sales_data(prompt, functions[2])
        elif self.role == "ผู้เชี่ยวชาญด้านการผลิต":
            return self.retrieve_production_data(prompt, functions[0], functions[1])
        elif self.role == "ผู้ดูแลลูกค้า":
            return self.handle_customer_service(prompt, functions[3])

    def retrieve_sales_data(self, prompt, sales_function):
        # ใช้ RAG เพื่อค้นหาข้อมูลยอดขาย
        return sales_function()

    def retrieve_production_data(self, prompt, inventory_function, cost_function):
        # ใช้ RAG เพื่อค้นหาข้อมูลวัตถุดิบหรือต้นทุนการผลิต
        if "จำนวน" in prompt:
            item = re.search(r"(วัตถุดิบ A|วัตถุดิบ B|วัตถุดิบ C)", prompt)
            if item:
                return inventory_function(item.group())
            else:
                return "กรุณาระบุชื่อวัตถุดิบให้ถูกต้อง"
        elif "ต้นทุนการผลิต" in prompt:
            try:
                units = int(prompt.split("ต้นทุนการผลิต")[1].strip())
                return cost_function(units)
            except ValueError:
                return "กรุณาระบุจำนวนหน่วยให้ถูกต้อง"

    def handle_customer_service(self, prompt, service_function):
        # ใช้ RAG เพื่อจัดการปัญหาลูกค้า
        issue = prompt.split("ปัญหา")[1].strip() if "ปัญหา" in prompt else prompt
        return self.rag_customer_service(issue)

    def rag_customer_service(self, issue):
        # ตัวอย่างการใช้ RAG สำหรับบริการลูกค้า
        # ค้นหาข้อมูลที่เกี่ยวข้องจากแหล่งข้อมูล
        knowledge_base = {
            "สินค้าชำรุด": "เราจะส่งทีมงานไปตรวจสอบและแก้ไขปัญหาโดยเร็วที่สุด",
            "บริการล่าช้า": "เราขออภัยในความล่าช้า และจะเร่งดำเนินการให้เร็วที่สุด",
            "สินค้าไม่ตรงตามสั่ง": "เราขออภัยในความไม่สะดวก กรุณาติดต่อฝ่ายบริการลูกค้าเพื่อจัดส่งสินค้าใหม่"
        }

        # ค้นหาข้อมูลที่เกี่ยวข้อง
        for keyword, response in knowledge_base.items():
            if keyword in issue:
                return response

        # หากไม่พบข้อมูลที่เกี่ยวข้อง
        return f"ขอบคุณสำหรับการแจ้งปัญหา '{issue}' เราจะดำเนินการแก้ไขโดยเร็วที่สุด"

# ตั้งค่า API Key
api_key = 'Api Key'  # กรุณาแทนค่าด้วย API Key ของคุณ
role_file = 'role.md'  # ไฟล์บทบาท (Role)
agent = MyAgent(api_key=api_key, role_file=role_file)

# ฟังก์ชันสำหรับการซื้อ (Procurement)
def check_inventory(item):
    inventory = {
        "วัตถุดิบ A": 100,
        "วัตถุดิบ B": 50,
        "วัตถุดิบ C": 200
    }
    return f"จำนวน {item} ในคลัง: {inventory.get(item, 0)} หน่วย"

# ฟังก์ชันสำหรับการผลิต (Production)
def calculate_production_cost(units):
    cost_per_unit = 50  # ต้นทุนต่อหน่วย
    total_cost = units * cost_per_unit
    return f"ต้นทุนการผลิตสำหรับ {units} หน่วย: {total_cost} บาท"

# ฟังก์ชันสำหรับการขาย (Sales)
def get_sales_report():
    sales_data = {
        "มกราคม": 50000,
        "กุมภาพันธ์": 60000,  # แก้ไขข้อมูลให้ถูกต้อง
        "มีนาคม": 70000
    }
    return f"รายงานยอดขาย: {sales_data}"

# ฟังก์ชันสำหรับการบริการ (Service)
def handle_customer_service(issue):
    # ตัวอย่างการแก้ไขปัญหาลูกค้า
    if "สินค้าไม่ตรงตามสั่ง" in issue:
        return "เราขออภัยในความไม่สะดวก กรุณาติดต่อฝ่ายบริการลูกค้าเพื่อจัดส่งสินค้าใหม่"
    elif "สินค้าชำรุด" in issue:
        return "เราจะส่งทีมงานไปตรวจสอบและแก้ไขปัญหาโดยเร็วที่สุด"
    elif "บริการล่าช้า" in issue:
        return "เราขออภัยในความล่าช้า และจะเร่งดำเนินการให้เร็วที่สุด"
    else:
        return f"ขอบคุณสำหรับการแจ้งปัญหา '{issue}' เราจะดำเนินการแก้ไขโดยเร็วที่สุด"

# รวมฟังก์ชันทั้งหมดเข้าไว้ใน Agent
functions = [
    check_inventory,
    calculate_production_cost,
    get_sales_report,
    handle_customer_service
]

# ฟังก์ชันหลักสำหรับรับคำถามและตอบกลับ
def organization_system(prompt):
    try:
        response = agent.chat_with_functions(prompt, functions=functions)
        return response
    except Exception as e:
        return f"เกิดข้อผิดพลาด: {e}"

# ทดสอบการทำงานของระบบ
if __name__ == "__main__":
    while True:
        print("\nโปรดพิมพ์คำถามของคุณ (พิมพ์ 'exit' เพื่อออกจากโปรแกรม):")
        user_input = input("คำถาม: ")
        
        if user_input.lower() == 'exit':
            print("ขอบคุณที่ใช้งานระบบของเรา!")
            break
        
        # ส่งคำถามไปยังระบบ
        result = organization_system(user_input)
        print("คำตอบ:", result)

 ตัวอย่างคำตอบ

โปรดพิมพ์คำถามของคุณ (พิมพ์ 'exit' เพื่อออกจากโปรแกรม):
คำถาม: สวัสดี
คำตอบ: สวัสดีครับ! มีอะไรให้ช่วยเหลือไหมครับ?

โปรดพิมพ์คำถามของคุณ (พิมพ์ 'exit' เพื่อออกจากโปรแกรม):
คำถาม: รายงานยอดขาย
คำตอบ: รายงานยอดขาย: {'มกราคม': 50000, 'กุมภาพันธ์': 60000, 'มีนาคม': 70000}

โปรดพิมพ์คำถามของคุณ (พิมพ์ 'exit' เพื่อออกจากโปรแกรม):
คำถาม: สินค้ามีปัญหา
คำตอบ: - เราขออภัยในความไม่สะดวก กรุณาติดต่อฝ่ายบริการลูกค้าเพื่อจัดส่งสินค้าใหม่

โปรดพิมพ์คำถามของคุณ (พิมพ์ 'exit' เพื่อออกจากโปรแกรม):
คำถาม: