kimi-k3-subagent

Kimi K3 Subagent 架构

Kimi K3 (MoonshotAI) 的子Agent调度架构，蒸馏为Python实现。

架构

SubagentHost (生命周期管理)
  ├── spawn()    — 创建新子Agent
  ├── resume()   — 恢复已有子Agent(保留上下文)
  ├── retry()    — 重试失败子Agent
  └── runQueued()— 批量执行

SubagentBatch (两阶段调度引擎)
  ├── 正常阶段: 5个立即启动, 之后每700ms启动1个
  └── 限流阶段: 指数退避(3s/6s/12s/24s) + 容量收缩/恢复(3分钟)

SwarmEventEmitter (生命周期事件)
  ├── spawned / started / completed / failed / suspended / batch_done
  └── on() / off() / emit() 监听机制

SummaryCheck (结果质量控制)
  └── 子Agent结果 ≥ 200字符, 不足自动补全

内置子Agent类型

使用方法

1. 批量并行任务

from scripts.lib.agent import SubagentHost, SubagentBatch, QueuedSubagentTask

host = SubagentHost(session, "main_agent")

tasks = [
    QueuedSubagentTask(
        data={"code": "600519"},
        prompt="分析贵州茅台资金流向",
        profile_name="explore",
    ),
    QueuedSubagentTask(
        data={"code": "000858"},
        prompt="分析五粮液资金流向",
        profile_name="explore",
    ),
]
results = await host.run_queued(tasks)

2. 生命周期管理

# 创建子Agent
handle = await host.spawn("coder", "扫描板块资金流")

# 恢复已有子Agent
handle = await host.resume(agent_id, "继续之前的分析")

# 重试失败
handle = await host.retry(agent_id, "重试板块扫描")

3. 事件监听

from scripts.swarm import SwarmEventType

scheduler = SwarmScheduler(executor)

def on_completed(event_type, data):
    print(f"Task {data['task_id']} completed in {data['result_len']} chars")

scheduler.on(SwarmEventType.COMPLETED, on_completed)
results = await scheduler.run(tasks)

4. 结果摘要检查

子Agent结果自动确保≥200字符, 不足时自动补全:

from scripts.swarm import _ensure_min_summary

result = _ensure_min_summary(short_result)
# 不足200字符自动追加说明

限流策略

正常阶段 → 遇到RateLimit → 限流阶段
  ├── 容量收缩: 每次限流-1, 最小1
  ├── 重试间隔: 3s/6s/12s/24s... 指数退避
  └── 容量恢复: 3分钟无新限流则+1

文件清单