การพัฒนาระบบสนทนา AI ที่สามารถจดจำบริบทจากการสนทนารอบก่อนหน้าเป็นความท้าทายสำคัญสำหรับนักพัฒนา ไม่ว่าจะเป็นแชทบอทบริการลูกค้า ระบบค้นหาข้อมูลอัจฉริยะ หรือโปรเจกต์ส่วนตัว บทความนี้จะพาคุณไปรู้จักกับเทคนิคการจัดการบริบทหลายรอบ (Multi-turn Context Management) พร้อมโค้ดตัวอย่างที่พร้อมใช้งานจริง และวิธีเลือก API ที่เหมาะสมกับงบประมาณของคุณ
กรณีศึกษา: แชทบอท AI ลูกค้าสัมพันธ์สำหรับอีคอมเมิร์ซ
ทีมพัฒนาของร้านค้าออนไลน์แห่งหนึ่งเผชิญปัญหาการสร้างแชทบอทที่ต้องติดตาม:
- ความต้องการของลูกค้าในหลายรอบการสนทนา
- รายการสินค้าที่ลูกค้าสนใจ
- สถานะตะกร้าสินค้าและการสั่งซื้อ
- ประวัติการสนทนาก่อนหน้า
วิธีแก้ปัญหาคือการสร้าง Session Manager ที่จัดการสถานะและบริบทการสนทนาอย่างเป็นระบบ ด้วยการใช้ HolySheep AI ที่มีความหน่วงต่ำกว่า 50 มิลลิวินาที ทำให้การตอบสนองรวดเร็วและลื่นไหล
import requests
import json
from datetime import datetime
from typing import Dict, List, Optional
class EcommerceSessionManager:
"""ตัวจัดการเซสชันสำหรับระบบอีคอมเมิร์ซ"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.sessions: Dict[str, dict] = {}
def create_session(self, user_id: str, initial_context: dict = None) -> str:
"""สร้างเซสชันใหม่พร้อมบริบทเริ่มต้น"""
session_id = f"session_{user_id}_{int(datetime.now().timestamp())}"
self.sessions[session_id] = {
"user_id": user_id,
"messages": [],
"cart": [],
"preferences": {},
"created_at": datetime.now().isoformat(),
"context": initial_context or {}
}
return session_id
def add_message(self, session_id: str, role: str, content: str):
"""เพิ่มข้อความในประวัติการสนทนา"""
if session_id in self.sessions:
self.sessions[session_id]["messages"].append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
def get_context_window(self, session_id: str, max_tokens: int = 4000) -> List[dict]:
"""ดึงบริบทที่อยู่ในขีดจำกัด token"""
if session_id not in self.sessions:
return []
session = self.sessions[session_id]
messages = session["messages"]
# คำนวณ token โดยประมาณ (1 token ≈ 4 ตัวอักษร)
total_chars = sum(len(m["content"]) for m in messages)
estimated_tokens = total_chars // 4
# ถ้าเกินขีดจำกัด ให้ดึงเฉพาะข้อความล่าสุด
if estimated_tokens > max_tokens:
# เริ่มจากข้อความล่าสุดย้อนกลับไป
context_messages = []
current_tokens = 0
for msg in reversed(messages):
msg_tokens = len(msg["content"]) // 4
if current_tokens + msg_tokens <= max_tokens:
context_messages.insert(0, msg)
current_tokens += msg_tokens
else:
break
return context_messages
return messages
def send_to_ai(self, session_id: str, user_message: str) -> dict:
"""ส่งข้อความไปยัง API และรับการตอบกลับ"""
self.add_message(session_id, "user", user_message)
context = self.get_context_window(session_id)
system_prompt = self._build_system_prompt(session_id)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": system_prompt},
*context
],
"temperature": 0.7,
"max_tokens": 1000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
ai_response = result["choices"][0]["message"]["content"]
self.add_message(session_id, "assistant", ai_response)
return {"success": True, "response": ai_response}
return {"success": False, "error": response.text}
def _build_system_prompt(self, session_id: str) -> str:
"""สร้าง prompt ระบบที่มีบริบทของเซสชัน"""
session = self.sessions[session_id]
cart_info = f"สินค้าในตะกร้า: {', '.join(session['cart'])}" if session['cart'] else "ตะกร้าว่าง"
prefs = session.get('preferences', {})
return f"""คุณเป็นผู้ช่วยขายสินค้าออนไลน์
{cart_info}
ความชอบของลูกค้า: {prefs}
ตอบกลับอย่างเป็นมิตรและเป็นประโยชน์"""
ระบบ RAG องค์กร: การจัดการบริบทเอกสาร
องค์กรที่ต้องการค้นหาข้อมูลจากเอกสารจำนวนมากผ่านระบบ RAG (Retrieval-Augmented Generation) มักเผชิญปัญหา:
- บริบทการค้นหาหลุดจากกรอบที่ต้องการ
- ผลลัพธ์ไม่สอดคล้องกับคำถามต่อเนื่อง
- การจัดการ document chunk ที่ซับซ้อน
การใช้ Context Manager ที่ออกแบบมาอย่างดีจะช่วยรักษาความต่อเนื่องของบริบทและปรับปรุงความแม่นยำในการค้นหา
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import numpy as np
@dataclass
class Document:
"""โครงสร้างข้อมูลเอกสาร"""
content: str
metadata: Dict = field(default_factory=dict)
embedding: Optional[np.ndarray] = None
class RAGContextManager:
"""ตัวจัดการบริบทสำหรับระบบ RAG"""
def __init__(self, api_key: str, max_context_docs: int = 5):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_context_docs = max_context_docs
self.document_store: List[Document] = []
self.conversation_history: List[Dict] = []
def add_documents(self, documents: List[Document]):
"""เพิ่มเอกสารเข้าสู่คลังเอกสาร"""
self.document_store.extend(documents)
def retrieve_relevant_docs(self, query: str, top_k: int = 3) -> List[Document]:
"""ค้นหาเอกสารที่เกี่ยวข้องกับคำถาม"""
# จำลองการค้นหาด้วย keyword matching
# ในระบบจริงควรใช้ vector similarity
query_keywords = set(query.lower().split())
scored_docs = []
for doc in self.document_store:
doc_keywords = set(doc.content.lower().split())
overlap = len(query_keywords & doc_keywords)
if overlap > 0:
scored_docs.append((overlap, doc))
scored_docs.sort(reverse=True, key=lambda x: x[0])
return [doc for _, doc in scored_docs[:top_k]]
def add_to_history(self, role: str, content: str):
"""เพิ่มบันทึกการสนทนา"""
self.conversation_history.append({"role": role, "content": content})
# จำกัดประวัติการสนทนาไว้ที่ 10 รายการล่าสุด
if len(self.conversation_history) > 10:
self.conversation_history = self.conversation_history[-10:]
def build_rag_prompt(self, user_query: str) -> str:
"""สร้าง prompt สำหรับ RAG"""
relevant_docs = self.retrieve_relevant_docs(user_query, self.max_context_docs)
# รวมเนื้อหาเอกสารที่เกี่ยวข้อง
context_text = "\n\n".join([
f"[เอกสาร {i+1}]: {doc.content}"
for i, doc in enumerate(relevant_docs)
])
# เพิ่มบริบทจากการสนทนาก่อนหน้า
conversation_context = ""
if self.conversation_history:
recent = self.conversation_history[-4:]
conversation_context = "บริบทการสนทนาก่อนหน้า:\n" + \
"\n".join([f"{h['role']}: {h['content']}" for h in recent])
prompt = f"""คุณเป็นผู้ช่วยค้นหาข้อมูลจากเอกสารองค์กร
{context_text}
{conversation_context}
คำถามปัจจุบัน: {user_query}
ตอบกลับโดยอ้างอิงจากเอกสารที่ให้ไว้ ถ้าไม่พบข้อมูลที่เกี่ยวข้องให้ตอบว่า "ไม่พบข้อมูลในเอกสาร""""
return prompt
def query(self, user_query: str) -> Dict:
"""ส่งคำถามไปยัง AI พร้อมบริบท RAG"""
self.add_to_history("user", user_query)
prompt = self.build_rag_prompt(user_query)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 800
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
answer = result["choices"][0]["message"]["content"]
self.add_to_history("assistant", answer)
return {"success": True, "answer": answer}
return {"success": False, "error": response.text}
โปรเจกต์นักพัฒนาอิสระ: Multi-Agent Context System
นักพัฒนาอิสระที่ต้องการสร้างแชทบอทสำหรับลูกค้าของตนเองมักมีข้อจำกัดด้านงบประมาณและความซับซ้อนในการจัดการ API การใช้ HolySheep AI ช่วยลดภาระด้านค่าใช้จ่ายได้มากกว่า 85% เมื่อเทียบกับบริการอื่น พร้อมรองรับการชำระเงินผ่าน WeChat และ Alipay
import asyncio
import aiohttp
from