调用 AgentCore 运行时代理

该InvokeAgentRuntime操作允许您向由其亚马逊资源名称 (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参数进行会话管理。通过在多个请求中提供相同的会话标识符，您可以维护对话上下文，从而允许代理引用以前的交互。

要开始新的对话，请生成唯一的会话标识符。要继续现有对话，请使用与之前请求相同的会话标识符。这种方法使您能够构建交互式应用程序，这些应用程序可以随着时间的推移保持上下文。

错误处理

使用该InvokeAgentRuntime操作时，您可能会遇到各种错误。以下是一些常见错误及其处理方法：

在应用程序中实施适当的错误处理，以提供更好的用户体验并有效地解决问题。

最佳实践

使用InvokeAgentRuntime操作时，请遵循以下最佳实践：