games-development-ai/tier2/gen-worker/worker/agent_loop/design_team.py

"""agent_loop/design_team.py —— tier2 阶段 1「工作室 Agent Team 星形多 agent 设计」(on AgentScope v2.0.2)。

这是直攻 tier2 头号过门杠杆(spike 期 ~33% 过门率)的设计阶段升级:把原来的「单 design agent 单轮发散」
换成「一个 leader 把题面拆成四面设计子任务 → 四位专家 worker 分头设计 → leader 汇总成一份连贯设计稿」。
设计阶段越充分,产出的富游戏规格越完整,阶段 2 单写就越容易把门逼绿。

设计依据:图说 C1(单写者只指阶段 2,阶段 1 仍多 agent 发散)、E3(阶段 1 工作室 Agent Team 星形)。

落地范式(AgentScope 2.0.2 源码核验,务必看清这条——图说里的 AgentCreate/TeamSay 是设计口径、不是纯库 API):
  - 2.0.2 删了进程内 MsgHub / pipeline 静态编排(`src/agentscope/agent/__init__.py` 只导出 `Agent`);
    部署态的 AgentCreate / TeamCreate / TeamSay 四件套**只在 app 服务层(create_app)可用**
    (见 .agents/knowledge/agentscope-2.0-facts.md「多 agent 只在 app 服务层」)。
  - 本线是纯库 `import Agent` 跑脚本(studio.py 无 create_app),拿不到部署态 Team 原语。故星形落成
    **worker-as-tool**:把每个专家 worker 跑一次设计 reply 包成 leader 的一个 `FunctionTool`,leader 在
    ReAct 循环里自己调度。这与 E3 的星形语义等价——leader 是唯一中心(拆解 + 汇总),专家之间不互通、
    各自只把本域结论回给 leader(因为专家工具彼此独立、leader 串起所有协调)。
  - 这是设计意图(E3 星形)与 2.0.2 纯库现实之间唯一最小、不拼接的落地:零 app 服务层、零新依赖,
    复用 toolkit.py 已验证的 `FunctionTool(func)` + `Toolkit(tools=[...])` 范式。

框架接缝(2.0.2,与 studio.py 同源,全兼容):
  - `from agentscope.agent import Agent, ReActConfig` / `from agentscope.message import UserMsg`;
  - `from agentscope.tool import Toolkit, FunctionTool`(worker-as-tool 的包壳);
  - leader 调专家工具需 BYPASS 权限(FunctionTool 默认 check_permissions 返回 ASK,无人值守会卡死——
    见 _adapters.py:99;复用 studio._bypass_state 的同款 AgentState(BYPASS) 注入)。

爆炸半径:本模块产出 = 一份**连贯设计稿 str**,形状与原 `_design_stage` 的 `str` 完全一致(只是内容更丰富),
  阶段 2 单写消费接口(roles.writer_system 的 design_text 形参)零改;失败/超预算一律 degrade 回单 agent。
"""

import asyncio

# 框架接缝(2.0.2):Agent + ReActConfig + UserMsg + Toolkit + FunctionTool + BYPASS 权限。
# 包内/直跑兼容导入(与 studio.py 同款兜底:直跑时把 gen-worker/ 加进 sys.path,使顶层包 worker 可解析)。
try:
    from .. import roles, genconfig
except ImportError:  # pragma: no cover —— 直接 python 跑兜底
    import sys
    from pathlib import Path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from worker import roles, genconfig  # type: ignore

from agentscope.agent import Agent, ReActConfig
from agentscope.message import UserMsg
from agentscope.tool import Toolkit, FunctionTool
from agentscope.state import AgentState
from agentscope.permission import PermissionContext, PermissionMode


def _user_msg(text: str) -> UserMsg:
    """2.0.1+:UserMsg 必须带 name;字符串 content 会被包成 TextBlock。"""
    return UserMsg(name="user", content=text)


def _text_of(resp) -> str:
    """从 AgentScope Msg 抽纯文本(content 是 block 列表,兼容 dict/对象;与 studio.text_of 同口径)。"""
    blocks = getattr(resp, "content", None)
    if blocks is None and hasattr(resp, "get"):
        blocks = resp.get("content")
    out = []
    for b in (blocks or []):
        if isinstance(b, dict):
            if b.get("type") == "text":
                out.append(b.get("text", "") or "")
        elif getattr(b, "type", None) == "text":
            out.append(getattr(b, "text", "") or "")
    return "".join(out)


def _bypass_state() -> AgentState:
    """造一个权限=BYPASS 的 AgentState(与 studio._bypass_state 同款),使 leader 调专家工具不卡在 ASK。

    FunctionTool 默认 check_permissions 返回 ASK(_adapters.py:99),无人值守会卡死;BYPASS 是沙箱/
    无人值守档位(permission/_types.py),跳过工具 ASK。本线运行在受控生成环境(专家工具只调本地模型、
    不碰文件/子进程),适用 BYPASS。
    """
    return AgentState(permission_context=PermissionContext(mode=PermissionMode.BYPASS))


class DesignTeamError(Exception):
    """工作室设计团队失败信号(模型超时/预算耗尽/leader 未产出有效设计稿)。

    studio 的 _design_stage 接住它 → degrade 回落原单 agent 单轮设计,绝不让设计阶段中断主链。

    kind:失败大类(供 studio 采集编排层 infra_flags 时区分);'timeout' = 整团队墙钟硬超时(降级归
      wall_timeout),其余(预算耗尽 / leader 空产出 / 团队异常)= None(只归 design_team_degraded)。
    """

    def __init__(self, message: str, *, kind: str | None = None) -> None:
        self.kind = kind
        super().__init__(message)


# ── 四面专家定义:(工具名, 工具中文描述, system prompt)──────────────────────────
#   工具名/描述会被 AgentScope 从 FunctionTool 抽成 leader 可见的工具 schema(描述即 leader 选工具的依据)。
#   工具名与 roles.design_leader_system 里列的四个工具名严格对齐(leader 提示词按这四个名字派活)。
_EXPERTS = [
    ("design_systems",
     "系统设计师:产出核心循环、多系统拆解与耦合点、可观测语义 state。富游戏系统骨架这一面。",
     roles.DESIGN_SYSTEMS_DESIGNER),
    ("design_economy",
     "经济数值设计师:产出数据表(条目+示例值)、经济流转、胜负条件(含失败态阈值)、难度曲线。",
     roles.DESIGN_ECONOMY_DESIGNER),
    ("design_level",
     "玩法关卡设计师:产出玩法节奏、关卡/阶段编排、操作手感、失败-重试体验。",
     roles.DESIGN_LEVEL_DESIGNER),
    ("design_presentation",
     "表现层与可玩性设计师:产出 Phaser scene 树、UI/HUD 布局与命中映射、进展反馈、可玩性(390×844 竖屏)。",
     roles.DESIGN_PRESENTATION_DESIGNER),
]


def _make_expert_tool(tool_name: str, tool_desc: str, system_prompt: str,
                      model_factory, *, model_sink: list,
                      counter: dict, per_expert_cap: int):
    """把一个专家 worker 包成 leader 的一个 FunctionTool(worker-as-tool)。

    返回的 async 工具函数:leader 调用时,起一个**无工具**的专家 Agent 跑一次 reply,把题面(及 leader 附带的
    上游上下文)交给它,返回该域设计文本(str)。FunctionTool 会把 str 自动包成 TextBlock 回给 leader
    (_adapters.py:_convert_func_result_to_chunk),故工具直接返回 str 即可。

    预算/熔断:每个专家至多被 leader 调 per_expert_cap 次(确定性预算闸,防 leader 反复刷同一专家烧 token);
      超限不抛、返回提示文本让 leader 改去汇总(degrade 友好)。专家 Agent 用到的 model 客户端 append 进
      model_sink,供 studio 汇总 token 成本(多 agent 烧的 token 必须被成本台账抓到,P0 纪律)。

    Args:
        tool_name: 工具名(= leader 提示词里列的四名之一)。
        tool_desc: 工具中文描述(被抽成 leader 可见 schema,作选工具依据)。
        system_prompt: 该专家的 system prompt(roles.DESIGN_*_DESIGNER)。
        model_factory: 零参 callable,每次调返回一个新的 ChatModel 客户端(= config.build_model 的偏函数)。
        model_sink: 收集本团队造的所有 model(供 studio 汇总 records → 成本)。
        counter: 共享调用计数 dict(key=tool_name);跨同名工具多次调用累计。
        per_expert_cap: 单专家被调用次数上限。
    """

    async def _expert(brief: str, context: str = "") -> str:
        """产出本域富游戏设计结论。

        Args:
            brief: 游戏题面(主设计师转交的设计需求)。
            context: 可选,主设计师附带的上游已收结论(让本专家衔接得上,如系统骨架/经济数值)。

        Returns:
            本域的设计结论文本(紧凑中文,给下游单写 agent 实现依据)。
        """
        # 确定性预算闸:单专家被调超上限即不再真跑模型,返回提示让 leader 收尾去汇总。
        counter[tool_name] = counter.get(tool_name, 0) + 1
        if counter[tool_name] > per_expert_cap:
            print(f"[tier2-design-team] 专家 {tool_name} 调用超上限({per_expert_cap}),"
                  "不再发散、提示 leader 汇总。", flush=True)
            return (f"(预算提示:{tool_name} 已达本局调用上限 {per_expert_cap} 次,"
                    "请基于已有结论汇总成最终设计稿,不要再重复调用本专家。)")

        # 每个专家 worker = 一个无工具、单轮的 Agent(只发散本域设计、不调工具、不自治)。
        model = model_factory()
        model_sink.append(model)  # 收集 model 供 studio 汇总 token 成本(多 agent 开销也要进台账)
        agent = Agent(
            name=tool_name,
            system_prompt=system_prompt,
            model=model,
            react_config=ReActConfig(max_iters=1),  # 专家只发散一轮、无 toolkit
        )
        # 组装专家输入:题面 + leader 附带的上游上下文(若有)。
        user_text = f"游戏题面:\n{(brief or '').strip()}"
        if (context or "").strip():
            user_text += f"\n\n【主设计师附带的上游已收结论(供你衔接)】\n{context.strip()}"
        user_text += "\n\n请只产出你这一域的设计结论。"
        try:
            resp = await agent.reply(_user_msg(user_text))
            out = _text_of(resp).strip()
        except Exception as e:  # noqa: BLE001 —— 单个专家失败不抛,返回降级提示让 leader 继续(团队鲁棒)
            print(f"[tier2-design-team] 专家 {tool_name} 调用失败(返回降级提示,不中断团队):"
                  f"{type(e).__name__}: {e}", flush=True)
            return f"({tool_name} 本次产出失败:{type(e).__name__};请 leader 据其余专家结论尽力汇总。)"
        if not out:
            return f"({tool_name} 产出为空;请 leader 据其余专家结论尽力汇总。)"
        print(f"[tier2-design-team] 专家 {tool_name} 产出 {len(out)} 字。", flush=True)
        return out

    # FunctionTool:name 取传入 tool_name(而非内层函数名 _expert),description 显式传(否则抽 docstring)。
    #   input_schema 由 AgentScope 从 _expert 的 (brief, context) 注解 + docstring 自动抽。
    return FunctionTool(_expert, name=tool_name, description=tool_desc)


async def run_design_team(
    brief: str,
    model_factory,
    *,
    model_sink: list | None = None,
    leader_max_iters: int | None = None,
    per_expert_cap: int | None = None,
    timeout_s: float | None = None,
) -> str:
    """跑工作室星形设计团队,返回一份连贯的富游戏设计稿(str;形状同原 _design_stage)。

    星形落地 = worker-as-tool:四专家各包成 leader 的一个工具,leader(ReAct + BYPASS)拆题面 → 调专家 →
    汇总。成功 = leader 跑完后产出非空最终设计稿;否则抛 DesignTeamError 由调用方 degrade。

    预算约束(多 agent 烧 token,必须有上限):
      - leader_max_iters:leader ReAct 总轮数上限(= 调专家 + 汇总的天花板;够调四专家各 ≤2 次 + 汇总);
      - per_expert_cap:单专家被调上限(防 leader 反复刷同一专家);
      - timeout_s:整团队墙钟硬超时(秒),兜住模型挂起/leader 不收敛。
    三者均为 None → 运行时读 generation.yaml 的 design_team 段(default 12 / 2 / 240.0;口径同 studio.py
    读 iteration.* 旋钮)。**这是配置外置的兑现点**:改 generation.yaml 的 design_team.timeout_s 即生效,
    不必改代码(此前硬编码默认 240 导致 YAML 调 timeout 不生效,已修)。
    任一预算触发都不让设计阶段挂死:超时/异常 → 抛 DesignTeamError → studio degrade 回单 agent。

    Args:
        brief: 游戏题面。
        model_factory: 零参 callable,每次返回一个新 ChatModel(= functools.partial(config.build_model, ...))。
            leader 与每个专家各取一个独立客户端(records 各自计 token,与 studio 成本汇总方式一致)。
        model_sink: 外部传入的列表,本函数把造的所有 model(leader + 各专家)append 进去,
            供 studio 遍历它们的 records 汇总 token 成本(多 agent 开销进台账,P0 纪律)。None → 内部自建(成本会漏算)。
        leader_max_iters / per_expert_cap / timeout_s: 预算上限(见上)。

    Returns:
        连贯的富游戏设计稿(str)。

    Raises:
        DesignTeamError: 超时 / leader 未产出有效设计稿 / 团队整体异常(调用方据此 degrade)。
    """
    if model_sink is None:
        model_sink = []  # 容错:调用方没传则内部建(此时成本汇总会漏算 team token,仅兜底用)

    # ── 旋钮外置(运行时读 generation.yaml 的 design_team 段;口径同 studio.py 读 iteration.*)──
    #   显式传入(非 None)优先;未传 → 从 genconfig 读,default 对齐历史硬编码值(12 / 2 / 240.0)。
    #   修复点:此前这三个是写死的函数默认值,studio 调用时不传 → generation.yaml 的 design_team.* 形同虚设
    #   (改 YAML 调 timeout 不生效)。现在 None→genconfig.get,配置外置真正兑现。
    if leader_max_iters is None:
        leader_max_iters = genconfig.get("design_team", "leader_max_iters", 12)
    if per_expert_cap is None:
        per_expert_cap = genconfig.get("design_team", "per_expert_cap", 2)
    if timeout_s is None:
        timeout_s = genconfig.get("design_team", "timeout_s", 240.0)

    # 每专家调用计数(跨同名工具多次调用累计;预算闸据它判超限)。
    counter: dict[str, int] = {}

    # ① 造四个专家工具(worker-as-tool),装进 leader 的 Toolkit。
    expert_tools = [
        _make_expert_tool(name, desc, system, model_factory,
                          model_sink=model_sink, counter=counter, per_expert_cap=per_expert_cap)
        for (name, desc, system) in _EXPERTS
    ]
    toolkit = Toolkit(tools=expert_tools)

    # ② leader = ReAct Agent(握四专家工具 + BYPASS 权限);它拆题面、调专家、最后汇总成连贯设计稿。
    leader_model = model_factory()
    model_sink.append(leader_model)  # leader 自己的 token 也进台账
    leader = Agent(
        name="design-leader",
        system_prompt=roles.design_leader_system(brief),
        model=leader_model,
        toolkit=toolkit,
        state=_bypass_state(),  # 不给 BYPASS,leader 调专家工具会卡在默认 ASK(_adapters.py:99)
        react_config=ReActConfig(max_iters=leader_max_iters),
    )

    kick = (
        "开始为这款富游戏做工作室设计。先想清大方向,再依次调用四个专家工具(系统/经济/玩法/表现层)"
        "拿到各面设计结论,把题面和已收到的上游结论一起转交给后调的专家以便衔接,"
        "最后把四面结论整合成一份连贯、可直接实现的最终设计稿(你的最后一条回答,不再调工具)。")

    # ③ 跑 leader(整团队墙钟硬超时兜底);超时/异常 → 抛 DesignTeamError 由 studio degrade。
    try:
        resp = await asyncio.wait_for(leader.reply(_user_msg(kick)), timeout=timeout_s)
    except asyncio.TimeoutError as e:
        # kind='timeout':整团队墙钟硬超时——studio 采集时据此归 infra_flags 的 wall_timeout(+ design_team_degraded)。
        raise DesignTeamError(f"工作室设计团队超时(> {timeout_s}s):{e}", kind="timeout") from e
    except Exception as e:  # noqa: BLE001 —— 任何团队异常都转成 DesignTeamError(让 studio 走 degrade)
        raise DesignTeamError(f"工作室设计团队异常:{type(e).__name__}: {e}") from e

    design = _text_of(resp).strip()
    calls = sum(counter.values())
    print(f"[tier2-design-team] 团队收敛:专家调用合计 {calls} 次,leader 汇总设计稿 {len(design)} 字。",
          flush=True)
    if not design:
        # leader 没产出有效设计稿(可能在调工具途中耗尽 max_iters、最后一回合是空文本)→ degrade。
        raise DesignTeamError("工作室设计团队未产出有效设计稿(leader 最终回答为空)。")
    return design