云信AI部署

Python + LangChain：开发智能内容审核系统

云信安装大师
90
AI 质量分
2 5 月, 2025
3 分钟阅读
0 阅读

Python + LangChain：开发智能内容审核系统

引言

在当今互联网时代，内容审核变得越来越重要。无论是社交媒体、论坛还是电商平台，都需要对用户生成的内容进行审核，以确保符合社区准则和法律法规。传统的关键词过滤方法已经不能满足需求，而基于人工智能的内容审核系统能够更智能地识别违规内容。

本文将教你如何使用Python和LangChain框架开发一个简单的智能内容审核系统。

准备工作

环境要求

  • Python 3.8+
  • pip包管理工具
  • OpenAI API密钥（或其他LLM提供商的API密钥）

安装必要的库

代码片段 
pip install langchain openai python-dotenv

获取API密钥

  1. 访问OpenAI官网(https://platform.openai.com/)注册账号
  2. 在API Keys页面创建新的密钥
  3. 将密钥保存在项目根目录的.env文件中：
代码片段 
OPENAI_API_KEY=你的API密钥

开发智能内容审核系统

1. 基础设置

首先，我们创建一个Python文件（如content_moderator.py）并设置环境：

代码片段 
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 初始化大语言模型(这里使用gpt-3.5-turbo)
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,  # 设置为0以获得更确定性的输出
    openai_api_key=os.getenv("OPENAI_API_KEY")
)

代码解释：
– load_dotenv(): 从.env文件加载环境变量
– ChatOpenAI: LangChain提供的OpenAI聊天模型封装
– temperature=0: 使模型输出更加确定，减少随机性

2. 创建审核提示模板

我们需要设计一个清晰的提示模板来指导模型进行内容审核：

代码片段 
# 定义内容审核提示模板
moderation_prompt = ChatPromptTemplate.from_template("""
你是一个专业的内容审核系统。请根据以下规则评估用户提交的内容:

[审核规则]
1. 禁止仇恨言论和歧视性内容
2. 禁止暴力威胁或煽动暴力
3. 禁止色情或露骨性内容
4. 禁止垃圾广告和诈骗信息

请对以下内容进行审核:
{content}

请按照以下格式回复:
[判断]: [通过/拒绝]
[原因]: [简要说明原因]
""")

设计原理：
– 明确告知模型它的角色是”专业的内容审核系统”
– 列出具体的审核规则作为判断依据
– {content}是占位符，将被实际内容替换

3. 构建处理链

LangChain的链式结构可以让我们轻松组合多个组件：

代码片段 
# 创建处理链: prompt -> llm -> output_parser
moderation_chain = moderation_prompt | llm | StrOutputParser()

链式结构解释：
1. moderation_prompt: 提供格式化后的提示词给LLM
2. llm: OpenAI的语言模型处理请求并生成响应
3. StrOutputParser: 将模型的输出解析为字符串格式

4. 测试审核系统

让我们测试几个例子看看效果：

代码片段 
# 测试案例1 - "正常评论"
normal_comment = "我很喜欢这个产品的设计理念，它解决了我的实际问题。"
result = moderation_chain.invoke({"content": normal_comment})
print(f"测试1 - {normal_comment}\n{result}\n")

# 测试案例2 - "仇恨言论"
hate_speech = "某些种族的人天生就比其他人低劣。"
result = moderation_chain.invoke({"content": hate_speech})
print(f"测试2 - {hate_speech}\n{result}\n")

# #测试案例3 - "暴力威胁"
violent_content = "如果有人不同意我的观点，我会用武力让他闭嘴。"
result = moderation_chain.invoke({"content": violent_content})
print(f"测试3 - {violent_content}\n{result}\n")

预期输出示例：

代码片段 
测试1 - "我很喜欢这个产品的设计理念，它解决了我的实际问题。"
[判断]: [通过]
[原因]: [内容积极正面，不违反任何规则]

测试2 - "某些种族的人天生就比其他人低劣。"
[判断]: [拒绝]
[原因]: [包含歧视性言论，违反规则1]

测试3 - "如果有人不同意我的观点，我会用武力让他闭嘴。"
[判断]: [拒绝]
[原因]: [包含暴力威胁，违反规则2]

进阶功能：批量处理和结果记录

在实际应用中，我们通常需要处理大量内容和记录结果：

代码片段 
import pandas as pd

def batch_moderate(contents):
    """批量处理多个内容"""
    results = []
    for content in contents:
        try:
            result = moderation_chain.invoke({"content": content})
            results.append({
                "content": content,
                "result": result,
                "status": "[通过]" if "[通过]" in result else "[拒绝]"
            })
        except Exception as e:
            results.append({
                "content": content,
                "result": f"处理出错: {str(e)}",
                "status": "[错误]"
            })

    # 转换为DataFrame方便分析
    return pd.DataFrame(results)

# #示例批量处理    
contents_to_check = [
    "这款手机拍照效果真棒！",
    "我要杀了那个背叛我的人！",
    "点击此链接领取免费奖品www.scam.com",
    "这家餐厅的服务态度很差劲。",
]

results_df = batch_moderate(contents_to_check)
print(results_df)

实践经验：
1. 错误处理很重要，因为API调用可能会失败
2. 结构化存储结果便于后续分析和报告生成
3. 批量处理可以提高效率（实际应用中可以考虑异步处理）

API调用优化与注意事项

API调用限制与优化策略

代码片段 
from langchain.callbacks import get_openai_callback

def moderate_with_monitoring(content):
    """带监控的单个内容审核"""
    with get_openai_callback() as cb:
        result = moderation_chain.invoke({"content": content})
        print(f"消耗token数: {cb.total_tokens}")
        print(f"总成本: ${cb.total_cost:.6f}")
        return result

# #示例调用        
print(moderate_with_monitoring("这是一个需要监控API调用的例子"))

注意事项：
1. Token消耗监控: OpenAI API按token计费
2. 速率限制: OpenAI API有每分钟请求限制(具体取决于账户类型)
3. 成本控制: GPT-4比GPT-3.5更精确但更昂贵

LangChain缓存机制（减少重复计算）

代码片段 
from langchain.cache import InMemoryCache  
from langchain.globals import set_llm_cache  

# #设置内存缓存  
set_llm_cache(InMemoryCache())  

# #第一次调用会实际请求API  
print(moderation_chain.invoke({"content": "重复的内容"}))  

# #第二次相同内容的调用会直接从缓存获取  
print(moderation_chain.invoke({"content": "重复的内容"}))

Web服务集成（Flask示例）

让我们把这个功能包装成一个简单的Web服务：

代码片段 
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/moderate', methods=['POST'])
def moderate_content():
    data = request.json

    if not data or 'content' not in data:
        return jsonify({"error": "'content'字段缺失"}), 400

    try:
        result = moderation_chain.invoke({"content": data['content']})
        return jsonify({
            "status": "[通过]" if "[通过]" in result else "[拒绝]",
            "reason": result.split("[原因]:")[1].strip() if "[原因]:" in result else ""
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(port=5000)

使用方法：
1. POST请求发送到/moderate
2. JSON格式请求体: {"content": "要审核的内容"}
3. JSON格式响应:

代码片段 
{
  "status": "[通过]/[拒绝]", 
  "reason": "..."
}<br>

总结与扩展建议

我们已经完成了一个基本的智能内容审核系统开发。关键点回顾：

1️⃣ 核心技术栈: Python + LangChain + OpenAI API
2️⃣ 核心组件: LLM + Prompt工程 + Chain结构
3️⃣ 扩展方向:
– 多语言支持: prompt中加入多语言识别指令
– 敏感度分级: [“通过”, “警告”, “拒绝”]三级分类
– 自定义规则库: RAG架构接入企业特定规则文档

完整代码已放在GitHub仓库中(假设的链接)，欢迎Star和贡献！

最终建议:
对于生产环境应用，建议考虑：
✅ Azure Content Moderator等专业服务作为补充
✅ Redis缓存提高性能
✅ Celery异步任务队列处理大批量内容

原创 高质量