从您的应用程序调用代理 - Amazon Bedrock
调用代理

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

从您的应用程序调用代理

通过使用 Amazon Bedrock 代理运行时端点发出 InvokeAgent 请求，在应用程序中使用您的代理。

默认情况下，来自 InvokeAgent 的响应包括一个分块，其中包含代理的完整响应，可能需要一段时间才能完成。或者，您可以将 InvokeAgent 配置为以多个较小的分块流式传输响应。这可以减少代理初始响应的延迟。

流式响应

您可以选择通过在流式传输配置（StreamingConfigurations）中将 streamFinalResponse 设置为 true 来启用响应流式传输。响应流包含多个事件，响应的每个部分按顺序分块。

要流式传输响应，请确保代理执行角色包含已配置代理模型的 bedrock:InvokeModelWithResponseStream 权限。有关更多信息，请参阅 代理服务角色的基于身份的权限。。

如果您的代理配置了护栏，则还可以在 StreamingConfigurations 中指定 applyGuardrailInterval，以控制每多少个传出响应字符（例如，每 50 个字符）发出一个 ApplyGuardrail 调用。

默认情况下，护栏间隔设置为 50 个字符。如果指定了较大的间隔，则响应将以较大的分块生成，并且 ApplyGuardrail 调用次数更少。下面的示例显示了为 Hello, I am an agent 输入字符串生成的响应。

分块响应示例：间隔设置为 3 个字符

'Hel', 'lo, ', 'I am', ' an', ' Age', 'nt'

除了最后一个分块外，每个分块至少包含 3 个字符。

分块响应示例：间隔设置为 20 个或更多字符

'Hello, I am an Agent'

调用代理

下面的 Python 示例说明如何调用 InvokeAgent 操作并显示代理的输出，以及该调用生成的任何跟踪信息。

调用代理

  1. 按照 手动创建和配置代理 中的说明创建代理。记下代理的 ID。如有必要，您可以在以后获取 ID。有关更多信息，请参阅 查看关于代理的信息。

  2. 按照 为代理创建别名 中的说明为代理创建别名。记下别名的 ID。如有必要，您可以在以后获取 ID。有关更多信息，请参阅 在 Amazon Bedrock 中查看有关代理别名的信息。

  3. 运行以下代码。将 agent_id 的值更新为代理的 ID，将 alias_id 的值更新为代理的别名 ID。要流式传输来自代理的响应，请将 streamFinalResponse 的值更改为 True。您还可以通过更改 applyGuardrailInterval 的值来更改护栏间隔。

    import boto3
import logging

from botocore.exceptions import ClientError


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def invoke_agent(client, agent_id, alias_id, prompt, session_id):
        response = client.invoke_agent(
            agentId=agent_id,
            agentAliasId=alias_id,
            enableTrace=True,
            sessionId = session_id,
            inputText=prompt,
            streamingConfigurations = { 
    "applyGuardrailInterval" : 20,
      "streamFinalResponse" : False
            }
        )
        completion = ""
        for event in response.get("completion"):
            #Collect agent output.
            if 'chunk' in event:
                chunk = event["chunk"]
                completion += chunk["bytes"].decode()
            
            # Log trace output.
            if 'trace' in event:
                trace_event = event.get("trace")
                trace = trace_event['trace']
                for key, value in trace.items():
                    logging.info("%s: %s",key,value)

        print(f"Agent response: {completion}")


if __name__ == "__main__":

    client=boto3.client(
            service_name="bedrock-agent-runtime"
        )
    
    agent_id = "AGENT_ID"
    alias_id = "ALIAS_ID"
    session_id = "MY_SESSION_ID"
    prompt = "Prompt to send to agent"

    try:

        invoke_agent(client, agent_id, alias_id, prompt, session_id)

    except ClientError as e:
        print(f"Client error: {str(e)}")
        logger.error("Client error: %s", {str(e)})