Ở bài viết trước, chúng ta đã xây dựng một AI Agent Memory API bằng Python và FastAPI. Tuy nhiên, việc lưu lịch sử hội thoại trong dictionary chỉ phù hợp với demo: dữ liệu mất khi process khởi động lại, còn nhiều instance của ứng dụng không tự động chia sẻ cùng một bộ nhớ.
Để khắc phục những hạn chế đó, chúng ta có thể sử dụng Redis làm nơi lưu trữ session và conversation history. Redis hỗ trợ truy cập dữ liệu nhanh, nhiều cấu trúc dữ liệu và cơ chế TTL (Time To Live) để key tự hết hạn sau một khoảng thời gian.
Trong bài viết này, MCNA Technology School sẽ hướng dẫn xây dựng một API quản lý lịch sử hội thoại bằng Redis và FastAPI, từ cấu trúc dữ liệu đến cách chạy thử và những lưu ý khi triển khai thực tế.
1. Vì sao AI Agent cần Redis Memory?
Khi một chatbot hoặc AI Agent được nhiều người sử dụng, hệ thống cần lưu trạng thái hội thoại ở nơi có thể được truy cập từ các request khác nhau.
Ví dụ, một ứng dụng có hai instance FastAPI:
User request
|
v
Load Balancer
/ \
v v
API A API B
\ /
\ /
Redis
Shared MemoryNếu lịch sử chỉ nằm trong RAM của API A, một request tiếp theo được chuyển đến API B có thể không tìm thấy dữ liệu. Khi cả hai instance sử dụng cùng một Redis và cùng quy tắc định danh key, chúng có thể truy cập chung kho dữ liệu hội thoại.
Redis giải quyết vấn đề gì?
- Chia sẻ trạng thái: Nhiều instance có thể truy cập chung dữ liệu Redis.
- TTL: Session có thể tự hết hạn sau một khoảng thời gian.
- Truy cập nhanh: Phù hợp với dữ liệu thường xuyên được đọc và cập nhật.
- Cấu trúc dữ liệu linh hoạt: Có thể sử dụng Lists, Hashes hoặc các kiểu dữ liệu khác tùy nhu cầu.
Redis không tự động giải quyết toàn bộ vấn đề về phân quyền, độ bền dữ liệu hoặc quản lý vòng đời hội thoại. Những phần này vẫn cần được thiết kế dựa trên yêu cầu cụ thể của ứng dụng.
Tài liệu tham khảo: Redis – Asynchronous operations with redis-py.
2. TTL là gì và tại sao cần thiết?
TTL là viết tắt của Time To Live, tức khoảng thời gian một key còn tồn tại trước khi hết hạn. Redis cho phép đặt thời gian hết hạn bằng các lệnh như EXPIRE hoặc thông qua tùy chọn thời hạn khi tạo key.
Ví dụ, một session được thiết lập TTL là 3.600 giây sẽ hết hạn sau khoảng một giờ nếu thời hạn không được gia hạn.
SET agent:session:123 "conversation data" EX 3600
TTL agent:session:123Lệnh đầu tiên lưu dữ liệu và đặt thời hạn một giờ. Lệnh thứ hai kiểm tra thời gian sống còn lại của key.
Đối với AI Agent, TTL giúp hạn chế việc lưu giữ session tạm thời quá lâu. Tuy nhiên, TTL của Redis key không thay thế chính sách lưu trữ dữ liệu, backup hoặc yêu cầu xóa dữ liệu của một hệ thống production.
Tham khảo: Redis – EXPIRE và Redis – TTL.
3. Thiết kế cấu trúc dữ liệu cho AI Agent Memory
Trong ví dụ này, mỗi cuộc hội thoại sẽ được lưu bằng một Redis List. Mỗi phần tử của List là một message được mã hóa thành JSON.
Quy ước key:
agent:session:{user_id}:{conversation_id}Ví dụ:
agent:session:user_001:conversation_abcDữ liệu bên trong có thể gồm các message như sau:
[
{
"role": "user",
"content": "Phân tích doanh thu tháng này"
},
{
"role": "assistant",
"content": "Doanh thu tăng 12% so với tháng trước."
}
]Trong ví dụ này, user ID được dùng để phân vùng dữ liệu theo người dùng, còn conversation ID xác định cuộc hội thoại cụ thể. Khi triển khai thật, user ID phải được lấy từ danh tính đã xác thực ở server, không được tin tưởng một giá trị do client tự gửi lên.
4. Chuẩn bị môi trường
Bước 1: Cài đặt thư viện Python
Tạo file requirements.txt:
fastapi
uvicorn[standard]
redisCài đặt các thư viện:
python -m pip install -r requirements.txtChúng ta sẽ dùng API bất đồng bộ của redis-py thông qua redis.asyncio. Cách này phù hợp với ứng dụng FastAPI sử dụng async/await khi thao tác với Redis qua mạng.
Bước 2: Khởi chạy Redis
Nếu đã cài Docker, có thể khởi chạy Redis bằng lệnh sau trong môi trường phát triển:
docker run --name ai-agent-redis \
-p 127.0.0.1:6379:6379 \
-d redis:7Lệnh trên dành cho thử nghiệm trên máy cá nhân, không phải cấu hình production. Trong môi trường thực tế, nên dùng phiên bản image được pin cụ thể, đặt thông tin xác thực và cấu hình mạng phù hợp; không mở Redis công khai ra Internet.
5. Xây dựng Redis Memory Store bằng Python
Tiếp theo, tạo file main.py. Ví dụ này có các chức năng tạo hội thoại, thêm message, lấy lịch sử và gia hạn TTL khi có hoạt động mới.
import json
import os
from contextlib import asynccontextmanager
from uuid import uuid4
import redis.asyncio as redis
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
REDIS_URL = os.getenv(
"REDIS_URL",
"redis://localhost:6379/0"
)
SESSION_TTL = 3600
MAX_MESSAGES = 20
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = redis.from_url(
REDIS_URL,
decode_responses=True
)
await app.state.redis.ping()
try:
yield
finally:
await app.state.redis.aclose()
app = FastAPI(
title="AI Agent Redis Memory API",
lifespan=lifespan
)
class ChatRequest(BaseModel):
user_id: str = Field(min_length=1, max_length=100)
conversation_id: str
message: str = Field(min_length=1, max_length=2000)
def session_key(user_id, conversation_id):
return (
f"agent:session:{user_id}:"
f"{conversation_id}"
)
@app.post("/conversations")
async def create_conversation(user_id: str):
conversation_id = str(uuid4())
key = session_key(user_id, conversation_id)
await app.state.redis.rpush(
key,
json.dumps({
"role": "system",
"content": (
"Bạn là trợ lý AI. "
"Trả lời rõ ràng và trung thực."
})
)
)
await app.state.redis.expire(
key,
SESSION_TTL
)
return {
"user_id": user_id,
"conversation_id": conversation_id,
"ttl_seconds": SESSION_TTL
}
@app.post("/chat")
async def chat(request: ChatRequest):
key = session_key(
request.user_id,
request.conversation_id
)
r = app.state.redis
if not await r.exists(key):
raise HTTPException(
status_code=404,
detail="Conversation not found or expired"
)
user_message = {
"role": "user",
"content": request.message
}
await r.rpush(
key,
json.dumps(user_message, ensure_ascii=False)
)
history_length = await r.llen(key)
# Giữ system message đầu tiên và các message gần nhất.
if history_length > MAX_MESSAGES + 1:
await r.ltrim(key, -MAX_MESSAGES, -1)
await r.lpush(
key,
json.dumps({
"role": "system",
"content": (
"Bạn là trợ lý AI. "
"Trả lời rõ ràng và trung thực."
)
})
)
# Gia hạn session khi có hoạt động mới.
await r.expire(key, SESSION_TTL)
answer = (
f"Đã lưu câu hỏi của bạn: {request.message}"
)
await r.rpush(
key,
json.dumps({
"role": "assistant",
"content": answer
}, ensure_ascii=False)
)
await r.expire(key, SESSION_TTL)
return {
"conversation_id": request.conversation_id,
"answer": answer
}
@app.get("/conversations/{conversation_id}")
async def get_history(
conversation_id: str,
user_id: str
):
key = session_key(user_id, conversation_id)
r = app.state.redis
messages = await r.lrange(key, 0, -1)
if not messages:
raise HTTPException(
status_code=404,
detail="Conversation not found or expired"
)
history = [
json.loads(message)
for message in messages
]
return {
"conversation_id": conversation_id,
"history": history,
"ttl_seconds": await r.ttl(key)
}Lưu ý về ví dụ: Đây là bản minh họa cách sử dụng Redis, chưa phải triển khai production hoàn chỉnh. Các endpoint đang nhận user ID từ request để dễ kiểm thử; trong hệ thống thực tế phải lấy user ID từ thông tin xác thực đã được server kiểm tra. Cần bổ sung xử lý lỗi Redis, giới hạn tốc độ và cơ chế cập nhật lịch sử an toàn khi có nhiều request đồng thời.
6. Chạy thử API
Khởi động ứng dụng:
uvicorn main:app --reloadMở địa chỉ http://127.0.0.1:8000/docs để kiểm thử API.
Bước 1: Tạo cuộc hội thoại
Gọi endpoint POST /conversations với user ID dùng cho môi trường demo. Ví dụ, hệ thống trả về:
{
"user_id": "user_001",
"conversation_id": "a1b2c3d4-...",
"ttl_seconds": 3600
}Bước 2: Gửi câu hỏi
Sử dụng conversation ID vừa nhận được để gọi POST /chat:
{
"user_id": "user_001",
"conversation_id": "a1b2c3d4-...",
"message": "Phân tích doanh thu tháng này"
}Ứng dụng sẽ lưu message vào Redis List và cập nhật thời hạn session. Phần trả lời trong code vẫn là câu xác nhận đơn giản; để tạo câu trả lời bằng AI, cần thay logic demo bằng lời gọi LLM và lưu kết quả thực tế của mô hình.
Bước 3: Lấy lịch sử hội thoại
Gọi endpoint GET /conversations/{conversation_id} với user ID tương ứng. API trả về danh sách message cùng TTL còn lại của key.
Nếu bạn khởi động hai instance FastAPI cùng kết nối tới một Redis, cả hai có thể đọc cùng dữ liệu hội thoại. Đây là khác biệt quan trọng so với dictionary lưu trong RAM của từng process.
7. TTL được gia hạn như thế nào?
Có hai chiến lược thường gặp khi quản lý thời hạn session:
| Chiến lược | Cách hoạt động | Ví dụ |
|---|---|---|
| Fixed expiration | Thời hạn được tính từ lúc tạo session và không tự kéo dài sau mỗi request. | Phiên hết hạn sau 1 giờ kể từ lúc tạo. |
| Sliding expiration | Thời hạn được gia hạn khi người dùng tiếp tục hoạt động. | Mỗi request hợp lệ đặt TTL trở lại 1 giờ. |
Code demo sử dụng sliding expiration: sau hoạt động ghi dữ liệu, key được đặt lại TTL một giờ. Vì vậy, nếu người dùng liên tục hoạt động, session có thể tiếp tục tồn tại lâu hơn một giờ tính từ lúc khởi tạo.
Đối với session đăng nhập hoặc dữ liệu nhạy cảm, chỉ dùng sliding expiration có thể không đủ. Hệ thống có thể cần thêm giới hạn thời gian tuyệt đối, yêu cầu xác thực lại và cơ chế thu hồi session.
8. Những vấn đề cần xử lý trước khi triển khai production
8.1. Xác thực và phân quyền
Không được xem user ID hoặc conversation ID do client gửi lên là bằng chứng về quyền truy cập. Server phải xác thực người dùng và kiểm tra cuộc hội thoại thuộc phạm vi dữ liệu mà người dùng đó được phép truy cập.
8.2. Ghi lịch sử an toàn khi có nhiều request đồng thời
Trong ví dụ đơn giản, các lệnh ghi message và cập nhật TTL được thực hiện theo nhiều bước. Nếu nhiều request đồng thời cập nhật cùng một cuộc hội thoại, thứ tự message hoặc giới hạn số message có thể không như mong muốn.
Ứng dụng thực tế có thể cần dùng pipeline, transaction hoặc một chiến lược ghi có tính nguyên tử phù hợp. Nếu Agent chạy nhiều tác vụ đồng thời trong cùng session, nên xác định rõ cách tuần tự hóa hoặc xử lý xung đột.
8.3. Giới hạn kích thước lịch sử
Giới hạn số message giúp kiểm soát dung lượng lưu trữ, nhưng số message không tương đương với số token. Một message rất dài vẫn có thể khiến request LLM vượt giới hạn ngữ cảnh.
Trong ứng dụng thật, nên kết hợp giới hạn độ dài đầu vào, số lượt hội thoại và ngân sách token. Với lịch sử dài, có thể tóm tắt các lượt cũ hoặc truy xuất riêng những thông tin liên quan thay vì đưa toàn bộ transcript vào mô hình.
8.4. Xử lý lỗi và khả năng sẵn sàng
Redis có thể không truy cập được do lỗi mạng, quá tải hoặc bảo trì. Cần thiết kế timeout, logging, cảnh báo và chính sách phản hồi phù hợp. Không nên âm thầm chuyển sang lưu RAM mà không thông báo, vì các instance sẽ tạo ra những bản lịch sử khác nhau.
8.5. Bảo vệ dữ liệu và cấu hình Redis
- Không mở cổng Redis công khai nếu không cần thiết.
- Sử dụng xác thực, phân quyền và mạng riêng theo môi trường triển khai.
- Không ghi nội dung nhạy cảm hoặc thông tin xác thực vào log không cần thiết.
- Thiết lập TTL và chính sách lưu trữ phù hợp với loại dữ liệu.
- Đánh giá persistence, backup và khả năng khôi phục nếu dữ liệu hội thoại cần được giữ lâu dài.
9. Redis Memory có thay thế Long-term Memory không?
Không nhất thiết. Redis có thể được dùng để lưu trạng thái hội thoại đang hoạt động, nhưng long-term memory có thể cần một cách lưu trữ và truy xuất khác.
Ví dụ, một hệ thống có thể sử dụng:
- Redis: Lưu session đang hoạt động và một cửa sổ message gần nhất.
- SQL database: Lưu lịch sử cần truy vấn, báo cáo hoặc kiểm toán.
- Vector database: Truy xuất thông tin liên quan theo ngữ nghĩa khi kiến trúc cần tìm kiếm vector.
Việc chọn công nghệ phụ thuộc vào mục tiêu của ứng dụng. Không phải AI Agent nào cũng cần vector database, và không phải toàn bộ lịch sử hội thoại đều cần được lưu vĩnh viễn.
Kết luận
Redis giúp nâng cấp AI Agent Memory từ bộ nhớ trong RAM thành một kho trạng thái có thể được chia sẻ giữa nhiều instance. Với Redis List, redis-py và FastAPI, bạn có thể xây dựng một API cơ bản để lưu lịch sử hội thoại, truy xuất message và tự động hết hạn session bằng TTL.
Tuy nhiên, Redis chỉ là một phần trong kiến trúc. Một hệ thống thực tế còn cần xác thực, phân quyền, xử lý ghi đồng thời, giới hạn ngữ cảnh, giám sát và chính sách lưu giữ dữ liệu. Khi những thành phần này được thiết kế đúng, AI Agent sẽ dễ mở rộng và vận hành ổn định hơn.
Học thêm về AI và Data tại MCNA Technology School
Nếu bạn muốn xây dựng nền tảng Python, phân tích dữ liệu và ứng dụng AI vào các bài toán thực tế, có thể tham khảo các chương trình học tại MCNA Technology School:
