🔧 DeepAgents 后端配置

"""
StateBackend - 默认内存存储
"""
from deepagents import create_deep_agent

# 方式 1：默认使用（无需指定）
agent = create_deep_agent(
    model="gpt-4o"
)

# 方式 2：显式指定
from deepagents.backends import StateBackend

agent = create_deep_agent(
    model="gpt-4o",
    backend=lambda rt: StateBackend(rt)
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "创建一个临时文件保存搜索结果"
    }]
})

# 特点：
# - 文件存储在内存中（Agent 状态）
# - 对话结束后数据消失
# - 最快的性能（无 I/O 开销）
# - 适用于临时工作空间

# 限制：
# - 不跨线程共享
# - 不持久化到磁盘
# - 占用内存空间

"""
FilesystemBackend - 本地磁盘存储
"""
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

# 配置磁盘后端
agent = create_deep_agent(
    model="gpt-4o",
    backend=FilesystemBackend(
        root_dir="/Users/myname/agent_workspace",  # 根目录
        virtual_mode=True  # 启用虚拟模式（安全）
    )
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "创建文件 /project/README.md，内容是项目说明"
    }]
})

# 执行后：
# - 文件实际写入：/Users/myname/agent_workspace/project/README.md
# - Agent 看到的路径：/project/README.md（虚拟路径）

# virtual_mode=True 的作用：
# 1. 防止路径遍历攻击（Agent 无法访问 root_dir 之外的路径）
# 2. 路径规范化（自动处理 ../ 和符号链接）
# 3. 安全沙箱（限制 Agent 操作范围）

# 特点：
# - 真实磁盘 I/O
# - 跨对话持久化
# - 适用于本地开发、CI 环境
# - 支持大文件（不受内存限制）

"""
虚拟模式示例
"""
# 场景：Agent 尝试路径遍历攻击
agent = create_deep_agent(
    backend=FilesystemBackend(
        root_dir="/workspace",
        virtual_mode=True
    )
)

# Agent 尝试访问系统文件
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "读取文件 /../../../etc/passwd"
    }]
})

# 结果：
# - virtual_mode=True：访问被拒绝（安全）
# - virtual_mode=False：可能访问成功（危险）

"""
StoreBackend - 持久化存储
"""
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend
from langgraph.store.memory import InMemoryStore

# 创建持久化 Store
store = InMemoryStore()  # 可替换为 RedisStore、PostgresStore

# 配置 StoreBackend
agent = create_deep_agent(
    model="gpt-4o",
    backend=lambda rt: StoreBackend(rt),
    store=store  # 传递 store 实例
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "保存用户偏好到 /memories/user_preferences.json"
    }]
})

# 特点：
# - 跨对话持久化
# - 跨线程共享数据
# - 支持数据库后端（Redis、PostgreSQL）
# - 适用于长期记忆、知识库

# 使用场景：
# - 用户配置和偏好
# - FAQ 知识库
# - 历史对话记录
# - 共享的参考数据

"""
StoreBackend - PostgreSQL 持久化
"""
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend
from langgraph.store.postgres import PostgresStore

# 连接 PostgreSQL
store = PostgresStore(
    connection_string="postgresql://user:pass@localhost:5432/deepagents"
)

agent = create_deep_agent(
    model="gpt-4o",
    backend=lambda rt: StoreBackend(rt),
    store=store
)

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

"""
CompositeBackend - 混合路由策略
"""
from deepagents import create_deep_agent
from deepagents.backends import (
    CompositeBackend,
    StateBackend,
    StoreBackend,
    FilesystemBackend
)
from langgraph.store.memory import InMemoryStore

# 创建 Store
store = InMemoryStore()

# 定义路由规则
composite_backend = lambda rt: CompositeBackend(
    default=StateBackend(rt),  # 默认使用内存
    routes={
        "/tmp/": StateBackend(rt),              # 临时数据 → 内存
        "/knowledge/": StoreBackend(rt),        # 知识库 → 持久化
        "/workspace/": FilesystemBackend(       # 工作区 → 磁盘
            root_dir="/Users/myname/workspace",
            virtual_mode=True
        ),
    }
)

# 创建 Agent
agent = create_deep_agent(
    model="gpt-4o",
    backend=composite_backend,
    store=store
)

# 使用示例：

# 1. 临时计算结果 → 内存
agent.invoke({
    "messages": [{
        "role": "user",
        "content": "保存搜索结果到 /tmp/search.txt"
    }]
})  # 使用 StateBackend（内存）

# 2. 知识库数据 → 持久化
agent.invoke({
    "messages": [{
        "role": "user",
        "content": "保存 FAQ 到 /knowledge/faq.json"
    }]
})  # 使用 StoreBackend（数据库）

# 3. 工作区文件 → 磁盘
agent.invoke({
    "messages": [{
        "role": "user",
        "content": "创建项目文件 /workspace/main.py"
    }]
})  # 使用 FilesystemBackend（磁盘）

# 4. 其他路径 → 默认内存
agent.invoke({
    "messages": [{
        "role": "user",
        "content": "保存临时数据到 /data/temp.txt"
    }]
})  # 使用默认 StateBackend（内存）

"""
复杂路由规则示例
"""
composite = lambda rt: CompositeBackend(
    default=StateBackend(rt),
    routes={
        "/memories/": FilesystemBackend(
            root_dir="/deepagents/myagent",
            virtual_mode=True
        ),
        "/memories/projects/": StoreBackend(rt),  # 更具体的前缀
    }
)

# 路由结果：
# - /memories/projects/code.py → StoreBackend（更长前缀优先）
# - /memories/notes.txt → FilesystemBackend
# - /tmp/data.txt → StateBackend（默认）

"""
自定义后端 - S3Backend 示例
"""
from typing import Protocol
from deepagents.backends import (
    BackendProtocol,
    LsResult,
    ReadResult,
    WriteResult,
    EditResult,
    GrepResult,
    GlobResult
)
import boto3

class S3Backend(BackendProtocol):
    """S3 存储后端"""

    def __init__(self, bucket_name: str, prefix: str = ""):
        self.s3 = boto3.client("s3")
        self.bucket = bucket_name
        self.prefix = prefix

    def ls_info(self, path: str) -> LsResult:
        """列出 S3 对象"""
        try:
            response = self.s3.list_objects_v2(
                Bucket=self.bucket,
                Prefix=self._normalize_path(path),
                Delimiter="/"
            )

            files = []
            for obj in response.get("Contents", []):
                files.append({
                    "name": obj["Key"].split("/")[-1],
                    "size": obj["Size"],
                    "modified": obj["LastModified"].isoformat()
                })

            return LsResult(files=files, files_update=None)
        except Exception as e:
            return LsResult(error=f"S3 列表失败: {str(e)}")

    def read(self, file_path: str, offset: int = 0,
             limit: int | None = None) -> ReadResult:
        """从 S3 读取文件"""
        try:
            key = self._normalize_path(file_path)
            response = self.s3.get_object(Bucket=self.bucket, Key=key)
            content = response["Body"].read().decode("utf-8")

            # 应用 offset 和 limit
            lines = content.splitlines()
            if offset or limit:
                lines = lines[offset:offset + limit if limit else None]

            return ReadResult(
                content="\n".join(lines),
                files_update=None
            )
        except Exception as e:
            return ReadResult(error=f"S3 读取失败: {str(e)}")

    def write(self, file_path: str, content: str) -> WriteResult:
        """写入 S3 对象"""
        try:
            key = self._normalize_path(file_path)
            self.s3.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=content.encode("utf-8")
            )
            return WriteResult(
                success=f"已写入 S3: s3://{self.bucket}/{key}",
                files_update=None
            )
        except Exception as e:
            return WriteResult(error=f"S3 写入失败: {str(e)}")

    def edit(self, file_path: str, old_string: str,
             new_string: str, replace_all: bool = False) -> EditResult:
        """编辑 S3 对象"""
        # 1. 读取内容
        read_result = self.read(file_path)
        if read_result.error:
            return EditResult(error=read_result.error)

        # 2. 执行替换
        content = read_result.content
        if replace_all:
            new_content = content.replace(old_string, new_string)
        else:
            new_content = content.replace(old_string, new_string, 1)

        # 3. 写回
        write_result = self.write(file_path, new_content)
        if write_result.error:
            return EditResult(error=write_result.error)

        return EditResult(
            success=f"已编辑 {file_path}",
            files_update=None
        )

    def grep_raw(self, pattern: str, path: str = "/",
                 glob: str | None = None) -> GrepResult:
        """S3 内容搜索（简化实现）"""
        return GrepResult(error="S3Backend 暂不支持 grep")

    def glob_info(self, pattern: str, path: str = "/") -> GlobResult:
        """S3 模式匹配（简化实现）"""
        return GlobResult(error="S3Backend 暂不支持 glob")

    def _normalize_path(self, path: str) -> str:
        """规范化路径"""
        path = path.lstrip("/")
        return f"{self.prefix}/{path}" if self.prefix else path


# 使用自定义后端
agent = create_deep_agent(
    model="gpt-4o",
    backend=S3Backend(
        bucket_name="my-agent-storage",
        prefix="agents/workspace"
    )
)

# Agent 的文件操作会使用 S3
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "保存报告到 /reports/summary.md"
    }]
})

# 实际操作：
# - 文件写入 S3: s3://my-agent-storage/agents/workspace/reports/summary.md

"""
策略钩子 - 子类化方式
"""
from deepagents.backends import FilesystemBackend, WriteResult

class GuardedBackend(FilesystemBackend):
    """带安全策略的文件系统后端"""

    def __init__(self, *, deny_prefixes: list[str], **kwargs):
        super().__init__(**kwargs)
        # 规范化拒绝前缀
        self.deny_prefixes = [
            p if p.endswith("/") else p + "/"
            for p in deny_prefixes
        ]

    def write(self, file_path: str, content: str) -> WriteResult:
        """写入前检查权限"""
        # 检查是否在拒绝列表中
        if any(file_path.startswith(p) for p in self.deny_prefixes):
            return WriteResult(
                error=f"写入被拒绝：{file_path} (受保护路径)"
            )

        # 允许写入
        return super().write(file_path, content)

    def edit(self, file_path: str, old_string: str,
             new_string: str, replace_all: bool = False):
        """编辑前检查权限"""
        if any(file_path.startswith(p) for p in self.deny_prefixes):
            return EditResult(
                error=f"编辑被拒绝：{file_path} (受保护路径)"
            )
        return super().edit(file_path, old_string, new_string, replace_all)


# 使用受保护的后端
agent = create_deep_agent(
    model="gpt-4o",
    backend=GuardedBackend(
        root_dir="/workspace",
        virtual_mode=True,
        deny_prefixes=[
            "/config/",      # 配置文件禁止写入
            "/production/",  # 生产数据禁止修改
        ]
    )
)

# Agent 尝试写入受保护路径
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "修改配置文件 /config/settings.json"
    }]
})

# 结果：写入被拒绝（安全策略生效）

"""
策略钩子 - 包装器模式
"""
from deepagents.backends import BackendProtocol, WriteResult

class PolicyWrapper(BackendProtocol):
    """策略包装器"""

    def __init__(self, inner: BackendProtocol,
                 deny_prefixes: list[str] | None = None,
                 max_file_size: int | None = None):
        self.inner = inner
        self.deny_prefixes = [
            p if p.endswith("/") else p + "/"
            for p in (deny_prefixes or [])
        ]
        self.max_file_size = max_file_size

    def write(self, file_path: str, content: str) -> WriteResult:
        """写入策略检查"""
        # 1. 路径检查
        if any(file_path.startswith(p) for p in self.deny_prefixes):
            return WriteResult(error=f"拒绝访问: {file_path}")

        # 2. 大小检查
        if self.max_file_size and len(content) > self.max_file_size:
            return WriteResult(
                error=f"文件过大: {len(content)} 字节 "
                      f"(最大 {self.max_file_size})"
            )

        # 3. 执行实际写入
        return self.inner.write(file_path, content)

    # 委托其他方法到内部后端
    def ls_info(self, path: str):
        return self.inner.ls_info(path)

    def read(self, file_path: str, offset: int = 0, limit: int | None = None):
        return self.inner.read(file_path, offset, limit)

    # ... 其他方法类似


# 使用包装器
base_backend = FilesystemBackend(root_dir="/workspace", virtual_mode=True)

protected_backend = PolicyWrapper(
    inner=base_backend,
    deny_prefixes=["/config/", "/secrets/"],
    max_file_size=10 * 1024 * 1024  # 10MB 限制
)

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

# 优势：
# - 保持后端解耦
# - 便于测试和组合
# - 可以嵌套多个包装器（链式策略）

# ✅ 推荐配置
composite = lambda rt: CompositeBackend(
    default=StateBackend(rt),        # 默认临时存储
    routes={
        "/tmp/": StateBackend(rt),           # 显式临时
        "/knowledge/": StoreBackend(rt),     # 知识库持久化
        "/workspace/": FilesystemBackend(    # 工作区磁盘
            root_dir="/data/workspace",
            virtual_mode=True
        ),
    }
)

# 创建共享 Store
shared_store = PostgresStore(connection_string="...")

# Agent 1
agent1 = create_deep_agent(
    backend=lambda rt: StoreBackend(rt),
    store=shared_store
)

# Agent 2（共享同一个 Store）
agent2 = create_deep_agent(
    backend=lambda rt: StoreBackend(rt),
    store=shared_store
)

# Agent 1 写入的数据，Agent 2 可以读取