LongArticlesJob/applications/llm_sensitivity.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

import json
from openai import OpenAI


def request_llm_api(prompt, text):
    client = OpenAI(
        api_key='sk-c1b18099dadc4dd1b48239bdde184f6c',
        base_url="https://api.deepseek.com"
    )
    chat_completion = client.chat.completions.create(
        messages=[
            {
                "role": "user",
                "content": prompt + text,
            }
        ],
        model="deepseek-chat",
        temperature=0.2,
        response_format={"type": "json_object"}
    )
    response = chat_completion.choices[0].message.content
    return response


def do_check_titles(text):
    """
    :param text:
    :return:
    """
    text_prompt = """
你是一位专业的微信公众号内容编辑，公众号的主要读者是中老年群体，主要内容来源为其它公众号发布文章。你需要对公众号待发布的文章标题进行可用性审核，判断其是否适合发布。不可用的情形包括以下7大类，每个大类情形中包括多个小类，以多级序号表示：
1.违规诱导、对抗平台审查规则
1.1 使用不完全或擦边的标题诱导用户点击
1.2 使用了生僻字/绕过字体等形式规避微信审核
2.违反国家法律法规
2.1 涉及国内政治敏感人物、敏感政治事件
3.有煽动、夸大、误导嫌疑
3.1 涉及国内外时局政治内容，尤其是与中国关系密切的国家的
3.2 在标题中使用内部、内参、解封等词语，将内容伪装成官方内部流出的
3.3 非官方通知或者公告，但标题假借官方名义或者暗喻为官方发布的
3.4 标题滥用“领导”、“主席”等头衔，容易造成曲解的
3.5 在标题中使用重大消息、通知、紧急提醒等，或以信息来源机密、看完即删来诱导用户的
3.6 在标题中使用过度夸张的词语，比如含有危害人身安全、恐吓侮辱、惊悚、极端内容，或者以命令式语气强迫用户阅读，煽动诱导的意味较强的
4.色情低俗内容
4.1 文字描述男女性隐私部位，或误导用户误以为是隐私部位
4.2 文字明示或暗示，容易让人联想到性、性行为、不正当性关系的
4.3 文字出现不正当婚恋观，如出轨、找小三、约炮等
5.宣扬封建迷信
5.1 标题捏造神迹或捏造所谓的特殊日子，用于接福、求福的
5.2 标题涉及民间迷信谚语、民俗，容易宣扬迷信思想的
6.不适合在非特定时间、向不特定群体发布的
6.1 具有较强新闻时效性的事件，尤其是包含“突发”、“刚刚”或特定日期的
6.2 面向特定范围群体（某个省市的居民、企事业单位员工等）的通知、宣传稿、新闻稿等
7.与中老年人阅读偏好不相关的
7.1 学习、影视等资源下载
7.2 其它与一般中老年人偏好不相关的情形
审核时，请特别注意以下事项：
 - 标题是否涉及政治敏感、色情低俗、封建迷信等内容。
 - 标题是否适合中老年群体的阅读偏好。
 - 标题是否具有时效性或面向特定群体，不适合广泛发布。例如：
  - 标题中包含“突破”、“刚刚”等词语时，需注意是否具有时效性（如年初、特定节日等）。
  - 标题中包含特定地名、群体名称（如省市区县旗盟乡镇）时，需注意是否面向特定群体，如果包含特定地域或人群但具有传播性和普适性则也可以发布。
审核结果分为2种：
 - 通过：标题未违反规则，适合发布。
 - 不通过：标题明确违反规则，不适合发布。
对于“不通过”的情形，请同时标注命中的规则大类序号（如 1、2、3 等）。
输出格式必须为 JSON 格式，示例如下：
[
{
"title": "标题",
"status": "不通过",
"hit_rule": 5
}
]
现在，请对以下文章标题的可用性进行判断：
"""
    return request_llm_api(text_prompt, text)

def check_titles(titles, return_map=False):
    n_titles = len(titles)

    batch_size = 20
    n_batches = (n_titles + batch_size - 1) // batch_size
    json_data = []
    for i in range(0, n_batches):
        offset_begin = i * batch_size
        offset_end = min(offset_begin + batch_size, n_titles)
        current_batch = titles[offset_begin:offset_end]
        try:
            res = do_check_titles('\n'.join(current_batch))
            json_data = json.loads(res)
        except Exception as e:
            print(e)
        if isinstance(json_data, dict):
            json_data = [json_data]
        if not json_data:
            for title in current_batch:
                try:
                    res = do_check_titles(title)
                    json_res = json.loads(res)
                    if isinstance(json_res, list):
                        json_data.append(json_res[0])
                    elif isinstance(json_res, dict):
                        json_data.append(json_res)
                    else:
                        raise Exception('Invalid Response')
                except Exception as e:
                    print(e)
                    json_data.append({'title': title, 'status': '通过'})
    for item in json_data:
        if item['status'] == '不通过':
            try:
                item['hit_rule'] = int(float(item['hit_rule']))
            except Exception as e:
                item['hit_rule'] = 99
            # 保证不为0 避免LLM返回异常
            if item['hit_rule'] == 0:
                item['hit_rule'] = 99
        else:
            item['hit_rule'] = 0
    if return_map:
        result_map = {}
        for item in json_data:
            if item['title'] not in result_map:
                result_map[item['title']] = item
            # 安全起见，保留最后一次不通过的记录
            elif item['status'] == '不通过':
                result_map[item['title']] = item
        return result_map
    else:
        return json_data