企业官网建设流程全解析

在 multi-agent-orchestrator 中使用 Amazon Bedrock Prompt Routing 优化模型选择与成本

【免费下载链接】agent-squadFlexible and powerful framework for managing multiple AI agents and handling complex conversations项目地址: https://gitcode.com/GitHub_Trending/mu/agent-squad

本指南基于仓库中的 examples/bedrock-prompt-routing/readme.md 与配套示例 examples/bedrock-prompt-routing/main.py，完整演示如何将 Amazon Bedrock Prompt Routing 能力接入 multi-agent-orchestrator 框架：把default-prompt-router的模型 ARN 同时用作BedrockClassifier的意图识别模型与BedrockLLMAgent的回复模型，从而根据输入模式自动优化模型选型，在保证回复质量的同时降低推理成本。读完本文，你将掌握 Prompt Routing 模型 ARN 的构造方式、在分类器与 Agent 两端的接入方法、完整可运行示例的逐段解析，以及 Orchestrator 配置项与底层源码调用链。

一、Bedrock Prompt Routing 是什么，为什么适合多 Agent 编排

Amazon Bedrock Prompt Routing 是 Bedrock 提供的一项模型路由能力：你无需为每条请求手动指定具体模型，而是向一个「路由端点」发送请求，由 Bedrock 根据输入内容的特点自动将请求分发到最合适的模型上。对 multi-agent-orchestrator 而言，这意味着两处关键收益：

1. 成本优化：简单问题自动落到小模型（如 Claude Haiku 级别），复杂推理才使用大模型，避免所有请求都按最贵模型计费；
2. 性能提升：路由端点按输入模式选型，能在延迟与质量之间取得更好平衡。

在多 Agent 系统中，Prompt Routing 有两个天然的应用位点：

• 分类器（Classifier）：意图识别本质上是一种「结构化输出」任务，路由端点足以胜任，且能显著降低高频分类调用的开销；
• 回复 Agent（Agent）：当不同 Agent 面向不同专业领域时，路由端点可以让每个 Agent 内部也享受自动选型的能力。

本仓库示例恰好演示了这两种用法同时存在的场景：分类器与健康咨询 Agent 都使用default-prompt-router路由端点，而技术 Agent 显式指定 Claude 3 Sonnet。

二、前置条件

运行本示例前需要准备以下环境（见 examples/bedrock-prompt-routing/readme.md）：

• 一个已开通 Amazon Bedrock 访问权限的 AWS 账户；
• Python 3.11 或更高版本；
• 已安装 AWS SDK for Python（Boto3）；
• 为 Bedrock 访问配置了合适的 IAM 权限（至少包含bedrock:InvokeModel、bedrock:InvokeModelWithResponseStream等运行时调用权限，以及读取账户信息的sts:GetCallerIdentity）；
• 当前 AWS 区域已启用 Prompt Routing 与所需基础模型。

三、安装依赖

在项目环境（如虚拟环境）中执行：

pip install boto3 multi-agent-orchestrator

multi-agent-orchestrator即本仓库 Python 侧发布包名（对应源码位于 python/src/multi_agent_orchestrator，构建配置见 python/pyproject.toml）。若后续还需要 Bedrock 以外的额外依赖，可参考文档 bedrock-classifier.mdx 中的pip install "multi-agent-orchestrator[aws]"方式。

示例代码在构造路由 ARN 时需要当前账户 ID，因此先导出环境变量：

export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)

这一步执行后，AWS_ACCOUNT_ID将包含类似123456789012的 12 位账户 ID，供下面的 ARN 模板使用。

四、运行示例

进入示例目录后直接运行：

python main.py

程序会启动一个交互式多 Agent 会话（源码见 main.py）：

Welcome to the interactive Multi-Agent system. Type 'quit' to exit. You:

在You:提示符后输入问题并按回车，即可观察 Orchestrator 完成「意图分类 → 路由到 Agent → 流式/非流式回复」的完整链路；输入quit退出程序。

五、示例核心代码逐段解析

5.1 引入模块与回调定义

from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig from multi_agent_orchestrator.agents import (BedrockLLMAgent, BedrockLLMAgentOptions, AgentResponse, AgentCallbacks) from multi_agent_orchestrator.types import ConversationMessage, ParticipantRole from multi_agent_orchestrator.classifiers import BedrockClassifier, BedrockClassifierOptions

其中LLMAgentCallbacks继承了AgentCallbacks，通过on_llm_new_token实现流式 token 的实时打印：

class LLMAgentCallbacks(AgentCallbacks): def on_llm_new_token(self, token: str) -> None: print(token, end='', flush=True)

该回调会被 bedrock_llm_agent.py 的handle_streaming_response在解析contentBlockDelta时逐 token 调用，与 Bedrock 的converse_streamAPI 一一对应。

5.2 请求处理与元数据打印

async def handle_request(_orchestrator, _user_input, _user_id, _session_id): response:AgentResponse = await _orchestrator.route_request(_user_input, _user_id, _session_id) print("\nMetadata:") print(f"Selected Agent: {response.metadata.agent_name}") if isinstance(response, AgentResponse) and response.streaming is False: if isinstance(response.output, str): print(response.output) elif isinstance(response.output, ConversationMessage): print(response.output.content[0].get('text'))

这里调用的route_request是 Orchestrator 的核心入口，其执行顺序可在 orchestrator.py 中确认：先classify_request做意图分类，再dispatch_to_agent分发到选中 Agent，最后agent_process_request负责保存会话消息并组装AgentResponse元数据。注意流式场景（streaming is True）下回复已经通过回调打印，这里只打印非流式输出。

5.3 自定义 payload 编解码函数（示例中预置，对应 LambdaAgent 选项）

示例定义了下面两个函数：

def custom_input_payload_encoder(input_text, chat_history, user_id, session_id, additional_params=None) -> str: return json.dumps({'hello':'world'}) def custom_output_payload_decoder(response: dict[str, Any]) -> Any: decoded_response = json.loads( json.loads( response['Payload'].read().decode('utf-8') )['body'])['response'] return ConversationMessage( role=ParticipantRole.ASSISTANT.value, content=[{'text': decoded_response}] )

它们在示例中属于「预置但未被本文件使用」的辅助函数，其签名分别对应框架中LambdaAgentOptions的input_payload_encoder与output_payload_decoder两个选项——前者把请求参数编码为发送给 Lambda 的 JSON payload，后者把 Lambda 返回的Payload解码为ConversationMessage。相关实现可参考 lambda_agent.py。如果你后续把某个 Agent 换成 Lambda 后端，可以直接复用这两个函数。

5.4 Orchestrator 与 BedrockClassifier 接入路由端点

orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig( LOG_AGENT_CHAT=True, LOG_CLASSIFIER_CHAT=True, LOG_CLASSIFIER_RAW_OUTPUT=True, LOG_CLASSIFIER_OUTPUT=True, LOG_EXECUTION_TIMES=True, MAX_RETRIES=3, USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True, MAX_MESSAGE_PAIRS_PER_AGENT=10, ), classifier=BedrockClassifier(BedrockClassifierOptions( model_id=f"arn:aws:bedrock:us-east-1:{os.getenv('AWS_ACCOUNT_ID')}:default-prompt-router/anthropic.claude:1")) )

这是整个示例最关键的一行：分类器的model_id不是普通模型 ID，而是一个Prompt Routing 端点 ARN，其通用模板为：

arn:aws:bedrock:<region>:<account_id>:default-prompt-router/<model-family>:<version>

• region：路由端点所在区域，示例用us-east-1；
• account_id：由AWS_ACCOUNT_ID环境变量注入；
• default-prompt-router/anthropic.claude:1：指向「Anthropic Claude 系列默认路由」，Bedrock 会按输入自动在 Claude 各型号间选型。

OrchestratorConfig各字段的默认值与含义见 types.py，其中MAX_RETRIES默认 3、USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED默认 True、MAX_MESSAGE_PAIRS_PER_AGENT默认 100，示例中显式改为 10，用于限制每个 Agent 会话保留的历史消息轮数。

5.5 两类 Agent：显式模型与路由端点混用

tech_agent = BedrockLLMAgent(BedrockLLMAgentOptions( name="Tech Agent", streaming=True, description="Specializes in technology areas including software development, hardware, AI, \ cybersecurity, blockchain, cloud computing, emerging tech innovations, and pricing/costs \ related to technology products and services.", model_id="anthropic.claude-3-sonnet-20240229-v1:0", callbacks=LLMAgentCallbacks() )) orchestrator.add_agent(tech_agent) health_agent = BedrockLLMAgent(BedrockLLMAgentOptions( name="Health Agent", streaming=False, model_id=f"arn:aws:bedrock:us-east-1:{os.getenv('AWS_ACCOUNT_ID')}:default-prompt-router/anthropic.claude:1", description="Specialized agent for giving health advice.", callbacks=LLMAgentCallbacks() )) orchestrator.add_agent(health_agent)

两个 Agent 展示了两种模式：

• Tech Agent：显式指定anthropic.claude-3-sonnet-20240229-v1:0（Claude 3 Sonnet），适合对质量稳定性要求高的技术问答；
• Health Agent：同样使用 Prompt Routing ARN，让 Bedrock 按健康咨询问题的复杂度自动选型，且关闭流式（streaming=False），与 Tech Agent 的流式输出形成对照。

BedrockLLMAgentOptions还支持region、inference_config、guardrail_config、retriever、tool_config、custom_system_prompt等扩展选项（见 bedrock_llm_agent.py），默认推理参数为maxTokens=1000、temperature=0.0、topP=0.9、stopSequences=[]，可通过inference_config覆盖。

5.6 交互主循环

USER_ID = "user123" SESSION_ID = str(uuid.uuid4()) while True: user_input = input("\nYou: ").strip() if user_input.lower() == 'quit': print("Exiting the program. Goodbye!") sys.exit() asyncio.run(handle_request(orchestrator, user_input, USER_ID, SESSION_ID))

• USER_ID固定为user123，SESSION_ID每次启动生成新的 UUID，二者共同作为会话存储的键（框架默认使用InMemoryChatStorage，见 orchestrator.py）；
• 每次输入通过asyncio.run驱动一次完整的异步路由请求。

六、底层原理：分类器如何调用 Prompt Routing 端点

将路由 ARN 作为model_id传给BedrockClassifier后，请求会走标准 Converse API。在 bedrock_classifier.py 中，process_request构造的请求体包括：

converse_cmd = { "modelId": self.model_id, # 此处即 default-prompt-router ARN "messages": [user_message.__dict__], "system": [{"text": self.system_prompt}], "toolConfig": toolConfig, "inferenceConfig": { "maxTokens": self.inference_config['maxTokens'], "temperature": self.inference_config['temperature'], "topP": self.inference_config['topP'], "stopSequences": self.inference_config['stopSequences'], }, }

几个值得注意的实现细节：

1. 工具强制结构化输出：分类器内置了名为analyzePrompt的工具（toolSpec定义见 bedrock_classifier.py），要求模型返回userinput、selected_agent、confidence三个字段。对于 Anthropic 与 Mistral Large 系列模型，还会附加toolChoice强制模型必须调用该工具（bedrock_classifier.py）。该工具并不真正执行，只是用来约束输出格式，返回结果直接映射为ClassifierResult（selected_agent+confidence）。
2. 路由端点同样适用工具约束：只要路由端点在 Anthropic 模型家族内选型，上述toolChoice逻辑即可生效，保证分类输出的结构化。
3. 分类器推理参数默认值：maxTokens=1000、temperature=0.0、topP=0.9、stopSequences=[]（bedrock_classifier.py）。由于分类是确定性问题，temperature=0.0有利于获得稳定一致的意图判断。
4. 分类提示词模板：默认系统提示词是「AgentMatcher」模板，包含 Agent 描述注入（{{AGENT_DESCRIPTIONS}}）与会话历史注入（{{HISTORY}}），并内置「追问/续接沿用上一 Agent」的规则，完整模板见 classifier.py。

对路由端点而言，toolConfig与inferenceConfig依然原样传递，Bedrock 在路由层完成模型选型后，由被选中的模型执行工具调用与生成。框架侧无需任何特殊分支，这正是接入成本低的原因。

七、Orchestrator 配置项速查

示例中使用的OrchestratorConfig字段及默认值（来源 types.py）：

将前四个日志开关全部打开后，终端会清晰呈现「输入文本 → 选中 Agent → 置信度」的分类过程（对应print_intent实现，见 orchestrator.py），非常适合排查路由效果。

八、最佳实践与注意事项

1. 区域一致性：示例硬编码了us-east-1，请确保该区域已开通 Prompt Routing，且你的 AWS 凭据对应的账户与该区域一致；如需其他区域，同步修改 ARN 中的 region 段与AWS_DEFAULT_REGION环境变量。
2. 权限最小化：运行示例至少需要sts:GetCallerIdentity与 Bedrock 运行时调用权限；生产环境建议按需收紧。
3. 分类器与 Agent 的解耦：意图分类是高频低复杂度调用，放在 Prompt Routing 端点上可显著降低分类成本；对质量敏感的回复场景可以像 Tech Agent 一样显式固定模型，形成「分类走路由、关键回复锁模型」的混合策略。
4. 善用日志定位问题：打开LOG_CLASSIFIER_OUTPUT与LOG_CLASSIFIER_RAW_OUTPUT可观察路由端点实际选型与置信度，便于调优 Agent 的description文案——分类准确度在很大程度上依赖 Agent 描述的质量（这一点在 bedrock-classifier.mdx 的 Limitations 一节也有说明）。
5. 会话存储扩展：示例使用默认内存存储，重启即丢失；生产部署可替换为仓库提供的 DynamoDB 存储 或 SQL 存储。
6. 流式与非流式并存：同一 Orchestrator 中可混用streaming=True/False的 Agent，框架通过AgentResponse.streaming字段区分处理方式，示例 5.2 节已展示对应分支。

九、小结

本示例用不足 110 行的 Python 代码，演示了 multi-agent-orchestrator 与 Amazon Bedrock Prompt Routing 的完整集成路径：通过default-prompt-routerARN 同时驱动BedrockClassifier的意图识别与BedrockLLMAgent的领域回复，实现按输入模式自动选型。结合框架源码可以看到，这一能力建立在 Converse API 与工具约束的结构化输出之上，无需修改框架内部即可获得路由收益。若需在 TypeScript 侧实现同等能力，可在 typescript/src/classifiers/bedrockClassifier.ts 与 typescript/src/agents/bedrockLLMAgent.ts 中找到对应的BedrockClassifier与BedrockLLMAgent实现，将同样的路由 ARN 传入modelId即可。

【免费下载链接】agent-squadFlexible and powerful framework for managing multiple AI agents and handling complex conversations项目地址: https://gitcode.com/GitHub_Trending/mu/agent-squad