Agent Dispatch Skill

版本: v1.0

创建日期: 2026-03-26

*作�?: 象腿 (main agent)

*用�?: 根据routing规则自动调度specialist agents

🎯 核心功能

Agent Dispatch是main agent的核心协调skill，负责：

1. 意图识别: 分析用户请求，提取关键特�?2. 路由匹配: 根据routing规则选择最合适的agent
2. 任务调度: 使用sessions_spawn调度specialist agent
3. 结果整合: 合并agent执行结果，形成连贯回�?5. 容错处理: 失败时自动重试或fallback到main

📋 Routing规则

规则优先�?

路由匹配逻辑

1. *优先级匹�?: 按priority顺序从低到高匹配
2. 正则表达�?: pattern使用正则表达式匹配用户输�?3. 首次命中*: 匹配到第一个rule即停止，返回target
3. 默认兜底: 如果都不匹配，fallback�?self"

🔄 调度流程

步骤1: 意图识别

步骤2: 路由匹配

def match_route(intent):
    """
    根据intent匹配routing规则

    Returns:
        dict: {
            "target": "coder|danao|writer|engineer|manager|self",
            "priority": 1-99,
            "confidence": 0.0-1.0
        }
    """
    for rule in routing_rules:
        pattern = rule["pattern"]
        if re.search(pattern, intent["user_input"]):
            return {
                "target": rule["target"],
                "priority": rule["priority"],
                "confidence": calculate_confidence(intent, rule)
            }

    # 默认fallback到self
    return {"target": "self", "priority": 99, "confidence": 0.5}

步骤3: 任务调度

步骤4: 结果整合

步骤5: 容错处理

🛠�?实现细节

sessions_spawn参数配置

# Coder Agent
coder:
  runtime: "acp"
  agentId: "codex"
  mode: "run"
  timeout: 300

# Danao Agent
danao:
  runtime: "subagent"
  agentId: "danao"
  mode: "session"
  timeout: 600

# Writer Agent
writer:
  runtime: "subagent"
  agentId: "writer"
  mode: "session"
  timeout: 600

# Engineer Agent
engineer:
  runtime: "subagent"
  agentId: "engineer"
  mode: "session"
  timeout: 600

# Manager Agent
manager:
  runtime: "subagent"
  agentId: "manager"
  mode: "session"
  timeout: 300

并行任务处理

def parallel_dispatch(tasks):
    """
    并行调度多个独立任务

    Args:
        tasks: [
            {"route": route1, "intent": intent1, "user_input": input1},
            {"route": route2, "intent": intent2, "user_input": input2}
        ]

    Returns:
        list: [result1, result2]
    """
    MAX_PARALLEL = 2

    # 限制并行数量
    tasks = tasks[:MAX_PARALLEL]

    # 并行执行
    results = []
    with ThreadPoolExecutor(max_workers=MAX_PARALLEL) as executor:
        futures = [
            executor.submit(dispatch_task, task["route"], task["intent"], task["user_input"])
            for task in tasks
        ]

        for future in as_completed(futures):
            results.append(future.result())

    return results

📊 性能监控

关键指标

日志记录

def log_dispatch(route, intent, result):
    """
    记录调度日志

    Format:
    [2026-03-26 12:00:00] [DISPATCH] target=coder, priority=1, confidence=0.95, success=true, time=3.45s
    """
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "type": "DISPATCH",
        "target": route["target"],
        "priority": route["priority"],
        "confidence": route["confidence"],
        "success": result["success"],
        "execution_time": result["metadata"]["execution_time"],
        "retry_count": result["metadata"]["retry_count"]
    }

    write_log(log_entry)

🎓 使用示例

示例1: 编程任务

示例2: 内容创作

# 用户输入
user_input = "写一篇关于AI工具推荐的小红书文案"

# 意图识别
intent = identify_intent(user_input)
# => {"task_type": "writing", "complexity": "medium"}

# 路由匹配
route = match_route(intent)
# => {"target": "writer", "priority": 3, "confidence": 0.92}

# 任务调度
result = dispatch_task(route, intent, user_input)

# 结果整合
response = integrate_results(result)
# => "【Writer】已处理完成（耗时15.23秒）\n\n小红书文案已生成..."

示例3: 并行任务

⚙️ 配置文件

agent-dispatch-config.json

🚀 未来优化

短期 (1-2�?

• [ ] 添加机器学习模型提升意图识别准确�?- [ ] 实现动态routing规则（基于历史数据优化）
• [ ] 添加agent性能评分，自动选择最优agent

中期 (1个月)

• [ ] 实现agent负载均衡
• [ ] 添加任务队列管理
• [ ] 实现跨agent知识共享

长期 (3个月)

• [ ] 引入强化学习优化routing策略
• [ ] 实现自适应并行度调�?- [ ] 构建agent性能预测模型

Skill版本: v1.0

最后更�? 2026-03-26

维护�? 象腿 (main agent)