From afc68af3a67b2f936429feb3a68c1f09e03ce930 Mon Sep 17 00:00:00 2001
From: zhaoqingang <zhaoqg0118@163.com>
Date: 星期五, 11 四月 2025 10:07:24 +0800
Subject: [PATCH] 智能体中心不显示自研智能体
---
app/service/v2/chat.py | 839 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 818 insertions(+), 21 deletions(-)
diff --git a/app/service/v2/chat.py b/app/service/v2/chat.py
index 136a113..a4d6986 100644
--- a/app/service/v2/chat.py
+++ b/app/service/v2/chat.py
@@ -1,28 +1,825 @@
+import asyncio
+import datetime
+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, RG_CHAT_UPDATE_URL, DF_WORKFLOW_DRAFT, DF_WORKFLOW_PUBLISH
+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 service_chat_dialog(chat_id:str, question: str, session_id: str):
- token = "ragflow-YzMzE1NDRjYzMyZjExZWY5ZjkxMDI0Mm"
- url = f"/api/v1/chats/{chat_id}/completions"
- chat = ChatDialog(token)
- data = {
- "question": question,
- "stream": True,
- "session_id": session_id
- }
- headers = {
- 'Content-Type': 'application/json',
- 'Authorization': f"Bearer {token}"
- }
+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:
- for ans in chat.chat_completions(url, data, headers):
-
- yield "data:" + json.dumps({"code": 0, "message": "", "data": ans}, ensure_ascii=False) + "\n\n"
- ChatSessionModel.update_by_id(conv.id, conv.to_dict())
+ 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:
- yield "data:" + json.dumps({"code": 500, "message": str(e),
- "data": {"answer": "**ERROR**: " + str(e), "reference": []}},
- ensure_ascii=False) + "\n\n"
- yield "data:" + json.dumps({"code": 0, "message": "", "data": True}, ensure_ascii=False) + "\n\n"
\ No newline at end of file
+ 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 get_workflow_token(db):
+ user_token = db.query(UserTokenModel).filter(UserTokenModel.id == workflow_server).first()
+ return user_token.access_token if user_token else ""
+
+
+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')
+ # 灏嗗瓧绗︿覆鍒囧垎鎴恡oken骞惰浆鎹负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 ""
+
+
+async def service_complex_model(db, chat_type, model_type, model_name, model_provider):
+ if chat_type == 1 and model_type == 1:
+ return await set_dialog_model(db, complex_knowledge_chat, model_name,model_provider)
+ elif chat_type == 1 and model_type == 2:
+ return await set_dialog_model(db, complex_knowledge_chat_deep, model_name, model_provider)
+ else:
+ if model_type == 1:
+ chats = [complex_dialog_chat, complex_network_chat, complex_mindmap_chat, complex_content_optimization_chat]
+ else:
+ chats = [complex_dialog_chat, complex_network_chat]
+ return await set_workflow_model(db,chats
+ , # , complex_network_chat, complex_mindmap_chat, complex_content_optimization_chat
+ model_type, model_name, model_provider)
+
+
+async def set_dialog_model(db, chat_mode, model_name, model_provider):
+ chat = await ComplexChatDao(db).get_complex_chat_by_mode(chat_mode)
+ if chat:
+ access_token = await get_chat_token(db, rg_api_token)
+ url = settings.fwr_base_url + RG_CHAT_UPDATE_URL.format(chat.id)
+ chat_base = ChatBaseApply()
+ payload = {
+ "name": chat.name,
+ "llm": {
+ "model_name": model_name
+ }
+ }
+ response = await chat_base.chat_put(url, payload, await chat_base.get_headers(access_token))
+ # print(response)
+ if not response:
+ return "鏈嶅姟寮傚父,淇敼澶辫触锛�"
+ await ComplexChatDao(db).update_complex_chat_by_id(chat.id, {"chat_model": model_name, "chat_model_ds": model_name, "chat_provider": model_provider, "update_date": datetime.datetime.now()})
+ return ""
+
+async def set_workflow_model(db, chat_modes, model_type, model_name, model_provider):
+ chat_base = ChatBaseApply()
+ token = await get_workflow_token(db)
+ for chat_mode in chat_modes:
+ chat = await ComplexChatDao(db).get_complex_chat_by_mode(chat_mode)
+ if chat:
+
+ get_draft_url = settings.dify_base_url + DF_WORKFLOW_DRAFT.format(chat.id)
+ draft_data = await chat_base.chat_get(get_draft_url, {}, await chat_base.get_headers(token))
+ if draft_data:
+ graph = draft_data.get("graph")
+ for node in graph.get("nodes"):
+ if node.get("data", {}).get("type") == "llm":
+ if model_type == 1 and "娣卞害鎼滅储" not in node.get("data", {}).get("title"):
+ node["data"]["model"]["name"] = model_name
+ node["data"]["model"]["provider"] = model_provider
+ elif model_type == 2 and "娣卞害鎼滅储" in node.get("data", {}).get("title"):
+ node["data"]["model"]["name"] = model_name
+ node["data"]["model"]["provider"] = model_provider
+
+ draft_data_query = {"conversation_variables": draft_data.get("conversation_variables"),
+ "environment_variables": draft_data.get("environment_variables"),
+ "hash": draft_data.get("hash"),
+ "features": draft_data.get("features"),
+ "graph": graph}
+ set_draft_data = await chat_base.chat_post(get_draft_url, draft_data_query, await chat_base.get_headers(token))
+ if set_draft_data and set_draft_data.get("result") == "success":
+ publish_url = settings.dify_base_url + DF_WORKFLOW_PUBLISH.format(chat.id)
+ publish_data = await chat_base.chat_post(publish_url, {}, await chat_base.get_headers(token))
+ if publish_data and publish_data.get("result") == "success":
+ update_kwargs = {"chat_provider": model_provider, "update_date": datetime.datetime.now()}
+ if model_type == 1:
+ update_kwargs["chat_model"] = model_name
+ else:
+ update_kwargs["chat_model_ds"] = model_name
+ await ComplexChatDao(db).update_complex_chat_by_id(chat.id, update_kwargs)
+
+
+async def service_get_complex_model(db):
+ res = {}
+ for complexs in await ComplexChatDao(db).aget_complex_chat():
+ if complexs.chat_mode == complex_knowledge_chat:
+ res["dialog"] = {"modelName": complexs.chat_model, "modelProvider": complexs.chat_provider}
+ elif complexs.chat_mode == complex_knowledge_chat_deep:
+ res["dialog_ds"] = {"modelName": complexs.chat_model_ds, "modelProvider": complexs.chat_provider}
+ else:
+ res["workflow"] = {"modelName": complexs.chat_model, "modelProvider": complexs.chat_provider}
+ res["workflow_ds"] = {"modelName": complexs.chat_model_ds, "modelProvider": complexs.chat_provider}
+
+ return json.dumps(res)
+
+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())
--
Gitblit v1.8.0