Feat/2.2.0 beta2 (#1667)

Author: zgqgit

docker/docker-compose.yml

@@ -40,7 +40,7 @@ services:
 
   backend:
     container_name: bisheng-backend
-    image: dataelement/bisheng-backend:v2.2.0-beta1
+    image: dataelement/bisheng-backend:v2.2.0-beta2
     ports:
       - "7860:7860"
     environment:
@@ -78,7 +78,7 @@ services:
 
   backend_worker:
     container_name: bisheng-backend-worker
-    image: dataelement/bisheng-backend:v2.2.0-beta1
+    image: dataelement/bisheng-backend:v2.2.0-beta2
     environment:
       TZ: Asia/Shanghai
       BS_MILVUS_CONNECTION_ARGS: '{"host":"milvus","port":"19530","user":"","password":"","secure":false}'
@@ -109,7 +109,7 @@ services:
 
   frontend:
     container_name: bisheng-frontend
-    image: dataelement/bisheng-frontend:v2.2.0-beta1
+    image: dataelement/bisheng-frontend:v2.2.0-beta2
     ports:
       - "3001:3001"
     environment:

src/backend/bisheng/__init__.py

@@ -7,7 +7,7 @@
 
 try:
     # 通过ci去自动修改
-    __version__ = '2.2.0-beta1'
+    __version__ = '2.2.0-beta2'
 except metadata.PackageNotFoundError:
     # Case where package metadata is not available.
     __version__ = ''

src/backend/bisheng/api/errcode/base.py

@@ -59,6 +59,15 @@ def to_sse_event_instance(self, event: str = "error", data: any = None) -> dict:
             })
         }
 
+    def to_sse_event_instance_str(self, event: str = "error", data: any = None) -> str:
+        data = data if data is not None else {"exception": str(self), **self.kwargs}
+        msg = json.dumps({
+            "status_code": self.code,
+            "status_message": self.message,
+            "data": data
+        }, ensure_ascii=False)
+        return f'event: {event}\ndata: {msg}\n\n'
+
     def to_dict(self, data: any = None) -> dict:
         data = data if data is not None else {"exception": str(self), **self.kwargs}
         return {

src/backend/bisheng/api/errcode/flow.py

@@ -75,3 +75,8 @@ class WorkFlowVersionUpdateError(BaseErrorCode):
 class WorkFlowTaskBusyError(BaseErrorCode):
     Code: int = 10540
     Msg: str = '服务器线程数已满，请稍候再试'
+
+# 工作流任务其它错误
+class WorkFlowTaskOtherError(BaseErrorCode):
+    Code: int = 10541
+    Msg: str = '工作流任务执行失败'

src/backend/bisheng/api/errcode/linsight.py

@@ -1,11 +1,16 @@
-from .base import BaseErrorCode
+from bisheng.api.errcode.base import BaseErrorCode
 
 
 class SopFileError(BaseErrorCode):
     Code: int = 11010
     Msg: str = 'SOP文件格式不符合要求'
 
 
+class SopShowcaseError(BaseErrorCode):
+    Code: int = 11011
+    Msg: str = 'SOP设置精选案例失败'
+
+
 class FileUploadError(BaseErrorCode):
     __doc__ = 'Linsight上传文件失败'
     Code: int = 11020
@@ -53,6 +58,7 @@ class LinsightBishengLLMError(BaseErrorCode):
     Code: int = 11090
     Msg: str = '灵思Bisheng LLM相关错误'
 
+
 # 生成SOP内容失败
 class LinsightGenerateSopError(BaseErrorCode):
     Code: int = 11100
@@ -70,27 +76,32 @@ class LinsightSessionVersionRunningError(BaseErrorCode):
     Code: int = 11120
     Msg: str = '灵思会话版本已完成或正在执行，无法再次执行'
 
+
 # 开始执行灵思任务失败
 class LinsightStartTaskError(BaseErrorCode):
     Code: int = 11130
     Msg: str = '开始执行灵思任务失败'
 
+
 # 获取灵思队列排队状态失败
 class LinsightQueueStatusError(BaseErrorCode):
     Code: int = 11140
     Msg: str = '获取灵思队列排队状态失败'
 
+
 # 添加指导手册失败，向量存储添加数据失败
 class LinsightAddSopError(BaseErrorCode):
     Code: int = 11150
     Msg: str = '添加指导手册失败，向量存储添加数据失败'
 
+
 # 更新指导手册失败，向量存储更新数据失败
 class LinsightUpdateSopError(BaseErrorCode):
     Code: int = 11160
     Msg: str = '更新指导手册失败，向量存储更新数据失败'
 
+
 # 删除指导手册失败，向量存储删除数据失败
 class LinsightDeleteSopError(BaseErrorCode):
     Code: int = 11170
-    Msg: str = '删除指导手册失败，向量存储删除数据失败'
+    Msg: str = '删除指导手册失败，向量存储删除数据失败'

src/backend/bisheng/api/services/chat_imp.py

@@ -5,12 +5,12 @@
 from collections import defaultdict
 from datetime import datetime, timedelta
 
+from pydantic import BaseModel
+from websockets import connect
+
 from bisheng.api.errcode.http_error import ServerError
-from bisheng.api.v1.schemas import resp_500
 from bisheng.database.base import session_getter
 from bisheng.database.models.message import ChatMessage
-from pydantic import BaseModel
-from websockets import connect
 
 # 维护一个连接池
 connection_pool = defaultdict(asyncio.Queue)
@@ -54,6 +54,8 @@ async def clean_inactive_queues(queue: defaultdict, timeout_threshold: timedelta
 
 # 维护一个连接池
 connection_pool = defaultdict(TimedQueue)
+
+
 # clean_inactive_queues(connection_pool, timedelta(minutes=5))
 
 
@@ -106,25 +108,24 @@ def __str__(self) -> str:
 
 
 async def event_stream(
-    webosocket: connect,
-    message: str,
-    session_id: str,
-    model: str,
-    streaming: bool,
+        webosocket: connect,
+        message: str,
+        session_id: str,
+        model: str,
+        streaming: bool,
 ):
-
     payload = {'inputs': message, 'flow_id': model, 'chat_id': session_id}
     try:
         await webosocket.send(json.dumps(payload, ensure_ascii=False))
     except Exception as e:
-        yield ServerError(exception=e).to_sse_event_instance()
+        yield ServerError(exception=e).to_sse_event_instance_str()
         return
     sync = ''
     while True:
         try:
             msg = await webosocket.recv()
         except Exception as e:
-            yield ServerError(exception=e).to_sse_event_instance()
+            yield ServerError(exception=e).to_sse_event_instance_str()
             break
         if msg is None:
             continue

src/backend/bisheng/api/services/linsight/sop_manage.py

@@ -1,7 +1,7 @@
 import io
 import json
 import uuid
-from typing import List, Dict
+from typing import List, Dict, Literal
 
 import openpyxl
 from fastapi import UploadFile
@@ -176,6 +176,9 @@ async def _sync_sop_record(sop_records: list[LinsightSOPRecord], override: bool
                         description=one_record.description,
                         content=one_record.content,
                         rating=one_record.rating,
+                        linsight_version_id=one_record.linsight_version_id,
+                        showcase=False,
+                        user_id=one_record.user_id,
                     ))
                     override_name_dict[one.name] = True
             # 再新增剩下的sop记录
@@ -187,6 +190,7 @@ async def _sync_sop_record(sop_records: list[LinsightSOPRecord], override: bool
                     description=one.description,
                     content=one.content,
                     rating=one.rating,
+                    linsight_version_id=one.linsight_version_id,
                 ), one.user_id)
         elif save_new:
             for one in sop_records:
@@ -199,6 +203,7 @@ async def _sync_sop_record(sop_records: list[LinsightSOPRecord], override: bool
                     description=one.description,
                     content=one.content,
                     rating=one.rating,
+                    linsight_version_id=one.linsight_version_id,
                 ), one.user_id)
         else:
             # 说明有重复的记录，需要用户确认
@@ -211,6 +216,7 @@ async def _sync_sop_record(sop_records: list[LinsightSOPRecord], override: bool
                     description=one.description,
                     content=one.content,
                     rating=one.rating,
+                    linsight_version_id=one.linsight_version_id,
                 ), one.user_id)
         if oversize_records:
             raise ValueError(f"{'、'.join(oversize_records)}内容超长")
@@ -285,6 +291,28 @@ async def upload_sop_file(cls, login_user: UserPayload, file: UploadFile, ignore
         records = [LinsightSOPRecord(**one, user_id=login_user.user_id) for one in success_rows]
         return await cls._sync_sop_record(records, override=override, save_new=save_new)
 
+    @classmethod
+    async def get_sop_list(cls, keywords: str = None, sort: Literal["asc", "desc"] = "desc", showcase: bool = False,
+                           page: int = 1, page_size: int = 10) -> dict:
+        """
+        获取SOP列表
+        :param keywords: 关键词
+        :param sort: 排序方式
+        :param page: 页码
+        :param page_size: 每页数量
+        :param showcase: 是否仅展示精选案例的SOP
+        :return: SOP列表和总数
+        """
+        sop_pages = await LinsightSOPDao.get_sop_page(keywords=keywords, showcase=showcase, page=page,
+                                                      page_size=page_size,
+                                                      sort=sort)
+        user_ids = list(set([one["user_id"] for one in sop_pages["items"]]))
+        user_map = UserDao.aget_user_by_ids(user_ids=user_ids)
+        user_map = {one.user_id: one.user_name for one in await user_map}
+        for one in sop_pages["items"]:
+            one["user_name"] = user_map.get(one["user_id"], str(one["user_id"]))
+        return sop_pages
+
     @staticmethod
     async def add_sop(sop_obj: SOPManagementSchema, user_id) -> UnifiedResponseModel | None:
         """
@@ -336,16 +364,20 @@ async def add_sop(sop_obj: SOPManagementSchema, user_id) -> UnifiedResponseModel
         return resp_200(data=sop_model)
 
     @staticmethod
-    async def update_sop(sop_obj: SOPManagementUpdateSchema) -> UnifiedResponseModel | None:
+    async def update_sop(sop_obj: SOPManagementUpdateSchema,
+                         update_version_id: bool = True) -> UnifiedResponseModel | None:
         """
         更新SOP
         :param sop_obj:
+        :param update_version_id: 是否更新版本ID
         :return: 更新后的SOP对象
         """
         # 校验SOP是否存在
         existing_sop = await LinsightSOPDao.get_sops_by_ids([sop_obj.id])
         if not existing_sop:
             return NotFoundError.return_resp()
+        if not update_version_id:
+            sop_obj.linsight_version_id = existing_sop[0].linsight_version_id
 
         if sop_obj.content != existing_sop[0].content:
 
@@ -419,6 +451,18 @@ async def remove_sop(sop_ids: list[int], login_user: UserPayload) -> UnifiedResp
 
         return resp_200(data=True)
 
+    @classmethod
+    async def get_sop_by_id(cls, sop_id: int) -> LinsightSOP | None:
+        """
+        根据ID获取SOP
+        :param sop_id: SOP唯一ID
+        :return: SOP对象
+        """
+        sop_models = await LinsightSOPDao.get_sops_by_ids([sop_id])
+        if not sop_models:
+            return None
+        return sop_models[0]
+
     # sop 库检索
     @classmethod
     async def search_sop(cls, query: str, k: int = 3) -> (List[Document], BaseErrorCode | None):

src/backend/bisheng/api/services/linsight/workbench_impl.py

@@ -317,7 +317,8 @@ async def generate_sop(cls, linsight_session_version_id: str,
                            feedback_content: Optional[str] = None,
                            reexecute: bool = False,
                            login_user: Optional[UserPayload] = None,
-                           knowledge_list: List[KnowledgeRead] = None) -> AsyncGenerator[Dict, None]:
+                           knowledge_list: List[KnowledgeRead] = None,
+                           sop_id: Optional[int] = None) -> AsyncGenerator[Dict, None]:
         """
         生成SOP内容
 
@@ -359,9 +360,14 @@ async def generate_sop(cls, linsight_session_version_id: str,
             if previous_session_version_id:
                 session_version = await LinsightSessionVersionDao.get_by_id(previous_session_version_id)
 
+            example_sop = None
+            if sop_id:
+                sop_db = await SOPManageService.get_sop_by_id(sop_id)
+                example_sop = sop_db.content if sop_db else None
+
             content = ""
             async for res in cls._generate_sop_content(
-                    agent, session_version, feedback_content, history_summary, knowledge_list
+                    agent, session_version, feedback_content, history_summary, knowledge_list, example_sop=example_sop
             ):
                 if isinstance(res, cls.SearchSOPError):
                     yield res.error_class.to_sse_event(event="search_sop_error")
@@ -497,12 +503,15 @@ async def _create_linsight_agent(cls, session_version: LinsightSessionVersion,
     async def _generate_sop_content(cls, agent, session_version: LinsightSessionVersion,
                                     feedback_content: Optional[str],
                                     history_summary: List[str],
-                                    knowledge_list: List[KnowledgeRead] = None) -> AsyncGenerator:
+                                    knowledge_list: List[KnowledgeRead] = None,
+                                    example_sop: str = None) -> AsyncGenerator:
         """生成SOP内容"""
         file_list = await cls.prepare_file_list(session_version)
         knowledge_list = await cls.prepare_knowledge_list(knowledge_list)
-
-        if feedback_content is None:
+        if example_sop:
+            async for res in agent.generate_sop(sop=example_sop, file_list=file_list, knowledge_list=knowledge_list):
+                yield res
+        elif feedback_content is None:
             # 检索SOP模板
             sop_template, search_sop_error = await SOPManageService.search_sop(
                 query=session_version.question, k=3

src/backend/bisheng/api/v1/flows.py

@@ -2,24 +2,24 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
-from fastapi_jwt_auth import AuthJWT
 from sqlmodel import select
 from starlette.responses import StreamingResponse
 
-from bisheng.api.errcode.http_error import UnAuthorizedError, ServerError, NotFoundError
 from bisheng.api.errcode.flow import FlowOnlineEditError, FlowNameExistsError
+from bisheng.api.errcode.http_error import UnAuthorizedError, ServerError, NotFoundError
 from bisheng.api.services.flow import FlowService
 from bisheng.api.services.user_service import UserPayload, get_login_user
-from bisheng.api.utils import build_flow_no_yield, get_L2_param_from_flow, remove_api_keys
+from bisheng.api.utils import build_flow_no_yield, remove_api_keys
 from bisheng.api.v1.schemas import (FlowCompareReq, FlowListRead, FlowVersionCreate, StreamData,
-                                    UnifiedResponseModel, resp_200)
+                                    resp_200)
 from bisheng.database.base import session_getter
-from bisheng.database.models.flow import (Flow, FlowCreate, FlowDao, FlowRead, FlowReadWithStyle, FlowType,
+from bisheng.database.models.flow import (Flow, FlowCreate, FlowDao, FlowRead, FlowType,
                                           FlowUpdate)
 from bisheng.database.models.flow_version import FlowVersionDao
 from bisheng.database.models.role_access import AccessType
 from bisheng.settings import settings
 from bisheng.utils.logger import logger
+from fastapi_jwt_auth import AuthJWT
 
 # build router
 router = APIRouter(prefix='/flows', tags=['Flows'], dependencies=[Depends(get_login_user)])
@@ -238,7 +238,7 @@ async def event_stream(req: FlowCompareReq):
             yield str(StreamData(event='message', data={'type': 'end', 'data': ''}))
         except Exception as e:
             logger.exception('compare flow stream error')
-            yield ServerError(exception=e).to_sse_event_instance()
+            yield ServerError(exception=e).to_sse_event_instance_str()
 
     try:
         return StreamingResponse(event_stream(item), media_type='text/event-stream')