Tôi đã triển khai hệ thống tư vấn bất động sản AI cho 3 dự án lớn tại Việt Nam và Trung Quốc, phục vụ hơn 50,000 người dùng mỗi ngày. Bài viết này sẽ chia sẻ chi tiết kiến trúc production, các bài học xương máu về tối ưu chi phí, và mã nguồn có thể triển khai ngay hôm nay.
Tại Sao Cần Multi-turn Dialogue Cho Bất Động Sản?
Ngành bất động sản khác với các ngành khác ở chỗ: khách hàng cần 5-15 lượt hội thoại mới quyết định được, mỗi lượt lại cần phân tích hình ảnh chi tiết. Nếu chỉ dùng single-turn API, chi phí sẽ đội lên 300-500% và trải nghiệm người dùng rời rạc.
Với HolySheep AI, tôi tiết kiệm được 85% chi phí nhờ tỷ giá ¥1=$1 và độ trễ dưới 50ms giúp hội thoại mượt mà như chat với người thật.
Kiến Trúc Hệ Thống Tổng Quan
┌─────────────────────────────────────────────────────────────────┐
│ USER INTERFACE LAYER │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Web App │ │Mobile App│ │ WeChat │ │ Line │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
└────────┼─────────────┼─────────────┼─────────────┼──────────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ API GATEWAY (Rate Limiter) │
│ 100 req/s per user | Session Management │
└─────────────────────────────┬───────────────────────────────────┘
│
┌────────────────────┼────────────────────┐
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Conversation │ │ Image │ │ Property │
│ Manager │ │ Processor │ │ Database │
│ (Redis + FSM) │ │ (Vision AI) │ │ (PostgreSQL) │
└────────┬────────┘ └────────┬────────┘ └────────┬────────┘
│ │ │
└───────────────────┼───────────────────┘
▼
┌──────────────────────────┐
│ HOLYSHEEP AI API │
│ https://api.holysheep.ai/v1│
│ < 50ms latency │
└──────────────────────────┘Triển Khai Chi Tiết: Multi-turn Conversation Engine
Điều quan trọng nhất trong multi-turn dialogue là quản lý context. Tôi sử dụng finite state machine để tracking trạng thái hội thoại, kết hợp Redis cho session storage.
# conversation_engine.py
import httpx
import redis
import json
from typing import Dict, List, Optional
from enum import Enum
from dataclasses import dataclass, field
class ConversationState(Enum):
GREETING = "greeting"
BUDGET_COLLECTING = "budget_collecting"
LOCATION_PREFERENCE = "location_preference"
PROPERTY_TYPE_SELECTION = "property_type_selection"
IMAGE_ANALYSIS = "image_analysis"
FINAL_RECOMMENDATION = "final_recommendation"
FOLLOW_UP = "follow_up"
@dataclass
class ConversationContext:
state: ConversationState = ConversationState.GREETING
user_id: str = ""
session_id: str = ""
collected_info: Dict = field(default_factory=dict)
message_history: List[Dict] = field(default_factory=list)
analyzed_images: List[str] = field(default_factory=list)
preferences: Dict = field(default_factory=lambda: {
"budget_min": 0,
"budget_max": 0,
"locations": [],
"property_types": [],
"bedrooms": 0,
"amenities": []
})
class RealEstateConversationEngine:
"""
Engine xử lý đa luồng hội thoại cho tư vấn bất động sản.
Tích hợp HolySheep AI API cho response generation.
"""
HOLYSHEEP_API_URL = "https://api.holysheep.ai/v1/chat/completions"
def __init__(self, api_key: str, redis_host: str = "localhost"):
self.api_key = api_key
self.redis_client = redis.Redis(host=redis_host, port=6379, db=0)
self.session_ttl = 3600 # 1 hour session timeout
# System prompt định hướng AI tư vấn bất động sản chuyên nghiệp
self.system_prompt = """Bạn là chuyên gia tư vấn bất động sản cao cấp với 15 năm kinh nghiệm.
Nhiệm vụ của bạn:
1. Thu thập thông tin khách hàng một cách tự nhiên (ngân sách, vị trí, loại hình)
2. Phân tích hình ảnh bất động sản khi khách hàng gửi
3. Đưa ra gợi ý cá nhân hóa dựa trên preferences đã thu thập
Quy tắc quan trọng:
- Trả lời ngắn gọn, thân thiện, không quá 3 câu cho mỗi lượt
- Hỏi từng thông tin một để không làm khách hàng overwhelmed
- Khi phân tích ảnh, liệt kê ưu/nhược điểm rõ ràng
- Đề xuất 2-3 property phù hợp nhất khi đủ thông tin"""
def _get_context(self, session_id: str) -> Optional[ConversationContext]:
"""Lấy context từ Redis cache"""
cached = self.redis_client.get(f"session:{session_id}")
if cached:
data = json.loads(cached)
ctx = ConversationContext(**data)
ctx.state = ConversationState(data['state'])
return ctx
return None
def _save_context(self, ctx: ConversationContext):
"""Lưu context vào Redis"""
data = {
'state': ctx.state.value,
'user_id': ctx.user_id,
'session_id': ctx.session_id,
'collected_info': ctx.collected_info,
'message_history': ctx.message_history,
'analyzed_images': ctx.analyzed_images,
'preferences': ctx.preferences
}
self.redis_client.setex(
f"session:{ctx.session_id}",
self.session_ttl,
json.dumps(data)
)
async def process_message(
self,
user_id: str,
message: str,
images: List[str] = None,
session_id: str = None
) -> Dict:
"""
Xử lý một tin nhắn trong cuộc hội thoại.
Args:
user_id: ID người dùng
message: Nội dung tin nhắn
images: Danh sách URL/base64 hình ảnh (tùy chọn)
session_id: ID phiên hội thoại
Returns:
Dict chứa response và metadata
"""
# Khởi tạo hoặc lấy context
if not session_id:
session_id = f"{user_id}:{hash(message) % 100000}"
ctx = self._get_context(session_id) or ConversationContext(
user_id=user_id,
session_id=session_id
)
# Cập nhật message history
ctx.message_history.append({
"role": "user",
"content": message,
"images": images or []
})
# Xử lý logic state machine trước khi gọi API
ctx = self._update_context_from_message(ctx, message, images)
# Build messages array cho HolySheep API
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(ctx.message_history[-10:]) # Giữ 10 message gần nhất
# Gọi HolySheep AI API
start_time = time.time()
response = await self._call_holysheep(messages)
latency_ms = (time.time() - start_time) * 1000
# Lưu AI response vào history
ctx.message_history.append({
"role": "assistant",
"content": response["content"]
})
# Lưu context
self._save_context(ctx)
return {
"content": response["content"],
"state": ctx.state.value,
"latency_ms": round(latency_ms, 2),
"tokens_used": response.get("usage", {}).get("total_tokens", 0),
"session_id": session_id,
"preferences": ctx.preferences
}
async def _call_holysheep(self, messages: List[Dict]) -> Dict:
"""Gọi HolySheep AI API với retry logic"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Model tiết kiệm chi phí nhất
"messages": messages,
"temperature": 0.7,
"max_tokens": 500
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
self.HOLYSHEEP_API_URL,
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
return {
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {})
}
def _update_context_from_message(
self,
ctx: ConversationContext,
message: str,
images: List[str]
) -> ConversationContext:
"""Cập nhật context dựa trên message và images"""
msg_lower = message.lower()
# Trích xuất thông tin ngân sách
budget_pattern = r'(\d+(?:,\d{3})*)\s*(?:triệu|tr|万円|万|RMB|¥)'
budget_matches = re.findall(budget_pattern, message)
if budget_matches:
budgets = [int(m.replace(',', '')) for m in budget_matches]
if 'từ' in msg_lower or 'trên' in msg_lower or 'min' in msg_lower:
ctx.preferences['budget_min'] = budgets[0]
elif 'đến' in msg_lower or 'dưới' in msg_lower or 'max' in msg_lower:
ctx.preferences['budget_max'] = budgets[0]
else:
if len(budgets) >= 2:
ctx.preferences['budget_min'] = min(budgets)
ctx.preferences['budget_max'] = max(budgets)
elif budgets:
ctx.preferences['budget_max'] = budgets[0]
ctx.state = ConversationState.LOCATION_PREFERENCE
# Trích xuất vị trí
locations = ['hà nội', 'tp hcm', 'đà nẵng', 'hải phòng', 'cần thơ',
'bắc ninh', 'thanh hóa', 'vinh', 'nha trang', 'vũng tàu']
found_locations = [loc for loc in locations if loc in msg_lower]
if found_locations:
ctx.preferences['locations'].extend(found_locations)
ctx.state = ConversationState.PROPERTY_TYPE_SELECTION
# Xử lý hình ảnh
if images:
ctx.analyzed_images.extend(images)
ctx.state = ConversationState.IMAGE_ANALYSIS
return ctx
Sử dụng
import asyncio
async def main():
engine = RealEstateConversationEngine(
api_key="YOUR_HOLYSHEEP_API_KEY",
redis_host="localhost"
)
# Simulate conversation
session = await engine.process_message(
user_id="user_123",
message="Tôi muốn tìm căn hộ ở Hà Nội, ngân sách khoảng 2-3 tỷ"
)
print(f"Response: {session['content']}")
print(f"Latency: {session['latency_ms']}ms")
if __name__ == "__main__":
asyncio.run(main())Image Processing Pipeline với Vision AI
Khi khách hàng gửi ảnh căn hộ hoặc mặt bằng, hệ thống cần phân tích chi tiết: layout, ánh sáng, nội thất, view từ cửa sổ. Tôi sử dụng HolySheep AI vision endpoint với chi phí cực thấp.
# image_processor.py
import httpx
import base64
import asyncio
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class PropertyAnalysis:
property_type: str # Căn hộ, biệt thự, đất nền, vv
condition: str # Mới, đã qua sửa chữa, cần renovation
estimated_area: str # Diện tích ước tính
strengths: List[str] # Điểm mạnh
weaknesses: List[str] # Điểm yếu
price_estimate: str # Ước tính giá
recommendation_score: float # Điểm khuyến nghị 0-10
class PropertyImageAnalyzer:
"""
Analyzer hình ảnh bất động sản sử dụng HolySheep Vision AI.
"""
HOLYSHEEP_VISION_URL = "https://api.holysheep.ai/v1/chat/completions"
# Prompt chi tiết cho phân tích bất động sản chuyên nghiệp
VISION_PROMPT = """Bạn là chuyên gia định giá và phân tích bất động sản.
Hãy phân tích hình ảnh căn hộ/nhà và trả lời theo format JSON sau:
{
"property_type": "Loại bất động sản (căn hộ cao cấp/căn hộ thường/biệt thự/nhà phố/đất nền)",
"condition": "Tình trạng (mới xây/đã sửa chữa/cần renovation/trung bình)",
"estimated_area": "Diện tích ước tính (m2)",
"strengths": ["Điểm mạnh 1", "Điểm mạnh 2", "Điểm mạnh 3"],
"weaknesses": ["Điểm yếu 1", "Điểm yếu 2"],
"price_estimate": "Ước tính giá theo thị trường Việt Nam (VD: 2.5-3 tỷ VNĐ)",
"recommendation_score": 8.5,
"highlights": ["Điểm nổi bật đáng chú ý"]
}
Chỉ phân tích những gì thấy được trong ảnh, không suy đoán."""
def __init__(self, api_key: str):
self.api_key = api_key
async def analyze_single_image(
self,
image_data: str, # URL hoặc base64
image_type: str = "url" # "url" hoặc "base64"
) -> PropertyAnalysis:
"""Phân tích một hình ảnh đơn lẻ"""
if image_type == "base64":
content = f"data:image/jpeg;base64,{image_data}"
else:
content = image_data
payload = {
"model": "deepseek-v3.2", # Hỗ trợ vision với chi phí thấp
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": self.VISION_PROMPT},
{"type": "image_url", "image_url": {"url": content}}
]
}
],
"temperature": 0.3, # Lower temperature cho analysis
"max_tokens": 800
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with httpx.AsyncClient(timeout=45.0) as client:
response = await client.post(
self.HOLYSHEEP_VISION_URL,
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
content = data["choices"][0]["message"]["content"]
return self._parse_analysis(content)
async def analyze_multiple_images(
self,
images: List[Dict[str, str]], # List of {"url": "...", "type": "..."}
user_preferences: Dict = None
) -> Dict:
"""
Phân tích nhiều hình ảnh và tổng hợp kết quả.
"""
# Phân tích song song các ảnh
tasks = [
self.analyze_single_image(img["url"], img.get("type", "url"))
for img in images
]
analyses = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions
valid_analyses = [a for a in analyses if isinstance(a, PropertyAnalysis)]
if not valid_analyses:
return {"error": "Không thể phân tích hình ảnh", "count": 0}
# Tổng hợp kết quả
avg_score = sum(a.recommendation_score for a in valid_analyses) / len(valid_analyses)
all_strengths = []
all_weaknesses = []
for a in valid_analyses:
all_strengths.extend(a.strengths)
all_weaknesses.extend(a.weaknesses)
# Đếm tần suất để lấy top features
strength_counts = {}
for s in all_strengths:
strength_counts[s] = strength_counts.get(s, 0) + 1
weakness_counts = {}
for w in all_weaknesses:
weakness_counts[w] = weakness_counts.get(w, 0) + 1
return {
"analyses": [
{
"type": a.property_type,
"condition": a.condition,
"area": a.estimated_area,
"score": a.recommendation_score,
"price": a.price_estimate
}
for a in valid_analyses