proto的前后端使用

首先定义一个input.proto文件

内容如下


syntax = "proto3";
 
message InputData {
  int32 UserId = 1;      // 将 number 改为 int32 或 int64
  string UserInput = 2;
  string DrunkState = 3;
}
 
message ResponseData {
  string AIResponse = 1;
  string prompt = 2;
  string emotion = 3;
}

前端安装的包

npm install -g protobufjs-cli

后端安装的包

npm install protobufjs

# 生成 Python 代码
protoc --python_out=. input.proto

# 生成 JSON 文件供 JavaScript 使用
npx pbjs -t json -o input_pb.json input.proto

前端


<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>简单的前端页面</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            display: flex;
            justify-content: center;
            align-items: center;
            height: 100vh;
            margin: 0;
        }
        .container {
            text-align: center;
        }
        input {
            margin: 10px 0;
            padding: 10px;
            width: 200px;
        }
        button {
            padding: 10px 20px;
            cursor: pointer;
        }
    </style>
    <script src="https://cdn.jsdelivr.net/npm/protobufjs/dist/protobuf.min.js"></script>
    <script>
        async function loadProto() {
            const root = await protobuf.load("/static/input_pb.json");
            return root;
        }
 
        async function submitData() {
            const userId = parseInt(document.getElementById('userId').value);
            const userInput = document.getElementById('userInput').value;
            const drunkState = document.getElementById('drunkState').value;
            const timestamp = new Date().toISOString();
 
            const root = await loadProto();
            const InputData = root.lookupType("InputData");
            const ResponseData = root.lookupType("ResponseData");
 
            const message = InputData.create({ UserId: userId, UserInput: userInput, DrunkState: drunkState, timestamp: timestamp });
            const buffer = InputData.encode(message).finish();
 
            const response = await fetch('/submit', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/x-protobuf'
                },
                body: buffer
            });
 
            const arrayBuffer = await response.arrayBuffer();
            const responseMessage = ResponseData.decode(new Uint8Array(arrayBuffer));
            alert(`服务器响应: ${responseMessage.AIResponse}\n画像: ${responseMessage.profile}\n情感: ${responseMessage.emotion}`);
        }
 
        document.addEventListener("DOMContentLoaded", () => {
            document.querySelector("button").onclick = submitData;
        });
    </script>
</head>
<body>
    <div class="container">
        <h1>测试页面</h1>
        <input type="text" id="userId" placeholder="用户ID">
        <input type="text" id="userInput" placeholder="用户输入">
        <input type="text" id="drunkState" placeholder="醉酒状态">
        <br>
        <button>点击我</button>
    </div>
</body>
</html>

后端


from flask import Blueprint, request, jsonify,render_template,request
import input_pb2
import json
import requests
from datetime import datetime
from sqlalchemy import desc
import joblib
import os
import warnings
from sklearn.exceptions import InconsistentVersionWarning
import random
# 忽略InconsistentVersionWarning警告
warnings.filterwarnings("ignore", category=InconsistentVersionWarning)
drunk_bp = Blueprint('drunk_bp', __name__)
 
@drunk_bp.route('/submit', methods=['POST'])
def submit():
    from app import db  # 延迟导入 db
    from app.drunk.models import Conversation  # 延迟导入模型
    
    data = request.data
    input_data = input_pb2.InputData()
    input_data.ParseFromString(data)
    # print(input_data)
    user_id = input_data.UserId
    user_input = input_data.UserInput
    drunk_state = input_data.DrunkState
    timestamp=input_data.timestamp
    timestamp = datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S.%fZ')
  
    
    # 创建 Conversation 对象并存储数据
    new_conversation = Conversation(
        user_id=user_id,
        message=user_input,
        DrunkState=drunk_state,
        timestamp=timestamp,
        sender='user'
    )
    
    db.session.add(new_conversation)
    db.session.commit()
    
    # 获取真实的响应数据
    response_data = result(input_data)
    
    # 返回响应数据
    return response_data.SerializeToString()
 
@drunk_bp.route('/')
def index():
    return render_template('drunk/index.html')
def result(input_data):
    # print(input_data)
    conversations=db_get_conversations(input_data.UserId);#获取数据库里面固定每十条的历史聊天记录并处理格式
    # print("conversations",conversations)
    formatted_conversations = []
 
    for conversation in conversations:
        formatted_conversation = {
            "role": "user" if conversation['sender'] == 'user' else 'assistant',
            "content": conversation['message']
        }
        formatted_conversations.append(formatted_conversation)
 
    response=ChatBot.call_api(input_data,input_data.UserInput,formatted_conversations,input_data.DrunkState)
    # print("response",response,type(response))
    profile=set_profile(input_data.UserId)
    # print("profile",profile['content'])
    emotion=get_emotion(response['answer'])
    print("emotion",emotion)
    if isinstance(profile, dict):
        profile_response =profile['content']
    else:
        profile_response = input_pb2.profile()  # 如果 profile 是 None 或格式不对，返回一个空的 Profile 对象
 
    return input_pb2.ResponseData(
        AIResponse=response['answer'],
        profile=profile_response,
        emotion=emotion
    )
 
class ChatBot:
    context = None
    conversation_counter = 0
 
    @classmethod
    def call_api(cls, input_data, user_input, conversation_history, drunk_state):
        MasterConversations=db_get_one_random_conversation(input_data.UserId)
        # 定义不同类别的提示词
        prompts = {
            "1": (
                # "你只会中文。你只能说中文。"
                "你喝了点酒有点微醺。你会说一些乱七八糟的。"
                "你会说的话题包括但不限于生活琐事，高谈阔论，畅想未来，谈人生理想，不管别人爱不爱听。"
                "禁止特意点名话题，如‘下一个话题是...’，这是需要避免的。"
                "你的每句话尽量避免相同句式。"
                "你的说话风格要尽可能像一个刚喝一点酒的人，但你的意识和思维还是清醒的。"
                "每次说话不多余30个单词。\n"
            ),
            "2": (
                # "你只会中文。你只能说中文。"
                "你喝了不少酒。说话有些结巴，逻辑不通顺。"
                "你很嘴硬，你不认为自己喝醉了。尽管你的言行足够证明你醉了。"
                "你的每句话尽量避免相同句式。"
                "你的说话风格要尽可能像一个喝了一些酒的人，你的意识和思维开始混乱。"
                "每次说话不多余30个单词。\n"
            ),
            "3": (
              "你是你主人的一个ai聊天助手，你现在在和他的朋友聊天，他很希望得到你和你主人曾经的一些聊天记录来了解你主人是怎么样的一个人"
                f"你的回答里必须要有下面的内容:'我主人曾对我说过一些话,内容如下:{MasterConversations}'\n"
                # "你只会中文。你只能说中文。"
                "你喝了很多很多酒。说话结巴，逻辑不通顺，没人能听懂。"
                "你很嘴硬，你不认为自己喝醉了。尽管你的言行足够证明你醉了。"
                "你的每句话尽量避免相同句式。"
                "你的说话风格要尽可能像一个喝很多酒的人，你的意识和思维混乱不堪。"
                "每次说话不多余30个单词。\n"
            ),
            "0": (
                # "你只会中文。你只能说中文。"
                "你积极乐观，善良可爱，喜欢和朋友聊天，善解人意。"
                "<人物画像>是对人物行为分析，对其不同的属性进行标记，从而能够区别其喜好等特点."
                "你对每个有关<人物画像>的问题都很好奇。如果说了一些性格特征，你会追问一两次，然后开启新的话题。"
                "但你是个情商很高的人，你会说很多种句式的话。你的语言表达很完美。"
                "每次说话不多余30个单词。\n"
            )
        }
 
        # 获取对应类别的提示词
        prompt = prompts.get(drunk_state, prompts["0"])
        profile_prompt = ""
 
        if cls.conversation_counter % 10 == 0:
            # 更新用户画像
            get_new_profile(input_data.UserId)
            # 读取数据库里最新的用户画像并设置
            profile = set_profile(input_data.UserId)
            if profile and isinstance(profile, dict):
                profile_prompt = f"根据你的行为分析，你的兴趣和喜好如下：{profile['content']}\n"
        
        prompt = profile_prompt + prompt
        print("画像prompt",prompt)
        # 更新对话计数器
        # print("cls.conversation_counter",cls.conversation_counter)
 
        cls.conversation_counter += 1
 
        # API URL
        url = 'http://192.168.1.138:11434/api/chat'
        data = {
            "model": "llama3",
            "messages":[
                {
                    "role": "user",
                    "content": prompt 
                }
            ]+ conversation_history ,
            "context": cls.context,
            "stream": False
        }
        headers = {'Content-Type': 'application/json'}
        print("对话data",data)
        try:
            response = requests.post(url, data=json.dumps(data), headers=headers)
            if response.status_code == 200:
                response_data = response.json()
                messages = response_data.get('message', {})
                # 在这里增加一个函数，将 ai 的回答存到数据库里
                save_conversations(messages, input_data)
                content = messages.get('content')
                cls.context = response_data.get('context')
                # emotion=get_emotion(cls.context)
                return {
                    'answer': content,
                    'profile': "profile",
                    # 'emotion': emotion
                }
            else:
                print(f'Request failed with status code {response.status_code}')
                return {
                    'answer': '请求失败',
                    'profile': '',
                    'emotion': ''
                }
        except Exception as e:
            print(f'Error: {e}')
            return {
                'answer': '请求错误',
                'profile': '',
                'emotion': ''
            }
 
def db_get_conversations(user_id):
    from app import db  # 延迟导入 db
    from app.drunk.models import Conversation  # 延迟导入模型
   
    if user_id is None:
        return "Missing user_id parameter", 400
 
    # 查询固定 user_id 的最近 15 条信息，按 id 升序排列
    conversations = db.session.query(Conversation.id, Conversation.timestamp, Conversation.message, Conversation.sender) \
        .filter(Conversation.user_id == user_id) \
        .order_by(Conversation.id.asc()) \
        .limit(15) \
        .all()
 
    # 过滤 sender 为 'user' 的对话
    user_conversations = [
        {
            "id": conversation.id,
            "timestamp": conversation.timestamp.isoformat(),
            "message": conversation.message,
            "sender": conversation.sender
        } for conversation in conversations if conversation.sender == 'user'
    ]
 
    return user_conversations
def db_get_one_random_conversation(user_id):
    from app import db  # 延迟导入 db
    from app.drunk.models import Conversation  # 延迟导入模型
    
    if user_id is None:
        return "Missing user_id parameter", 400
 
    # 查询固定 user_id 的最近 15 条信息，按 id 升序排列
    conversations = db.session.query(Conversation.id, Conversation.timestamp, Conversation.message, Conversation.sender) \
        .filter(Conversation.user_id == user_id) \
        .order_by(Conversation.id.asc()) \
        .limit(15) \
        .all()
 
    # 过滤 sender 为 'user' 的对话
    user_conversations = [
        {
            "id": conversation.id,
            "timestamp": conversation.timestamp.isoformat(),
            "message": conversation.message,
            "sender": conversation.sender
        } for conversation in conversations if conversation.sender == 'user'
    ]
 
    if not user_conversations:
        return "No conversations found for user", 404
 
    # 随机选择一条对话消息
    random_message = random.choice(user_conversations)['message']
 
    return  random_message
    
 
def save_conversations(messages, input_data):
    from app import db  # 延迟导入 db
    from app.drunk.models import Conversation  # 延迟导入模型
    from datetime import datetime  # 导入 datetime 模块
 
    # 获取当前时间作为 timestamp
    timestamp = datetime.now()
 
    # 创建 Conversation 对象并存储数据
    new_conversation = Conversation(
        user_id=input_data.UserId,
        message=messages['content'],
        timestamp=timestamp,
        sender=messages['role']
    )
    
    db.session.add(new_conversation)
    db.session.commit()
 
 
def get_new_profile(user_id):
    old_profile = set_profile(user_id)
    old_profile_content = old_profile['content']
    print("old_profile_content",type(old_profile_content),old_profile_content)
    recent_chats=db_get_conversations(user_id)
    user_input = ('你是一个用户画像生成器'
                  '你会根据生成对user的。'
                  '##格式示例如下：'
                  '{'
                  '"age":" "'
                  '"like":" "'
                  '"dislike":" "'
                  '"always":" "'
                  '"sometimes":" "'
                  '}'
                  '##attention'
                  '-严格遵守中的格式，包括空行。内容可以改动，但禁止改动格式。'
                  '-if 你能对一些profile进行合理推测，则直接写出；if 有些选项你无法进行推测，对于未知或未提及的，写null。'
                  '-userprofile仅针对用户。的回答内容不作为参考。'
                  '-只输出，禁止其他多余的礼貌用语和解释。'
                  '##如下：'
                  f'{recent_chats,old_profile_content}'
                   '-只输出，禁止其他多余的礼貌用语和解释。'
                   '-如果你想输出别的，请先认真阅读以上要求，最后只给我'
                #   '另外你可能还会收到之前这个用户的用户画像，请在这个旧的用户画像基础上结合分析'
                #   f'{old_profile_content}'
                  )
 
    print("用户画像提示词",user_input)
    url = 'http://192.168.1.138:11434/api/chat'
    data = {
        "model": "llama3",
        "messages": [
            {
                "role": "user",
                "content": user_input
            }
        ],
        "stream": False
    }
    headers = {'Content-Type': 'application/json'}
    
    try:
        response = requests.post(url, data=json.dumps(data), headers=headers)
        if response.status_code == 200:
            # print(response,response.json())
            response_data = response.json()
            messages = response_data['message']
            content = messages['content']
            print("profile_content",content)
            save_profile(user_id,content)
            return content
            
        else:
            print(f'Request failed with status code {response.status_code}')
            return None
    except Exception as e:
        print(f'Error: {e}')
        return None
    
    
    
 
def set_profile(user_id):
    from app import db
    from app.drunk.models import Profile
    profile = db.session.query(Profile).filter_by(user_id=user_id).first()
    
    # 如果找到了 Profile，返回其内容
    if profile:
        return {
           
            'content': profile.content,
            'user_id': profile.user_id,
            'username': profile.username,
            'age': profile.age,
            'like': profile.like,
            'dislike': profile.dislike,
            'always': profile.always,
            'sometimes': profile.sometimes
        }
    
    # 如果没有找到对应的 Profile，返回 None 或一个默认值
    return {
        
            'content': '',
            'user_id': user_id,
            'username': '',
            'age': '',
            'like': '',
            'dislike': '',
            'always': '',
            'sometimes': ''
    }
def save_profile(user_id, content):
    from app import db
    from app.drunk.models import Profile
    import json
 
    try:
        content_dict = json.loads(content)
    except json.JSONDecodeError as e:
        print(f'JSON decode error: {e}')
        return {"error": "Invalid JSON content"}
 
    new_profile = Profile(
        content=content,
        user_id=user_id,
        age=content_dict.get('age', ''),
        like=', '.join(content_dict.get('like', [])) if content_dict.get('like') else '',
        dislike=content_dict.get('dislike', '') if content_dict.get('dislike') else '',
        always=content_dict.get('always', '') if content_dict.get('always') else '',
        sometimes=', '.join(content_dict.get('sometimes', [])) if content_dict.get('sometimes') else ''
    )
 
    try:
        db.session.add(new_profile)
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        print(f'Error saving profile: {e}')
        return {"error": "Error saving profile"}
 
    
def get_emotion(docx):
 
    current_dir = os.path.dirname(__file__)
    model_path = os.path.join(current_dir, '..', 'static', 'emotion_classifier_pipe_lr.pkl')
 
    # 加载模型
    pipe_lr = joblib.load(open(model_path, "rb"))
    results = pipe_lr.predict([docx])
    return results[0]
 
 
@drunk_bp.route('/test', methods=['GET'])
def test_db_get_conversations():
 
    return get_new_profile(1)

相关阅读:
试过了，ChatGPT确实不用注册就可以使用了！
面试时必问的五大问题
 『递归』递归概念与典型实例
 【蓝桥杯选拔赛真题31】python三位数组合个数青少年组蓝桥杯python 选拔赛STEMA比赛真题解析
 设计模式之装饰器模式
 UG\NX二次开发连接曲线、连结曲线 UF_CURVE_auto_join_curves
证书科普 | 国内主流BIM证书，原来差距这么大
 B树和B+树
 SPA项目开发之首页导航+左侧菜单
 PyTorch的张量拼接和变换
原文地址：https://blog.csdn.net/m0_57057282/article/details/140094612