子代理系统

子代理的核心概念

子代理是一种委托机制，允许主代理将特定任务分配给专门化的代理来处理。这种设计解决了在复杂任务中可能出现的"上下文膨胀"问题。

子代理解决的问题

当主代理需要处理包含多个信息密集型操作（如网络搜索、文件读取、数据库查询）的任务时，中间结果会大量堆积在对话历史中，导致：

1. 上下文窗口浪费：用于存储中间结果而非有用信息
2. 推理效率下降：模型需要在冗长的上下文中进行推理
3. 成本增加：处理更多的 token 需要更高的 API 成本
4. 响应变慢：整个系统的响应时间增加

解决方案

子代理隔离了详细的工作过程，主代理只需要接收最终结果：

何时使用子代理

应该使用子代理

• ✅ 多步骤任务需要大量的工具调用
• ✅ 任务需要专门的模型配置或工具集
• ✅ 需要完全隔离某个工作域（上下文隔离）
• ✅ 需要对子任务使用不同的 LLM 模型
• ✅ 需要跨大数据集进行处理

不应该使用子代理

• ❌ 简单的单步操作
• ❌ 需要频繁往返沟通的任务
• ❌ 需要保持全局上下文的任务
• ❌ 任务足够简单，主代理能高效处理

基本用法

定义和使用子代理

from deepagents import create_deep_agent

# 定义子代理规范
researcher_subagent = {
    "name": "researcher",
    "description": "进行深度研究和信息收集",
    "system_prompt": """你是一个资深的研究员。
    
你的职责是：
1. 进行全面的信息搜索
2. 分析和总结找到的信息
3. 提出关键发现和结论
4. 仅返回经过验证的信息""",
    
    "tools": [web_search, fetch_article],  # 子代理专用工具
}

writer_subagent = {
    "name": "writer",
    "description": "撰写和编辑内容",
    "system_prompt": """你是一个专业的技术写手。
    
你的职责是：
1. 根据提供的信息撰写清晰的内容
2. 组织内容的逻辑结构
3. 确保文章的专业性和准确性""",
    
    "tools": [save_document, format_text],
}

# 创建主代理，包含子代理
main_agent = create_deep_agent(
    model="anthropic:claude-3-5-sonnet-20241022",
    system_prompt="你是一个内容生产管理者。协调研究员和写手生成高质量的内容。",
    subagents=[researcher_subagent, writer_subagent],
)

# 使用代理
result = main_agent.invoke({
    "messages": [{
        "role": "user",
        "content": "请写一篇关于量子计算的综合性文章"
    }]
})

子代理配置

完整的子代理规范

subagent_config = {
    # 必需字段
    "name": "data_analyst",                    # 子代理的唯一标识
    "description": "专业的数据分析和可视化工作",  # 主代理用这个决定何时使用
    "system_prompt": """你是一个数据科学家...""",  # 子代理的系统提示
    "tools": [analyze_data, plot_chart],       # 仅此子代理可用的工具
    
    # 可选字段
    "model": "openai:gpt-4o",                  # 覆盖主代理的模型
    "middleware": [custom_middleware],         # 自定义中间件
    "interrupt_on": {                          # 控制何时需要人工审批
        "delete_file": {"allowed_decisions": ["approve", "reject"]},
    },
    "skills": ["/skills/data-analysis/"],      # 子代理特有的技能
}

字段详解

字段	类型	必需	说明
name	str	✓	子代理标识符，主代理通过名称调用
description	str	✓	清晰描述子代理的职能，帮助主代理决定何时使用
system_prompt	str	✓	给子代理的详细指导
tools	list	✓	子代理可用的工具集
model	str/Model	✗	不指定则继承主代理的模型
middleware	list	✗	子代理特有的中间件
interrupt_on	dict	✗	哪些操作需要人工审批
skills	list[str]	✗	子代理的技能目录路径

实际例子

例子1：内容审核系统

from deepagents import create_deep_agent

def check_toxicity(text: str) -> dict[str, float]:
    """检查文本中的有害内容程度"""
    # 实现有害内容检测
    return {"toxicity": 0.2, "safety": 0.8}

def verify_facts(claims: list[str]) -> dict[str, bool]:
    """验证事实陈述"""
    return {claim: True for claim in claims}

def generate_response(original: str, notes: str) -> str:
    """生成修订后的响应"""
    return f"修订后的响应：{original}"

# 审核子代理
moderator = {
    "name": "moderator",
    "description": "检查内容安全性和事实准确性",
    "system_prompt": """你是一个内容审核专家。
    
检查清单：
1. 检查有害或不当内容
2. 验证关键事实
3. 标记需要修订的部分""",
    "tools": [check_toxicity, verify_facts],
}

# 写手子代理
writer = {
    "name": "writer",
    "description": "撰写内容",
    "system_prompt": "你是一个创意写手",
    "tools": [generate_response],
}

# 主代理
agent = create_deep_agent(
    system_prompt="你是编辑。协调审核员检查内容，然后让写手进行必要的修改。",
    subagents=[moderator, writer],
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "帮我起草一份关于科技趋势的文章摘要"
    }]
})

例子2：多步骤项目规划

def analyze_requirements(requirements: str) -> dict:
    """分析项目需求"""
    pass

def create_project_plan(analysis: dict) -> str:
    """创建详细的项目计划"""
    pass

def estimate_resources(plan: str) -> dict[str, any]:
    """估算所需资源"""
    pass

# 需求分析子代理
requirements_agent = {
    "name": "requirements_analyst",
    "description": "分析和理解项目需求",
    "system_prompt": """你是需求分析师。
    
你的职责是：
1. 理解用户需求
2. 识别关键功能和非功能需求
3. 列出潜在的风险和依赖关系""",
    "tools": [analyze_requirements],
}

# 规划子代理
planner = {
    "name": "project_planner",
    "description": "创建项目执行计划",
    "system_prompt": """你是项目规划师。
    
基于需求分析结果，创建：
1. 阶段分解（WBS）
2. 时间表
3. 关键里程碑""",
    "tools": [create_project_plan],
}

# 资源评估子代理
resource_agent = {
    "name": "resource_optimizer",
    "description": "估算和优化资源配置",
    "system_prompt": """你是资源规划专家。
    
在给定计划的情况下：
1. 估算所需的人力、时间和成本
2. 识别资源瓶颈
3. 提出优化建议""",
    "tools": [estimate_resources],
}

# 主协调代理
orchestrator = create_deep_agent(
    model="anthropic:claude-3-5-sonnet-20241022",
    system_prompt="""你是项目经理。协调需求分析、规划和资源估算三个步骤。
    
流程：
1. 要求需求分析师分析需求
2. 基于分析结果要求规划师制定计划
3. 基于计划要求资源优化师估算资源
4. 综合所有信息生成最终的项目计划""",
    subagents=[requirements_agent, planner, resource_agent],
)

result = orchestrator.invoke({
    "messages": [{
        "role": "user",
        "content": "我需要构建一个电商平台。主要功能包括：用户认证、商品目录、购物车、订单管理、支付集成。"
    }]
})

子代理间的沟通

顺序执行

# 主代理维护任务队列，按顺序分配给子代理
main_agent_prompt = """你是任务协调者。

对于用户的请求，按以下步骤执行：
1. 首先使用 'researcher' 代理进行研究
2. 然后使用 'analyzer' 代理进行分析
3. 最后使用 'writer' 代理生成报告

每个步骤的输出作为下一步的输入。"""

并行执行

虽然标准子代理是顺序执行的，但可以通过异步子代理实现并行执行。

共享状态

子代理通过共享的文件系统进行数据交换：

research_agent = {
    "name": "researcher",
    "system_prompt": """研究主题并将结果保存到 /research_data.md""",
    "tools": [web_search, write_file],
}

analysis_agent = {
    "name": "analyzer",
    "system_prompt": """从 /research_data.md 读取研究结果并进行分析""",
    "tools": [read_file, analyze_data, write_file],
}

高级配置

针对特定操作的人工审批

# 对不同操作设置不同的审批要求
gene_deletion_agent = {
    "name": "data_engineer",
    "description": "执行数据库操作",
    "system_prompt": "你是数据工程师",
    "tools": [delete_record, modify_data],
    "interrupt_on": {
        "delete_record": {
            "allowed_decisions": ["approve", "reject"]  # 删除需要审批
        },
        "modify_data": {
            "allowed_decisions": ["approve", "edit", "reject"]  # 修改可编辑后再执行
        }
    }
}

子代理使用不同的 LLM

# 对于涉及复杂推理的子代理使用更强大的模型
expensive_analyzer = {
    "name": "advanced_analyzer",
    "description": "进行复杂的数据分析",
    "system_prompt": "你是数据科学家",
    "tools": [analyze_data, visualize_results],
    "model": "openai:gpt-4o",  # 使用更强大的模型
}

# 对于简单任务使用廉价的模型
simple_formatter = {
    "name": "formatter",
    "description": "格式化输出",
    "system_prompt": "你是格式化专家",
    "tools": [format_text],
    "model": "openai:gpt-4-mini",  # 使用轻量级模型节省成本
}

子代理特有的技能

# 只有特定的子代理可以访问特定的技能
specialized_agent = {
    "name": "ml_specialist",
    "description": "机器学习相关任务",
    "system_prompt": "你是机器学习工程师",
    "tools": [train_model, evaluate_model],
    "skills": [
        "/skills/machine-learning/",
        "/skills/data-preprocessing/",
    ]
}

調試子代理

启用追踪

import logging

logging.basicConfig(level=logging.DEBUG)

# 现在所有子代理的执行会被记录
result = agent.invoke({"messages": [...]})

检查执行流

# 使用流式 API 查看子代理的执行过程
for event in agent.stream(
    {"messages": [{"role": "user", "content": "..."}]},
    stream_mode="updates",
):
    print(f"事件：{event}")

最佳实践

1. 清晰的职责划分

# ❌ 职责重叠
agent1_description = "处理各种任务"
agent2_description = "处理各种任务"

# ✅ 职责清晰
agent1_description = "从多个来源搜索和聚合信息"
agent2_description = "合成信息并生成连贯的摘要"

2. 合理的粒度

# ❌ 太细粒度
agents = [
    {"name": "searcher", ...},
    {"name": "parser", ...},
    {"name": "filter", ...},
    {"name": "summarizer", ...},
]

# ✅ 合理粒度
agents = [
    {"name": "researcher", ...},  # 包含搜索、解析、过滤
    {"name": "synthesizer", ...},  # 合成和总结
]

3. 明确的沟通接口

# 子代理使用标准化的输出格式
output_spec = """
你的输出必须是以下格式的 JSON：
{
    "success": bool,
    "summary": "简短总结（最多 200 字）",
    "details": ["详细项1", "详细项2", ...],
    "confidence": 0.0-1.0
}
"""

下一步

• 异步子代理
• 技能与扩展
• 生产部署