github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-09-09 13:39:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(knowledge): add read-only RAGFlow retrieval (#4955)

Browse Source 
* feat(knowledge): add read-only RAGFlow retrieval

* test(knowledge): cover RAGFlow retrieval contracts

* docs(knowledge): document retrieval-only RAGFlow setup

* refactor(knowledge): move RAGFlow settings to tool config

* fix(ragflow): bind retrieval to configured datasets

* docs(ragflow): record validated response versions

* fix(ragflow): bind retrieval by dataset id

* fix(ragflow): search all datasets by default

* fix(ragflow): retrieve mixed embeddings by group

* docs(ragflow): keep feature details out of agent guides

* docs(ragflow): remove agent guide changes

* docs(ragflow): remove root readme changes

* fix(ragflow): handle unresolved and empty datasets

* fix(ragflow): harden dataset scope and errors
This commit is contained in:
zhangwei139623 2026-08-25 16:57:55 +08:00 committed by GitHub
parent 1c219b6864
commit 431892e16a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 1792 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
backend/docs/CONFIGURATION.md
View File

@ -218,6 +218,67 @@ models:
`PatchedChatMiMo` preserves MiMo's `choices[].message.reasoning_content`, streaming `delta.reasoning_content`, and request-history assistant `reasoning_content` fields. It does not reuse the DeepSeek provider.
### RAGFlow Knowledge Retrieval
RAGFlow integration is disabled by default. It adds one read-only Agent tool,
`knowledge_search`. DeerFlow does not persist a copy of dataset or document
metadata; RAGFlow is the sole source of truth. The configured API key is
tenant-scoped. An optional operator-controlled `datasets` list restricts every
Agent on this deployment to the same dataset-ID allowlist; omitting it searches
all datasets visible to that tenant API key. An explicitly empty `datasets: []`
is rejected rather than being treated as tenant-wide access.
```yaml
tool_groups:
- name: knowledge
tools:
- name: knowledge_search
group: knowledge
use: deerflow.community.ragflow.tools:knowledge_search_tool
base_url: http://localhost:9380
api_key: $RAGFLOW_API_KEY
datasets:
- 0123456789abcdef0123456789abcdef
- fedcba9876543210fedcba9876543210
timeout: 30
page_size: 8
similarity_threshold: 0.2
vector_similarity_weight: 0.3
top_k: 256
max_chars_per_chunk: 800
max_total_chars: 8000
```
The tool is opt-in through the normal `tools:` list. `datasets` is optional but,
when present, must contain at least one ID. If
it contains RAGFlow dataset IDs selected by the deployment operator, DeerFlow
does not validate their existence while loading configuration; on each search
it verifies them with ID-filtered requests. If `datasets` is omitted, each
search paginates through the tenant-visible dataset catalog. Both paths resolve
current names, embedding models, and chunk counts. Empty datasets are ignored;
an empty dataset that has no embedding-model metadata is also skipped with a
server warning. The remaining datasets are grouped by the exact embedding-model
identifier and each group is sent to RAGFlow with a non-empty `dataset_ids`
list. At most four groups are retrieved concurrently. Because raw similarity
scores from different embedding spaces are not globally comparable, DeerFlow
preserves each group's RAGFlow ranking, interleaves equal rank positions, omits
score labels when more than one group is searched, and applies `page_size` as a
single global chunk limit. If any searchable group fails, the whole tool call
fails rather than silently omitting part of the configured scope. A deleted or
inaccessible configured dataset identifies its ordinal entry in
`knowledge_search.datasets` and produces guidance to check `config.yaml`.
Dataset IDs and catalog listing are not exposed to the Agent.
Use an allowlist to narrow the tenant-wide scope; compatible embedding models
are no longer required across selected datasets. `base_url` must not contain
embedded username or password information. For Docker or Kubernetes, it must be
reachable from the Gateway container or Pod; `localhost` refers to that
container or Pod, not the host machine.
This integration is retrieval-only. Dataset creation, uploads, parsing, and
deletion remain in RAGFlow and are not exposed as Agent tools or DeerFlow APIs.
### Tool Groups
Organize tools into logical groups:

0
backend/packages/harness/deerflow/community/ragflow/__init__.py Normal file
View File

187
backend/packages/harness/deerflow/community/ragflow/client.py Normal file
View File

@ -0,0 +1,187 @@
"""Minimal asynchronous client for the RAGFlow APIs DeerFlow consumes."""
from __future__ import annotations
from typing import Any
import httpx
_DATASET_PAGE_SIZE = 100
_MAX_DATASET_PAGES = 100
class RAGFlowError(Exception):
"""Base class for normalized RAGFlow failures."""
class RAGFlowAPIError(RAGFlowError):
"""RAGFlow returned a valid response envelope with a non-zero code."""
def __init__(self, message: str, *, code: object = None) -> None:
self.code = code
super().__init__(message)
class RAGFlowConnectionError(RAGFlowError):
"""RAGFlow could not be reached or timed out."""
class RAGFlowProtocolError(RAGFlowError):
"""RAGFlow returned an invalid or unexpected HTTP response."""
class RAGFlowClient:
"""Direct HTTP client for DeerFlow's read-only retrieval tools.
The client deliberately owns no cache or persistent state. A fresh HTTP
session is opened for each method call so callers do not need to manage a
client lifecycle.
"""
def __init__(
self,
*,
base_url: str,
api_key: str,
timeout: float = 30,
transport: httpx.AsyncBaseTransport | None = None,
) -> None:
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self._api_key = api_key
self._transport = transport
def _redact(self, value: object) -> str:
text = str(value)
if self._api_key:
text = text.replace(self._api_key, "[REDACTED]")
return text
async def _request(
self,
method: str,
path: str,
*,
params: dict[str, object] | list[tuple[str, str]] | None = None,
json: dict[str, Any] | None = None,
) -> dict[str, Any]:
request_headers = {
"Authorization": f"Bearer {self._api_key}",
"Accept": "application/json",
}
client_kwargs: dict[str, Any] = {
"base_url": f"{self.base_url}/api/v1",
"headers": request_headers,
"timeout": self.timeout,
}
if self._transport is not None:
client_kwargs["transport"] = self._transport
try:
async with httpx.AsyncClient(**client_kwargs) as client:
response = await client.request(method, path, params=params, json=json)
except httpx.TimeoutException:
raise RAGFlowConnectionError(f"RAGFlow request timed out after {self.timeout:g} seconds.") from None
except httpx.RequestError as exc:
detail = self._redact(exc)
raise RAGFlowConnectionError(f"{type(exc).__name__}: {detail}") from None
if response.is_error:
try:
error_payload = response.json()
except ValueError:
error_payload = None
if isinstance(error_payload, dict) and error_payload.get("code") not in (None, 0):
message = self._redact(error_payload.get("message") or f"RAGFlow API error (HTTP {response.status_code})")
raise RAGFlowAPIError(message, code=error_payload.get("code"))
raise RAGFlowProtocolError(f"RAGFlow request failed (HTTP {response.status_code}).")
try:
payload = response.json()
except ValueError:
raise RAGFlowProtocolError("RAGFlow returned invalid JSON.") from None
if not isinstance(payload, dict):
raise RAGFlowProtocolError("RAGFlow returned a non-object JSON payload.")
code = payload.get("code")
if code != 0:
message = self._redact(payload.get("message") or "RAGFlow request failed.")
raise RAGFlowAPIError(message, code=code)
return payload
async def list_datasets(self, *, dataset_id: str | None = None) -> list[dict[str, Any]]:
"""Resolve one dataset ID, or enumerate every page when no ID is given."""
if dataset_id is not None:
dataset_id = dataset_id.strip()
if not dataset_id:
raise ValueError("dataset_id must not be empty")
# RAGFlow's singular `id` filter returns the generic DATA_ERROR code
# for an inaccessible or missing dataset, which is indistinguishable
# from several provider failures. Its `ids` filter instead returns a
# successful empty list for an inaccessible ID, allowing callers to
# classify only that result as a missing binding while preserving all
# real API errors.
payload = await self._request("GET", "/datasets", params={"ids": dataset_id})
data = payload.get("data")
if not isinstance(data, list):
raise RAGFlowProtocolError("RAGFlow returned an invalid dataset list.")
return [item for item in data if isinstance(item, dict)]
datasets: list[dict[str, Any]] = []
received_count = 0
for page in range(1, _MAX_DATASET_PAGES + 1):
payload = await self._request(
"GET",
"/datasets",
params={"page": page, "page_size": _DATASET_PAGE_SIZE},
)
data = payload.get("data")
if not isinstance(data, list):
raise RAGFlowProtocolError("RAGFlow returned an invalid dataset list.")
datasets.extend(item for item in data if isinstance(item, dict))
received_count += len(data)
total = payload.get("total")
if not (isinstance(total, int) and not isinstance(total, bool) and total >= 0):
total = payload.get("total_datasets")
has_valid_total = isinstance(total, int) and not isinstance(total, bool) and total >= 0
if has_valid_total:
if received_count >= total:
return datasets
if not data:
raise RAGFlowProtocolError("RAGFlow dataset listing ended before the reported total.")
elif len(data) < _DATASET_PAGE_SIZE:
return datasets
raise RAGFlowProtocolError(f"RAGFlow dataset listing exceeded {_MAX_DATASET_PAGES} pages.")
async def retrieve(
self,
query: str,
*,
dataset_ids: list[str],
page_size: int = 8,
similarity_threshold: float = 0.2,
vector_similarity_weight: float = 0.3,
top_k: int = 256,
) -> dict[str, Any]:
"""Retrieve chunks from an explicit, non-empty dataset allowlist."""
if not dataset_ids or not all(isinstance(dataset_id, str) and dataset_id.strip() for dataset_id in dataset_ids):
raise ValueError("dataset_ids must contain at least one dataset ID")
request_body: dict[str, object] = {
"question": query,
"dataset_ids": dataset_ids,
"page_size": page_size,
"similarity_threshold": similarity_threshold,
"vector_similarity_weight": vector_similarity_weight,
"top_k": top_k,
}
payload = await self._request("POST", "/retrieval", json=request_body)
data = payload.get("data")
if not isinstance(data, dict):
raise RAGFlowProtocolError("RAGFlow returned an invalid retrieval result.")
return data

96
backend/packages/harness/deerflow/community/ragflow/formatting.py Normal file
View File

@ -0,0 +1,96 @@
"""Compact, citation-friendly formatting for RAGFlow retrieval results."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
def _truncate(value: str, max_chars: int, *, marker: str = "…") -> str:
if len(value) <= max_chars:
return value
if max_chars <= len(marker):
return marker[:max_chars]
return f"{value[: max_chars - len(marker)].rstrip()}{marker}"
def _document_aggregates(value: object) -> list[Mapping[str, Any]]:
if isinstance(value, list):
return [item for item in value if isinstance(item, Mapping)]
if isinstance(value, Mapping):
return [item for item in value.values() if isinstance(item, Mapping)]
return []
def _score(value: object) -> float | None:
if isinstance(value, bool):
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def format_retrieval_result(
result: Mapping[str, Any],
*,
dataset_names_by_id: Mapping[str, str],
max_chars_per_chunk: int = 800,
max_total_chars: int = 8000,
) -> str:
"""Format one RAGFlow retrieval response into compact cited text.
Verified against RAGFlow v0.26.4 and v0.27.0: the REST retrieval endpoint
normalizes response chunk fields before returning them (for example,
``kb_id`` becomes ``dataset_id``). Only those public response field names
are consumed, and dataset IDs are mapped back to the operator-configured
names before anything reaches the model.
"""
raw_chunks = result.get("chunks")
if not isinstance(raw_chunks, list):
raw_chunks = []
chunks = [chunk for chunk in raw_chunks if isinstance(chunk, Mapping)]
if not chunks:
return "No relevant content found."
aggregates = _document_aggregates(result.get("doc_aggs"))
document_names_by_id = {str(item["doc_id"]): str(item["doc_name"]) for item in aggregates if item.get("doc_id") and item.get("doc_name")}
entries: list[str] = []
for index, chunk in enumerate(chunks, start=1):
dataset_id = chunk.get("dataset_id")
dataset_name = dataset_names_by_id.get(str(dataset_id), "Unknown dataset")
document_id = chunk.get("document_id")
document_name = chunk.get("document_keyword")
if not document_name and document_id:
document_name = document_names_by_id.get(str(document_id))
document_name = str(document_name or "Unknown document")
similarity = _score(chunk.get("similarity"))
score_suffix = f" (score {similarity:.2f})" if similarity is not None else ""
content = str(chunk.get("content") or "").strip()
content = _truncate(content, max_chars_per_chunk)
entries.append(f"[{index}] {dataset_name} / {document_name}{score_suffix}\n{content}")
if aggregates:
summaries: list[str] = []
for item in aggregates:
name = item.get("doc_name")
if not name:
continue
count = item.get("count")
count_text = str(count) if isinstance(count, int) and not isinstance(count, bool) else "?"
unit = "chunk" if count == 1 else "chunks"
summaries.append(f"{name} ({count_text} {unit})")
if summaries:
entries.append(f"Matched documents: {', '.join(summaries)}")
formatted = "\n\n".join(entries)
truncation_marker = "… (response truncated)"
if len(formatted) <= max_total_chars:
return formatted
if max_total_chars <= len(truncation_marker):
return truncation_marker[:max_total_chars]
prefix_length = max_total_chars - len(truncation_marker)
return f"{formatted[:prefix_length].rstrip()}{truncation_marker}"

397
backend/packages/harness/deerflow/community/ragflow/tools.py Normal file
View File

@ -0,0 +1,397 @@
"""Read-only Agent tool for operator-scoped RAGFlow knowledge retrieval."""
from __future__ import annotations
import asyncio
import logging
import re
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from langchain_core.tools import StructuredTool
from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, SecretStr, ValidationError, field_validator
from deerflow.config import get_app_config
from .client import RAGFlowAPIError, RAGFlowClient, RAGFlowConnectionError, RAGFlowProtocolError
from .formatting import format_retrieval_result
logger = logging.getLogger(__name__)
_warned: set[str] = set()
_RAGFLOW_UUID_PATTERN = re.compile(r"(?<![0-9A-Fa-f])(?:[0-9A-Fa-f]{32}|[0-9A-Fa-f]{8}(?:-[0-9A-Fa-f]{4}){3}-[0-9A-Fa-f]{12})(?![0-9A-Fa-f])")
_MAX_PARALLEL_RETRIEVAL_GROUPS = 4
_NO_RELEVANT_CONTENT = "No relevant content found."
@dataclass(frozen=True, slots=True)
class _ResolvedDataset:
dataset_id: str
name: str
embedding_model: str
chunk_count: int | None
class _RAGFlowRetrievalSettings(BaseModel):
"""Validated provider settings stored on the knowledge_search tool entry."""
model_config = ConfigDict(validate_default=True)
datasets: list[str] | None = Field(default=None, max_length=100)
base_url: AnyHttpUrl = Field(default="http://localhost:9380")
api_key: SecretStr | None = Field(default=None)
timeout: float = Field(default=30, gt=0, le=600)
page_size: int = Field(default=8, ge=1, le=100)
similarity_threshold: float = Field(default=0.2, ge=0, le=1)
vector_similarity_weight: float = Field(default=0.3, ge=0, le=1)
top_k: int = Field(default=256, ge=1, le=1024)
max_chars_per_chunk: int = Field(default=800, ge=1, le=100_000)
max_total_chars: int = Field(default=8000, ge=1, le=1_000_000)
@field_validator("datasets")
@classmethod
def _normalize_dataset_ids(cls, value: list[str] | None) -> list[str] | None:
if value is None:
return None
if not value:
raise ValueError("datasets must not be empty when configured; omit it to search all accessible datasets")
normalized: list[str] = []
seen: set[str] = set()
for dataset_id in value:
clean_id = dataset_id.strip()
if not clean_id or len(clean_id) > 256:
raise ValueError("dataset IDs must contain between 1 and 256 characters")
if clean_id not in seen:
normalized.append(clean_id)
seen.add(clean_id)
return normalized
@field_validator("base_url")
@classmethod
def _reject_url_userinfo(cls, value: AnyHttpUrl) -> AnyHttpUrl:
if value.username is not None or value.password is not None:
raise ValueError("base_url must not contain username or password information")
return value
def _api_key(settings: _RAGFlowRetrievalSettings) -> str | None:
value = settings.api_key
if isinstance(value, SecretStr):
value = value.get_secret_value()
if isinstance(value, str) and value.strip():
return value.strip()
return None
def _redact_api_key(value: object, api_key: str | None) -> str:
text = str(value)
if api_key:
text = text.replace(api_key, "[REDACTED]")
return text
def _redact_error(value: object, api_key: str | None) -> str:
"""Redact provider credentials and opaque dataset IDs on error paths."""
return _RAGFLOW_UUID_PATTERN.sub("[DATASET_ID]", _redact_api_key(value, api_key))
def _settings_from_extra(extra: Mapping[str, object]) -> _RAGFlowRetrievalSettings:
return _RAGFlowRetrievalSettings.model_validate(dict(extra))
def _settings_or_error() -> tuple[_RAGFlowRetrievalSettings | None, str | None]:
tool_config = get_app_config().get_tool_config("knowledge_search")
if tool_config is None:
return None, "Error: knowledge_search is not configured; add its RAGFlow settings to the tools list in config.yaml."
try:
settings = _settings_from_extra(tool_config.model_extra or {})
except ValidationError:
logger.warning("RAGFlow knowledge_search tool configuration is invalid")
return None, "Error: Invalid RAGFlow settings for knowledge_search; check config.yaml."
if not _api_key(settings):
if "api_key" not in _warned:
_warned.add("api_key")
logger.warning("RAGFlow API key is not configured; set knowledge_search.api_key in config.yaml, preferably via $RAGFLOW_API_KEY.")
return None, "Error: RAGFlow API key is not configured; set knowledge_search.api_key in config.yaml (prefer $RAGFLOW_API_KEY)."
return settings, None
def _build_client(settings: _RAGFlowRetrievalSettings) -> RAGFlowClient:
api_key = _api_key(settings)
if api_key is None: # Guarded by _settings_or_error; keeps this helper total.
raise ValueError("RAGFlow API key is missing")
return RAGFlowClient(
base_url=str(settings.base_url).rstrip("/"),
api_key=api_key,
timeout=settings.timeout,
)
def _tool_error(exc: Exception, settings: _RAGFlowRetrievalSettings) -> str:
key = _api_key(settings)
safe_detail = _redact_error(exc, key)
base_url = _redact_error(str(settings.base_url).rstrip("/"), key)
if isinstance(exc, RAGFlowAPIError):
logger.warning("RAGFlow API rejected a read-only tool request (code=%s)", exc.code)
return f"Error: {safe_detail}"
if isinstance(exc, RAGFlowConnectionError):
logger.warning("RAGFlow connection failed for %s (%s)", base_url, type(exc).__name__)
return f"Error: Unable to connect to RAGFlow ({base_url}): {safe_detail}"
if isinstance(exc, RAGFlowProtocolError):
logger.warning("RAGFlow returned an invalid response for a read-only tool request (%s)", type(exc).__name__)
return f"Error: RAGFlow request failed: {safe_detail}"
logger.warning("Unexpected RAGFlow read-only tool failure (%s)", type(exc).__name__)
return "Error: An unexpected RAGFlow retrieval error occurred; try again later."
def _resolved_dataset(dataset: Mapping[str, object], *, expected_id: str | None = None) -> _ResolvedDataset | None:
dataset_id = dataset.get("id")
if not isinstance(dataset_id, str) or not dataset_id.strip():
return None
clean_id = dataset_id.strip()
if expected_id is not None and clean_id != expected_id:
return None
name = dataset.get("name")
raw_chunk_count = dataset.get("chunk_count")
chunk_count = raw_chunk_count if isinstance(raw_chunk_count, int) and not isinstance(raw_chunk_count, bool) and raw_chunk_count >= 0 else None
embedding_model = dataset.get("embedding_model")
if not isinstance(embedding_model, str) or not embedding_model.strip():
if chunk_count == 0:
warning_key = f"empty_embedding:{clean_id}"
if warning_key not in _warned:
_warned.add(warning_key)
logger.warning("Skipping empty RAGFlow dataset without embedding model metadata (dataset_id=%s)", clean_id)
embedding_model = ""
else:
raise RAGFlowProtocolError("RAGFlow returned a searchable dataset without embedding model metadata.")
return _ResolvedDataset(
dataset_id=clean_id,
name=str(name).strip() if name else "Unknown dataset",
embedding_model=embedding_model.strip(),
chunk_count=chunk_count,
)
def _current_dataset(datasets: list[dict], bound_id: str) -> _ResolvedDataset | None:
for dataset in datasets:
resolved = _resolved_dataset(dataset, expected_id=bound_id)
if resolved is not None:
return resolved
return None
def _ordinal(value: int) -> str:
if 10 <= value % 100 <= 20:
suffix = "th"
else:
suffix = {1: "st", 2: "nd", 3: "rd"}.get(value % 10, "th")
return f"{value}{suffix}"
def _missing_dataset_error(position: int) -> str:
return f"Error: The {_ordinal(position)} entry of knowledge_search.datasets was not found or is inaccessible; check config.yaml."
def _log_missing_dataset(*, position: int, dataset_id: str, code: object = None) -> None:
logger.warning(
"Configured RAGFlow dataset binding could not be resolved (position=%d, dataset_id=%s, code=%s)",
position,
dataset_id,
code,
)
async def _resolve_datasets(
client: RAGFlowClient,
settings: _RAGFlowRetrievalSettings,
) -> tuple[list[_ResolvedDataset] | None, str | None]:
if settings.datasets is None:
datasets = await client.list_datasets()
resolved_by_id: dict[str, _ResolvedDataset] = {}
for dataset in datasets:
resolved = _resolved_dataset(dataset)
if resolved is None:
continue
resolved_by_id.setdefault(resolved.dataset_id, resolved)
if not resolved_by_id:
return (
None,
"Error: No accessible RAGFlow datasets were found; configure knowledge_search.datasets or add a dataset in RAGFlow.",
)
return list(resolved_by_id.values()), None
resolved_datasets: list[_ResolvedDataset] = []
for position, bound_id in enumerate(settings.datasets, start=1):
datasets = await client.list_datasets(dataset_id=bound_id)
resolved = _current_dataset(datasets, bound_id)
if resolved is None:
_log_missing_dataset(position=position, dataset_id=bound_id)
return None, _missing_dataset_error(position)
resolved_datasets.append(resolved)
return resolved_datasets, None
def _group_searchable_datasets(datasets: list[_ResolvedDataset]) -> list[tuple[str, list[str]]]:
groups: dict[str, list[str]] = {}
for dataset in datasets:
if dataset.chunk_count == 0:
continue
groups.setdefault(dataset.embedding_model, []).append(dataset.dataset_id)
return sorted(groups.items())
def _result_chunks(result: Mapping[str, Any]) -> list[Mapping[str, Any]]:
chunks = result.get("chunks")
if not isinstance(chunks, list):
return []
return [chunk for chunk in chunks if isinstance(chunk, Mapping)]
def _result_document_aggregates(result: Mapping[str, Any]) -> list[Mapping[str, Any]]:
aggregates = result.get("doc_aggs")
if isinstance(aggregates, list):
return [aggregate for aggregate in aggregates if isinstance(aggregate, Mapping)]
if isinstance(aggregates, Mapping):
return [aggregate for aggregate in aggregates.values() if isinstance(aggregate, Mapping)]
return []
def _merge_group_results(results: list[dict[str, Any]], *, page_size: int) -> dict[str, Any]:
chunk_groups = [_result_chunks(result) for result in results]
merged_chunks: list[Mapping[str, Any]] = []
max_group_size = max((len(chunks) for chunks in chunk_groups), default=0)
hide_cross_group_scores = len(results) > 1
# Similarity scores from different embedding spaces are not globally
# calibrated. Preserve each provider-ranked list and interleave equal rank
# positions instead of comparing raw scores across models.
for rank in range(max_group_size):
for chunks in chunk_groups:
if rank < len(chunks):
chunk = chunks[rank]
if hide_cross_group_scores and "similarity" in chunk:
chunk = {key: value for key, value in chunk.items() if key != "similarity"}
merged_chunks.append(chunk)
if len(merged_chunks) >= page_size:
break
if len(merged_chunks) >= page_size:
break
selected_document_ids: list[str] = []
for chunk in merged_chunks:
document_id = chunk.get("document_id")
if document_id is not None:
clean_id = str(document_id)
if clean_id not in selected_document_ids:
selected_document_ids.append(clean_id)
aggregates_by_document_id: dict[str, Mapping[str, Any]] = {}
for result in results:
for aggregate in _result_document_aggregates(result):
document_id = aggregate.get("doc_id")
if document_id is not None:
aggregates_by_document_id.setdefault(str(document_id), aggregate)
total = 0
for result in results:
value = result.get("total")
if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
total += value
return {
"chunks": merged_chunks,
"doc_aggs": [aggregates_by_document_id[document_id] for document_id in selected_document_ids if document_id in aggregates_by_document_id],
"total": total,
}
async def _retrieve_dataset_groups(
client: RAGFlowClient,
settings: _RAGFlowRetrievalSettings,
query: str,
groups: list[tuple[str, list[str]]],
) -> dict[str, Any]:
semaphore = asyncio.Semaphore(_MAX_PARALLEL_RETRIEVAL_GROUPS)
async def retrieve_group(dataset_ids: list[str]) -> dict[str, Any]:
async with semaphore:
return await client.retrieve(
query,
dataset_ids=dataset_ids,
page_size=settings.page_size,
similarity_threshold=settings.similarity_threshold,
vector_similarity_weight=settings.vector_similarity_weight,
top_k=settings.top_k,
)
results = await asyncio.gather(*(retrieve_group(dataset_ids) for _, dataset_ids in groups), return_exceptions=True)
successful_results: list[dict[str, Any]] = []
for result in results:
if isinstance(result, BaseException):
raise result
successful_results.append(result)
return _merge_group_results(successful_results, page_size=settings.page_size)
async def knowledge_search(query: str) -> str:
"""Search the configured RAGFlow scope, defaulting to every accessible dataset."""
query = query.strip()
if not query:
return "Error: query must not be empty."
settings, error = _settings_or_error()
if settings is None:
return error or "Error: Invalid RAGFlow settings for knowledge_search; check config.yaml."
client = _build_client(settings)
try:
datasets, resolution_error = await _resolve_datasets(client, settings)
if resolution_error is not None:
return resolution_error
if not datasets: # Defensive; both resolution paths return a non-empty scope.
return "Error: No RAGFlow datasets could be resolved; check knowledge_search in config.yaml."
groups = _group_searchable_datasets(datasets)
if not groups:
return _NO_RELEVANT_CONTENT
result = await _retrieve_dataset_groups(client, settings, query, groups)
names_by_id = {dataset.dataset_id: dataset.name for dataset in datasets}
formatted = format_retrieval_result(
result,
dataset_names_by_id=names_by_id,
max_chars_per_chunk=settings.max_chars_per_chunk,
max_total_chars=settings.max_total_chars,
)
# API-key redaction remains mandatory on success. UUID redaction is
# deliberately error-only so valid checksums and trace IDs survive.
return _redact_api_key(formatted, _api_key(settings))
except Exception as exc:
return _tool_error(exc, settings)
def _tool_description() -> str:
base = "Search the operator-approved RAGFlow datasets and return compact, citation-numbered source chunks."
return f"{base} If knowledge_search.datasets is omitted, all datasets accessible to the configured RAGFlow API key are searched. Dataset IDs are never shown to the model."
async def _knowledge_search_entrypoint(query: str) -> str:
"""Search the configured RAGFlow datasets, or every accessible dataset by default.
Args:
query: Specific question or search terms to retrieve from the configured private documents.
"""
return await knowledge_search(query)
knowledge_search_tool = StructuredTool.from_function(
coroutine=_knowledge_search_entrypoint,
name="knowledge_search",
description=_tool_description(),
parse_docstring=True,
)

257
backend/tests/test_ragflow_client.py Normal file
View File

@ -0,0 +1,257 @@
import json
import httpx
import pytest
from deerflow.community.ragflow.client import (
RAGFlowAPIError,
RAGFlowClient,
RAGFlowConnectionError,
RAGFlowProtocolError,
)
@pytest.mark.anyio
async def test_list_datasets_filters_by_bound_id_in_one_request() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
assert request.method == "GET"
assert request.url == httpx.URL("http://ragflow.test/api/v1/datasets?ids=dataset-1")
assert request.headers["Authorization"] == "Bearer ragflow-secret"
return httpx.Response(
200,
json={
"code": 0,
"data": [{"id": "dataset-1", "name": "HR Policies"}],
"total": 1,
},
)
client = RAGFlowClient(
base_url="http://ragflow.test/",
api_key="ragflow-secret",
timeout=12,
transport=httpx.MockTransport(handler),
)
assert await client.list_datasets(dataset_id="dataset-1") == [{"id": "dataset-1", "name": "HR Policies"}]
assert len(requests) == 1
@pytest.mark.anyio
async def test_list_datasets_without_id_fetches_every_page() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
page = int(request.url.params["page"])
assert request.url.params["page_size"] == "100"
if page == 1:
data = [{"id": f"dataset-{index}", "name": f"Dataset {index}"} for index in range(100)]
elif page == 2:
data = [{"id": "dataset-100", "name": "Dataset 100"}]
else:
pytest.fail(f"unexpected page {page}")
return httpx.Response(200, json={"code": 0, "data": data, "total": 101})
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
datasets = await client.list_datasets()
assert len(datasets) == 101
assert [request.url.params["page"] for request in requests] == ["1", "2"]
@pytest.mark.anyio
async def test_list_datasets_without_id_has_a_hard_page_cap() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
data = [{"id": f"dataset-{index}", "name": f"Dataset {index}"} for index in range(100)]
return httpx.Response(200, json={"code": 0, "data": data})
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(RAGFlowProtocolError, match="exceeded 100 pages"):
await client.list_datasets()
assert len(requests) == 100
@pytest.mark.anyio
async def test_list_datasets_accepts_reported_total_at_the_page_cap() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
page = int(request.url.params["page"])
start = (page - 1) * 100
data = [{"id": f"dataset-{index}", "name": f"Dataset {index}"} for index in range(start, start + 100)]
return httpx.Response(200, json={"code": 0, "data": data, "total": 10_000})
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
datasets = await client.list_datasets()
assert len(datasets) == 10_000
assert len(requests) == 100
@pytest.mark.anyio
async def test_retrieve_always_sends_nonempty_dataset_ids() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
assert request.method == "POST"
assert request.url == httpx.URL("http://ragflow.test/api/v1/retrieval")
assert json.loads(request.content) == {
"question": "annual leave",
"dataset_ids": ["dataset-1"],
"page_size": 8,
"similarity_threshold": 0.2,
"vector_similarity_weight": 0.3,
"top_k": 256,
}
return httpx.Response(200, json={"code": 0, "data": {"chunks": [], "doc_aggs": [], "total": 0}})
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
result = await client.retrieve(
"annual leave",
dataset_ids=["dataset-1"],
page_size=8,
similarity_threshold=0.2,
vector_similarity_weight=0.3,
top_k=256,
)
assert result["total"] == 0
@pytest.mark.anyio
async def test_retrieve_rejects_empty_dataset_ids_before_request() -> None:
called = False
async def handler(request: httpx.Request) -> httpx.Response:
nonlocal called
called = True
return httpx.Response(500)
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(ValueError, match="dataset_ids must contain at least one dataset ID"):
await client.retrieve("fallback search", dataset_ids=[])
assert called is False
@pytest.mark.anyio
async def test_nonzero_api_code_is_normalized_and_redacts_api_key() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={"code": 102, "message": "invalid credential ragflow-secret"},
)
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(RAGFlowAPIError) as exc_info:
await client.list_datasets(dataset_id="dataset-1")
assert exc_info.value.code == 102
assert "invalid credential" in str(exc_info.value)
assert "ragflow-secret" not in str(exc_info.value)
assert "[REDACTED]" in str(exc_info.value)
@pytest.mark.anyio
async def test_timeout_is_english_and_does_not_leak_api_key(caplog: pytest.LogCaptureFixture) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
raise httpx.ReadTimeout("timed out with ragflow-secret", request=request)
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
timeout=2,
transport=httpx.MockTransport(handler),
)
with pytest.raises(RAGFlowConnectionError) as exc_info:
await client.list_datasets(dataset_id="dataset-1")
assert str(exc_info.value) == "RAGFlow request timed out after 2 seconds."
assert "ragflow-secret" not in str(exc_info.value)
assert "ragflow-secret" not in caplog.text
@pytest.mark.anyio
async def test_http_error_body_cannot_echo_api_key() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(401, text="unauthorized: ragflow-secret")
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(RAGFlowProtocolError) as exc_info:
await client.list_datasets(dataset_id="dataset-1")
assert str(exc_info.value) == "RAGFlow request failed (HTTP 401)."
assert "ragflow-secret" not in str(exc_info.value)
@pytest.mark.anyio
async def test_invalid_json_response_is_normalized_in_english() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, text="not-json")
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(RAGFlowProtocolError, match="RAGFlow returned invalid JSON"):
await client.list_datasets(dataset_id="dataset-1")
@pytest.mark.anyio
async def test_list_datasets_rejects_unexpected_data_shape_in_english() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"code": 0, "data": {"id": "not-a-list"}})
client = RAGFlowClient(
base_url="http://ragflow.test",
api_key="ragflow-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(RAGFlowProtocolError, match="invalid dataset list"):
await client.list_datasets(dataset_id="dataset-1")

747
backend/tests/test_ragflow_tools.py Normal file
View File

@ -0,0 +1,747 @@
import asyncio
import logging
from collections.abc import Mapping
from pathlib import Path
from types import SimpleNamespace
import pytest
import deerflow.community.ragflow.tools as ragflow_tools
from deerflow.community.ragflow.client import RAGFlowAPIError, RAGFlowConnectionError
from deerflow.community.ragflow.formatting import format_retrieval_result
from deerflow.config.tool_config import ToolConfig
from deerflow.tools.tools import get_available_tools
DATASET_ID_1 = "0123456789abcdef0123456789abcdef"
DATASET_ID_2 = "fedcba9876543210fedcba9876543210"
MISSING_DATASET_ID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
EMBEDDING_V2 = "text-embedding-v2@primary@Tongyi-Qianwen"
EMBEDDING_V3 = "text-embedding-v3@primary@Tongyi-Qianwen"
def _dataset(
dataset_id: str,
name: str,
*,
embedding_model: str = EMBEDDING_V3,
chunk_count: int = 1,
) -> dict:
return {
"id": dataset_id,
"name": name,
"embedding_model": embedding_model,
"chunk_count": chunk_count,
}
class FakeRAGFlowClient:
def __init__(
self,
*,
datasets_by_id: Mapping[str, list[dict]] | None = None,
dataset_errors_by_id: Mapping[str, Exception] | None = None,
all_datasets: list[dict] | None = None,
retrieval: dict | None = None,
retrieval_by_dataset_ids: Mapping[tuple[str, ...], dict] | None = None,
retrieval_errors_by_dataset_ids: Mapping[tuple[str, ...], Exception] | None = None,
error: Exception | None = None,
) -> None:
self.datasets_by_id = dict(datasets_by_id or {})
self.dataset_errors_by_id = dict(dataset_errors_by_id or {})
self.all_datasets = list(all_datasets or [])
self.retrieval = retrieval or {"chunks": [], "doc_aggs": [], "total": 0}
self.retrieval_by_dataset_ids = dict(retrieval_by_dataset_ids or {})
self.retrieval_errors_by_dataset_ids = dict(retrieval_errors_by_dataset_ids or {})
self.error = error
self.list_calls: list[str | None] = []
self.retrieve_calls: list[tuple[str, dict]] = []
async def list_datasets(self, *, dataset_id: str | None = None) -> list[dict]:
if self.error is not None:
raise self.error
self.list_calls.append(dataset_id)
if dataset_id is None:
return self.all_datasets
if error := self.dataset_errors_by_id.get(dataset_id):
raise error
return self.datasets_by_id.get(dataset_id, [])
async def retrieve(self, query: str, **kwargs: object) -> dict:
if self.error is not None:
raise self.error
self.retrieve_calls.append((query, kwargs))
dataset_ids = kwargs.get("dataset_ids")
key = tuple(dataset_ids) if isinstance(dataset_ids, list) else ()
if error := self.retrieval_errors_by_dataset_ids.get(key):
raise error
if key in self.retrieval_by_dataset_ids:
return self.retrieval_by_dataset_ids[key]
return self.retrieval
@pytest.fixture(autouse=True)
def reset_warning_deduplication() -> None:
ragflow_tools._warned.clear()
def _config(
*,
configured: bool = True,
api_key: str | None = "ragflow-secret",
base_url: str = "http://ragflow.test",
datasets: list[str] | None = None,
page_size: int = 8,
) -> SimpleNamespace:
extra: dict[str, object] = {
"base_url": base_url,
"api_key": api_key,
"timeout": 30,
"page_size": page_size,
"similarity_threshold": 0.2,
"vector_similarity_weight": 0.3,
"top_k": 256,
"max_chars_per_chunk": 800,
"max_total_chars": 8000,
}
if datasets is not None:
extra["datasets"] = datasets
search_config = ToolConfig(
name="knowledge_search",
group="knowledge",
use="deerflow.community.ragflow.tools:knowledge_search_tool",
**extra,
)
return SimpleNamespace(
get_tool_config=lambda name: search_config if configured and name == "knowledge_search" else None,
)
def _install(monkeypatch: pytest.MonkeyPatch, fake: FakeRAGFlowClient, *, config: SimpleNamespace | None = None) -> None:
monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: config or _config(datasets=[DATASET_ID_1]))
monkeypatch.setattr(ragflow_tools, "_build_client", lambda settings: fake)
@pytest.mark.anyio
async def test_knowledge_search_resolves_configured_ids_to_current_names(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
datasets_by_id={
DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")],
DATASET_ID_2: [_dataset(DATASET_ID_2, "Engineering")],
},
retrieval={
"chunks": [
{
"dataset_id": DATASET_ID_1,
"document_id": "doc-1",
"document_keyword": "handbook.pdf",
"content": "Annual leave is based on years of service.",
"similarity": 0.874,
}
],
"doc_aggs": [{"doc_id": "doc-1", "doc_name": "handbook.pdf", "count": 1}],
"total": 1,
},
)
_install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, DATASET_ID_2]))
result = await ragflow_tools.knowledge_search("annual leave")
assert fake.list_calls == [DATASET_ID_1, DATASET_ID_2]
assert fake.retrieve_calls == [
(
"annual leave",
{
"dataset_ids": [DATASET_ID_1, DATASET_ID_2],
"page_size": 8,
"similarity_threshold": 0.2,
"vector_similarity_weight": 0.3,
"top_k": 256,
},
)
]
assert "[1] HR Policies / handbook.pdf (score 0.87)" in result
assert "Annual leave" in result
assert "Matched documents: handbook.pdf (1 chunk)" in result
assert DATASET_ID_1 not in result
assert DATASET_ID_2 not in result
@pytest.mark.anyio
async def test_knowledge_search_uses_id_filter_and_survives_dataset_rename(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Renamed Policies")]},
retrieval={"chunks": [{"dataset_id": DATASET_ID_1, "document_keyword": "policy.pdf", "content": "Current policy."}]},
)
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search("leave")
assert fake.list_calls == [DATASET_ID_1]
assert fake.retrieve_calls[0][1]["dataset_ids"] == [DATASET_ID_1]
assert "Renamed Policies / policy.pdf" in result
assert DATASET_ID_1 not in result
@pytest.mark.anyio
async def test_missing_bound_dataset_returns_indexed_operator_guidance(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeRAGFlowClient(
datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Existing")]},
)
_install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, MISSING_DATASET_ID]))
with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"):
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: The 2nd entry of knowledge_search.datasets was not found or is inaccessible; check config.yaml."
assert MISSING_DATASET_ID not in result
assert fake.list_calls == [DATASET_ID_1, MISSING_DATASET_ID]
assert fake.retrieve_calls == []
assert MISSING_DATASET_ID in caplog.text
assert "code=None" in caplog.text
@pytest.mark.anyio
async def test_missing_bound_dataset_error_does_not_expose_configured_id(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient()
_install(monkeypatch, fake, config=_config(datasets=[MISSING_DATASET_ID]))
result = await ragflow_tools.knowledge_search("leave")
assert MISSING_DATASET_ID not in result
assert "[DATASET_ID]" not in result
@pytest.mark.anyio
async def test_bound_dataset_api_error_uses_normal_redacted_error_handler(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeRAGFlowClient(
dataset_errors_by_id={
DATASET_ID_1: RAGFlowAPIError("invalid credential ragflow-secret", code=102),
}
)
_install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1]))
with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"):
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: invalid credential [REDACTED]"
assert "code=102" in caplog.text
assert fake.retrieve_calls == []
@pytest.mark.anyio
async def test_mismatched_id_filtered_response_returns_operator_guidance(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_2, "Wrong dataset")]})
_install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1]))
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: The 1st entry of knowledge_search.datasets was not found or is inaccessible; check config.yaml."
assert fake.retrieve_calls == []
@pytest.mark.anyio
async def test_missing_dataset_binding_lists_all_and_passes_every_id(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
all_datasets=[
_dataset(DATASET_ID_1, "HR Policies"),
_dataset(DATASET_ID_2, "Engineering"),
],
retrieval={"chunks": [{"dataset_id": DATASET_ID_2, "document_keyword": "guide.pdf", "content": "Build guide."}]},
)
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("leave")
assert fake.list_calls == [None]
assert fake.retrieve_calls[0][1]["dataset_ids"] == [DATASET_ID_1, DATASET_ID_2]
assert "Engineering / guide.pdf" in result
assert DATASET_ID_1 not in result
assert DATASET_ID_2 not in result
@pytest.mark.anyio
async def test_mixed_embedding_models_are_retrieved_in_parallel_groups_and_rank_interleaved(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
datasets_by_id={
DATASET_ID_1: [_dataset(DATASET_ID_1, "Legacy", embedding_model=EMBEDDING_V2)],
DATASET_ID_2: [_dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3)],
},
retrieval_by_dataset_ids={
(DATASET_ID_1,): {
"chunks": [
{"dataset_id": DATASET_ID_1, "document_id": "legacy-1", "document_keyword": "legacy-1.txt", "content": "Legacy rank one.", "similarity": 0.41},
{"dataset_id": DATASET_ID_1, "document_id": "legacy-2", "document_keyword": "legacy-2.txt", "content": "Legacy rank two.", "similarity": 0.99},
],
"doc_aggs": [
{"doc_id": "legacy-1", "doc_name": "legacy-1.txt", "count": 1},
{"doc_id": "legacy-2", "doc_name": "legacy-2.txt", "count": 1},
],
"total": 2,
},
(DATASET_ID_2,): {
"chunks": [
{"dataset_id": DATASET_ID_2, "document_id": "current-1", "document_keyword": "current-1.txt", "content": "Current rank one.", "similarity": 0.87},
{"dataset_id": DATASET_ID_2, "document_id": "current-2", "document_keyword": "current-2.txt", "content": "Current rank two.", "similarity": 0.86},
],
"doc_aggs": [
{"doc_id": "current-1", "doc_name": "current-1.txt", "count": 1},
{"doc_id": "current-2", "doc_name": "current-2.txt", "count": 1},
],
"total": 2,
},
},
)
_install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, DATASET_ID_2], page_size=3))
result = await ragflow_tools.knowledge_search("policy")
assert [call[1]["dataset_ids"] for call in fake.retrieve_calls] == [[DATASET_ID_1], [DATASET_ID_2]]
assert result.index("Legacy rank one.") < result.index("Current rank one.") < result.index("Legacy rank two.")
assert "Current rank two." not in result
assert "current-2.txt (1 chunk)" not in result
assert "(score " not in result
assert DATASET_ID_1 not in result
assert DATASET_ID_2 not in result
@pytest.mark.anyio
async def test_all_dataset_scope_skips_empty_datasets_before_grouped_retrieval(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
all_datasets=[
_dataset(DATASET_ID_1, "Empty legacy", embedding_model=EMBEDDING_V2, chunk_count=0),
_dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3, chunk_count=4),
],
retrieval_by_dataset_ids={(DATASET_ID_2,): {"chunks": [{"dataset_id": DATASET_ID_2, "document_keyword": "guide.txt", "content": "Searchable."}], "doc_aggs": [], "total": 1}},
)
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("searchable")
assert [call[1]["dataset_ids"] for call in fake.retrieve_calls] == [[DATASET_ID_2]]
assert "Searchable." in result
@pytest.mark.anyio
async def test_all_dataset_scope_skips_empty_dataset_without_embedding_model_and_warns(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeRAGFlowClient(
all_datasets=[
_dataset(DATASET_ID_1, "Unconfigured empty", embedding_model="", chunk_count=0),
_dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3, chunk_count=4),
],
retrieval_by_dataset_ids={
(DATASET_ID_2,): {
"chunks": [
{
"dataset_id": DATASET_ID_2,
"document_keyword": "guide.txt",
"content": "Remaining dataset result.",
"similarity": 0.75,
}
],
"doc_aggs": [],
"total": 1,
}
},
)
_install(monkeypatch, fake, config=_config(datasets=None))
with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"):
result = await ragflow_tools.knowledge_search("searchable")
assert [call[1]["dataset_ids"] for call in fake.retrieve_calls] == [[DATASET_ID_2]]
assert "Remaining dataset result." in result
assert "(score 0.75)" in result
assert DATASET_ID_1 not in result
assert "Skipping empty RAGFlow dataset without embedding model metadata" in caplog.text
assert DATASET_ID_1 in caplog.text
@pytest.mark.anyio
async def test_all_empty_dataset_scope_returns_no_content_without_retrieval(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
all_datasets=[
_dataset(DATASET_ID_1, "Empty legacy", embedding_model=EMBEDDING_V2, chunk_count=0),
_dataset(DATASET_ID_2, "Empty current", embedding_model=EMBEDDING_V3, chunk_count=0),
]
)
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("anything")
assert result == "No relevant content found."
assert fake.retrieve_calls == []
@pytest.mark.anyio
async def test_grouped_retrieval_limits_concurrency_to_four(monkeypatch: pytest.MonkeyPatch) -> None:
class ConcurrencyTrackingClient(FakeRAGFlowClient):
def __init__(self) -> None:
super().__init__(all_datasets=[_dataset(f"dataset-{index}", f"Dataset {index}", embedding_model=f"embedding-{index}@provider") for index in range(5)])
self.active_retrievals = 0
self.max_active_retrievals = 0
async def retrieve(self, query: str, **kwargs: object) -> dict:
self.retrieve_calls.append((query, kwargs))
self.active_retrievals += 1
self.max_active_retrievals = max(self.max_active_retrievals, self.active_retrievals)
try:
await asyncio.sleep(0.05)
return {"chunks": [], "doc_aggs": [], "total": 0}
finally:
self.active_retrievals -= 1
fake = ConcurrencyTrackingClient()
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("anything")
assert result == "No relevant content found."
assert len(fake.retrieve_calls) == 5
assert fake.max_active_retrievals == 4
@pytest.mark.anyio
async def test_dataset_without_embedding_metadata_returns_protocol_error(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(all_datasets=[{"id": DATASET_ID_1, "name": "Broken", "chunk_count": 1}])
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("anything")
assert result == "Error: RAGFlow request failed: RAGFlow returned a searchable dataset without embedding model metadata."
assert fake.retrieve_calls == []
@pytest.mark.anyio
async def test_group_failure_remains_strict_and_redacts_secret_and_dataset_id(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
all_datasets=[
_dataset(DATASET_ID_1, "Legacy", embedding_model=EMBEDDING_V2),
_dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3),
],
retrieval_by_dataset_ids={(DATASET_ID_2,): {"chunks": [], "doc_aggs": [], "total": 0}},
retrieval_errors_by_dataset_ids={(DATASET_ID_1,): RAGFlowAPIError(f"dataset {DATASET_ID_1} rejected ragflow-secret", code=102)},
)
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("anything")
assert result == "Error: dataset [DATASET_ID] rejected [REDACTED]"
assert len(fake.retrieve_calls) == 2
@pytest.mark.anyio
async def test_missing_dataset_binding_with_empty_catalog_returns_guidance(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient()
_install(monkeypatch, fake, config=_config(datasets=None))
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: No accessible RAGFlow datasets were found; configure knowledge_search.datasets or add a dataset in RAGFlow."
assert fake.list_calls == [None]
assert fake.retrieve_calls == []
@pytest.mark.anyio
async def test_missing_api_key_returns_english_guidance_and_warns_only_once(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeRAGFlowClient()
_install(monkeypatch, fake, config=_config(api_key=None, datasets=[DATASET_ID_1]))
with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"):
first = await ragflow_tools.knowledge_search("leave")
second = await ragflow_tools.knowledge_search("benefits")
assert first == "Error: RAGFlow API key is not configured; set knowledge_search.api_key in config.yaml (prefer $RAGFLOW_API_KEY)."
assert second == first
assert caplog.text.count("RAGFlow API key is not configured") == 1
assert fake.list_calls == []
@pytest.mark.anyio
async def test_missing_knowledge_search_config_returns_english_guidance(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient()
_install(monkeypatch, fake, config=_config(configured=False, datasets=[DATASET_ID_1]))
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: knowledge_search is not configured; add its RAGFlow settings to the tools list in config.yaml."
assert fake.list_calls == []
@pytest.mark.anyio
async def test_api_error_is_returned_as_readable_text(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Policies")]},
retrieval_errors_by_dataset_ids={(DATASET_ID_1,): RAGFlowAPIError("embedding models do not match", code=102)},
)
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: embedding models do not match"
@pytest.mark.anyio
async def test_error_path_redacts_dataset_uuid(monkeypatch: pytest.MonkeyPatch) -> None:
dataset_id = "0123456789abcdef0123456789abcdef"
fake = FakeRAGFlowClient(
datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Policies")]},
retrieval_errors_by_dataset_ids={(DATASET_ID_1,): RAGFlowAPIError(f"dataset {dataset_id} failed", code=102)},
)
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search("leave")
assert dataset_id not in result
assert "[DATASET_ID]" in result
@pytest.mark.anyio
async def test_success_path_preserves_legitimate_uuid_and_md5_text(monkeypatch: pytest.MonkeyPatch) -> None:
uuid = "123e4567-e89b-12d3-a456-426614174000"
md5 = "d41d8cd98f00b204e9800998ecf8427e"
fake = FakeRAGFlowClient(
datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")]},
retrieval={
"chunks": [
{
"dataset_id": DATASET_ID_1,
"document_keyword": "checksums.txt",
"content": f"Trace {uuid}; checksum {md5}.",
}
]
},
)
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search("trace")
assert uuid in result
assert md5 in result
assert "[DATASET_ID]" not in result
@pytest.mark.anyio
async def test_success_path_still_redacts_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(
datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")]},
retrieval={
"chunks": [
{
"dataset_id": DATASET_ID_1,
"document_keyword": "secret.txt",
"content": "Accidental echo: ragflow-secret",
}
]
},
)
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search("secret")
assert "ragflow-secret" not in result
assert "[REDACTED]" in result
@pytest.mark.anyio
async def test_connection_error_is_english_and_does_not_leak_key(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeRAGFlowClient(error=RAGFlowConnectionError("ConnectError: refused ragflow-secret"))
_install(monkeypatch, fake)
with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"):
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: Unable to connect to RAGFlow (http://ragflow.test): ConnectError: refused [REDACTED]"
assert "ragflow-secret" not in result
assert "ragflow-secret" not in caplog.text
@pytest.mark.anyio
@pytest.mark.parametrize(
"base_url",
[
"http://ragflow-secret@ragflow.test",
"http://ragflow%2Dsecret@ragflow.test",
"http://user:ragflow-secret@ragflow.test",
],
)
async def test_base_url_with_plain_or_encoded_userinfo_is_rejected_without_leaking_credentials(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
base_url: str,
) -> None:
fake = FakeRAGFlowClient()
_install(monkeypatch, fake, config=_config(base_url=base_url, datasets=[DATASET_ID_1]))
with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"):
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: Invalid RAGFlow settings for knowledge_search; check config.yaml."
assert "ragflow-secret" not in result
assert "ragflow-secret" not in caplog.text
assert "ragflow%2Dsecret" not in caplog.text
assert fake.list_calls == []
@pytest.mark.anyio
async def test_empty_query_has_english_error(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient()
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search(" ")
assert result == "Error: query must not be empty."
assert fake.list_calls == []
@pytest.mark.anyio
async def test_empty_retrieval_has_explicit_english_message(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")]})
_install(monkeypatch, fake)
result = await ragflow_tools.knowledge_search("nothing")
assert result == "No relevant content found."
def test_formatting_uses_only_normalized_chunk_fields() -> None:
result = format_retrieval_result(
{
"chunks": [
{
"kb_id": "dataset-1",
"doc_id": "doc-legacy",
"docnm_kwd": "legacy.pdf",
"content": "abcdefghij",
"similarity": 0.5,
}
]
},
dataset_names_by_id={"dataset-1": "HR Policies"},
max_chars_per_chunk=5,
max_total_chars=1000,
)
assert "Unknown dataset / Unknown document" in result
assert "HR Policies" not in result
assert "legacy.pdf" not in result
assert "abcd…" in result
assert "abcdefghij" not in result
assert "dataset-1" not in result
def test_formatting_applies_total_response_truncation_in_english() -> None:
result = format_retrieval_result(
{
"chunks": [
{
"dataset_id": "dataset-1",
"document_keyword": f"document-{index}.txt",
"content": "content " * 20,
"similarity": 0.5,
}
for index in range(4)
]
},
dataset_names_by_id={"dataset-1": "Policies"},
max_chars_per_chunk=100,
max_total_chars=120,
)
assert len(result) <= 120
assert result.endswith("… (response truncated)")
def test_retrieval_settings_load_bound_dataset_ids_and_hide_secret(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config(datasets=[DATASET_ID_1, DATASET_ID_2]))
config, error = ragflow_tools._settings_or_error()
assert error is None
assert config is not None
assert config.datasets == [DATASET_ID_1, DATASET_ID_2]
assert str(config.base_url).rstrip("/") == "http://ragflow.test"
assert config.page_size == 8
assert config.max_chars_per_chunk == 800
assert config.max_total_chars == 8000
assert "ragflow-secret" not in repr(config)
def test_retrieval_settings_allow_omitting_dataset_ids(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config(datasets=None))
config, error = ragflow_tools._settings_or_error()
assert error is None
assert config is not None
assert config.datasets is None
@pytest.mark.anyio
async def test_explicitly_empty_dataset_allowlist_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeRAGFlowClient(all_datasets=[_dataset(DATASET_ID_1, "Must remain inaccessible")])
_install(monkeypatch, fake, config=_config(datasets=[]))
result = await ragflow_tools.knowledge_search("leave")
assert result == "Error: Invalid RAGFlow settings for knowledge_search; check config.yaml."
assert fake.list_calls == []
assert fake.retrieve_calls == []
def test_agent_exposes_only_query_on_single_search_tool() -> None:
assert not hasattr(ragflow_tools, "list_knowledge_bases_tool")
assert not hasattr(ragflow_tools, "list_knowledge_bases")
assert ragflow_tools.knowledge_search_tool.name == "knowledge_search"
assert ragflow_tools.knowledge_search_tool.coroutine is not None
assert set(ragflow_tools.knowledge_search_tool.tool_call_schema.model_fields) == {"query"}
def test_tool_assembly_hides_bound_dataset_ids_without_network_io(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(ragflow_tools, "_build_client", lambda settings: pytest.fail("tool assembly must not perform network IO"))
tool_config = ToolConfig(
name="knowledge_search",
group="knowledge",
use="deerflow.community.ragflow.tools:knowledge_search_tool",
base_url="http://ragflow.test",
api_key="ragflow-secret",
datasets=[DATASET_ID_1, DATASET_ID_2],
)
config = SimpleNamespace(
tools=[tool_config],
sandbox=SimpleNamespace(use="example.remote:Sandbox"),
skill_evolution=SimpleNamespace(enabled=False),
models=[],
acp_agents={},
get_model_config=lambda name: None,
)
tools = get_available_tools(include_mcp=False, app_config=config)
assembled = next(tool for tool in tools if tool.name == "knowledge_search")
assert "If knowledge_search.datasets is omitted" in assembled.description
assert "all datasets accessible to the configured RAGFlow API key" in assembled.description
assert DATASET_ID_1 not in assembled.description
assert DATASET_ID_2 not in assembled.description
assert "ragflow-secret" not in assembled.description
assert {tool.name for tool in tools}.isdisjoint({"list_knowledge_bases"})
def test_ragflow_package_has_explicit_init_file() -> None:
package_dir = Path(ragflow_tools.__file__).resolve().parent
assert (package_dir / "__init__.py").is_file()

25
config.example.yaml
View File

@ -669,6 +669,7 @@ tool_groups:
- name: file:write
- name: bash
- name: browser
- name: knowledge
# ============================================================================
# Tools Configuration
@ -676,6 +677,30 @@ tool_groups:
# Configure available tools for the agent to use
tools:
# RAGFlow knowledge retrieval (read-only). Uncomment this single entry.
# `datasets` is optional. Omit it to list every tenant-visible dataset at
# search time; an explicit `datasets: []` is invalid. Empty datasets are
# skipped; searchable datasets are grouped by
# embedding model and retrieved with at most four groups in parallel. Group
# ranks are interleaved under the global `page_size` limit; scores are omitted
# when multiple groups are searched because they are not comparable.
# Configure stable IDs only to restrict scope. IDs and catalog listing never
# reach the model.
# - name: knowledge_search
# group: knowledge
# use: deerflow.community.ragflow.tools:knowledge_search_tool
# base_url: http://localhost:9380 # Docker: use a backend-reachable URL
# api_key: $RAGFLOW_API_KEY
# datasets: # Optional operator-controlled allowlist
# - 0123456789abcdef0123456789abcdef # Replace with a RAGFlow dataset ID
# timeout: 30
# page_size: 8
# similarity_threshold: 0.2
# vector_similarity_weight: 0.3
# top_k: 256
# max_chars_per_chunk: 800
# max_total_chars: 8000
# Web search tool (uses DuckDuckGo, no API key required)
- name: web_search
group: web

22
deploy/helm/deer-flow/values.yaml
View File

@ -307,8 +307,30 @@ config: |
- name: file:read
- name: file:write
- name: bash
# - name: knowledge # Enable with the RAGFlow tool below.
tools:
# Optional tenant-shared, read-only RAGFlow retrieval. Put RAGFLOW_API_KEY
# in `secrets`. `datasets` is an optional stable-ID allowlist; omit it to
# search all tenant-visible datasets. An explicit empty list is invalid.
# Empty datasets are skipped; remaining
# datasets are grouped by embedding model with up to four parallel retrieval
# requests and one global `page_size` limit. Multi-group scores are omitted
# because they are not comparable. The Agent cannot see the IDs.
# - name: knowledge_search
# group: knowledge
# use: deerflow.community.ragflow.tools:knowledge_search_tool
# base_url: http://ragflow:9380
# api_key: $RAGFLOW_API_KEY
# datasets: # Optional operator-controlled allowlist
# - 0123456789abcdef0123456789abcdef
# timeout: 30
# page_size: 8
# similarity_threshold: 0.2
# vector_similarity_weight: 0.3
# top_k: 256
# max_chars_per_chunk: 800
# max_total_chars: 8000
- name: web_search
group: web
use: deerflow.community.ddg_search.tools:web_search_tool