🧠 DeepAgents 长期记忆

"""
长期记忆配置 - InMemoryStore（开发环境）
"""
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore

# 步骤 1: 创建 Store
store = InMemoryStore()

# 步骤 2: 配置后端路由
def make_backend(runtime):
    return CompositeBackend(
        default=StateBackend(runtime),  # 默认：临时存储
        routes={
            "/memories/": StoreBackend(runtime)  # /memories/ 路径：持久化
        }
    )

# 步骤 3: 创建 Agent
agent = create_deep_agent(
    model="gpt-4o",
    store=store,
    backend=make_backend,
    system_prompt="""你是助手。当用户告诉你偏好时，
    保存到 /memories/user_preferences.txt 以便未来参考。"""
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我喜欢简洁的回答，记住这一点"
    }]
})

# Agent 会自动将偏好写入 /memories/user_preferences.txt
# 下次对话可以读取这个文件

"""
长期记忆配置 - PostgresStore（生产环境）
"""
import os
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.postgres import PostgresStore

# 步骤 1: 连接 PostgreSQL Store
store = PostgresStore(
    connection_string=os.environ["DATABASE_URL"]
)

# 步骤 2: 配置后端路由（同上）
def make_backend(runtime):
    return CompositeBackend(
        default=StateBackend(runtime),
        routes={
            "/memories/": StoreBackend(runtime)
        }
    )

# 步骤 3: 创建 Agent
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    store=store,
    backend=make_backend,
    system_prompt="""你是智能助手。

## 长期记忆管理

你可以将重要信息保存到 /memories/ 路径下：

1. 用户偏好：/memories/user_preferences.txt
2. 研究笔记：/memories/research/topic_name/notes.txt
3. 项目记录：/memories/projects/project_name/status.md

这些文件会永久保存，跨所有对话访问。"""
)

# 优势：
# - 真正的持久化（数据库级别）
# - 跨实例共享（多个 Agent 实例访问同一数据）
# - 支持大规模数据
# - 生产环境推荐

"""
跨线程访问长期记忆
"""
import uuid
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore

# 创建共享的 Store
store = InMemoryStore()

# 创建 Agent
agent = create_deep_agent(
    model="gpt-4o",
    store=store,
    backend=lambda rt: CompositeBackend(
        default=StateBackend(rt),
        routes={"/memories/": StoreBackend(rt)}
    ),
    system_prompt="""你可以保存和读取用户偏好。

使用 /memories/user_preferences.txt 存储用户偏好。"""
)

# ===== 线程 1：保存偏好 =====
print("=== 线程 1：保存偏好 ===")
config1 = {"configurable": {"thread_id": str(uuid.uuid4())}}

result1 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我喜欢技术细节和代码示例，请记住"
    }]
}, config=config1)

print(result1["messages"][-1].content)

# ===== 线程 2：读取偏好（不同的线程 ID）=====
print("\n=== 线程 2：读取偏好 ===")
config2 = {"configurable": {"thread_id": str(uuid.uuid4())}}  # 不同的 thread_id

result2 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我之前告诉过你什么偏好？"
    }]
}, config=config2)

print(result2["messages"][-1].content)

# 输出示例：
# === 线程 1：保存偏好 ===
# 我已经记住了你的偏好：喜欢技术细节和代码示例。
# 这已保存到长期记忆中，未来对话我会参考这一点。
#
# === 线程 2：读取偏好 ===
# 根据记录，你喜欢技术细节和代码示例。
# 我会在回答中提供更多的技术深度和实际代码。

# 关键点：
# - 两个线程使用不同的 thread_id
# - 线程 2 能够访问线程 1 保存的数据
# - 这是因为数据存储在共享的 Store 中，而非线程状态

/memories/
├── user_preferences.txt              # 用户偏好
├── research/                         # 研究笔记
│   ├── ai_agents/
│   │   ├── sources.txt
│   │   └── summary.md
│   └── langchain/
│       ├── rag_notes.txt
│       └── best_practices.md
├── projects/                         # 项目记录
│   ├── customer_service_bot/
│   │   ├── requirements.md
│   │   ├── progress.txt
│   │   └── issues.txt
│   └── data_pipeline/
│       ├── design.md
│       └── status.txt
└── knowledge_base/                   # 知识库
    ├── faq.json
    └── product_docs.md

"""
完整实战示例 - 知识助手（支持长期记忆）
"""
import os
import uuid
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.postgres import PostgresStore
from langgraph.checkpoint.memory import MemorySaver

# ========== 1. 初始化工具 ==========

tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

def internet_search(query: str, max_results: int = 5) -> dict:
    """搜索网络信息"""
    return tavily_client.search(query, max_results=max_results)

# ========== 2. 配置持久化 ==========

# Store：长期记忆（跨对话）
store = PostgresStore(
    connection_string=os.environ["DATABASE_URL"]
)

# Checkpointer：短期记忆（对话状态）
checkpointer = MemorySaver()

# Backend：混合路由
def make_backend(runtime):
    return CompositeBackend(
        default=StateBackend(runtime),  # 临时数据
        routes={
            "/memories/": StoreBackend(runtime),  # 长期记忆
        }
    )

# ========== 3. 创建知识助手 ==========

knowledge_assistant_prompt = """你是一位专业的知识助手。你的任务是帮助用户进行研究、管理笔记和积累知识。

## 长期记忆管理

你可以将信息保存到以下路径（所有路径下的文件永久保存，跨对话访问）：

### 1. 用户偏好
- 路径：/memories/user_preferences.txt
- 内容：用户的个人偏好、习惯、专业领域

### 2. 研究笔记
- 路径：/memories/research/[主题]/
- 子文件：
  - sources.txt：信息来源和链接
  - summary.md：研究摘要
  - notes.txt：详细笔记

### 3. 知识库
- 路径：/memories/knowledge_base/
- 子文件：
  - [主题].md：特定主题的知识总结
  - faq.json：常见问题解答

## 工作流程

### 研究任务流程
1. 接收用户的研究主题
2. 使用 internet_search 搜索相关信息
3. 整理和分析搜索结果
4. 将信息保存到 /memories/research/[主题]/ 下
5. 生成研究报告

### 偏好管理流程
1. 当用户首次交互时，读取 /memories/user_preferences.txt
2. 如不存在，创建新文件
3. 根据用户反馈更新偏好
4. 在后续对话中应用偏好

### 知识积累流程
1. 在研究过程中发现有价值的通用知识
2. 提取关键信息
3. 保存到 /memories/knowledge_base/ 下相应主题文件
4. 在未来研究中参考已有知识

## 注意事项

- 每次对话开始时，检查是否有相关的长期记忆
- 研究笔记应结构化、易于检索
- 定期整理和更新知识库
- 引用来源时提供完整的 URL
"""

agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[internet_search],
    store=store,
    checkpointer=checkpointer,
    backend=make_backend,
    system_prompt=knowledge_assistant_prompt
)

# ========== 4. 使用知识助手 ==========

# 场景 1：首次对话，设置偏好
print("=== 场景 1：首次对话 ===")
config1 = {"configurable": {"thread_id": str(uuid.uuid4())}}

result1 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我是一名 Python 开发者，对 AI Agent 和 LangChain 很感兴趣。请记住这一点。"
    }]
}, config=config1)

print(result1["messages"][-1].content)

# 场景 2：研究任务
print("\n=== 场景 2：研究 RAG 技术 ===")
config2 = {"configurable": {"thread_id": str(uuid.uuid4())}}

result2 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "帮我研究 RAG（检索增强生成）的最佳实践，并保存研究笔记。"
    }]
}, config=config2)

print(result2["messages"][-1].content)

# 场景 3：后续对话，引用偏好和研究
print("\n=== 场景 3：后续对话 ===")
config3 = {"configurable": {"thread_id": str(uuid.uuid4())}}  # 新线程

result3 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "根据我之前的研究，总结 RAG 在 Python 项目中的应用。"
    }]
}, config=config3)

print(result3["messages"][-1].content)

# Agent 会：
# 1. 读取 /memories/user_preferences.txt（Python 开发者）
# 2. 读取 /memories/research/rag/summary.md（之前的研究）
# 3. 结合这些信息生成针对性的回答

# ========== 5. 查看长期记忆内容 ==========

print("\n=== 长期记忆文件列表 ===")
# Agent 可以使用 ls 工具列出文件
result4 = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "列出我的所有长期记忆文件"
    }]
}, config=config3)

print(result4["messages"][-1].content)

# 输出示例：
# === 场景 1：首次对话 ===
# 我已经记住了你的背景：Python 开发者，对 AI Agent 和 LangChain 感兴趣。
# 这些信息已保存到长期记忆中，未来对话我会参考这些偏好。
#
# === 场景 2：研究 RAG 技术 ===
# 我已完成 RAG 最佳实践的研究，并保存了以下文件：
# - /memories/research/rag/sources.txt：权威资源链接
# - /memories/research/rag/summary.md：核心概念和最佳实践摘要
# - /memories/research/rag/notes.txt：详细技术笔记
#
# === 场景 3：后续对话 ===
# 根据你的背景（Python 开发者）和之前的 RAG 研究，
# 以下是 RAG 在 Python 项目中的应用要点：...
# [结合偏好和研究笔记生成的回答]
#
# === 长期记忆文件列表 ===
# /memories/user_preferences.txt
# /memories/research/rag/sources.txt
# /memories/research/rag/summary.md
# /memories/research/rag/notes.txt

"""
高级路由配置 - 多存储后端
"""
from deepagents.backends import (
    CompositeBackend,
    StateBackend,
    StoreBackend,
    FilesystemBackend
)
from langgraph.store.postgres import PostgresStore

# 创建不同的 Store 实例
user_store = PostgresStore(connection_string=os.environ["USER_DB_URL"])
knowledge_store = PostgresStore(connection_string=os.environ["KNOWLEDGE_DB_URL"])

def make_advanced_backend(runtime):
    return CompositeBackend(
        default=StateBackend(runtime),  # 默认：临时内存
        routes={
            # 用户数据：专用数据库
            "/memories/users/": StoreBackend(runtime, store=user_store),

            # 知识库：共享数据库
            "/memories/knowledge/": StoreBackend(runtime, store=knowledge_store),

            # 项目文件：磁盘存储
            "/memories/projects/": FilesystemBackend(
                root_dir="/data/projects",
                virtual_mode=True
            ),

            # 临时工作区：内存
            "/tmp/": StateBackend(runtime),
        }
    )

agent = create_deep_agent(
    model="gpt-4o",
    backend=make_advanced_backend
)

# 使用场景：
# 1. /memories/users/user123.json → PostgreSQL (user_store)
# 2. /memories/knowledge/faq.md → PostgreSQL (knowledge_store)
# 3. /memories/projects/code.py → 磁盘文件
# 4. /tmp/scratch.txt → 内存（不持久化）
# 5. /data.txt（其他路径）→ 内存（默认）

# 优势：
# - 数据隔离：用户数据和知识库分离
# - 混合存储：数据库 + 文件系统 + 内存
# - 灵活路由：根据路径自动选择存储方式
# - 性能优化：临时数据用内存，重要数据用数据库

# 定期清理过期数据
system_prompt = """...

## 数据清理规则

1. 临时研究笔记：30 天后自动清理
2. 用户偏好：永久保留
3. 项目记录：项目完成后 90 天清理
4. 每月第一天执行数据清理任务
"""

routes={
    "/memories/users/": StoreBackend(runtime, store=user_store),
    "/memories/shared/": StoreBackend(runtime, store=shared_store),
}

# 设置环境变量
export DATABASE_URL="postgresql://user:pass@localhost:5432/deepagents"

# Python 代码
store = PostgresStore(connection_string=os.environ["DATABASE_URL"])

# 动态创建租户隔离的路径
def make_tenant_backend(tenant_id: str):
    def backend_factory(runtime):
        return CompositeBackend(
            default=StateBackend(runtime),
            routes={
                f"/memories/{tenant_id}/": StoreBackend(runtime)
            }
        )
    return backend_factory

# 为不同租户创建 Agent
agent_tenant1 = create_deep_agent(
    backend=make_tenant_backend("tenant_123")
)

agent_tenant2 = create_deep_agent(
    backend=make_tenant_backend("tenant_456")
)

# 数据完全隔离：
# tenant_123 的数据：/memories/tenant_123/
# tenant_456 的数据：/memories/tenant_456/