为记忆插件开发者提供协议适配器开发指南，教你将自有存储格式与…

# Redis Hash ↔ 标准记忆协议 适配器（LangChain 兼容）

一、设计思路

• 标准协议：以 LangChain 的 BaseChatMemory / BaseMemory 为基类，消息统一转为 HumanMessage / AIMessage。
• Redis 结构：推荐 Hash 存元数据 + List 存有序消息（Hash 适合存 profile/摘要，List 天然保序）。若坚持纯 Hash，字段用 msg:{递增序号}，配合 HKEYS + 排序读取。
• 核心方法：load_memory_variables（读）、save_context（写）、export / import（迁移/备份）。

二、核心代码

```python

import json

import redis

from langchain.memory import BaseChatMemory # 或 BaseMemory

from langchain.schema import HumanMessage, AIMessage

class RedisHashMemory(BaseChatMemory):

def __init__(self, session_id: str, url="redis://localhost:6379/0",

max_messages: int = 50, **kwargs):

super().__init__(**kwargs)

self.r = redis.Redis.from_url(url, decode_responses=True)

self.key = f"memory:{session_id}"

self.max_messages = max_messages

def load_memory_variables(self, inputs=None):

vals = self.r.hvals(self.key)

msgs = sorted(

(json.loads(v) for v in vals),

key=lambda m: m.get("seq", 0)

)[-self.max_messages:]

messages = [

HumanMessage(m["content"]) if m["role"] == "human"

else AIMessage(m["content"])

for m in msgs

]

return {"history": messages}

def save_context(self, inputs, outputs):

seq = self.r.hincrby(f"{self.key}:meta", "seq", 1)

self.r.hset(self.key, f"msg:{seq}", json.dumps({

"seq": seq,

"role": "human",

"content": str(list(inputs.values())[0]),

"ts": seq, # 可换 time.time()