-
Notifications
You must be signed in to change notification settings - Fork 0
/
seus_app.py
110 lines (81 loc) · 3.1 KB
/
seus_app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
import os
import sys
from flask import Flask, render_template, request, jsonify, send_from_directory
seus_app = Flask(__name__)
sys.path.append(os.path.join(os.path.dirname(__file__), 'python_script'))
from parameters import load_config
global DATA_PATH
load_config('seus')
from parameters import CHROMA_ROOT_PATH, EMBEDDING_MODEL, LLM_MODEL, PROMPT_TEMPLATE, DATA_PATH, REPHRASING_PROMPT, STANDALONE_PROMPT, ROUTER_DECISION_PROMPT
from get_llm_function import get_llm_function
from get_rag_chain import get_rag_chain
from ConversationalRagChain import ConversationalRagChain
root = "seus"
@seus_app.route("/")
def index():
return render_template('seus_index.html', root=root)
def init_app():
load_rag()
seus_app.config['UPLOAD_FOLDER'] = DATA_PATH
def load_rag(settings = None):
global rag_conv
if settings is None :
rag_conv = ConversationalRagChain.from_llm(
rag_chain=get_rag_chain(),
llm=get_llm_function(model_name = LLM_MODEL),
callbacks=None
)
else:
rag_conv = ConversationalRagChain.from_llm(
rag_chain=get_rag_chain(settings),
llm=get_llm_function(model_name = settings["llm_model"]),
callbacks=None
)
# Route to get the document list
@seus_app.route('/documents', methods=['GET'])
def list_documents():
files = os.listdir(seus_app.config['UPLOAD_FOLDER'])
documents = [{"name": f, "url": f"/{root}/files/{f}", "extension": os.path.splitext(f)[1][1:]} for f in files]
return jsonify(documents)
# Route to get a single document
@seus_app.route('/documents/<document_name>', methods=['GET'])
def get_document(document_name):
files = os.listdir(seus_app.config['UPLOAD_FOLDER'])
documents = [{"name": f, "url": f"/{root}/files/{f}", "extension": os.path.splitext(f)[1][1:]} for f in files]
document = next((doc for doc in documents if doc["name"] == document_name), None)
if document is None:
return jsonify({'error': 'Document not found'}), 404
return jsonify(document)
# Route to show the pdf
@seus_app.route('/files/<filename>', methods=['GET'])
def serve_file(filename):
return send_from_directory(seus_app.config['UPLOAD_FOLDER'], filename)
@seus_app.route("/get", methods=["POST"])
def chat():
data = request.get_json()
msg = data.get("msg", "")
return get_Chat_response(msg)
def get_Chat_response(query):
inputs = {
"query": str(query),
"chat_history": []
}
res = rag_conv._call(inputs)
output = jsonify({
'response': res['result'],
'context': res['context'],
'metadatas': res['metadatas']
})
return output
@seus_app.route('/update-settings', methods=['POST'])
def update_settings():
data = request.get_json()
load_rag(settings=data)
return jsonify({'status': 'success', 'message': 'Settings updated successfully'}), 200
@seus_app.route('/clear_chat_history', methods=['POST'])
def clear_chat_history():
rag_conv.clear_chat_history()
return jsonify({"status": "success"}), 200
if __name__ == '__main__':
init_app()
seus_app.run(port=6666,debug=False)