feat(knowledge): add read-only RAGFlow retrieval

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:

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

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}"