150 lines
4.8 KiB
Python
150 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
小唯看板中枢 - 任务自动路由与执行监督
|
||
"""
|
||
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
KANTAB_DIR = Path.home() / ".hermes" / "kanban"
|
||
ROUTES_FILE = Path.home() / ".hermes" / "scripts" / "kanban-route.py"
|
||
|
||
# 可用 worker profile
|
||
WORKER_PROFILES = ["opencode", "dsh", "research"]
|
||
|
||
def get_profile_description(profile: str) -> str:
|
||
"""获取 profile 描述"""
|
||
descriptions = {
|
||
"default": "小唯本体（orchestrator + 通用任务）",
|
||
"opencode": "代码执行专用（编码、插件、安装部署）",
|
||
"dsh": "DSH研究专用（本地模型、DeepSeek Harness测试）",
|
||
"research": "研究分析专用（资料搜集、竞品分析、报告）",
|
||
}
|
||
return descriptions.get(profile, profile)
|
||
|
||
def dispatch_task(task_desc: str, manual_assignee: str = None) -> dict:
|
||
"""分发任务到合适的 worker"""
|
||
assignee = manual_assignee or detect_profile(task_desc)
|
||
|
||
# 调用路由脚本
|
||
cmd = ["python3", str(ROUTES_FILE), task_desc, "--assignee", assignee, "--json"]
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
||
if result.returncode == 0:
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except:
|
||
pass
|
||
|
||
# 降级：直接创建
|
||
return create_basic_task(task_desc, assignee)
|
||
|
||
def create_basic_task(title: str, assignee: str) -> dict:
|
||
"""创建基础看板任务"""
|
||
cmd = [
|
||
"hermes", "kanban", "create",
|
||
"--tenant", "xiaowei",
|
||
"--assignee", assignee,
|
||
"--json",
|
||
title,
|
||
]
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except:
|
||
pass
|
||
|
||
return {"error": result.stderr}
|
||
|
||
def detect_profile(text: str) -> str:
|
||
"""检测任务适合的 profile"""
|
||
text_lower = text.lower()
|
||
|
||
# 编码任务
|
||
if any(k in text_lower for k in ["代码", "实现", "写", "改", "修", "bug", "opencode", "API", "插件", "安装", "部署", "审查", "review"]):
|
||
return "opencode"
|
||
|
||
# DSH/模型
|
||
if any(k in text_lower for k in ["DSH", "dsh", "deepseek", "模型测试", "本地模型", "本地推理"]):
|
||
return "dsh"
|
||
|
||
# 研究
|
||
if any(k in text_lower for k in ["调研", "分析", "竞品", "对比", "报告", "选型", "方案", "设计"]):
|
||
return "research"
|
||
|
||
# 默认
|
||
return "research"
|
||
|
||
def dispatch_swarm(goal: str, workers: list) -> list:
|
||
"""启动 Swarm 并行执行"""
|
||
cmd = ["python3", str(ROUTES_FILE), goal, "--swarm"] + workers
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
||
if result.returncode == 0:
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except:
|
||
pass
|
||
|
||
print(f"❌ Swarm 启动失败: {result.stderr}")
|
||
return []
|
||
|
||
def list_tasks(status: str = None) -> list:
|
||
"""列出看板任务"""
|
||
cmd = ["hermes", "kanban", "list", "--json"]
|
||
if status:
|
||
cmd.extend(["--status", status])
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except:
|
||
pass
|
||
|
||
return []
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="小唯看板中枢")
|
||
parser.add_argument("action", choices=["dispatch", "swarm", "list", "status"], help="动作")
|
||
parser.add_argument("task", nargs="?", help="任务描述")
|
||
parser.add_argument("--assignee", help="手动指定 profile")
|
||
parser.add_argument("--worker", action="append", help="Swarm worker (repeatable)")
|
||
parser.add_argument("--status", help="任务状态过滤")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.action == "dispatch":
|
||
if not args.task:
|
||
print("❌ 需要提供任务描述")
|
||
sys.exit(1)
|
||
result = dispatch_task(args.task, args.assignee)
|
||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||
|
||
elif args.action == "swarm":
|
||
if not args.task:
|
||
print("❌ 需要提供目标描述")
|
||
sys.exit(1)
|
||
if not args.worker:
|
||
print("❌ 需要提供至少一个 worker")
|
||
sys.exit(1)
|
||
result = dispatch_swarm(args.task, args.worker)
|
||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||
|
||
elif args.action == "list":
|
||
result = list_tasks(args.status)
|
||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||
|
||
elif args.action == "status":
|
||
profiles = subprocess.run(["hermes", "profile", "list"], capture_output=True, text=True)
|
||
print(profiles.stdout)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|