Buổi sáng thứ Hai, tôi nhận được cuộc gọi từ đội devops: "Hệ thống chatbot AI của khách hàng bị crash sau 3 ngày hoạt động. Tất cả người dùng bị đăng xuất, conversation history biến mất." Đó là lần thứ ba trong tháng, và mỗi lần chi phí downtime lên tới hàng trăm triệu đồng.

Vấn đề nằm ở đâu? Memory management của AI Agent. Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi đánh giá và so sánh các hệ thống quản lý memory phổ biến nhất, đồng thời hướng dẫn bạn xây dựng giải pháp tối ưu với chi phí thấp nhất.

Tại sao AI Agent Memory lại quan trọng?

Khi xây dựng AI Agent, bạn cần hệ thống memory để:

Không có memory system hiệu quả, agent của bạn sẽ trở thành "cá vàng" — mỗi câu hỏi đều như bắt đầu từ đầu.

Các giải pháp AI Agent Memory phổ biến nhất 2026

1. Redis — Giải pháp in-memory cache

Redis là lựa chọn phổ biến với độ trễ cực thấp, nhưng yêu cầu cấu hình phức tạp và chi phí server cao.

# Cài đặt Redis connection
import redis
import json

class RedisMemoryStore:
    def __init__(self, host='localhost', port=6379, db=0):
        self.client = redis.Redis(
            host=host,
            port=port,
            db=db,
            decode_responses=True,
            socket_connect_timeout=5,
            socket_timeout=5
        )
    
    def save_conversation(self, session_id, messages, ttl=86400):
        """Lưu conversation với TTL 24 giờ"""
        key = f"conversation:{session_id}"
        data = json.dumps(messages)
        self.client.setex(key, ttl, data)
        
    def get_conversation(self, session_id):
        """Lấy conversation history"""
        key = f"conversation:{session_id}"
        data = self.client.get(key)
        return json.loads(data) if data else []
    
    def add_vector_cache(self, session_id, query, vector, ttl=3600):
        """Cache vector embedding"""
        key = f"vector:{session_id}:{hash(query)}"
        self.client.setex(key, ttl, json.dumps(vector))


Sử dụng

memory = RedisMemoryStore(host='prod-redis.internal', port=6379)
memory.save_conversation(
    "user_12345",
    [{"role": "user", "content": "Xin chào"},
     {"role": "assistant", "content": "Chào bạn!"}]
)
conversations = memory.get_conversation("user_12345")

2. PostgreSQL + pgvector — Database vector hibrid

Kết hợp relational database với vector search, phù hợp cho ứng dụng cần query phức tạp.

# PostgreSQL + pgvector setup
from sqlalchemy import create_engine, Column, String, DateTime, Text
from sqlalchemy.dialects.postgresql import ARRAY, JSONB
from pgvector.sqlalchemy import Vector
from datetime import datetime
import numpy as np

DATABASE_URL = "postgresql://user:pass@localhost:5432/agent_memory"

engine = create_engine(DATABASE_URL)

class ConversationMemory(Base):
    __tablename__ = 'conversation_memories'
    
    id = Column(String, primary_key=True)
    session_id = Column(String, index=True)
    user_id = Column(String, index=True)
    messages = Column(JSONB)
    embedding = Column(Vector(1536))  # OpenAI embedding dimension
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, onupdate=datetime.utcnow)
    metadata = Column(JSONB)

Base.metadata.create_all(engine)

def semantic_search_similar(query_embedding, session_id=None, top_k=5):
    """Tìm kiếm semantic trong lịch sử"""
    from sqlalchemy.orm import Session
    
    with Session(engine) as session:
        query = session.query(ConversationMemory)
        
        if session_id:
            query = query.filter(ConversationMemory.session_id == session_id)
        
        # Vector similarity search với cosine distance
        results = session.query(
            ConversationMemory,
            ConversationMemory.embedding.cosine_distance(query_embedding).label('distance')
        ).order_by('distance').limit(top_k).all()
        
        return [(r[0], r[1]) for r in results]


Tạo embedding và lưu

def save_with_embedding(session_id, user_id, messages, api_key):
    """Lưu conversation kèm embedding"""
    import requests
    
    # Tạo embedding cho toàn bộ conversation
    content = "\n".join([m['content'] for m in messages])
    
    response = requests.post(
        "https://api.holysheep.ai/v1/embeddings",
        headers={
            "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "input": content[:8000],  # Giới hạn token
            "model": "text-embedding-3-small"
        }
    )
    
    embedding = response.json()['data'][0]['embedding']
    
    with Session(engine) as session:
        memory = ConversationMemory(
            id=f"{session_id}_{datetime.utcnow().timestamp()}",
            session_id=session_id,
            user_id=user_id,
            messages=messages,
            embedding=np.array(embedding)
        )
        session.add(memory)
        session.commit()

3. Qdrant — Vector database chuyên dụng

# Qdrant setup cho AI Agent Memory
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Payload
import uuid

class QdrantMemoryStore:
    def __init__(self, host="localhost", port=6333):
        self.client = QdrantClient(host=host, port=port)
        self.collection_name = "agent_memory"
        self._init_collection()
    
    def _init_collection(self):
        """Khởi tạo collection với vector size phù hợp"""
        collections = self.client.get_collections().collections
        if self.collection_name not in [c.name for c in collections]:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
                optimizers_config={
                    "memmap_threshold_kb": 20000,
                    "indexing_threshold": 20000
                }
            )
    
    def store_memory(self, user_id, session_id, message, vector):
        """Lưu memory với vector embedding"""
        point = PointStruct(
            id=str(uuid.uuid4()),
            vector=vector,
            payload={
                "user_id": user_id,
                "session_id": session_id,
                "message": message,
                "timestamp": datetime.utcnow().isoformat()
            }
        )
        
        self.client.upsert(
            collection_name=self.collection_name,
            points=[point]
        )
    
    def retrieve_similar(self, query_vector, user_id, top_k=5, threshold=0.7):
        """Tìm memory tương tự cho context augmentation"""
        results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            query_filter={
                "must": [
                    {"key": "user_id", "match": {"value": user_id}}
                ]
            },
            score_threshold=threshold,
            limit=top_k
        )
        
        return [
            {"id": r.id, "score": r.score, "payload": r.payload}
            for r in results
        ]
    
    def get_session_context(self, session_id, limit=20):
        """Lấy context của một session cụ thể"""
        results = self.client.scroll(
            collection_name=self.collection_name,
            scroll_filter={
                "must": [
                    {"key": "session_id", "match": {"value": session_id}}
                ]
            },
            limit=limit
        )
        return [r.payload for r in results[0]]


Sử dụng kết hợp với HolySheep API cho embedding

def semantic_memory_lookup(question: str, user_id: str):
    """Tìm kiếm memory liên quan đến câu hỏi"""
    import requests
    
    # Tạo embedding từ HolySheep
    response = requests.post(
        "https://api.holysheep.ai/v1/embeddings",
        headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
        json={"input": question, "model": "text-embedding-3-small"}
    )
    
    query_vector = response.json()['data'][0]['embedding']
    
    memory_store = QdrantMemoryStore()
    similar_memories = memory_store.retrieve_similar(query_vector, user_id)
    
    # Ghép context vào prompt
    context = "\n".join([m['payload']['message'] for m in similar_memories])
    return f"Context from previous conversations:\n{context}"

4. HolySheep AI — Giải pháp tích hợp API + Memory

Sau khi thử nghiệm nhiều giải pháp, team của tôi đã chuyển sang HolySheep AI cho phần backend API và tích hợp custom memory layer. Đây là giải pháp tối ưu về chi phí và độ trễ.

# HolySheep AI Agent Memory System
import requests
import hashlib
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional

class HolySheepAgentMemory:
    """Memory system tích hợp HolySheep API"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.memory_cache = {}  # In-memory cache cho demo
        self.conversation_window = 10  # Số tin nhắn giữ lại
        
    def _get_embedding(self, text: str) -> List[float]:
        """Tạo embedding qua HolySheep API"""
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "input": text,
                "model": "text-embedding-3-small"
            },
            timeout=10
        )
        response.raise_for_status()
        return response.json()['data'][0]['embedding']
    
    def _chat_completion(self, messages: List[Dict], **kwargs):
        """Gọi chat completion qua HolySheep"""
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": kwargs.get("model", "gpt-4o-mini"),
                "messages": messages,
                "temperature": kwargs.get("temperature", 0.7),
                "max_tokens": kwargs.get("max_tokens", 1000)
            },
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    
    def add_to_memory(self, session_id: str, role: str, content: str):
        """Thêm message vào session memory"""
        if session_id not in self.memory_cache:
            self.memory_cache[session_id] = {
                "messages": [],
                "created_at": datetime.utcnow()
            }
        
        self.memory_cache[session_id]["messages"].append({
            "role": role,
            "content": content,
            "timestamp": datetime.utcnow().isoformat()
        })
        
        # Trim nếu quá dài
        if len(self.memory_cache[session_id]["messages"]) > 50:
            self.memory_cache[session_id]["messages"] = \
                self.memory_cache[session_id]["messages"][-50:]
    
    def get_recent_context(self, session_id: str, last_n: int = None) -> List[Dict]:
        """Lấy context gần đây từ memory"""
        if session_id not in self.memory_cache:
            return []
        
        messages = self.memory_cache[session_id]["messages"]
        if last_n:
            return messages[-last_n:]
        return messages[-self.conversation_window:]
    
    def semantic_recall(self, session_id: str, query: str, top_k: int = 3) -> str:
        """Tìm kiếm semantic trong lịch sử cuộc trối chuyện"""
        messages = self.memory_cache.get(session_id, {}).get("messages", [])
        
        if not messages:
            return ""
        
        # Tạo embedding cho query
        query_embedding = self._get_embedding(query)
        
        # Tính similarity với từng message
        scored_messages = []
        for msg in messages:
            msg_embedding = self._get_embedding(msg['content'])
            similarity = self._cosine_similarity(query_embedding, msg_embedding)
            scored_messages.append((similarity, msg))
        
        # Sắp xếp và lấy top_k
        scored_messages.sort(key=lambda x: x[0], reverse=True)
        relevant = scored_messages[:top_k]
        
        # Format kết quả
        context_parts = [
            f"[{msg[1]['role']}]: {msg[1]['content']}"
            for score, msg in relevant
            if score > 0.5
        ]
        
        return "\n".join(context_parts)
    
    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        """Tính cosine similarity"""
        dot = sum(x * y for x, y in zip(a, b))
        norm_a = sum(x * x for x in a) ** 0.5
        norm_b = sum(x * x for x in b) ** 0.5
        return dot / (norm_a * norm_b) if norm_a and norm_b else 0
    
    def chat(self, session_id: str, user_message: str, 
             use_memory: bool = True, use_recall: bool = True) -> str:
        """Chat với memory context"""
        # Thêm user message vào memory
        self.add_to_memory(session_id, "user", user_message)
        
        # Build system prompt với memory
        system_prompt = """Bạn là một AI Assistant thông minh. 
Bạn có khả năng nhớ ngữ cảnh cuộc trò chuyện trước đó."""
        
        messages = [{"role": "system", "content": system_prompt}]
        
        if use_memory:
            recent = self.get_recent_context(session_id)
            for msg in recent:
                messages.append({"role": msg["role"], "content": msg["content"]})
        
        if use_recall and len(user_message) > 10:
            recall_context = self.semantic_recall(session_id, user_message)
            if recall_context:
                messages.append({
                    "role": "system",
                    "content": f"Thông tin liên quan từ cuộc trò chuyện trước:\n{recall_context}"
                })
        
        # Thêm user message hiện tại
        messages.append({"role": "user", "content": user_message})
        
        # Gọi API
        response = self._chat_completion(messages, model="gpt-4o-mini")
        assistant_reply = response['choices'][0]['message']['content']
        
        # Lưu assistant response vào memory
        self.add_to_memory(session_id, "assistant", assistant_reply)
        
        return assistant_reply
    
    def clear_session(self, session_id: str):
        """Xóa memory của một session"""
        if session_id in self.memory_cache:
            del self.memory_cache[session_id]
    
    def get_session_stats(self, session_id: str) -> Dict:
        """Lấy thống kê session"""
        if session_id not in self.memory_cache:
            return {"exists": False}
        
        session = self.memory_cache[session_id]
        return {
            "exists": True,
            "message_count": len(session["messages"]),
            "created_at": session["created_at"].isoformat(),
            "last_message": session["messages"][-1]["timestamp"] if session["messages"] else None
        }



============== DEMO ==============

if __name__ == "__main__":
    # Khởi tạo với API key
    memory_system = HolySheepAgentMemory("YOUR_HOLYSHEEP_API_KEY")
    
    session_id = "user_session_001"
    
    # Cuộc trò chuyện 1
    print("=== Cuộc trò chuyện 1 ===")
    reply1 = memory_system.chat(session_id, "Tôi đang làm dự án AI chatbot bằng Python")
    print(f"Assistant: {reply1}")
    
    # Cuộc trò chuyện 2 (hỏi về project cũ)
    print("\n=== Cuộc trò chuyện 2 ===")
    reply2 = memory_system.chat(session_id, "Dự án đó dùng framework gì?")
    print(f"Assistant: {reply2}")
    
    # Kiểm tra memory
    print("\n=== Memory Stats ===")
    stats = memory_system.get_session_stats(session_id)
    print(f"Session info: {json.dumps(stats, indent=2)}")

Bảng so sánh chi tiết các giải pháp

Tiêu chí Redis PostgreSQL + pgvector Qdrant HolySheep AI
Độ trễ trung bình <5ms 20-50ms 10-30ms <50ms (API)
Chi phí hàng tháng $50-500 $30-300 $40-400 $0-50
Setup phức tạp Cao Trung bình Trung bình Thấp
Tích hợp Vector Search Cần module riêng Có (pgvector) Native Qua API
Scalability Manual sharding Replication Distributed Managed
Bảo mật Tự quản lý Tự quản lý Tự quản lý Enterprise-grade
Phù hợp cho Cache thời gian thực Database chính Vector retrieval AI Agent production

So sánh chi phí thực tế theo tháng

Dưới đây là bảng chi phí thực tế khi vận hành AI Agent với 10,000 người dùng hoạt động:

Hạng mục OpenAI API Claude API HolySheep AI
Model GPT-4.1 Claude Sonnet 4.5 DeepSeek V3.2
Giá/MT $8.00 $15.00 $0.42
Token/người/ngày 5,000 5,000 5,000
Tổng token/tháng 1.5 tỷ 1.5 tỷ 1.5 tỷ
Chi phí API/tháng $12,000 $22,500 $630
Chi phí Infra (memory) $200 $200 $50
Tổng chi phí $12,200 $22,700 $680
Tiết kiệm so với OpenAI — -86% -94%

Lỗi thường gặp và cách khắc phục

1. Lỗi kết nối Redis: "ConnectionError: Error 111 Connection Refused"

# ❌ Sai: Không có retry và timeout
client = redis.Redis(host='localhost', port=6379)
data = client.get('key')  # Crash nếu Redis down


✅ Đúng: Retry pattern với exponential backoff

import time
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (redis.ConnectionError, redis.TimeoutError) as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    print(f"Retry {attempt + 1}/{max_retries} sau {delay}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

class ResilientRedisClient:
    def __init__(self, hosts: list):
        self.hosts = hosts
        self.client = None
        self.current_host = 0
        self._connect()
    
    def _connect(self):
        for i, host in enumerate(self.hosts):
            try:
                self.client = redis.Redis(
                    host=host['host'],
                    port=host['port'],
                    socket_connect_timeout=3,
                    socket_timeout=5,
                    retry_on_timeout=True
                )
                self.client.ping()
                self.current_host = i
                print(f"✅ Kết nối Redis thành công: {host['host']}")
                return
            except Exception as e:
                print(f"❌ Không kết nối được {host['host']}: {e}")
                continue
        raise Exception("Không có Redis server khả dụng")
    
    @retry_with_backoff(max_retries=3, base_delay=2)
    def get(self, key: str):
        return self.client.get(key)
    
    @retry_with_backoff(max_retries=3, base_delay=2)
    def set(self, key: str, value: str, ex: int = None):
        return self.client.set(key, value, ex=ex)
    
    def failover(self):
        """Chuyển sang server dự phòng"""
        self.current_host = (self.current_host + 1) % len(self.hosts)
        self._connect()


Sử dụng

redis_client = ResilientRedisClient([
    {'host': 'redis-primary.internal', 'port': 6379},
    {'host': 'redis-replica.internal', 'port': 6379}
])

2. Lỗi Memory Leak: "Response 200 nhưng context dần trống"

# ❌ Nguyên nhân: Memory không được cleanup, dict grow vô hạn
class BadMemoryStore:
    def __init__(self):
        self.all_conversations = {}  # Never cleaned!
    
    def add(self, session_id, message):
        if session_id not in self.all_conversations:
            self.all_conversations[session_id] = []
        self.all_conversations[session_id].append(message)
        # Dict grow forever → OOM crash


✅ Đúng: TTL + size limit + periodic cleanup

from collections import OrderedDict
from threading import Lock
import time

class MemoryStoreWithCleanup:
    def __init__(self, max_sessions=10000, max_messages_per_session=100, ttl_seconds=86400):
        self.max_sessions = max_sessions
        self.max_messages = max_messages_per_session
        self.ttl = ttl_seconds
        
        # OrderedDict để LRU eviction
        self.sessions = OrderedDict()
        self.locks = {}
        self.timestamps = {}
        self.lock = Lock()
    
    def _get_session_lock(self, session_id):
        if session_id not in self.locks:
            self.locks[session_id] = Lock()
        return self.locks[session_id]
    
    def add_message(self, session_id: str, role: str, content: str):
        with self._get_session_lock(session_id):
            current_time = time.time()
            
            # Cleanup expired sessions
            self._cleanup_expired()
            
            # LRU eviction nếu đầy
            if session_id not in self.sessions and len(self.sessions) >= self.max_sessions:
                self._evict_lru()
            
            if session_id not in self.sessions:
                self.sessions[session_id] = []
                self.timestamps[session_id] = current_time
            
            self.sessions[session_id].append({
                'role': role,
                'content': content,
                'timestamp': current_time
            })
            
            # Trim nếu quá dài
            if len(self.sessions[session_id]) > self.max_messages:
                self.sessions[session_id] = self.sessions[session_id][-self.max_messages:]
            
            # Move to end (most recently used)
            self.sessions.move_to_end(session_id)
    
    def get_messages(self, session_id: str, limit: int = None):
        with self._get_session_lock(session_id):
            if session_id not in self.sessions:
                return []
            
            messages = self.sessions[session_id]
            if limit:
                messages = messages[-limit:]
            return messages.copy()
    
    def _cleanup_expired(self):
        current_time = time.time()
        expired = [
            sid for sid, ts in self.timestamps.items()
            if current_time - ts > self.ttl
        ]
        for sid in expired:
            self._remove_session(sid)
    
    def _evict_lru(self):
        if self.sessions:
            lru_session = next(iter(self.sessions))
            self._remove_session(lru_session)
            print(f"🗑️ Evicted LRU session: {lru_session}")
    
    def _remove_session(self, session_id):
        if session_id in self.sessions:
            del self.sessions[session_id]
        if session_id in self.timestamps:
            del self.timestamps[session_id]
        if session_id in self.locks:
            del self.locks[session_id]
    
    def get_stats(self):
        return {
            'active_sessions': len(self.sessions),
            'max_sessions': self.max_sessions,
            'memory_usage_pct': len(self.sessions) / self.max_sessions * 100
        }


Periodic cleanup thread

import threading
def start_cleanup_scheduler(store: MemoryStoreWithCleanup, interval=3600):
    def cleanup_loop():
        while True:
            time.sleep(interval)
            with store.lock:
                stats = store.get_stats()
                print(f"📊 Memory stats: {stats}")
                store._cleanup_expired()
    
    thread = threading.Thread(target=cleanup_loop, daemon=True)
    thread.start()
    return thread


Sử dụng

memory = MemoryStoreWithCleanup(
    max_sessions=5000,
    max_messages_per_session=50,
    ttl_seconds=3600  # 1 giờ
)
start_cleanup_scheduler(memory, interval=1800)

3. Lỗi Token Overflow: "429 Too Many Requests" hoặc "Context length exceeded"

# ❌ Sai: Không kiểm soát context length
def chat(messages):
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=messages  # Có thể exceed limit!
    )
    return response


✅ Đúng: Dynamic context truncation

import tiktoken

class TokenAwareMemory:
    def __init__(self, api_key: str, max_tokens: int = 6000):
        self.api_key = api_key
        self.max_tokens = max_tokens
        self.base_url = "https://api.holysheep.ai/v1"
        # Encoding cho model tương ứng
        try:
            self.encoding = tiktoken.get_encoding("cl100k_base")
        except:
            self.encoding = None
    
    def count_tokens(self, text: str) -> int:
        if self.encoding:
            return len(self.encoding.encode(text))
        # Fallback estimation
        return len(text) // 4
    
    def truncate_messages(self, messages: list, reserve_tokens: int = 500) -> list:
        """Truncate messages để fit vào context limit"""
        available_tokens = self.max_tokens - reserve_tokens
        
        # System message giữ lại nếu có
        system_msg = None
        working_messages = []
        
        for msg in messages:
            if msg.get('role') == 'system':
                system_msg = msg
            else:
                working_messages.append(msg)
        
        # Tính tokens hiện tại
        current_tokens = sum(
            self.count_tokens(m['content']) for m in working_messages
        )
        
        # Truncate từ cũ nhất nếu quá
        while current_tokens > available_tokens and working_messages:
            removed = working_messages.pop(0)
            current_tokens -= self.count_tokens(removed['content'])
        
        # Ghép lại
        result = []
        if system_msg:
            result.append(system_msg)
        result.extend(working_messages)
        
        return result
    
    def smart_chat(self, session_messages: list, new_message: str) -> dict:
        """Chat với token awareness"""
        messages = session_messages + [{"role": "user", "content": new_message}]
        
        # Truncate nếu cần
        messages = self.truncate_messages(messages)
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "deepseek-chat",
                "messages": messages,
                "max_tokens": 1000
            },
            timeout=30
        )
        
        if response.status_code == 429:
            # Rate limit → retry sau
            time.sleep(5)
            return self.smart_chat(session_messages, new_message)
        
        response.raise_for_status()
        return response.json()


Sử dụng

token_manager = TokenAwareMemory("YOUR_HOLYSHEEP_API_KEY")

messages = [{"role": "user", "content": "Nói về lịch sử Việt Nam"}]

Thêm nhiều messages...


result = token_manager.smart_chat(messages, "Tiếp tục đi")

Kiến trúc hoàn chỉnh cho AI Agent Memory System

Dựa trên kinh nghiệm triển khai thực tế, đây là kiến trúc tối ưu:

# Complete AI Agent Memory Architecture
import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime
import json

@dataclass
class Message:
    role: str
    content: str
    timestamp: datetime = field(default_factory=datetime

Tài nguyên liên quan
Bài viết liên quan