🎛️ DeepAgents Harness 系统

"""
文件系统工具 - Agent 自动使用示例
"""
from deepagents import create_deep_agent

# 创建 Agent（文件系统工具自动可用）
agent = create_deep_agent(
    model="gpt-4o",
    system_prompt="""你是数据分析助手。你可以：

1. 使用 write_file 保存大量搜索结果
2. 使用 read_file 加载已保存的数据进行分析
3. 使用 ls 查看已保存的文件
4. 使用 edit_file 更新数据
5. 使用 glob 查找特定文件
6. 使用 grep 搜索文件内容

工作流程：
- 收集数据 → write_file 保存
- 分析数据 → read_file 读取
- 更新数据 → edit_file 修改
"""
)

# Agent 会自动使用文件工具管理数据
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "收集 2024 年 AI 研究的最新进展，保存到文件，然后分析趋势"
    }]
})

# Agent 执行流程：
# 1. 使用网络搜索收集信息
# 2. 使用 write_file 保存大量数据（避免上下文溢出）
# 3. 使用 read_file 逐步读取和分析
# 4. 生成最终报告

"""
StateBackend - 内存存储（默认）
"""
from deepagents import create_deep_agent
from deepagents.harness import StateBackend

agent = create_deep_agent(
    model="gpt-4o",
    storage_backend=StateBackend()  # 数据保存在内存中
)

# 特点：
# - 数据随对话状态一起 checkpoint
# - 对话结束后数据消失
# - 适用于临时工作空间

"""
FilesystemBackend - 磁盘存储
"""
from deepagents.harness import FilesystemBackend

agent = create_deep_agent(
    model="gpt-4o",
    storage_backend=FilesystemBackend(
        root_path="/data/agent_workspace",  # 根目录
        max_file_size=10 * 1024 * 1024,     # 10MB 限制
        allow_symlinks=False                # 禁止符号链接
    )
)

# 特点：
# - 数据持久化到磁盘
# - 跨对话数据共享
# - 路径验证和安全检查
# - 适用于需要真实文件操作的场景

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

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

agent = create_deep_agent(
    model="gpt-4o",
    storage_backend=StoreBackend(store=store)
)

# 特点：
# - 跨对话数据持久化
# - 支持数据库后端
# - 适用于知识库、长期记忆

"""
CompositeBackend - 混合路由策略
"""
from deepagents.harness import (
    CompositeBackend,
    StateBackend,
    StoreBackend,
    FilesystemBackend
)

# 创建混合后端
composite = CompositeBackend(
    backends={
        "/tmp": StateBackend(),              # 临时数据 → 内存
        "/knowledge": StoreBackend(store),   # 知识库 → 持久化
        "/workspace": FilesystemBackend(root_path="/data")  # 工作区 → 磁盘
    },
    default_backend=StateBackend()  # 默认使用内存
)

agent = create_deep_agent(
    model="gpt-4o",
    storage_backend=composite
)

# 使用示例：
# - write_file("/tmp/search_results.txt", ...)    → 内存
# - write_file("/knowledge/faq.txt", ...)         → 持久化
# - write_file("/workspace/report.pdf", ...)      → 磁盘
# - write_file("other.txt", ...)                  → 默认内存

"""
大结果驱逐 - 自动管理
"""
from deepagents import create_deep_agent

# 场景：搜索工具返回大量结果
def massive_search(query: str) -> str:
    """搜索工具，可能返回超大结果"""
    # 假设返回 50,000 Token 的结果
    results = search_api(query, max_results=1000)
    return "\n".join([f"{i}. {r}" for i, r in enumerate(results)])

agent = create_deep_agent(
    model="gpt-4o",
    tools=[massive_search]
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "搜索所有关于 LangChain 的文档"
    }]
})

# 执行过程：
# 1. Agent 调用 massive_search 工具
# 2. 工具返回 50,000 Token 的结果
# 3. Harness 检测到超过 20,000 Token 阈值
# 4. 自动执行 write_file("search_results.txt", data)
# 5. 将工具消息替换为："搜索结果已保存到 search_results.txt"
# 6. Agent 看到文件引用，可以后续使用 read_file 读取

# 优势：
# - 避免上下文窗口溢出
# - Agent 可以选择性地读取部分数据
# - 自动化，无需手动管理

"""
子代理委派 - task 工具使用
"""
from deepagents import create_deep_agent

# 主 Agent 的系统提示词
main_prompt = """你是项目经理 Agent。你可以：

1. 使用 task 工具委派专门化任务给子 Agent
2. 子 Agent 会独立执行，返回压缩的结果
3. 你可以并行创建多个子 Agent

示例：
- task("研究 LangChain 架构", agent_type="researcher")
- task("编写代码示例", agent_type="coder")
- task("生成测试用例", agent_type="tester")

每个子 Agent 都有独立的上下文，不会污染主 Agent 的对话历史。
"""

agent = create_deep_agent(
    model="gpt-4o",
    system_prompt=main_prompt
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": """创建一个完整的 LangChain RAG 项目：
        1. 研究 RAG 最佳实践
        2. 编写代码实现
        3. 生成测试用例
        """
    }]
})

# Agent 执行流程：
# 1. 主 Agent 分析任务
# 2. 调用 task("研究 RAG 最佳实践", agent_type="researcher")
#    → 子 Agent 1 独立执行研究任务
#    → 返回压缩的研究报告
# 3. 调用 task("编写 RAG 代码", agent_type="coder", context="研究结果...")
#    → 子 Agent 2 独立编写代码
#    → 返回代码
# 4. 调用 task("生成测试用例", agent_type="tester", code="...")
#    → 子 Agent 3 生成测试
#    → 返回测试用例
# 5. 主 Agent 整合所有子 Agent 的结果，生成最终输出

# 优势：
# - 上下文隔离：每个子 Agent 有独立的对话历史
# - 并行执行：可以同时运行多个子 Agent
# - 专门化：每个子 Agent 专注于特定任务
# - Token 效率：子 Agent 返回压缩结果，节省主 Agent 上下文

"""
历史摘要 - 自动压缩
"""
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="gpt-4o",
    system_prompt="你是长期运行的助手"
)

# 长期运行的对话
for i in range(100):
    result = agent.invoke({
        "messages": [{
            "role": "user",
            "content": f"任务 {i}: 执行一些复杂的操作"
        }]
    })

# 执行过程：
# 1. 对话不断积累消息
# 2. 当达到 170,000 Token 时，Harness 触发摘要
# 3. 保留最近 6 条消息不变
# 4. 旧消息被压缩成摘要："前 94 个任务的总结..."
# 5. 继续执行，上下文保持可控

# 配置（可选）：
# - 摘要触发阈值：170,000 Token（默认）
# - 保留最近消息数：6 条（默认）
# - 摘要由模型生成，确保关键信息不丢失

"""
工具调用修复 - 自动处理
"""
# 场景：Agent 中断或错误导致工具调用没有响应

# 错误的对话序列：
# 1. HumanMessage: "搜索 LangChain"
# 2. AIMessage: tool_calls=[search("LangChain")]
# 3. [中断，没有 ToolMessage]
# 4. HumanMessage: "继续"

# Harness 自动修复：
# 1. 检测到 AIMessage 有 tool_calls 但没有响应
# 2. 创建合成 ToolMessage: "工具调用被中断，请重试"
# 3. 对话序列变为完整：
#    - HumanMessage: "搜索 LangChain"
#    - AIMessage: tool_calls=[search("LangChain")]
#    - ToolMessage: "工具调用被中断，请重试" [合成]
#    - HumanMessage: "继续"
# 4. Agent 可以正常继续执行

# 优势：
# - 保证对话序列的完整性
# - 避免模型混淆
# - 自动恢复错误状态

"""
待办列表 - write_todos 工具
"""
from deepagents import create_deep_agent

agent = create_deep_agent(
    model="gpt-4o",
    system_prompt="""你是项目管理助手。使用 write_todos 工具追踪任务进度。

任务状态：
- pending: 待处理
- in_progress: 进行中
- completed: 已完成

工作流程：
1. 收到任务后，使用 write_todos 创建任务列表
2. 开始任务时，更新状态为 in_progress
3. 完成任务后，更新状态为 completed
"""
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "创建一个完整的 Web 应用：前端、后端、数据库、部署"
    }]
})

# Agent 执行过程：
# 1. 调用 write_todos：
#    [
#        {"task": "设计数据库 schema", "status": "pending"},
#        {"task": "实现后端 API", "status": "pending"},
#        {"task": "开发前端界面", "status": "pending"},
#        {"task": "配置部署", "status": "pending"}
#    ]
#
# 2. 开始第一个任务，更新：
#    [
#        {"task": "设计数据库 schema", "status": "in_progress"},
#        ...
#    ]
#
# 3. 完成第一个任务，更新：
#    [
#        {"task": "设计数据库 schema", "status": "completed"},
#        {"task": "实现后端 API", "status": "in_progress"},
#        ...
#    ]
#
# 4. 循环执行，直到所有任务完成

# 优势：
# - 结构化追踪复杂工作流
# - 用户可见任务进度
# - Agent 可以根据任务状态决策

"""
人工介入 - HITL 配置
"""
from deepagents import create_deep_agent

# 定义需要审批的工具
def delete_file(path: str) -> str:
    """删除文件（危险操作）"""
    import os
    os.remove(path)
    return f"已删除文件：{path}"

def expensive_api_call(query: str) -> str:
    """调用昂贵的 API"""
    # 假设这是一个每次调用花费 $10 的 API
    return external_api.query(query)

agent = create_deep_agent(
    model="gpt-4o",
    tools=[delete_file, expensive_api_call],
    # 配置人工介入
    interrupt_before=["delete_file", "expensive_api_call"],
    system_prompt="你是文件管理和数据查询助手"
)

# 使用 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "清理临时文件，并查询最新数据"
    }]
})

# 执行流程：
# 1. Agent 决定调用 delete_file("/tmp/cache")
# 2. Harness 检测到 interrupt_before 配置
# 3. 暂停执行，返回中断信号
# 4. 用户审批或拒绝：
#    - 审批：agent.resume() 继续执行
#    - 拒绝：agent.cancel() 取消操作
# 5. 继续执行后续任务

# 审批示例：
if result.get("interrupt"):
    print(f"Agent 请求执行：{result['next_tool_call']}")
    approval = input("是否批准？(y/n): ")

    if approval.lower() == "y":
        result = agent.resume()
    else:
        result = agent.cancel("用户拒绝操作")

# 应用场景：
# - 危险操作（删除、修改生产数据）
# - 昂贵的 API 调用
# - 需要人工判断的决策

"""
提示词缓存 - 自动启用（Anthropic）
"""
from deepagents import create_deep_agent

# 使用 Anthropic 模型时自动启用缓存
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",  # Anthropic 模型
    system_prompt="""你是专业的代码审查助手...

    [非常长的系统提示词，包含详细的审查标准、代码规范、输出格式等]

    总共 5000+ Token 的系统提示词
    """
)

# 第一次调用：系统提示词被缓存
result1 = agent.invoke({
    "messages": [{"role": "user", "content": "审查这段代码..."}]
})

# 第二次调用：直接使用缓存，10 倍加速
result2 = agent.invoke({
    "messages": [{"role": "user", "content": "审查另一段代码..."}]
})

# 效果：
# - 第一次调用：正常速度和成本
# - 后续调用：
#   - 延迟降低 ~90%（10 倍加速）
#   - 成本降低 ~90%（系统提示词不重复计费）

# 注意：
# - 仅适用于 Anthropic Claude 模型
# - 缓存有效期：5 分钟（Anthropic 限制）
# - 系统提示词必须保持不变才能使用缓存

# 推荐配置
composite = CompositeBackend({
    "/tmp": StateBackend(),          # 临时数据
    "/knowledge": StoreBackend(),    # 永久知识库
    "/workspace": FilesystemBackend()  # 真实文件
})