🤖🤖🤖 OpenClaw多智能体协作实战

🎯 为什么需要多智能体？

🚀 sessions_spawn详解

# 核心参数
sessions_spawn(
    task="任务描述",           # 必需：给子Agent的指令
    runtime="subagent",       # 运行环境：subagent | acp
    mode="run",               # 执行模式：run(一次性) | session(持久)
    
    # 可选参数
    agentId="coder",          # 指定Agent（需runtime="acp"）
    model="claude-3-5",       # 模型覆盖
    thinking="on",            # 推理模式
    timeoutSeconds=300,       # 超时时间
    thread=True,              # Discord线程绑定
    
    # 文件附件
    attachments=[{
        "name": "code.py",
        "content": "...",
        "mimeType": "text/x-python"
    }],
    
    # 启动后自动attach
    attachAs={"mountPath": "/data"}
)

mode选择指南

• mode="run"：一次性任务，执行完自动清理。适合：数据采集、代码生成、分析任务
• mode="session"：持久会话，可多次交互。适合：长期项目、编码会话、调试任务

🏗️ 设计模式：Agent团队

# Director Agent
class Orchestrator:
    def execute_complex_task(self, task):
        # 1. 拆解任务
        subtasks = self.break_down(task)
        
        # 2. 并行派发
        workers = []
        for subtask in subtasks:
            worker = sessions_spawn(
                task=subtask.description,
                mode="run",
                runtime="subagent"
            )
            workers.append(worker)
        
        # 3. 收集结果
        results = [w.result for w in workers]
        
        # 4. 汇总输出
        return self.synthesize(results)

# Content Pipeline
class ContentPipeline:
    def create_article(self, topic):
        # Step 1: Researcher收集资料
        research = sessions_spawn(
            task=f"搜索{topic}的相关资料，整理关键信息",
            mode="run"
        )
        
        # Step 2: Writer撰写文章
        draft = sessions_spawn(
            task=f"基于以下资料撰写文章：{research.output}",
            mode="run"
        )
        
        # Step 3: Editor润色
        final = sessions_spawn(
            task=f"编辑以下文章，提升可读性：{draft.output}",
            mode="run"
        )
        
        return final.output

# Voting for accuracy
def consensus_answer(question, n_agents=3):
    answers = []
    
    for i in range(n_agents):
        agent = sessions_spawn(
            task=f"回答这个问题：{question}",
            mode="run",
            temperature=0.7  # 增加随机性
        )
        answers.append(agent.output)
    
    # 聚合答案
    return majority_vote(answers)

# Adversarial Debate
def debate(topic, rounds=3):
    pro_agent = sessions_spawn(
        task=f"你是正方，论证'{topic}'的优势",
        mode="session"
    )
    
    con_agent = sessions_spawn(
        task=f"你是反方，论证'{topic}'的劣势",
        mode="session"
    )
    
    for round in range(rounds):
        pro_arg = pro_agent.send(f"反驳反方的观点")
        con_arg = con_agent.send(f"反驳正方的观点：{pro_arg}")
    
    # 由Judge Agent综合双方观点
    judge = sessions_spawn(
        task=f"基于以下辩论给出客观结论：{transcript}",
        mode="run"
    )
    
    return judge.output

📊 状态监控：subagents工具

# 列出当前请求者的所有子Agent
subagents(action="list")

# 查看最近运行的子Agent
subagents(action="list", recentMinutes=60)

# 终止特定子Agent
subagents(action="kill", target="session-id")

# 向子Agent发送指令
subagents(action="steer", target="session-id", message="加快进度")

🔧 实战：SEO内容工厂

# 每日批量生成SEO页面的多Agent架构
class SEOContentFactory:
    def generate_daily_pages(self, topics):
        results = []
        
        for topic in topics:
            # 每个主题启动一个独立Agent
            worker = sessions_spawn(
                task=f"""
                生成'{topic}'的SEO教程页面：
                1. 搜索相关信息
                2. 撰写包含h1-h3的HTML
                3. 添加代码示例和最佳实践
                4. 保存到 /var/www/miaoquai/tools/{kebab_case(topic)}.html
                5. 返回文件路径和字数统计
                """,
                mode="run",
                timeoutSeconds=300
            )
            results.append(worker)
        
        # 等待全部完成
        for r in results:
            print(f"完成: {r.result['file']} ({r.result['words']}字)")
        
        return results

# 使用示例
factory = SEOContentFactory()
topics = [
    "OpenClaw定时任务配置",
    "Agent Skills工作流设计",
    "MCP协议集成指南",
    "多智能体协作模式"
]
factory.generate_daily_pages(topics)
# 4个Agent并行执行，总耗时 ~5分钟 vs 串行20分钟

⚡ 高级：ACP harness集成

对于复杂编码任务，可以接入专业AI编码Agent：

# 接入Codex/Claude Code/Cursor等ACP harness
sessions_spawn(
    task="实现一个MCP Server，支持GitHub API调用",
    runtime="acp",           # 使用ACP运行时
    agentId="codex",         # 指定Codex Agent
    mode="session",          # 需要多轮交互
    thread=True              # Discord线程绑定
)

# ACP harness优势：
# - 专业编码能力
# - 长期会话保持
# - 文件系统访问
# - 终端执行能力