From 44666921016339e839d6984e4bb32db6526d429f Mon Sep 17 00:00:00 2001 From: hexiaonan-800 Date: Tue, 20 Oct 2026 16:55:51 +0800 Subject: [PATCH] fix: event loop closed bug --- apps/application/workflow/common.py | 103 ++++++++++++------ .../workflow/nodes/ai_chat_node/agent.py | 14 +-- .../nodes/ai_chat_node/ai_chat_node.py | 6 +- .../nodes/ai_chat_node/tools/tool/__init__.py | 2 +- .../nodes/ai_chat_node/tools/tool/custom.py | 2 +- .../image_understand_node.py | 5 +- .../workflow/nodes/llm_node/llm_node.py | 14 ++- .../nodes/question_node/question_node.py | 3 +- .../video_understand_node.py | 5 +- 9 files changed, 105 insertions(+), 49 deletions(-) diff --git a/apps/application/workflow/common.py b/apps/application/workflow/common.py index 1fc5da8a578..42d8fbccfd6 100644 --- a/apps/application/workflow/common.py +++ b/apps/application/workflow/common.py @@ -8,8 +8,9 @@ """ import asyncio -import contextlib +import os import threading +from concurrent.futures import CancelledError from enum import Enum from typing import List, Dict @@ -257,44 +258,82 @@ class CancelledException(Exception): pass +class _SharedEventLoop: + """ + 进程内常驻事件循环(后台守护线程)。 + + 模型实例按 model_id 缓存,其 httpx 异步连接池随实例复用;连接绑定在创建它的事件循环上。 + 协程内的阻塞调用会卡住所有工作流,需用 asyncio.to_thread 挪出。 + """ + + _loop = None + _pid = None + _lock = threading.Lock() + + @classmethod + def get(cls): + loop = cls._loop + # gunicorn 等 fork 出的子进程不继承父进程线程,按 pid 判断是否需要重建 + if loop is not None and cls._pid == os.getpid() and loop.is_running(): + return loop + with cls._lock: + if cls._loop is None or cls._pid != os.getpid() or not cls._loop.is_running(): + loop = asyncio.new_event_loop() + ready = threading.Event() + + def run(): + asyncio.set_event_loop(loop) + loop.call_soon(ready.set) + loop.run_forever() + + thread = threading.Thread(target=run, name="maxkb-workflow-event-loop", daemon=True) + loop._thread = thread + thread.start() + ready.wait() + cls._loop, cls._pid = loop, os.getpid() + return cls._loop + + class AsyncRunner: """跨线程取消感知地运行异步协程;cancel() 置位事件使阻塞中的流式中断并抛 CancelledException。""" def __init__(self): self.cancel_event = threading.Event() + self._lock = threading.Lock() + self._loop = None + self._task = None def cancel(self): - """请求取消(任意线程可调);只置旗标,真正中断发生在事件循环内。""" + """请求取消(任意线程可调):置位旗标并取消正在运行的协程。""" self.cancel_event.set() + with self._lock: + loop, task = self._loop, self._task + if loop is not None and task is not None: + loop.call_soon_threadsafe(task.cancel) def run_async(self, async_factory): - """在独立事件循环里运行 async_factory() 的协程;被取消则抛 CancelledException。""" - return asyncio.run(self.await_or_cancel(async_factory())) - - async def _wait_cancel(self): - # 轮询取消旗标,置位即结束,充当竞速对手 - while not self.cancel_event.is_set(): - await asyncio.sleep(0) - - async def await_or_cancel(self, _async): - """竞速等待:流式完成返回其结果;取消先到则抛 CancelledException。""" - - # 真正要跑的流式协程,包成 task 才能与 waiter 一起放进 asyncio.wait - stream = asyncio.create_task(_async) - # 把"是否已取消"伪装成一个可等待的 task,充当 stream 的竞速对手。 - waiter = asyncio.create_task(self._wait_cancel()) - # 同时等两个 task,谁先完成(返回)谁进入 done;FIRST_COMPLETED = 取先到者。 - done, _ = await asyncio.wait({stream, waiter}, return_when=asyncio.FIRST_COMPLETED) - # stream_done 为真 -> 流式自己先跑完;为假 -> 是 waiter 先完成,即取消先到。 - stream_done = stream in done - # 无论谁赢,都把两个 task cancel 并等它们退干净,防止"输掉的那一方" - # 还悬在事件循环里。await 吞掉 CancelledError(取消是本意,不算错误)。 - for task in (stream, waiter): - task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await task - - # 流式正常结束:把它的返回值交给调用方。 - if stream_done: - return stream.result() - raise CancelledException() + """在进程内常驻事件循环中运行 async_factory() 的协程并阻塞等待结果;被取消则抛 CancelledException。""" + if self.cancel_event.is_set(): + raise CancelledException() + loop = _SharedEventLoop.get() + if threading.current_thread() is loop._thread: + # 在常驻循环线程内同步等待自身会死锁;协程内应直接 await + raise RuntimeError("AsyncRunner.run_async cannot be called from the shared event loop thread") + future = asyncio.run_coroutine_threadsafe(self._run(async_factory), loop) + try: + # 取消时任务先在循环内执行完 finally 清理(落盘已累积内容、关闭流),再以 CancelledError 结束 + return future.result() + except CancelledError: + raise CancelledException() + + async def _run(self, async_factory): + with self._lock: + self._loop, self._task = asyncio.get_running_loop(), asyncio.current_task() + try: + # 登记任务前已请求取消:直接结束,不发起模型请求 + if self.cancel_event.is_set(): + raise CancelledException() + return await async_factory() + finally: + with self._lock: + self._loop, self._task = None, None diff --git a/apps/application/workflow/nodes/ai_chat_node/agent.py b/apps/application/workflow/nodes/ai_chat_node/agent.py index f00b09166aa..4a204f79f8a 100644 --- a/apps/application/workflow/nodes/ai_chat_node/agent.py +++ b/apps/application/workflow/nodes/ai_chat_node/agent.py @@ -186,13 +186,11 @@ async def _run(): stream_mode="messages", ) - async def consume(): - # runner 取消该 task 会让 CancelledError 注入在途的工具 await,从而打断正在运行的工具。 - async for chunk in response: - msg = chunk[0] - call_back.on_next(msg) - - await async_runner.await_or_cancel(consume()) + # 取消:run_async 的 task.cancel() 会让 CancelledError 注入在途的 chunk await, + # 从而打断正在运行的 MCP/工具调用。 + async for chunk in response: + msg = chunk[0] + call_back.on_next(msg) def _classify_error(e): # 取消：原样保留（保持节点取消语义）；MCP TaskGroup 的 ExceptionGroup：展开取真实异常并包成 RuntimeError @@ -205,7 +203,7 @@ def _classify_error(e): error = None try: - asyncio.run(_run()) + async_runner.run_async(_run) except Exception as e: error = _classify_error(e) finally: diff --git a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py index fc1adee0c4b..d0ad602d21c 100644 --- a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py +++ b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py @@ -7,6 +7,7 @@ @desc: """ +import asyncio import base64 import contextlib import json @@ -473,7 +474,9 @@ async def _stream_response( # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接 finally: - self._write_final_context(chat_model, message_list, question, answer, reasoning_content) + await asyncio.to_thread( + self._write_final_context, chat_model, message_list, question, answer, reasoning_content + ) with contextlib.suppress(Exception): await response.aclose() @@ -561,7 +564,6 @@ def _handle_mcp( ToolExecutor().validate_mcp_transport(json.dumps(mcp_servers_config)) if tools or mcp_servers_config or skill_tool_ids: - node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.RUNNING) # 使用可变状态在回调间共享（answer 累积、当前文本 content id、工具 content id 映射） state = {"answer": ""} tool_stream = ToolCallStreamManagement() diff --git a/apps/application/workflow/nodes/ai_chat_node/tools/tool/__init__.py b/apps/application/workflow/nodes/ai_chat_node/tools/tool/__init__.py index e44c2f9beff..4dc888b2e52 100644 --- a/apps/application/workflow/nodes/ai_chat_node/tools/tool/__init__.py +++ b/apps/application/workflow/nodes/ai_chat_node/tools/tool/__init__.py @@ -22,5 +22,5 @@ def get_tool_tools(source_type, source_id, tool_ids, workspace_id, cancel_event, if not tool_ids: return [] return get_workflow_tools( - source_type, source_id, tool_ids, workspace_id, workflow_params, cancel_event + source_type, source_id, tool_ids, workspace_id, cancel_event, workflow_params ) + get_custom_tools(source_type, source_id, tool_ids, workspace_id, cancel_event) diff --git a/apps/application/workflow/nodes/ai_chat_node/tools/tool/custom.py b/apps/application/workflow/nodes/ai_chat_node/tools/tool/custom.py index 737f6b36530..025d6d7901d 100644 --- a/apps/application/workflow/nodes/ai_chat_node/tools/tool/custom.py +++ b/apps/application/workflow/nodes/ai_chat_node/tools/tool/custom.py @@ -85,7 +85,7 @@ def _run(): threading.Thread(target=_run, daemon=True).start() if await_complete_or_cancel(done_event, cancel_event, on_cancel): raise CancelledException() - if result["error"]: + if result.get("error"): raise result["error"] return result["value"] diff --git a/apps/application/workflow/nodes/image_understand_node/image_understand_node.py b/apps/application/workflow/nodes/image_understand_node/image_understand_node.py index 2378fb83995..fa388a01a5c 100644 --- a/apps/application/workflow/nodes/image_understand_node/image_understand_node.py +++ b/apps/application/workflow/nodes/image_understand_node/image_understand_node.py @@ -1,4 +1,5 @@ # coding=utf-8 +import asyncio import base64 import contextlib import uuid_utils.compat as uuid @@ -258,7 +259,9 @@ async def _stream_response( # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接 finally: - self._write_final_context(chat_model, message_list, question, answer, reasoning_content) + await asyncio.to_thread( + self._write_final_context, chat_model, message_list, question, answer, reasoning_content + ) with contextlib.suppress(Exception): await response.aclose() diff --git a/apps/application/workflow/nodes/llm_node/llm_node.py b/apps/application/workflow/nodes/llm_node/llm_node.py index abc5a3f2496..a8dbd6f65b5 100644 --- a/apps/application/workflow/nodes/llm_node/llm_node.py +++ b/apps/application/workflow/nodes/llm_node/llm_node.py @@ -7,6 +7,7 @@ @desc: LLM 节点:单次调用大模型(不执行工具),输出 content/reasoning_content/refusal/is_refusal/usage/tool_calls/finish_reason """ +import asyncio import contextlib import json @@ -61,6 +62,9 @@ def is_valid(self, *, raise_exception=False): def _get_default_model_params_setting(model_id): model = QuerySet(Model).filter(id=model_id).first() + if model is None: + # 模型已删除或不可见(如默认模型/引用变量指向了失效的模型) + raise AppApiException(500, _("The model of the node does not exist")) credential = get_model_credential(model.provider, model.model_type, model.model_name) return credential.get_model_params_setting_form(model.model_name).get_default_form_data() @@ -290,8 +294,14 @@ async def _stream_response(self, model, chat_model, message_list): ) ) # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接 - self._write_final_context( - chat_model, message_list, full, reasoning_content, finish_reason, tool_call_accumulator.tool_calls() + await asyncio.to_thread( + self._write_final_context, + chat_model, + message_list, + full, + reasoning_content, + finish_reason, + tool_call_accumulator.tool_calls(), ) with contextlib.suppress(Exception): await response.aclose() diff --git a/apps/application/workflow/nodes/question_node/question_node.py b/apps/application/workflow/nodes/question_node/question_node.py index 7db242b690f..60d58c9f8f1 100644 --- a/apps/application/workflow/nodes/question_node/question_node.py +++ b/apps/application/workflow/nodes/question_node/question_node.py @@ -5,6 +5,7 @@ @desc: """ +import asyncio import re import contextlib from functools import reduce @@ -173,7 +174,7 @@ async def _stream_response(self, response, chat_model, message_list, is_result): ) # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接 finally: - self._write_final_context(chat_model, message_list, answer) + await asyncio.to_thread(self._write_final_context, chat_model, message_list, answer) with contextlib.suppress(Exception): await response.aclose() diff --git a/apps/application/workflow/nodes/video_understand_node/video_understand_node.py b/apps/application/workflow/nodes/video_understand_node/video_understand_node.py index 956af5333c9..8a2a1116d5d 100644 --- a/apps/application/workflow/nodes/video_understand_node/video_understand_node.py +++ b/apps/application/workflow/nodes/video_understand_node/video_understand_node.py @@ -1,4 +1,5 @@ # coding=utf-8 +import asyncio import contextlib import uuid_utils.compat as uuid from functools import reduce @@ -256,7 +257,9 @@ async def _stream_response( # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接 finally: - self._write_final_context(chat_model, message_list, question, answer, reasoning_content) + await asyncio.to_thread( + self._write_final_context, chat_model, message_list, question, answer, reasoning_content + ) with contextlib.suppress(Exception): await response.aclose()