multi-role-transfer

第10章 · 多 Agent 协作 · 配套项目 chapter10/multi-role-transfer

项目说明

实验 10-2：多角色转换 / transfer_to_agent（★★）

《深入理解 AI Agent》配套代码。演示共享上下文下的链式移交（handoff）： 一个会话里存在多个专业角色 Agent（各有独立系统提示词与专属工具集）， 通过一个 transfer_to_agent(target_role, reason) 工具在角色间自主移交控制权。

这个实验想说明什么

• 与 10-1（软件开发单任务的预定义阶段流水线）不同，10-2 强调跨领域、 由 Agent 自主判断该切换到哪个专业角色——不是预先规划好的线性流程， 而是根据任务进展动态切换。
• 因为共享同一段对话历史，移交时完整历史天然保留， 新角色自动继承此前所有内容（无需显式传参）。
• 机制重点是「自主角色移交」，而非工具本身多强，因此工具用轻量真实实现 / 可控 mock。

架构

                        共享对话历史 history（user/assistant/tool 消息，全程保留）
                                        ▲   ▲
   每轮调用大模型时：                     │   │
   [ 当前角色的 system prompt ] + history ┘   └ 只暴露 [ 当前角色工具集 + transfer_to_agent ]

   模型两种动作：
     ① 调用自己的专属工具（普通 function calling）
     ② 调用 transfer_to_agent(target_role, reason)
        → 编排器换掉「系统提示词 + 工具集」，history 原样不动
        → 新角色继承全部历史（共享上下文）

5 个角色（roles.py）：

角色	说明	专属工具集
triage	前台分诊 / 默认入口，拆解需求并按序移交、最后收尾	仅 transfer_to_agent
research	信息检索	web_search（内置知识库 mock）
coding	编程	execute_python（真实执行并捕获输出）
data_analysis	数据分析 / 计算	calculate、descriptive_stats
writing	润色写作	count_characters

每个角色都额外持有 transfer_to_agent，可自主把控制权交给同事。

代码结构：

• tools.py —— 各角色专属工具的实现 + OpenAI function-calling schema
• roles.py —— 5 个角色定义（系统提示词 + 工具集）+ transfer_to_agent schema
• orchestrator.py —— 移交编排器（共享历史 + 换系统提示词/工具集的主循环，含防死循环/拒绝自我移交）
• demo.py —— 一条命令的演示入口

运行方式

pip install -r requirements.txt

# 配置 key（二选一）
export OPENAI_API_KEY=sk-...        # 直接 export
# 或： cp env.example .env 后填写

python demo.py

可配环境变量（均有默认值）： OPENAI_API_KEY、OPENAI_BASE_URL（默认 https://api.openai.com/v1）、 OPENAI_MODEL（默认 gpt-5.6-luna）。

通用回退：优先用 OPENAI_API_KEY 直连 OpenAI；若未设置该变量但设了 OPENROUTER_API_KEY，则自动改走 OpenRouter，并把模型名映射到其命名空间 （gpt-5.6-luna → openai/gpt-5.6-luna）。提示：gpt-5.6 系列直连 OpenAI 需组织验证， 只填 OPENROUTER_API_KEY（不填 OPENAI_API_KEY）即可强制走 OpenRouter，更省事。

命令行参数

所有参数均可选，不传则行为与最初版本完全一致（跑默认 cagr 场景）。运行 python demo.py --help 查看完整中文说明。

参数	作用
--list-roles	离线自检：只打印角色花名册 + 内置场景后退出，无需 API Key
--scenario {cagr,solar,coding}	选内置场景（默认 cagr）；coding 会路由到 coding 角色真正跑代码
--task "..."	自定义任务文本，覆盖 --scenario
--role {triage,research,coding,data_analysis,writing}	指定起始角色（别名 --starting-role，默认 triage）
--interactive	交互式多轮：复用同一编排器，角色与共享历史跨轮保留
--model gpt-5.6-luna	临时覆盖 OPENAI_MODEL
--max-steps 30	单条消息的最大 LLM 轮数硬上限（默认 20，防死循环）

例：

python demo.py --list-roles            # 离线看角色/场景清单，不调用 API
python demo.py --scenario coding       # 路由到 coding 角色的场景
python demo.py --task "帮我调研并总结…" # 自定义任务
python demo.py --role research         # 从 research 角色起步
python demo.py --interactive           # 交互式多轮，输入 exit 退出

三个内置场景（SCENARIOS）：cagr（默认，新能源汽车销量→CAGR→投资总结）、 solar（同类链路换一组光伏装机数据）、coding（路由到 coding 角色用 execute_python 真正跑斐波那契脚本，再由 writing/triage 收尾）。

演示说明

demo.py 抛出一个需要多次跨领域切换的复合任务：

查中国 2021—2023 三年新能源汽车销量 → 算出年均复合增长率(CAGR) → 写成一段面向投资人的中文总结

预期看到 Agent 自主完成移交链：

triage → research → data_analysis → writing

• triage 判断第一步要查数据，移交 research；
• research 用 web_search 查到三年销量，移交 data_analysis；
• data_analysis 用 calculate 算出 CAGR ≈ 64.22%，移交 writing；
• writing 综合此前历史里的销量数据与 CAGR，直接写出最终成稿。

writing 从未自己检索或计算，却能引用准确的销量数字和增长率—— 这正是共享上下文的证据。运行结束会打印完整移交链、每次移交的 from→to 与 reason， 以及各角色分工总览（谁调用了哪些专属工具、谁产出了最终回复），一眼看清 「同一段历史上不同专业角色各司其职地接力」。

注：真实 LLM 输出有随机性，某次运行的具体措辞/步数可能略有不同，但移交机制一致。

预期输出示例（真实运行截取）

以下是一次 python demo.py（model=gpt-5.6-luna，经 OpenRouter 路由）真实运行的关键片段，未做任何编造或修饰：

=== 角色花名册（共 5 个专业角色）===
• triage — 前台分诊（默认入口）
    工具集: ['transfer_to_agent']
    系统提示词(首句): 你是通用助理系统的『前台分诊』角色，也是默认入口。
• research — 信息检索专家
    工具集: ['web_search', 'transfer_to_agent']
    ...（其余角色略，完整列表见上方角色表）

┌── 当前角色: 信息检索专家 (research)  工具: ['web_search', 'transfer_to_agent']
└── 🔧 调用工具 web_search args={'query': '中国 2021年 2022年 2023年 新能源汽车销量 乘联会 中汽协'}
    → 【检索结果·中国乘用车市场信息联席会/中汽协】…2021 年：352.1 万辆 / 2022 年：688.7 万辆 / 2023 年：949.5 万辆
┌── 当前角色: 数据分析专家 (data_analysis)  工具: ['calculate', 'descriptive_stats', 'transfer_to_agent']
└── 🔧 调用工具 calculate args={'expression': '(949.5/352.1)**(1/2)-1'}
    → (949.5/352.1)**(1/2)-1 = 0.6421562289791105

================ 运行汇总 ================
自主移交链: triage → research → data_analysis → writing → triage
移交次数: 4
  1. triage → research  |  reason: 需要先检索中国2021、2022、2023年新能源汽车销量及可靠来源，为后续CAGR计算和投资人摘要提供数据依据。
  2. research → data_analysis  |  reason: 已检索到2021、2022、2023年新能源汽车销量，请计算2021至2023年的两年CAGR，并给出结果供后续写作。
  3. data_analysis → writing  |  reason: 销量数据与CAGR已完成：2021年352.1万辆、2022年688.7万辆、2023年949.5万辆；2021—2023年CAGR=(949.5/352.1)^(1/2)-1=64.22%。请据此写不超过120字的投资人中文总结。
  4. writing → triage  |  reason: 已完成投资人摘要并核对篇幅（101字符，不超过120字）…请做最终收尾确认。

各角色分工（谁用了什么工具、谁产出最终回复）:
  triage        : （仅路由/移交，未用专属工具）  ⇒ 产出最终回复
  research      : web_search
  data_analysis : calculate
  writing       : count_characters

最终成果:
据中汽协公开数据，中国新能源汽车销量由2021年的352.1万辆增至2022年的688.7万辆、2023年的949.5万辆。2021—2023年两年CAGR达64.2%，市场保持高速扩张，成长潜力显著。

局限

• 默认模型为 gpt-5.6-luna；移交是否按预期链路发生，很大程度依赖所选模型的指令遵循能力，换模型效果可能不同。
• research 角色的 web_search 是内置知识库 mock，并非真实联网检索，仅能命中预置的少量关键词（新能源汽车销量、光伏装机、Python GIL），换查询词可能查不到。
• 真实 LLM 输出存在随机性：具体移交步数、每次 reason 的措辞、是否途经 coding 角色等，不同次运行可能不同，但移交机制本身一致。
• orchestrator.py 设有 max_steps（默认 20）硬上限，以及「同一 (角色, 工具, 参数) 连续调用 ≥3 次」的纠偏提示，用于防止模型死循环；这是兜底保护，不代表每次运行都会用满这些步数。

源代码

demo.py

"""
demo.py —— 实验 10-2 演示入口：多角色转换 / transfer_to_agent

最简运行（一条命令，跑默认复合任务）：
    python demo.py

其它常用方式：
    python demo.py --list-roles              # 离线：只打印角色花名册后退出（无需 API Key）
    python demo.py --scenario coding         # 换一个内置场景（会路由到 coding 角色）
    python demo.py --task "..."              # 自定义任务
    python demo.py --role research            # 指定起始角色（默认 triage 前台分诊）
    python demo.py --interactive             # 交互式多轮对话（角色与共享历史跨轮保留）
    python demo.py --model gpt-5.6-luna --max-steps 30

演示一个需要【多次跨领域切换】的复合任务，预期出现
    triage → research → data_analysis → writing
的自主移交链——每次移交都由当前角色自己判断并调用 transfer_to_agent 触发。
"""

from __future__ import annotations

import argparse
import os
import sys

from openai import OpenAI

from roles import ROLES, DEFAULT_ROLE
from orchestrator import MultiRoleOrchestrator, C


def _to_openrouter_model(model: str) -> str:
    """把模型名映射到 OpenRouter 命名空间（用于无 OPENAI_API_KEY 的回退路径）。"""
    if "/" in model:
        return model                      # 已是 OpenRouter 命名空间，原样使用
    if model.startswith("gpt-"):
        return "openai/" + model          # gpt-* -> openai/gpt-*
    if model.startswith("claude-"):
        return "anthropic/claude-opus-4.8"
    return "openai/gpt-5.6-luna"          # 兜底：当前便宜旗舰

# 尽量读取 .env（可选依赖，没装也能跑，只要 shell 里已 export）
try:
    from dotenv import load_dotenv

    load_dotenv()
except ImportError:
    pass


# ---------------------------------------------------------------------------
# 内置场景：每个都刻意跨多个领域，以逼出多次自主移交。
# 键名用于 --scenario；值为 (任务文本, 一句话说明)。
# ---------------------------------------------------------------------------
COMPOSITE_TASK = (
    "我在准备一份给投资人看的材料。请帮我：\n"
    "1) 查一下中国 2021、2022、2023 三年的新能源汽车销量；\n"
    "2) 据此算出这三年的年均复合增长率(CAGR)；\n"
    "3) 把数据和这个增长率结论，写成一段面向投资人的、不超过 120 字的中文总结。"
)

SCENARIOS: dict[str, tuple[str, str]] = {
    "cagr": (
        COMPOSITE_TASK,
        "默认场景。跨检索/计算/写作三领域：查销量 → 算 CAGR → 写投资总结，"
        "预期链路 triage → research → data_analysis → writing。",
    ),
    "solar": (
        "帮我查一下中国 2021、2022、2023 三年的光伏新增装机量，"
        "算出这三年的年均复合增长率(CAGR)，再写成一句话面向读者的结论。",
        "另一组数据的同类链路（research → data_analysis → writing），验证机制而非记住答案。",
    ),
    "coding": (
        "请写一个 Python 脚本：计算斐波那契数列前 20 项，并求它们的和；"
        "运行脚本得到结果后，用一句话向非技术读者解释这个结果。",
        "路由到 coding 角色用 execute_python 真正跑代码，再由 writing/triage 收尾。",
    ),
}
DEFAULT_SCENARIO = "cagr"


def print_roster():
    """打印角色花名册，证明存在 5 个角色、各有不同系统提示词/工具集。"""
    print(f"{C.BOLD}=== 角色花名册（共 {len(ROLES)} 个专业角色）==={C.RESET}")
    for name, role in ROLES.items():
        default_tag = "（默认入口）" if name == DEFAULT_ROLE else ""
        tools = role.tools + ["transfer_to_agent"]
        first_line = role.system_prompt.strip().splitlines()[0]
        print(
            f"{C.CYAN}• {name}{C.RESET} — {role.title}{default_tag}\n"
            f"    工具集: {tools}\n"
            f"    系统提示词(首句): {first_line}"
        )
    print()


def print_scenarios():
    """打印内置场景列表（供 --help / --list-roles 参考）。"""
    print(f"{C.BOLD}=== 内置场景（--scenario）==={C.RESET}")
    for key, (_task, desc) in SCENARIOS.items():
        default_tag = "（默认）" if key == DEFAULT_SCENARIO else ""
        print(f"{C.CYAN}• {key}{C.RESET}{default_tag} — {desc}")
    print()


def parse_args() -> argparse.Namespace:
    """命令行参数——均为可选，不传时行为与最初版本完全一致（跑默认复合任务）。"""
    parser = argparse.ArgumentParser(
        prog="demo.py",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=(
            "实验 10-2 演示：多角色转换 / transfer_to_agent。\n"
            "在一段【共享对话历史】上，5 个专业角色通过 transfer_to_agent 自主接力，"
            "触发形如 triage → research → data_analysis → writing 的移交链。"
        ),
        epilog=(
            "示例：\n"
            "  python demo.py                     # 跑默认场景（新能源汽车 CAGR 投资总结）\n"
            "  python demo.py --list-roles        # 离线：只看角色/场景清单，不调用 API\n"
            "  python demo.py --scenario coding   # 换到会路由至 coding 角色的场景\n"
            "  python demo.py --task '帮我...'    # 自定义任务\n"
            "  python demo.py --role research     # 从 research 角色起步\n"
            "  python demo.py --interactive       # 交互式多轮，角色与共享历史跨轮保留\n"
        ),
    )
    parser.add_argument(
        "--scenario",
        choices=list(SCENARIOS.keys()),
        default=DEFAULT_SCENARIO,
        help=f"选择一个内置场景（默认 {DEFAULT_SCENARIO}）；被 --task 覆盖。可选：{list(SCENARIOS.keys())}",
    )
    parser.add_argument(
        "--task",
        default=None,
        help="自定义任务文本，覆盖 --scenario；不传则使用所选内置场景。",
    )
    parser.add_argument(
        "--role",
        "--starting-role",
        dest="role",
        choices=list(ROLES.keys()),
        default=DEFAULT_ROLE,
        help=f"指定起始角色（默认 {DEFAULT_ROLE} 前台分诊）。可选：{list(ROLES.keys())}",
    )
    parser.add_argument(
        "--interactive",
        action="store_true",
        help="交互式多轮模式：复用同一编排器，角色与共享历史跨轮保留（Ctrl-C / 输入 exit 退出）。",
    )
    parser.add_argument(
        "--model",
        default=None,
        help="覆盖 OPENAI_MODEL 环境变量（默认沿用环境变量，未设置则为 gpt-5.6-luna）。",
    )
    parser.add_argument(
        "--max-steps",
        type=int,
        default=20,
        help="单条用户消息的最大 LLM 轮数硬上限，防止死循环（默认 20）。",
    )
    parser.add_argument(
        "--list-roles",
        action="store_true",
        help="离线打印角色花名册与内置场景后退出，不需要 API Key（用于自检）。",
    )
    return parser.parse_args()


def print_run_summary(orch: MultiRoleOrchestrator, final: str):
    """打印一次运行的移交链、分工总览与最终成果。"""
    print(f"\n{C.BOLD}================ 运行汇总 ================{C.RESET}")
    print(f"{C.MAGENTA}自主移交链:{C.RESET} {orch.handoff_chain_str()}")
    print(f"{C.MAGENTA}移交次数:{C.RESET} {len(orch.handoffs)}")
    for i, h in enumerate(orch.handoffs, 1):
        print(f"  {i}. {h.from_role} → {h.to_role}  |  reason: {h.reason}")
    print(f"\n{C.MAGENTA}各角色分工（谁用了什么工具、谁产出最终回复）:{C.RESET}")
    print(orch.role_work_summary())
    print(f"\n{C.GREEN}最终成果:{C.RESET}\n{final}")


def run_interactive(orch: MultiRoleOrchestrator):
    """交互式多轮：同一编排器跨轮复用，共享历史与当前角色持续保留。"""
    print(
        f"{C.BOLD}=== 交互式多轮模式 ==={C.RESET}\n"
        f"{C.DIM}输入你的请求后回车；输入 exit / quit 或按 Ctrl-C 退出。"
        f"角色与对话历史会跨轮保留（共享上下文）。{C.RESET}"
    )
    turn = 0
    while True:
        try:
            user_message = input(f"\n{C.BOLD}👤 你（当前控制权在 {orch.current_role}）> {C.RESET}").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n已退出交互模式。")
            break
        if not user_message:
            continue
        if user_message.lower() in {"exit", "quit", "q"}:
            print("已退出交互模式。")
            break
        turn += 1
        final = orch.run(user_message)
        print_run_summary(orch, final)


def main():
    args = parse_args()

    # ---- 离线自检路径：无需 API Key ----
    if args.list_roles:
        print_roster()
        print_scenarios()
        return

    model = args.model or os.environ.get("OPENAI_MODEL", "gpt-5.6-luna")

    # 通用回退：优先直连 OPENAI_API_KEY；否则用 OPENROUTER_API_KEY 走 OpenRouter；
    # 都没有则报清晰错误。
    # 特例：gpt-5.x 系列直连 OpenAI 需组织验证，且其 /v1/chat/completions 对带工具的
    # 推理模型支持受限（reasoning_effort 限制）。因此只要设置了 OPENROUTER_API_KEY，
    # 就对 gpt-5.x 优先改走 OpenRouter，避免直连报错。
    prefer_openrouter = model.startswith("gpt-5") and os.environ.get("OPENROUTER_API_KEY")
    api_key = None if prefer_openrouter else os.environ.get("OPENAI_API_KEY")
    if api_key:
        base_url = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
    elif os.environ.get("OPENROUTER_API_KEY"):
        api_key = os.environ["OPENROUTER_API_KEY"]
        base_url = "https://openrouter.ai/api/v1"
        model = _to_openrouter_model(model)
        why = "gpt-5.x 优先走 OpenRouter" if prefer_openrouter else "未检测到 OPENAI_API_KEY"
        print(f"（{why}，改用 OpenRouter；模型映射为 {model}）")
    else:
        print("错误：未找到环境变量 OPENAI_API_KEY 或 OPENROUTER_API_KEY。请先设置后重试。",
              file=sys.stderr)
        print("（提示：只想看角色/场景清单可运行 `python demo.py --list-roles`，无需 Key。）",
              file=sys.stderr)
        sys.exit(1)

    client = OpenAI(api_key=api_key, base_url=base_url)

    print_roster()

    orch = MultiRoleOrchestrator(
        client=client,
        model=model,
        max_steps=args.max_steps,
        verbose=True,
        start_role=args.role,
    )

    if args.interactive:
        print(f"{C.BOLD}=== 模型 model={model}，起始角色 {args.role} ==={C.RESET}")
        run_interactive(orch)
        return

    # ---- 脚本化：单条复合任务，端到端跑完一次 ----
    task = args.task if args.task is not None else SCENARIOS[args.scenario][0]
    scenario_tag = "自定义任务" if args.task is not None else f"场景 {args.scenario}"
    print(f"{C.BOLD}=== 开始执行（{scenario_tag}，model={model}，起始角色={args.role}）==={C.RESET}")

    final = orch.run(task)
    print_run_summary(orch, final)


if __name__ == "__main__":
    main()

orchestrator.py

"""
orchestrator.py —— 多角色移交（handoff）编排器。

核心机制（实验 10-2）：
- 全程维护一段【共享对话历史】history（user/assistant/tool 消息）。
- 每次调用大模型时，把【当前角色】的系统提示词临时拼到 history 前面，
  并只暴露【当前角色的工具集 + transfer_to_agent】。
- 模型可以：
    1) 调用自己的专属工具（正常 function calling）；
    2) 调用 transfer_to_agent 把控制权移交给别的角色——
       此时编排器换掉「系统提示词 + 工具集」，但 history 原样保留，
       于是新角色天然继承了全部对话历史（共享上下文）。
- 循环直到某个角色给出「没有工具调用」的最终回复。
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Dict, List, Optional

from openai import OpenAI

from roles import ROLES, DEFAULT_ROLE, transfer_tool_schema
from tools import TOOL_SCHEMAS, TOOL_IMPLEMENTATIONS


# ---- 终端着色（无第三方依赖）----
class C:
    RESET = "\033[0m"
    DIM = "\033[2m"
    BOLD = "\033[1m"
    CYAN = "\033[36m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    MAGENTA = "\033[35m"
    BLUE = "\033[34m"
    RED = "\033[31m"


@dataclass
class Handoff:
    from_role: str
    to_role: str
    reason: str


class MultiRoleOrchestrator:
    def __init__(
        self,
        client: OpenAI,
        model: str = "gpt-5.6-luna",
        max_steps: int = 20,
        verbose: bool = True,
        start_role: str = DEFAULT_ROLE,
    ):
        if start_role not in ROLES:
            raise ValueError(f"未知的起始角色 {start_role!r}，可选：{list(ROLES.keys())}")
        self.client = client
        self.model = model
        self.max_steps = max_steps
        self.verbose = verbose

        self.history: List[dict] = []          # 共享对话历史（不含 system）
        self.current_role: str = start_role    # 当前控制权所在角色（可自定义起始角色）
        self.handoffs: List[Handoff] = []      # 记录移交链
        self._tool_call_counts: Dict[str, int] = {}  # 相同工具调用去重计数（防死循环）
        # 分工记录：(role, kind, detail)，kind ∈ {"tool", "transfer", "final"}，
        # 用于运行结束后打印「哪个角色做了什么」的分工总览。
        self.activity: List[tuple] = []

    # -------------------------------------------------------------- 工具装配
    def _tools_for_current_role(self) -> List[dict]:
        """当前角色可见的工具 = 专属工具集 + transfer_to_agent。"""
        role = ROLES[self.current_role]
        schemas = [TOOL_SCHEMAS[name] for name in role.tools]
        schemas.append(transfer_tool_schema())  # 每个角色都能移交
        return schemas

    def _messages_for_api(self) -> List[dict]:
        """把当前角色的系统提示词拼到共享历史前面。"""
        system_msg = {"role": "system", "content": ROLES[self.current_role].system_prompt}
        return [system_msg] + self.history

    # -------------------------------------------------------------- 日志
    def _log(self, msg: str):
        if self.verbose:
            print(msg)

    def _log_role_banner(self):
        role = ROLES[self.current_role]
        self._log(
            f"\n{C.BOLD}{C.CYAN}┌── 当前角色: {role.title} ({role.name}){C.RESET}"
            f"{C.DIM}  工具: {role.tools + ['transfer_to_agent']}{C.RESET}"
        )

    # -------------------------------------------------------------- 单步
    def _run_one_llm_turn(self) -> Optional[str]:
        """
        执行一次「模型调用 + 工具处理」。
        返回值：
          - None  表示还要继续循环（发生了工具调用/移交）
          - str   表示这是最终回复（模型没有再调用工具），流程结束
        """
        self._log_role_banner()

        response = self.client.chat.completions.create(
            model=self.model,
            messages=self._messages_for_api(),
            tools=self._tools_for_current_role(),
            temperature=0,
        )
        msg = response.choices[0].message

        # 没有工具调用 => 最终回复
        if not msg.tool_calls:
            content = msg.content or ""
            self.history.append({"role": "assistant", "content": content})
            self.activity.append((self.current_role, "final", ""))
            self._log(f"{C.GREEN}└── [{self.current_role}] 最终回复:{C.RESET}\n{content}")
            return content

        # 有工具调用：先把 assistant 消息（含 tool_calls）写进历史
        self.history.append(
            {
                "role": "assistant",
                "content": msg.content or "",
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                    }
                    for tc in msg.tool_calls
                ],
            }
        )

        pending_transfer: Optional[Handoff] = None

        # 逐个处理工具调用，并为每个调用回填一条 tool 消息（OpenAI 要求）
        for tc in msg.tool_calls:
            name = tc.function.name
            try:
                args = json.loads(tc.function.arguments or "{}")
            except json.JSONDecodeError:
                args = {}

            if name == "transfer_to_agent":
                target = args.get("target_role", "")
                reason = args.get("reason", "")
                if target == self.current_role:
                    # 拒绝自我移交：让模型改用自己的工具或选别的角色
                    result = (
                        f"移交失败：你已经是 {target} 角色，不能移交给自己。"
                        "请直接使用你自己的工具完成当前部分，或移交给其他角色。"
                    )
                    self._log(f"{C.RED}└── transfer 被拒: 不能移交给自己 ({target}){C.RESET}")
                elif target in ROLES:
                    pending_transfer = Handoff(self.current_role, target, reason)
                    self.activity.append((self.current_role, "transfer", target))
                    result = f"已移交给 {target}。对方将继承完整对话历史并继续处理。"
                    self._log(
                        f"{C.MAGENTA}└── ⇢ transfer_to_agent: "
                        f"{self.current_role} → {target}{C.RESET}\n"
                        f"    {C.YELLOW}reason:{C.RESET} {reason}"
                    )
                else:
                    result = f"移交失败：未知角色 {target!r}。可选：{list(ROLES.keys())}"
                    self._log(f"{C.RED}└── transfer 失败: 未知角色 {target!r}{C.RESET}")
            else:
                impl = TOOL_IMPLEMENTATIONS.get(name)
                if impl is None:
                    result = f"工具 {name} 不存在。"
                else:
                    try:
                        result = impl(**args)
                    except (TypeError, ValueError) as exc:
                        # 模型偶尔会传错/漏参数（如 {"q": ...} 而非 {"query": ...}）
                        # 或给出无法转换的值；把错误作为工具结果回给模型让它自行纠正，
                        # 而不是让整个移交流程崩溃。
                        result = f"工具 {name} 调用失败：{exc}。请检查参数名与取值后重试。"
                    self.activity.append((self.current_role, "tool", name))
                # 防死循环：同一 (角色,工具,参数) 反复调用时给出纠偏提示
                sig = f"{self.current_role}:{name}:{tc.function.arguments}"
                self._tool_call_counts[sig] = self._tool_call_counts.get(sig, 0) + 1
                if self._tool_call_counts[sig] >= 3:
                    result += (
                        "\n[系统提示] 你已多次重复完全相同的调用。请停止重复，"
                        "直接给出最终文本，或调用 transfer_to_agent 移交给下一个角色。"
                    )
                self._log(
                    f"{C.BLUE}└── 🔧 调用工具 {name}{C.RESET} "
                    f"{C.DIM}args={args}{C.RESET}\n"
                    f"    {C.DIM}→ {result[:300]}{C.RESET}"
                )

            self.history.append(
                {"role": "tool", "tool_call_id": tc.id, "content": str(result)}
            )

        # 处理完本轮所有工具调用后，如有移交则切换角色（保留 history）
        if pending_transfer is not None:
            self.handoffs.append(pending_transfer)
            self.current_role = pending_transfer.to_role

        return None  # 继续循环

    # -------------------------------------------------------------- 主循环
    def run(self, user_message: str) -> str:
        """处理一条用户消息，跑完整个多角色移交流程，返回最终回复。"""
        self.history.append({"role": "user", "content": user_message})
        self._log(f"{C.BOLD}👤 用户:{C.RESET} {user_message}")

        final_answer = ""
        for step in range(self.max_steps):
            result = self._run_one_llm_turn()
            if result is not None:
                final_answer = result
                break
        else:
            final_answer = "（达到最大步数上限，流程终止）"
            self._log(f"{C.RED}{final_answer}{C.RESET}")

        return final_answer

    # -------------------------------------------------------------- 汇总
    def handoff_chain_str(self) -> str:
        """返回可读的移交链，如 triage → research → data_analysis → writing → triage。"""
        if not self.handoffs:
            return DEFAULT_ROLE + "（未发生移交）"
        chain = [self.handoffs[0].from_role]
        for h in self.handoffs:
            chain.append(h.to_role)
        return " → ".join(chain)

    def role_work_summary(self) -> str:
        """
        返回「哪个角色做了什么」的分工总览——按角色首次出场顺序，
        列出每个角色实际调用过的专属工具，以及谁产出了最终回复。
        这直接印证：同一段共享历史上，不同专业角色各司其职地接力完成任务。
        """
        order: List[str] = []
        tools_by_role: Dict[str, List[str]] = {}
        final_role: Optional[str] = None
        for role, kind, detail in self.activity:
            if role not in order:
                order.append(role)
                tools_by_role[role] = []
            if kind == "tool" and detail not in tools_by_role[role]:
                tools_by_role[role].append(detail)
            elif kind == "final":
                final_role = role
        if not order:
            return "（无角色活动记录）"
        width = max(len(r) for r in order)
        lines: List[str] = []
        for role in order:
            used = tools_by_role[role]
            desc = "、".join(used) if used else "（仅路由/移交，未用专属工具）"
            if role == final_role:
                desc += "  ⇒ 产出最终回复"
            lines.append(f"  {role.ljust(width)} : {desc}")
        return "\n".join(lines)

roles.py

"""
roles.py —— 定义多个「专业角色 Agent」。

实验 10-2 的核心：一个会话里存在多个专业角色，每个角色有
  (1) 独立的系统提示词（system prompt）
  (2) 专属工具集（tools）
角色之间通过 transfer_to_agent(target_role, reason) 自主移交控制权。

与 10-1（软件开发单任务的预定义阶段流水线）不同，这里强调跨领域、
由 Agent 自主判断该切换到哪个角色——不是预先规划好的线性流程。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Dict, List


@dataclass
class Role:
    name: str            # 角色标识，用作 transfer_to_agent 的 target_role
    title: str           # 中文名称（打印用）
    system_prompt: str   # 该角色的系统提示词
    tools: List[str] = field(default_factory=list)  # 该角色的专属工具名（不含 transfer）


# 所有可移交的目标角色说明（会拼进每个角色的系统提示词，让它知道有哪些同事）。
_ROSTER_DESC = (
    "- triage：前台分诊（默认角色），负责理解需求、拆解任务、把控制权移交给合适的专业角色，"
    "并在全部子任务完成后做收尾确认。\n"
    "- research：信息检索专家，擅长用 web_search 查数据、事实、资料。\n"
    "- coding：编程专家，擅长用 execute_python 写并运行代码解决逻辑/脚本问题。\n"
    "- data_analysis：数据分析专家，擅长用 calculate / descriptive_stats 做计算与统计（如增长率、均值）。\n"
    "- writing：写作专家，擅长把零散结论润色成通顺、面向特定读者的成稿。\n"
)

# 每个角色系统提示词共用的移交纪律。
_HANDOFF_RULES = (
    "\n\n【团队协作规则】\n"
    f"当前会话中有以下专业角色（同事）：\n{_ROSTER_DESC}"
    "你们共享同一段对话历史，因此移交后新同事能看到此前的全部内容。\n"
    "如果当前任务超出你的职责范围，必须调用 transfer_to_agent(target_role, reason) "
    "把控制权移交给更合适的同事，而不要勉强自己做。\n"
    "reason 里要简述『为什么移交、请对方做什么』。\n"
    "只有当属于你职责范围内的部分做完时，才移交或收尾；不要一次移交给多个角色。"
)


ROLES: Dict[str, Role] = {
    "triage": Role(
        name="triage",
        title="前台分诊",
        tools=[],  # triage 没有专业工具，只有 transfer
        system_prompt=(
            "你是通用助理系统的『前台分诊』角色，也是默认入口。\n"
            "你的职责：理解用户的整体需求，把它拆成有先后顺序的子任务，"
            "然后【一步一步】把控制权移交给合适的专业角色去完成。\n"
            "典型顺序是：先移交 research 检索数据 → 再移交 data_analysis 计算指标 → "
            "最后移交 writing 成文。因此当任务包含『查数据』时，你的第一步一般就是移交给 research。\n"
            "你自己不做检索/编程/计算/长文写作——这些都要移交。\n"
            "当所有子任务都完成、最终成稿已经在对话里产出时，由你向用户做一句话收尾确认，"
            "并把最终成稿原文再复述一遍；此时不要再移交，直接输出结束语。"
        ) + _HANDOFF_RULES,
    ),
    "research": Role(
        name="research",
        title="信息检索专家",
        tools=["web_search"],
        system_prompt=(
            "你是『信息检索专家』。你的职责：用 web_search 工具查找用户需要的数据、"
            "事实或资料，并把检索到的关键信息清晰列出来（写进对话，供后续同事使用）。\n"
            "你不做数值计算，也不写最终成稿。检索完成后，如果接下来需要计算或写作，"
            "就移交给对应角色。"
        ) + _HANDOFF_RULES,
    ),
    "coding": Role(
        name="coding",
        title="编程专家",
        tools=["execute_python"],
        system_prompt=(
            "你是『编程专家』。你的职责：用 execute_python 写并运行代码来解决"
            "偏程序逻辑/脚本类的问题，并汇报运行结果。\n"
            "纯数学指标计算更适合 data_analysis；查资料更适合 research；"
            "写成稿更适合 writing。完成你的部分后按需移交。"
        ) + _HANDOFF_RULES,
    ),
    "data_analysis": Role(
        name="data_analysis",
        title="数据分析专家",
        tools=["calculate", "descriptive_stats"],
        system_prompt=(
            "你是『数据分析专家』。你的职责：基于对话里已有的数据，用 calculate / "
            "descriptive_stats 工具做定量计算与统计（如同比增长率、年均复合增长率 CAGR、"
            "均值等），并用文字清楚说明计算过程与结果。\n"
            "你不查资料也不写最终成稿。算完后如需润色成文，移交给 writing。"
        ) + _HANDOFF_RULES,
    ),
    "writing": Role(
        name="writing",
        title="写作专家",
        tools=["count_characters"],
        system_prompt=(
            "你是『写作专家』。你的职责：综合对话历史里检索到的数据和计算结论，"
            "写出一段通顺、结构清晰、面向指定读者的成稿。\n"
            "可以【最多一次】用 count_characters 粗略检查篇幅（这里的『字』指中文字符数）；"
            "不要反复核对字数，长度大致合适即可，切勿因为差几个字就反复重算。\n"
            "写好成稿后，立即调用 transfer_to_agent 把控制权移交回 triage 做收尾确认，"
            "不要停留在自己这一步。"
        ) + _HANDOFF_RULES,
    ),
}

DEFAULT_ROLE = "triage"


def transfer_tool_schema() -> dict:
    """transfer_to_agent 工具的 OpenAI schema —— 所有角色都持有它。"""
    return {
        "type": "function",
        "function": {
            "name": "transfer_to_agent",
            "description": (
                "把当前会话的控制权移交给另一个更合适的专业角色。"
                "移交后对方会继承完整对话历史。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "target_role": {
                        "type": "string",
                        "enum": list(ROLES.keys()),
                        "description": "要移交到的目标角色名",
                    },
                    "reason": {
                        "type": "string",
                        "description": "为什么移交、请对方做什么（简述）",
                    },
                },
                "required": ["target_role", "reason"],
            },
        },
    }

test_count_characters_null.py

from tools import count_characters


def test_count_characters_null_text():
    result = count_characters(None)
    assert result == "总字符数=0, 其中中文字符=0"


def test_count_characters_normal():
    result = count_characters("你好hi")
    assert "总字符数=4" in result
    assert "中文字符=2" in result

test_tool_dispatch_errors.py

"""回归测试：模型传错/漏工具参数时，编排器不应崩溃，而应把错误作为工具结果
回给模型（让它自行纠正），流程继续推进到最终回复。

此前 orchestrator.py 的 `impl(**args)` 未加保护：{"q": ...} 这类错键名、
缺必填参数、或无法 float() 转换的取值都会以 TypeError/ValueError 炸掉整个
多角色移交流程。
"""

import json
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace

sys.path.insert(0, str(Path(__file__).parent))

try:
    import openai  # noqa: F401
except ImportError:
    sys.modules["openai"] = ModuleType("openai")
    sys.modules["openai"].OpenAI = object

from orchestrator import MultiRoleOrchestrator

FINAL_TEXT = "已查完，最终汇报。"


def _tool_call_msg(name, arguments):
    tc = SimpleNamespace(
        id="call_1", type="function",
        function=SimpleNamespace(name=name, arguments=arguments))
    return SimpleNamespace(choices=[SimpleNamespace(
        message=SimpleNamespace(content=None, tool_calls=[tc]))])


def _final_msg():
    return SimpleNamespace(choices=[SimpleNamespace(
        message=SimpleNamespace(content=FINAL_TEXT, tool_calls=None))])


def _fake_client(responses):
    queue = list(responses)
    return SimpleNamespace(chat=SimpleNamespace(
        completions=SimpleNamespace(create=lambda **kw: queue.pop(0))))


def _run_with_bad_tool_args(tool_name, arguments):
    orch = MultiRoleOrchestrator(
        client=_fake_client([_tool_call_msg(tool_name, arguments), _final_msg()]),
        verbose=False, start_role="research")
    final = orch.run("查一下新能源汽车销量")
    tool_results = [m["content"] for m in orch.history if m["role"] == "tool"]
    return final, tool_results


def test_wrong_arg_name_returns_error_string_not_crash():
    final, tool_results = _run_with_bad_tool_args(
        "web_search", json.dumps({"q": "新能源汽车销量"}))
    assert final == FINAL_TEXT
    assert any("调用失败" in r for r in tool_results)


def test_missing_required_arg_returns_error_string_not_crash():
    final, tool_results = _run_with_bad_tool_args("web_search", "{}")
    assert final == FINAL_TEXT
    assert any("调用失败" in r for r in tool_results)


def test_non_numeric_stats_input_returns_error_string_not_crash():
    final, tool_results = _run_with_bad_tool_args(
        "descriptive_stats", json.dumps({"numbers": ["a", "b"]}))
    assert final == FINAL_TEXT
    assert any("调用失败" in r for r in tool_results)


def test_valid_tool_call_still_works():
    final, tool_results = _run_with_bad_tool_args(
        "web_search", json.dumps({"query": "新能源汽车 销量"}))
    assert final == FINAL_TEXT
    assert any("检索结果" in r for r in tool_results)

tools.py

"""
tools.py —— 各专业角色的专属工具实现 + OpenAI function-calling schema。

设计原则（配合实验 10-2）：
- 工具用「轻量真实实现」或「可控 mock」，重点不是工具多强，
  而是展示「自主角色移交」这一机制。
- research.web_search：内置知识库 mock（可控、可复现），
  未命中时诚实返回「未检索到」。
- coding.execute_python：真实执行 Python 代码并捕获标准输出（受限命名空间）。
- data_analysis.calculate / descriptive_stats：真实的安全计算。
- writing.count_characters：真实的中英文字数统计。

每个工具函数签名为 func(**kwargs) -> str（统一返回字符串，方便塞回对话历史）。
"""

from __future__ import annotations

import ast
import io
import math
import operator
import contextlib
from typing import Callable, Dict, List


# ---------------------------------------------------------------------------
# research 角色：web_search —— 内置知识库 mock
# ---------------------------------------------------------------------------

# 一个极小的「联网检索结果」知识库。命中关键词即返回内置资料，
# 保证 demo 可复现，同时不依赖真实外网。
_KNOWLEDGE_BASE = [
    {
        "keywords": ["新能源", "汽车", "销量", "nev", "电动车"],
        "content": (
            "【检索结果·中国乘用车市场信息联席会/中汽协】\n"
            "中国新能源汽车年度销量（单位：万辆）：\n"
            "- 2021 年：352.1 万辆\n"
            "- 2022 年：688.7 万辆\n"
            "- 2023 年：949.5 万辆\n"
            "备注：包含纯电动(BEV)与插电混动(PHEV)乘用车。"
        ),
    },
    {
        "keywords": ["光伏", "装机", "太阳能"],
        "content": (
            "【检索结果·国家能源局】\n"
            "中国光伏新增装机量（单位：GW）：\n"
            "- 2021 年：54.9 GW\n"
            "- 2022 年：87.4 GW\n"
            "- 2023 年：216.9 GW"
        ),
    },
    {
        "keywords": ["python", "gil", "全局解释器锁"],
        "content": (
            "【检索结果·技术资料】\n"
            "CPython 的 GIL(全局解释器锁)保证同一时刻只有一个线程执行字节码，"
            "因此 CPU 密集型任务用多线程无法并行，需改用多进程或 C 扩展。"
            "PEP 703 提出可选的 no-GIL 构建，Python 3.13 起以实验特性提供。"
        ),
    },
]


def web_search(query: str) -> str:
    """在内置知识库中检索（mock 联网检索）。"""
    q = (query or "").lower()
    hits: List[str] = []
    for entry in _KNOWLEDGE_BASE:
        if any(kw.lower() in q for kw in entry["keywords"]):
            hits.append(entry["content"])
    if hits:
        return "\n\n".join(hits)
    return (
        f"未在内置知识库中检索到与「{query}」直接相关的权威数据。"
        "请换一个更具体的检索词，或说明需要哪一年的数据。"
    )


# ---------------------------------------------------------------------------
# coding 角色：execute_python —— 真实执行代码并捕获 stdout
# ---------------------------------------------------------------------------

def execute_python(code: str) -> str:
    """在受限命名空间中执行 Python 代码，返回其标准输出。"""
    safe_globals: Dict[str, object] = {
        "__builtins__": {
            "print": print,
            "range": range,
            "len": len,
            "sum": sum,
            "min": min,
            "max": max,
            "abs": abs,
            "round": round,
            "sorted": sorted,
            "enumerate": enumerate,
            "zip": zip,
            "map": map,
            "list": list,
            "dict": dict,
            "tuple": tuple,
            "float": float,
            "int": int,
            "str": str,
        },
        "math": math,
    }
    buf = io.StringIO()
    try:
        with contextlib.redirect_stdout(buf):
            exec(code, safe_globals, {})  # noqa: S102 —— 受限命名空间下的教学示例
    except Exception as exc:  # noqa: BLE001
        return f"代码执行出错：{type(exc).__name__}: {exc}\n已捕获输出：\n{buf.getvalue()}"
    out = buf.getvalue().strip()
    return out if out else "（代码已执行，但没有任何 print 输出）"


# ---------------------------------------------------------------------------
# data_analysis 角色：calculate（安全表达式求值）+ descriptive_stats
# ---------------------------------------------------------------------------

_ALLOWED_OPERATORS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.Mod: operator.mod,
    ast.USub: operator.neg,
    ast.UAdd: operator.pos,
}


def _safe_eval(node: ast.AST) -> float:
    """只支持四则运算/幂/取模的安全表达式求值（不走 Python 内置 eval）。"""
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return float(node.value)
    if isinstance(node, ast.BinOp) and type(node.op) in _ALLOWED_OPERATORS:
        return _ALLOWED_OPERATORS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
    if isinstance(node, ast.UnaryOp) and type(node.op) in _ALLOWED_OPERATORS:
        return _ALLOWED_OPERATORS[type(node.op)](_safe_eval(node.operand))
    raise ValueError("表达式包含不被支持的运算，只允许 + - * / ** % 与括号。")


def calculate(expression: str) -> str:
    """安全地计算一个纯数学表达式，例如 (949.5/352.1)**(1/2)-1 。"""
    try:
        tree = ast.parse(expression, mode="eval")
        result = _safe_eval(tree.body)
    except Exception as exc:  # noqa: BLE001
        return f"计算失败：{exc}"
    return f"{expression} = {result}"


def descriptive_stats(numbers: List[float]) -> str:
    """给一组数值返回基本描述统计（均值/最大/最小/极差）。"""
    if not numbers:
        return "输入为空，无法统计。"
    nums = [float(x) for x in numbers]
    n = len(nums)
    mean = sum(nums) / n
    return (
        f"样本量={n}, 均值={mean:.4f}, 最小={min(nums)}, "
        f"最大={max(nums)}, 极差={max(nums) - min(nums)}"
    )


# ---------------------------------------------------------------------------
# writing 角色：count_characters —— 中英文字数统计
# ---------------------------------------------------------------------------

def count_characters(text: str) -> str:
    """统计文本的字符数与中文字符数，帮助控制篇幅。"""
    if text is None:
        text = ""
    total = len(text)
    chinese = sum(1 for ch in text if "一" <= ch <= "鿿")
    return f"总字符数={total}, 其中中文字符={chinese}"


# ---------------------------------------------------------------------------
# 工具注册表：名称 -> (实现函数, OpenAI schema)
# ---------------------------------------------------------------------------

# 每个工具的 OpenAI function-calling schema。
TOOL_SCHEMAS: Dict[str, dict] = {
    "web_search": {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "联网检索信息（本实验用内置知识库 mock）。用于查数据、事实、资料。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "检索关键词或问题"},
                },
                "required": ["query"],
            },
        },
    },
    "execute_python": {
        "type": "function",
        "function": {
            "name": "execute_python",
            "description": "执行一段 Python 代码并返回其 print 输出。适合写脚本、跑逻辑。",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "要执行的 Python 源码，用 print 输出结果"},
                },
                "required": ["code"],
            },
        },
    },
    "calculate": {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "安全计算一个数学表达式，支持 + - * / ** % 和括号。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式，如 (949.5/352.1)**(1/2)-1"},
                },
                "required": ["expression"],
            },
        },
    },
    "descriptive_stats": {
        "type": "function",
        "function": {
            "name": "descriptive_stats",
            "description": "对一组数值做基本描述统计（均值/最大/最小/极差）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "numbers": {
                        "type": "array",
                        "items": {"type": "number"},
                        "description": "数值数组",
                    },
                },
                "required": ["numbers"],
            },
        },
    },
    "count_characters": {
        "type": "function",
        "function": {
            "name": "count_characters",
            "description": "统计文本字符数与中文字符数，帮助控制篇幅。",
            "parameters": {
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "要统计的文本"},
                },
                "required": ["text"],
            },
        },
    },
}

# 工具名 -> 实现函数
TOOL_IMPLEMENTATIONS: Dict[str, Callable[..., str]] = {
    "web_search": web_search,
    "execute_python": execute_python,
    "calculate": calculate,
    "descriptive_stats": descriptive_stats,
    "count_characters": count_characters,
}