本文为您提供大模型应用防火墙接口调用的示例代码。实际使用时，您可以替换成 SDK 中已经支持的任意接口。

前提条件 #

调用前，请确保已经完成Python 安装和初始化。

调用接口 #

CheckLLMPrompt #

调用CheckLLMPrompt接口可以判别输入的提示词是否合规。大模型应用防火墙检测并分析提示词内容，并返回对应内容的判别结果、处置动作和命中的内容。
CheckLLMPrompt相关参数的详细说明，可参考CheckLLMPrompt - 检查用户提示词是否合规。

from __future__ import print_function
import volcenginesdkcore
import volcenginesdkwaf
from volcenginesdkcore.rest import ApiException

if __name__ == '__main__':
    configuration = volcenginesdkcore.Configuration()
    configuration.ak = "YOUR AK"  # Access Key，用于身份验证
    configuration.sk = "YOUR SK"  # Secret Key，用于签名
    configuration.region = "cn-beijing"  # 大模型应用防火墙所属地域对应的 ID
    
    # 设置默认配置
    volcenginesdkcore.Configuration.set_default(configuration)

    # 创建大模型应用防火墙 API 实例
    api_instance = volcenginesdkwaf.WAFApi()
    
    # 创建内容检测请求对象
    check_llm_prompt_request = volcenginesdkwaf.CheckLLMPromptRequest(
        content="请输入您想要校验的内容，文本内容或资源URL",  # 需要检测的内容
        content_type=1,  # 检测的内容类型，当前支持 1-文本
        host="fd****.access.omni-shield.volces.com",  # 大模型应用防火墙生成的接入点域名
        msg_class=0,  # 场景类型，当前默认为标准类型，填入 0 即可
        region="cn-beijing",  # 大模型应用防火墙所属地域对应的 ID
    )
    
    try:
        # 复制代码运行示例，请自行打印API返回值。
        data = api_instance.check_llm_prompt(check_llm_prompt_request)
        print("response:" , data )

    except ApiException as e:
        # 复制代码运行示例，请自行打印API错误信息。
        print("Exception when calling api: %s\n" % e)
        pass

CheckLLMResponseStream #

调用CheckLLMResponseStream接口可判别大模型响应的内容是否合规。大模型应用防火墙会检测并分析您的大模型响应内容，并返回对应内容的判别结果、处置动作和命中的内容。
CheckLLMResponseStream相关参数的详细说明，可参考CheckLLMResponseStream - 检查大模型响应内容是否合规。以下分别介绍非流式响应检测和流式响应检测的示例代码。

非流式响应检测

from __future__ import print_function
import volcenginesdkcore
import volcenginesdkwaf
from volcenginesdkcore.rest import ApiException

if __name__ == '__main__':
    
    configuration = volcenginesdkcore.Configuration()
    configuration.ak = "YOUR AK"  # 替换为实际 AK
    configuration.sk = "YOUR SK"  # 替换为实际 SK
    configuration.region = "cn-beijing" # 大模型应用防火墙所属地域对应的 ID

    volcenginesdkcore.Configuration.set_default(configuration)

    api_instance = volcenginesdkwaf.WAFApi()
    check_llm_response_stream_request = volcenginesdkwaf.CheckLLMResponseStreamRequest(
        content="此处输入需要检测的文本内容",  # 待检测的文本内容
        content_type=1,  # 检测的内容类型，当前支持 1-文本
        host="0d****.access.omni-shield.volces.com", # 大模型应用防火墙生成的接入点域名
        msg_class=0,  # 场景类型，当前默认为标准类型，填入 0 即可
        region="cn-beijing",  # 大模型应用防火墙所属地域对应的 ID
     )
    
    try:
        # 复制代码运行示例，请自行打印API返回值。
        api_instance.check_llm_response_stream(check_llm_response_stream_request)
    except ApiException as e:
        # 复制代码运行示例，请自行打印API错误信息。
        # print("Exception when calling api: %s\n" % e)
        pass

流式响应检测

from __future__ import print_function
import pprint

import volcenginesdkcore
import volcenginesdkwafruntime
from volcenginesdkcore.rest import ApiException
from volcenginesdkwafruntime import WAFRuntimeApi
from volcenginesdkwafruntime import LLMStreamSession


def split_text_into_chunks(text, chunk_size=9):
    """将文本按指定长度拆分为子串列表"""
    return [text[i:i + chunk_size] for i in range(0, len(text), chunk_size)]


if __name__ == '__main__':
    # 注意：请替换为实际的AK/SK和域名
    configuration = volcenginesdkcore.Configuration()
    configuration.ak = "YOUR AK"  # 替换为实际 AK
    configuration.sk = "YOUR SK"  # 替换为实际 SK
    configuration.region = "cn-beijing"  # 大模型应用防火墙所属地域对应的 ID
    volcenginesdkcore.Configuration.set_default(configuration)

    long_text = "这是一个特别特别长的字符串，用来模拟一个输入流，模拟模拟这是一个特别特别长的字符串，用来模拟一个输入流，模拟模拟这是一个特别特别长的字符串，用来模拟一个输入流"
    chunked_text = split_text_into_chunks(long_text)

    # 初始化会话对象
    llm_session = LLMStreamSession()
    api_instance = WAFRuntimeApi()
    # 配置请求基础参数
    base_request = volcenginesdkwafruntime.CheckLLMResponseStreamRequest(
        content="", # 检测内容分片
        content_type=1, # 检测的内容类型，当前支持 1-文本
        host="0d****.access.omni-shield.volces.com",   # 大模型应用防火墙生成的接入点域名
        msg_class=0, # 场景类型，当前默认为标准类型，填入 0 即可
        region="cn-beijing", # 大模型应用防火墙所属地域对应的 ID
        use_stream=1,  # 是否开启流式计算：0-非流式检查，1-流式检查，2-流式检查结束
    )

    # 获取最后一块的索引
    last_index = len(chunked_text) - 1

    # 循环处理每个文本块
    for index, chunk in enumerate(chunked_text):
        #print(f"处理第 {index + 1} 块（{len(chunk)}字符）: {chunk}")

        # 设置当前块内容
        base_request.content = chunk

        # 当处理到最后一块时，将use_stream设置为2
        if index == last_index:
            base_request.use_stream = 2
            #print(f"注意：最后一块，use_stream已设置为 {base_request.use_stream}")
        else:
            base_request.use_stream = 1

        try:
            # 调用API（自动传递session）
            resp = api_instance.check_llm_response_stream(
                body=base_request,
                session=llm_session
            )
            # 打印响应摘要
            print(f"第 {index + 1} 次调用响应:")
            if resp:
                pprint.pprint(resp)
            else:
                print("无有效响应")

        except ApiException as e:
            print(f"第 {index + 1} 次调用异常: {e.body}")
            break  # 遇到异常时终止循环

    print("\n全部块处理完成")

    

上述代码中流式内容检测的实现逻辑如下流程图所示：

QueryLLMGenerate #

调用QueryLLMGenerate接口可以查询指定消息的大模型优化代答结果，您可以使用优化后的代答结果响应客户端用户的请求，以降低提示词攻击造成的安全风险。
QueryLLMGenerate相关参数的详细说明，可参考QueryLLMGenerate - 查询大模型优化代答结果。

from __future__ import print_function
import volcenginesdkcore
import volcenginesdkwaf
from volcenginesdkcore.rest import ApiException

if __name__ == '__main__':
   
    configuration = volcenginesdkcore.Configuration()
     configuration.ak = "YOUR AK"  # Access Key，用于身份验证
    configuration.sk = "YOUR SK"  # Secret Key，用于签名
    configuration.region = "cn-beijing"  # 大模型应用防火墙所属地域对应的 ID
    
    # 设置默认配置
    volcenginesdkcore.Configuration.set_default(configuration)

    # 创建大模型应用防火墙 API 实例
    api_instance = volcenginesdkwaf.WAFApi()
    query_llm_generate_request = volcenginesdkwaf.QueryLLMGenerateRequest(
        msg_id="c4b08d3503e64645a88253fb8e49****",  # 需要查询的请求消息 ID
        use_stream=False  # 设置是否返回流式结果，SDK 调用仅支持 false 非流式返回
    )
    
    try:
        # 复制代码运行示例，请自行打印API返回值。
        api_instance.query_llm_generate(query_llm_generate_request)
    except ApiException as e:
        # 复制代码运行示例，请自行打印API错误信息。
        # print("Exception when calling api: %s\n" % e)
        pass