向量记忆与长期记忆

第 12 章的会话存档是"短期记忆"：上下文窗口一满或会话一换就没了。要跨会话记住"用户喜欢喝无糖美式""下周二去杭州出差"这类长期信息，得单独落库并按语义相关性召回。本章实现一套完整的向量记忆：memory 表（key/向量/文本/时间）+ embedding 封装 + 余弦召回，最后演示智能体如何利用记忆作答。

长期记忆要解决什么

上下文窗口 = 短期记忆，只活在当次对话里。长期记忆的思路是：每次对话结束后，把值得记住的信息（偏好、事实、计划）连同它的向量存起来；下次用户提问时，先按"问题的向量"召回最相关的几条记忆，拼进 system 提示词再让模型回答——模型本身不"记得"，但每次都能"查到"。

存储：sqlite 一张 memory 表

一行一条记忆：key 做主键（业务去重用，如 user:01:pet）、text 存原文、vector 存向量、updated_at 存时间：

# agent_demo/memory.py —— 第 13 章
import json
import os
import sqlite3
import time
import zlib

DB_FILE = os.path.join(os.path.dirname(__file__), "memory.db")

def _conn():
    conn = sqlite3.connect(DB_FILE)
    conn.execute("""CREATE TABLE IF NOT EXISTS memory(
        key TEXT PRIMARY KEY,
        text TEXT NOT NULL,
        vector TEXT NOT NULL,          -- 向量以 JSON 数组文本存储
        updated_at TEXT NOT NULL)""")
    return conn

embedding：三路可切换的封装

向量从哪来？三种方式按环境变量 EMBED_MODE 切换，以你本机可用者为准：

def debug_embedding(text: str, dim: int = 256) -> list:
    """离线调试伪向量：字符二元组哈希到固定维度。无语义，仅供验证流程。"""
    v = [0.0] * dim
    for i in range(len(text) - 1):
        v[zlib.crc32(text[i:i + 2].encode("utf-8")) % dim] += 1.0
    norm = (sum(x * x for x in v)) ** 0.5 or 1.0
    return [x / norm for x in v]

def get_embedding(text: str) -> list:
    """EMBED_MODE：api=OpenAI 兼容接口 / ollama=本地 Ollama / debug=离线。"""
    mode = os.environ.get("EMBED_MODE", "api")
    if mode == "debug":
        return debug_embedding(text)
    if mode == "ollama":                     # 本地 Ollama 原生接口
        import requests
        url = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434") + "/api/embeddings"
        resp = requests.post(url, timeout=60, json={
            "model": os.environ.get("EMBED_MODEL", "bge-m3"),   # 或 nomic-embed-text
            "prompt": text})
        resp.raise_for_status()
        return resp.json()["embedding"]
    # 默认 OpenAI 兼容：text-embedding-3-small；若服务商无 embedding
    #（如 deepseek），请设 EMBED_MODE=ollama 或把 EMBED_BASE_URL 指向通义等
    from openai import OpenAI
    client = OpenAI(api_key=os.environ.get("EMBED_API_KEY",
                                           os.environ.get("LLM_API_KEY")),
                    base_url=os.environ.get("EMBED_BASE_URL",
                                            os.environ["LLM_BASE_URL"]))
    model = os.environ.get("EMBED_MODEL", "text-embedding-3-small")
    r = client.embeddings.create(model=model, input=text)
    return r.data[0].embedding

写入与召回（B 路：手写余弦，零向量库依赖）

def remember(key: str, text: str) -> str:
    """写入一条记忆；同 key 再次写入视为更新，天然去重。"""
    ts = time.strftime("%Y-%m-%d %H:%M:%S")
    conn = _conn()
    conn.execute("""INSERT INTO memory(key, text, vector, updated_at)
                    VALUES(?, ?, ?, ?)
                    ON CONFLICT(key) DO UPDATE SET
                      text = excluded.text, vector = excluded.vector,
                      updated_at = excluded.updated_at""",
                 (key, text, json.dumps(get_embedding(text)), ts))
    conn.commit()
    conn.close()
    return f"已记住（{key}）：{text}"

def _cosine(a: list, b: list) -> float:
    if len(a) != len(b):
        return 0.0
    dot = sum(x * y for x, y in zip(a, b))
    na = (sum(x * x for x in a)) ** 0.5 or 1.0
    nb = (sum(y * y for y in b)) ** 0.5 or 1.0
    return dot / (na * nb)

def recall(query: str, top_k: int = 3) -> list:
    """按语义相关性召回 top_k 条记忆（纯 Python 余弦，无向量库）。"""
    qv = get_embedding(query)
    conn = _conn()
    rows = conn.execute("SELECT key, text, vector, updated_at FROM memory").fetchall()
    conn.close()
    scored = [(_cosine(qv, json.loads(vec)), text, ts)
              for _, text, vec, ts in rows]
    scored.sort(key=lambda x: x[0], reverse=True)
    return [{"text": t, "score": round(s, 3), "updated_at": ts}
            for s, t, ts in scored[:top_k]]

装了 numpy 可一行算整批相似度；装了 chromadb 则走 A 路（可选）：col.add(ids=[key], documents=[text], embeddings=[get_embedding(text)])，查询用 col.query(query_embeddings=[get_embedding(q)], n_results=3)，召回逻辑完全一样。

全流程演示：不联网也能跑通

debug 模式用固定伪向量，写入→召回全流程离线可跑（真实环境去掉 EMBED_MODE 那行即可）。api/ollama 模式需先让 .env 生效（import llm_client 或自行 load_dotenv），并确认所用服务/模型真的提供 embedding 接口：

if __name__ == "__main__":
    os.environ["EMBED_MODE"] = "debug"          # 离线演示；线上删掉此行
    print(remember("user:01:hobby", "用户喜欢喝无糖美式咖啡"))
    # 输出：已记住（user:01:hobby）：用户喜欢喝无糖美式咖啡
    remember("user:01:pet", "用户养了一只叫年糕的橘猫")
    remember("user:01:plan", "用户下周二要去杭州出差")
    for hit in recall("我的猫叫什么名字？"):
        print(hit)
    # 输出（debug 向量得分仅供参考）：
    # {'text': '用户养了一只叫年糕的橘猫', 'score': 0.26, 'updated_at': '...'}
    # {'text': '用户喜欢喝无糖美式咖啡', 'score': 0.03, 'updated_at': '...'}

Agent 利用记忆的对话轨迹

真实运行时把召回结果拼进 system，模型作答就像"记得你"一样：

# demo_memory_agent.py —— 需真实 embedding 服务 + .env（见第 2 章）
from llm_client import LLMClient       # 沿用第 2 章客户端
from memory import remember, recall    # 本章 memory.py（与脚本同目录）

llm = LLMClient()

def learn_from(user_text: str) -> None:
    """每轮对话结束后调用：用简单规则抓事实（正式项目可让模型抽取再存）。"""
    for word, key in (("喜欢", "hobby"), ("养了", "pet"), ("下周", "plan")):
        if word in user_text:
            remember(f"user:01:{key}", user_text)

def ask(user_input: str) -> str:
    facts = recall(user_input, top_k=3)          # 1) 按问题召回记忆
    lines = "\n".join(f"- {f['text']}" for f in facts)
    system = "你是贴心助手。以下是与用户相关的长期记忆，可参考：\n" + lines
    return llm.chat_text([{"role": "system", "content": system},
                          {"role": "user", "content": user_input}])  # 2) 带记忆作答

if __name__ == "__main__":
    learn_from("我养了一只叫年糕的橘猫，特别黏人。")   # 3) 事后沉淀
    print(ask("想给猫挑零食，结合它的性格起个名"))
    # 输出（示例轨迹，实际以模型回复为准）：
    # [记忆] 命中：用户养了一只叫年糕的橘猫，特别黏人。
    # 助手：年糕这么黏人，零食可以叫"贴贴糕"——又黏又甜，很配它。

写与读分离后，智能体就拥有了可跨会话累积的长期记忆：对话结束沉淀（remember），对话开始前召回（recall）并拼进 system；同 key 覆盖更新避免越记越杂，检索阈值与 top_k 可按需要调。

小结：向量记忆 = 把"值得记住的话"连同向量存进 sqlite（key/vector/text/时间四要素），提问时用余弦相似度按相关性召回 top_k 拼进 system；embedding 支持 OpenAI 兼容/本地 Ollama 两路真实接口，另留 debug 伪向量路保证无网也能跑通全流程。