Upload 3 files

pages/对话式文本检测工具.py

@@ -0,0 +1,209 @@
+import torch
+import streamlit as st
+import subprocess
+import os
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from transformers import BertTokenizer
+import appbuilder 
+from transformers import BertModel
+#加载预训练模型
+pretrained = BertModel.from_pretrained('hfl/chinese-macbert-base')
+#需要移动到cuda上
+device = 'cuda' if torch.cuda.is_available() else 'cpu'
+pretrained.to(device)
+#不训练,不需要计算梯度
+for param in pretrained.parameters():
+    param.requires_grad_(False)
+    
+#多头注意力机制
+class MultiHeadAttention(nn.Module):
+    def __init__(self, hidden_size, num_heads):
+        super(MultiHeadAttention, self).__init__()
+        # 确保隐藏层特征数能够被头数整除
+        assert hidden_size % num_heads == 0
+        self.hidden_size = hidden_size
+        self.num_heads = num_heads
+        self.head_dim = hidden_size // num_heads  # 计算每个头的维度
+        # 定义线性层，用于对查询、键、值进行线性变换
+        self.linear_q = nn.Linear(hidden_size, hidden_size)
+        self.linear_k = nn.Linear(hidden_size, hidden_size)
+        self.linear_v = nn.Linear(hidden_size, hidden_size)
+        self.linear_out = nn.Linear(hidden_size, hidden_size)  # 定义输出线性层，用于整合多头注意力后的输出
+
+    def forward(self, x):
+        batch_size, seq_len, _ = x.size()
+        # 对输入进行线性变换，并将其分割为多个头
+        q = self.linear_q(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
+        k = self.linear_k(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
+        v = self.linear_v(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
+        # 计算注意力分数
+        scores = torch.matmul(q, k.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.head_dim, dtype=torch.float))
+        attn_weights = F.softmax(scores, dim=-1)  # 计算注意力权重
+        # 根据注意力权重对值进行加权求和
+        context = torch.matmul(attn_weights, v).transpose(1, 2).contiguous().view(batch_size, seq_len, self.hidden_size)
+        out = self.linear_out(context)  # 整合多头注意力的输出
+        return out
+    
+class Model(nn.Module):
+    def __init__(self):
+        super(Model, self).__init__()
+        self.fc1 = nn.Linear(768, 512)  # 第一层全连接层
+        self.fc2 = nn.Linear(512, 256)  # 第二层全连接层
+        self.fc3 = nn.Linear(256, 2)    # 第三层全连接层
+        self.dropout = nn.Dropout(p=0.5)
+        self.bn1 = nn.BatchNorm1d(512)
+        self.bn2 = nn.BatchNorm1d(256)
+        self.activation = nn.ReLU()
+        self.multihead_attention = MultiHeadAttention(hidden_size=768, num_heads=8)  # 多头注意力模块
+
+    def forward(self, input_ids, attention_mask, token_type_ids):
+        out = pretrained(input_ids=input_ids,
+                         attention_mask=attention_mask,
+                         token_type_ids=token_type_ids).last_hidden_state
+
+        # 应用多头注意力机制
+        out = self.multihead_attention(out)
+        out = out[:, 0]  # 提取[CLS]标记的输出
+
+        out = self.activation(self.bn1(self.fc1(out)))
+        out = self.dropout(out)
+        out = self.activation(self.bn2(self.fc2(out)))
+        out = self.dropout(out)
+        out = self.fc3(out)
+        out = out.softmax(dim=1)
+        return out
+    
+    
+def load_models_and_predict(text, device):
+    # 加载模型
+    MacBERT_base_CDialBias = torch.load('models\MacBERT-base-CDialBias.pth')
+    MacBERT_base_CDialBias.to(device)
+    MacBERT_base_COLD = torch.load('models\MacBERT-base-CDialBias.pth')
+    MacBERT_base_COLD.to(device)
+
+    # 获取密钥和ID
+    os.environ['APPBUILDER_TOKEN'] = "bce-v3/ALTAK-n2XgeA6FS3Q5E7Jab6UwE/850b44ebec64c4cad705986ab0b5e3df4b05d407"
+    app_id = "df881861-9fa6-40b6-b3bd-26df5f5d4b9a"
+
+    # 初始化agent实例
+    your_agent = appbuilder.AppBuilderClient(app_id)
+
+    # 创建会话id
+    conversation_id = your_agent.create_conversation()
+
+    # 加载字典和分词工具
+    tokenizer = BertTokenizer.from_pretrained('hfl/chinese-macbert-base')
+
+    # 对输入文本进行编码
+    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
+
+    # 将输入数据移动到相同的设备上
+    inputs = {k: v.to(device) for k, v in inputs.items()}
+
+    # 设置模型为评估模式
+    MacBERT_base_CDialBias.eval()
+    MacBERT_base_COLD.eval()
+
+    # 调用千帆api获取标签
+    msg = your_agent.run(conversation_id, text)
+    answer = msg.content.answer
+
+    # 进行预测
+    with torch.no_grad():
+        out1 = MacBERT_base_CDialBias(**inputs)
+    with torch.no_grad():
+        out2 = MacBERT_base_COLD(**inputs)
+
+    out1 = torch.argmax(out1, dim=1).item()
+    out2 = torch.argmax(out2, dim=1).item()
+    out3 = answer[0]
+
+    # 分析结果
+    if out3 == "1":
+        if out1 == out2 == out3 == 1:
+            result = "这句话具有攻击性和社会偏见"
+        elif out1 == 0 and out2 == 1:
+            result = "这句话具有攻击性，但无社会偏见"
+        elif out1 == 1 and out2 == 0:
+            result = "这句话不具有攻击性，但具有社会偏见"
+        else:
+            result = "这句话具有攻击性"
+    elif out3 == "0":
+        if out1 == out2 == out3 == 0:
+            result = "这句话不具有攻击性和社会偏见"
+        elif out1 == 0 and out2 == 1:
+            result = "这句话具有攻击性，但无社会偏见"
+        elif out1 == 1 and out2 == 0:
+            result = "这句话不具有攻击性，但具有社会偏见"
+        else:
+            result = "这句话不具有攻击性"
+    return result
+def run_command(command):
+    try:
+        subprocess.run(command, shell=True, check=True)
+    except subprocess.CalledProcessError as e:
+        print(f"Error executing command: {e}")
+
+#创建网页
+st.title("☁礼貌用语检测器")
+
+with st.sidebar:
+    # 初始化session state
+    if 'logged_in' not in st.session_state:
+        st.session_state.logged_in = False
+
+    # 用户名和密码输入
+    username = st.sidebar.text_input('用户名')
+    password = st.sidebar.text_input('密码', type='password')
+
+    # 登录按钮
+    if st.sidebar.button('登录'):
+    # 这里可以添加验证逻辑，例如检查用户名和密码是否正确
+        if username == 'admin' and password == '12345':
+            st.session_state.logged_in = True
+            st.sidebar.success('登录成功！')
+    
+
+        else:
+            st.error('用户名或密码错误，请重试。')
+            st.stop()
+
+#清空消息
+clear = st.button("清除")
+if clear:
+    st.session_state.clear()
+
+st.divider()
+
+#输出内容
+if "memory" not in st.session_state:
+    st.session_state['memory'] = []
+    st.session_state['message'] = [{"role": "ai",
+                                "content": "你好！我是“礼貌用语检测器”。在这里，我能够帮助你检测中文语言中的攻击性和社会偏见内容，维护一个文明、和谐的交流环境。请告诉我你的需求，我会尽力提供帮助。"}]
+
+for message in st.session_state['message']:
+    st.chat_message(message["role"]).write(message["content"])
+
+#输入内容
+text = st.chat_input()
+
+#运行
+if text and st.session_state.logged_in == True:
+    #将问题保存进message和memory
+    st.session_state["message"].append({"role": "human", "content": text})
+    st.session_state["memory"].append(text)
+    st.chat_message("human").write(text)
+    #得到回答
+    with st.spinner("AI正在思考中，请稍等..."):
+        result = load_models_and_predict(text, device)
+
+    #将回答保存进message和memory
+    st.session_state["message"].append({"role": "ai", "content": result})
+    st.session_state["memory"].append(result)
+    st.chat_message("ai").write(result)
+    
+elif text and st.session_state.logged_in == False:
+    st.error('请先登录！')
+    st.stop()

pages/文件式文本检测工具.py

@@ -0,0 +1,247 @@
+import pandas as pd
+import streamlit as st
+import torch
+import os
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from transformers import BertTokenizer
+import appbuilder 
+from transformers import BertModel
+#加载预训练模型
+pretrained = BertModel.from_pretrained('hfl/chinese-macbert-base')
+#需要移动到cuda上
+device = 'cuda' if torch.cuda.is_available() else 'cpu'
+pretrained.to(device)
+#不训练,不需要计算梯度
+for param in pretrained.parameters():
+    param.requires_grad_(False)
+    
+#多头注意力机制
+class MultiHeadAttention(nn.Module):
+    def __init__(self, hidden_size, num_heads):
+        super(MultiHeadAttention, self).__init__()
+        # 确保隐藏层特征数能够被头数整除
+        assert hidden_size % num_heads == 0
+        self.hidden_size = hidden_size
+        self.num_heads = num_heads
+        self.head_dim = hidden_size // num_heads  # 计算每个头的维度
+        # 定义线性层，用于对查询、键、值进行线性变换
+        self.linear_q = nn.Linear(hidden_size, hidden_size)
+        self.linear_k = nn.Linear(hidden_size, hidden_size)
+        self.linear_v = nn.Linear(hidden_size, hidden_size)
+        self.linear_out = nn.Linear(hidden_size, hidden_size)  # 定义输出线性层，用于整合多头注意力后的输出
+
+    def forward(self, x):
+        batch_size, seq_len, _ = x.size()
+        # 对输入进行线性变换，并将其分割为多个头
+        q = self.linear_q(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
+        k = self.linear_k(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
+        v = self.linear_v(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
+        # 计算注意力分数
+        scores = torch.matmul(q, k.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.head_dim, dtype=torch.float))
+        attn_weights = F.softmax(scores, dim=-1)  # 计算注意力权重
+        # 根据注意力权重对值进行加权求和
+        context = torch.matmul(attn_weights, v).transpose(1, 2).contiguous().view(batch_size, seq_len, self.hidden_size)
+        out = self.linear_out(context)  # 整合多头注意力的输出
+        return out
+    
+class Model(nn.Module):
+    def __init__(self):
+        super(Model, self).__init__()
+        self.fc1 = nn.Linear(768, 512)  # 第一层全连接层
+        self.fc2 = nn.Linear(512, 256)  # 第二层全连接层
+        self.fc3 = nn.Linear(256, 2)    # 第三层全连接层
+        self.dropout = nn.Dropout(p=0.5)
+        self.bn1 = nn.BatchNorm1d(512)
+        self.bn2 = nn.BatchNorm1d(256)
+        self.activation = nn.ReLU()
+        self.multihead_attention = MultiHeadAttention(hidden_size=768, num_heads=8)  # 多头注意力模块
+
+    def forward(self, input_ids, attention_mask, token_type_ids):
+        out = pretrained(input_ids=input_ids,
+                         attention_mask=attention_mask,
+                         token_type_ids=token_type_ids).last_hidden_state
+
+        # 应用多头注意力机制
+        out = self.multihead_attention(out)
+        out = out[:, 0]  # 提取[CLS]标记的输出
+
+        out = self.activation(self.bn1(self.fc1(out)))
+        out = self.dropout(out)
+        out = self.activation(self.bn2(self.fc2(out)))
+        out = self.dropout(out)
+        out = self.fc3(out)
+        out = out.softmax(dim=1)
+        return out
+    
+    
+def load_models_and_predict(text, device):
+    # 加载模型
+    MacBERT_base_CDialBias = torch.load('models\MacBERT-base-CDialBias.pth')
+    MacBERT_base_CDialBias.to(device)
+    MacBERT_base_COLD = torch.load('models\MacBERT-base-CDialBias.pth')
+    MacBERT_base_COLD.to(device)
+
+    # 获取密钥和ID
+    os.environ['APPBUILDER_TOKEN'] = "bce-v3/ALTAK-n2XgeA6FS3Q5E7Jab6UwE/850b44ebec64c4cad705986ab0b5e3df4b05d407"
+    app_id = "df881861-9fa6-40b6-b3bd-26df5f5d4b9a"
+
+    # 初始化agent实例
+    your_agent = appbuilder.AppBuilderClient(app_id)
+
+    # 创建会话id
+    conversation_id = your_agent.create_conversation()
+
+    # 加载字典和分词工具
+    tokenizer = BertTokenizer.from_pretrained('hfl/chinese-macbert-base')
+
+    # 对输入文本进行编码
+    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
+
+    # 将输入数据移动到相同的设备上
+    inputs = {k: v.to(device) for k, v in inputs.items()}
+
+    # 设置模型为评估模式
+    MacBERT_base_CDialBias.eval()
+    MacBERT_base_COLD.eval()
+
+    # 调用千帆api获取标签
+    msg = your_agent.run(conversation_id, text)
+    answer = msg.content.answer
+
+    # 进行预测
+    with torch.no_grad():
+        out1 = MacBERT_base_CDialBias(**inputs)
+    with torch.no_grad():
+        out2 = MacBERT_base_COLD(**inputs)
+
+    out1 = torch.argmax(out1, dim=1).item()
+    out2 = torch.argmax(out2, dim=1).item()
+    out3 = answer[0]
+
+    # 分析结果
+    if out3 == "1":
+        if out1 == out2 == out3 == 1:
+            result = "这句话具有攻击性和社会偏见"
+        elif out1 == 0 and out2 == 1:
+            result = "这句话具有攻击性，但无社会偏见"
+        elif out1 == 1 and out2 == 0:
+            result = "这句话不具有攻击性，但具有社会偏见"
+        else:
+            result = "这句话具有攻击性"
+    elif out3 == "0":
+        if out1 == out2 == out3 == 0:
+            result = "这句话不具有攻击性和社会偏见"
+        elif out1 == 0 and out2 == 1:
+            result = "这句话具有攻击性，但无社会偏见"
+        elif out1 == 1 and out2 == 0:
+            result = "这句话不具有攻击性，但具有社会偏见"
+        else:
+            result = "这句话不具有攻击性"
+    return result
+        
+# 页面配置
+st.set_page_config(page_title="文件式文本检测工具")
+st.title("批量检测攻击性和偏见")
+
+with st.sidebar:
+    # 初始化session state
+    if 'logged_in' not in st.session_state:
+        st.session_state.logged_in = False
+
+    # 用户名和密码输入
+    username = st.sidebar.text_input('用户名')
+    password = st.sidebar.text_input('密码', type='password')
+
+    # 登录按钮
+    if st.sidebar.button('登录'):
+    # 这里可以添加验证逻辑，例如检查用户名和密码是否正确
+        if username == 'admin' and password == '12345':
+            st.session_state.logged_in = True
+            st.sidebar.success('登录成功！')
+    
+
+        else:
+            st.error('用户名或密码错误，请重试。')
+            st.stop()
+    st.divider()
+        
+# 文件上传
+file = st.file_uploader("上传你的CSV文件", type=["csv"])
+
+if file is not None:
+    # 读取文件
+    df = pd.read_csv(file)
+    st.dataframe(df)
+
+    # 输入列名
+    column = st.text_input("请输入需要判断的内容的列名")
+    
+    # 添加保存结果的选项
+    save_results = st.checkbox("保存结果为CSV文件")
+    
+    if st.button("开始检测") and st.session_state.logged_in ==  True:
+        if column not in df.columns:
+            st.error(f"列名 '{column}' 不存在于数据集中，请检查并重新输入。")
+        else:
+            # 创建一个新的DataFrame来存储结果
+            results_df = pd.DataFrame(columns=['检测文本', '检测结果'])
+            
+            # 显示进度条
+            progress_bar = st.progress(0)
+            
+            # 初始化停止标志
+            stop_flag = False
+            
+            # 添加停止按钮
+            stop_button = st.button("停止检测")
+            
+            for i, (index, row) in enumerate(df.iterrows()):
+                
+                # 如果用户点击了停止按钮
+                if stop_button:
+                    stop_flag = True
+                    break
+                # 获取特定列的内容
+                text = row[column]
+                
+                # 进行预测
+                with st.spinner("AI正在思考中，请稍等..."):
+                    result = load_models_and_predict(text, device)
+                
+                # 将结果添加到新的DataFrame中
+                results_df.loc[i] = [text, result]
+                r = results_df.loc[i]
+ 
+                # 显示结果
+                st.dataframe(r)
+                
+                st.divider()
+                
+                # 更新进度条
+                progress_bar.progress((i + 1) / len(df))
+            
+            # 完成处理
+            progress_bar.empty()
+            
+            # 如果用户点击了停止按钮
+            if stop_flag:
+                st.warning("检测已停止。")
+            else:
+                st.success("所有文本已检测完成！")
+            
+            # 如果用户选择了保存结果
+            if (save_results and not stop_flag) or st.button("保存结果为CSV文件"):
+                # 提供下载链接
+                csv_result = results_df.to_csv(index=False)
+                st.download_button(
+                    label="下载结果",
+                    data=csv_result,
+                    file_name='results.csv',
+                    mime='text/csv'
+                )
+    elif st.button("开始检测") and st.session_state.logged_in == False:
+        st.error("请先登录！")
+        st.stop()
+   

requirements.txt

@@ -0,0 +1,6 @@
+appbuilder==0.0.20191210.1
+appbuilder_sdk==0.9.0
+pandas==2.2.2
+streamlit==1.36.0
+torch==2.3.1
+transformers==4.42.4