15308555518/ MaxKB
1
0
Fork 0
mirror of https://github.com/1Panel-dev/MaxKB.git synced 2025-12-25 17:22:55 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity
MaxKB/apps/application/views/chat_views.py

517 lines
32 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# coding=utf-8
"""
@project: maxkb
@Author：虎
@file： chat_views.py
@date：2023/11/14 9:53
@desc:
"""
from django.utils.translation import gettext_lazy as _
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from rest_framework.decorators import action
from rest_framework.parsers import MultiPartParser
from rest_framework.request import Request
from rest_framework.views import APIView
from application.serializers.chat_message_serializers import ChatMessageSerializer, OpenAIChatSerializer
from application.serializers.chat_serializers import ChatSerializers, ChatRecordSerializer
from application.swagger_api.chat_api import ChatApi, VoteApi, ChatRecordApi, ImproveApi, ChatRecordImproveApi, \
ChatClientHistoryApi, OpenAIChatApi
from application.views import get_application_operation_object
from common.auth import TokenAuth, has_permissions, OpenAIKeyAuth
from common.constants.authentication_type import AuthenticationType
from common.constants.permission_constants import Permission, Group, Operate, \
RoleConstants, ViewPermission, CompareConstants
from common.log.log import log
from common.response import result
from common.util.common import query_params_to_single_dict
from dataset.serializers.file_serializers import FileSerializer
class Openai(APIView):
authentication_classes = [OpenAIKeyAuth]
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("OpenAI Interface Dialogue"),
operation_id=_("OpenAI Interface Dialogue"),
request_body=OpenAIChatApi.get_request_body_api(),
responses=OpenAIChatApi.get_response_body_api(),
tags=[_("OpenAI Dialogue")])
def post(self, request: Request, application_id: str):
return OpenAIChatSerializer(data={'application_id': application_id, 'client_id': request.auth.client_id,
'client_type': request.auth.client_type}).chat(request.data)
class ChatView(APIView):
authentication_classes = [TokenAuth]
class Export(APIView):
authentication_classes = [TokenAuth]
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("Export conversation"),
operation_id=_("Export conversation"),
manual_parameters=ChatApi.get_request_params_api(),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
@log(menu='Conversation Log', operate="Export conversation",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def post(self, request: Request, application_id: str):
return ChatSerializers.Query(
data={**query_params_to_single_dict(request.query_params), 'application_id': application_id,
'user_id': request.user.id}).export(request.data)
class Open(APIView):
authentication_classes = [TokenAuth]
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get the session id according to the application id"),
operation_id=_("Get the session id according to the application id"),
manual_parameters=ChatApi.OpenChat.get_request_params_api(),
tags=[_("Application/Chat")])
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_ACCESS_TOKEN,
RoleConstants.APPLICATION_KEY],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))],
compare=CompareConstants.AND)
)
def get(self, request: Request, application_id: str):
return result.success(ChatSerializers.OpenChat(
data={'user_id': request.user.id, 'application_id': application_id}).open())
class OpenWorkFlowTemp(APIView):
authentication_classes = [TokenAuth]
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("Get the workflow temporary session id"),
operation_id=_("Get the workflow temporary session id"),
request_body=ChatApi.OpenWorkFlowTemp.get_request_body_api(),
responses=result.get_api_response(ChatApi.OpenTempChat.get_response_body_api()),
tags=[_("Application/Chat")])
def post(self, request: Request):
return result.success(ChatSerializers.OpenWorkFlowChat(
data={'user_id': request.user.id, **request.data}).open())
class OpenTemp(APIView):
authentication_classes = [TokenAuth]
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("Get a temporary session id"),
operation_id=_("Get a temporary session id"),
request_body=ChatApi.OpenTempChat.get_request_body_api(),
responses=result.get_api_response(ChatApi.OpenTempChat.get_response_body_api()),
tags=[_("Application/Chat")])
@has_permissions(RoleConstants.ADMIN, RoleConstants.USER)
def post(self, request: Request):
return result.success(ChatSerializers.OpenTempChat(
data={**request.data, 'user_id': request.user.id}).open())
class Message(APIView):
authentication_classes = [TokenAuth]
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("dialogue"),
operation_id=_("dialogue"),
request_body=ChatApi.get_request_body_api(),
tags=[_("Application/Chat")])
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY,
RoleConstants.APPLICATION_ACCESS_TOKEN],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def post(self, request: Request, chat_id: str):
return ChatMessageSerializer(data={'chat_id': chat_id, 'message': request.data.get('message'),
're_chat': (request.data.get(
're_chat') if 're_chat' in request.data else False),
'stream': (request.data.get(
'stream') if 'stream' in request.data else True),
'application_id': (request.auth.keywords.get(
'application_id') if request.auth.client_type == AuthenticationType.APPLICATION_ACCESS_TOKEN.value else None),
'client_id': request.auth.client_id,
'form_data': (request.data.get(
'form_data') if 'form_data' in request.data else {}),
'image_list': request.data.get(
'image_list') if 'image_list' in request.data else [],
'document_list': request.data.get(
'document_list') if 'document_list' in request.data else [],
'audio_list': request.data.get(
'audio_list') if 'audio_list' in request.data else [],
'other_list': request.data.get(
'other_list') if 'other_list' in request.data else [],
'client_type': request.auth.client_type,
'node_id': request.data.get('node_id', None),
'runtime_node_id': request.data.get('runtime_node_id', None),
'node_data': request.data.get('node_data', {}),
'chat_record_id': request.data.get('chat_record_id'),
'child_node': request.data.get('child_node')}
).chat()
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get the conversation list"),
operation_id=_("Get the conversation list"),
manual_parameters=ChatApi.get_request_params_api(),
responses=result.get_api_array_response(ChatApi.get_response_body_api()),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def get(self, request: Request, application_id: str):
return result.success(ChatSerializers.Query(
data={**query_params_to_single_dict(request.query_params), 'application_id': application_id,
'user_id': request.user.id}).list())
class Operate(APIView):
authentication_classes = [TokenAuth]
@action(methods=['DELETE'], detail=False)
@swagger_auto_schema(operation_summary=_("Delete a conversation"),
operation_id=_("Delete a conversation"),
tags=[_("Application/Conversation Log")])
@has_permissions(ViewPermission(
[RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.MANAGE,
dynamic_tag=keywords.get('application_id'))],
compare=CompareConstants.AND),
compare=CompareConstants.AND)
@log(menu='Conversation Log', operate="Delete a conversation",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def delete(self, request: Request, application_id: str, chat_id: str):
return result.success(
ChatSerializers.Operate(
data={'application_id': application_id, 'user_id': request.user.id,
'chat_id': chat_id}).delete())
class ClientChatHistoryPage(APIView):
authentication_classes = [TokenAuth]
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get client conversation list by paging"),
operation_id=_("Get client conversation list by paging"),
manual_parameters=result.get_page_request_params(
ChatClientHistoryApi.get_request_params_api()),
responses=result.get_page_api_response(ChatApi.get_response_body_api()),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.APPLICATION_ACCESS_TOKEN],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def get(self, request: Request, application_id: str, current_page: int, page_size: int):
return result.success(ChatSerializers.ClientChatHistory(
data={'client_id': request.auth.client_id, 'application_id': application_id}).page(
current_page=current_page,
page_size=page_size))
class Operate(APIView):
authentication_classes = [TokenAuth]
@action(methods=['DELETE'], detail=False)
@swagger_auto_schema(operation_summary=_("Client deletes conversation"),
operation_id=_("Client deletes conversation"),
tags=[_("Application/Conversation Log")])
@has_permissions(ViewPermission(
[RoleConstants.APPLICATION_ACCESS_TOKEN],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))],
compare=CompareConstants.AND),
compare=CompareConstants.AND)
@log(menu='Conversation Log', operate="Client deletes conversation",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def delete(self, request: Request, application_id: str, chat_id: str):
return result.success(
ChatSerializers.Operate(
data={'application_id': application_id, 'user_id': request.user.id,
'chat_id': chat_id}).logic_delete())
@action(methods=['PUT'], detail=False)
@swagger_auto_schema(operation_summary=_("Client modifies dialogue summary"),
operation_id=_("Client modifies dialogue summary"),
request_body=ChatClientHistoryApi.Operate.ReAbstract.get_request_body_api(),
responses=result.get_default_response(),
tags=[_("Application/Conversation Log")])
@has_permissions(ViewPermission(
[RoleConstants.APPLICATION_ACCESS_TOKEN, RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))],
compare=CompareConstants.AND),
compare=CompareConstants.AND)
@log(menu='Conversation Log', operate="Client modifies dialogue summary",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def put(self, request: Request, application_id: str, chat_id: str):
return result.success(
ChatSerializers.Operate(
data={'application_id': application_id, 'user_id': request.user.id,
'chat_id': chat_id}).re_abstract(request.data))
class Page(APIView):
authentication_classes = [TokenAuth]
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get the conversation list by page"),
operation_id=_("Get the conversation list by page"),
manual_parameters=result.get_page_request_params(ChatApi.get_request_params_api()),
responses=result.get_page_api_response(ChatApi.get_response_body_api()),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def get(self, request: Request, application_id: str, current_page: int, page_size: int):
return result.success(ChatSerializers.Query(
data={**query_params_to_single_dict(request.query_params), 'application_id': application_id,
'user_id': request.user.id}).page(current_page=current_page,
page_size=page_size))
class ChatRecord(APIView):
authentication_classes = [TokenAuth]
class Operate(APIView):
authentication_classes = [TokenAuth]
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get conversation record details"),
operation_id=_("Get conversation record details"),
manual_parameters=ChatRecordApi.get_request_params_api(),
responses=result.get_api_array_response(ChatRecordApi.get_response_body_api()),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY,
RoleConstants.APPLICATION_ACCESS_TOKEN],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def get(self, request: Request, application_id: str, chat_id: str, chat_record_id: str):
return result.success(ChatRecordSerializer.Operate(
data={'application_id': application_id,
'chat_id': chat_id,
'chat_record_id': chat_record_id}).one(request.auth.current_role))
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get a list of conversation records"),
operation_id=_("Get a list of conversation records"),
manual_parameters=ChatRecordApi.get_request_params_api(),
responses=result.get_api_array_response(ChatRecordApi.get_response_body_api()),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def get(self, request: Request, application_id: str, chat_id: str):
return result.success(ChatRecordSerializer.Query(
data={'application_id': application_id,
'chat_id': chat_id, 'order_asc': request.query_params.get('order_asc')}).list())
class Page(APIView):
authentication_classes = [TokenAuth]
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get the conversation history list by page"),
operation_id=_("Get the conversation history list by page"),
manual_parameters=result.get_page_request_params(
ChatRecordApi.get_request_params_api()),
responses=result.get_page_api_response(ChatRecordApi.get_response_body_api()),
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def get(self, request: Request, application_id: str, chat_id: str, current_page: int, page_size: int):
return result.success(ChatRecordSerializer.Query(
data={'application_id': application_id,
'chat_id': chat_id, 'order_asc': request.query_params.get('order_asc')}).page(current_page,
page_size))
class Vote(APIView):
authentication_classes = [TokenAuth]
@action(methods=['PUT'], detail=False)
@swagger_auto_schema(operation_summary=_("Like, Dislike"),
operation_id=_("Like, Dislike"),
manual_parameters=VoteApi.get_request_params_api(),
request_body=VoteApi.get_request_body_api(),
responses=result.get_default_response(),
tags=[_("Application/Chat")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY,
RoleConstants.APPLICATION_ACCESS_TOKEN],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
@log(menu='Conversation Log', operate="Like, Dislike",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def put(self, request: Request, application_id: str, chat_id: str, chat_record_id: str):
return result.success(ChatRecordSerializer.Vote(
data={'vote_status': request.data.get('vote_status'), 'chat_id': chat_id,
'chat_record_id': chat_record_id}).vote())
class ChatRecordImprove(APIView):
authentication_classes = [TokenAuth]
@action(methods=['GET'], detail=False)
@swagger_auto_schema(operation_summary=_("Get the list of marked paragraphs"),
operation_id=_("Get the list of marked paragraphs"),
manual_parameters=ChatRecordImproveApi.get_request_params_api(),
responses=result.get_api_response(ChatRecordImproveApi.get_response_body_api()),
tags=[_("Application/Conversation Log/Annotation")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))]
))
def get(self, request: Request, application_id: str, chat_id: str, chat_record_id: str):
return result.success(ChatRecordSerializer.ChatRecordImprove(
data={'chat_id': chat_id, 'chat_record_id': chat_record_id}).get())
class Improve(APIView):
authentication_classes = [TokenAuth]
@action(methods=['PUT'], detail=False)
@swagger_auto_schema(operation_summary=_("Annotation"),
operation_id=_("Annotation"),
manual_parameters=ImproveApi.get_request_params_api(),
request_body=ImproveApi.get_request_body_api(),
responses=result.get_api_response(ChatRecordApi.get_response_body_api()),
tags=[_("Application/Conversation Log/Annotation")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))],
), ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.DATASET,
operate=Operate.MANAGE,
dynamic_tag=keywords.get(
'dataset_id'))],
compare=CompareConstants.AND
), compare=CompareConstants.AND)
@log(menu='Conversation Log', operate="Annotation",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def put(self, request: Request, application_id: str, chat_id: str, chat_record_id: str, dataset_id: str,
document_id: str):
return result.success(ChatRecordSerializer.Improve(
data={'chat_id': chat_id, 'chat_record_id': chat_record_id,
'dataset_id': dataset_id, 'document_id': document_id}).improve(request.data))
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("Add to Knowledge Base"),
operation_id=_("Add to Knowledge Base"),
manual_parameters=ImproveApi.get_request_params_api_post(),
request_body=ImproveApi.get_request_body_api_post(),
responses=result.get_default_response(),
tags=[_("Application/Conversation Log/Add to Knowledge Base")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))],
), ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.DATASET,
operate=Operate.MANAGE,
dynamic_tag=keywords.get(
'dataset_id'))],
compare=CompareConstants.AND
), compare=CompareConstants.AND)
@log(menu='Conversation Log', operate="Add to Knowledge Base",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def post(self, request: Request, application_id: str, dataset_id: str):
return result.success(ChatRecordSerializer.PostImprove().post_improve(request.data))
class Operate(APIView):
authentication_classes = [TokenAuth]
@action(methods=['DELETE'], detail=False)
@swagger_auto_schema(operation_summary=_("Delete a Annotation"),
operation_id=_("Delete a Annotation"),
manual_parameters=ImproveApi.get_request_params_api(),
responses=result.get_api_response(ChatRecordApi.get_response_body_api()),
tags=[_("Application/Conversation Log/Annotation")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))],
), ViewPermission([RoleConstants.ADMIN, RoleConstants.USER],
[lambda r, keywords: Permission(group=Group.DATASET,
operate=Operate.MANAGE,
dynamic_tag=keywords.get(
'dataset_id'))],
compare=CompareConstants.AND
), compare=CompareConstants.AND)
@log(menu='Conversation Log', operate="Delete a Annotation",
get_operation_object=lambda r, k: get_application_operation_object(k.get('application_id')))
def delete(self, request: Request, application_id: str, chat_id: str, chat_record_id: str,
dataset_id: str,
document_id: str, paragraph_id: str):
return result.success(ChatRecordSerializer.Improve.Operate(
data={'chat_id': chat_id, 'chat_record_id': chat_record_id,
'dataset_id': dataset_id, 'document_id': document_id,
'paragraph_id': paragraph_id}).delete())
class UploadFile(APIView):
authentication_classes = [TokenAuth]
parser_classes = [MultiPartParser]
@action(methods=['POST'], detail=False)
@swagger_auto_schema(operation_summary=_("Upload files"),
operation_id=_("Upload files"),
manual_parameters=[
openapi.Parameter(name='application_id',
in_=openapi.IN_PATH,
type=openapi.TYPE_STRING,
required=True,
description=_('Application ID')),
openapi.Parameter(name='chat_id',
in_=openapi.IN_PATH,
type=openapi.TYPE_STRING,
required=True,
description=_('Conversation ID')),
openapi.Parameter(name='file',
in_=openapi.IN_FORM,
type=openapi.TYPE_FILE,
required=True,
description=_('Upload file'))
],
tags=[_("Application/Conversation Log")]
)
@has_permissions(
ViewPermission([RoleConstants.ADMIN, RoleConstants.USER, RoleConstants.APPLICATION_KEY,
RoleConstants.APPLICATION_ACCESS_TOKEN],
[lambda r, keywords: Permission(group=Group.APPLICATION, operate=Operate.USE,
dynamic_tag=keywords.get('application_id'))])
)
def post(self, request: Request, application_id: str, chat_id: str):
files = request.FILES.getlist('file')
file_ids = []
debug = request.data.get("debug", "false").lower() == "true"
meta = {'application_id': application_id, 'chat_id': chat_id, 'debug': debug}
for file in files:
file_url = FileSerializer(data={'file': file, 'meta': meta}).upload()
file_ids.append({'name': file.name, 'url': file_url, 'file_id': file_url.split('/')[-1]})
return result.success(file_ids)
Reference in New Issue View Git Blame Copy Permalink