diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index bee2139cee5..862027c30bb 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -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: diff --git a/backend/packages/harness/deerflow/community/ragflow/__init__.py b/backend/packages/harness/deerflow/community/ragflow/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/backend/packages/harness/deerflow/community/ragflow/client.py b/backend/packages/harness/deerflow/community/ragflow/client.py new file mode 100644 index 00000000000..5d9d69eecc5 --- /dev/null +++ b/backend/packages/harness/deerflow/community/ragflow/client.py @@ -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 diff --git a/backend/packages/harness/deerflow/community/ragflow/formatting.py b/backend/packages/harness/deerflow/community/ragflow/formatting.py new file mode 100644 index 00000000000..7a179273a40 --- /dev/null +++ b/backend/packages/harness/deerflow/community/ragflow/formatting.py @@ -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}" diff --git a/backend/packages/harness/deerflow/community/ragflow/tools.py b/backend/packages/harness/deerflow/community/ragflow/tools.py new file mode 100644 index 00000000000..df41a15ca89 --- /dev/null +++ b/backend/packages/harness/deerflow/community/ragflow/tools.py @@ -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"(? 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, +) diff --git a/backend/tests/test_ragflow_client.py b/backend/tests/test_ragflow_client.py new file mode 100644 index 00000000000..62b96360ad3 --- /dev/null +++ b/backend/tests/test_ragflow_client.py @@ -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") diff --git a/backend/tests/test_ragflow_tools.py b/backend/tests/test_ragflow_tools.py new file mode 100644 index 00000000000..03e015e0e9d --- /dev/null +++ b/backend/tests/test_ragflow_tools.py @@ -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() diff --git a/config.example.yaml b/config.example.yaml index eeb611395f2..6997ebb2f55 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -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 diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index 940f76ebc96..3e6d752d683 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -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