import asyncio
|
import io
|
import json
|
import time
|
import uuid
|
|
import fitz
|
from fastapi import HTTPException
|
from sqlalchemy import or_
|
|
from Log import logger
|
from app.config.agent_base_url import RG_CHAT_DIALOG, DF_CHAT_AGENT, DF_CHAT_PARAMETERS, RG_CHAT_SESSIONS, \
|
DF_CHAT_WORKFLOW, DF_UPLOAD_FILE, RG_ORIGINAL_URL
|
from app.config.config import settings
|
from app.config.const import *
|
from app.models import DialogModel, ApiTokenModel, UserTokenModel, ComplexChatSessionDao, ChatDataRequest, \
|
ComplexChatDao, KnowledgeModel, UserModel, KnowledgeUserModel
|
from app.models.v2.session_model import ChatSessionDao, ChatData
|
from app.service.v2.app_driver.chat_agent import ChatAgent
|
from app.service.v2.app_driver.chat_data import ChatBaseApply
|
from app.service.v2.app_driver.chat_dialog import ChatDialog
|
from app.service.v2.app_driver.chat_workflow import ChatWorkflow
|
from docx import Document
|
from dashscope import get_tokenizer # dashscope版本 >= 1.14.0
|
|
|
async def update_session_log(db, session_id: str, message: dict, conversation_id: str):
|
await ChatSessionDao(db).update_session_by_id(
|
session_id=session_id,
|
session=None,
|
message=message,
|
conversation_id=conversation_id
|
)
|
|
|
async def add_session_log(db, session_id: str, question: str, chat_id: str, user_id, event_type: str,
|
conversation_id: str, agent_type, query: dict=None):
|
try:
|
session = await ChatSessionDao(db).update_or_insert_by_id(
|
session_id=session_id,
|
name=question[:255],
|
agent_id=chat_id,
|
agent_type=agent_type,
|
tenant_id=user_id,
|
message={"role": "user", "content": question, "query": query},
|
conversation_id=conversation_id,
|
event_type=event_type
|
)
|
return session
|
except Exception as e:
|
logger.error(e)
|
return None
|
|
|
async def get_app_token(db, app_id):
|
app_token = db.query(UserTokenModel).filter_by(id=app_id).first()
|
if app_token:
|
return app_token.access_token
|
return ""
|
|
|
async def get_chat_token(db, app_id):
|
app_token = db.query(ApiTokenModel).filter_by(app_id=app_id).first()
|
if app_token:
|
return app_token.token
|
return ""
|
|
|
async def add_chat_token(db, data):
|
try:
|
api_token = ApiTokenModel(**data)
|
db.add(api_token)
|
db.commit()
|
except Exception as e:
|
logger.error(e)
|
|
|
async def get_chat_info(db, chat_id: str):
|
return db.query(DialogModel).filter_by(id=chat_id, status=Dialog_STATSU_ON).first()
|
|
|
async def get_chat_object(mode):
|
if mode == workflow_chat:
|
url = settings.dify_base_url + DF_CHAT_WORKFLOW
|
return ChatWorkflow(), url
|
else:
|
url = settings.dify_base_url + DF_CHAT_AGENT
|
return ChatAgent(), url
|
|
|
|
async def get_user_kb(db, user_id: int, kb_ids: list) -> list:
|
res = []
|
user = db.query(UserModel).filter(UserModel.id == user_id).first()
|
if user is None:
|
return res
|
query = db.query(KnowledgeModel)
|
if user.permission != "admin":
|
klg_list = [j.id for i in user.groups for j in i.knowledges]
|
for i in db.query(KnowledgeUserModel).filter(KnowledgeUserModel.user_id == user_id, KnowledgeUserModel.status == 1).all():
|
if i.kb_id not in klg_list:
|
klg_list.append(i.kb_id)
|
query = query.filter(or_(KnowledgeModel.id.in_(klg_list), KnowledgeModel.tenant_id == str(user_id)))
|
kb_list= query.all()
|
for kb in kb_list:
|
if kb.id in kb_ids:
|
if kb.permission == "team":
|
res.append(kb.id)
|
elif kb.tenant_id == str(user_id):
|
res.append(kb.id)
|
return res
|
else:
|
return kb_ids
|
|
|
async def service_chat_dialog(db, chat_id: str, question: str, session_id: str, user_id: int, mode: str, kb_ids: list):
|
conversation_id = ""
|
token = await get_chat_token(db, rg_api_token)
|
url = settings.fwr_base_url + RG_CHAT_DIALOG.format(chat_id)
|
kb_id = await get_user_kb(db, user_id, kb_ids)
|
if not kb_id:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: The agent has no knowledge base to work with!", "status": http_400},
|
ensure_ascii=False) + "\n\n"
|
return
|
chat = ChatDialog()
|
session = await add_session_log(db, session_id, question, chat_id, user_id, mode, session_id, RG_TYPE)
|
if session:
|
conversation_id = session.conversation_id
|
message = {"role": "assistant", "answer": "", "reference": {}}
|
try:
|
async for ans in chat.chat_completions(url, await chat.complex_request_data(question, kb_id, conversation_id),
|
await chat.get_headers(token)):
|
data = {}
|
error = ""
|
status = http_200
|
if ans.get("code", None) == 102:
|
error = ans.get("message", "error！")
|
status = http_400
|
event = smart_message_error
|
else:
|
if isinstance(ans.get("data"), bool) and ans.get("data") is True:
|
event = smart_message_end
|
else:
|
data = ans.get("data", {})
|
# conversation_id = data.get("session_id", "")
|
if "session_id" in data:
|
del data["session_id"]
|
message = data
|
event = smart_message_cover
|
message_str = "data: " + json.dumps(
|
{"event": event, "data": data, "error": error, "status": status, "session_id": session_id},
|
ensure_ascii=False) + "\n\n"
|
for i in range(0, len(message_str), max_chunk_size):
|
chunk = message_str[i:i + max_chunk_size]
|
# print(chunk)
|
yield chunk # 发送分块消息
|
except Exception as e:
|
|
logger.error(e)
|
try:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: " + str(e), "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
except:
|
...
|
finally:
|
message["role"] = "assistant"
|
await update_session_log(db, session_id, message, conversation_id)
|
|
|
async def data_process(data):
|
if isinstance(data, str):
|
return data.replace("dify", "smart")
|
elif isinstance(data, dict):
|
for k in list(data.keys()):
|
if isinstance(k, str) and "dify" in k:
|
new_k = k.replace("dify", "smart")
|
data[new_k] = await data_process(data[k])
|
del data[k]
|
else:
|
data[k] = await data_process(data[k])
|
return data
|
elif isinstance(data, list):
|
for i in range(len(data)):
|
data[i] = await data_process(data[i])
|
return data
|
else:
|
return data
|
|
|
async def service_chat_workflow(db, chat_id: str, chat_data: ChatData, session_id: str, user_id, mode: str):
|
conversation_id = ""
|
answer_event = ""
|
answer_agent = ""
|
answer_workflow = ""
|
download_url = ""
|
message_id = ""
|
task_id = ""
|
error = ""
|
files = []
|
node_list = []
|
token = await get_chat_token(db, chat_id)
|
chat, url = await get_chat_object(mode)
|
if hasattr(chat_data, "query"):
|
query = chat_data.query
|
else:
|
query = "start new conversation"
|
session = await add_session_log(db, session_id, query if query else "start new conversation", chat_id, user_id,
|
mode, conversation_id, DF_TYPE, chat_data.to_dict())
|
if session:
|
conversation_id = session.conversation_id
|
try:
|
async for ans in chat.chat_completions(url,
|
await chat.request_data(query, conversation_id, str(user_id), chat_data),
|
await chat.get_headers(token)):
|
data = {}
|
status = http_200
|
conversation_id = ans.get("conversation_id")
|
task_id = ans.get("task_id")
|
if ans.get("event") == message_error:
|
error = ans.get("message", "参数异常！")
|
status = http_400
|
event = smart_message_error
|
elif ans.get("event") == message_agent:
|
data = {"answer": ans.get("answer", ""), "id": ans.get("message_id", "")}
|
answer_agent += ans.get("answer", "")
|
message_id = ans.get("message_id", "")
|
event = smart_message_stream
|
elif ans.get("event") == message_event:
|
data = {"answer": ans.get("answer", ""), "id": ans.get("message_id", "")}
|
answer_event += ans.get("answer", "")
|
message_id = ans.get("message_id", "")
|
event = smart_message_stream
|
elif ans.get("event") == message_file:
|
data = {"url": ans.get("url", ""), "id": ans.get("id", ""),
|
"type": ans.get("type", "")}
|
files.append(data)
|
event = smart_message_file
|
elif ans.get("event") in [workflow_started, node_started, node_finished]:
|
data = ans.get("data", {})
|
data["inputs"] = await data_process(data.get("inputs", {}))
|
data["outputs"] = await data_process(data.get("outputs", {}))
|
data["files"] = await data_process(data.get("files", []))
|
data["process_data"] = ""
|
if data.get("status") == "failed":
|
status = http_500
|
error = data.get("error", "")
|
node_list.append(ans)
|
event = [smart_workflow_started, smart_node_started, smart_node_finished][
|
[workflow_started, node_started, node_finished].index(ans.get("event"))]
|
elif ans.get("event") == workflow_finished:
|
data = ans.get("data", {})
|
answer_workflow = data.get("outputs", {}).get("output", data.get("outputs", {}).get("answer"))
|
download_url = data.get("outputs", {}).get("download_url")
|
event = smart_workflow_finished
|
if data.get("status") == "failed":
|
status = http_500
|
error = data.get("error", "")
|
node_list.append(ans)
|
|
elif ans.get("event") == message_end:
|
event = smart_message_end
|
else:
|
continue
|
|
yield "data: " + json.dumps(
|
{"event": event, "data": data, "error": error, "status": status, "task_id": task_id,
|
"session_id": session_id},
|
ensure_ascii=False) + "\n\n"
|
|
except Exception as e:
|
logger.error(e)
|
try:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: " + str(e), "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
except:
|
...
|
finally:
|
await update_session_log(db, session_id, {"role": "assistant",
|
"answer": answer_event or answer_agent or answer_workflow or error,
|
"download_url": download_url,
|
"node_list": node_list, "task_id": task_id, "id": message_id,
|
"error": error}, conversation_id)
|
|
|
|
|
async def service_chat_basic(db, chat_id: str, chat_data: ChatData, session_id: str, user_id, mode: str):
|
|
if chat_id == basic_report_talk:
|
complex_chat = await ComplexChatDao(db).get_complex_chat_by_mode(chat_data.report_mode)
|
if complex_chat:
|
...
|
|
|
|
async def service_chat_parameters(db, chat_id, user_id):
|
chat_info = db.query(DialogModel).filter_by(id=chat_id).first()
|
if not chat_info:
|
return {}
|
return chat_info.parameters
|
|
|
async def service_chat_sessions(db, chat_id, name):
|
token = await get_chat_token(db, rg_api_token)
|
# print(token)
|
if not token:
|
return {}
|
url = settings.fwr_base_url + RG_CHAT_SESSIONS.format(chat_id)
|
chat = ChatDialog()
|
return await chat.chat_sessions(url, {"name": name}, await chat.get_headers(token))
|
|
|
async def service_chat_sessions_list(db, chat_id, current, page_size, user_id, keyword):
|
total, session_list = await ChatSessionDao(db).get_session_list(
|
user_id=user_id,
|
agent_id=chat_id,
|
keyword=keyword,
|
page=current,
|
page_size=page_size
|
)
|
return json.dumps({"total": total, "rows": [session.to_dict() for session in session_list]})
|
|
|
async def service_chat_session_log(db, session_id):
|
session_log = await ChatSessionDao(db).get_session_by_id(session_id)
|
if not session_log:
|
return {}
|
log_info =session_log.log_to_json()
|
if session_log.event_type == complex_chat:
|
|
total, message_list = await ComplexChatSessionDao(db).get_session_list(session_id)
|
log_info["message"] = [message.log_to_json() for message in message_list[::-1]]
|
|
return json.dumps(log_info)
|
|
|
async def service_chat_upload(db, chat_id, file, user_id):
|
files = []
|
token = await get_chat_token(db, chat_id)
|
if not token:
|
return files
|
url = settings.dify_base_url + DF_UPLOAD_FILE
|
chat = ChatBaseApply()
|
for f in file:
|
try:
|
file_content = await f.read()
|
file_upload = await chat.chat_upload(url, {"file": (f.filename, file_content)}, {"user": str(user_id)},
|
{'Authorization': f'Bearer {token}'})
|
try:
|
tokens = await read_file(file_content, f.filename, f.content_type)
|
file_upload["tokens"] = tokens
|
except:
|
...
|
files.append(file_upload)
|
except Exception as e:
|
logger.error(e)
|
return json.dumps(files) if files else ""
|
|
|
async def get_str_token(input_str):
|
# 获取tokenizer对象，目前只支持通义千问系列模型
|
tokenizer = get_tokenizer('qwen-turbo')
|
# 将字符串切分成token并转换为token id
|
tokens = tokenizer.encode(input_str)
|
return len(tokens)
|
|
|
async def read_pdf(pdf_stream):
|
text = ""
|
with fitz.open(stream=pdf_stream, filetype="pdf") as pdf_document:
|
for page in pdf_document:
|
text += page.get_text()
|
return text
|
|
|
async def read_word(word_stream):
|
# 使用 python-docx 打开 Word 文件流
|
doc = Document(io.BytesIO(word_stream))
|
|
# 提取每个段落的文本
|
text = ""
|
for para in doc.paragraphs:
|
text += para.text
|
|
return text
|
|
|
async def read_file(file, filename, content_type):
|
text = ""
|
if content_type == "application/pdf" or filename.endswith('.pdf'):
|
|
# 提取 PDF 内容
|
text = await read_pdf(file)
|
elif content_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document" or filename.endswith(
|
'.docx'):
|
text = await read_word(file)
|
|
return await get_str_token(text)
|
|
|
async def service_chunk_retrieval(query, knowledge_id, top_k, similarity_threshold, api_key):
|
# print(query)
|
|
try:
|
request_data = json.loads(query)
|
payload = {
|
"question": request_data.get("query", ""),
|
"dataset_ids": request_data.get("dataset_ids", []),
|
"page_size": top_k,
|
"similarity_threshold": similarity_threshold if similarity_threshold else 0.2
|
}
|
except json.JSONDecodeError as e:
|
fixed_json = query.replace("'", '"')
|
try:
|
request_data = json.loads(fixed_json)
|
payload = {
|
"question": request_data.get("query", ""),
|
"dataset_ids": request_data.get("dataset_ids", []),
|
"page_size": top_k,
|
"similarity_threshold": similarity_threshold if similarity_threshold else 0.2
|
}
|
except Exception:
|
payload = {
|
"question": query,
|
"dataset_ids": [knowledge_id],
|
"page_size": top_k,
|
"similarity_threshold": similarity_threshold if similarity_threshold else 0.2
|
}
|
# print(payload)
|
url = settings.fwr_base_url + RG_ORIGINAL_URL
|
chat = ChatBaseApply()
|
response = await chat.chat_post(url, payload, await chat.get_headers(api_key))
|
if not response:
|
raise HTTPException(status_code=500, detail="服务异常！")
|
records = [
|
{
|
"content": chunk["content"],
|
"score": chunk["similarity"],
|
"title": chunk.get("document_keyword", "Unknown Document"),
|
"metadata": {"document_id": chunk["document_id"],
|
"path": f"{settings.fwr_base_url}/document/{chunk['document_id']}?ext={chunk.get('document_keyword').split('.')[-1]}&prefix=document",
|
'highlight': chunk.get("highlight"), "image_id": chunk.get("image_id"),
|
"positions": chunk.get("positions"), }
|
}
|
for chunk in response.get("data", {}).get("chunks", [])
|
]
|
# print(len(records))
|
# print(records)
|
return records
|
|
|
async def service_base_chunk_retrieval(query, knowledge_id, top_k, similarity_threshold, api_key):
|
# request_data = json.loads(query)
|
payload = {
|
"question": query,
|
"dataset_ids": [knowledge_id],
|
"page_size": top_k,
|
"similarity_threshold": similarity_threshold
|
}
|
url = settings.fwr_base_url + RG_ORIGINAL_URL
|
# url = "http://192.168.20.116:11080/" + RG_ORIGINAL_URL
|
chat = ChatBaseApply()
|
response = await chat.chat_post(url, payload, await chat.get_headers(api_key))
|
if not response:
|
raise HTTPException(status_code=500, detail="服务异常！")
|
records = [
|
{
|
"content": chunk["content"],
|
"score": chunk["similarity"],
|
"title": chunk.get("document_keyword", "Unknown Document"),
|
"metadata": {"document_id": chunk["document_id"]}
|
}
|
for chunk in response.get("data", {}).get("chunks", [])
|
]
|
return records
|
|
|
async def add_complex_log(db, message_id, chat_id, session_id, chat_mode, query, user_id, mode, agent_type, message_type, conversation_id="", node_data=None, query_data=None):
|
if not node_data:
|
node_data = []
|
if not query_data:
|
query_data = {}
|
# print(node_data)
|
# print("--------------------------------------------------------")
|
# print(query_data)
|
try:
|
complex_log = ComplexChatSessionDao(db)
|
if not conversation_id:
|
session = await complex_log.get_session_by_session_id(session_id, chat_id)
|
if session:
|
conversation_id = session.conversation_id
|
await complex_log.create_session(message_id,
|
chat_id=chat_id,
|
session_id=session_id,
|
chat_mode=chat_mode,
|
message_type=message_type,
|
content=query,
|
event_type=mode,
|
tenant_id=user_id,
|
conversation_id=conversation_id,
|
node_data=json.dumps(node_data),
|
query=json.dumps(query_data),
|
agent_type=agent_type)
|
return conversation_id, True
|
|
except Exception as e:
|
logger.error(e)
|
return conversation_id, False
|
|
async def add_query_files(db, message_id):
|
query = {}
|
complex_log = await ComplexChatSessionDao(db).get_session_by_id(message_id)
|
if complex_log:
|
query = json.loads(complex_log.query)
|
return query.get("files", [])
|
|
async def service_complex_chat(db, chat_id, mode, user_id, chat_request: ChatDataRequest):
|
answer_event = ""
|
answer_agent = ""
|
answer_dialog = ""
|
answer_workflow = ""
|
download_url = ""
|
message_id = ""
|
task_id = ""
|
error = ""
|
node_list = []
|
reference= {}
|
conversation_id = ""
|
query_data = chat_request.to_dict()
|
new_message_id = str(uuid.uuid4())
|
inputs = {"is_deep": chat_request.isDeep}
|
files = chat_request.files
|
if chat_request.chatMode == complex_content_optimization_chat:
|
inputs["type"] = chat_request.optimizeType
|
elif chat_request.chatMode == complex_dialog_chat:
|
if not files and chat_request.parentId:
|
files = await add_query_files(db, chat_request.parentId)
|
if chat_request.chatMode != complex_content_optimization_chat:
|
await add_session_log(db, chat_request.sessionId, chat_request.query if chat_request.query else "未命名会话", chat_id, user_id,
|
mode, "", DF_TYPE)
|
conversation_id, message = await add_complex_log(db, new_message_id, chat_id, chat_request.sessionId, chat_request.chatMode, chat_request.query, user_id, mode, DF_TYPE, 1, query_data=query_data)
|
if not message:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: 创建会话失败！", "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
return
|
|
query_data["parentId"] = new_message_id
|
try:
|
if chat_request.chatMode == complex_knowledge_chat or chat_request.chatMode == complex_knowledge_chat_deep:
|
if not conversation_id:
|
session = await service_chat_sessions(db, chat_id, chat_request.query)
|
# print(session)
|
if not session or session.get("code") != 0:
|
yield "data: " + json.dumps(
|
{"message": smart_message_error, "error": "\n**ERROR**: chat agent error", "status": http_500})
|
return
|
conversation_id = session.get("data", {}).get("id")
|
token = await get_chat_token(db, rg_api_token)
|
url = settings.fwr_base_url + RG_CHAT_DIALOG.format(chat_id)
|
chat = ChatDialog()
|
try:
|
async for ans in chat.chat_completions(url, await chat.complex_request_data(chat_request.query, chat_request.knowledgeId, conversation_id),
|
await chat.get_headers(token)):
|
data = {}
|
error = ""
|
status = http_200
|
if ans.get("code", None) == 102:
|
error = ans.get("message", "error！")
|
status = http_400
|
event = smart_message_error
|
else:
|
if isinstance(ans.get("data"), bool) and ans.get("data") is True:
|
event = smart_message_end
|
else:
|
data = ans.get("data", {})
|
# conversation_id = data.get("session_id", "")
|
if "session_id" in data:
|
del data["session_id"]
|
data["prompt"] = ""
|
if not message_id:
|
message_id = data.get("id", "")
|
answer_dialog = data.get("answer", "")
|
reference = data.get("reference", {})
|
event = smart_message_cover
|
message_str = "data: " + json.dumps(
|
{"event": event, "data": data, "error": error, "status": status, "message_id":message_id,
|
"parent_id": new_message_id,
|
"session_id": chat_request.sessionId},
|
ensure_ascii=False) + "\n\n"
|
for i in range(0, len(message_str), max_chunk_size):
|
chunk = message_str[i:i + max_chunk_size]
|
# print(chunk)
|
yield chunk # 发送分块消息
|
except Exception as e:
|
|
logger.error(e)
|
try:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: " + str(e), "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
except:
|
...
|
else:
|
token = await get_chat_token(db, chat_id)
|
chat, url = await get_chat_object(mode)
|
async for ans in chat.chat_completions(url,
|
await chat.complex_request_data(chat_request.query, conversation_id, str(user_id), files=files, inputs=inputs),
|
await chat.get_headers(token)):
|
# print(ans)
|
data = {}
|
status = http_200
|
conversation_id = ans.get("conversation_id")
|
task_id = ans.get("task_id")
|
if ans.get("event") == message_error:
|
error = ans.get("message", "参数异常！")
|
status = http_400
|
event = smart_message_error
|
elif ans.get("event") == message_agent:
|
data = {"answer": ans.get("answer", ""), "id": ans.get("message_id", "")}
|
answer_agent += ans.get("answer", "")
|
message_id = ans.get("message_id", "")
|
event = smart_message_stream
|
elif ans.get("event") == message_event:
|
data = {"answer": ans.get("answer", ""), "id": ans.get("message_id", "")}
|
answer_event += ans.get("answer", "")
|
message_id = ans.get("message_id", "")
|
event = smart_message_stream
|
elif ans.get("event") == message_file:
|
data = {"url": ans.get("url", ""), "id": ans.get("id", ""),
|
"type": ans.get("type", "")}
|
files.append(data)
|
event = smart_message_file
|
elif ans.get("event") in [workflow_started, node_started, node_finished]:
|
data = ans.get("data", {})
|
data["inputs"] = await data_process(data.get("inputs", {}))
|
data["outputs"] = await data_process(data.get("outputs", {}))
|
data["files"] = await data_process(data.get("files", []))
|
data["process_data"] = ""
|
if data.get("status") == "failed":
|
status = http_500
|
error = data.get("error", "")
|
node_list.append(ans)
|
event = [smart_workflow_started, smart_node_started, smart_node_finished][
|
[workflow_started, node_started, node_finished].index(ans.get("event"))]
|
elif ans.get("event") == workflow_finished:
|
data = ans.get("data", {})
|
answer_workflow = data.get("outputs", {}).get("output", data.get("outputs", {}).get("answer"))
|
download_url = data.get("outputs", {}).get("download_url")
|
event = smart_workflow_finished
|
if data.get("status") == "failed":
|
status = http_500
|
error = data.get("error", "")
|
node_list.append(ans)
|
|
elif ans.get("event") == message_end:
|
event = smart_message_end
|
else:
|
continue
|
|
yield "data: " + json.dumps(
|
{"event": event, "data": data, "error": error, "status": status, "task_id": task_id, "message_id":message_id,
|
"parent_id": new_message_id,
|
"session_id": chat_request.sessionId},
|
ensure_ascii=False) + "\n\n"
|
|
except Exception as e:
|
logger.error(e)
|
try:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: " + str(e), "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
except:
|
...
|
finally:
|
# await update_session_log(db, session_id, {"role": "assistant",
|
# "answer": answer_event or answer_agent or answer_workflow or error,
|
# "download_url": download_url,
|
# "node_list": node_list, "task_id": task_id, "id": message_id,
|
# "error": error}, conversation_id)
|
if message_id:
|
await add_complex_log(db, message_id, chat_id, chat_request.sessionId, chat_request.chatMode, answer_event or answer_agent or answer_workflow or answer_dialog or error, user_id, mode, DF_TYPE, 2, conversation_id, node_data=node_list or reference, query_data=query_data)
|
|
async def service_complex_upload(db, chat_id, file, user_id):
|
files = []
|
token = await get_chat_token(db, chat_id)
|
if not token:
|
return files
|
url = settings.dify_base_url + DF_UPLOAD_FILE
|
chat = ChatBaseApply()
|
for f in file:
|
try:
|
file_content = await f.read()
|
file_upload = await chat.chat_upload(url, {"file": (f.filename, file_content)}, {"user": str(user_id)},
|
{'Authorization': f'Bearer {token}'})
|
# try:
|
# tokens = await read_file(file_content, f.filename, f.content_type)
|
# file_upload["tokens"] = tokens
|
# except:
|
# ...
|
files.append(file_upload)
|
except Exception as e:
|
logger.error(e)
|
return json.dumps(files) if files else ""
|
|
if __name__ == "__main__":
|
q = json.dumps({"query": "设备", "dataset_ids": ["fc68db52f43111efb94a0242ac120004"]})
|
top_k = 2
|
similarity_threshold = 0.5
|
api_key = "ragflow-Y4MGYwY2JlZjM2YjExZWY4ZWU5MDI0Mm"
|
|
|
# a = service_chunk_retrieval(q, top_k, similarity_threshold, api_key)
|
# print(a)
|
async def a():
|
b = await service_chunk_retrieval(q, top_k, similarity_threshold, api_key)
|
print(b)
|
|
|
asyncio.run(a())
|
