Redis 作为 Agent 基础设施

# FLAT：暴力精确搜索，适合向量数 < 5 万的小集合
FT.CREATE idx:docs ON HASH PREFIX 1 doc: SCHEMA
  content TEXT
  embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE

# HNSW：近似图索引，亿级向量仍可低延迟搜索
# M=16: 每节点连接数；EF_CONSTRUCTION=200: 构建时搜索宽度
FT.CREATE idx:docs_hnsw ON HASH PREFIX 1 doc: SCHEMA
  content TEXT
  embedding VECTOR HNSW 10 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE M 16 EF_CONSTRUCTION 200

# 向量相似度搜索：找最近 5 个，同时按 tag 预过滤
FT.SEARCH idx:docs_hnsw
  "(@tag:{news})=>[KNN 5 @embedding $vec AS score]"
  PARAMS 2 vec <binary_blob>
  SORTBY score
  DIALECT 2

# 插入向量（默认 int8 量化；NOQUANT=全精度 float32；BIN=二值化）
VADD myvecs VALUES 1536 0.12 0.34 0.56 ... doc:42
VADD myvecs NOQUANT VALUES 1536 0.12 0.34 0.56 ... doc:43

# 相似度搜索：返回最近 5 个成员及距离
VSIM myvecs VALUES 1536 0.11 0.33 0.55 ... COUNT 5 WITHSCORES

# 查看集合大小
VCARD myvecs

from redisvl.extensions.llmcache import SemanticCache
from redisvl.utils.vectorize import OpenAITextVectorizer

vectorizer = OpenAITextVectorizer(model="text-embedding-3-small")

cache = SemanticCache(
    name="llm_cache",
    vectorizer=vectorizer,
    redis_url="redis://localhost:6379",
    distance_threshold=0.10,   # COSINE 距离，0.0–2.0，默认 0.1
)

# 查缓存（返回 None 时需调 LLM）
result = cache.check(prompt="Redis 怎么做限流")
if result:
    answer = result[0]["response"]
else:
    answer = call_llm("Redis 怎么做限流")
    cache.store(
        prompt="Redis 怎么做限流",
        response=answer,
    )

# 运行时调整阈值（无需重建索引）
cache.set_threshold(0.15)

from langchain_redis import RedisChatMessageHistory

history = RedisChatMessageHistory(
    session_id="user:42:session:001",
    redis_url="redis://localhost:6379",
    ttl=3600,   # 会话 1 小时无活动后自动过期
)

history.add_user_message("Redis 的 HNSW 是什么")
history.add_ai_message("HNSW 是分层可导航小世界图...")

# 取最近 20 条消息传给 LLM
messages = history.messages[-20:]

from langgraph.graph import StateGraph
from langgraph_checkpoint_redis import RedisSaver

# 创建 checkpointer（同步版；异步版用 AsyncRedisSaver）
checkpointer = RedisSaver.from_conn_string("redis://localhost:6379")
checkpointer.setup()   # 初始化 Redis 索引结构

# 编译 graph 时注入 checkpointer
graph = StateGraph(AgentState)
# ... add_node / add_edge ...
app = graph.compile(checkpointer=checkpointer)

# 执行时带 thread_id：同一 thread 自动从上次检查点恢复
config = {"configurable": {"thread_id": "session:42"}}
result = app.invoke({"messages": [("user", "继续上次的任务")]}, config)

-- KEYS[1]: bucket key, e.g. "ratelimit:tenant:42"
-- ARGV[1]: capacity  (max tokens, e.g. 1000)
-- ARGV[2]: refill_rate  (tokens/second)
-- ARGV[3]: requested  (tokens this request needs)
-- ARGV[4]: now  (current unix timestamp in ms, from client)

local key      = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate     = tonumber(ARGV[2])
local req      = tonumber(ARGV[3])
local now      = tonumber(ARGV[4])

local data = redis.call("HMGET", key, "tokens", "last_refill")
local tokens     = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now

-- 按时间差补充令牌，不超过上限
local elapsed = math.max(0, now - last_refill) / 1000.0
tokens = math.min(capacity, tokens + elapsed * rate)

if tokens >= req then
    tokens = tokens - req
    redis.call("HMSET", key, "tokens", tokens, "last_refill", now)
    redis.call("PEXPIRE", key, 60000)   -- 60s 无请求自动清理
    return 1   -- 允许
else
    redis.call("HMSET", key, "tokens", tokens, "last_refill", now)
    return 0   -- 拒绝
end

# 创建 stream + 消费组（从头读）
XGROUP CREATE agent:tasks grp1 $ MKSTREAM

# 生产者推任务
XADD agent:tasks "*" task_type "summarize" payload "..." agent_hint "worker-1"

# 消费者拉取（最多 10 条，阻塞 2000ms）
XREADGROUP GROUP grp1 worker-1 COUNT 10 BLOCK 2000 STREAMS agent:tasks ">"

# 处理完后确认（消息从 PEL 移除）
XACK agent:tasks grp1 1718000000000-0

# 查看 PEL（超时未 ack 的消息，可用于故障恢复）
XPENDING agent:tasks grp1 - + 10