ในยุคที่ AI ต้องเข้าใจทั้งข้อความและรูปภาพ การสร้าง Multi-Modal RAG (Retrieval-Augmented Generation) จึงกลายเป็นทักษะที่ Developer ทุกคนควรมี ในบทความนี้ผมจะพาทุกท่านไปสัมผัสประสบการณ์จริงในการสร้าง Knowledge Base ที่รองรับทั้งรูปภาพและเอกสารข้อความ พร้อมวัดผลด้านความหน่วง อัตราความสำเร็จ และต้นทุนการใช้งาน
Multi-Modal RAG คืออะไร และทำไมต้องสนใจ
Multi-Modal RAG คือระบบที่ผสานความสามารถในการค้นหาข้อมูลจากหลายรูปแบบ (รูปภาพ ข้อความ ตาราง) เข้าด้วยกัน ทำให้ AI สามารถตอบคำถามที่ต้องอาศัยข้อมูลจากทั้งสองแหล่งได้อย่างแม่นยำ สมมติว่าคุณมีคู่มือผลิตภัณฑ์ที่มีทั้งรูปภาพประกอบและคำอธิบายเทคนิค Multi-Modal RAG จะช่วยให้ผู้ใช้ถามคำถามเช่น "รูปนี้คือส่วนไหนของเครื่อง" หรือ "ชิ้นส่วนในรูปเข้ากันได้กับรุ่นไหนบ้าง" ได้ทันที
สถาปัตยกรรมระบบ Multi-Modal RAG
ระบบที่ผมพัฒนาขึ้นประกอบด้วย 4 ส่วนหลัก ได้แก่ Vector Store สำหรับเก็บ Embeddings, Image Encoder สำหรับแปลงรูปภาพเป็นเวกเตอร์, Text Splitter สำหรับแบ่งเอกสารข้อความ และ Fusion Retrieval สำหรับรวมผลลัพธ์จากทั้งสองแหล่ง โดยใช้ HolySheep AI สมัครที่นี่ เป็น LLM Backend หลัก ซึ่งให้อัตราแลกเปลี่ยนที่คุ้มค่ามาก: ¥1=$1 ประหยัดสูงสุด 85% เมื่อเทียบกับผู้ให้บริการอื่น รองรับการชำระเงินผ่าน WeChat และ Alipay และมีความหน่วงต่ำกว่า 50 มิลลิวินาที
การติดตั้งและเตรียม Environment
# ติดตั้ง Library ที่จำเป็น
pip install langchain openai tiktoken pillow torch transformers faiss-cpu
pip install python-multipart pydantic
สร้างไฟล์ .env สำหรับเก็บ API Key
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
EOF
ตรวจสอบการเชื่อมต่อ
python -c "from openai import OpenAI; \
client = OpenAI(api_key='YOUR_HOLYSHEEP_API_KEY', base_url='https://api.holysheep.ai/v1'); \
print(client.models.list())"โค้ดสำหรับสร้าง Multi-Modal Knowledge Base
import os
import base64
from pathlib import Path
from typing import List, Dict, Tuple
from PIL import Image
import faiss
import numpy as np
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader, TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
class MultiModalKnowledgeBase:
"""ระบบ Knowledge Base รองรับทั้งรูปภาพและข้อความ"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
from openai import OpenAI
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.embeddings = OpenAIEmbeddings(
model="text-embedding-3-large",
openai_api_key=api_key,
openai_api_base=base_url
)
# สร้าง FAISS Index สำหรับ Text และ Image แยกกัน
self.text_index = None
self.image_index = None
self.text_chunks = []
self.image_metadata = []
def encode_image_to_base64(self, image_path: str) -> str:
"""แปลงรูปภาพเป็น Base64 สำหรับส่งไปยัง API"""
with open(image_path, "rb") as img_file:
return base64.b64encode(img_file.read()).decode('utf-8')
def get_image_embedding(self, image_path: str) -> np.ndarray:
"""สร้าง Embedding จากรูปภาพโดยใช้ Vision Model"""
base64_image = self.encode_image_to_base64(image_path)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}},
{"type": "text", "text": "Describe this image in detail for semantic search. Include objects, colors, text, layout, and any technical details."}
]
}],
max_tokens=500
)
# ใช้คำตอบจาก Vision Model สร้าง Text Embedding
description = response.choices[0].message.content
return np.array(self.embeddings.embed_query(description))
def process_documents(self, documents_path: str) -> List:
"""ประมวลผลเอกสารข้อความ PDF และ TXT"""
docs = []
path = Path(documents_path)
for file in path.glob("**/*"):
if file.suffix == '.pdf':
loader = PyPDFLoader(str(file))
docs.extend(loader.load())
elif file.suffix == '.txt':
loader = TextLoader(str(file))
docs.extend(loader.load())
# แบ่งเอกสารเป็น Chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
return text_splitter.split_documents(docs)
def process_images(self, images_path: str) -> List[Dict]:
"""ประมวลผลรูปภาพทั้งหมดในโฟลเดอร์"""
results = []
path = Path(images_path)
for img_file in path.glob("**/*"):
if img_file.suffix.lower() in ['.jpg', '.jpeg', '.png', '.gif', '.webp']:
try:
# ตรวจสอบขนาดรูปก่อนประมวลผล
with Image.open(img_file) as img:
width, height = img.size
print(f"กำลังประมวลผล: {img_file.name} ({width}x{height})")
embedding = self.get_image_embedding(str(img_file))
results.append({
"path": str(img_file),
"filename": img_file.name,
"embedding": embedding,
"size": f"{width}x{height}"
})
except Exception as e:
print(f"ข้อผิดพลาดกับ {img_file.name}: {e}")
return results
def build_indexes(self, text_docs: List, image_data: List[Dict]):
"""สร้าง Vector Index สำหรับ Text และ Image"""
# สร้าง Text Index
if text_docs:
self.text_chunks = text_docs
self.text_index = FAISS.from_documents(text_docs, self.embeddings)
print(f"สร้าง Text Index สำเร็จ: {len(text_docs)} chunks")
# สร้าง Image Index
if image_data:
embeddings_matrix = np.array([item["embedding"] for item in image_data])
dimension = embeddings_matrix.shape[1]
self.image_index = faiss.IndexFlatL2(dimension)
self.image_index.add(embeddings_matrix)
self.image_metadata = [
{"path": item["path"], "filename": item["filename"], "size": item["size"]}
for item in image_data
]
print(f"สร้าง Image Index สำเร็จ: {len(image_data)} images")
def search(self, query: str, top_k: int = 4) -> Dict:
"""ค้นหาข้อมูลจากทั้ง Text และ Image Index"""
results = {"text_results": [], "image_results": [], "combined": []}
# ค้นหาใน Text Index
if self.text_index:
text_results = self.text_index.similarity_search(query, k=top_k)
results["text_results"] = [
{"content": doc.page_content[:200], "source": doc.metadata.get("source", "unknown")}
for doc in text_results
]
# ค้นหาใน Image Index
if self.image_index:
query_embedding = np.array(self.embeddings.embed_query(query)).reshape(1, -1)
distances, indices = self.image_index.search(query_embedding, top_k)
results["image_results"] = [
{
"filename": self.image_metadata[idx]["filename"],
"path": self.image_metadata[idx]["path"],
"size": self.image_metadata[idx]["size"],
"distance": float(distances[0][i])
}
for i, idx in enumerate(indices[0]) if idx < len(self.image_metadata)
]
# รวมผลลัพธ์ (Weighted Fusion)
results["combined"] = self._fusion_results(results["text_results"], results["image_results"])
return results
def _fusion_results(self, text_results: List, image_results: List) -> List:
"""รวมผลลัพธ์จาก Text และ Image ด้วย Weighted Score"""
combined = []
for i, tr in enumerate(text_results):
score = 1.0 / (i + 1) * 0.6 # Text Weight: 60%
combined.append({"type": "text", "score": score, **tr})
for i, ir in enumerate(image_results):
score = 1.0 / (i + 1) * 0.4 # Image Weight: 40%
combined.append({"type": "image", "score": score, **ir})
return sorted(combined, key=lambda x: x["score"], reverse=True)
ตัวอย่างการใช้งาน
def main():
api_key = "YOUR_HOLYSHEEP_API_KEY"
kb = MultiModalKnowledgeBase(api_key)
# ประมวลผลเอกสารและรูปภาพ
print("เริ่มสร้าง Knowledge Base...")
text_docs = kb.process_documents("./documents")
image_data = kb.process_images("./images")
# สร้าง Index
kb.build_indexes(text_docs, image_data)
# ค้นหาข้อมูล
query = "วิธีการติดตั้งชิ้นส่วน A1-B2"
results = kb.search(query)
print(f"\nผลการค้นหาสำหรับ: '{query}'")
print(f"- พบ Text: {len(results['text_results'])} รายการ")
print(f"- พบ Image: {len(results['image_results'])} รายการ")
if __name__ == "__main__":
main()โค้ด RAG Chain สำหรับตอบคำถาม Multi-Modal
import time
from openai import OpenAI
class MultiModalRAGChain:
"""RAG Chain ที่รวมข้อมูลจากทั้ง Text และ Image"""
def __init__(self, knowledge_base, api_key: str):
self.kb = knowledge_base
self.client = OpenAI(api_key=api_key, base_url="https://api.holysheep.ai/v1")
def generate_response(self, query: str, model: str = "gpt-4o") -> Dict:
"""สร้างคำตอบโดยใช้ RAG กับ Multi-Modal Context"""
start_time = time.time()
# 1. ค้นหาข้อมูลที่เกี่ยวข้อง
retrieved = self.kb.search(query, top_k=4)
# 2. สร้าง Context สำหรับ LLM
context_parts = []
# เพิ่ม Text Context
for tr in retrieved["text_results"][:2]:
context_parts.append(f"[Text from {tr['source']}]:\n{tr['content']}\n")
# เพิ่ม Image Context (แปลงเป็น Base64)
for ir in retrieved["image_results"][:2]:
try:
with open(ir["path"], "rb") as img_file:
img_b64 = base64.b64encode(img_file.read()).decode('utf-8')
context_parts.append(
f"[Image: {ir['filename']} ({ir['size']})]\n"
f"![{ir['filename']}](data:image/jpeg;base64,{img_b64})"
)
except Exception as e:
print(f"ไม่สามารถโหลดรูป {ir['filename']}: {e}")
context = "\n---\n".join(context_parts)
# 3. สร้าง Prompt สำหรับ Multi-Modal RAG
system_prompt = """คุณเป็นผู้ช่วยผู้เชี่ยวชาญ ตอบคำถามโดยอ้างอิงจาก Context ที่ได้รับ
หาก Context มีรูปภาพ ให้อธิบายหรืออ้างอิงถึงรูปภาพนั้นด้วย
ตอบเป็นภาษาไทย ชัดเจน และมีประโยชน์"""
user_prompt = f"""Context:
{context}
คำถาม: {query