本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
调用 AgentCore 运行时代理
该InvokeAgentRuntime操作允许您向由其 Amazon 资源名称 (ARN) 标识的特定 AgentCore 运行时终端节点发送请求,并接收包含代理输出的流式响应。该 API 支持通过会话标识符进行会话管理,使您能够维护多次交互中的对话上下文。您可以使用可选的限定符来定位特定的代理端点。
要拨打电话InvokeAgentRuntime,你需要bedrock-agentcore:InvokeAgentRuntime权限。在通话中,您还可以传递代理可用于用户身份验证的持有者令牌。
该InvokeAgentRuntime操作将您的请求负载作为二进制数据接受,大小不超过 100 MB,并返回流式响应,该响应在代理处理您的请求时实时传送数据块。这种流式传输方法使您可以立即收到部分结果,而不必等待完整的响应,因此非常适合交互式应用程序。
要在同一个会话中执行 shell 命令(例如运行测试、git 操作或环境设置),请使用 “在 AgentCore 运行时会话中执行 shell 命令” 操作。这两个操作都在同一个代理运行时和会话上运行。
如果您计划将代理与 OAuth 集成,则无法使用 AWS SDK 进行调用。InvokeAgentRuntime改为向发出 HTTPS 请求 InvokeAgentRuntime。有关更多信息,请参阅使用入站身份验证和出站身份验证进行身份验证和授权。
调用流媒体代理
以下示例显示如何使用 boto3 调用代理运行时:
import boto3 import json # Initialize the Bedrock AgentCore client agent_core_client = boto3.client('bedrock-agentcore') # Prepare the payload payload = json.dumps({"prompt": prompt}).encode() # Invoke the agent response = agent_core_client.invoke_agent_runtime( agentRuntimeArn=agent_arn, runtimeSessionId=session_id, payload=payload ) # Process and print the response if "text/event-stream" in response.get("contentType", ""): # Handle streaming response content = [] for line in response["response"].iter_lines(chunk_size=10): if line: line = line.decode("utf-8") if line.startswith("data: "): line = line[6:] print(line) content.append(line) print("\nComplete response:", "\n".join(content)) elif response.get("contentType") == "application/json": # Handle standard JSON response content = [] for chunk in response.get("response", []): content.append(chunk.decode('utf-8')) print(json.loads(''.join(content))) else: # Print raw response for other content types print(response)
调用多模式代理
您可以使用该InvokeAgentRuntime操作发送包含文本和图像的多模式请求。以下示例显示如何调用多模式代理:
import boto3 import json import base64 # Read and encode image with open("image.jpg", "rb") as image_file: image_data = base64.b64encode(image_file.read()).decode('utf-8') # Prepare multi-modal payload payload = json.dumps({ "prompt": "Describe what you see in this image", "media": { "type": "image", "format": "jpeg", "data": image_data } }).encode() # Invoke the agent response = agent_core_client.invoke_agent_runtime( agentRuntimeArn=agent_arn, runtimeSessionId=session_id, payload=payload )
会话管理
该InvokeAgentRuntime操作通过runtimeSessionId参数支持会话管理。通过在多个请求中提供相同的会话标识符,您可以维护对话上下文,从而允许代理引用以前的交互。
要开始新的对话,请生成一个唯一的会话标识符。要继续现有对话,请使用与先前请求相同的会话标识符。这种方法使您能够构建可随时间推移保持上下文的交互式应用程序。
提示
为获得最佳结果,请为会话 ID 使用 UUID 或其他唯一标识符,以避免不同用户或对话之间发生冲突。
错误处理
使用该InvokeAgentRuntime操作时,您可能会遇到各种错误。以下是一些常见错误及其处理方法:
- ValidationException
-
当请求参数无效时发生。检查您的代理 ARN、会话 ID 和负载的格式是否正确。
- ResourceNotFoundException
-
当找不到指定的代理运行时发生。验证代理 ARN 是否正确以及您的 AWS 账户中是否存在该代理。
- AccessDeniedException
-
当你没有必要的权限时发生。确保您的 IAM 政策包含
bedrock-agentcore:InvokeAgentRuntime权限。 - ThrottlingException
-
当您超过请求速率限制时发生。在应用程序中实现指数退避和重试逻辑。
- RetryableConflictException
-
当服务正在配置或关闭该会话时,第二个操作以会话为目标时发生 (HTTP 409)。消息是
Session operation in progress, please retry。这种情况是暂时的,可以重试。窗口很短,已经运行的会话不受影响。使用短暂的指数退避重试。启用默认重试后, AWS SDK 会自动重试此异常。如果您禁用了重试或直接调用 API,请自己重试。
在应用程序中实施适当的错误处理,以提供更好的用户体验并有效地解决问题。
最佳实践
使用该InvokeAgentRuntime操作时,请遵循以下最佳实践:
-
验证提示字段是代理入口点中的字符串 ——有效负载以解析的 JSON 形式到达,因此该
prompt字段可以是任何 JSON 类型(字符串、列表、对象)。如果包含toolUse内容块的非字符串值到达您的代理框架,则该框架可能会直接执行命名工具。绕过了模型推理和护栏评估。isinstance(prompt, str)在将输入传递给代理之前,请务必强制执行。有关更多信息,请参阅 AgentCore 运行时安全最佳实践。 -
使用会话管理来维护对话上下文,以获得更好的用户体验。
-
逐步处理流媒体响应,为用户提供实时反馈。
-
为强大的应用程序实现正确的错误处理和重试逻辑。
-
发送请求时,请考虑有效负载大小限制(100 MB),尤其是对于多模态内容。
-
使用适当的限定符来定位特定的代理版本或端点。
-
必要时使用持有者令牌实施身份验证机制。
-
InvokeAgentRuntimeCommand用于确定性操作(测试、git、构建),而不是通过代理的 LLM 进行路由。请参见在 AgentCore 运行时会话中执行 shell 命令。