mirror of
https://github.com/aAAaqwq/AGI-Super-Team.git
synced 2026-09-14 16:33:39 +08:00
aAAaqwq
301 lines
13 KiB
Python
301 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Build the Codex C-suite leaf agents and packaged team contracts."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import tempfile
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
ADAPTER_PATH = Path("config/harness-adapters/codex.json")
|
||
PLUGIN_ROOT = Path("plugins/agi-super-team-codex")
|
||
CONTRACTS_PATH = PLUGIN_ROOT / "skills/c-suite-team/references/team-contracts.json"
|
||
KIT_REFERENCES_PATH = PLUGIN_ROOT / "skills/c-suite-team/references/kits"
|
||
HIERARCHY_PATH = Path("config/agent-hierarchy.json")
|
||
PACKAGED_HIERARCHY_PATH = PLUGIN_ROOT / "skills/c-suite-team/references/agent-hierarchy.json"
|
||
PACKAGED_ROUTES_PATH = PLUGIN_ROOT / "skills/c-suite-team/references/routes"
|
||
GENERATED_NOTICE = "Generated by scripts/build_codex_csuite_adapter.py; edit config/team-manifest.json instead."
|
||
|
||
|
||
def _json_string(value: str) -> str:
|
||
"""A JSON string is also a valid TOML basic string."""
|
||
|
||
return json.dumps(value, ensure_ascii=False)
|
||
|
||
|
||
def _agent_instructions(agent: dict[str, Any], registry: dict[str, Any] | None) -> str:
|
||
outputs = ", ".join(agent["outputs"])
|
||
if registry:
|
||
manager = agent["id"]
|
||
specialist_routes = [
|
||
f"ast-{manager}-{item['id']}: 当{item['trigger']} 不用于{item['doNotUseWhen']}"
|
||
for item in registry["specialists"]
|
||
]
|
||
role_routes = [
|
||
f"ast-{item['id']}: 当{item['trigger']} 不用于{item['doNotUseWhen']}"
|
||
for item in registry["roleRoutes"]
|
||
]
|
||
routes = " ".join(role_routes + specialist_routes)
|
||
execution = (
|
||
"你是受限管理节点，不是叶子。先识别工作对象、专业领域和交付阶段；只选择正向触发匹配且未命中排除条件的一个主责子 Agent，"
|
||
"仅当存在独立下游交付物时再增加一个协作角色。关键输入缺失或两个路由仍冲突时，先提出一个简短澄清问题，不得广播。"
|
||
f"你只能调用下列已安装的 ast-{manager}-* 角色和 canonical 角色引用，fork_turns 必须为 none，最多两个叶子并发，总深度不得超过二。"
|
||
"所有直属子 Agent 都是叶子，不得继续创建 Agent。路由表：" + routes + " "
|
||
)
|
||
else:
|
||
execution = "你是叶子 Agent，不得创建子 Agent 或协调其他叶子。"
|
||
return (
|
||
f"你是 AGI Super Team 的 {agent['name']}。职责重点：{agent['focus']} 标准交付物：{outputs}。"
|
||
f"职责边界：{agent['boundary']} 接受来自上级的单一、边界清楚的任务契约，只检查已提供的证据和获授权工作区，"
|
||
f"回传产物、检查、限制和下一步。{execution}"
|
||
"结构检查不能冒充运行验证。未经人类明确批准，不得发布、部署、交易、联系第三方、使用凭证或执行破坏性/不可逆操作；"
|
||
"不得提交、推送、合并或创建 PR。"
|
||
)
|
||
|
||
|
||
def _render_agent(agent: dict[str, Any], registry: dict[str, Any] | None) -> str:
|
||
agent_name = f"ast-{agent['id']}"
|
||
sandbox = "workspace-write" if agent["id"] in {"pe", "cco"} else "read-only"
|
||
kind = "受限管理节点" if registry else "叶子"
|
||
description = f"C-suite {agent['id'].upper()} {kind}：{agent['focus']}"
|
||
return "\n".join(
|
||
[
|
||
f"# {GENERATED_NOTICE}",
|
||
f"name = {_json_string(agent_name)}",
|
||
f"description = {_json_string(description)}",
|
||
f"nickname_candidates = [{_json_string(agent['id'].upper())}]",
|
||
'model_reasoning_effort = "high"',
|
||
f'sandbox_mode = "{sandbox}"',
|
||
f"developer_instructions = {_json_string(_agent_instructions(agent, registry))}",
|
||
"",
|
||
]
|
||
)
|
||
|
||
|
||
def _render_specialist(root: Path, manager: str, specialist: dict[str, Any]) -> str:
|
||
agent_name = f"ast-{manager}-{specialist['id']}"
|
||
description = f"{manager.upper()} 子专家｜{specialist['name']}：{specialist['trigger']}"
|
||
inputs = "、".join(specialist["requiredInputs"])
|
||
outputs = "、".join(specialist["outputs"])
|
||
acceptance = "；".join(specialist["acceptance"])
|
||
partners = ", ".join(specialist["partners"])
|
||
adaptation = (
|
||
f" Source role: {specialist['sourceRole']}. Adaptation: {specialist['adaptation']}"
|
||
if specialist.get("sourceRole")
|
||
else ""
|
||
)
|
||
upstream = (
|
||
root / "agents" / manager / "subagents" / specialist["id"] / "AGENTS.md"
|
||
).read_text(encoding="utf-8")
|
||
envelope = (
|
||
f"你是 {manager.upper()} 直属的{specialist['name']}叶子 Agent。"
|
||
f"调用条件：{specialist['trigger']}"
|
||
f"不要调用：{specialist['doNotUseWhen']}"
|
||
f"必需输入：{inputs}。缺少会改变结论的输入时，返回缺口，不得臆测。"
|
||
f"标准交付物：{outputs}。验收标准：{acceptance}。"
|
||
f"常见协作角色：{partners or '无'}；你只建议协作，不得自行调用。"
|
||
f"边界：{specialist['boundary']}"
|
||
f"上游来源：jnMetaCode/agency-agents-zh@{specialist['sourcePath']}。{adaptation} "
|
||
"只接受一个有明确目标、范围、输入、交付物和验收标准的任务包。"
|
||
"返回事实、假设、产物、检查、限制和下一步。你是叶子 Agent，不得创建子 Agent。"
|
||
"禁止登录未授权系统、使用凭证、抓取登录态数据、自动发布、部署、改基础设施、交易、付费或联系第三方。"
|
||
"不得保证未经验证的性能、安全、可靠性、排名、收入、收益、播放、粉丝、转化、税务处理、法律结论或 AI 引用；动态平台、市场、法律、财务、税务、技术和标准结论必须注明需要当前一手来源验证。"
|
||
"不要提交、推送、合并或创建 PR。"
|
||
)
|
||
instructions = f"{upstream}\n\n---\n\n# AGI Super Team 路由与安全信封\n\n{envelope}"
|
||
return "\n".join(
|
||
[
|
||
f"# Generated by scripts/build_codex_csuite_adapter.py; edit config/{manager}-specialists.json or the vendored source instead.",
|
||
f"name = {_json_string(agent_name)}",
|
||
f"description = {_json_string(description)}",
|
||
f"nickname_candidates = [{_json_string(specialist['id'])}]",
|
||
'model_reasoning_effort = "high"',
|
||
'sandbox_mode = "read-only"',
|
||
f"developer_instructions = {_json_string(instructions)}",
|
||
"",
|
||
]
|
||
)
|
||
|
||
|
||
def _packaged_contract(manifest: dict[str, Any]) -> dict[str, Any]:
|
||
agent_fields = ("id", "name", "focus", "outputs", "boundary")
|
||
kit_fields = (
|
||
"id",
|
||
"name",
|
||
"outcome",
|
||
"entrypoint",
|
||
"coordinator",
|
||
"reviewers",
|
||
"coreAgents",
|
||
"agents",
|
||
"outputs",
|
||
"checks",
|
||
)
|
||
packaged_kits = []
|
||
for kit in manifest["kits"]:
|
||
packaged_kit = {
|
||
field: kit[field] for field in kit_fields if field in kit
|
||
}
|
||
packaged_kit["entrypoint"] = f"references/kits/{kit['id']}.md"
|
||
packaged_kits.append(packaged_kit)
|
||
|
||
return {
|
||
"schemaVersion": manifest["schemaVersion"],
|
||
"generatedFrom": "config/team-manifest.json",
|
||
"runtimeEvidence": "pending",
|
||
"agents": [
|
||
{field: agent[field] for field in agent_fields if field in agent}
|
||
for agent in manifest["agents"]
|
||
],
|
||
"kits": packaged_kits,
|
||
}
|
||
|
||
|
||
def build_outputs(root: Path) -> dict[Path, str]:
|
||
manifest = json.loads(
|
||
(root / "config/team-manifest.json").read_text(encoding="utf-8")
|
||
)
|
||
hierarchy = json.loads((root / HIERARCHY_PATH).read_text(encoding="utf-8"))
|
||
registries = {
|
||
manager: json.loads((root / settings["routingFile"]).read_text(encoding="utf-8"))
|
||
for manager, settings in hierarchy["managers"].items()
|
||
}
|
||
leaf_agents = [agent for agent in manifest["agents"] if agent["id"] != "ceo"]
|
||
adapter = {
|
||
"$schema": "./codex.schema.json",
|
||
"schemaVersion": 1,
|
||
"harness": "codex",
|
||
"coordinator": "ceo",
|
||
"independentReviewer": "governor",
|
||
"maxParallelLeaves": 3,
|
||
"fallbackModes": ["native", "manual", "sequential"],
|
||
"runtimeEvidence": "pending",
|
||
"agentMap": {
|
||
agent["id"]: f"ast-{agent['id']}" for agent in leaf_agents
|
||
},
|
||
"requiredMaxDepth": hierarchy["requiredMaxDepth"],
|
||
"managerAgentMap": {
|
||
manager: {
|
||
"agent": f"ast-{manager}",
|
||
"requiredMaxDepth": hierarchy["requiredMaxDepth"],
|
||
"maxConcurrentChildren": settings["maxConcurrentChildren"],
|
||
"fallbackMode": "parent-flat",
|
||
"delegates": {
|
||
item["id"]: f"ast-{manager}-{item['id']}"
|
||
for item in registries[manager]["specialists"]
|
||
},
|
||
"roleRefs": {
|
||
role_id: f"ast-{role_id}" for role_id in settings["roleRefs"]
|
||
},
|
||
}
|
||
for manager, settings in hierarchy["managers"].items()
|
||
},
|
||
}
|
||
outputs = {
|
||
root / ADAPTER_PATH: json.dumps(adapter, indent=2, ensure_ascii=False) + "\n",
|
||
root / CONTRACTS_PATH: json.dumps(
|
||
_packaged_contract(manifest), indent=2, ensure_ascii=False
|
||
)
|
||
+ "\n",
|
||
root / PACKAGED_HIERARCHY_PATH: json.dumps(
|
||
hierarchy, indent=2, ensure_ascii=False
|
||
)
|
||
+ "\n",
|
||
}
|
||
for manager, registry in registries.items():
|
||
outputs[root / PACKAGED_ROUTES_PATH / f"{manager}.json"] = (
|
||
json.dumps(registry, indent=2, ensure_ascii=False) + "\n"
|
||
)
|
||
for agent in leaf_agents:
|
||
outputs[
|
||
root / PLUGIN_ROOT / "payload/agents" / f"ast-{agent['id']}.toml"
|
||
] = _render_agent(agent, registries.get(agent["id"]))
|
||
for manager, registry in registries.items():
|
||
for specialist in registry["specialists"]:
|
||
outputs[
|
||
root
|
||
/ PLUGIN_ROOT
|
||
/ "payload/agents"
|
||
/ f"ast-{manager}-{specialist['id']}.toml"
|
||
] = _render_specialist(root, manager, specialist)
|
||
for kit in manifest["kits"]:
|
||
outputs[root / KIT_REFERENCES_PATH / f"{kit['id']}.md"] = (
|
||
root / kit["entrypoint"]
|
||
).read_text(encoding="utf-8")
|
||
return outputs
|
||
|
||
|
||
def _atomic_write(path: Path, content: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.NamedTemporaryFile(
|
||
"w", encoding="utf-8", dir=path.parent, delete=False
|
||
) as handle:
|
||
handle.write(content)
|
||
temporary = Path(handle.name)
|
||
temporary.replace(path)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
|
||
parser.add_argument("--check", action="store_true")
|
||
arguments = parser.parse_args()
|
||
root = arguments.root.resolve()
|
||
outputs = build_outputs(root)
|
||
stale = [
|
||
path
|
||
for path, content in outputs.items()
|
||
if path.is_symlink()
|
||
or not path.is_file()
|
||
or path.read_text(encoding="utf-8") != content
|
||
]
|
||
expected_agents = {
|
||
path.absolute()
|
||
for path in outputs
|
||
if path.parent == (root / PLUGIN_ROOT / "payload/agents").absolute()
|
||
}
|
||
extra_agents = sorted(
|
||
path
|
||
for path in (root / PLUGIN_ROOT / "payload/agents").glob("ast-*.toml")
|
||
if path.absolute() not in expected_agents
|
||
)
|
||
expected_runbooks = {
|
||
path.absolute()
|
||
for path in outputs
|
||
if path.parent == (root / KIT_REFERENCES_PATH).absolute()
|
||
}
|
||
extra_runbooks = sorted(
|
||
path
|
||
for path in (root / KIT_REFERENCES_PATH).glob("*.md")
|
||
if path.absolute() not in expected_runbooks
|
||
)
|
||
if arguments.check:
|
||
for path in stale:
|
||
print(f"Codex C-suite adapter is stale: {path.relative_to(root)}")
|
||
for path in extra_agents:
|
||
print(f"Unexpected generated C-suite agent: {path.relative_to(root)}")
|
||
for path in extra_runbooks:
|
||
print(f"Unexpected generated Kit runbook: {path.relative_to(root)}")
|
||
return 1 if stale or extra_agents or extra_runbooks else 0
|
||
for path in stale:
|
||
_atomic_write(path, outputs[path])
|
||
print(f"Wrote {path.relative_to(root)}")
|
||
if extra_agents:
|
||
raise RuntimeError(
|
||
"remove obsolete generated agents explicitly: "
|
||
+ ", ".join(str(path.relative_to(root)) for path in extra_agents)
|
||
)
|
||
if extra_runbooks:
|
||
raise RuntimeError(
|
||
"remove obsolete generated Kit runbooks explicitly: "
|
||
+ ", ".join(str(path.relative_to(root)) for path in extra_runbooks)
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|