From aa4e43a2bcc8ffc3838b012c4583f561a0e199be Mon Sep 17 00:00:00 2001 From: ZJPex <144683084+ZJPex@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:09:36 +0800 Subject: [PATCH] fix(ragflow): batch validation for large document selections (#5572) * fix(ragflow): batch validation for large document selections * docs(ragflow): align documentation language with repository conventions * docs(ragflow): preserve spacing before validation heading --------- Co-authored-by: Willem Jiang --- README.md | 2 + README_zh.md | 2 + .../deerflow/community/ragflow/AGENTS.md | 10 ++ .../deerflow/community/ragflow/tools.py | 9 +- backend/tests/test_ragflow_tools.py | 105 ++++++++++++++++++ 5 files changed, 126 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b0667dfe2..8816f8d9b 100644 --- a/README.md +++ b/README.md @@ -1168,6 +1168,8 @@ This release does not add an independent Knowledge item to the workspace sidebar or a DeerFlow knowledge-management page; create, upload, parse, and delete datasets and documents directly in RAGFlow. +Each message can still select up to 1000 documents. When more than 100 documents are selected from a single dataset, DeerFlow validates them in batches of at most 100 while preserving the complete selection. If any batch contains an inaccessible or non-searchable document, retrieval is rejected. + Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. Every HTTP route that starts or enables a future Agent run requires `runs:create`: this includes the stateless `POST /api/runs/stream` and `POST /api/runs/wait` endpoints plus scheduled-task create, update, resume, and manual-trigger mutations. Scheduled-task mutations retain their existing `threads:write` requirement, and the stateless routes separately enforce ownership when the optional thread ID is supplied in the request body. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. Model access follows the same provider: the Gateway `models` list is filtered per principal, `model:use` is enforced on model detail requests and again when the runtime resolves the agent's model, and a denied default model falls back to the first remaining candidate that also passes `model:use`. The built-in RBAC provider supports per-role `tools`, `routes`, `models`, `skills`, and `sandbox` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md). Advanced deployments can also extend the agent runtime itself by declaring `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. Each entry is a `module.path:ClassName` string (zero-argument constructor) or an object `{class, kwargs}` whose `kwargs` are passed to the constructor. `kwargs` values must be JSON types (object, array, string, number, boolean, or null); YAML dates and timestamps are coerced to ISO strings so they match JSON. DeerFlow loads the same configured list into the lead-agent and subagent pipelines after their built-in runtime middlewares and loop/token guards, but before the terminal-response/safety/clarification tail, so enterprise forks can add domain guardrails, tool-call governance, or observability hooks without patching the built-in middleware builders. Missing packages, invalid classes, broken modules, and constructor errors fail loudly at agent creation. Treat `config.yaml` and `extensions_config.json` as trusted operator-controlled files: middleware paths are code execution, just like custom tool, model, sandbox, guardrail, MCP server, and MCP interceptor declarations. Gateway skill/MCP toggle endpoints preserve this field but do not expose an API write path for `extensions.middlewares`. Separate lead-only/subagent-only middleware lists are not supported yet. diff --git a/README_zh.md b/README_zh.md index 981b40c98..6bb065d74 100644 --- a/README_zh.md +++ b/README_zh.md @@ -643,6 +643,8 @@ DeerFlow 可连接租户级 RAGFlow，并通过 `knowledge_search` 按 embedding 使用内置 RAGFlow `knowledge_search` provider 时，可在 `config.yaml` 中设置 `knowledge_base.scope_selection_enabled: true`，为主智能体和自定义智能体聊天开放模式选择器右侧的纯图标“知识库”按钮。图标持续高亮表示知识检索已启用，普通状态表示本轮检索已关闭。用户可选择全部允许知识库、指定知识库/文件或关闭本轮检索。同一个配置开关统一控制两类聊天；关闭时两类输入框都不显示、也不提交知识范围。选择仅保存在当前页面内，刷新或切换对话后恢复“全部”；每条已发送的人类消息保留不可变的范围快照，用于历史回显、重试和恢复。回复待处理的澄清问题或编辑后重新生成时，若提交了当前选择器快照则以该新范围为准，未提交时继承来源轮次已接纳的范围；知识库仍处于“全部可检索文件”时，展开文件区域不会加载目录，切换为“指定文件”后才加载。Gateway 会校验快照、与运维 allowlist 取交集，把仅含执行字段的范围传递给 native/durable 子智能体，并在模型输入和外部 trace 中清除完整范围。`knowledge_base` 是与 provider 无关的能力开关，只控制知识能力和选择器是否启用；RAGFlow 的连接、dataset allowlist 和检索参数（`base_url`、`api_key`、`datasets`、`page_size`、阈值及输出上限）必须配置在 `tools[].name: knowledge_search` 条目中，`knowledge_base` 中的这些字段不会被读取。 +每条消息仍可选择最多 1000 份文档；同一知识库超过 100 份时，DeerFlow 会按每批最多 100 份校验，保留完整选择范围。任何一批文档不可访问或不可检索，都会拒绝本次检索。 + 自定义智能体聊天请求会同时携带该智能体名称作为 `assistant_id` 和 `context.agent_name`，确保 Gateway 的范围校验与运行时加载的是同一个智能体；主智能体聊天使用 `lead_agent`，两者都只有在共享配置启用 RAGFlow provider 时才会提交知识范围。 diff --git a/backend/packages/harness/deerflow/community/ragflow/AGENTS.md b/backend/packages/harness/deerflow/community/ragflow/AGENTS.md index 21ffac0f8..0f6de4199 100644 --- a/backend/packages/harness/deerflow/community/ragflow/AGENTS.md +++ b/backend/packages/harness/deerflow/community/ragflow/AGENTS.md @@ -18,3 +18,13 @@ together, including delegated results and model-request history. Never shorten an excerpt under an existing ID. Drop entries that cannot fit, with an omission notice, while preserving unrelated artifact fields. This honors per-tool and fallback limits without exempting citation-bearing results from the budget. + +## Document validation + +`tools.py` validates each dataset's selected documents in batches of at most +100 IDs. All batches share the `_bounded_gather` concurrency limit of four. +Merge validated IDs in input order to retain the complete retrieval scope; +any batch error, missing document, or non-searchable document rejects the +whole scope. Keep the application-level 1000-document selection limit separate +from the provider's per-request limit. Regression coverage lives in +`backend/tests/test_ragflow_tools.py` under `test_large_document_scope_*`. diff --git a/backend/packages/harness/deerflow/community/ragflow/tools.py b/backend/packages/harness/deerflow/community/ragflow/tools.py index 2836472b2..e8579a58e 100644 --- a/backend/packages/harness/deerflow/community/ragflow/tools.py +++ b/backend/packages/harness/deerflow/community/ragflow/tools.py @@ -28,6 +28,7 @@ logger = logging.getLogger(__name__) _warned: set[str] = set() _RAGFLOW_UUID_PATTERN = re.compile(r"(? None: message = {"type": "tool", "name": "knowledge_search", "artifact": {"knowledge_sources": {"version": 1, "sources": [source]}}} command = _task_result_command(tool_call_id="task-1", status="completed", result="Answer [citation:1](#knowledge-abc)", source_messages=[message]) assert command.update["messages"][0].artifact["knowledge_sources"]["sources"] == [source] + + +@pytest.mark.anyio +@pytest.mark.parametrize("count", [1, 100, 101, 200, 1000]) +async def test_large_document_scope_batches_provider_validation(monkeypatch: pytest.MonkeyPatch, count: int) -> None: + document_ids = [f"doc-{index:04d}" for index in range(count)] + + class StrictClient(FakeRAGFlowClient): + async def list_documents(self, dataset_id: str, *, params: list[tuple[str, str]]) -> dict: + ids = [value for key, value in params if key == "ids"] + assert 1 <= len(ids) <= 100 + assert dict(params)["page"] == "1" + assert int(dict(params)["page_size"]) == len(ids) + return await super().list_documents(dataset_id, params=params) + + fake = StrictClient( + datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Documents")]}, + documents_by_dataset_id={DATASET_ID_1: [{"id": doc_id, "run": "DONE", "chunk_count": 1} for doc_id in document_ids]}, + ) + _install(monkeypatch, fake) + result = await ragflow_tools.knowledge_search( + "anything", + knowledge_scope={"version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_1], "document_filters": [{"dataset_id": DATASET_ID_1, "document_ids": document_ids}]}, + ) + assert result == "No relevant content found." + assert len(fake.document_list_calls) == (count + 99) // 100 + assert [value for _, params in fake.document_list_calls for key, value in params if key == "ids"] == document_ids + assert len(fake.retrieve_calls) == 1 + assert fake.retrieve_calls[0][1]["document_ids"] == document_ids + + +@pytest.mark.anyio +@pytest.mark.parametrize("failure", ["missing", "running", "empty", "connection", "protocol"]) +async def test_large_document_scope_later_batch_fails_closed(monkeypatch: pytest.MonkeyPatch, failure: str) -> None: + document_ids = [f"doc-{index:04d}" for index in range(101)] + + class LaterBatchFailureClient(FakeRAGFlowClient): + async def list_documents(self, dataset_id: str, *, params: list[tuple[str, str]]) -> dict: + ids = [value for key, value in params if key == "ids"] + assert len(ids) <= 100 + if document_ids[-1] in ids: + if failure == "connection": + raise RAGFlowConnectionError("unavailable") + if failure == "protocol": + return {"data": {"docs": None}} + return await super().list_documents(dataset_id, params=params) + + documents = [{"id": doc_id, "run": "DONE", "chunk_count": 1} for doc_id in document_ids] + if failure == "missing": + documents.pop() + elif failure == "running": + documents[-1]["run"] = "RUNNING" + elif failure == "empty": + documents[-1]["chunk_count"] = 0 + fake = LaterBatchFailureClient( + datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Documents")]}, + documents_by_dataset_id={DATASET_ID_1: documents}, + ) + _install(monkeypatch, fake) + result = await ragflow_tools.knowledge_search( + "anything", + knowledge_scope={"version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_1], "document_filters": [{"dataset_id": DATASET_ID_1, "document_ids": document_ids}]}, + ) + if failure in {"missing", "running", "empty"}: + assert "selected knowledge scope is no longer available" in result + elif failure == "connection": + assert "Unable to connect" in result + else: + assert "invalid document list" in result + assert fake.retrieve_calls == [] + + +@pytest.mark.anyio +async def test_large_document_scope_batches_share_global_concurrency_limit() -> None: + filters = [{"dataset_id": f"dataset-{index}", "document_ids": [f"doc-{index}-{number}" for number in range(201)]} for index in range(3)] + release = asyncio.Event() + saturated = asyncio.Event() + active = 0 + peak = 0 + + class BarrierClient(FakeRAGFlowClient): + async def list_documents(self, dataset_id: str, *, params: list[tuple[str, str]]) -> dict: + nonlocal active, peak + ids = [value for key, value in params if key == "ids"] + assert len(ids) <= 100 + active += 1 + peak = max(peak, active) + if active >= 4: + saturated.set() + try: + await release.wait() + return {"data": {"docs": [{"id": doc_id, "run": "DONE", "chunk_count": 1} for doc_id in reversed(ids)]}} + finally: + active -= 1 + + task = asyncio.create_task(ragflow_tools._validate_document_filters(BarrierClient(), filters)) + try: + await asyncio.wait_for(saturated.wait(), timeout=2) + assert peak == 4 + finally: + release.set() + result, error = await task + assert error is None + assert result == {item["dataset_id"]: item["document_ids"] for item in filters} + assert peak == 4