View a markdown version of this page

在运行时使用配置包 - 亚马逊基岩 AgentCore

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

在运行时使用配置包

您的代理在运行时从包中读取配置以应用动态设置,而无需重新部署代码。网关和运行时通过 W3C 行李标头传播包引用,因此您的代理代码无需知道哪个版本处于活动状态;它会读取当前请求上下文中的任何配置。

注意

配置包集成需要bedrock-agentcore-sdk-python版本 1.8 或更高版本。该get_config_bundle()方法从此版本开始可用。BedrockAgentCoreContext

网关注入捆绑包 ARN,并使用不同的捆绑包版本拆分传入流量。相同的代理代码在同一个运行时运行,其行为因其接收的包版本而异。

行李标题传播

当 A/B 测试处于活动状态时, AgentCore Gateway 会将每个会话分配给一个变体,并将相应的配置包引用作为 W3C Baggage 标头注入到请求中。运行时会自动解析此标头,并通过BedrockAgentCoreContext以下方式将捆绑配置提供给您的代理代码。

行李包含两把钥匙:

  • aws.agentcore.configbundle_arn— 配置包的完整 ARN

  • aws.agentcore.configbundle_version— 捆绑包的版本 ID

在直接调用代理时(例如,在测试期间),您也可以手动传递行李:

import boto3 import json import uuid rt_client = boto3.client("bedrock-agentcore", region_name="us-west-2") BUNDLE_ARN = "arn:aws:bedrock-agentcore:us-west-2:123456789012:configuration-bundle/myAgentConfig-a1b2c3d4e5" BUNDLE_VERSION = "12345678-1234-1234-1234-123456789012" baggage = ( f"aws.agentcore.configbundle_arn={BUNDLE_ARN}," f"aws.agentcore.configbundle_version={BUNDLE_VERSION}" ) response = rt_client.invoke_agent_runtime( agentRuntimeArn="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/MyAgent-abc123", runtimeSessionId=str(uuid.uuid4()), payload=json.dumps({"prompt": "What is the status of order ORD-1001?"}).encode(), baggage=baggage, ) print(response["response"].read().decode("utf-8"))

在生产中,您无需手动构造行李。网关在 A/B 测试期间会自动处理此问题。

BedrockAgentCoreContext 整合

该BedrockAgentCoreContext类(来自 bedrock-agentcore SDK)提供了一get_config_bundle()种返回当前请求配置的方法。会BedrockAgentCoreApp自动解析行李标头,解析控制平面 API 中的捆绑版本并缓存结果。

from bedrock_agentcore.runtime import BedrockAgentCoreContext # Returns the configuration dict for your component, or {} if no bundle is in context config = BedrockAgentCoreContext.get_config_bundle() system_prompt = config.get("system_prompt", "You are a helpful assistant.") model_id = config.get("model_id", "global.anthropic.claude-sonnet-4-5-20250929-v1:0")

get_config_bundle()返回与您的运行时 ARN 相匹配的组件的configuration对象。如果请求中不存在捆绑引用(例如,当没有 A/B 测试处于活动状态且未通过任何行李时),它将返回一个空字典。

你也可以查看原始包参考资料:

ref = BedrockAgentCoreContext.get_config_bundle_ref() if ref: print(f"Bundle ID: {ref.bundle_id}") print(f"Bundle ARN: {ref.bundle_arn}") print(f"Version: {ref.bundle_version}")

带挂钩的 BeforeModelCallEvent 绞线剂

Strands 代理的推荐模式是使用BeforeModelCallEvent挂钩,在每次调用模型之前动态更新代理的系统提示符。代理在模块级别创建一次,挂钩会根据请求对其进行修改:

from strands import Agent from strands.models.bedrock import BedrockModel from strands.hooks.events import BeforeModelCallEvent from bedrock_agentcore.runtime import BedrockAgentCoreApp, BedrockAgentCoreContext app = BedrockAgentCoreApp() DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0" DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant." def dynamic_config_hook(event: BeforeModelCallEvent): """Read config bundle and apply system prompt before every model call.""" config_bundle = BedrockAgentCoreContext.get_config_bundle() event.agent.system_prompt = config_bundle.get("system_prompt", DEFAULT_SYSTEM_PROMPT) agent = Agent( model=BedrockModel(model_id=DEFAULT_MODEL_ID), system_prompt=DEFAULT_SYSTEM_PROMPT, ) agent.hooks.add_callback(BeforeModelCallEvent, dynamic_config_hook) @app.entrypoint def invoke(payload, context): result = agent(str(payload.get("prompt", "Hello"))) return {"response": str(result)} if __name__ == "__main__": app.run()

BeforeModelCallEvent挂钩会在每次 LLM 调用之前触发,因此配置包更改会立即生效,无需重新启动运行时。

按请求构造的 Strands 代理

如果您需要应用更多配置字段(模型 ID、温度、工具),请为每个请求构建一个新的代理,而不是使用挂钩:

from strands import Agent from strands.models.bedrock import BedrockModel from bedrock_agentcore.runtime import BedrockAgentCoreApp, BedrockAgentCoreContext app = BedrockAgentCoreApp() DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0" DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant." def build_agent() -> Agent: """Build a fresh agent per request with config bundle applied.""" config = BedrockAgentCoreContext.get_config_bundle() model_id = config.get("model_id", DEFAULT_MODEL_ID) system_prompt = config.get("system_prompt", DEFAULT_SYSTEM_PROMPT) model_kwargs = {"model_id": model_id} temperature = config.get("temperature") if temperature is not None: model_kwargs["temperature"] = temperature return Agent( model=BedrockModel(**model_kwargs), system_prompt=system_prompt, ) @app.entrypoint def invoke(payload, context): agent = build_agent() result = agent(str(payload.get("prompt", "Hello"))) return {"response": str(result)} if __name__ == "__main__": app.run()

LangGraph 代理人

对于 LangGraph 代理,请在每次调用开始时读取配置包并将值传递给图表:

from langchain_aws import ChatBedrock from langgraph.graph import StateGraph, MessagesState, START, END from bedrock_agentcore.runtime import BedrockAgentCoreApp, BedrockAgentCoreContext app = BedrockAgentCoreApp() DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0" DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant." def build_graph(): """Build a LangGraph graph with config bundle applied.""" config = BedrockAgentCoreContext.get_config_bundle() model_id = config.get("model_id", DEFAULT_MODEL_ID) system_prompt = config.get("system_prompt", DEFAULT_SYSTEM_PROMPT) model = ChatBedrock(model_id=model_id) def call_model(state: MessagesState): messages = [{"role": "system", "content": system_prompt}] + state["messages"] response = model.invoke(messages) return {"messages": [response]} graph = StateGraph(MessagesState) graph.add_node("model", call_model) graph.add_edge(START, "model") graph.add_edge("model", END) return graph.compile() @app.entrypoint def invoke(payload, context): graph = build_graph() result = graph.invoke({"messages": [{"role": "user", "content": str(payload.get("prompt", "Hello"))}]}) return {"response": result["messages"][-1].content} if __name__ == "__main__": app.run()

谷歌 ADK 代理

对于使用 Google 代理开发套件 (ADK) 构建的代理,请在构造代理时阅读配置包:

from google.adk.agents import LlmAgent from google.adk.models.lite_llm import LiteLlm from bedrock_agentcore.runtime import BedrockAgentCoreApp, BedrockAgentCoreContext app = BedrockAgentCoreApp() DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0" DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant." def build_agent() -> LlmAgent: """Build an ADK agent with config bundle applied.""" config = BedrockAgentCoreContext.get_config_bundle() model_id = config.get("model_id", DEFAULT_MODEL_ID) system_prompt = config.get("system_prompt", DEFAULT_SYSTEM_PROMPT) return LlmAgent( name="my_agent", model=LiteLlm(model=f"bedrock/{model_id}"), instruction=system_prompt, ) @app.entrypoint def invoke(payload, context): agent = build_agent() # ADK agent invocation logic result = agent.invoke(str(payload.get("prompt", "Hello"))) return {"response": str(result)} if __name__ == "__main__": app.run()

openAI SDK 代理

对于在亚马逊 Bedrock 中使用 OpenAI 软件开发工具包的代理,请阅读配置包以设置模型和系统提示:

from openai import OpenAI from bedrock_agentcore.runtime import BedrockAgentCoreApp, BedrockAgentCoreContext app = BedrockAgentCoreApp() DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0" DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant." client = OpenAI() @app.entrypoint def invoke(payload, context): config = BedrockAgentCoreContext.get_config_bundle() model_id = config.get("model_id", DEFAULT_MODEL_ID) system_prompt = config.get("system_prompt", DEFAULT_SYSTEM_PROMPT) response = client.chat.completions.create( model=model_id, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": str(payload.get("prompt", "Hello"))}, ], ) return {"response": response.choices[0].message.content} if __name__ == "__main__": app.run()

优雅的后备

读取捆绑配置时,请务必提供默认值。这样可以确保即使在没有 A/B 测试处于活动状态、捆绑包提取失败或捆绑包不包含预期密钥的情况下,您的代理也能正常运行。

get_config_bundle()当不存在捆绑引用时,返回一个空字典。如果底层 API 调用失败,异常就会传播。将通话打包成 try/except 优雅降级:

from bedrock_agentcore.runtime import BedrockAgentCoreContext DEFAULT_SYSTEM_PROMPT = "You are a helpful customer support assistant." DEFAULT_MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0" def get_config_with_fallback(): """Read config bundle with graceful fallback to defaults.""" try: config = BedrockAgentCoreContext.get_config_bundle() if not config: return {"system_prompt": DEFAULT_SYSTEM_PROMPT, "model_id": DEFAULT_MODEL_ID} return config except Exception: return {"system_prompt": DEFAULT_SYSTEM_PROMPT, "model_id": DEFAULT_MODEL_ID}