W5-D6 · 记忆管理成图：把 W3 的 MemoryManager 翻译成一张状态图

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

                START
                  │
                  ▼
            route_node            ← 原来 MemoryManager 里的 if/else 决策
                  │
        ┌─────────┼──────────┐
        ▼         ▼          ▼
   long_node  compress_node  short_node   ← 三条记忆路径 = 原来三个组件
        │         │          │
        │     summary 写入    │
        │     recent 窗口更新  │
        └────┬────┴────┬─────┘
             ▼         ▼
          agent_node（ReAct）   ← 原来 ChatSession 的对话循环
             │
       tools_condition
        ╱          ╲
  有 tool_calls    无 → END
      ▼
 tools_node（save / recall 长期记忆）
      │
      └──→ 回 agent（循环）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class MemoryManager:
    def __init__(self, window=8, max_tokens=2000, keep=4, llm=None):
        self.stm = ShortTermMemory(window=window, maxtokens=max_tokens)  # 短期
        self.ltm = LongTermMemory()        # 长期（Chroma 向量库）
        self.keep, self.max_tokens, self.llm = keep, max_tokens, llm

    def add(self, msg: dict, persist: bool = False):
        self.stm.add(msg)                                   # 永远进短期
        if persist and msg.get("role") == "user":           # 只有明确要记的才落长期
            self.ltm.add_fact(msg["content"], fid=_fid(msg["content"]))

    def get_context(self, query: str = None) -> list[dict]:
        base = self.stm.context()                           # 取滑动窗口
        if not query:
            return base
        facts = self.ltm.recall(query, k=3)                 # 有 query 才召回长期
        if not facts:
            return base
        facts_msg = {"role": "system", "content": "已知用户长期事实：" + "|".join(facts)}
        # 拼成：系统提示 + 长期事实 + 窗口消息
        return sys_msgs + [facts_msg] + others

    def maybe_compress(self, llm=None):                     # 超阈值自动压缩
        msgs = _auto_compress(self.stm.buffer, llm, max_tokens=self.max_tokens, keep=self.keep)
        if msgs is not self.stm.buffer:                     # 真的压了才写回 buffer
            self.stm.buffer = msgs
        return self.stm.buffer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from typing import Annotated, Any
from pydantic import BaseModel, Field
from langgraph.graph.message import add_messages

class MemoryState(BaseModel):
    messages: Annotated[list[Any], add_messages] = Field(default_factory=list)  # 完整历史
    recent: list[Any] = Field(default_factory=list)    # 压缩产物：当前窗口（无 reducer）
    summary: str = Field(default="", description="压缩后保留的历史摘要")
    retrieved: list[str] = Field(default_factory=list, description="长期记忆召回内容")
    memory_path: str = Field(default="", description="本次走的路径: short/long/compress")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

memory_store = {}    # {关键词: 内容} —— 玩具级记忆库，正式版换回 W3 的 Chroma

@tool
def save_long_term(key: str, content: str) -> str:
    """把重要信息存入长期记忆。当用户说'记住...'或提到个人偏好时调用。"""
    memory_store[key] = content
    return f"已记住: {key} = {content}"

@tool
def recall_long_term(query: str) -> str:
    """从长期记忆检索信息。当用户询问之前提过的事实/偏好时调用。
    query 请传 1~3 个关键词（如"咖啡、预算"），不要传完整问句。"""
    kws = [w for w in query.replace(",", " ").replace("，", " ").split() if w]
    hits = []
    for k, v in memory_store.items():
        if any(kw in k or kw in v for kw in kws) or query in k or query in v:
            hits.append(f"{k}: {v}")
    return "\n".join(hits) if hits else "(长期记忆中没有相关内容)"

1
2
3
4
5
6

llm = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com",
    api_key=os.getenv("DEEPSEEK_API_KEY") or os.getenv("OPENAI_API_KEY"),
)
llm_with_tools = llm.bind_tools([save_long_term, recall_long_term])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def route_node(state: MemoryState) -> dict:
    # 规则优先：消息超阈值直接压缩（确定性，不赌模型心情；也是 W3 maybe_compress 的阈值语义）
    if len(state.messages) > 10:
        return {"memory_path": "compress"}

    # 其余交给 LLM 语义判断：要长期记忆，还是普通对话？
    prompt = f"""判断这段对话最需要哪种记忆处理，只输出一个词：
- long: 用户明确要求记住信息(说'记住...')，或询问之前提过的事实/偏好
- short: 普通对话，无需特殊处理
对话历史:
{fmt(state.messages[-4:])}"""
    decision = llm.invoke(prompt).content.strip().lower()
    if "long" in decision:
        return {"memory_path": "long"}
    return {"memory_path": "short"}

1
2
3
4
5

def fmt(msgs) -> str:
    return "\n".join(
        m.content if hasattr(m, "content") and m.content else str(m)
        for m in msgs
    )

1
2
3

def short_node(state: MemoryState) -> dict:
    """普通对话：消息已在 messages 自然累加，recent 同步为当前窗口"""
    return {"recent": state.messages[-6:], "memory_path": "short"}

1
2
3

def long_node(state: MemoryState) -> dict:
    """长期记忆路径：真正的'存什么/取什么'由 agent 调工具完成，这里只标记路径"""
    return {"memory_path": "long"}

1
2
3
4
5
6
7
8
9

def compress_node(state: MemoryState) -> dict:
    """压缩：旧消息 → LLM 摘要；裁剪结果写进 recent（普通字段），不碰 messages"""
    if len(state.messages) <= 4:
        return {"recent": state.messages, "summary": state.summary, "memory_path": "compress"}
    old, recent = state.messages[:-4], state.messages[-4:]
    sp = (f"把以下对话压缩成一句摘要，保留关键事实(人名/偏好/结论):\n{fmt(old)}\n"
          f"(已有摘要:{state.summary or '无'})")
    new_summary = llm.invoke(sp).content.strip()
    return {"recent": recent, "summary": new_summary, "memory_path": "compress"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def agent_node(state: MemoryState) -> dict:
    # 读窗口：压缩过就用 recent（裁剪后的 4 条），没压缩过退回全量 messages
    window = state.recent if state.recent else state.messages

    ctx = []
    if state.summary:                      # 压缩摘要 → system
        ctx.append(("system", f"历史摘要:{state.summary}"))
    if state.retrieved:                    # 长期记忆直连召回结果 → system（本版预留）
        ctx.append(("system", f"长期记忆:\n{'，'.join(state.retrieved)}"))
    ctx.extend(window)

    # 兜底闸：工具已经调过 3 次还没收敛 → 强制作答，别再烧 API
    tool_turns = [m for m in state.messages if getattr(m, "type", "") == "tool"]
    if len(tool_turns) >= 3:
        ctx.append(("user", "请基于已有工具结果直接给出最终答案；若没有查到，就如实说没查到，不要再调用工具。"))

    return {"messages": [llm_with_tools.invoke(ctx)]}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

builder = StateGraph(MemoryState)
builder.add_node("route", route_node)
builder.add_node("short", short_node)
builder.add_node("long", long_node)
builder.add_node("compress", compress_node)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode([save_long_term, recall_long_term]))

builder.add_edge(START, "route")
builder.add_conditional_edges(
    "route",
    lambda s: s.memory_path,                          # 路由函数：读 memory_path 当路标
    {"short": "short", "long": "long", "compress": "compress"},
)
for n in ["short", "long", "compress"]:
    builder.add_edge(n, "agent")                      # 三路径都汇入 agent
builder.add_conditional_edges("agent", tools_condition)   # 有 tool_calls → tools，否则 END
builder.add_edge("tools", "agent")                    # 工具执行完回 agent

memory_graph = builder.compile()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

"""W5-D6 项目日：MemoryManager(类) → MemoryManagerGraph(图)
目标：先跑通。记忆库用进程内 dict（玩具级），正式版换 W3 的 Chroma。"""
import os
from typing import Annotated, Any
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# ══════════ 1. State：档案(messages) 与 窗口(recent) 分开 ══════════
class MemoryState(BaseModel):
    messages: Annotated[list[Any], add_messages] = Field(default_factory=list)  # 完整历史，只追加
    recent: list[Any] = Field(default_factory=list)      # 当前窗口（无 reducer → 可整体覆盖）
    summary: str = Field(default="", description="压缩后保留的历史摘要")
    retrieved: list[str] = Field(default_factory=list, description="长期记忆召回内容(预留)")
    memory_path: str = Field(default="", description="本次走的路径: short/long/compress")

# ══════════ 2. 长期记忆：dict 顶班 + 两个工具 ══════════
memory_store = {}    # {关键词: 内容}；正式版换回 W3 的 Chroma 向量检索

@tool
def save_long_term(key: str, content: str) -> str:
    """把重要信息存入长期记忆。当用户说'记住...'或提到个人偏好时调用。"""
    memory_store[key] = content
    return f"已记住: {key} = {content}"

@tool
def recall_long_term(query: str) -> str:
    """从长期记忆检索信息。当用户询问之前提过的事实/偏好时调用。
    query 请传 1~3 个关键词（如'咖啡、预算'），不要传完整问句。"""
    kws = [w for w in query.replace(",", " ").replace("，", " ").split() if w]
    hits = [f"{k}: {v}" for k, v in memory_store.items()
            if any(kw in k or kw in v for kw in kws) or query in k or query in v]
    return "\n".join(hits) if hits else "(长期记忆中没有相关内容)"

llm = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com",
    api_key=os.getenv("DEEPSEEK_API_KEY") or os.getenv("OPENAI_API_KEY"),
)
llm_with_tools = llm.bind_tools([save_long_term, recall_long_term])   # 只有 agent 用它

# ══════════ 3. 工具函数 ══════════
def fmt(msgs) -> str:
    """消息对象摊成纯文本，给 LLM 看"""
    return "\n".join(
        m.content if hasattr(m, "content") and m.content else str(m)
        for m in msgs
    )

# ══════════ 4. route：规则管压缩，LLM 管 long/short ══════════
def route_node(state: MemoryState) -> dict:
    if len(state.messages) > 10:                 # 规则优先：超阈值必压缩（确定性）
        return {"memory_path": "compress"}
    prompt = f"""判断这段对话最需要哪种记忆处理，只输出一个词：
- long: 用户明确要求记住信息(说'记住...')，或询问之前提过的事实/偏好
- short: 普通对话，无需特殊处理
对话历史:
{fmt(state.messages[-4:])}"""
    decision = llm.invoke(prompt).content.strip().lower()
    if "long" in decision:
        return {"memory_path": "long"}
    return {"memory_path": "short"}

# ══════════ 5. 三条记忆路径 ══════════
def short_node(state: MemoryState) -> dict:
    """普通对话：recent 同步为当前窗口"""
    return {"recent": state.messages[-6:], "memory_path": "short"}

def long_node(state: MemoryState) -> dict:
    """长期路径：只打标；真正的存/取由 agent 调工具完成"""
    return {"memory_path": "long"}

def compress_node(state: MemoryState) -> dict:
    """压缩：旧消息 → LLM 摘要；裁剪写 recent，绝不返回 messages（reducer 会追加）"""
    if len(state.messages) <= 4:
        return {"recent": state.messages, "summary": state.summary, "memory_path": "compress"}
    old, recent = state.messages[:-4], state.messages[-4:]
    sp = (f"把以下对话压缩成一句摘要，保留关键事实(人名/偏好/结论):\n{fmt(old)}\n"
          f"(已有摘要:{state.summary or '无'})")
    new_summary = llm.invoke(sp).content.strip()
    return {"recent": recent, "summary": new_summary, "memory_path": "compress"}

# ══════════ 6. agent：窗口 + 摘要 + 工具轮数闸 ══════════
def agent_node(state: MemoryState) -> dict:
    window = state.recent if state.recent else state.messages     # 只读窗口
    ctx = []
    if state.summary:
        ctx.append(("system", f"历史摘要:{state.summary}"))
    if state.retrieved:
        ctx.append(("system", f"长期记忆:\n{'，'.join(state.retrieved)}"))
    ctx.extend(window)
    tool_turns = [m for m in state.messages if getattr(m, "type", "") == "tool"]
    if len(tool_turns) >= 3:                          # 工具调过 3 次还不收敛 → 强制作答
        ctx.append(("user", "请基于已有工具结果直接给出最终答案；若没有查到，就如实说没查到，不要再调用工具。"))
    return {"messages": [llm_with_tools.invoke(ctx)]}

# ══════════ 7. 构图 ══════════
builder = StateGraph(MemoryState)
builder.add_node("route", route_node)
builder.add_node("short", short_node)
builder.add_node("long", long_node)
builder.add_node("compress", compress_node)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode([save_long_term, recall_long_term]))

builder.add_edge(START, "route")
builder.add_conditional_edges("route", lambda s: s.memory_path,
                              {"short": "short", "long": "long", "compress": "compress"})
for n in ["short", "long", "compress"]:
    builder.add_edge(n, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")

memory_graph = builder.compile()
print("✅ 图编译成功")

# ══════════ 8. 四组测试 ══════════
if __name__ == "__main__":
    # 测试 A：compress —— 12 条超阈值，规则锁死压缩路径
    r1 = memory_graph.invoke({"messages": [("user", f"第{i}句闲聊内容{i}") for i in range(12)]})
    print("\n[A compress] 路径:", r1["memory_path"])
    print("  压缩后 recent 消息数:", len(r1["recent"]), "| 摘要:", (r1.get("summary") or "")[:40])

    # 测试 B：long 记住（落库）
    r2 = memory_graph.invoke({"messages": [("user", "记住我喜欢喝美式咖啡")]})
    print("\n[B 记住] 路径:", r2["memory_path"], "| 记忆库:", memory_store)

    # 测试 C：long 召回（跨轮验证——同进程内 dict 共享）
    r3 = memory_graph.invoke({"messages": [("user", "我喜欢喝什么咖啡？")]})
    print("[C 召回] 路径:", r3["memory_path"])
    for m in r3["messages"][-2:]:
        m.pretty_print()

    # 测试 D：short 普通对话
    r4 = memory_graph.invoke({"messages": [("user", "你好，简单介绍下你自己")]})
    print("\n[D short] 路径:", r4["memory_path"], "| 回复:", r4["messages"][-1].content[:30])

1
2
3

print("r1 keys:", r1.keys())                      # 让代码自己说话：到底有哪些键
print("路径:", r1["memory_path"])                 # 大概率是 short → 证实 compress 没走
print("摘要:", r1.get("summary", "")[:40])        # 键不存在返回 ""，不报错

1

"我喜欢喝什么咖啡" in "用户喜欢喝美式咖啡"  → False

1
2
3
4
5

START → route
route ──(memory_path)──→ short | long | compress   条件边，读 state 字段当路标
short / long / compress → agent                     三路径汇合
agent ──(tools_condition)──→ tools | END           有 tool_calls 就调，否则结束
tools → agent                                      回边（ReAct 循环）