用护栏呼叫匡威 API 使用匡威时处理响应 API 使用带护栏的 Converse API 的示例代码

包括匡威的护栏 API

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

主题

用护栏呼叫匡威 API
使用匡威时处理响应 API
使用带护栏的 Converse API 的示例代码

用护栏呼叫匡威 API

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

将护栏配置为与匡威配合使用 API

您可以在guardrailConfig输入参数中指定护栏的配置信息。配置包括您要使用的护栏的 ID 和版本。您还可以启用对护栏的跟踪，它会提供有关护栏屏蔽的内容的信息。

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


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

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

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

当您将消息（消息）传递给模型时，护栏会评估消息中的内容。或者，您可以通过指定 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字段的行为方式有所不同。

	系统提示有护栏屏蔽	系统提示符没有护栏屏蔽
消息有护栏屏蔽	系统：Guardrail 正在调查护栏块中的内容消息：Guardrail 正在调查 Guardrail block 中的内容	系统：Guardrail 什么也没调查消息：Guardrail 正在调查 Guardrail block 中的内容
消息没有护栏屏蔽	系统：Guardrail 正在调查护栏块中的内容消息：Guardrail 会调查所有内容	系统：Guardrail 什么也没调查消息：Guardrail 会调查所有内容

系统提示有护栏屏蔽

系统提示符没有护栏屏蔽

消息有护栏屏蔽

系统：Guardrail 正在调查护栏块中的内容

消息：Guardrail 正在调查 Guardrail block 中的内容

系统：Guardrail 什么也没调查

消息：Guardrail 正在调查 Guardrail block 中的内容

消息没有护栏屏蔽

系统：Guardrail 正在调查护栏块中的内容

消息：Guardrail 会调查所有内容

系统：Guardrail 什么也没调查

消息：Guardrail 会调查所有内容

使用匡威时处理响应 API

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

响应中的stopReason字段设置为guardrail_intervened。
如果您启用了跟踪，则跟踪在 trace (ConverseTrace) 字段中可用。使用ConverseStream，跟踪位于操作返回的元数据 (ConverseStreamMetadataEvent) 中。
在 output (ConverseOutput) 字段中返回您在护栏中配置的屏蔽内容文本。内容被屏蔽ConverseStream后，直播消息中会显示文本。

以下部分响应显示了屏蔽的内容文本和护栏评估中的痕迹。护栏在消息中屏蔽了 “重金属” 一词。


{
    "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操作的对话。该示例说明如何防止模特创建包含重金属流派歌曲的播放列表。

为了保护对话

按照中的说明创建护栏。创建护栏在步骤 6a 中，输入以下信息以创建被拒绝的主题：
- 名称 — 输入 He avy metal。
- 主题的定义 —输入避免提及来自重金属音乐流派的歌曲.
- 添加示例短语 —输入创建重金属歌曲的播放列表.
在步骤 9 中，输入以下内容：
- 显示屏蔽提示的消息 — 输入对不起，我无法回答有关重金属音乐的问题。
- 回复被屏蔽的消息 — 输入对不起，模型生成了一个提及重金属音乐的答案。
您可以配置其他护栏选项，但在本示例中不是必需的。
按照中的说明创建护栏版本。创建护栏的版本
在以下代码示例（Converse 和 ConverseStream）中，设置以下变量：
- guardrail_id— 您在步骤 1 中创建的护栏的 ID。
- guardrail_version— 您在步骤 2 中创建的护栏版本。
- text— 使用 Create a playlist of heavy metal songs.
运行代码示例。输出应显示护栏评估和输出消息。Text: Sorry, I can't answer questions about heavy metal music.护栏输入评估表明，该模型在输入消息中检测到了 “重金属” 一词。
（可选）通过将的值更改为 “列出所有类型的摇滚音乐”，测试护栏是否屏蔽了模型生成的不恰当文本。text 。再次运行示例。您应该会在响应中看到产出评估。

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 Converse 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()

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

文档惯例

直播回复

ApplyGuardrail API在您的应用程序中使用