👥 DeepAgents 子代理

"""
字典形式定义子代理
"""
from deepagents import create_deep_agent

# 定义子代理
research_subagent = {
    # 必需字段
    "name": "research-agent",  # 唯一标识符，主 Agent 通过此名调用
    "description": "用于深入研究问题，收集和综合信息",  # 清晰的行动导向说明
    "system_prompt": """你是一名优秀的研究者。你的任务是：

1. 将研究问题分解为可搜索的查询
2. 使用 internet_search 查找相关信息
3. 综合发现成简明总结
4. 引用来源

输出格式：
- 摘要（2-3 段）
- 关键发现（要点）
- 来源（带 URL）

⚠️ 重要：保持回复在 500 字以内以维持清洁上下文。""",
    "tools": [internet_search],  # 子代理可用的工具列表

    # 可选字段
    "model": "openai:gpt-4o",  # 覆盖主 Agent 的模型
}

# 创建主 Agent，传入子代理
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    subagents=[research_subagent]
)

# 主 Agent 可以调用子代理
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "研究量子计算的最新进展"
    }]
})

# 主 Agent 会自动决定何时调用子代理
# 子代理完成任务后，主 Agent 只接收简洁的摘要

"""
CompiledSubAgent - 基于 LangGraph 的复杂子代理
"""
from deepagents import create_deep_agent, CompiledSubAgent
from langgraph.graph import StateGraph

# 定义复杂的工作流图
class AnalysisState(TypedDict):
    raw_data: str
    processed_data: dict
    insights: list

def load_data(state: AnalysisState) -> AnalysisState:
    # 加载数据的逻辑
    state["raw_data"] = "..."
    return state

def process_data(state: AnalysisState) -> AnalysisState:
    # 处理数据的逻辑
    state["processed_data"] = {...}
    return state

def generate_insights(state: AnalysisState) -> AnalysisState:
    # 生成洞察的逻辑
    state["insights"] = [...]
    return state

# 构建 LangGraph 工作流
workflow = StateGraph(AnalysisState)
workflow.add_node("load", load_data)
workflow.add_node("process", process_data)
workflow.add_node("insights", generate_insights)

workflow.set_entry_point("load")
workflow.add_edge("load", "process")
workflow.add_edge("process", "insights")
workflow.add_edge("insights", END)

custom_graph = workflow.compile()

# 创建 CompiledSubAgent
custom_subagent = CompiledSubAgent(
    name="data-analyzer",
    description="复杂数据分析任务的专属代理，包含多步骤工作流",
    runnable=custom_graph
)

# 使用 CompiledSubAgent
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    subagents=[custom_subagent]
)

# 主 Agent 可以调用此复杂子代理
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "分析用户行为数据并生成洞察报告"
    }]
})

"""
使用通用目的子代理
"""
from deepagents import create_deep_agent

# 创建 Agent（无需显式定义 general-purpose 子代理）
agent = create_deep_agent(
    model="gpt-4o",
    tools=[internet_search],
    system_prompt="你是研究助手，可以搜索和分析信息。"
)

# 主 Agent 会自动使用 general-purpose 子代理
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "研究 AI 安全的最新进展，并生成详细报告"
    }]
})

# 执行过程：
# 1. 主 Agent 识别到这是复杂的多步骤任务
# 2. 自动调用 general-purpose 子代理
# 3. 子代理执行搜索、分析、报告生成
# 4. 子代理返回简洁的最终报告给主 Agent
# 5. 主 Agent 将结果返回给用户

# 优势：
# - 无需定义专门的子代理
# - 主 Agent 上下文保持清洁
# - 子代理的中间步骤被隔离

"""
多个专业化子代理 - 完整协作示例
"""
import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent

# 初始化工具
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

def internet_search(query: str, max_results: int = 5):
    """网络搜索工具"""
    return tavily_client.search(query, max_results=max_results)

def database_query(sql: str):
    """数据库查询工具（模拟）"""
    # 实际会连接真实数据库
    return {"status": "success", "data": [...]}

def statistical_analysis(data: dict):
    """统计分析工具（模拟）"""
    # 实际会执行统计分析
    return {"mean": 42.5, "median": 40.0, "insights": [...]}

def format_document(content: str, format: Literal["pdf", "markdown", "html"]):
    """文档格式化工具（模拟）"""
    # 实际会生成格式化文档
    return f"formatted_{format}_document.pdf"

# 定义多个专业化子代理
subagents = [
    {
        "name": "data-collector",
        "description": "从各种来源收集原始数据：网络搜索、API 调用、数据库查询",
        "system_prompt": """你是数据收集专家。你的任务是：

1. 使用 internet_search 搜索网络信息
2. 使用 database_query 查询数据库
3. 整理和清理收集到的数据

输出格式：
- 数据来源列表
- 数据质量评估
- 关键数据点摘要（< 300 字）

保持输出简洁，主 Agent 不需要看到所有原始数据。""",
        "tools": [internet_search, database_query],
        "model": "openai:gpt-4o",  # 使用快速模型
    },
    {
        "name": "data-analyzer",
        "description": "分析数据并生成统计洞察和业务建议",
        "system_prompt": """你是数据分析专家。你的任务是：

1. 使用 statistical_analysis 执行统计分析
2. 识别趋势和异常值
3. 生成可操作的业务洞察

输出格式：
- 关键发现（3-5 个要点）
- 业务建议（具体可行）
- 数据质量说明

保持输出在 400 字以内。""",
        "tools": [statistical_analysis],
        "model": "claude-sonnet-4-5-20250929",  # 使用推理能力强的模型
    },
    {
        "name": "report-writer",
        "description": "从分析结果生成精美的专业报告",
        "system_prompt": """你是报告撰写专家。你的任务是：

1. 整合分析结果
2. 使用 format_document 生成格式化报告
3. 确保报告清晰、专业、易读

输出格式：
- 执行摘要
- 详细分析章节
- 结论和建议

使用清晰的标题和要点列表。""",
        "tools": [format_document],
        "model": "claude-sonnet-4-5-20250929",
    },
]

# 创建主 Agent
agent = create_deep_agent(
    model="claude-sonnet-4-5-20250929",
    subagents=subagents,
    system_prompt="""你是项目经理 Agent。你负责协调专业化的子代理团队。

可用的子代理：
1. data-collector: 收集数据
2. data-analyzer: 分析数据
3. report-writer: 生成报告

工作流程：
1. 接收用户请求
2. 分解任务
3. 按顺序委派给相应的子代理
4. 整合子代理的结果
5. 返回最终输出给用户

你只需要协调，不需要直接执行数据收集、分析或报告生成。"""
)

# 使用主 Agent
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": """请分析我们公司 2024 年的销售数据：
        1. 从数据库和网络收集相关数据
        2. 执行统计分析
        3. 生成专业的 PDF 报告
        """
    }]
})

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

# 执行流程：
# 1. 主 Agent 分解任务为 3 个子任务
# 2. 调用 data-collector 子代理收集数据
#    → 返回简洁的数据摘要（< 300 字）
# 3. 调用 data-analyzer 子代理分析数据
#    → 返回关键发现和建议（< 400 字）
# 4. 调用 report-writer 子代理生成报告
#    → 返回报告文件路径和摘要
# 5. 主 Agent 整合所有结果，返回给用户

# 优势：
# - 每个子代理专注于特定任务
# - 主 Agent 的上下文始终清洁
# - 可以使用最适合每个任务的模型
# - 易于维护和扩展

"""
详细的子代理系统提示词示例
"""
research_subagent = {
    "name": "research-agent",
    "description": "深入研究技术主题，综合多个来源的信息",
    "system_prompt": """你是一名彻底的研究者。你的工作是：

## 职责
1. 将研究问题分解为可搜索的查询
2. 使用 internet_search 查找相关信息
3. 综合发现成简明总结
4. 引用来源

## 工作流程
1. 分析研究问题的核心要点
2. 设计 3-5 个有针对性的搜索查询
3. 执行搜索并评估结果质量
4. 交叉验证重要信息
5. 综合成结构化摘要

## 输出格式
### 摘要
[2-3 段总结核心发现]

### 关键发现
- 发现 1
- 发现 2
- 发现 3

### 来源
- [标题](URL)
- [标题](URL)

## 注意事项
- 优先使用权威来源（.edu, .gov, 知名媒体）
- 标注信息的时效性
- 如有矛盾，说明不同观点
- ⚠️ **重要**：保持回复在 500 字以内以维持清洁上下文
""",
    "tools": [internet_search],
}

# ❌ 不好：给子代理过多工具
subagent = {
    "name": "data-collector",
    "tools": [search, database, api_call, file_read, email_send, slack_notify, ...]  # 太多了！
}

# ✅ 好：只给必需工具
subagent = {
    "name": "data-collector",
    "tools": [search, database]  # 仅收集数据所需
}

# 在主 Agent 的系统提示中指导委派
agent = create_deep_agent(
    model="gpt-4o",
    subagents=[research_subagent],
    system_prompt="""你是协调助手。你可以将任务委派给专门的子代理：

- research-agent: 用于深入研究问题

当遇到需要深入研究的任务时，应该委派给 research-agent。"""
)