用 AWS Multi-Agent Orchestrator 做多智能体路由编排：分类器自动派单实操

框架本体用 pip 安装，[anthropic] 额外依赖会拉入官方 Anthropic SDK，用于分类器与 Agent。

# 建并激活虚拟环境（Windows 用 venv\Scripts\activate）
python -m venv venv
source venv/bin/activate

# 安装 Anthropic 后端（分类器 + Agent 都走 Anthropic）
pip install "multi-agent-orchestrator[anthropic]"

# 准备密钥：导出 Anthropic API Key
export ANTHROPIC_API_KEY="sk-ant-xxxxxxxxxxxxxxxx"

用 MultiAgentOrchestrator 初始化编排器，并传入 OrchestratorConfig。先把分类日志打开，方便后面排查「为什么分错了 Agent」。

import asyncio
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig

orchestrator = MultiAgentOrchestrator(
    options=OrchestratorConfig(
        LOG_AGENT_CHAT=True,
        LOG_CLASSIFIER_CHAT=True,
        LOG_CLASSIFIER_OUTPUT=True,   # 打印每次选中的 Agent 与置信度
        LOG_EXECUTION_TIMES=True,
        MAX_RETRIES=3,
        USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
        MAX_MESSAGE_PAIRS_PER_AGENT=10,
    )
)

每个 AnthropicAgent 用 name + description 描述职责边界。描述写得越具体、彼此越不重叠，路由越准。

from multi_agent_orchestrator.agents import AnthropicAgent, AnthropicAgentOptions
from multi_agent_orchestrator.classifiers import AnthropicClassifier, AnthropicClassifierOptions

# 分类器也用 Anthropic（替代默认的 Bedrock）
orchestrator = MultiAgentOrchestrator(
    options=OrchestratorConfig(LOG_CLASSIFIER_OUTPUT=True),
    classifier=AnthropicClassifier(
        options=AnthropicClassifierOptions(api_key="sk-ant-xxxxxxxxxxxxxxxx")
    ),
)

tech_agent = AnthropicAgent(AnthropicAgentOptions(
    name="Tech Agent",
    description=("专门解答软件架构、编程语言、云基础设施与调试类问题，"
                 "不处理订单、物流或账单等商务问题。"),
    api_key="sk-ant-xxxxxxxxxxxxxxxx",
    model_id="claude-3-5-sonnet-20240620",
    streaming=False,
))

biz_agent = AnthropicAgent(AnthropicAgentOptions(
    name="Biz Agent",
    description=("处理订单状态、退换货政策、物流与发票等商务咨询，"
                 "不回答技术实现细节。"),
    api_key="sk-ant-xxxxxxxxxxxxxxxx",
    model_id="claude-3-5-sonnet-20240620",
    streaming=False,
))

用 add_agent 注册，再用 route_request(text, user_id=..., session_id=...) 发起请求。分类器会选中一个 Agent 并把回答写到该会话历史。

orchestrator.add_agent(tech_agent)
orchestrator.add_agent(biz_agent)

async def main():
    response = await orchestrator.route_request(
        "我的 Kubernetes Pod 一直 CrashLoopBackOff，怎么排查？",
        user_id="u1",
        session_id="s1",
    )
    print("由哪个 Agent 回答：", response.metadata.agent_name)
    print("回答内容：", response.output.content)

asyncio.run(main())

连发两条不同领域的问题，再发一句短追问，确认：① 各自落到正确的 Agent；② 短追问被送回上一个还在等待的 Agent（上下文隔离但连续）。

async def main():
    # 技术类
    r1 = await orchestrator.route_request("Pod 一直重启怎么办？", user_id="u1", session_id="s1")
    print(r1.metadata.agent_name)   # -> Tech Agent

    # 商务类
    r2 = await orchestrator.route_request("我的订单能开发票吗？", user_id="u1", session_id="s2")
    print(r2.metadata.agent_name)   # -> Biz Agent

    # 同一会话的短追问，应回到 Tech Agent 而非重新分类
    r3 = await orchestrator.route_request("那日志里一直 OOMKilled 呢？", user_id="u1", session_id="s1")
    print(r3.metadata.agent_name)   # -> Tech Agent

asyncio.run(main())

当分类器无法确定归属时，回退到 default_agent 比直接报错更友好。配合 USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED 即可。

from multi_agent_orchestrator.agents import AnthropicAgent, AnthropicAgentOptions

default_agent = AnthropicAgent(AnthropicAgentOptions(
    name="Default Agent",
    description="处理未被其他专职 Agent 覆盖的通用问题。",
    api_key="sk-ant-xxxxxxxxxxxxxxxx",
    model_id="claude-3-5-sonnet-20240620",
))

orchestrator = MultiAgentOrchestrator(
    options=OrchestratorConfig(
        USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
        NO_SELECTED_AGENT_MESSAGE="这个问题我不太确定归谁管，麻烦再描述详细一点？",
    ),
    classifier=AnthropicClassifier(options=AnthropicClassifierOptions(api_key="sk-ant-xxxxxxxxxxxxxxxx")),
    default_agent=default_agent,
)