difyq 接入

zhaoqingang

2024-11-25 dc478b065693dd24e4cae719186d6aafb2d24f6d

 app/api/chat.py

@@ -14,6 +14,7 @@
from app.models.user_model import UserModel
from app.service.dialog import update_session_history
from app.service.basic import BasicService
from app.service.difyService import DifyService
from app.service.ragflow import RagflowService
from app.service.service_token import get_bisheng_token, get_ragflow_token
from app.service.session import SessionService
@@ -286,6 +287,93 @@
        finally:
            await websocket.close()
            print(f"Client {agent_id} disconnected")
    if agent_type == AgentType.DIFY:
        dify_service = DifyService(settings.dify_base_url)
        # token = get_dify_token(db, current_user.id)
        token = settings.dify_api_token
        try:
            async def forward_to_dify():
                while True:
                    conversation_id = ""
                    receive_message = await websocket.receive_json()
                    print(f"Received from client {chat_id}: {receive_message}")
                    upload_file_id = receive_message.get('upload_file_id', [])
                    question = receive_message.get('message', "")
                    if not question and not image_url:
                        await websocket.send_json({"message": "Invalid request", "type": "error"})
                        continue
                    try:
                        session = SessionService(db).create_session(
                            chat_id,
                            question,
                            agent_id,
                            AgentType.DIFY,
                            current_user.id
                        )
                        conversation_id = session.conversation_id
                    except Exception as e:
                        logger.error(e)
                    complete_response = ""
                    async for rag_response in dify_service.chat(token, chat_id, question, upload_file_id, conversation_id):
                        try:
                            if rag_response[:5] == "data:":
                                # 如果是，则截取掉前5个字符，并去除首尾空白符
                                text = rag_response[5:].strip()
                            else:
                                # 否则，保持原样
                                text = rag_response
                            complete_response += text
                            try:
                                data = json.loads(complete_response)
                                # data = json_data.get("data")
                                if "answer" not in  data:  # 信息过滤
                                    continue
                                else:  # 正常输出
                                    answer = data.get("answer", "")

                                    result = {"message": answer, "type": "message"}
                                    try:
                                        SessionService(db).update_session(chat_id,
                                                                          message={"role": "assistant", "content": data, "conversation_id": data.get("conversation_id")})
                                    except Exception as e:
                                        logger.error(e)
                                await websocket.send_json(result)
                                complete_response = ""
                            except json.JSONDecodeError as e:
                                print(f"Error decoding JSON: {e}")
                                # print(f"Response text: {text}")
                        except Exception as e2:
                            result = {"message": f"内部错误： {e2}", "type": "close"}
                            await websocket.send_json(result)
                            print(f"Error process message of ragflow: {e2}")
                    try:
                        dialog_chat_history = await ragflow_service.get_session_history(token, chat_id, 1)
                        await update_session_history(db, dialog_chat_history, current_user.id)
                    except Exception as e:
                        logger.error(e)
                        logger.error("-----------------保存ragflow的历史会话异常-----------------")

            # 启动任务处理客户端消息
            tasks = [
                asyncio.create_task(forward_to_dify())
            ]
            await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        except WebSocketDisconnect as e1:
            print(f"Client {chat_id} disconnected: {e1}")
            await websocket.close()
        except Exception as e:
            print(f"Exception occurred: {e}")

        finally:
            print("Cleaning up resources of ragflow")
            # 取消所有任务
            for task in tasks:
                if not task.done():
                    task.cancel()
                    try:
                        await task
                    except asyncio.CancelledError:
                        pass
    else:
        ret = {"message": "Agent not found", "type": "close"}
        await websocket.send_json(ret)