第 15 章 记忆管理

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, RemoveMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def chat(state: ChatState) -> dict:
    from langchain_core.messages import AIMessage
    return {"messages": [AIMessage("好的，已记录您的偏好")]}

# 独立的裁剪节点：图结构的一部分，而非节点内的隐藏动作
def trim(state: ChatState) -> dict:
    KEEP = 4                                   # 保留最近 4 条
    if len(state["messages"]) <= KEEP:
        return {}                              # 不够长，无需裁剪
    to_remove = state["messages"][:-KEEP]      # 更早的全部删掉
    return {"messages": [RemoveMessage(id=m.id) for m in to_remove]}

builder = StateGraph(ChatState)
builder.add_node("chat", chat)
builder.add_node("trim", trim)
builder.add_edge(START, "chat")
builder.add_edge("chat", "trim")
builder.add_edge("trim", END)
# 跨轮累积消息必须有 checkpointer（线程记忆的前提，见第 12 章）
graph = builder.compile(checkpointer=InMemorySaver())

from langchain_core.messages import HumanMessage
config = {"configurable": {"thread_id": "mem-001"}}

msgs = [HumanMessage(f"第 {i} 轮问题", id=f"h{i}") for i in range(1, 7)]
for m in msgs:
    graph.invoke({"messages": [m]}, config)

final = graph.get_state(config).values["messages"]
print(len(final), [m.content for m in final])
# 4 ['第 5 轮问题', '好的，已记录您的偏好', '第 6 轮问题', '好的，已记录您的偏好']
# 第 1~4 轮被陆续裁掉：每轮追加"1 问 + 1 答"共 2 条，超过 4 条即裁掉最旧 2 条（恰好一问一答）

import os
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import AnyMessage, RemoveMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()
llm = ChatOpenAI(model=os.getenv("LLM_MODEL", "gpt-4o-mini"), temperature=0)

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    summary: str                        # 滚动摘要：短期记忆的"魂"

def condense(state: ChatState) -> dict:
    KEEP = 4
    if len(state["messages"]) <= KEEP:
        return {}
    old, recent = state["messages"][:-KEEP], state["messages"][-KEEP:]

    # 老消息并入滚动摘要（调一次模型做压缩）
    transcript = state.get("summary", "") + "\n".join(
        f"{'用户' if isinstance(m, HumanMessage) else '助手'}：{m.content}" for m in old
    )
    new_summary = llm.invoke([
        SystemMessage("将以下对话历史压缩为不超过 100 字的要点摘要，保留关键事实与偏好。"),
        HumanMessage(transcript),
    ]).content

    # 删旧 + 存魂
    return {
        "messages": [RemoveMessage(id=m.id) for m in old],
        "summary": new_summary,         # 注意 summary 是覆盖语义字段
    }

def respond(state: ChatState) -> dict:
    # 回复时：摘要作为 system 上下文 + 近期原文
    history = (
        ([SystemMessage(f"此前对话摘要：{state['summary']}")] if state.get("summary") else [])
        + state["messages"]
    )
    reply = llm.invoke(history)
    return {"messages": [reply]}

builder = StateGraph(ChatState)
builder.add_node("condense", condense)
builder.add_node("respond", respond)
builder.add_edge(START, "condense")
builder.add_edge("condense", "respond")
builder.add_edge("respond", END)
graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "mem-002"}}
for i in range(1, 8):
    graph.invoke({"messages": [HumanMessage(f"第 {i} 轮：了解我的技术栈偏好", id=f"h{i}")]}, config)

final = graph.get_state(config).values
print("摘要：", final["summary"])
print("消息数：", len(final["messages"]))

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 写：命名空间 + 键 + 值
store.put(
    ("candidates", "李思远"),                # 命名空间：候选人维度
    "assessment",                            # 键：这类事实的名称
    {"conclusion": "算法偏弱，工程能力扎实",   # 值：任意可序列化 dict
     "history": ["2026-06 投递后端岗", "2026-08 复投"],
     "score": 72},
)

# 读：命名空间 + 键
item = store.get(("candidates", "李思远"), "assessment")
print(item.value["conclusion"])     # 算法偏弱，工程能力扎实

# 列：该命名空间下全部条目
for item in store.search(("candidates", "李思远")):
    print(item.key, item.value)

from typing import TypedDict
from langgraph.config import get_store
from langgraph.graph import StateGraph, START, END
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()   # 全新的空 Store：演示"首次投递 → 建立画像 → 复投读取"

def load_history(state: dict) -> dict:
    store = get_store()
    item = store.get(("candidates", state["name"]), "assessment")
    prior = item.value if item else None
    # 旧结论进状态，供评估节点参考
    return {"prior_assessment": prior["conclusion"] if prior else "首次投递"}

def save_assessment(state: dict) -> dict:
    store = get_store()
    store.put(
        ("candidates", state["name"]),
        "assessment",
        {"conclusion": state["conclusion"], "score": state["score"],
         "updated": "2026-08-28"},
    )
    return {}

# 组装：评估前读画像，收尾时写回
class ScreenState(TypedDict):
    name: str
    resume_text: str
    prior_assessment: str
    conclusion: str
    score: int

builder = StateGraph(ScreenState)
builder.add_node("load_history", load_history)
builder.add_node("evaluate", lambda s: {
    "conclusion": f"综合评估（参考历史：{s['prior_assessment']}）",
    "score": 80,
})
builder.add_node("save_assessment", save_assessment)
builder.add_edge(START, "load_history")
builder.add_edge("load_history", "evaluate")
builder.add_edge("evaluate", "save_assessment")
builder.add_edge("save_assessment", END)
graph = builder.compile(store=store)       # 挂载 Store

# 第一次投递：Store 为空，建立画像
r1 = graph.invoke({"name": "李思远", "resume_text": "..."})
print(r1["prior_assessment"])   # 首次投递

# 第二次投递（完全独立的一次运行！）：读到了 r1 写回的画像
r2 = graph.invoke({"name": "李思远", "resume_text": "..."})
print(r2["prior_assessment"])   # 综合评估（参考历史：首次投递）—— r1 的 save_assessment 写入