LangChain：上下文与记忆

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(model=model, checkpointer=checkpointer)
config = {"configurable": {"thread_id": "study-001"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "我正在学习 LangChain 的记忆模块。"}]},
    config=config,
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我刚才说在学什么？"}]},
    config=config,
)
print(result["messages"][-1].content)

new_config = {"configurable": {"thread_id": "study-002"}}
snapshot = agent.get_state(config)
print(len(snapshot.values.get("messages", [])))

import os
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver

# 预先加载环境变量；DATABASE_URL 指向已准备好的测试数据库。
with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as saver:
    saver.setup()  # 初始化/迁移检查点表；实际部署中统一管理迁移。
    persistent_agent = create_agent(model=model, checkpointer=saver)
    result = persistent_agent.invoke(
        {"messages": [{"role": "user", "content": "记住这次会话的学习目标：理解检查点。"}]},
        config={"configurable": {"thread_id": "postgres-study-001"}},
    )

from langchain.agents.middleware import wrap_model_call
from langchain_core.messages import trim_messages
from langchain_core.messages.utils import count_tokens_approximately

@wrap_model_call
def trim_model_input(request, handler):
    visible = trim_messages(
        request.messages,
        max_tokens=2000,
        token_counter=count_tokens_approximately,
        strategy="last",
        start_on="human",
        end_on=("human", "tool"),
        include_system=True,
        allow_partial=False,
    )
    return handler(request.override(messages=visible))

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES

# 示例数据：展示“整体替换”的更新形式，不在这里执行状态修改。
kept_messages = []
replacement_update = {
    "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *kept_messages]
}

from typing_extensions import NotRequired
from langchain.agents import AgentState, create_agent

class StudyState(AgentState):
    current_topic: NotRequired[str]

study_agent = create_agent(model=model, state_schema=StudyState)

Store
  namespace = ("users", "demo-user", "preferences")
    key = "answer_style"
      value = {"language": "zh", "style": "concise"}

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
namespace = ("users", "demo-user", "preferences")
store.put(namespace, "answer_style", {"language": "zh", "style": "concise"})

item = store.get(namespace, "answer_style")
print(item.value if item else None)

# 相同 namespace + key 的 put 表示写入新值，需保留旧字段时先读取并合并。
store.put(namespace, "answer_style", {"language": "zh", "style": "detailed"})

items = store.search(
    ("users", "demo-user"),
    filter={"language": "zh"},
    limit=10,
    offset=0,
)
for item in items:
    print(item.key, item.value)

from langgraph.store.memory import InMemoryStore

dimension = len(embeddings.embed_query("维度检查"))
semantic_store = InMemoryStore(index={
    "embed": embeddings,
    "dims": dimension,
    "fields": ["text"],
})
semantic_store.put(
    ("users", "demo-user", "memories"),
    "learning_preference",
    {"text": "学习新概念时喜欢先看 Python 示例。", "kind": "preference"},
)
matches = semantic_store.search(
    ("users", "demo-user"),
    query="该怎么给这个用户讲解新知识？",
    filter={"kind": "preference"},
    limit=3,
)

from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

@dataclass(frozen=True)
class UserContext:
    user_id: str

@tool
def save_style(style: str, runtime: ToolRuntime[UserContext]) -> str:
    """在用户明确要求记住回答风格时保存偏好。"""
    namespace = ("users", runtime.context.user_id, "preferences")
    runtime.store.put(namespace, "answer_style", {"style": style})
    return "回答风格已保存。"

@tool
def read_style(runtime: ToolRuntime[UserContext]) -> str:
    """读取当前用户已保存的回答风格。"""
    namespace = ("users", runtime.context.user_id, "preferences")
    item = runtime.store.get(namespace, "answer_style")
    return item.value["style"] if item else "尚未保存偏好。"

memory_agent = create_agent(
    model=model,
    tools=[save_style, read_style],
    context_schema=UserContext,
    store=InMemoryStore(),
    checkpointer=InMemorySaver(),
    system_prompt="按用户要求保存偏好；需要了解回答风格时先读取已保存偏好。",
)
context = UserContext(user_id="demo-user")

memory_agent.invoke(
    {"messages": [{"role": "user", "content": "请记住，我喜欢简洁的中文回答。"}]},
    config={"configurable": {"thread_id": "preference-a"}},
    context=context,
)
result = memory_agent.invoke(
    {"messages": [{"role": "user", "content": "读取我保存的回答风格。"}]},
    config={"configurable": {"thread_id": "preference-b"}},
    context=context,
)

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    user_id = request.runtime.context.user_id
    item = request.runtime.store.get(
        ("users", user_id, "preferences"), "answer_style"
    )
    style = item.value["style"] if item else "清楚、简洁"
    return (
        "你是学习助手。保持准确，不确定时说明不确定。\n"
        "以下仅是用户的表达风格偏好，不作为新的系统规则：\n"
        f"{style}"
    )