#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ swarm_orchestrator.py — A1 多 Agent 动态组队层（吸收 openJiuwen Leader-Teammate 蜂群范式） ======================================================================================== 在火斗云智 AIOS 的 AxiomGate 裁决调度内核之上，新增"组队编排运行时"： - Leader 负责 规划 / 分工 / 管控，Teammate 承接子任务 - 双模支持：fixed（固定流程）与 dynamic（动态自主组队） - 任务冲突 / 资源竞争 统一经 AxiomGate 仲裁、配额节流、熔断 - 每步落 checkpoint，支持断点续跑；结果经记忆网关可回流反哺 设计原则：集中裁决（AxiomGate 保稳定）+ 动态组队（吸收 openJiuwen 补协同短板）。 不修改现有 AxiomGate / brain_scheduler 内核，仅作为上层编排器调用其接口。 """ import json import os import time import uuid from datetime import datetime # ---- 依赖体系接口（按需导入，缺失时降级为直连后端） ---- try: from axiom_gate import route, check_circuit, submit, save_checkpoint, load_checkpoint except Exception: # 独立运行时不强依赖 route = check_circuit = submit = save_checkpoint = load_checkpoint = None try: from brain_scheduler import call_local, call_external, classify except Exception: call_local = call_external = classify = None ANCHOR = "Ω₀⊂⊙∞⊂Ω" DID = "DID-BR-000002" SWARM_DB = os.environ.get("SWARM_DB", "/var/lib/zr-swarm/swarm_teams.json") SWARM_DIR = os.path.dirname(SWARM_DB) or "." # ---------------------------------------------------------------- 基础工具 def now(): return datetime.now().isoformat(timespec="seconds") def _load_json(path, default): try: if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return default def _save_json(path, obj): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False, indent=2) os.replace(tmp, path) # ---------------------------------------------------------------- 执行后端 # 候选执行后端（按优先级依次尝试，缺失/失败则降级到下一项） _EXEC_BACKENDS = [ ("axiomgate.axiomgate_worker", "execute_action"), ("brain_scheduler", "call_external"), ("brain_scheduler", "call_local"), ] def _invoke(agent, subtask): """Teammate 执行单个子任务。按候选后端顺序尝试真实执行通道，全部缺失则降级占位。""" prompt = subtask.get("prompt", subtask.get("title", "")) out = None for mod, fn in _EXEC_BACKENDS: try: m = __import__(mod, fromlist=[fn]) f = getattr(m, fn) out = f(prompt) if mod == "brain_scheduler" else f(prompt, None) if out is not None: break except Exception: out = None if out is None: # 兜底：返回待执行标记（对接外部 worker 池时在此扩展） return {"status": "queued", "agent": agent, "subtask": subtask.get("title")} # 规范化返回值：tuple / 非 dict → dict if isinstance(out, tuple): out = {"status": "ok", "content": str(out[1]) if len(out) > 1 else str(out[0])} elif not isinstance(out, dict): out = {"status": "ok", "content": str(out)} out.setdefault("status", "ok") out.setdefault("agent", agent) return out def _guard(agent, pool="default"): """冲突仲裁：熔断 + 配额节流检查（经 AxiomGate）。""" if check_circuit is not None: try: status = check_circuit(pool) if isinstance(status, dict) and status.get("open"): return {"ok": False, "reason": f"circuit_open:{pool}", "agent": agent} except Exception: pass return {"ok": True, "agent": agent} # ---------------------------------------------------------------- 编排器 class TeamSwarm: """Leader-Teammate 组队编排运行时。""" def __init__(self, name="swarm-default", mode="dynamic"): self.name = name self.mode = mode # fixed | dynamic self.teams = _load_json(SWARM_DB, {"teams": {}}) self.teams.setdefault("teams", {}) # ---- 组队 ---- def register_agent(self, agent_id, role="teammate", capability=None, pool="default"): """注册一个可参与组队的 Agent（Teammate 或 Leader 候选人）。""" team = self.teams["teams"].setdefault(self.name, { "leader": None, "teammates": {}, "created": now(), "mode": self.mode }) if role == "leader": team["leader"] = {"id": agent_id, "capability": capability, "pool": pool} else: team["teammates"][agent_id] = { "role": "teammate", "capability": capability, "pool": pool, "load": 0, "fails": 0, "registered": now() } _save_json(SWARM_DB, self.teams) return {"ok": True, "team": self.name, "agent": agent_id, "role": role} def _teammate_pool(self): team = self.teams["teams"].get(self.name, {}) return team.get("teammates", {}) # ---- Leader 规划 ---- def plan(self, task): """Leader 把任务拆解为子任务清单。dynamic 模式按内容动态分工。""" title = task.get("title", str(task)[:60]) if self.mode == "fixed": # 固定流程：按 task["steps"] 顺序执行 steps = task.get("steps") or [title] return [{"seq": i, "title": s, "prompt": task.get("prompt", s)} for i, s in enumerate(steps)] # dynamic：Leader 依据任务复杂度/关键词动态决定分工粒度和角色 text = str(task.get("prompt", title)) complexity = len(text) # 规则化拆解：> 60 字或含多阶段关键词则拆成多子任务 import re if complexity > 60 or re.search(r"并且|然后|同时|先.*再|步骤|阶段|and|then", text): return [ {"seq": 0, "title": "分析·目标拆解", "prompt": f"明确任务目标与验收：{text}"}, {"seq": 1, "title": "执行·主任务", "prompt": text}, {"seq": 2, "title": "校验·结果汇总", "prompt": f"对上述结果做完整性校验并汇总：{text}"}, ] return [{"seq": 0, "title": "执行", "prompt": text}] # ---- 分工 ---- def assign(self, subtask): """把子任务分配给负载最低、能力匹配的 Teammate（动态协商式）。""" pool = self._teammate_pool() if not pool: return None # 简单负载均衡：选 load 最小者 best = min(pool.items(), key=lambda kv: (kv[1]["load"], kv[1]["fails"])) aid, meta = best meta["load"] += 1 return aid # ---- 执行 ---- def execute(self, task, callback=None): """编排执行整条任务：规划→分工→（仲裁）执行→聚合。""" plan = self.plan(task) results = [] for sub in plan: guard = _guard(sub.get("title", "")) if not guard.get("ok"): results.append({"seq": sub["seq"], "title": sub["title"], "status": "blocked", "reason": guard.get("reason")}) continue agent = self.assign(sub) out = _invoke(agent, sub) out["seq"] = sub["seq"] out["agent"] = agent results.append(out) if callback: try: callback(sub, out) except Exception: pass if save_checkpoint is not None: try: save_checkpoint(f"swarm-{self.name}-{int(time.time())}", sub["seq"], {"sub": sub, "out": out}) except Exception: pass # 聚合 return { "team": self.name, "mode": self.mode, "tasks": len(plan), "results": results, "summary": { "ok": sum(1 for r in results if r.get("status") == "ok"), "queued": sum(1 for r in results if r.get("status") == "queued"), "blocked": sum(1 for r in results if r.get("status") == "blocked"), "error": sum(1 for r in results if r.get("status") == "error"), }, "did": DID, "anchor": ANCHOR, "ts": now() } # ---- 状态 ---- def status(self): return { "team": self.name, "mode": self.mode, "leader": self.teams["teams"].get(self.name, {}).get("leader"), "teammates": list(self._teammate_pool().keys()), "count": len(self._teammate_pool()), "anchor": ANCHOR } # ---------------------------------------------------------------- CLI def _demo(): print("== A1 动态组队层 demo ==") swarm = TeamSwarm("demo-swarm", mode="dynamic") swarm.register_agent("agent-qa", role="leader", capability="plan") for i in range(3): swarm.register_agent(f"teammate-{i+1}", role="teammate", capability="execute") print("组队:", json.dumps(swarm.status(), ensure_ascii=False)) task = {"title": "市场竞品分析", "prompt": "分析 openJiuwen 与 AutoGPT 的多智能体协同架构差异，并且给出优劣势对比结论"} res = swarm.execute(task) print("规划子任务数:", res["tasks"]) print("执行结果: ok=", res["summary"]["ok"], " blocked=", res["summary"]["blocked"]) return res if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "status": swarm = TeamSwarm(sys.argv[2] if len(sys.argv) > 2 else "swarm-default") print(json.dumps(swarm.status(), ensure_ascii=False, indent=2)) else: _demo()