包括带护栏的 Converse API

你可以使用护栏来保护你使用匡威创建的对话应用程序。API例如，如果您使用创建聊天应用程序 Converse API，您可以使用护栏屏蔽用户输入的不当内容和模特生成的不当内容。有关信息 Converse API，请参阅与... 进行对话 Converse API 操作。

打电话给 Converse API带护栏

要使用护栏，请在调用 C onverse 或 ConverseStream（用于流式响应）操作中包含护栏的配置信息。或者，您可以选择您希望防护机制评估的消息中的特定内容。有关可与护栏配合使用的型号的信息，以及 Converse API，请参阅支持的模型和模型功能。

主题

• 配置要使用的护栏 Converse API

• 使用保护邮件以评估有害内容 APIs

• 保护发送给的系统提示符 Converse API

• 防护机制在保护消息和系统提示时的行为

配置要使用的护栏 Converse API

您需要在 guardrailConfig 输入参数中指定防护机制的配置信息。配置包括您要使用的防护机制的 ID 和版本。您还可以为防护机制启用跟踪功能，该功能可以提供关于被防护机制屏蔽的内容的信息。

在Converse操作中，guardrailConfig是一个GuardrailConfiguration对象，如以下示例所示。

{
        "guardrailIdentifier": "Guardrail ID",
        "guardrailVersion": "Guardrail version",
        "trace": "enabled"
}

如果使用ConverseStream，则传递一个GuardrailStreamConfiguration对象。或者，您可以使用 streamProcessingMode 字段，指定您希望模型在返回流式响应块之前完成防护机制评估。或者，您可以让模型在做出异步响应的同时，防护机制继续在后台进行评估。有关更多信息，请参阅 配置流式响应行为以筛选内容。

使用保护邮件以评估有害内容 APIs

当您将消息（Message）传递给模型时，防护机制会评估消息中的内容。或者，您可以通过指定 guardContent (GuardrailConverseContentBlock) 字段来保护消息中的选定内容。防护机制仅评估 guardContent 字段中的内容，而不评估消息的其余部分。这有助于让防护机制仅评估对话中最重要的部分（如以下示例所示）。

[
    {
        "role": "user",
        "content": [
            {
                "text": "Create a playlist of 2 pop songs."
            }
        ]
    },
    {
        "role": "assistant",
        "content": [
            {
                "text": " Sure! Here are two pop songs:\n1. \"Bad Habits\" by Ed Sheeran\n2. \"All Of The Lights\" by Kanye West\n\nWould you like to add any more songs to this playlist? "
            }
        ]
    },
    {
        "role": "user",
        "content": [
            {
                "guardContent": {
                    "text": {
                        "text": "Create a playlist of 2 heavy metal songs."
                    }
                }
            }
        ]
    }
]

另一种用途是为消息提供额外的上下文，而不让防护机制评估这些额外的上下文。

[
    {
        "role": "user",
        "content": [
            {
                "text": "Only answer with a list of songs."
            },
            {
                "guardContent": {
                    "text": {
                        "text": "Create a playlist of heavy metal songs."
                    }
                }
            }
        ]
    }
]

注意

使用该guardContent字段类似于使用带有InvokeModel和的输入标签。InvokeModelWithResponseStream有关更多信息，请参阅 用标签标记用户输入以筛选内容。

保护发送给的系统提示符 Converse API

您可以将护栏与发送到的系统提示一起使用 Converse API。 要保护系统提示符，请在传递给的系统提示符中指定 guardContent (SystemContentBlock) 字段API，如以下示例所示。

[
    {
        "guardContent": {
            "text": {
                "text": "Only respond with Welsh heavy metal songs."
            }
        }
    }
]
            

如果您没有提供 guardContent 字段，防护机制不会评估系统提示消息。

防护机制在保护消息和系统提示时的行为

防护机制在处理系统提示和您传递的消息时，评估 guardContent 字段的方式有会有所不同。

	系统提示包含防护机制块	系统提示不包含防护机制块
消息包含防护机制块	系统：防护机制评估防护机制块中的内容 消息：防护机制评估防护机制块中的内容	系统：防护机制不评估任何内容 消息：防护机制评估防护机制块中的内容
消息不包含防护机制块	系统：防护机制评估防护机制块中的内容 消息：防护机制评估所有内容	系统：防护机制不评估任何内容 消息：防护机制评估所有内容

使用时处理响应 Converse API

当您调用 Converse 操作时，防护机制会评估您发送的消息。如果防护机制检测到被屏蔽的内容，会发生以下情况。

• 响应中的 stopReason 字段会被设置为 guardrail_intervened。

• 如果您启用了跟踪，则跟踪在 trace (ConverseTrace) 字段中可用。使用ConverseStream，跟踪位于操作返回的元数据 (ConverseStreamMetadataEvent) 中。

• 在 output (ConverseOutput) 字段中返回您在护栏中配置的屏蔽内容文本。如果您使用 ConverseStream，被屏蔽的内容文本会显示在流式传输消息中。

以下提供了部分响应，展示了被屏蔽的内容文本和来自防护机制评估的跟踪信息。防护机制屏蔽了消息中的 Heavy metal 这一词汇。

{
    "output": {
        "message": {
            "role": "assistant",
            "content": [
                {
                    "text": "Sorry, I can't answer questions about heavy metal music."
                }
            ]
        }
    },
    "stopReason": "guardrail_intervened",
    "usage": {
        "inputTokens": 0,
        "outputTokens": 0,
        "totalTokens": 0
    },
    "metrics": {
        "latencyMs": 721
    },
    "trace": {
        "guardrail": {
            "inputAssessment": {
                "3o06191495ze": {
                    "topicPolicy": {
                        "topics": [
                            {
                                "name": "Heavy metal",
                                "type": "DENY",
                                "action": "BLOCKED"
                            }
                        ]
                    },
                    "invocationMetrics": {
                        "guardrailProcessingLatency": 240,
                        "usage": {
                            "topicPolicyUnits": 1,
                            "contentPolicyUnits": 0,
                            "wordPolicyUnits": 0,
                            "sensitiveInformationPolicyUnits": 0,
                            "sensitiveInformationPolicyFreeUnits": 0,
                            "contextualGroundingPolicyUnits": 0
                        },
                        "guardrailCoverage": {
                            "textCharacters": {
                                "guarded": 39,
                                "total": 72
                            }
                        }
                    }
                }
            }
        }
    }
}

使用的示例代码 Converse API带护栏

以下示例展示了如何使用 Converse 和 ConverseStream 操作来保护对话，还展示了如何防止模型创建包含重金属歌曲的播放列表。

保护对话

1. 按照创建防护机制中的说明创建防护机制。在步骤 6a 中，输入以下信息以创建拒绝的主题：

   • 名称 – 输入 Heavy metal。

   • 主题定义 – 输入 Avoid mentioning songs that are from the heavy metal genre of music。

   • 添加示例短语 – 输入 Create a playlist of heavy metal songs。

   在步骤 9 中，输入以下内容：

   • 对于被屏蔽的提示，系统显示的消息 – 输入 Sorry, I can't answer questions about heavy metal music。

   • 对于被屏蔽的响应，系统显示的消息 – 输入 Sorry, the model generated an answer that mentioned heavy metal music。

   您可以配置其他防护机制选项，但在本示例中不是必填项。

2. 按照创建防护机制的版本中的说明创建防护机制的版本。

3. 在以下代码示例（Converse 和 ConverseStream）中，设置以下变量：

   • guardrail_id – 您在第 1 步中创建的防护机制的 ID。

   • guardrail_version – 您在第 2 步中创建的防护机制的版本。

   • text – 使用 Create a playlist of heavy metal songs.。

4. 运行示例代码。输出应该显示防护机制评估和输出消息 Text: Sorry, I can't answer questions about heavy metal music.。防护机制输入评估表明，模型在输入消息中检测到了 heavy metal 这一词汇。

5. 通过将 text 的值改为 List all genres of rock music.，测试防护机制能否屏蔽模型生成的不当文本内容。再次运行示例代码。您应该会在响应中看到输出评估。

Converse

以下代码将您的防护机制与 Converse 结合使用。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Shows how to use a guardrail with the <noloc>Converse</noloc> API.
"""

import logging
import json
import boto3


from botocore.exceptions import ClientError


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


def generate_conversation(bedrock_client,
                          model_id,
                          messages,
                          guardrail_config):
    """
    Sends a message to a model.
    Args:
        bedrock_client: The Boto3 Bedrock runtime client.
        model_id (str): The model ID to use.
        messages JSON): The message to send to the model.
        guardrail_config : Configuration for the guardrail.

    Returns:
        response (JSON): The conversation that the model generated.

    """

    logger.info("Generating message with model %s", model_id)

    # Send the message.
    response = bedrock_client.converse(
        modelId=model_id,
        messages=messages,
        guardrailConfig=guardrail_config
    )

    return response


def main():
    """
    Entrypoint for example.
    """

    logging.basicConfig(level=logging.INFO,
                        format="%(levelname)s: %(message)s")

    # The model to use.
    model_id="meta.llama3-8b-instruct-v1:0"

    # The ID and version of the guardrail.
    guardrail_id = "Your guardrail ID"
    guardrail_version = "DRAFT"

    # Configuration for the guardrail.
    guardrail_config = {
        "guardrailIdentifier": guardrail_id,
        "guardrailVersion": guardrail_version,
        "trace": "enabled"
    }

    text = "Create a playlist of 2 heavy metal songs."
    context_text = "Only answer with a list of songs."

    # The message for the model and the content that you want the guardrail to assess.
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "text": context_text,
                },
                {
                    "guardContent": {
                        "text": {
                            "text": text
                        }
                    }
                }
            ]
        }
    ]

    try:

        print(json.dumps(messages, indent=4))

        bedrock_client = boto3.client(service_name='bedrock-runtime')

        response = generate_conversation(
            bedrock_client, model_id, messages, guardrail_config)

        output_message = response['output']['message']

        if response['stopReason'] == "guardrail_intervened":
            trace = response['trace']
            print("Guardrail trace:")
            print(json.dumps(trace['guardrail'], indent=4))

        for content in output_message['content']:
            print(f"Text: {content['text']}")

    except ClientError as err:
        message = err.response['Error']['Message']
        logger.error("A client error occurred: %s", message)
        print(f"A client error occured: {message}")

    else:
        print(
            f"Finished generating text with model {model_id}.")


if __name__ == "__main__":
    main()

ConverseStream

以下代码将您的防护机制与 ConverseStream 结合使用。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Shows how to use a guardrail with the ConverseStream operation.
"""

import logging
import json
import boto3


from botocore.exceptions import ClientError


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


def stream_conversation(bedrock_client,
                    model_id,
                    messages,
                    guardrail_config):
    """
    Sends messages to a model and streams the response.
    Args:
        bedrock_client: The Boto3 Bedrock runtime client.
        model_id (str): The model ID to use.
        messages (JSON) : The messages to send.
        guardrail_config : Configuration for the guardrail.


    Returns:
        Nothing.

    """

    logger.info("Streaming messages with model %s", model_id)

    response = bedrock_client.converse_stream(
        modelId=model_id,
        messages=messages,
        guardrailConfig=guardrail_config
    )

    stream = response.get('stream')
    if stream:
        for event in stream:

            if 'messageStart' in event:
                print(f"\nRole: {event['messageStart']['role']}")

            if 'contentBlockDelta' in event:
                print(event['contentBlockDelta']['delta']['text'], end="")

            if 'messageStop' in event:
                print(f"\nStop reason: {event['messageStop']['stopReason']}")

            if 'metadata' in event:
                metadata = event['metadata']
                if 'trace' in metadata:
                    print("\nAssessment")
                    print(json.dumps(metadata['trace'], indent=4))


def main():
    """
    Entrypoint for streaming message API response example.
    """

    logging.basicConfig(level=logging.INFO,
                        format="%(levelname)s: %(message)s")

    # The model to use.
    model_id = "amazon.titan-text-express-v1"

    # The ID and version of the guardrail.
    guardrail_id = "Change to your guardrail ID"
    guardrail_version = "DRAFT"

    # Configuration for the guardrail.
    guardrail_config = {
        "guardrailIdentifier": guardrail_id,
        "guardrailVersion": guardrail_version,
        "trace": "enabled",
        "streamProcessingMode" : "sync"
    }

    text = "Create a playlist of heavy metal songs."
  
    # The message for the model and the content that you want the guardrail to assess.
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "text": text,
                },
                {
                    "guardContent": {
                        "text": {
                            "text": text
                        }
                    }
                }
            ]
        }
    ]

    try:
        bedrock_client = boto3.client(service_name='bedrock-runtime')

        stream_conversation(bedrock_client, model_id, messages,
                        guardrail_config)

    except ClientError as err:
        message = err.response['Error']['Message']
        logger.error("A client error occurred: %s", message)
        print("A client error occured: " +
              format(message))

    else:
        print(
            f"Finished streaming messages with model {model_id}.")


if __name__ == "__main__":
    main()