mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
parent
a558884c3a
commit
c2f31e440e
10
CHANGELOG.md
10
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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
161
backend/tests/blocking_io/test_knowledge_router.py
Normal file
161
backend/tests/blocking_io/test_knowledge_router.py
Normal file
@ -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()
|
||||
Loading…
x
Reference in New Issue
Block a user