From c2f31e440e2196b5c25ea4a111d952c3a360c1ae Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 5 Oct 2026 23:17:09 +0900 Subject: [PATCH] fix(knowledge): load the catalog's agent config off the event loop (#6313) * fix(knowledge): load the catalog's agent config off the event loop Both retrieval-catalog handlers called load_agent_config() directly from async code. The agent store is synchronous: file IO on the file backend, a sync SQLAlchemy round trip on the db backend, so a slow disk or database stalled every other Gateway request. Offload the load with asyncio.to_thread, matching agents.py, scheduled_tasks.py and the run-admission path in services.py. The strict Blockbuster anchors drive both handlers end to end against real file and SQLite agent stores, plus the unknown-agent 404 path. * docs(changelog): reference #6313 for the knowledge catalog fix * Added the Chinese changelog entry back --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 10 ++ CHANGELOG_zh.md | 7 + backend/app/gateway/routers/knowledge.py | 11 +- .../blocking_io/test_knowledge_router.py | 161 ++++++++++++++++++ 4 files changed, 185 insertions(+), 4 deletions(-) create mode 100644 backend/tests/blocking_io/test_knowledge_router.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a7f6567311..2255c102e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -479,6 +479,15 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **gateway:** The knowledge retrieval catalog no longer blocks the Gateway event + loop while it loads a custom agent's config. Both + `GET /api/knowledge/retrieval-catalog/datasets` and + `.../datasets/{id}/documents` read the agent through the sync agent store on + the loop, which means file IO on the `file` backend and a synchronous + SQLAlchemy round trip on the `db` backend, so a slow disk or database stalled + every other request. The load now runs in `asyncio.to_thread`, like the other + Gateway routes that read agent configs; responses and the 404 for an unknown + agent are unchanged. ([#6313]) - **gateway:** Deleting a thread with a large workspace no longer freezes every other Gateway request while its files are removed. `DELETE /api/threads/{id}` ran `shutil.rmtree` over the thread directory on the event loop, so other @@ -7787,5 +7796,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6305]: https://github.com/bytedance/deer-flow/pull/6305 [#6306]: https://github.com/bytedance/deer-flow/pull/6306 [#6307]: https://github.com/bytedance/deer-flow/pull/6307 +[#6313]: https://github.com/bytedance/deer-flow/pull/6313 [#6319]: https://github.com/bytedance/deer-flow/pull/6319 [#6326]: https://github.com/bytedance/deer-flow/pull/6326 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8ebfb00f19..da1ab2a605 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -432,6 +432,12 @@ ### 修复 +- **网关：** 知识检索目录加载自定义 Agent 配置时不再阻塞网关事件循环。 + `GET /api/knowledge/retrieval-catalog/datasets` 与 `.../datasets/{id}/documents` + 此前在事件循环上通过同步 Agent 存储读取 Agent：`file` 后端为文件 IO，`db` 后端 + 为一次同步 SQLAlchemy 往返，磁盘或数据库变慢时会拖住所有其他请求。现在与其他 + 读取 Agent 配置的网关路由一样，在 `asyncio.to_thread` 中加载；响应内容以及未知 + Agent 返回的 404 保持不变。([#6313]) - **网关：** 删除工作区较大的线程时，移除文件期间不再冻结 Gateway 的其他所有请求。 `DELETE /api/threads/{id}` 此前在事件循环上对线程目录执行 `shutil.rmtree`， 因此其他请求与进行中的 SSE 流都要等到整棵目录树删除完毕（本地 SSD 上 20,000 个 @@ -6492,5 +6498,6 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6305]: https://github.com/bytedance/deer-flow/pull/6305 [#6306]: https://github.com/bytedance/deer-flow/pull/6306 [#6307]: https://github.com/bytedance/deer-flow/pull/6307 +[#6313]: https://github.com/bytedance/deer-flow/pull/6313 [#6319]: https://github.com/bytedance/deer-flow/pull/6319 [#6326]: https://github.com/bytedance/deer-flow/pull/6326 diff --git a/backend/app/gateway/routers/knowledge.py b/backend/app/gateway/routers/knowledge.py index bae5ac5ef1..a60ea4998c 100644 --- a/backend/app/gateway/routers/knowledge.py +++ b/backend/app/gateway/routers/knowledge.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import logging from collections.abc import Awaitable from typing import Annotated, Any @@ -62,12 +63,14 @@ async def _catalog_result[Result](operation: Awaitable[Result]) -> Result: raise HTTPException(status_code=502, detail="RAGFlow request failed.") from None -def _scope_catalog(config: AppConfig, agent_name: str): +async def _scope_catalog(config: AppConfig, agent_name: str): knowledge_base = config.knowledge_base agent_config = None if agent_name != "lead_agent": try: - agent_config = load_agent_config( + # The agent store reads files or makes a sync DB round trip. + agent_config = await asyncio.to_thread( + load_agent_config, agent_name, user_id=get_effective_user_id(), ) @@ -118,7 +121,7 @@ async def list_retrieval_catalog_datasets( config: AppConfig = Depends(get_config), ) -> dict[str, Any]: """Return only datasets that the operator permits this agent to retrieve.""" - settings = _scope_catalog(config, agent_name) + settings = await _scope_catalog(config, agent_name) client = _build_retrieval_client(settings) datasets, error = await _catalog_result( resolve_ragflow_datasets(client, settings), @@ -164,7 +167,7 @@ async def list_retrieval_catalog_documents( config: AppConfig = Depends(get_config), ) -> dict[str, Any]: """Return a normalized, read-only document page inside operator scope.""" - settings = _scope_catalog(config, agent_name) + settings = await _scope_catalog(config, agent_name) if settings.datasets is not None and dataset_id not in set(settings.datasets): raise HTTPException(status_code=404, detail="Knowledge base not found.") client = _build_retrieval_client(settings) diff --git a/backend/tests/blocking_io/test_knowledge_router.py b/backend/tests/blocking_io/test_knowledge_router.py new file mode 100644 index 0000000000..6467dc650b --- /dev/null +++ b/backend/tests/blocking_io/test_knowledge_router.py @@ -0,0 +1,161 @@ +"""Regression anchors: the knowledge catalog must not block the event loop. + +Both retrieval-catalog handlers load the custom agent's config to decide +whether it may select a knowledge scope. That read goes through the sync agent +store — file IO on the ``file`` backend, a sync SQLAlchemy round trip on the +``db`` backend — so it is offloaded via ``asyncio.to_thread``. If it regresses +onto the event loop, the strict Blockbuster gate raises ``BlockingError``. + +The handlers are driven end to end (not the scope helper) so the anchors fail +if a call site stops awaiting the offloaded load. ``@require_permission`` is +bypassed via ``__wrapped__``; RAGFlow is replaced with an in-memory fake. +Imports sit at module top so import-time IO runs at collection, outside the +gate. +""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from fastapi import HTTPException +from sqlalchemy import create_engine + +from app.gateway.knowledge_scope_admission import RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER +from app.gateway.routers import knowledge +from deerflow.config.paths import get_paths +from deerflow.persistence.agents.model import AgentRow +from deerflow.persistence.agents.sql import SqlAgentStore +from deerflow.persistence.base import Base +from deerflow.runtime.user_context import get_effective_user_id + +pytestmark = pytest.mark.asyncio + +# The undecorated coroutines (``require_permission`` uses ``functools.wraps``). +_list_datasets = knowledge.list_retrieval_catalog_datasets.__wrapped__ +_list_documents = knowledge.list_retrieval_catalog_documents.__wrapped__ + +_AGENT = "kb-agent" + + +@pytest.fixture(autouse=True) +def _isolate_agent_store_config(tmp_path: Path, monkeypatch) -> None: + monkeypatch.delenv("DEER_FLOW_CONFIG_PATH", raising=False) + monkeypatch.setenv("DEER_FLOW_PROJECT_ROOT", str(tmp_path)) + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr("deerflow.config.app_config._legacy_config_candidates", lambda: ()) + monkeypatch.setattr("deerflow.config.paths._paths", None) + + +def _seed_file_agent() -> None: + agent_dir = get_paths().user_agent_dir(get_effective_user_id(), _AGENT) + agent_dir.mkdir(parents=True, exist_ok=True) + (agent_dir / "config.yaml").write_text(f"name: {_AGENT}\ntool_groups: [knowledge]\n", encoding="utf-8") + + +def _seed_db_agent(tmp_path: Path) -> SqlAgentStore: + url = f"sqlite:///{tmp_path}/agents.db" + engine = create_engine(url) + Base.metadata.create_all(engine, tables=[AgentRow.__table__]) + engine.dispose() + store = SqlAgentStore(url) + store.create(_AGENT, {"name": _AGENT, "tool_groups": ["knowledge"]}, None, user_id=get_effective_user_id()) + return store + + +@pytest.fixture(params=["file", "db"]) +def seeded_agent_store(request, tmp_path: Path, monkeypatch) -> None: + # Sync fixture: seeding runs before the loop starts, so only the handlers' + # own IO is exercised on it. + if request.param == "file": + _seed_file_agent() + return + store = _seed_db_agent(tmp_path) + monkeypatch.setattr("deerflow.persistence.agents.get_agent_store", lambda: store) + + +def _config() -> SimpleNamespace: + tool = SimpleNamespace( + use=RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER, + model_extra={ + "base_url": "http://ragflow.test", + "api_key": "ragflow-secret", + "timeout": 30, + "datasets": ["dataset-1"], + }, + ) + return SimpleNamespace( + knowledge_base=SimpleNamespace(enabled=True, scope_selection_enabled=True), + get_tool_config=lambda name: tool if name == "knowledge_search" else None, + ) + + +def _fake_ragflow(monkeypatch) -> SimpleNamespace: + ragflow = SimpleNamespace( + list_datasets=AsyncMock( + return_value=[ + { + "id": "dataset-1", + "name": "Policies", + "embedding_model": "embed-a", + "chunk_count": 3, + } + ] + ), + list_documents=AsyncMock( + return_value={ + "code": 0, + "data": { + "total": 1, + "docs": [{"id": "doc-1", "name": "Ready.pdf", "run": "DONE", "chunk_count": 2}], + }, + } + ), + ) + monkeypatch.setattr(knowledge, "_build_retrieval_client", lambda settings: ragflow) + return ragflow + + +@pytest.mark.usefixtures("seeded_agent_store") +async def test_catalog_handlers_do_not_block_event_loop(monkeypatch) -> None: + _fake_ragflow(monkeypatch) + + datasets = await _list_datasets( + request=None, + agent_name=_AGENT, + page=1, + page_size=20, + search="", + config=_config(), + ) + documents = await _list_documents( + "dataset-1", + request=None, + agent_name=_AGENT, + page=1, + page_size=20, + search="", + config=_config(), + ) + + assert datasets["items"] == [{"id": "dataset-1", "name": "Policies", "selectable": True}] + assert documents["items"] == [{"id": "doc-1", "name": "Ready.pdf", "selectable": True}] + + +async def test_unknown_agent_is_not_found_without_blocking_event_loop(monkeypatch) -> None: + ragflow = _fake_ragflow(monkeypatch) + + with pytest.raises(HTTPException) as exc_info: + await _list_datasets( + request=None, + agent_name="missing-agent", + page=1, + page_size=20, + search="", + config=_config(), + ) + + assert exc_info.value.status_code == 404 + ragflow.list_datasets.assert_not_awaited()