步骤 5：通过 Python 代码调用代理 - Amazon Bedrock

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

步骤 5：通过 Python 代码调用代理

在本步骤中，您将学习如何使用 AWS SDK for Python (Boto) 以编程方式与代理交互。本示例代码演示如何使用 InvokeAgent 操作，该操作需要 AGENT ID 和 ALIAS ID 作为参数来调用代理。该代码演示了如何向代理发送提示、处理响应，以及处理流式和非流式响应模式。这使您能够将 Bedrock 代理集成到自己的 Python 应用程序中。

有关更多信息，请参阅 从您的应用程序调用代理。

通过 Python 代码调用代理

  1. 获取代理的 ID。有关更多信息，请参阅 查看关于代理的信息。

  2. 获取代理别名的 ID。有关更多信息，请参阅 在 Amazon Bedrock 中查看有关代理别名的信息。

  3. 运行以下代码。更新以下值：

    • AGENT_ID – 更新为您的代理的 ID。

    • ALIAS_ID – 更新为您的代理的别名 ID。

    • REGION – 更新为您在其中创建代理的 AWS 区域，例如 us-east-1。

    要流式传输来自代理的响应，请将 streamFinalResponse 的值更改为 True。

    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",
            region_name="REGION") 
    
    agent_id = "AGENT_ID"
    alias_id = "ALIAS_ID"
    session_id = "123456"
    prompt = "What's the current time?"

    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)})