大模型本身是"无状态"的——每次调用它都不记得上一次说了什么。但一个好用的 Agent 需要三种"记性"：

• 记住这次对话：多轮追问要接得上（这是"工作记忆"）；
• 记住历史经验：这个故障以前遇到过吗、当时怎么解决的（"情景记忆"）；
• 记住团队知识：相关的 SOP、运维手册、架构拓扑（"语义记忆"）。

本框架把这些抽象成四层 Memory Protocol，代码在 packages/ai-trust-toolkit/src/ai_trust_toolkit/memory/。每一层是一个文件，职责清晰。

情景/语义记忆的核心是"按语义相似度找"，而不是"按关键词精确匹配"。比如查"服务 A 变慢"，它应该能召回"service-A 响应延迟升高"这种意思相近但用词不同的历史 case。这靠向量检索（vector search）实现：

所以一次记忆召回 = "把查询文本 embed 成向量 → 在向量库里搜最近的几条 → 返回对应的原文"。

向量库里存的每一条叫 Record（vector_store.py:17），是最基础的数据结构：

@dataclass
class Record:                                 # vector_store.py:17
    id: str                                   # 唯一主键（如 case_id "abc-2025-04-15"）
    text: str                                 # 原文（召回后要塞进 prompt 的那段）
    embedding: list[float]                    # 文本转成的向量（一串浮点数）
    metadata: dict[str, Any] = field(default_factory=dict)
    # 常用 metadata：type(case/sop)、service、alert_type、created_at

问题来了：本地开发想用"内存里的假向量库 + 假 embedder"（快、免费、可复现），生产想用"PostgreSQL+pgvector + Ollama 真 embedder"。怎么做到切换存储时业务代码一行都不用改？

答案是 Python 的 Protocol（协议/接口）——只约定"必须有哪些方法"，不关心具体实现。业务只依赖协议，跑时注入哪个实现都行。真身在 vector_store.py:28：

@runtime_checkable                            # vector_store.py:28
class VectorStore(Protocol):
    async def search(self, query_embedding, top_k=3, filter=None,
                     min_score=-1.0) -> list[tuple[Record, float]]:
        """返回 [(record, similarity), ...] 按相似度降序。"""
        ...
    async def add(self, records: list[Record]) -> None: ...
    async def delete(self, ids: list[str]) -> int: ...
    async def count(self, filter=None) -> int: ...

两种实现都满足同一个 VectorStore / Embedder 协议，所以业务节点拿到的接口一模一样。PgVectorStore 还做了防注入——表名走白名单校验（vector_store.py:149）。

测试/本地用的实现 InMemoryVectorStore（vector_store.py:58）——用一个 dict 当数据库，纯 Python、0 外部依赖。最能说明"向量搜索到底在算什么"的是它的 search（vector_store.py:87）：

class InMemoryVectorStore:                     # vector_store.py:58
    def __init__(self):
        self._records: dict[str, Record] = {}  # 就是个字典当库

    async def search(self, query_embedding, top_k=3, filter=None, min_score=-1.0):
        results = []
        for r in self._records.values():
            if filter and not _match_filter(r, filter):   # 先按 metadata 过滤
                continue
            score = _cosine_similarity(query_embedding, r.embedding)  # 算余弦相似度
            if score < min_score:              # 低于阈值丢掉
                continue
            results.append((r, score))
        results.sort(key=lambda x: x[1], reverse=True)     # 按相似度降序
        return results[:top_k]                 # 取前 K 条

余弦相似度的真实实现（vector_store.py:367）也很朴素——点积除以两个模长：

def _cosine_similarity(a, b):                  # vector_store.py:367
    if not a or not b or len(a) != len(b):
        return 0.0                             # ← 边界：空/长度不齐直接判 0
    dot = sum(x * y for x, y in zip(a, b, strict=True))
    na = sum(x * x for x in a) ** 0.5
    nb = sum(y * y for y in b) ** 0.5
    if not na or not nb:
        return 0.0                             # ← 边界：零向量除零保护
    return dot / (na * nb)

生产用 PgVectorStore（vector_store.py:112）——PostgreSQL 16+ 加 pgvector 扩展。它和内存实现满足同一个协议、方法签名一模一样，但内部是真数据库。先看构造函数里的两处安全校验（vector_store.py:144-150）：

def __init__(self, dsn, *, table="vector_records", dim=768,
             index_type="ivfflat", ...):       # vector_store.py:133
    if dim <= 0:
        raise ValueError(f"dim 必须 > 0 · got {dim}")
    if index_type not in ("ivfflat", "hnsw"):
        raise ValueError(f"index_type 必须 ivfflat 或 hnsw · got {index_type!r}")
    # 防 SQL 注入 · table 名严格限白名单字符
    if not table.replace("_", "").isalnum():   # vector_store.py:149
        raise ValueError(f"table 名只接受 [A-Za-z0-9_] · got {table!r}")

它的 search（vector_store.py:287）把相似度计算下推给数据库，用 pgvector 的 <=> 距离算子 + 索引，不再全扫：

async def search(self, query_embedding, top_k=3, filter=None, min_score=-1.0):
    if len(query_embedding) != self.dim:       # ← 边界：维度不符直接报错（不像内存版返回0）
        raise ValueError("query_embedding dim mismatch ...")
    if filter:
        sql = ("SELECT id, text, embedding, metadata, "
               "1 - (embedding <=> %s::vector) AS sim "   # 1 - 余弦距离 = 相似度
               "FROM {table} WHERE metadata @> %s::jsonb "  # @> = jsonb 包含，走 GIN 索引
               "ORDER BY embedding <=> %s::vector LIMIT %s")
        args = (query_embedding, json.dumps(filter), query_embedding, top_k)
    ...

这两层都是薄薄一层封装——持有一个 VectorStore + 一个 Embedder，把"embed 再搜/存"的样板包好。

Episodic 情景记忆（episodic.py）

EpisodicStore（episodic.py:16）是个 dataclass，recall（episodic.py:39）读、writeback（episodic.py:58）写：

@dataclass
class EpisodicStore:                           # episodic.py:16
    vector_store: VectorStore                  # 依赖协议，不认具体实现
    embedder: Embedder
    record_type: str = "case"                  # 本层固定 type=case

    async def recall(self, query, top_k=3, extra_filter=None, min_score=-1.0):
        [embedding] = await self.embedder.embed([query])   # ① 查询转向量
        filter = {"type": self.record_type}    # ② 只搜 type=case！
        if extra_filter: filter.update(extra_filter)
        return await self.vector_store.search(query_embedding=embedding,
                                              top_k=top_k, filter=filter, min_score=min_score)

    async def writeback(self, case_id, text, metadata=None):
        [embedding] = await self.embedder.embed([text])    # 原文转向量
        meta = {"type": self.record_type}      # 打上 type=case 标记
        if metadata: meta.update(metadata)
        await self.vector_store.add([Record(id=case_id, text=text,
                                            embedding=embedding, metadata=meta)])

Semantic 语义记忆（semantic.py）

SemanticStore（semantic.py:16）和情景记忆共用同一个向量库，seed（semantic.py:40）离线灌库、lookup（semantic.py:60）在线检索：

    async def lookup(self, query, record_type=None, top_k=3,
                     extra_filter=None, min_score=-1.0):    # semantic.py:60
        [embedding] = await self.embedder.embed([query])
        filter = {}
        if record_type: filter["type"] = record_type       # 传 "sop" 就只搜 SOP
        if extra_filter: filter.update(extra_filter)
        return await self.vector_store.search(query_embedding=embedding, top_k=top_k,
                                              filter=filter or None, min_score=min_score)

第①层工作记忆就是 Day 03 讲的 checkpointer（存档器），管"会话内 state + 断点续跑"。框架把后端选择做成工厂 make_checkpointer_from_env()（checkpointer.py:24），按环境变量自动挑：

def make_checkpointer_from_env():              # checkpointer.py:24
    redis_url = os.environ.get("REDIS_URL")
    if redis_url:
        log.info("...REDIS_URL 设但需 lifespan 集成 · 降级 memory...")  # 见下方诚实注释
    pg_dsn = os.environ.get("CHECKPOINTER_PG_DSN")
    if pg_dsn:
        log.info("checkpointer · backend=postgres")
        try:
            saver = make_checkpointer("postgres", url=pg_dsn)
            ...
            return saver
        except ImportError as e:
            log.warning("postgres backend 不可用(%s)· 降级 memory ...", e)
    return make_checkpointer("memory")         # 兜底：内存版

底层 make_checkpointer(backend)（checkpointer.py:70）是真正 new 出实例的地方，三种后端惰性 import：

def make_checkpointer(backend="memory", *, url=None, **kwargs):  # checkpointer.py:70
    if backend == "memory":
        from langgraph.checkpoint.memory import InMemorySaver
        return InMemorySaver(**kwargs)
    if backend == "redis":
        if not url: raise ValueError("redis backend 需要 url='redis://...'")
        try:
            from langgraph.checkpoint.redis import RedisSaver
        except ImportError as e:
            raise ImportError("redis backend 需要 `pip install langgraph-checkpoint-redis`") from e
        return RedisSaver.from_conn_string(url, **kwargs)
    ...  # postgres 同理
    raise ValueError(f"未知 backend: {backend!r}（支持 memory/redis/postgres）")

又是"依赖倒置"套路——同一份业务代码，本地用内存、生产用 Redis/PG，靠环境变量切换，业务 builder.compile(checkpointer=...) 一行不改。

回到 Day 05 的 sre-rca，四层记忆在它的图里各就各位：

# builder.py 里的组装（本地默认全走内存 mock，测试友好）
store = EpisodicStore(vector_store=InMemoryVectorStore(),
                      embedder=MockEmbedder())
checkpointer = make_checkpointer_from_env()
graph = builder.compile(checkpointer=checkpointer)

✋ 动手

# 1. 看 memory 模块清单
sed -n '1,39p' packages/ai-trust-toolkit/src/ai_trust_toolkit/memory/__init__.py

# 2. 读向量库 Protocol + 两种实现
sed -n '17,120p' packages/ai-trust-toolkit/src/ai_trust_toolkit/memory/vector_store.py

# 3. 读情景/语义记忆
cat packages/ai-trust-toolkit/src/ai_trust_toolkit/memory/episodic.py
cat packages/ai-trust-toolkit/src/ai_trust_toolkit/memory/semantic.py

# 4. 读 checkpointer 工厂（含诚实降级注释）
sed -n '24,105p' packages/ai-trust-toolkit/src/ai_trust_toolkit/memory/checkpointer.py

# 5. 跑记忆单测
uv run pytest packages/ai-trust-toolkit/tests/test_memory.py -q