在运行时使用配置包
您的代理在运行时从捆绑包中读取配置以应用动态设置,而无需重新部署代码。网关和运行时通过 W3C 行李标头传播捆绑包引用,因此您的代理代码无需知道哪个版本处于活动状态;它会读取当前请求上下文中的任何配置。
注意
配置包集成需要bedrock-agentcore-sdk-python版本 1.8 或更高版本。从此版本起BedrockAgentCoreContext,可以使用 on get_config_bundle() 的方法。
网关注入捆绑包 ARN,并使用不同的捆绑包版本拆分传入流量。在同一个运行时运行的相同代理代码的行为会根据它收到的捆绑包版本而有所不同。
行李头部传播
当 A/B 测试处于活动状态时, AgentCore Gateway 会将每个会话分配给一个变体,并将相应的配置包引用作为 W3CBedrockAgentCoreContext将包配置提供给您的代理代码。
行李里有两把钥匙:
-
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(payload.get("prompt", "Hello")) return {"response": str(result)} if __name__ == "__main__": app.run()
该BeforeModelCallEvent挂钩会在每次 LLM 调用之前触发,因此配置包更改无需重新启动运行时即可立即生效。
使用按请求构造分散代理
如果您需要应用更多配置字段(型号 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(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": 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(payload.get("prompt", "Hello")) return {"response": str(result)} if __name__ == "__main__": app.run()
OpenAI SDK 代理
对于使用带有 Amazon Bedrock 的 OpenAI SDK 的代理,请阅读配置包以设置模型和系统提示:
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": 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}