Giới thiệu: Tại sao tôi cần thiết kế lại hệ thống
Sau 18 tháng vận hành hệ thống chatbot AI cho startup của tôi với hơn 2 triệu cuộc hội thoại được lưu trữ, tôi gặp phải vấn đề nghiêm trọng: chi phí API gốc tăng 340% trong năm qua. Đỉnh điểm là tháng 12, chúng tôi chi 12.000 USD chỉ riêng tiền API cho Claude Opus, trong khi doanh thu không tăng tương xứng.
Tôi bắt đầu tìm kiếm giải pháp thay thế và tình cờ phát hiện HolySheep AI — một API relay với tỷ giá chỉ ¥1=$1, tiết kiệm 85%+ so với giá gốc. Sau 6 tuần migration và tối ưu hóa, hệ thống của tôi không chỉ giảm 78% chi phí mà còn cải thiện độ trễ từ 890ms xuống còn 47ms trung bình.
Tại sao PostgreSQL + Vector Extension là lựa chọn tối ưu
Trước khi đi vào chi tiết migration, tôi muốn giải thích vì sao PostgreSQL với pgvector extension là stack hoàn hảo cho bài toán này:
- Chi phí vận hành thấp: Không cần Elasticsearch hay Pinecone riêng
- ACID compliance: Đảm bảo tính nhất quán cho metadata hội thoại
- Hybrid search: Kết hợp vector similarity với SQL query truyền thống
- Mature ecosystem: Backup, replication, monitoring đều có công cụ sẵn có
Kiến trúc hệ thống đề xuất
Schema thiết kế cho conversation history
-- Extension bắt buộc
CREATE EXTENSION IF NOT EXISTS vector;
-- Bảng chính lưu trữ hội thoại
CREATE TABLE conversations (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id UUID NOT NULL REFERENCES users(id),
session_id VARCHAR(128) NOT NULL,
-- Metadata
model VARCHAR(50) NOT NULL DEFAULT 'claude-opus-4-5',
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
token_usage INTEGER DEFAULT 0,
-- Trạng thái
status VARCHAR(20) DEFAULT 'active',
metadata JSONB DEFAULT '{}'::jsonb
);
-- Bảng messages với vector embedding
CREATE TABLE messages (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
conversation_id UUID REFERENCES conversations(id) ON DELETE CASCADE,
-- Nội dung
role VARCHAR(20) NOT NULL CHECK (role IN ('user', 'assistant', 'system')),
content TEXT NOT NULL,
-- Embedding vector (1536 chiều cho Claude embedding model)
embedding VECTOR(1536),
-- Timing
created_at TIMESTAMPTZ DEFAULT NOW(),
-- Usage stats
input_tokens INTEGER,
output_tokens INTEGER,
-- Index cho hybrid search
INDEX idx_conv_id (conversation_id),
INDEX idx_created (created_at)
);
-- Vector index cho semantic search
CREATE INDEX idx_message_embedding
ON messages USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
-- Partition theo tháng cho performance
CREATE TABLE messages_2026_01 PARTITION OF messages
FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
-- Function auto-update updated_at
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = NOW();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER conversations_updated_at
BEFORE UPDATE ON conversations
FOR EACH ROW EXECUTE FUNCTION update_updated_at();Service layer Python: HolySheep API Integration
# config.py
import os
HolySheep AI Configuration
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": os.environ.get("YOUR_HOLYSHEEP_API_KEY"),
"timeout": 30,
"max_retries": 3
}
Model pricing (USD per million tokens - 2026)
MODEL_PRICING = {
"claude-opus-4-5": {"input": 15.00, "output": 75.00},
"gpt-4.1": {"input": 8.00, "output": 24.00},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00},
"deepseek-v3.2": {"input": 0.42, "output": 1.68}
}
Database
DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://localhost:5432/chatbot")
Rate limiting
MAX_TOKENS_PER_MINUTE = 100000# services/holy_sheep_client.py
import httpx
from typing import Optional, List, Dict, Any
import asyncio
from datetime import datetime
import tiktoken
class HolySheepClient:
"""Client cho HolySheep AI API - thay thế api.anthropic.com"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
timeout=30.0,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
async def create_embedding(
self,
input_text: str,
model: str = "embedding-3"
) -> List[float]:
"""Tạo embedding cho văn bản"""
response = await self.client.post(
f"{self.base_url}/embeddings",
json={
"input": input_text,
"model": model,
"encoding_format": "float"
}
)
response.raise_for_status()
data = response.json()
return data["data"][0]["embedding"]
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "claude-opus-4-5",
temperature: float = 0.7,
max_tokens: int = 4096,
**kwargs
) -> Dict[str, Any]:
"""Gọi API chat completion"""
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
start_time = datetime.now()
response = await self.client.post(
f"{self.base_url}/chat/completions",
json=payload
)
latency = (datetime.now() - start_time).total_seconds() * 1000
response.raise_for_status()
result = response.json()
result["_latency_ms"] = latency
return result
async def batch_create_embeddings(
self,
texts: List[str],
model: str = "embedding-3"
) -> List[List[float]]:
"""Batch embedding cho performance"""
tasks = [self.create_embedding(text, model) for text in texts]
return await asyncio.gather(*tasks)
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Tính chi phí theo token"""
pricing = MODEL_PRICING.get(model, MODEL_PRICING["claude-opus-4-5"])
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6)
async def close(self):
await self.client.aclose()
Singleton instance
_client: Optional[HolySheepClient] = None
def get_holy_sheep_client() -> HolySheepClient:
global _client
if _client is None:
_client = HolySheepClient(
api_key=HOLYSHEEP_CONFIG["api_key"],
base_url=HOLYSHEEP_CONFIG["base_url"]
)
return _client# services/conversation_service.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select, update, delete, func, text
from sqlalchemy.dialects.postgresql import insert
from models.conversation import Conversation, Message
from services.holy_sheep_client import get_holy_sheep_client, HOLYSHEEP_CONFIG, MODEL_PRICING
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
class ConversationService:
def __init__(self, session: AsyncSession):
self.session = session
self.client = get_holy_sheep_client()
async def create_conversation(
self,
user_id: str,
session_id: str,
model: str = "claude-opus-4-5"
) -> Conversation:
"""Tạo cuộc hội thoại mới"""
conversation = Conversation(
user_id=user_id,
session_id=session_id,
model=model
)
self.session.add(conversation)
await self.session.commit()
await self.session.refresh(conversation)
return conversation
async def add_message(
self,
conversation_id: str,
role: str,
content: str,
generate_embedding: bool = True
) -> Message:
"""Thêm message và tự động tạo embedding"""
message = Message(
conversation_id=conversation_id,
role=role,
content=content
)
# Tạo embedding nếu cần
if generate_embedding and role == "user":
try:
embedding = await self.client.create_embedding(content)
message.embedding = embedding
logger.info(f"Embedding created for message, dimension: {len(embedding)}")
except Exception as e:
logger.warning(f"Failed to create embedding: {e}")
self.session.add(message)
await self.session.commit()
await self.session.refresh(message)
return message
async def chat_with_conversation(
self,
conversation_id: str,
user_message: str,
system_prompt: Optional[str] = None,
temperature: float = 0.7
) -> Dict[str, Any]:
"""Gửi message và nhận response từ AI"""
# Lấy lịch sử hội thoại
messages_history = await self.get_conversation_messages(conversation_id)
# Build messages list
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.extend(messages_history)
messages.append({"role": "user", "content": user_message})
# Gọi API
conversation = await self.session.get(Conversation, conversation_id)
model = conversation.model if conversation else "claude-opus-4-5"
response = await self.client.chat_completion(
messages=messages,
model=model,
temperature=temperature
)
# Lưu user message với embedding
await self.add_message(
conversation_id=conversation_id,
role="user",
content=user_message,
generate_embedding=True
)
# Lưu assistant response (không cần embedding)
assistant_content = response["choices"][0]["message"]["content"]
await self.add_message(
conversation_id=conversation_id,
role="assistant",
content=assistant_content,
generate_embedding=False
)
# Update token usage
usage = response.get("usage", {})
await self.update_token_usage(
conversation_id,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
return {
"content": assistant_content,
"latency_ms": response.get("_latency_ms"),
"usage": usage,
"cost": self.client.calculate_cost(
model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
}
async def semantic_search(
self,
user_id: str,
query: str,
limit: int = 5,
threshold: float = 0.7
) -> List[Dict[str, Any]]:
"""Tìm kiếm semantic trong lịch sử hội thoại"""
# Tạo embedding cho query
query_embedding = await self.client.create_embedding(query)
# Vector search
stmt = (
select(
Message,
func.vector_cosine_distance(Message.embedding, query_embedding).label("distance")
)
.join(Conversation)
.where(Conversation.user_id == user_id)
.where(Message.embedding.isnot(None))
.where(func.vector_cosine_distance(Message.embedding, query_embedding) < (1 - threshold))
.order_by("distance")
.limit(limit)
)
result = await self.session.execute(stmt)
rows = result.all()
return [
{
"message_id": str(msg.id),
"content": msg.content,
"role": msg.role,
"similarity": round(1 - dist, 4),
"created_at": msg.created_at.isoformat()
}
for msg, dist in rows
]
async def get_conversation_messages(
self,
conversation_id: str,
limit: int = 50
) -> List[Dict[str, str]]:
"""Lấy lịch sử messages cho context"""
stmt = (
select(Message)
.where(Message.conversation_id == conversation_id)
.order_by(Message.created_at.desc())
.limit(limit)
)
result = await self.session.execute(stmt)
messages = result.scalars().all()
return [
{"role": msg.role, "content": msg.content}
for msg in reversed(messages)
]
async def update_token_usage(
self,
conversation_id: str,
input_tokens: int,
output_tokens: int
):
"""Cập nhật tổng token usage"""
await self.session.execute(
update(Conversation)
.where(Conversation.id == conversation_id)
.values(
token_usage=Conversation.token_usage + input_tokens + output_tokens,
updated_at=datetime.now()
)
)
await self.session.commit()
async def get_conversation_stats(
self,
conversation_id: str
) -> Dict[str, Any]:
"""Lấy thống kê cuộc hội thoại"""
stmt = select(
func.count(Message.id).label("message_count"),
func.sum(Message.input_tokens).label("total_input"),
func.sum(Message.output_tokens).label("total_output")
).where(Message.conversation_id == conversation_id)
result = await self.session.execute(stmt)
row = result.one()
conversation = await self.session.get(Conversation, conversation_id)
return {
"conversation_id": str(conversation_id),
"message_count": row.message_count,
"total_input_tokens": row.total_input or 0,
"total_output_tokens": row.total_output or 0,
"model": conversation.model,
"created_at": conversation.created_at.isoformat(),
"estimated_cost": self.client.calculate_cost(
conversation.model,
row.total_input or 0,
row.total_output or 0
)
}Kế hoạch Migration từng bước
Phase 1: Preparation (Tuần 1-2)
# scripts/migration_preparation.py
"""
Script chuẩn bị migration - chạy trước khi switch sang HolySheep
"""
import asyncio
import httpx
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv
load_dotenv()
Kết nối database cũ
OLD_DB_URL = os.environ.get("OLD_DATABASE_URL")
NEW_DB_URL = os.environ.get("NEW_DATABASE_URL") # Sẽ dùng sau migration
def analyze_current_usage():
"""Phân tích usage hiện tại để ước tính chi phí"""
engine = create_engine(OLD_DB_URL)
with engine.connect() as conn:
# Đếm tổng messages
result = conn.execute(text("""
SELECT COUNT(*) as total_messages,
COUNT(DISTINCT conversation_id) as total_conversations,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output
FROM messages
WHERE created_at > NOW() - INTERVAL '90 days'
"""))
stats = result.fetchone()
print("=" * 60)
print("PHÂN TÍCH USAGE 90 NGÀY GẦN NHẤT")
print("=" * 60)
print(f"Tổng messages: {stats.total_messages:,}")
print(f"Tổng conversations: {stats.total_conversations:,}")
print(f"Tổng input tokens: {stats.total_input:,}")
print(f"Tổng output tokens: {stats.total_output:,}")
# Ước tính chi phí với các provider
providers = {
"API gốc (Anthropic)": {"input": 15.00, "output": 75.00},
"HolySheep Claude Sonnet 4.5": {"input": 15.00, "output": 75.00},
"HolySheep GPT-4.1": {"input": 8.00, "output": 24.00},
"HolySheep DeepSeek V3.2": {"input": 0.42, "output": 1.68}
}
print("\nƯỚC TÍNH CHI PHÍ VỚI HOLYSHEEP:")
print("-" * 60)
for provider, pricing in providers.items():
input_cost = (stats.total_input / 1_000_000) * pricing["input"]
output_cost = (stats.total_output / 1_000_000) * pricing["output"]
total = input_cost + output_cost
print(f"{provider}:")
print(f" - Input: ${input_cost:.2f}")
print(f" - Output: ${output_cost:.2f}")
print(f" - Tổng: ${total:.2f}")
print()
async def test_holy_sheep_connection():
"""Test kết nối HolySheep API"""
async with httpx.AsyncClient(timeout=30.0) as client:
# Test authentication
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {os.environ.get('YOUR_HOLYSHEEP_API_KEY')}",
"Content-Type": "application/json"
},
json={
"model": "claude-sonnet-4.5",
"messages": [{"role": "user", "content": "Test message"}],
"max_tokens": 10
}
)
if response.status_code == 200:
print("✓ Kết nối HolySheep API thành công!")
data = response.json()
print(f" - Model: {data.get('model')}")
print(f" - Latency: {response.elapsed.total_seconds() * 1000:.2f}ms")
return True
else:
print(f"✗ Lỗi kết nối: {response.status_code}")
print(f" Response: {response.text}")
return False
async def benchmark_embedding_performance():
"""Benchmark tốc độ embedding với HolySheep"""
test_texts = [
"Xin chào, tôi cần hỗ trợ về sản phẩm của bạn",
"Làm thế nào để đổi mật khẩu?",
"Tôi muốn biết về các gói premium"
] * 10 # 30 texts
async with httpx.AsyncClient(timeout=60.0) as client:
latencies = []
for text in test_texts:
start = asyncio.get_event_loop().time()
response = await client.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer {os.environ.get('YOUR_HOLYSHEEP_API_KEY')}",
"Content-Type": "application/json"
},
json={
"input": text,
"model": "embedding-3"
}
)
latency = (asyncio.get_event_loop().time() - start) * 1000
latencies.append(latency)
avg_latency = sum(latencies) / len(latencies)
print(f"\nBENCHMARK EMBEDDING PERFORMANCE:")
print(f" - Số lượng: {len(test_texts)} texts")
print(f" - Latency trung bình: {avg_latency:.2f}ms")
print(f" - Min: {min(latencies):.2f}ms")
print(f" - Max: {max(latencies):.2f}ms")
if __name__ == "__main__":
print("BẮT ĐẦU MIGRATION PREPARATION\n")
# Phân tích usage hiện tại
analyze_current_usage()
# Test HolySheep connection
asyncio.run(test_holy_sheep_connection())
# Benchmark
asyncio.run(benchmark_embedding_performance())Phase 2: Blue-Green Deployment
# scripts/blue_green_switch.py
"""
Blue-Green deployment để switch API provider an toàn
"""
import os
import asyncio
import httpx
from enum import Enum
from datetime import datetime
from typing import Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class Environment(Enum):
BLUE = "blue" # Production cũ (API gốc)
GREEN = "green" # Staging mới (HolySheep)
class BlueGreenSwitcher:
def __init__(self):
self.current_env = Environment.BLUE
self.fallback_enabled = True
self.error_count = 0
self.error_threshold = 5
# HolySheep config
self.holy_sheep_base = "https://api.holysheep.ai/v1"
self.holy_sheep_key = os.environ.get("YOUR_HOLYSHEEP_API_KEY")
# Original API config (sẽ ngừng sau khi stable)
self.original_base = os.environ.get("ORIGINAL_API_BASE")
self.original_key = os.environ.get("ORIGINAL_API_KEY")
async def call_api(
self,
messages: list,
model: str = "claude-sonnet-4.5",
**kwargs
) -> dict:
"""Gọi API với automatic fallback"""
# Thử HolySheep trước (primary)
try:
result = await self._call_holy_sheep(messages, model, **kwargs)
self.error_count = 0
return result
except Exception as e:
logger.warning(f"HolySheep error: {e}")
self.error_count += 1
# Fallback sang API gốc nếu cần
if self.fallback_enabled and self.current_env == Environment.GREEN:
logger.info("Falling back to original API...")
try:
return await self._call_original(messages, model, **kwargs)
except Exception as fallback_error:
logger.error(f"Fallback also failed: {fallback_error}")
raise
raise Exception("All API providers failed")
async def _call_holy_sheep(
self,
messages: list,
model: str,
**kwargs
) -> dict:
"""Gọi HolySheep API"""
async with httpx.AsyncClient(timeout=30.0) as client:
start_time = datetime.now()
response = await client.post(
f"{self.holy_sheep_base}/chat/completions",
headers={
"Authorization": f"Bearer {self.holy_sheep_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
**kwargs
}
)
latency = (datetime.now() - start_time).total_seconds() * 1000
if response.status_code != 200:
raise Exception(f"HolySheep API error: {response.status_code}")
result = response.json()
result["_provider"] = "holy_sheep"
result["_latency_ms"] = latency
logger.info(f"HolySheep success: {latency:.2f}ms")
return result
async def _call_original(
self,
messages: list,
model: str,
**kwargs
) -> dict:
"""Gọi API gốc (fallback)"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.original_base}/chat/completions",
headers={
"Authorization": f"Bearer {self.original_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
**kwargs
}
)
if response.status_code != 200:
raise Exception(f"Original API error: {response.status_code}")
result = response.json()
result["_provider"] = "original"
logger.info("Original API used (fallback)")
return result
async def switch_to_green(self):
"""Chuyển sang môi trường Green (HolySheep)"""
logger.info("Switching to GREEN environment (HolySheep)...")
self.current_env = Environment.GREEN
await self.health_check()
async def switch_to_blue(self):
"""Quay lại môi trường Blue (API gốc)"""
logger.info("Rolling back to BLUE environment...")
self.current_env = Environment.BLUE
async def health_check(self) -> bool:
"""Kiểm tra health của HolySheep"""
try:
result = await self._call_holy_sheep(
[{"role": "user", "content": "Health check"}],
model="claude-sonnet-4.5",
max_tokens=5
)
return True
except:
return False
def should_rollback(self) -> bool:
"""Kiểm tra xem có nên rollback không"""
return self.error_count >= self.error_threshold
Global instance
switcher = BlueGreenSwitcher()
API endpoint cho router
async def route_chat_completion(request: dict) -> dict:
"""Main entry point cho chat completion"""
# Automatic switch logic
if switcher.should_rollback():
logger.warning("Error threshold reached, initiating rollback...")
await switcher.switch_to_blue()
if switcher.current_env == Environment.GREEN:
return await switcher.call_api(
messages=request["messages"],
model=request.get("model", "claude-sonnet-4.5"),
temperature=request.get("temperature", 0.7),
max_tokens=request.get("max_tokens", 4096)
)
else:
return await switcher._call_original(
messages=request["messages"],
model=request.get("model", "claude-sonnet-4.5"),
temperature=request.get("temperature", 0.7),
max_tokens=request.get("max_tokens", 4096)
)Rollback Plan chi tiết
Khi thiết kế hệ thống, tôi luôn chuẩn bị kế hoạch rollback. Đây là những gì tôi đã làm:
# scripts/rollback_plan.py
"""
Kế hoạch Rollback chi tiết - chạy khi cần quay về API cũ
"""
import asyncio
import os
from datetime import datetime, timedelta
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RollbackManager:
def __init__(self):
self.db_url = os.environ.get("DATABASE_URL")
self.engine = create_engine(self.db_url)
self.Session = sessionmaker(bind=self.engine)
# Backup location
self.backup_path = f"/backups/rollback_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
async def pre_rollback_checks(self) -> bool:
"""Kiểm tra trước khi rollback"""
checks = {
"database_accessible": False,
"original_api_available": False,
"recent_backup_exists": False,
"no_active_migrations": True
}
# Check database
try:
with self.engine.connect() as conn:
conn.execute(text("SELECT 1"))
checks["database_accessible"] = True
logger.info("✓ Database accessible")
except Exception as e:
logger.error(f"✗ Database check failed: {e}")
# Check original API
try:
import httpx
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{os.environ.get('ORIGINAL_API_BASE')}/chat/completions",
headers={"Authorization": f"Bearer {os.environ.get('ORIGINAL_API_KEY')}"},
json={
"model": "claude-opus-4-5",
"messages": [{"role": "user", "content": "test"}],
"max_tokens": 1
}
)
checks["original_api_available"] = response.status_code == 200
logger.info(f"✓ Original API: {response.status_code}")
except Exception as e:
logger.warning(f"Original API check: {e}")
# Check backup
import os
backup_dir = "/backups"
if os.path.exists(backup_dir):
backups = sorted(os.listdir(backup_dir), reverse=True)
if backups:
latest = backups[0]
backup_time = datetime.fromisoformat(latest.split('_')[1])
checks["recent_backup_exists"] = datetime.now() - backup_time < timedelta(hours=24)
logger.info(f"✓ Backup exists: {latest}")
# Check for active migrations
# (Implement your migration lock check here)
return all(checks.values())
async def execute_rollback(self) -> dict:
"""Thực hiện rollback"""
logger.info("=" * 60)
logger.info("BẮT ĐẦU ROLLBACK PROCESS")
logger.info("=" * 60)
results = {
"success": False,
"steps_completed": [],
"errors": []
}
# Step 1: Stop all new requests
logger.info("[1/5] Draining active connections...")
try:
# Set maintenance mode
os.environ["MAINTENANCE_MODE"] = "true"
results["steps_completed"].append("maintenance_mode_enabled")
await asyncio.sleep(5) # Wait for requests to complete
except Exception as e:
results["errors"].append(f"Step 1 failed: {e}")
# Step 2: Update environment variables
logger.info("[2/5] Switching environment variables...")
try:
# Revert to original API
os.environ["ACTIVE_PROVIDER"] = "original"
os.environ["API_BASE_URL"] = os.environ.get("ORIGINAL_API_BASE", "")
results["steps_completed"].append("env_vars_updated")
except Exception as e:
results["errors"].append(f"Step 2 failed: {e}")
# Step 3: Restore configuration
logger.info("[3/5] Restoring configuration from backup...")
try:
# Restore from backup if needed
# (Implement restore logic)
results["steps_completed"].append("config_restored")
except Exception as e:
results["errors"].append(f"Step 3 failed: {e}")
# Step 4: Verify database consistency
logger.info("[4/5] Verifying database consistency...")
try:
with self.engine.connect() as conn:
# Check for orphaned messages
orphaned = conn.execute(text("""
SELECT COUNT(*) FROM messages
WHERE conversation_id NOT IN (SELECT id FROM conversations)
""")).scalar()
if orphaned > 0:
logger.warning(f"Found {orphaned} orphaned messages")
# Optionally clean up
results["steps_completed"].append("database_verified")
except Exception as e:
results["errors"].append(f"Step 4 failed: {e}")
# Step 5: Re-enable service
logger.info("[5/5] Re-enabling service...")
try:
os.environ["MAINTENANCE_MODE"] = "false"
results["steps_completed"].append("service_reenabled")
except Exception as e:
results["errors"].append(f"Step 5 failed: {e}")
results["success"] = len(results["errors"]) == 0
logger.info(f"Rollback completed: {results['success']}")
return results
async def generate_rollback_report(self) -> str:
"""Tạo báo cáo rollback"""
session = self.Session()