在 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 而言这意味着两处关键收益成本优化简单问题自动落到小模型如 Claude Haiku 级别复杂推理才使用大模型避免所有请求都按最贵模型计费性能提升路由端点按输入模式选型能在延迟与质量之间取得更好平衡。在多 Agent 系统中Prompt Routing 有两个天然的应用位点分类器Classifier意图识别本质上是一种「结构化输出」任务路由端点足以胜任且能显著降低高频分类调用的开销回复 AgentAgent当不同 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 PythonBoto3为 Bedrock 访问配置了合适的 IAM 权限至少包含bedrock:InvokeModel、bedrock:InvokeModelWithResponseStream等运行时调用权限以及读取账户信息的sts:GetCallerIdentity当前 AWS 区域已启用 Prompt Routing 与所需基础模型。三、安装依赖在项目环境如虚拟环境中执行pip install boto3 multi-agent-orchestratormulti-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.pyWelcome 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, flushTrue)该回调会被 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(fSelected 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_paramsNone) - 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( roleParticipantRole.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(optionsOrchestratorConfig( LOG_AGENT_CHATTrue, LOG_CLASSIFIER_CHATTrue, LOG_CLASSIFIER_RAW_OUTPUTTrue, LOG_CLASSIFIER_OUTPUTTrue, LOG_EXECUTION_TIMESTrue, MAX_RETRIES3, USE_DEFAULT_AGENT_IF_NONE_IDENTIFIEDTrue, MAX_MESSAGE_PAIRS_PER_AGENT10, ), classifierBedrockClassifier(BedrockClassifierOptions( model_idfarn: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:versionregion路由端点所在区域示例用us-east-1account_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( nameTech Agent, streamingTrue, descriptionSpecializes 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_idanthropic.claude-3-sonnet-20240229-v1:0, callbacksLLMAgentCallbacks() )) orchestrator.add_agent(tech_agent) health_agent BedrockLLMAgent(BedrockLLMAgentOptions( nameHealth Agent, streamingFalse, model_idfarn:aws:bedrock:us-east-1:{os.getenv(AWS_ACCOUNT_ID)}:default-prompt-router/anthropic.claude:1, descriptionSpecialized agent for giving health advice., callbacksLLMAgentCallbacks() )) orchestrator.add_agent(health_agent)两个 Agent 展示了两种模式Tech Agent显式指定anthropic.claude-3-sonnet-20240229-v1:0Claude 3 Sonnet适合对质量稳定性要求高的技术问答Health Agent同样使用 Prompt Routing ARN让 Bedrock 按健康咨询问题的复杂度自动选型且关闭流式streamingFalse与 Tech Agent 的流式输出形成对照。BedrockLLMAgentOptions还支持region、inference_config、guardrail_config、retriever、tool_config、custom_system_prompt等扩展选项见 bedrock_llm_agent.py默认推理参数为maxTokens1000、temperature0.0、topP0.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固定为user123SESSION_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], }, }几个值得注意的实现细节工具强制结构化输出分类器内置了名为analyzePrompt的工具toolSpec定义见 bedrock_classifier.py要求模型返回userinput、selected_agent、confidence三个字段。对于 Anthropic 与 Mistral Large 系列模型还会附加toolChoice强制模型必须调用该工具bedrock_classifier.py。该工具并不真正执行只是用来约束输出格式返回结果直接映射为ClassifierResultselected_agentconfidence。路由端点同样适用工具约束只要路由端点在 Anthropic 模型家族内选型上述toolChoice逻辑即可生效保证分类输出的结构化。分类器推理参数默认值maxTokens1000、temperature0.0、topP0.9、stopSequences[]bedrock_classifier.py。由于分类是确定性问题temperature0.0有利于获得稳定一致的意图判断。分类提示词模板默认系统提示词是「AgentMatcher」模板包含 Agent 描述注入{{AGENT_DESCRIPTIONS}}与会话历史注入{{HISTORY}}并内置「追问/续接沿用上一 Agent」的规则完整模板见 classifier.py。对路由端点而言toolConfig与inferenceConfig依然原样传递Bedrock 在路由层完成模型选型后由被选中的模型执行工具调用与生成。框架侧无需任何特殊分支这正是接入成本低的原因。七、Orchestrator 配置项速查示例中使用的OrchestratorConfig字段及默认值来源 types.py配置项默认值示例中的值作用LOG_AGENT_CHATFalseTrue打印每个 Agent 的对话历史LOG_CLASSIFIER_CHATFalseTrue打印送入分类器的会话历史LOG_CLASSIFIER_RAW_OUTPUTFalseTrue打印分类器原始模型输出LOG_CLASSIFIER_OUTPUTFalseTrue打印解析后的分类结果选中 Agent 与置信度LOG_EXECUTION_TIMESFalseTrue统计并输出分类与 Agent 推理耗时MAX_RETRIES33请求失败时的最大重试次数USE_DEFAULT_AGENT_IF_NONE_IDENTIFIEDTrueTrue分类失败时回退到默认 AgentMAX_MESSAGE_PAIRS_PER_AGENT10010每个 Agent 会话最多保留的消息轮数将前四个日志开关全部打开后终端会清晰呈现「输入文本 → 选中 Agent → 置信度」的分类过程对应print_intent实现见 orchestrator.py非常适合排查路由效果。八、最佳实践与注意事项区域一致性示例硬编码了us-east-1请确保该区域已开通 Prompt Routing且你的 AWS 凭据对应的账户与该区域一致如需其他区域同步修改 ARN 中的 region 段与AWS_DEFAULT_REGION环境变量。权限最小化运行示例至少需要sts:GetCallerIdentity与 Bedrock 运行时调用权限生产环境建议按需收紧。分类器与 Agent 的解耦意图分类是高频低复杂度调用放在 Prompt Routing 端点上可显著降低分类成本对质量敏感的回复场景可以像 Tech Agent 一样显式固定模型形成「分类走路由、关键回复锁模型」的混合策略。善用日志定位问题打开LOG_CLASSIFIER_OUTPUT与LOG_CLASSIFIER_RAW_OUTPUT可观察路由端点实际选型与置信度便于调优 Agent 的description文案——分类准确度在很大程度上依赖 Agent 描述的质量这一点在 bedrock-classifier.mdx 的 Limitations 一节也有说明。会话存储扩展示例使用默认内存存储重启即丢失生产部署可替换为仓库提供的 DynamoDB 存储 或 SQL 存储。流式与非流式并存同一 Orchestrator 中可混用streamingTrue/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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考