Mở Đầu: Khi Đoạn Hội Thoại 23 Lượt Bị "Quên" Giữa Chừng
Tháng 9 năm ngoái, tôi nhận được một cuộc gọi lúc 2 giờ sáng từ đội vận hành. Hệ thống chat AI của một trang thương mại điện tử lớn tại Việt Nam — nơi tôi đang tư vấn kiến trúc — đang trong tình trạng chết máy. Nguyên nhân không phải server quá tải hay mạng lag, mà là một bug logic kinh điển: Agent "quên" mất rằng khách hàng đã chọn màu đen cho đôi giày, và bắt đầu đề xuất lại các phương án màu sắc từ đầu.
Đó là lúc tôi nhận ra rằng quản lý trạng thái hội thoại (Dialog State Management) không phải thứ bạn có thể xử lý bằng một biến session đơn giản và hy vọng mọi thứ sẽ ổn. Sau 3 tháng nghiên cứu, benchmark thực tế trên 50 triệu đoạn hội thoại, và triển khai thành công cho 4 doanh nghiệp, tôi muốn chia sẻ với bạn bài phân tích toàn diện nhất về ba phương pháp quản lý trạng thái Agent: **FSM (Finite State Machine)**, **Graph-based** và **LLM Router**.
Tại Sao Trạng Thái Hội Thoại Lại Quan Trọng Đến Vậy?
Trước khi đi sâu vào kỹ thuật, hãy hiểu rõ bản chất vấn đề. Một Agent đối thoại không đơn thuần là "hỏi - đáp". Nó phải:
- Duy trì ngữ cảnh xuyên suốt nhiều lượt hội thoại (cross-turn context)
- Theo dõi mục đích của người dùng (intent tracking)
- Cập nhật thông tin đã xác nhận (slot filling)
- Xử lý các luồng phức tạp như quay lại bước trước, hủy tác vụ, hoặc chuyển hướng giữa chừng
- Đảm bảo tính nhất quán khi tích hợp nhiều tool/API
Một hệ thống có thể trả lời đúng từng câu một cách riêng lẻ, nhưng lại thất bại thảm hại khi người dùng nói: *"Đợi đã, tôi muốn đổi sang màu đỏ nhưng size 42 thôi, và giao cho tôi ở Quận 1"*. Đó là lúc bạn cần một kiến trúc quản lý trạng thái được thiết kế bài bản.
Ba Phương Pháp Quản Lý Trạng Thái Hội Thoại
1. FSM (Finite State Machine) — Đơn Giản Nhưng Mạnh Mẽ
FSM là phương pháp kinh điển nhất, nơi hội thoại được mô hình hóa thành các trạng thái rời rạc với các transition được định nghĩa trước.
"""
FSM-based Dialog State Manager cho hệ thống đặt hàng
Triển k khai bởi HolySheep AI — holy.sheep
"""
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, Callable, Dict
import json
class DialogState(Enum):
START = "start"
GREETING = "greeting"
COLLECTING_INTENT = "collecting_intent"
COLLECTING_PRODUCT = "collecting_product"
COLLECTING_COLOR = "collecting_color"
COLLECTING_SIZE = "collecting_size"
COLLECTING_ADDRESS = "collecting_address"
CONFIRMATION = "confirmation"
ORDER_PLACED = "order_placed"
HANDLING_ISSUE = "handling_issue"
END = "end"
@dataclass
class DialogContext:
"""Lưu trữ trạng thái và dữ liệu hội thoại"""
state: DialogState = DialogState.START
user_id: str = ""
intent: Optional[str] = None
product: Optional[str] = None
color: Optional[str] = None
size: Optional[str] = None
address: Optional[str] = None
history: list = field(default_factory=list)
error_count: int = 0
class FSMDialogManager:
"""
FSM quản lý hội thoại đặt hàng thương mại điện tử
Transition được định nghĩa rõ ràng, không ambiguous
"""
def __init__(self):
self.contexts: Dict[str, DialogContext] = {}
# Định nghĩa transition: (state hiện tại, intent) -> state tiếp theo
self.transitions: Dict[tuple, DialogState] = {
(DialogState.START, "greet"): DialogState.GREETING,
(DialogState.GREETING, "order"): DialogState.COLLECTING_PRODUCT,
(DialogState.COLLECTING_PRODUCT, "product_selected"): DialogState.COLLECTING_COLOR,
(DialogState.COLLECTING_COLOR, "color_selected"): DialogState.COLLECTING_SIZE,
(DialogState.COLLECTING_SIZE, "size_selected"): DialogState.COLLECTING_ADDRESS,
(DialogState.COLLECTING_ADDRESS, "address_confirmed"): DialogState.CONFIRMATION,
(DialogState.CONFIRMATION, "confirm"): DialogState.ORDER_PLACED,
(DialogState.CONFIRMATION, "modify"): DialogState.COLLECTING_PRODUCT,
(DialogState.ORDER_PLACED, "ask_issue"): DialogState.HANDLING_ISSUE,
(DialogState.HANDLING_ISSUE, "resolved"): DialogState.END,
# Xử lý fallback
(DialogState.COLLECTING_PRODUCT, "change_mind"): DialogState.COLLECTING_PRODUCT,
(DialogState.COLLECTING_COLOR, "skip"): DialogState.COLLECTING_SIZE,
}
def get_context(self, session_id: str) -> DialogContext:
if session_id not in self.contexts:
self.contexts[session_id] = DialogContext(user_id=session_id)
return self.contexts[session_id]
def process_input(self, session_id: str, user_input: str,
extracted_intent: str = None, slots: dict = None) -> tuple:
"""
Xử lý input từ người dùng và trả về response cùng state mới
Returns: (response_text, new_state, context_updated)
"""
ctx = self.get_context(session_id)
slots = slots or {}
# Cập nhật slots vào context
self._update_slots(ctx, slots)
# Xác định intent nếu chưa có
if not extracted_intent:
extracted_intent = self._classify_intent(user_input, ctx)
# Tìm transition hợp lệ
transition_key = (ctx.state, extracted_intent)
if transition_key in self.transitions:
new_state = self.transitions[transition_key]
ctx.state = new_state
else:
# Fallback: ở lại state hiện tại, tăng error count
ctx.error_count += 1
# Sinh response dựa trên state
response = self._generate_response(ctx)
# Lưu vào history
ctx.history.append({
"input": user_input,
"intent": extracted_intent,
"state": ctx.state.value,
"slots": slots
})
return response, ctx.state, ctx
def _update_slots(self, ctx: DialogContext, slots: dict):
"""Cập nhật các slot đã xác nhận"""
if "product" in slots:
ctx.product = slots["product"]
if "color" in slots:
ctx.color = slots["color"]
if "size" in slots:
ctx.size = slots["size"]
if "address" in slots:
ctx.address = slots["address"]
def _classify_intent(self, text: str, ctx: DialogContext) -> str:
"""Intent classification đơn giản — có thể thay bằng LLM"""
text_lower = text.lower()
if any(w in text_lower for w in ["mua", "order", "đặt", "want", "buy"]):
return "order"
if any(w in text_lower for w in ["đỏ", "xanh", "đen", "trắng", "red", "blue", "black"]):
return "color_selected"
if any(w in text_lower for w in ["size", "37", "38", "39", "40", "41", "42"]):
return "size_selected"
if any(w in text_lower for w in ["xác nhận", "ok", "đồng ý", "confirm", "yes", "đặt đi"]):
return "confirm"
if any(w in text_lower for w in ["sửa", "đổi", "thay", "change", "modify"]):
return "modify"
if any(w in text_lower for w in ["chào", "hello", "hi", "xin chào"]):
return "greet"
return "unknown"
def _generate_response(self, ctx: DialogContext) -> str:
"""Sinh response dựa trên state hiện tại"""
responses = {
DialogState.GREETING: f"Xin chào! Tôi có thể giúp bạn đặt hàng hôm nay. Bạn muốn mua sản phẩm gì?",
DialogState.COLLECTING_PRODUCT: "Bạn muốn đặt sản phẩm nào ạ?",
DialogState.COLLECTING_COLOR: f"Bạn chọn màu nào: đỏ, xanh, đen hay trắng?",
DialogState.COLLECTING_SIZE: f"Cho tôi biết size giày bạn cần (37-42)?",
DialogState.COLLECTING_ADDRESS: "Vui lòng cung cấp địa chỉ giao hàng:",
DialogState.CONFIRMATION: f"Tóm tắt đơn hàng:\n- Sản phẩm: {ctx.product}\n- Màu: {ctx.color}\n- Size: {ctx.size}\n- Địa chỉ: {ctx.address}\n\nXác nhận đặt hàng?",
DialogState.ORDER_PLACED: f"Đơn hàng #{ctx.user_id[:8].upper()} đã được đặt thành công! Cảm ơn bạn!",
DialogState.HANDLING_ISSUE: "Tôi đang hỗ trợ bạn giải quyết vấn đề..."
}
return responses.get(ctx.state, "Xin lỗi, tôi chưa hiểu. Bạn có thể nói rõ hơn không?")
Sử dụng FSM với HolySheep AI cho intent classification
import aiohttp
async def classify_with_holysheep(session_id: str, user_input: str, api_key: str):
"""
Sử dụng HolySheep AI (base_url: https://api.holysheep.ai/v1)
Chi phí chỉ $0.42/MTok với DeepSeek V3.2 — tiết kiệm 85%+ so với GPT-4
"""
url = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "Bạn là classifier intent cho chatbot thương mại điện tử Việt Nam. Trả về JSON: {\"intent\": \"...\", \"slots\": {...}}"},
{"role": "user", "content": user_input}
],
"temperature": 0.1
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as resp:
result = await resp.json()
content = result["choices"][0]["message"]["content"]
# Parse JSON intent
import json
try:
return json.loads(content)
except:
return {"intent": "unknown", "slots": {}}
Demo
async def main():
fsm = FSMDialogManager()
session_id = "user_12345"
# Lượt 1: Chào hỏi
resp1, state1, _ = fsm.process_input(session_id, "Xin chào")
print(f"State: {state1.value}, Response: {resp1}")
# Lượt 2: Muốn đặt giày
resp2, state2, ctx = fsm.process_input(session_id, "Tôi muốn đặt đôi giày thể thao Nike",
slots={"product": "Nike Air Max"})
print(f"State: {state2.value}, Response: {resp2}")
print(f"Context: product={ctx.product}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Ưu điểm của FSM
- **Dễ debug**: Mọi transition đều có thể trace, không có "magic" hay hidden state
- **Performance cao**: O(1) lookup cho mỗi transition, không cần LLM inference
- **Audit trail hoàn chỉnh**: Dễ dàng log và monitor mọi bước
- **Compliance-ready**: Phù hợp với các ngành cần giải thích được quyết định (fintech, healthcare)
Nhược điểm của FSM
- Cứng nhắc với các luồng hội thoại tự nhiên
- Khó xử lý multi-turn context phức tạp
- Yêu cầu định nghĩa trước tất cả các state và transition
2. Graph-based Dialog Management — Linh Hoạt Cho Hệ Thống Phức Tạp
Khi FSM trở nên quá phức tạp với hàng trăm transition, Graph-based approach mang đến sự linh hoạt vượt trội. Mỗi node là một "step" trong hội thoại, và các edge thể hiện các điều kiện/chuyển đổi có thể có.
"""
Graph-based Dialog Manager cho RAG Enterprise System
Xử lý multi-domain, multi-turn với ngữ cảnh phức tạp
Triển khai bởi HolySheep AI — holy.sheep
"""
from typing import Dict, List, Optional, Set, Any
from dataclasses import dataclass, field
from enum import Enum
import networkx as nx
from collections import defaultdict
class NodeType(Enum):
QUESTION = "question" # Hỏi thông tin
ACTION = "action" # Thực hiện action (API call, DB query)
CONDITION = "condition" # Kiểm tra điều kiện
RESPONSE = "response" # Trả lời người dùng
SUBGRAPH = "subgraph" # Gọi subgraph khác
END = "end" # Kết thúc
@dataclass
class DialogNode:
node_id: str
node_type: NodeType
content: str
required_slots: Set[str] = field(default_factory=set)
produced_slots: Set[str] = field(default_factory=set)
conditions: Dict[str, Any] = field(default_factory=dict)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class DialogEdge:
source: str
target: str
condition: str # Biểu thức điều kiện
priority: int = 0 # Thứ tự ưu tiên khi có nhiều edge thỏa mãn
class GraphDialogManager:
"""
Dialog Manager dựa trên Directed Graph
- Hỗ trợ parallel paths (multi-domain)
- Dynamic routing theo context
- Subgraph reuse cho các luồng phổ biến
"""
def __init__(self):
self.graph = nx.DiGraph()
self.nodes: Dict[str, DialogNode] = {}
self.edges: List[DialogEdge] = []
self.global_slots: Dict[str, Any] = {}
def add_node(self, node: DialogNode):
"""Thêm node vào graph"""
self.nodes[node.node_id] = node
self.graph.add_node(node.node_id, **node.metadata)
def add_edge(self, edge: DialogEdge):
"""Thêm edge với điều kiện chuyển"""
self.edges.append(edge)
self.graph.add_edge(edge.source, edge.target,
condition=edge.condition, priority=edge.priority)
def build_order_flow(self):
"""Build subgraph cho luồng đặt hàng"""
# Nodes cho luồng đặt hàng
self.add_node(DialogNode(
node_id="order_start",
node_type=NodeType.QUESTION,
content="Bạn muốn đặt sản phẩm gì?",
produced_slots={"product_query"}
))
self.add_node(DialogNode(
node_id="search_product",
node_type=NodeType.ACTION,
content="Đang tìm sản phẩm...",
required_slots={"product_query"},
produced_slots={"product_list"}
))
self.add_node(DialogNode(
node_id="product_selection",
node_type=NodeType.CONDITION,
content="Chọn sản phẩm",
conditions={"type": "selection", "options": "product_list"}
))
# Vòng lặp thu thập thông tin
self.add_node(DialogNode(
node_id="collect_color",
node_type=NodeType.QUESTION,
content="Bạn muốn màu nào?",
produced_slots={"color"}
))
self.add_node(DialogNode(
node_id="collect_size",
node_type=NodeType.QUESTION,
content="Size bao nhiêu?",
produced_slots={"size"}
))
self.add_node(DialogNode(
node_id="collect_address",
node_type=NodeType.QUESTION,
content="Địa chỉ giao hàng?",
produced_slots={"address"}
))
self.add_node(DialogNode(
node_id="confirmation",
node_type=NodeType.RESPONSE,
content="Xác nhận đơn hàng"
))
# Edges với conditions
self.add_edge(DialogEdge("order_start", "search_product", "true"))
self.add_edge(DialogEdge("search_product", "product_selection", "product_list not empty"))
self.add_edge(DialogEdge("product_selection", "collect_color", "product_selected"))
self.add_edge(DialogEdge("collect_color", "collect_size", "color filled"))
self.add_edge(DialogEdge("collect_size", "collect_address", "size filled"))
self.add_edge(DialogEdge("collect_address", "confirmation", "address filled"))
# Edge quay lại để sửa
self.add_edge(DialogEdge("confirmation", "collect_color", "want_modify"))
def build_inquiry_flow(self):
"""Build subgraph cho luồng hỏi thông tin (RAG-based)"""
self.add_node(DialogNode(
node_id="inquiry_start",
node_type=NodeType.QUESTION,
content="Bạn muốn hỏi về sản phẩm/policy gì?",
produced_slots={"inquiry_query"}
))
self.add_node(DialogNode(
node_id="rag_query",
node_type=NodeType.ACTION,
content="Đang truy vấn knowledge base...",
required_slots={"inquiry_query"},
produced_slots={"rag_results"}
))
self.add_node(DialogNode(
node_id="inquiry_response",
node_type=NodeType.RESPONSE,
content="Trả lời dựa trên RAG results"
))
self.add_edge(DialogEdge("inquiry_start", "rag_query", "true"))
self.add_edge(DialogEdge("rag_query", "inquiry_response", "rag_results found"))
self.add_edge(DialogEdge("inquiry_response", "inquiry_start", "has_followup"))
def find_next_node(self, current_node_id: str, context: Dict) -> Optional[str]:
"""Tìm node tiếp theo dựa trên conditions và context"""
# Lấy tất cả outgoing edges
outgoing_edges = [(e, self.graph[e.source][e.target])
for e in self.edges
if e.source == current_node_id]
if not outgoing_edges:
return None
# Sort theo priority giảm dần
outgoing_edges.sort(key=lambda x: x[0].priority, reverse=True)
for edge, edge_data in outgoing_edges:
if self._evaluate_condition(edge.condition, context):
return edge.target
return None
def _evaluate_condition(self, condition: str, context: Dict) -> bool:
"""Evaluate condition string với context"""
# Đơn giản hóa: thực tế nên dùng expression parser
if condition == "true":
return True
# Ví dụ: "color filled" -> context.get("color") is not None
if "filled" in condition:
slot_name = condition.split()[0]
return context.get(slot_name) is not None
if "not empty" in condition:
slot_name = condition.split()[0]
val = context.get(slot_name)
return val is not None and (isinstance(val, list) and len(val) > 0 or len(str(val)) > 0)
return False
async def execute_with_holysheep_rag(self, query: str, api_key: str) -> Dict:
"""
Thực thi RAG query sử dụng HolySheep AI
Hỗ trợ context window 128K tokens với DeepSeek V3.2
"""
url = "https://api.holysheep.ai/v1/chat/completions"
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": """Bạn là assistant cho hệ thống RAG enterprise.
Sử dụng retrieved context để trả lời chính xác.
Nếu không tìm thấy thông tin, hãy nói rõ 'Tôi không tìm thấy thông tin về...'"""},
{"role": "user", "content": f"Context: {self._get_relevant_context(query)}\n\nQuery: {query}"}
],
"temperature": 0.3,
"max_tokens": 2000
}
import aiohttp
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as resp:
result = await resp.json()
return {
"answer": result["choices"][0]["message"]["content"],
"model": "deepseek-v3.2",
"latency_ms": resp.headers.get("X-Response-Time", "N/A")
}
def _get_relevant_context(self, query: str) -> str:
"""Lấy relevant context từ knowledge base (simulated)"""
# Trong thực tế, đây sẽ là vector search
return "[Retrieved from enterprise KB based on query similarity]"
Demo Graph Dialog Manager
async def demo_graph_manager():
manager = GraphDialogManager()
manager.build_order_flow()
manager.build_inquiry_flow()
context = {
"product_query": "giày Nike",
"product_list": ["Nike Air Max 90", "Nike Jordan 1"],
"product_selected": True,
"color": "đen",
"size": "42"
}
# Simulate navigation
current = "order_start"
while current:
node = manager.nodes.get(current)
print(f"📍 Node: {current} ({node.node_type.value})")
print(f" Content: {node.content}")
if node.node_type == NodeType.END:
break
current = manager.find_next_node(current, context)
print()
if __name__ == "__main__":
import asyncio
asyncio.run(demo_graph_manager())
Ưu điểm của Graph-based
- **Linh hoạt**: Dễ dàng thêm/sửa/remove nodes và edges mà không ảnh hưởng toàn bộ hệ thống
- **Visualizable**: Có thể visualize bằng tools như Graphviz để review flow
- **Reusable subgraphs**: Các subgraph có thể được reuse giữa các domain
- **Parallel paths**: Hỗ trợ multiple valid paths cho cùng một intent
Nhược điểm của Graph-based
- Complexity tăng nhanh với số lượng nodes
- Cần quản lý state graph (serialization, versioning)
- Debugging có thể khó hơn FSM khi có nhiều paths
3. LLM Router — AI-Native Approach Cho Tự Nhiên
LLM Router là approach hiện đại nhất, nơi LLM đóng vai trò "bộ não" quyết định state transitions. Thay vì định nghĩa trước mọi thứ, bạn cung cấp system prompt và để LLM xử lý logic.
"""
LLM Router cho Dialog State Management
Sử dụng HolySheep AI với DeepSeek V3.2 — chi phí thấp, latency <50ms
Triển khai bởi HolySheep AI — holy.sheep
"""
import json
from typing import Dict, List, Optional, Literal
from dataclasses import dataclass, field
from enum import Enum
import aiohttp
class AgentMode(Enum):
"""Các chế độ hoạt động của Agent"""
GREETING = "greeting"
ORDER = "order"
INQUIRY = "inquiry"
COMPLAINT = "complaint"
GENERAL = "general"
@dataclass
class ConversationTurn:
role: Literal["user", "assistant", "system"]
content: str
mode: Optional[AgentMode] = None
@dataclass
class DialogStateV2:
"""Enhanced state với LLM-native approach"""
session_id: str
current_mode: AgentMode = AgentMode.GENERAL
collected_info: Dict[str, any] = field(default_factory=dict)
conversation_history: List[ConversationTurn] = field(default_factory=list)
pending_actions: List[str] = field(default_factory=list)
turn_count: int = 0
class LLM DialogRouter:
"""
LLM Router — để LLM quyết định state transitions
Phù hợp cho hội thoại tự nhiên, không cần define trước mọi flow
"""
SYSTEM_PROMPT = """Bạn là AI Assistant cho hệ thống thương mại điện tử.
Nhiệm vụ của bạn:
1. Phân tích input của user
2. Quyết định mode hoạt động (greeting/order/inquiry/complaint/general)
3. Cập nhật collected_info nếu có thông tin mới
4. Quyết định action tiếp theo (ask_question/answer_inquiry/confirm_order/...)
5. Xác định is_complete để biết có thể kết thúc hội thoại không
Luôn trả về JSON format:
{
"mode": "order|inquiry|greeting|complaint|general",
"action": "ask_question|answer|confirm|clarify|...",
"collected_info": {"key": "value"},
"response": "Nội dung phản hồi cho user",
"is_complete": true|false,
"next_slots_needed": ["slot1", "slot2"] // thông tin cần thu thập tiếp
}
Quy tắc:
- Nếu thiếu thông tin cần thiết cho action, hỏi user
- Nếu user cung cấp nhiều thông tin một lúc, parse và thu thập tất cả
- Nếu user muốn sửa thông tin đã cung cấp, cập nhật collected_info
- Ưu tiên trả lời chính xác và đầy đủ trước khi hỏi thêm
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1" # HolySheep endpoint
self.sessions: Dict[str, DialogStateV2] = {}
def get_session(self, session_id: str) -> DialogStateV2:
if session_id not in self.sessions:
self.sessions[session_id] = DialogStateV2(session_id=session_id)
return self.sessions[session_id]
async def process(self, session_id: str, user_input: str) -> Dict:
"""
Xử lý một lượt hội thoại
Trả về response và updated state
"""
session = self.get_session(session_id)
session.turn_count += 1
# Build messages cho LLM
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT},
]
# Thêm context về current state
context_msg = f"""
Current State:
- Mode: {session.current_mode.value}
- Collected Info: {json.dumps(session.collected_info, ensure_ascii=False)}
- Turn: {session.turn_count}
- History Length: {len(session.conversation_history)} turns
"""
messages.append({"role": "system", "content": context_msg})
# Thêm conversation history (giới hạn để tiết kiệm tokens)
for turn in session.conversation_history[-10:]:
messages.append({
"role": turn.role,
"
Tài nguyên liên quan
Bài viết liên quan