Redis 高级特性——Stream / PubSub / Module / LLM 应用
📅 创建时间：2026-06-02 🏷️ 标签：#Redis #Stream #PubSub #RediSearch #RedisJSON #RedisAI #LLM #向量搜索 #实时统计 📚 前置知识：[[00-redis-overview]]（Redis 全景） [[01-redis-architecture]]（架构分析） [[02-redis-data-structures]]（数据结构） [[03-redis-cache-patterns]]（缓存三剑客） 📚 相关知识：[[/03-web/06-databases-and-data-access/05-redis]]（Redis 命令速查） [[/03-web/06-databases-and-data-access/08-vector-database]]（向量数据库） [[/03-web/07-middleware/02-mq-kafka]]（消息队列）
场景：Redis 不只是缓存，它是全能数据平台
┌─────────────────────────────────────────────────────────────┐
│ │
│ 很多人以为 Redis 只是 KV 缓存。 │
│ │
│ 但 Redis 6.0+ 早已进化为一个多功能数据平台： │
│ │
│ 你可以用它做消息队列（Stream） │
│ 你可以用它做实时订阅（PubSub） │
│ 你可以用它做搜索引擎（RediSearch） │
│ 你可以用它存 JSON（RedisJSON） │
│ 你可以用它做向量搜索（RedisVSS） │
│ 你可以用它做 AI Agent 的记忆层（Session 缓存/Token 限流）│
│ │
│ 这一章，我们探索 Redis 的高级能力和前沿应用。 │
│ │
└─────────────────────────────────────────────────────────────┘第1节：Stream——Redis 原生消息队列
为什么 Stream：List 做队列的致命缺陷
┌─────────────────────────────────────────────────────────────┐
│ List 队列 vs Stream 的核心差距 │
├─────────────────────────────────────────────────────────────┤
│ │
│ List 队列（LPUSH + BRPOP）的缺陷： │
│ │
│ 1. 无 ACK 确认： │
│ 消费者拿到消息后崩溃 → 消息丢失，无法重投 │
│ │
│ 2. 无消费者组： │
│ 多个消费者无法协作，每条消息只能被一个消费者处理 │
│ （无法实现"消费者组内公平分发"） │
│ │
│ 3. 无消息 ID： │
│ 消息没有唯一标识，无法追踪、处理状态 │
│ │
│ 4. 无法查看待处理消息： │
│ 无法知道"哪些消息还没被处理" │
│ │
│ Stream 解决了以上所有问题： │
│ • ACK 确认 → 处理失败可重投 │
│ • 消费者组 → 多个消费者公平分摊消息 │
│ • 消息 ID → 唯一标识，可追踪 │
│ • XPENDING → 可见待处理消息列表 │
│ │
└─────────────────────────────────────────────────────────────┘Stream 核心概念
┌─────────────────────────────────────────────────────────────┐
│ Stream 的核心概念 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Stream = 一个持久化的消息流，类似于 Append-Only 日志 │
│ │
│ 关键概念： │
│ • Entry（条目）：消息，由 ID + 字段值组成 │
│ • Consumer Group（消费者组）：多个消费者协作处理 │
│ • Consumer（消费者）：消费者组中的一个实例 │
│ • Pending（待确认）：已取出但未 ACK 的消息 │
│ │
│ 消息 ID 格式：{timestamp}-{sequence} │
│ 1700000000000-0 → 时间戳 1700000000000，第 0 条 │
│ * 表示自动生成（推荐） │
│ │
│ 与 Kafka 的概念对应： │
│ • Kafka Topic → Redis Stream │
│ • Kafka Partition → Consumer Group 内的不同 Consumer │
│ • Kafka Consumer Group → Redis Consumer Group │
│ • Kafka offset → Redis Message ID │
│ │
└─────────────────────────────────────────────────────────────┘Stream 实战：订单超时处理
┌─────────────────────────────────────────────────────────────┐
│ 场景：订单超时自动取消 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 业务流程： │
│ 1. 用户下单 → 写入订单表（MySQL） │
│ 2. 同时写入 Redis Stream（带超时时间戳） │
│ 3. 定时任务扫描 Stream，取已超时的订单 │
│ 4. 检查订单状态，如果未支付 → 自动取消 │
│ │
└─────────────────────────────────────────────────────────────┘python
# 1. 下单时写入 Stream
def create_order(order_id, user_id):
# 写 MySQL
db.execute("INSERT INTO orders (id, user_id, status) VALUES (%s, %s, 'pending')",
order_id, user_id)
# 写 Stream（30 分钟后超时）
import time
timeout_ts = int((time.time() + 1800) * 1000) # 30 分钟后
r.xadd("orders:timeout", {"order_id": str(order_id)}, id=f"{timeout_ts}-0")
# 2. 创建消费者组
def init_consumer_group():
try:
r.xgroup_create("orders:timeout", "order-processor", id="0", mkstream=True)
except redis.exceptions.ResponseError:
pass # 组已存在
# 3. 消费者：扫描超时订单
def process_timeout_orders():
while True:
# 阻塞读取，超时订单（ID 小于当前时间）
now_ms = int(time.time() * 1000)
messages = r.xreadgroup(
groupname="order-processor",
consumername="worker-1",
streams={"orders:timeout": ">"}, # 只取新的
count=10,
block=1000 # 1 秒超时
)
for stream, entries in messages or []:
for msg_id, fields in entries:
order_id = fields["order_id"].decode()
# 检查订单状态
order = db.query("SELECT * FROM orders WHERE id = %s", order_id)
if order and order["status"] == "pending":
# 未支付，取消订单
db.execute("UPDATE orders SET status = 'cancelled' WHERE id = %s", order_id)
print(f"订单 {order_id} 超时取消")
# ACK 确认消息
r.xack("orders:timeout", "order-processor", msg_id)
time.sleep(1)XPENDING：查看未确认消息
python
# 查看待确认消息（消费者崩溃后，消息会在这里）
pending = r.xpending("orders:timeout", "order-processor")
# pending = {
# 'pel': [('1700000000000-0', b'worker-1', 60000, 1)],
# 'count': 1,
# 'first': '1700000000000-0',
# 'last': '1700000000000-0'
# }
# → 消息 1700000000000-0 被 worker-1 取出，等待了 60 秒还未 ACK
# 认领超时的消息（转移给其他消费者）
r.xclaim("orders:timeout", "order-processor", "worker-2",
min_idle_time=30000, message_ids=["1700000000000-0"])
# → 30 秒未被 ACK 的消息，转移给 worker-2 重试第2节：PubSub——实时发布订阅
PubSub 原理
┌─────────────────────────────────────────────────────────────┐
│ PubSub 工作原理 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ PUBLISH ┌──────────┐ │
│ │ 发布者 │ ──────────▶ │ Channel │ │
│ └─────────┘ │ "news" │ │
│ └────┬─────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │订阅者1 │ │订阅者2 │ │订阅者3 │ │
│ └────────┘ └────────┘ └────────┘ │
│ │
│ PUBLISH "news" "Redis 7.0 发布！" │
│ → 所有订阅了 "news" 频道的客户端同时收到消息 │
│ │
│ 特点： │
│ • 实时推送（无轮询） │
│ • 消息不持久化（订阅者不在线 → 消息丢失） │
│ • 无 ACK 确认 │
│ │
└─────────────────────────────────────────────────────────────┘典型使用场景
python
# 场景 1：实时通知
# 后台任务完成后，发布通知
r.publish("user:1001:notify", json.dumps({
"type": "order_shipped",
"message": "您的订单已发货"
}))
# 订阅者（WebSocket 服务）
pubsub = r.pubsub()
pubsub.subscribe("user:1001:notify")
for message in pubsub.listen():
if message["type"] == "message":
send_websocket(message["data"])
# 场景 2：跨服务事件通知
# 服务 A 完成了某个操作，通知服务 B
r.publish("system:events", json.dumps({
"event": "user_registered",
"user_id": 1001,
"timestamp": time.time()
}))
# 场景 3：模式匹配订阅
pubsub = r.pubsub()
pubsub.psubscribe("user:*:notify") # 监听所有用户通知
for message in pubsub.listen():
# 收到 user:1001:notify, user:1002:notify 等
pass
# 场景 4：多级缓存通知同步
# 数据更新时，通知所有服务实例清理本地缓存
r.publish("cache:invalidate", json.dumps({
"pattern": "product:*",
"keys": ["product:1001", "product:1002"]
}))PubSub vs Stream 对比
┌─────────────────────────────────────────────────────────────┐
│ PubSub vs Stream 选型指南 │
├─────────────────────────────────────────────────────────────┤
│ │
│ │ 维度 │ PubSub │ Stream │
│ ├───────────────┼────────────────────┼────────────────────┤
│ │ 消息持久化 │ 无 │ 有（Redis 持久化） │
│ ├───────────────┼────────────────────┼────────────────────┤
│ │ 消息确认 │ 无 │ 有（ACK） │
│ ├───────────────┼────────────────────┼────────────────────┤
│ │ 消费者组 │ 无 │ 有 │
│ ├───────────────┼────────────────────┼────────────────────┤
│ │ 离线消息 │ 不保留 │ 保留 │
│ ├───────────────┼────────────────────┼────────────────────┤
│ │ 适用场景 │ 实时通知/事件广播 │ 异步任务/消息队列 │
│ ├───────────────┼────────────────────┼────────────────────┤
│ │ 吞吐量 │ 高 │ 中（受 Redis 限制） │
│ │
│ 选型口诀： │
│ • 需要可靠消息 → Stream │
│ • 只需要实时通知 → PubSub │
│ • 需要消费者组 → Stream │
│ • 100 万+ QPS 消息队列 → Kafka / RabbitMQ │
│ │
└─────────────────────────────────────────────────────────────┘第3节：Redis Module 生态——扩展 Redis 能力
Redis Module 是什么
┌─────────────────────────────────────────────────────────────┐
│ Redis Module 扩展生态 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Redis 从 4.0 开始支持 Module 机制： │
│ → 允许用 C/Rust 等语言编写扩展模块 │
│ → 挂在 Redis 进程中，高性能运行 │
│ → 安装即用，无需改 Redis 源码 │
│ │
│ 官方推荐的 Module： │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ RedisJSON │ JSON 文档存储和查询 │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ RediSearch │ 全文搜索引擎 │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ RedisBloom │ 布隆过滤器（原生支持） │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ RedisGraph │ 图数据库 │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ RedisTimeSeries │ 时序数据库 │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ RedisVSS │ 向量相似度搜索（AI 应用） │ │
│ ├─────────────────────────────────────────────────────┤ │
│ │ RedisAI │ AI 模型 serving（推理） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘RedisJSON：直接操作 JSON 文档
┌─────────────────────────────────────────────────────────────┐
│ RedisJSON：像 MongoDB 一样存文档 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 传统方式（String + JSON）： │
│ GET product:001 # → '{"name":"iPhone","price":6999}' │
│ # 只能整体读写，无法按字段操作 │
│ │
│ RedisJSON 方式： │
│ JSON.SET product:001 $ '{"name":"iPhone","price":6999}' │
│ JSON.GET product:001 $.name # → "iPhone" │
│ JSON.NUMINCRBY product:001 $.price 100 # price +100 │
│ │
│ 支持的操作： │
│ JSON.GET / JSON.SET / JSON.DEL │
│ JSON.NUMINCRBY / JSON.STRAPPEND / JSON.ARRAPPEND │
│ JSON.ARRINSERT / JSON.ARRTRIM / JSON.OBJKEYS │
│ JSON.TYPE / JSON.TOGGLE / JSON.CLEAR │
│ │
│ 适用场景： │
│ • 需要按字段更新的大型 JSON 文档 │
│ • 半结构化数据存储（替代 MongoDB 的轻量选择） │
│ │
└─────────────────────────────────────────────────────────────┘RediSearch：Redis 原生全文搜索引擎
┌─────────────────────────────────────────────────────────────┐
│ RediSearch：Redis 内置的搜索引擎 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 场景：商品搜索 │
│ │
│ 搜索 "小米手机"却找不到"小米 13"： │
│ → LIKE '%小米手机%' 是全表扫描，慢 │
│ → 换 Elasticsearch 太重 │
│ → RediSearch：轻量、集成在 Redis 中 │
│ │
│ FT.CREATE products:idx ON JSON SCHEMA │
│ name TEXT WEIGHT 3 │
│ description TEXT │
│ category TAG │
│ price NUMERIC │
│ │
│ FT.ADD products:idx product:001 1.0 │
│ FIELDS name "小米手机13" description "骁龙8gen2" │
│ │
│ FT.SEARCH products:idx "小米手机" │
│ INFILEDDS name,description │
│ LIMIT 0 20 │
│ │
│ FT.AGG products:idx "@category:{手机} @price:[5000 8000]"│
│ → 按分类和价格范围过滤 │
│ │
│ 特点： │
│ • 倒排索引 + 增量索引（不用全量重建） │
│ • 支持中文分词（需配置 ANALYSIS） │
│ • 聚合查询（GROUP BY / SUM / AVG 等） │
│ • 性能比 Elasticsearch 好（小数据量场景） │
│ │
└─────────────────────────────────────────────────────────────┘RedisVSS：向量相似度搜索（AI 应用核心）
┌─────────────────────────────────────────────────────────────┐
│ RedisVSS：向量相似度搜索 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 场景：AI 应用中的语义搜索 │
│ │
│ Embedding（向量）= 把文本/图片转成一段数字 │
│ "苹果手机" → [0.23, -0.45, 0.89, ...] (1536 维) │
│ │
│ RedisVSS 工作流： │
│ 1. 把文档转成向量 │
│ 2. 存到 RedisVSS（使用向量索引） │
│ 3. 查询时：把查询文本也转成向量 │
│ 4. RedisVSS 返回最相似的 N 个结果 │
│ │
│ 示例： │
│ VSS.ADD article:idx article:001 │
│ EMBEDDINGS {"model": "text-embedding-3", │
│ "vector": [0.23, -0.45, ...]} │
│ │
│ VSS.SEARCH article:idx │
│ EMBEDDINGS {"vector": [0.12, -0.33, ...]} │
│ DISTANCE_TYPE COSINE │
│ TOP_K 5 │
│ │
│ 适用场景： │
│ • RAG（检索增强生成）的向量存储 │
│ • 图片相似度搜索 │
│ • 推荐系统的 item embedding 相似度召回 │
│ • AI Agent 的记忆层（存历史对话的语义向量） │
│ │
│ 替代方案对比： │
│ • Pinecone / Weaviate / Milvus → 专用向量数据库 │
│ • RedisVSS → 已有 Redis 基础设施的场景，性价比赛 │
│ │
└─────────────────────────────────────────────────────────────┘第4节：Redis 在 LLM/Agent 中的应用
AI 时代 Redis 的新角色
┌─────────────────────────────────────────────────────────────┐
│ Redis 在 AI 应用中的五大角色 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 1. LLM 响应缓存（语义缓存） │ │
│ │ │ │
│ │ 场景：相同问题多次被问到，节省 API 费用 │ │
│ │ │ │
│ │ 方案： │ │
│ │ prompt_hash = SHA256(question + model + temp) │ │
│ │ key = f"llm:cache:{prompt_hash}" │ │
│ │ if cached := redis.get(key): return cached │ │
│ │ response := llm.ask(question) │ │
│ │ redis.setex(key, ttl, response) │ │
│ │ │ │
│ │ 效果：相同问题避免重复调用，节省 30-70% API 费用 │ │
│ │ LangChain 有内置的 Redis 缓存实现 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 2. Agent 会话状态存储 │ │
│ │ │ │
│ │ 场景：分布式部署多个 Agent 实例，共用 Session │ │
│ │ │ │
│ │ Hash: session:{id} │ │
│ │ messages: [历史消息 JSON] │ │
│ │ context_tokens: 当前上下文 token 数 │ │
│ │ tools_used: [已使用的工具列表] │ │
│ │ last_active: 最后活跃时间 │ │
│ │ status: running/waiting/completed │ │
│ │ │ │
│ │ 过期策略：按 last_active 排序，超时自动清理 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 3. Token 速率限制（Rate Limiting） │ │
│ │ │ │
│ │ 场景：防止用户过度使用 LLM API，产生巨额账单 │ │
│ │ │ │
│ │ ZSet: ratelimit:{user_id}:{minute} │ │
│ │ ZADD ratelimit:user:1001 {timestamp} {request_id}│ │
│ │ ZREMRANGEBYSCORE ratelimit:user:1001 0 {now-60s} │ │
│ │ count = ZCARD ratelimit:user:1001 │ │
│ │ if count > MAX_TOKENS_PER_MINUTE: reject() │ │
│ │ │ │
│ │ 也可以用 Redis Cell（漏桶算法）： │ │
│ │ CL.THROTTLE user:1001:llm 60 60 60 1 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 4. 流式输出缓冲（Stream） │ │
│ │ │ │
│ │ 场景：LLM 流式输出（chunk by chunk），多端同步 │ │
│ │ │ │
│ │ Stream: llm:stream:{session_id} │ │
│ │ → WebSocket 多端同步（手机 + 电脑看到同一进度） │ │
│ │ → SSE 断线重连续传（不从头开始） │ │
│ │ → 记录每个 chunk 的 ID，支持断点续传 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 5. Agent 任务队列（异步执行） │ │
│ │ │ │
│ │ 场景：长时间 LLM 任务（生成报告/分析），不阻塞用户 │ │
│ │ │ │
│ │ List: agent:tasks:pending │ │
│ │ ZSet: agent:tasks:priority │ │
│ │ score = priority（高优先级先执行） │ │
│ │ │ │
│ │ 前端轮询：GET /tasks/{id}/status │ │
│ │ 完成后：Stream 推送通知前端 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘完整示例：AI Agent 的会话管理
python
class AgentSession:
"""用 Redis 管理 AI Agent 的会话状态"""
def __init__(self, redis_client, session_id):
self.r = redis_client
self.session_id = session_id
self.key = f"agent:session:{session_id}"
def init(self, system_prompt):
"""初始化会话"""
data = {
"messages": json.dumps([{"role": "system", "content": system_prompt}]),
"context_tokens": count_tokens(system_prompt),
"status": "running",
"created_at": time.time()
}
self.r.hset(self.key, mapping={k: str(v) if not isinstance(v, str) else v for k, v in data.items()})
self.r.expire(self.key, 7200) # 2 小时超时
def add_message(self, role, content):
"""添加消息"""
messages = json.loads(self.r.hget(self.key, "messages") or "[]")
tokens = count_tokens(content)
messages.append({"role": role, "content": content})
self.r.hset(self.key, "messages", json.dumps(messages))
self.r.hincrby(self.key, "context_tokens", tokens)
def get_context(self, max_tokens=6000):
"""获取上下文（不超过 max_tokens）"""
messages = json.loads(self.r.hget(self.key, "messages") or "[]")
# 从最新消息往前截断，直到 token 数不超标
# ... token 截断逻辑 ...
return messages
def add_tool_call(self, tool_name, args, result):
"""记录工具调用"""
tool_log_key = f"agent:tools:{self.session_id}"
self.r.lpush(tool_log_key, json.dumps({
"tool": tool_name,
"args": args,
"result": result,
"timestamp": time.time()
}))
self.r.ltrim(tool_log_key, 0, 999) # 只保留最新 1000 条
def complete(self):
"""标记会话完成"""
self.r.hset(self.key, "status", "completed")
self.r.expire(self.key, 300) # 完成后保留 5 分钟供查询第5节：Redis 的边界——什么时候不用 Redis
┌─────────────────────────────────────────────────────────────┐
│ Redis 的边界和不适用的场景 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Redis 不是银弹，以下场景不适合用 Redis： │
│ │
│ ❌ 海量数据存储（>100GB 热数据）： │
│ Redis 内存成本高 → 用 MySQL + 冷热分层 │
│ │
│ ❌ 强一致性要求（如转账/支付）： │
│ Redis 主从异步复制 → 可能丢数据 │
│ → 用数据库事务 / Zookeeper / etcd │
│ │
│ ❌ 文档型数据（复杂嵌套 JSON，频繁部分更新）： │
│ String(JSON) 要反序列化+序列化 │
│ → 用 MongoDB / PostgreSQL JSONB │
│ │
│ ❌ 关系查询（JOIN / 聚合分析）： │
│ Redis 没有 JOIN 能力 │
│ → 用 MySQL / PostgreSQL / ClickHouse │
│ │
│ ❌ 超高吞吐量消息队列（100 万+/s）： │
│ Redis 是单进程，有性能上限 │
│ → 用 Kafka / RabbitMQ │
│ │
│ ❌ 长期归档数据（审计日志 / 操作记录）： │
│ Redis 内存有限，不适合存归档数据 │
│ → 用对象存储（S3）+ OLAP 数据库 │
│ │
│ 选型原则： │
│ Redis + MySQL 配合使用，各取所长。 │
│ Redis 处理热数据和快速操作，MySQL 处理持久化和复杂查询。 │
│ │
└─────────────────────────────────────────────────────────────┘升华：Redis 的演进路径
┌─────────────────────────────────────────────────────────────┐
│ Redis 的演进：从缓存到全能数据平台 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 2009 Redis 1.0：简单 KV 缓存 │
│ 2012 Redis 2.6：Lua 脚本、Pub/Sub │
│ 2015 Redis 3.0：Cluster 分片 │
│ 2017 Redis 4.0：Module 机制、混合持久化 │
│ 2018 Redis 5.0：Stream 消息队列 │
│ 2020 Redis 6.0：多线程 I/O、ACL 权限 │
│ 2022 Redis 7.0：Function（更灵活的脚本）、Shard-level 订阅│
│ 2023+ RedisVSS（向量搜索）、RedisAI（模型 serving） │
│ │
│ 演进主线： │
│ 缓存 → 缓存 + 消息队列 → 缓存 + 消息队列 + 搜索 │
│ → 缓存 + 消息队列 + 搜索 + 向量 + AI 应用支持 │
│ │
│ Redis 的边界在扩展，但核心优势不变： │
│ • 内存级访问速度（纳秒级） │
│ • 丰富的数据结构 │
│ • 简单部署和运维 │
│ │
│ 无论 Redis 功能如何演进， │
│ 理解它的核心优势和使用边界，才能用好它。 │
│ │
└─────────────────────────────────────────────────────────────┘"AI 可查 vs 必须理解"清单
AI 可查：
✅ RediSearch 的具体查询语法和中文分词配置
✅ RedisJSON 的 JSONPath 查询语法
✅ RedisVSS 的向量维度配置和距离算法选择
✅ 各 Module 的具体安装和启动命令
必须理解：
🔴 Stream 和 PubSub 的核心区别（持久化+ACK vs 实时推送）
🔴 Stream 消费者组的概念和 ACK 机制
🔴 Redis Module 生态：RedisJSON / RediSearch / RedisVSS 各自的适用场景
🔴 RedisVSS 在 RAG/AI 应用中的角色（向量存储）
🔴 AI Agent 场景下 Redis 的五大用途（会话/缓存/限流/流式/队列）
🔴 Redis 的边界：不适合强一致性、海量存储、关系查询、超高吞吐消息队列学习状态：🟡 开始学习