From 942ae7655f1d79bd04dc2e4edbd2f203ecfc83e2 Mon Sep 17 00:00:00 2001 From: Wenruli Date: Wed, 6 Aug 2025 12:04:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E5=AE=9E=E7=8E=B0=E7=81=B5=E6=80=9D?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E4=B8=AD=E9=97=B4=E6=96=87=E4=BB=B6=E5=B1=95?= =?UTF-8?q?=E7=A4=BA=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/backend/bisheng/api/v1/linsight.py | 16 +++- src/backend/bisheng/linsight/task_exec.py | 75 ++++++++++--------- src/backend/bisheng/linsight/utils.py | 63 +++++++++++++++- .../bisheng_langchain/linsight/event.py | 2 + 4 files changed, 120 insertions(+), 36 deletions(-) diff --git a/src/backend/bisheng/api/v1/linsight.py b/src/backend/bisheng/api/v1/linsight.py index 0880520dd..07041de37 100644 --- a/src/backend/bisheng/api/v1/linsight.py +++ b/src/backend/bisheng/api/v1/linsight.py @@ -1,5 +1,6 @@ import json import os +from datetime import datetime from typing import List, Literal, Optional from urllib import parse @@ -26,7 +27,7 @@ from bisheng.database.models.knowledge import KnowledgeTypeEnum, KnowledgeDao from bisheng.database.models.linsight_session_version import LinsightSessionVersionDao, SessionVersionStatusEnum, \ LinsightSessionVersion from bisheng.database.models.linsight_sop import LinsightSOPDao, LinsightSOPRecord -from bisheng.linsight.state_message_manager import LinsightStateMessageManager +from bisheng.linsight.state_message_manager import LinsightStateMessageManager, MessageData, MessageEventType from bisheng.settings import settings router = APIRouter(prefix="/linsight", tags=["灵思"]) @@ -457,6 +458,19 @@ async def terminate_execute( await state_message_manager.set_session_version_info(session_version_model) + state_message_manager = LinsightStateMessageManager(session_version_id=session_version_model.id) + # 推送终止消息 + await state_message_manager.push_message( + MessageData( + event_type=MessageEventType.TASK_TERMINATED, + data={ + "message": "任务已被用户主动停止", + "session_id": session_version_model.id, + "terminated_at": datetime.now().isoformat() + } + ) + ) + return resp_200(data=True, message="灵思执行已终止") diff --git a/src/backend/bisheng/linsight/task_exec.py b/src/backend/bisheng/linsight/task_exec.py index 6946359f5..82988db99 100644 --- a/src/backend/bisheng/linsight/task_exec.py +++ b/src/backend/bisheng/linsight/task_exec.py @@ -54,16 +54,17 @@ class LinsightWorkflowTask: self._is_terminated = False self._termination_task: Optional[asyncio.Task] = None self._final_result: Optional[TaskEnd] = None + self.file_dir: Optional[str] = None + self.session_version_id: Optional[str] = None # ==================== 资源管理 ==================== @asynccontextmanager - async def _managed_execution(self, session_version_id: str): + async def _managed_execution(self): """管理执行资源的上下文管理器""" - file_dir = None - self._state_manager = LinsightStateMessageManager(session_version_id) - session_model = await self._get_session_model(session_version_id) + self._state_manager = LinsightStateMessageManager(self.session_version_id) + session_model = await self._get_session_model(self.session_version_id) # 检查会话状态 if await self._is_session_in_progress(session_model): @@ -75,22 +76,22 @@ class LinsightWorkflowTask: await self._start_termination_monitor(session_model) # 初始化文件目录 - file_dir = await self._init_file_directory(session_model) + self.file_dir = await self._init_file_directory(session_model) - yield session_model, file_dir + yield session_model finally: - await self._cleanup_resources(file_dir) + await self._cleanup_resources() - async def _cleanup_resources(self, file_dir: Optional[str] = None): + async def _cleanup_resources(self): """清理资源""" try: # 停止终止监控 await self._stop_termination_monitor() # 清理文件目录 - if file_dir and os.path.exists(file_dir): - shutil.rmtree(file_dir, ignore_errors=True) + if self.file_dir and os.path.exists(self.file_dir): + shutil.rmtree(self.file_dir, ignore_errors=True) except Exception as e: logger.error(f"资源清理失败: {e}") @@ -100,26 +101,28 @@ class LinsightWorkflowTask: async def async_run(self, session_version_id: str) -> None: """异步任务执行入口""" - with logger.contextualize(trace_id=session_version_id): - logger.info(f"开始执行任务: session_version_id={session_version_id}") + self.session_version_id = session_version_id + + with logger.contextualize(trace_id=self.session_version_id): + logger.info(f"开始执行任务: session_version_id={self.session_version_id}") try: - async with self._managed_execution(session_version_id) as (session_model, file_dir): - await self._execute_workflow(session_model, file_dir) + async with self._managed_execution() as session_model: + await self._execute_workflow(session_model) except UserTerminationError: - logger.info(f"任务被用户主动终止: session_version_id={session_version_id}") + logger.info(f"任务被用户主动终止: session_version_id={self.session_version_id}") except TaskAlreadyInProgressError: - logger.warning(f"任务已在进行中: session_version_id={session_version_id}") + logger.warning(f"任务已在进行中: session_version_id={self.session_version_id}") except TaskExecutionError as e: - logger.exception(f"任务执行失败: session_version_id={session_version_id}") - await self._handle_execution_error(session_version_id, e) + logger.exception(f"任务执行失败: session_version_id={self.session_version_id}") + await self._handle_execution_error(e) except Exception as e: - logger.exception(f"未知错误: session_version_id={session_version_id}, error={e}") - await self._handle_execution_error(session_version_id, e) + logger.exception(f"未知错误: session_version_id={self.session_version_id}, error={e}") + await self._handle_execution_error(e) - async def _execute_workflow(self, session_model: LinsightSessionVersion, file_dir: str): + async def _execute_workflow(self, session_model: LinsightSessionVersion): """执行工作流的核心逻辑""" # 更新会话状态为进行中 @@ -129,10 +132,10 @@ class LinsightWorkflowTask: llm = await self._get_llm() # 生成工具列表 tools = await self._generate_tools(session_model, llm) - linsight_tools = await ToolServices.init_linsight_tools(root_path=file_dir) + linsight_tools = await ToolServices.init_linsight_tools(root_path=self.file_dir) tools.extend(linsight_tools) # 创建智能体 - agent = await self._create_agent(session_model, llm, tools, file_dir) + agent = await self._create_agent(session_model, llm, tools) # 检查是否在初始化过程中被终止 self._check_termination() @@ -145,7 +148,7 @@ class LinsightWorkflowTask: success = await self._execute_agent_tasks(agent, task_info, session_model) if success: - await self._handle_task_completion(session_model, llm, file_dir) + await self._handle_task_completion(session_model, llm) else: await self._handle_user_termination(session_model) raise UserTerminationError("任务被用户终止") @@ -185,6 +188,7 @@ class LinsightWorkflowTask: async def _init_file_directory(self, session_model: LinsightSessionVersion) -> str: """初始化文件目录""" file_dir = os.path.join(CACHE_DIR, "linsight", session_model.id[:8]) + file_dir = os.path.normpath(file_dir) os.makedirs(file_dir, exist_ok=True) if not session_model.files: @@ -236,8 +240,7 @@ class LinsightWorkflowTask: return await LinsightWorkbenchImpl.init_linsight_config_tools(session_version=session_model, llm=llm) - async def _create_agent(self, session_model: LinsightSessionVersion, llm: BishengLLM, tools: List, - file_dir: str) -> LinsightAgent: + async def _create_agent(self, session_model: LinsightSessionVersion, llm: BishengLLM, tools: List) -> LinsightAgent: workbench_conf = await LLMService.get_workbench_llm() linsight_conf = settings.get_linsight_conf() @@ -248,7 +251,7 @@ class LinsightWorkflowTask: llm=llm, query=session_model.question, tools=tools, - file_dir=file_dir, + file_dir=self.file_dir, task_mode=workbench_conf.linsight_executor_mode, exec_config=exec_config, ) @@ -412,6 +415,10 @@ class LinsightWorkflowTask: async def _handle_exec_step(self, agent: LinsightAgent, event: ExecStep, session_model: LinsightSessionVersion): """处理执行步骤事件""" + + # 额外处理步骤事件 + event = await linsight_execute_utils.handle_step_event_extra(event, self) + await self._state_manager.add_execution_task_step(event.task_id, step=event) await self._state_manager.push_message( MessageData(event_type=MessageEventType.TASK_EXECUTE_STEP, data=event.model_dump()) @@ -538,22 +545,22 @@ class LinsightWorkflowTask: # ==================== 任务完成处理 ==================== - async def _handle_task_completion(self, session_model: LinsightSessionVersion, llm: BishengLLM, file_dir: str): + async def _handle_task_completion(self, session_model: LinsightSessionVersion, llm: BishengLLM): """处理任务完成""" if not self._final_result: logger.error("没有找到最终任务结果") return if self._final_result.status == TaskStatus.SUCCESS.value: - await self._handle_task_success(session_model, llm, file_dir) + await self._handle_task_success(session_model, llm) else: await self._handle_task_failure(session_model, "任务执行失败") - async def _handle_task_success(self, session_model: LinsightSessionVersion, llm: BishengLLM, file_dir: str): + async def _handle_task_success(self, session_model: LinsightSessionVersion, llm: BishengLLM): """处理任务成功""" try: # 读取文件目录文件详情 - file_details = await linsight_execute_utils.read_file_directory(file_dir) + file_details = await linsight_execute_utils.read_file_directory(self.file_dir) logger.debug(f"读取文件目录文件详情: {file_details}") final_result_files = await linsight_execute_utils.get_final_result_file( @@ -625,10 +632,10 @@ class LinsightWorkflowTask: if linsight_invitation_code: await InviteCodeService.revoke_invite_code(user_id=session_model.user_id) - async def _handle_execution_error(self, session_version_id: str, error: Exception): + async def _handle_execution_error(self, error: Exception): """处理执行错误""" try: - session_model = await LinsightSessionVersionDao.get_by_id(session_version_id) + session_model = await LinsightSessionVersionDao.get_by_id(self.session_version_id) await self._handle_task_failure(session_model, str(error)) except Exception as e: - logger.error(f"处理执行错误失败: session_version_id={session_version_id}, error={e}") + logger.error(f"处理执行错误失败: session_version_id={self.session_version_id}, error={e}") diff --git a/src/backend/bisheng/linsight/utils.py b/src/backend/bisheng/linsight/utils.py index 0c089660e..38219304a 100644 --- a/src/backend/bisheng/linsight/utils.py +++ b/src/backend/bisheng/linsight/utils.py @@ -7,13 +7,21 @@ from bisheng.database.models import LinsightSessionVersion, LinsightExecuteTask from bisheng.utils import util from bisheng.utils.minio_client import minio_client from bisheng.utils.util import sync_func_to_async +from bisheng_langchain.linsight.event import ExecStep # 灵思文件处理工具对应文件参数名 local_file_tool_dict = { - "write_text_file": "file_path", + "add_text_to_file": "file_path", "replace_file_lines": "file_path" } +# 步骤事件额外处理工具对应参数名、 +step_event_extra_tool_dict = { + "add_text_to_file": "file_path", + "replace_file_lines": "file_path", + "read_text_file": "file_path" +} + # 获取任务中的所有操作过的文件 async def get_all_files_from_session(execution_tasks: List[LinsightExecuteTask], file_details: List[Dict]) -> list[ @@ -191,3 +199,56 @@ async def get_final_result_file(session_model: LinsightSessionVersion, file_deta logger.warning(f"部分文件上传失败: {len(failed_uploads)} 个文件") return final_result_files + + +# 步骤事件额外处理 +async def handle_step_event_extra(event: ExecStep, task_exec_obj) -> ExecStep: + """ + 处理步骤事件的额外逻辑 + :param task_exec_obj: + :param event: 事件对象 + """ + logger.debug(f"步骤事件额外处理，call_id: {event.call_id}, name: {event.name}, status: {event.status}") + try: + if event.status == "end" and event.name in step_event_extra_tool_dict.keys(): + file_path = event.params.get(step_event_extra_tool_dict[event.name], "") + if not file_path: + return event + + file_name = os.path.basename(file_path) + logger.debug(f"步骤事件额外处理，文件名: {file_name}") + + # 文件路径处理 + if not os.path.isabs(file_path): + # 相对路径，转换为绝对路径 + file_path = os.path.join(task_exec_obj.file_dir, file_path) + file_path = os.path.normpath(file_path) + + logger.debug(f"步骤事件额外处理，转换后的文件路径: {file_path}") + + if not os.path.exists(file_path): + logger.error(f"步骤事件额外处理，文件不存在: {file_path}") + return event + + file_md5 = await util.async_calculate_md5(file_path) + + object_name = f"linsight/step_event/{task_exec_obj.session_version_id}/{uuid.uuid4().hex[:8]}.{file_name.split('.')[-1]}" + logger.debug(f"步骤事件额外处理，上传文件到MinIO: {object_name}") + + # 上传文件到MinIO + await sync_func_to_async(minio_client.upload_minio)( + bucket_name=minio_client.bucket, + object_name=object_name, + file_path=file_path + ) + event.extra_info["file_info"] = { + "file_name": file_name, + "file_md5": file_md5, + "file_url": object_name + } + + except Exception as e: + logger.error(f"步骤事件额外处理异常: {e}") + # 发生异常时，返回原始事件，不做任何修改 + + return event diff --git a/src/backend/bisheng_langchain/linsight/event.py b/src/backend/bisheng_langchain/linsight/event.py index 7b26791ba..3e80e60bc 100644 --- a/src/backend/bisheng_langchain/linsight/event.py +++ b/src/backend/bisheng_langchain/linsight/event.py @@ -18,6 +18,8 @@ class ExecStep(BaseEvent): step_type: Optional[str] = Field(default="tool_call", description='步骤类型。tool_call: 工具调用；react_step: 固定步骤或回答;') status: str = Field(..., description='执行状态，start: 开始执行；end：执行结束') + # 额外信息 + extra_info: Optional[dict] = Field(default={}, description='额外信息，包含文件上传等其他信息') # 生成子任务的事件