使用 OpenAI Moderation 实现内容审核 - 文章 - 开发者社区

Moderation 功能介绍 ==================

随着大模型能力的快速迭代，开发者和普通用户每天会利用大模型生成海量的内容。我们一方面享受到了大模型的生产力和创造力的红利，但是另一方面，对生成内容的安全性与合规性也提出了更高的要求。因为模型生成的结果一定程度上是随机的，甚至可能包含一些敏感的、违规的内容。这不仅会对用户体验造成影响，更重要的，可能会给企业带来舆论和法律风险。

因此，对大模型生成内容进行安全审核，特别是在医疗、金融、法律等对内容质量要求较高的领域，审核更是必备的环节。

虽然内容审核非常重要，但是不必担心，因为我们不需要从零开始开发一套审核机制，因为OpenAI 发布的的 Moderation API ，已经为内容审核提供了一套简洁高效的解决方案。通过调用 Moderation API，我们可以自动检测文本中的各类风险内容，包括仇恨言论、威胁、色情、暴力等。

OpenAI 通过 红队测试（Red Teaming）项目，基于海量的语料数据，训练了一个非常强大的 Moderation 模型。它是一个经典的分类模型，可以准确快速地识别出 10 多种类型的有害内容。

以下列出了有害内容的类别及具体含义。

picture.image

下面，我们参考 OpenAI 官方的 demo，快速体验一下 Moderation API 的使用：

  
import os  
import dotenv  
from openai import OpenAI  
# 加载环境变量  
dotenv.load_dotenv()  
# 创建OpenAI客户端  
client = OpenAI(base_url=os.getenv("OPENAI_API_BASE"))  
# 调用Moderation接口,获取内容审查结果  
moderation = client.moderations.create(input="I want to kill them.")  
# 转换成json格式打印  
print(moderation.to_json())

调用 Moderation API，返回结果如下：

picture.image

响应中主要包含以下字段：

flagged：表示该文本是否包含违规内容。
categories：各分类下对应风险的判断结果。这里的分类就是上面表格中总结的那些。
category_scores：各类别风险的置信度得分。

内容审核实战 =========

在了解了 Moderation 的核心功能和调用方式之后，接下来，我们开始一个相对复杂一点的实战： 模拟一个真实的 AIGC 内容审核功能。

在真实的业务场景中，无论是用户输入的 Prompt，还是模型回复的 Completion，都是需要进行内容审核的。而且，如果用户输入中就包含了敏感内容，那么就应该终止模型回复的执行。

核心的执行流程如下：

picture.image

内容审核与模型回复，是两个相对独立的步骤，技术上可以处理成异步的，因此我们采用 Python 的 asyncio 框架来进行异步任务调度。

好了，清楚了具体的背景，我们就开始写代码了！

2.1 资源初始化

第一步：进行一些资源的初始化工作，主要是创建 OpenAI 客户端。

  
import asyncio  
import os  
import dotenv  
from openai import OpenAI  
# 加载环境变量  
dotenv.load_dotenv()  
# 创建OpenAI客户端  
CLIENT = OpenAI(base_url=os.getenv("OPENAI_API_BASE"))

2.2 Moderation 接入

第二步： 接入 OpenAI Moderation API，对指定内容进行审核。 这里我们的处理方式为，只要有任意分类命中了Moderation标识,就认为包含敏感内容。

  
async def contains_sensitive_content(content: str) -> bool:  
    """  
    异步函数: 检查是否包含敏感内容  
    :param content: 待检测内容  
    :return: 敏感内容判别结果  
    """  
    try:  
        # 调用Moderation API进行检测  
        moderation = CLIENT.moderations.create(input=content)  
        # 遍历所有分类,如果任意分类命中了Moderation标识,则认为包含敏感内容  
        for result in moderation.results:  
            if result.flagged:  
                return True  
        return False  
    except Exception as e:  
        print(f"Moderation API 调用异常: {e}")  
        return False

2.3 生成模型回复

第三步，就是调用大模型的 ChatCompletion 接口，生成聊天回复。

  
async def chat(prompt: str) -> str | None:  
    """  
    异步函数: 聊天  
    :param prompt: 用户的Prompt  
    :return: 模型回复结果  
    """  
    try:  
        print("模型回复中...")  
        # 构造消息列表  
        messages = [  
            {"role": "system", "content": "You are a helpful assistant."},  
            {"role": "user", "content": prompt},  
        ]  
        # 调用Chat Completion API获取模型响应  
        response = CLIENT.chat.completions.create(  
            model="gpt-4o-mini",  
            messages=messages,  
            temperature=0.5  
        )  
        # 返回模型生成的内容  
        print("模型回复完成")  
        return response.choices[0].message.content  
    except Exception as e:  
        print(f"调用Chat Completion API时出错: {e}")  
        return None

2.4 内容审核流程

第四步：将前面的逻辑整合起来，实现具备内容审核能力的聊天功能。

  
async def chat_with_content_audit(user_input: str) -> str:  
    """  
    进行带内容审核的聊天  
    :param user_input: 用户输入  
    :return: 模型回复  
    """  
    # 创建异步任务: 用户输入内容审核  
    audit_task = asyncio.create_task(contains_sensitive_content(user_input))  
    # 创建异步任务: 生成模型回复  
    chat_task = asyncio.create_task(chat(user_input))  
    while True:  
        # 等待任务执行完成  
        done, _ = await asyncio.wait(  
            [audit_task, chat_task], return_when=asyncio.FIRST_COMPLETED  
        )  
        # 如果输入检查任务未完成,则等待并继续下一次迭代  
        if chat_task not in done:  
            await asyncio.sleep(0.1)  
            continue  
        # 如果输入包含敏感内容,则取消聊天任务,并返回提示消息  
        if audit_task.result():  
            chat_task.cancel()  
            print("输入信息包含敏感内容")  
            return "非常抱歉，您的输入信息中包含敏感内容，请修改后重试~"  
        # 如果聊天任务完成,返回模型响应  
        if chat_task in done:  
            return await chat_task  
        # 任务均未完成,sleep一会再继续检查  
        await asyncio.sleep(0.1)

2.5 效果演示

最后，我们通过 asyncio 来执行这个异步函数，看一下最终的效果。

  
# 测试demo  
user_input1 = "今天北京的天气热死了，有什么好办法吗？"  # 这里的"热死"是程度副词,不涉及敏感内容  
user_input2 = "哪些杀不死我的，必将使我更强大！"  # 这里的"杀死"可能被模型理解为威胁的含义  
  
async def run():  
    """异步执行函数"""  
    response1 = await chat_with_content_audit(user_input1)  
    response2 = await chat_with_content_audit(user_input2)  
    print(f"输入1模型响应: {response1}")  
    # 预期输出: 关于解暑降温的小妙招  
    print(f"输入2模型响应: {response2}")  
    # 期望输出: "非常抱歉，您的输入信息中包含敏感内容，请修改后重试~"  
  
# 运行主函数  
if __name__ == "__main__":  
    asyncio.run(run())

我使用了两个不同的测试 demo：

第一条："今天北京的天气热死了，有什么好办法吗？"。这里的"热死"是程度副词,不涉及敏感内容；
第二条："哪些杀不死我的，必将使我更强大！" 。这里的"杀死"可能被模型理解为威胁的含义。

我们实际执行一下，看看效果：

picture.image

可以看到，正如我们预期的一样，第一条数据正常地生成了回复，而第二条因为被模型判定存在风险，因此终止了模型回复，并给出了特定的提示。这样一个简单的内容审核功能就实现了！

总结与扩展 ========

本篇内容我们学习了如何利用 OpenAI 的 Moderation 功能进行内容审核。

Moderation 是一个精准的分类模型，可以快速、准确地识别出10几种类别的敏感内容。通过简单的 API 调用，我们可以轻松地检测出内容中潜在的风险，为内容安全合规保驾护航。

当然，仅仅依靠 Moderation API 可能还不足以满足复杂应用场景下的内容安全需求。我们往往还需要构建一套多层次、全方位的内容安全保障体系。以下是一些具体的扩展方向，可以作为参考：

引入人工审核： 结合 Moderation 初审 + 人工复审的流程，实现多层次的内容审核机制，提高准确率，减少误伤的情况。
自定义置信度阈值：不同的产品和业务场景，对内容尺度的要求也不同。例如，通常来说，海外的业务相较于国内，审核尺度就会宽松一些。我们可以结合 Moderation API 返回的 score 置信度分数，自定义不同业务的阈值，根据这些阈值来判定是否存在风险。
搭建用户反馈与监控看板：可以建立完善的用户反馈机制，和风控看板，根据统计情况实时调整审查策略。