LangChain v1Note 06

第 3 章:短期记忆

短期记忆是同一对话 thread 的 Agent state 持久化,不是模型自己的记忆。给 createagent 传 checkpointer=InMemorySaver(),每次调用传相同 config={"configurable": {"threadid": "..."}},第二轮就会恢复第一轮写入的 mes

短期记忆是同一对话 thread 的 Agent state 持久化,不是模型自己的记忆。给 create_agentcheckpointer=InMemorySaver(),每次调用传相同 config={"configurable": {"thread_id": "..."}},第二轮就会恢复第一轮写入的 messages 和中间工具消息。

agent = create_agent(model, tools=[...], checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "lesson-memory-1"}}
await agent.ainvoke({"messages": [("user", "我叫小王")]}, config=config)
await agent.ainvoke({"messages": [("user", "我叫什么?")]}, config=config)
项目 边界
state / checkpoint 当前 thread 的 messages、工具调用、结构化结果;可恢复、可检查
thread_id checkpoint 的会话键,必须稳定且由服务端授权,不能直接信任任意客户端值
context 本次运行依赖;不会因 thread_id 自动持久化
Store 跨 thread 长期数据;不是短期聊天历史的替代品

InMemorySaver 仅适合本地学习和测试,进程退出即丢失。生产选择持久 checkpointer 并制定删除/保留策略。运行 07_short_term_memory.py 会发起两次短模型调用;第二次的问题必须引用第一轮名称。

当前项目第 7 章讲解同一机制。deep_researchermessages 属于 AgentState,调用方可用 thread config 恢复执行;Configuration 属于 Runtime.context,不应改放 checkpoint。

相关资源

  • 查看示例代码:docs/langchain/examples/07_short_term_memory.py
    """Chapter 3: retain one conversation with a checkpoint and thread id."""
    
    import asyncio
    
    from dotenv import load_dotenv
    from langchain.agents import create_agent
    from langchain.chat_models import init_chat_model
    from langgraph.checkpoint.memory import InMemorySaver
    
    from open_deep_research.configuration import Configuration
    
    
    load_dotenv()
    
    
    async def main() -> None:
        settings = Configuration.from_env()
        agent = create_agent(
            init_chat_model(model=settings.research_model, max_tokens=80),
            system_prompt="回答简洁。只能根据本次对话中已知的事实回答。",
            checkpointer=InMemorySaver(),
        )
        config = {"configurable": {"thread_id": "lesson-short-memory"}}
        await agent.ainvoke(
            {"messages": [("user", "请记住:我叫小王。只回答已记住。 ")]},
            config=config,
        )
        result = await agent.ainvoke(
            {"messages": [("user", "我叫什么?只输出名字。")]}, config=config
        )
        answer = result["messages"][-1].content
        assert "小王" in str(answer), answer
        print(f"第二轮回答: {answer}")
    
    
    if __name__ == "__main__":
        asyncio.run(main())