feat(knowledge): add read-only LightRAG retrieval, fixes #5208 (#5209)

This commit is contained in:
hataa 2026-09-06 10:21:58 +08:00 committed by GitHub
parent a4ff4b0b3b
commit aec7d73890
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 1313 additions and 0 deletions

View File

@ -279,6 +279,67 @@ 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.
### LightRAG Knowledge Retrieval
LightRAG integration is disabled by default. It is an alternative provider for
the same read-only `knowledge_search` tool: an operator picks RAGFlow or
LightRAG by which entry appears in the `tools:` list — the two entries share
one name, and on duplicate names DeerFlow keeps the **first** configured
entry, so configure exactly one. Requires LightRAG v1.4.9 or newer: v1.4.8
introduced the data-retrieval endpoint but returned a pre-envelope response
shape, and the `status`/`data` envelope plus the citation fields consumed
here shipped in v1.4.9. DeerFlow does not persist any index
metadata; LightRAG stays the sole source of truth, and the deployment's
single indexed workspace is always searched.
```yaml
tool_groups:
- name: knowledge
tools:
- name: knowledge_search
group: knowledge
use: deerflow.community.lightrag.tools:knowledge_search_tool
base_url: http://localhost:9621
api_key: $LIGHTRAG_API_KEY
mode: mix
timeout: 30
top_k: 60
chunk_top_k: 8
max_chars_per_chunk: 800
max_total_chars: 8000
```
The tool is opt-in through the normal `tools:` list. Retrieval uses LightRAG's
`POST /query/data` endpoint, which performs no LLM generation and returns
structured entities, relationships, chunks, and references; DeerFlow keeps the
chunks — the document text the selected mode already ranked as relevant — and
formats them as citation-numbered text, dropping the graph objects to stay
compact and keep the citation shape shared with the RAGFlow provider. `mode`
selects the retrieval strategy (`naive`, `local`, `global`, `hybrid`, or
`mix`; default `mix`, matching the LightRAG API's own `QueryRequest` default;
`bypass` is rejected because it skips the index entirely). `top_k` bounds the
entities retrieved in `local` mode or relationships in `global` mode, and the
optional `chunk_top_k` bounds the text chunks retrieved and kept after
reranking; both are capped at 1000 by the LightRAG server. Short queries that
fail LightRAG's minimum-length validation surface the server's readable
message. `max_chars_per_chunk` / `max_total_chars` bound the model-visible
output size.
`api_key` is optional because LightRAG may run without authentication. Only
omit it for loopback or trusted-network deployments — a network-exposed
LightRAG must have authentication enabled, and then the key is sent as the
`X-API-Key` header and redacted from every model-visible error and from
server logs. Blank values are treated as unauthenticated. `base_url` must not
contain embedded username or password information, and for Docker or
Kubernetes it must be reachable from the Gateway container or Pod.
Internal identifiers (chunk IDs and the response-local reference IDs) are
never exposed to the Agent; citations use the operator-readable `file_path`.
This integration is retrieval-only. Document insertion, indexing, and graph
mutation remain in LightRAG and are not exposed as Agent tools or DeerFlow
APIs.
### Tool Groups
Organize tools into logical groups:

View File

@ -0,0 +1,170 @@
"""Minimal asynchronous client for the LightRAG APIs DeerFlow consumes."""
from __future__ import annotations
from typing import Any
import httpx
QUERY_MODES = ("naive", "local", "global", "hybrid", "mix")
class LightRAGError(Exception):
"""Base class for normalized LightRAG failures."""
class LightRAGAPIError(LightRAGError):
"""LightRAG rejected the request with a readable failure."""
class LightRAGConnectionError(LightRAGError):
"""LightRAG could not be reached or timed out."""
class LightRAGProtocolError(LightRAGError):
"""LightRAG returned an invalid or unexpected HTTP response."""
class LightRAGClient:
"""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. The optional API key is sent as the ``X-API-Key``
request header, the single credential form LightRAG documents for
API-key-authenticated servers; unauthenticated deployments simply omit it.
"""
def __init__(
self,
*,
base_url: str,
api_key: str | None,
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
def _error_message(self, payload: object, status_code: int) -> str | None:
"""Extract a redacted, human-readable message from an error payload.
LightRAG failures carry text in ``message`` (QueryDataResponse
envelope) or ``detail`` (FastAPI error handler, either a string or a
list of validation objects whose ``msg`` holds the reason, prefixed by
pydantic's "Value error, "). Anything else — structured bodies, plain
text, missing payloads yields ``None`` so the caller falls back to a
stable protocol error instead of dumping raw JSON at the model.
"""
if not isinstance(payload, dict):
return None
candidate = payload.get("message")
if not isinstance(candidate, str) or not candidate.strip():
detail = payload.get("detail")
if isinstance(detail, str):
candidate = detail
elif isinstance(detail, list):
candidate = self._first_validation_message(detail)
if isinstance(candidate, str) and candidate.strip():
text = candidate.removeprefix("Value error, ").strip()
return self._redact(text)
return None
@staticmethod
def _first_validation_message(items: list[object]) -> str | None:
for item in items:
if isinstance(item, dict):
message = item.get("msg")
if isinstance(message, str) and message.strip():
return message
return None
async def _request(self, method: str, path: str, *, json: dict[str, Any] | None = None) -> dict[str, Any]:
request_headers = {"Accept": "application/json"}
if self._api_key:
request_headers["X-API-Key"] = self._api_key
client_kwargs: dict[str, Any] = {
"base_url": self.base_url,
"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, json=json)
except httpx.TimeoutException:
raise LightRAGConnectionError(f"LightRAG request timed out after {self.timeout:g} seconds.") from None
except httpx.RequestError as exc:
detail = self._redact(exc)
raise LightRAGConnectionError(f"{type(exc).__name__}: {detail}") from None
if response.is_error:
# A 404 on the data-retrieval endpoint means either a wrong
# base_url or a LightRAG older than v1.4.9, where /query/data did
# not exist yet; the default "Not Found" body helps neither case.
if response.status_code == 404:
raise LightRAGAPIError("LightRAG data-retrieval endpoint not found; check base_url or upgrade LightRAG to v1.4.9 or newer.")
try:
error_payload = response.json()
except ValueError:
raise LightRAGProtocolError(f"LightRAG request failed (HTTP {response.status_code}).") from None
message = self._error_message(error_payload, response.status_code)
if message is not None:
raise LightRAGAPIError(message)
raise LightRAGProtocolError(f"LightRAG request failed (HTTP {response.status_code}).")
try:
payload = response.json()
except ValueError:
raise LightRAGProtocolError("LightRAG returned invalid JSON.") from None
if not isinstance(payload, dict):
raise LightRAGProtocolError("LightRAG returned a non-object JSON payload.")
status = payload.get("status")
if status != "success":
if "chunks" in payload or "entities" in payload:
# v1.4.8 answers with a flat {entities, relationships,
# chunks, metadata} payload; the status/data envelope and the
# reference-bearing chunk fields both shipped in v1.4.9.
raise LightRAGAPIError("LightRAG server response predates v1.4.9; upgrade LightRAG to v1.4.9 or newer to use the data-retrieval endpoint.")
message = payload.get("message")
text = self._redact(message) if isinstance(message, str) and message.strip() else "LightRAG request failed."
raise LightRAGAPIError(text)
return payload
async def query_data(
self,
query: str,
*,
mode: str = "hybrid",
top_k: int = 60,
chunk_top_k: int | None = None,
) -> dict[str, Any]:
"""Run one read-only structured retrieval against ``POST /query/data``.
The data endpoint performs no LLM generation and always returns
entities, relationships, chunks, and references, which is exactly the
read-only shape DeerFlow's knowledge tool consumes.
"""
if mode not in QUERY_MODES:
raise ValueError(f"mode must be one of {QUERY_MODES}")
request_body: dict[str, object] = {"query": query, "mode": mode, "top_k": top_k}
if chunk_top_k is not None:
request_body["chunk_top_k"] = chunk_top_k
payload = await self._request("POST", "/query/data", json=request_body)
data = payload.get("data")
if not isinstance(data, dict):
raise LightRAGProtocolError("LightRAG returned an invalid retrieval result.")
return data

View File

@ -0,0 +1,89 @@
"""Compact, citation-friendly formatting for LightRAG 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 _chunks(value: object) -> list[Mapping[str, Any]]:
if not isinstance(value, list):
return []
return [chunk for chunk in value if isinstance(chunk, Mapping)]
def _reference_file_paths(value: object) -> dict[str, str]:
if not isinstance(value, list):
return {}
file_paths: dict[str, str] = {}
for reference in value:
if not isinstance(reference, Mapping):
continue
reference_id = reference.get("reference_id")
file_path = reference.get("file_path")
if isinstance(reference_id, str) and isinstance(file_path, str) and file_path.strip():
file_paths.setdefault(reference_id, file_path.strip())
return file_paths
def format_retrieval_result(
result: Mapping[str, Any],
*,
max_chars_per_chunk: int = 800,
max_total_chars: int = 8000,
) -> str:
"""Format one LightRAG ``/query/data`` payload into compact cited text.
The consumed field names (``chunks[].content/file_path/chunk_id/reference_id``
and ``references[].reference_id/file_path``) are cross-checked against the
LightRAG v1.5.7 source tree; the endpoint shipped in v1.4.8 but only
v1.4.9 introduced the status/data envelope and these citation fields. Opaque
internal identifiers (``chunk_id`` and the response-local
``reference_id``) are never emitted; the operator-readable ``file_path``
labels each citation instead. Entities and relationships are deliberately
dropped: the chunks are the document text the selected query mode already
ranked as relevant, and keeping the output compact preserves the citation
shape shared with the RAGFlow provider.
"""
raw_chunks = _chunks(result.get("chunks"))
if not raw_chunks:
return "No relevant content found."
file_paths_by_reference = _reference_file_paths(result.get("references"))
entries: list[str] = []
matched_documents: list[str] = []
counts_by_document: dict[str, int] = {}
for index, chunk in enumerate(raw_chunks, start=1):
file_path = chunk.get("file_path")
if not isinstance(file_path, str) or not file_path.strip():
reference_id = chunk.get("reference_id")
file_path = file_paths_by_reference.get(str(reference_id)) if reference_id is not None else None
document_name = str(file_path).strip() if file_path and str(file_path).strip() else "Unknown document"
counts_by_document[document_name] = counts_by_document.get(document_name, 0) + 1
content = str(chunk.get("content") or "").strip()
content = _truncate(content, max_chars_per_chunk)
entries.append(f"[{index}] {document_name}\n{content}")
for document_name, count in counts_by_document.items():
unit = "chunk" if count == 1 else "chunks"
matched_documents.append(f"{document_name} ({count} {unit})")
entries.append(f"Matched documents: {', '.join(matched_documents)}")
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}"

View File

@ -0,0 +1,164 @@
"""Read-only Agent tool for operator-scoped LightRAG knowledge retrieval."""
from __future__ import annotations
import logging
from collections.abc import Mapping
from typing import Literal
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 LightRAGAPIError, LightRAGClient, LightRAGConnectionError, LightRAGProtocolError
from .formatting import format_retrieval_result
logger = logging.getLogger(__name__)
_NO_RELEVANT_CONTENT = "No relevant content found."
class _LightRAGRetrievalSettings(BaseModel):
"""Validated provider settings stored on the knowledge_search tool entry."""
model_config = ConfigDict(validate_default=True)
base_url: AnyHttpUrl = Field(default="http://localhost:9621")
api_key: SecretStr | None = Field(default=None)
# LightRAG's QueryRequest also accepts "bypass", which skips the index and
# answers straight from the LLM; that would defeat a retrieval tool, so it
# is excluded on purpose. "mix" matches the LightRAG API's own default.
mode: Literal["naive", "local", "global", "hybrid", "mix"] = Field(default="mix")
timeout: float = Field(default=30, gt=0, le=600)
# The server caps both fields at MAX_QUERY_TOP_K = 1000; match that limit
# instead of inventing a tighter client-side one.
top_k: int = Field(default=60, ge=1, le=1000)
chunk_top_k: int | None = Field(default=None, ge=1, le=1000)
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("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: _LightRAGRetrievalSettings) -> str | None:
# LightRAG may run without authentication, so a missing key stays valid;
# blank values are treated as unconfigured rather than rejected.
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 _settings_from_extra(extra: Mapping[str, object]) -> _LightRAGRetrievalSettings:
return _LightRAGRetrievalSettings.model_validate(dict(extra))
def _settings_or_error() -> tuple[_LightRAGRetrievalSettings | 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 LightRAG settings to the tools list in config.yaml."
try:
settings = _settings_from_extra(tool_config.model_extra or {})
except ValidationError:
logger.warning("LightRAG knowledge_search tool configuration is invalid")
return None, "Error: Invalid LightRAG settings for knowledge_search; check config.yaml."
return settings, None
def _build_client(settings: _LightRAGRetrievalSettings) -> LightRAGClient:
return LightRAGClient(
base_url=str(settings.base_url).rstrip("/"),
api_key=_api_key(settings),
timeout=settings.timeout,
)
def _tool_error(exc: Exception, settings: _LightRAGRetrievalSettings) -> str:
key = _api_key(settings)
safe_detail = _redact_api_key(exc, key)
base_url = _redact_api_key(str(settings.base_url).rstrip("/"), key)
if isinstance(exc, LightRAGAPIError):
logger.warning("LightRAG API rejected a read-only tool request: %s", safe_detail)
return f"Error: {safe_detail}"
if isinstance(exc, LightRAGConnectionError):
logger.warning("LightRAG connection failed for %s (%s)", base_url, type(exc).__name__)
return f"Error: Unable to connect to LightRAG ({base_url}): {safe_detail}"
if isinstance(exc, LightRAGProtocolError):
logger.warning("LightRAG returned an invalid response for a read-only tool request (%s)", type(exc).__name__)
return f"Error: LightRAG request failed: {safe_detail}"
logger.warning("Unexpected LightRAG read-only tool failure (%s)", type(exc).__name__)
return "Error: An unexpected LightRAG retrieval error occurred; try again later."
async def knowledge_search(query: str) -> str:
"""Search the operator-configured LightRAG instance.
LightRAG has no dataset catalog to scope: the deployment's single indexed
workspace is always searched with the configured retrieval mode, so no
binding resolution happens before the one read-only request.
"""
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 LightRAG settings for knowledge_search; check config.yaml."
client = _build_client(settings)
try:
result = await client.query_data(
query,
mode=settings.mode,
top_k=settings.top_k,
chunk_top_k=settings.chunk_top_k,
)
formatted = format_retrieval_result(
result,
max_chars_per_chunk=settings.max_chars_per_chunk,
max_total_chars=settings.max_total_chars,
)
# API-key redaction remains mandatory on the success path; chunk and
# reference identifiers never enter the formatted text at all.
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 LightRAG knowledge base and return compact, citation-numbered source chunks retrieved with the configured graph/vector mode."
return f"{base} Internal identifiers are never shown to the model."
async def _knowledge_search_entrypoint(query: str) -> str:
"""Search the operator-configured LightRAG knowledge base.
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,
)

View File

@ -0,0 +1,368 @@
import json
import httpx
import pytest
from deerflow.community.lightrag.client import (
LightRAGAPIError,
LightRAGClient,
LightRAGConnectionError,
LightRAGProtocolError,
)
@pytest.mark.anyio
async def test_query_data_sends_documented_body_and_api_key_header() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
assert request.method == "POST"
assert request.url == httpx.URL("http://lightrag.test/query/data")
assert request.headers["X-API-Key"] == "lightrag-secret"
return httpx.Response(
200,
json={
"status": "success",
"message": None,
"data": {"chunks": [], "references": [], "entities": [], "relationships": []},
},
)
client = LightRAGClient(
base_url="http://lightrag.test/",
api_key="lightrag-secret",
timeout=12,
transport=httpx.MockTransport(handler),
)
data = await client.query_data("annual leave", mode="hybrid", top_k=60)
assert data == {"chunks": [], "references": [], "entities": [], "relationships": []}
assert len(requests) == 1
assert json.loads(requests[0].content) == {"query": "annual leave", "mode": "hybrid", "top_k": 60}
@pytest.mark.anyio
async def test_query_data_omits_api_key_header_when_unauthenticated() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
assert "Authorization" not in request.headers
assert "X-API-Key" not in request.headers
return httpx.Response(200, json={"status": "success", "data": {"chunks": []}})
client = LightRAGClient(
base_url="http://lightrag.test",
api_key=None,
transport=httpx.MockTransport(handler),
)
await client.query_data("annual leave")
assert len(requests) == 1
@pytest.mark.anyio
async def test_query_data_sends_chunk_top_k_only_when_configured() -> None:
requests: list[httpx.Request] = []
async def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(200, json={"status": "success", "data": {"chunks": []}})
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
await client.query_data("annual leave", mode="mix", top_k=15, chunk_top_k=8)
await client.query_data("annual leave", mode="mix", top_k=15)
assert json.loads(requests[0].content) == {"query": "annual leave", "mode": "mix", "top_k": 15, "chunk_top_k": 8}
assert json.loads(requests[1].content) == {"query": "annual leave", "mode": "mix", "top_k": 15}
@pytest.mark.anyio
async def test_query_data_rejects_unknown_mode_before_request() -> None:
called = False
async def handler(request: httpx.Request) -> httpx.Response:
nonlocal called
called = True
return httpx.Response(500)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(ValueError, match="mode must be one of"):
await client.query_data("fallback search", mode="vector")
assert called is False
@pytest.mark.anyio
async def test_non_success_envelope_is_normalized_and_redacts_api_key() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={"status": "failure", "message": "invalid credential lightrag-secret"},
)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError) as exc_info:
await client.query_data("annual leave")
assert "invalid credential" in str(exc_info.value)
assert "lightrag-secret" not in str(exc_info.value)
assert "[REDACTED]" in str(exc_info.value)
@pytest.mark.anyio
async def test_http_error_envelope_message_is_used() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
400,
json={"status": "failure", "message": "RAG query is too short"},
)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError, match="RAG query is too short"):
await client.query_data("ab")
@pytest.mark.anyio
async def test_http_error_detail_string_is_used() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(401, json={"detail": "Invalid API key provided"})
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError, match="Invalid API key provided"):
await client.query_data("annual leave")
@pytest.mark.anyio
async def test_structured_validation_error_message_is_extracted_from_detail() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
422,
json={
"detail": [
{
"type": "value_error",
"loc": ["body", "query"],
"msg": "Value error, RAG query is too short. Enter at least 3 English characters or an equivalent combination where each Chinese, Japanese or Korean character counts as 2.",
}
]
},
)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError) as exc_info:
await client.query_data("ab")
assert "RAG query is too short" in str(exc_info.value)
assert "Value error," not in str(exc_info.value)
@pytest.mark.anyio
async def test_structured_validation_error_without_readable_msg_falls_back_to_protocol_error() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(422, json={"detail": [{"type": "missing", "loc": ["body"]}]})
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGProtocolError, match=r"LightRAG request failed \(HTTP 422\)"):
await client.query_data("annual leave")
@pytest.mark.anyio
@pytest.mark.parametrize("body", [None, '{"detail": "Not Found"}'])
async def test_404_returns_upgrade_guidance_instead_of_plain_not_found(body: str | None) -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(404, text=body, headers={"Content-Type": "application/json"} if body else None)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError, match="v1.4.9"):
await client.query_data("annual leave")
@pytest.mark.anyio
async def test_pre_v149_flat_success_payload_returns_upgrade_error() -> None:
"""v1.4.8 answers 200 with a flat payload (no status/data envelope).
Pinning this shape matters: the documented minimum is v1.4.9, so a v1.4.8
server must fail with explicit upgrade guidance instead of a generic
"request failed" that hides the version incompatibility.
"""
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={
"entities": [{"entity_name": "LightRAG"}],
"relationships": [],
"chunks": [{"content": "v1.4.8 chunk without citation fields."}],
"metadata": {"query_mode": "hybrid"},
},
)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError, match="predates v1.4.9"):
await client.query_data("annual leave")
@pytest.mark.anyio
async def test_http_error_text_body_cannot_echo_api_key() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(500, text="internal error: lightrag-secret")
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGProtocolError) as exc_info:
await client.query_data("annual leave")
assert str(exc_info.value) == "LightRAG request failed (HTTP 500)."
assert "lightrag-secret" not in str(exc_info.value)
@pytest.mark.anyio
async def test_server_500_detail_string_is_used_verbatim() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(500, json={"detail": "Internal server error"})
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGAPIError, match="^Internal server error$"):
await client.query_data("annual leave")
@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 lightrag-secret", request=request)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
timeout=2,
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGConnectionError) as exc_info:
await client.query_data("annual leave")
assert str(exc_info.value) == "LightRAG request timed out after 2 seconds."
assert "lightrag-secret" not in str(exc_info.value)
assert "lightrag-secret" not in caplog.text
@pytest.mark.anyio
async def test_request_error_is_normalized_and_redacts_api_key() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("refused with lightrag-secret", request=request)
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGConnectionError) as exc_info:
await client.query_data("annual leave")
assert "ConnectError" in str(exc_info.value)
assert "lightrag-secret" not in str(exc_info.value)
assert "[REDACTED]" 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 = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGProtocolError, match="LightRAG returned invalid JSON"):
await client.query_data("annual leave")
@pytest.mark.anyio
async def test_non_object_json_payload_is_rejected() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=["not", "an", "object"])
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGProtocolError, match="non-object JSON payload"):
await client.query_data("annual leave")
@pytest.mark.anyio
async def test_non_dict_data_payload_is_rejected() -> None:
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"status": "success", "data": ["not-a-dict"]})
client = LightRAGClient(
base_url="http://lightrag.test",
api_key="lightrag-secret",
transport=httpx.MockTransport(handler),
)
with pytest.raises(LightRAGProtocolError, match="invalid retrieval result"):
await client.query_data("annual leave")

View File

@ -0,0 +1,419 @@
import logging
from pathlib import Path
from types import SimpleNamespace
from typing import Any
import pytest
import deerflow.community.lightrag.tools as lightrag_tools
from deerflow.community.lightrag.client import LightRAGAPIError, LightRAGConnectionError
from deerflow.community.lightrag.formatting import format_retrieval_result
from deerflow.config.tool_config import ToolConfig
from deerflow.tools.tools import get_available_tools
CHUNK_ID = "71a4613a-e91b-4d3b-bdff-6f45d9ac1f80"
def _data(
*,
chunks: list[dict[str, Any]] | None = None,
references: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
return {
"chunks": chunks if chunks is not None else [],
"references": references if references is not None else [],
"entities": [],
"relationships": [],
}
class FakeLightRAGClient:
def __init__(self, *, data: dict | None = None, error: Exception | None = None) -> None:
self.data = data if data is not None else _data()
self.error = error
self.query_calls: list[tuple[str, dict]] = []
async def query_data(self, query: str, **kwargs: object) -> dict:
if self.error is not None:
raise self.error
self.query_calls.append((query, kwargs))
return self.data
def _config(
*,
configured: bool = True,
api_key: str | None = "lightrag-secret",
base_url: str = "http://lightrag.test",
mode: str = "mix",
extra: dict[str, object] | None = None,
) -> SimpleNamespace:
settings: dict[str, object] = {
"base_url": base_url,
"api_key": api_key,
"mode": mode,
"timeout": 30,
"top_k": 60,
"max_chars_per_chunk": 800,
"max_total_chars": 8000,
}
settings.update(extra or {})
search_config = ToolConfig(
name="knowledge_search",
group="knowledge",
use="deerflow.community.lightrag.tools:knowledge_search_tool",
**settings,
)
return SimpleNamespace(
get_tool_config=lambda name: search_config if configured and name == "knowledge_search" else None,
)
def _install(monkeypatch: pytest.MonkeyPatch, fake: FakeLightRAGClient, *, config: SimpleNamespace | None = None) -> None:
monkeypatch.setattr(lightrag_tools, "get_app_config", lambda: config or _config())
monkeypatch.setattr(lightrag_tools, "_build_client", lambda settings: fake)
@pytest.mark.anyio
async def test_knowledge_search_formats_citation_numbered_chunks(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient(
data=_data(
chunks=[
{
"content": "Annual leave is based on years of service.",
"file_path": "documents/handbook.md",
"chunk_id": CHUNK_ID,
"reference_id": "3",
},
{
"content": "Sick leave requires a medical certificate.",
"file_path": "documents/handbook.md",
"chunk_id": "71a4613a-e91b-4d3b-bdff-6f45d9ac1f81",
"reference_id": "3",
},
],
references=[{"reference_id": "3", "file_path": "documents/handbook.md"}],
)
)
_install(monkeypatch, fake)
result = await lightrag_tools.knowledge_search("annual leave")
assert fake.query_calls == [("annual leave", {"mode": "mix", "top_k": 60, "chunk_top_k": None})]
assert "[1] documents/handbook.md\nAnnual leave is based on years of service." in result
assert "[2] documents/handbook.md\nSick leave requires a medical certificate." in result
assert "Matched documents: documents/handbook.md (2 chunks)" in result
assert CHUNK_ID not in result
@pytest.mark.anyio
async def test_knowledge_search_resolves_missing_chunk_file_path_from_references(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient(
data=_data(
chunks=[{"content": "Graph-reconstructed context.", "chunk_id": CHUNK_ID, "reference_id": "7"}],
references=[{"reference_id": "7", "file_path": "documents/graph-notes.md"}],
)
)
_install(monkeypatch, fake)
result = await lightrag_tools.knowledge_search("graph")
assert "[1] documents/graph-notes.md\nGraph-reconstructed context." in result
assert "Unknown document" not in result
@pytest.mark.anyio
async def test_knowledge_search_sends_configured_retrieval_settings(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient(data=_data(chunks=[{"content": "Guide.", "file_path": "docs/guide.md"}]))
_install(monkeypatch, fake, config=_config(mode="local", extra={"top_k": 12, "chunk_top_k": 6}))
result = await lightrag_tools.knowledge_search("guide")
assert fake.query_calls == [("guide", {"mode": "local", "top_k": 12, "chunk_top_k": 6})]
assert "Guide." in result
@pytest.mark.anyio
async def test_knowledge_search_works_without_api_key_for_unauthenticated_servers(
monkeypatch: pytest.MonkeyPatch,
) -> None:
fake = FakeLightRAGClient(data=_data(chunks=[{"content": "Open knowledge.", "file_path": "docs/open.md"}]))
_install(monkeypatch, fake, config=_config(api_key=None))
result = await lightrag_tools.knowledge_search("open")
assert "Open knowledge." in result
assert fake.query_calls
@pytest.mark.anyio
async def test_blank_api_key_is_treated_as_unauthenticated(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient(data=_data(chunks=[{"content": "Open knowledge.", "file_path": "docs/open.md"}]))
_install(monkeypatch, fake, config=_config(api_key=" "))
result = await lightrag_tools.knowledge_search("open")
assert "Open knowledge." in result
assert fake.query_calls
@pytest.mark.anyio
async def test_missing_knowledge_search_config_returns_english_guidance(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient()
_install(monkeypatch, fake, config=_config(configured=False))
result = await lightrag_tools.knowledge_search("leave")
assert result == "Error: knowledge_search is not configured; add its LightRAG settings to the tools list in config.yaml."
assert fake.query_calls == []
@pytest.mark.anyio
async def test_invalid_settings_return_english_guidance_without_leaking_values(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeLightRAGClient()
_install(monkeypatch, fake, config=_config(mode="vector"))
with caplog.at_level(logging.WARNING, logger="deerflow.community.lightrag.tools"):
result = await lightrag_tools.knowledge_search("leave")
assert result == "Error: Invalid LightRAG settings for knowledge_search; check config.yaml."
assert "vector" not in caplog.text
assert fake.query_calls == []
@pytest.mark.anyio
@pytest.mark.parametrize(
"base_url",
[
"http://lightrag-secret@lightrag.test",
"http://lightrag%2Dsecret@lightrag.test",
"http://user:lightrag-secret@lightrag.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 = FakeLightRAGClient()
_install(monkeypatch, fake, config=_config(base_url=base_url))
with caplog.at_level(logging.WARNING, logger="deerflow.community.lightrag.tools"):
result = await lightrag_tools.knowledge_search("leave")
assert result == "Error: Invalid LightRAG settings for knowledge_search; check config.yaml."
assert "lightrag-secret" not in result
assert "lightrag-secret" not in caplog.text
assert "lightrag%2Dsecret" not in caplog.text
assert fake.query_calls == []
@pytest.mark.anyio
async def test_empty_query_has_english_error(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient()
_install(monkeypatch, fake)
result = await lightrag_tools.knowledge_search(" ")
assert result == "Error: query must not be empty."
assert fake.query_calls == []
@pytest.mark.anyio
async def test_empty_retrieval_has_explicit_english_message(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient()
_install(monkeypatch, fake)
result = await lightrag_tools.knowledge_search("nothing")
assert result == "No relevant content found."
@pytest.mark.anyio
async def test_api_error_is_returned_as_readable_text_and_logged(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeLightRAGClient(error=LightRAGAPIError("RAG query is too short"))
_install(monkeypatch, fake)
with caplog.at_level(logging.WARNING, logger="deerflow.community.lightrag.tools"):
result = await lightrag_tools.knowledge_search("ab")
assert result == "Error: RAG query is too short"
assert "too short" in caplog.text
@pytest.mark.anyio
async def test_connection_error_is_english_and_does_not_leak_key(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
fake = FakeLightRAGClient(error=LightRAGConnectionError("ConnectError: refused lightrag-secret"))
_install(monkeypatch, fake)
with caplog.at_level(logging.WARNING, logger="deerflow.community.lightrag.tools"):
result = await lightrag_tools.knowledge_search("leave")
assert result == "Error: Unable to connect to LightRAG (http://lightrag.test): ConnectError: refused [REDACTED]"
assert "lightrag-secret" not in result
assert "lightrag-secret" not in caplog.text
@pytest.mark.anyio
async def test_success_path_still_redacts_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
fake = FakeLightRAGClient(data=_data(chunks=[{"content": "Accidental echo: lightrag-secret", "file_path": "docs/secret.md"}]))
_install(monkeypatch, fake)
result = await lightrag_tools.knowledge_search("secret")
assert "lightrag-secret" not in result
assert "[REDACTED]" in result
def test_formatting_uses_only_documented_chunk_fields() -> None:
result = format_retrieval_result(
{
"chunks": [
{
"content": "abcdefghij",
"file_path": "docs/policy.md",
"chunk_id": CHUNK_ID,
"reference_id": "1",
"unexpected_future_field": "ignored",
}
],
"references": [{"reference_id": "1", "file_path": "docs/policy.md"}],
},
max_chars_per_chunk=5,
max_total_chars=1000,
)
assert "[1] docs/policy.md" in result
assert "abcd…" in result
assert "abcdefghij" not in result
assert CHUNK_ID not in result
def test_formatting_without_chunk_file_path_or_reference_labels_unknown_document() -> None:
result = format_retrieval_result({"chunks": [{"content": "Orphan chunk.", "chunk_id": CHUNK_ID}]})
assert "[1] Unknown document\nOrphan chunk." in result
assert CHUNK_ID not in result
def test_formatting_applies_total_response_truncation_in_english() -> None:
result = format_retrieval_result(
{"chunks": [{"content": "content " * 20, "file_path": f"documents/document-{index}.md"} for index in range(4)]},
max_chars_per_chunk=100,
max_total_chars=120,
)
assert len(result) <= 120
assert result.endswith("… (response truncated)")
def test_retrieval_settings_load_provider_fields_and_hide_secret(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(lightrag_tools, "get_app_config", lambda: _config(mode="mix", extra={"top_k": 12, "chunk_top_k": 6}))
config, error = lightrag_tools._settings_or_error()
assert error is None
assert config is not None
assert str(config.base_url).rstrip("/") == "http://lightrag.test"
assert config.mode == "mix"
assert config.top_k == 12
assert config.chunk_top_k == 6
assert config.max_chars_per_chunk == 800
assert config.max_total_chars == 8000
assert "lightrag-secret" not in repr(config)
def test_retrieval_settings_allow_omitting_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(lightrag_tools, "get_app_config", lambda: _config(api_key=None))
config, error = lightrag_tools._settings_or_error()
assert error is None
assert config is not None
assert config.api_key is None
def test_agent_exposes_only_query_on_single_search_tool() -> None:
assert not hasattr(lightrag_tools, "list_knowledge_bases_tool")
assert not hasattr(lightrag_tools, "list_knowledge_bases")
assert lightrag_tools.knowledge_search_tool.name == "knowledge_search"
assert lightrag_tools.knowledge_search_tool.coroutine is not None
assert set(lightrag_tools.knowledge_search_tool.tool_call_schema.model_fields) == {"query"}
def test_tool_assembly_hides_credentials_without_network_io(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(lightrag_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.lightrag.tools:knowledge_search_tool",
base_url="http://lightrag.test",
api_key="lightrag-secret",
mode="hybrid",
)
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 "LightRAG" in assembled.description
assert "lightrag-secret" not in assembled.description
assert {tool.name for tool in tools}.isdisjoint({"list_knowledge_bases"})
def test_retrieval_settings_default_mode_matches_lightrag_request_default(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(lightrag_tools, "get_app_config", lambda: _config(mode="mix"))
config, error = lightrag_tools._settings_or_error()
assert error is None
assert config is not None
assert config.mode == "mix"
def test_shared_knowledge_search_name_keeps_first_configured_entry() -> None:
"""RAGFlow and LightRAG share the ``knowledge_search`` tool name.
This pins the config-layer behavior for that shared name so an operator
accidentally configuring both entries gets a documented outcome (the
first entry wins) instead of silent provider swapping.
"""
from deerflow.config.app_config import AppConfig
from deerflow.config.sandbox_config import SandboxConfig
ragflow_entry = ToolConfig(
name="knowledge_search",
group="knowledge",
use="deerflow.community.ragflow.tools:knowledge_search_tool",
)
lightrag_entry = ToolConfig(
name="knowledge_search",
group="knowledge",
use="deerflow.community.lightrag.tools:knowledge_search_tool",
)
config = AppConfig(tools=[ragflow_entry, lightrag_entry], sandbox=SandboxConfig(use="example.remote:Sandbox"))
assert config.get_tool_config("knowledge_search") is ragflow_entry
def test_lightrag_package_has_explicit_init_file() -> None:
package_dir = Path(lightrag_tools.__file__).resolve().parent
assert (package_dir / "__init__.py").is_file()

View File

@ -746,6 +746,28 @@ tools:
# max_chars_per_chunk: 800
# max_total_chars: 8000
# LightRAG knowledge retrieval (read-only). Alternative provider for the same
# knowledge_search tool; uncomment this entry instead of the RAGFlow one —
# duplicate names keep the first entry, so configure exactly one. Requires
# LightRAG v1.4.9+ (response envelope and citation fields). Searches the deployment's
# single indexed workspace through the data retrieval endpoint (no LLM
# generation inside LightRAG). `api_key` is optional; only omit it for
# loopback/trusted-network servers — when auth is enabled the key is sent as
# the X-API-Key header. `mode` picks the retrieval strategy: naive, local,
# global, hybrid, or mix (default mix, matching LightRAG's own default).
# Internal chunk and reference identifiers never reach the model.
# - name: knowledge_search
# group: knowledge
# use: deerflow.community.lightrag.tools:knowledge_search_tool
# base_url: http://localhost:9621 # Docker: use a backend-reachable URL
# api_key: $LIGHTRAG_API_KEY # Omit only for unauthenticated trusted servers
# mode: mix
# timeout: 30
# top_k: 60
# chunk_top_k: 8 # Optional; omit to use the server default
# max_chars_per_chunk: 800
# max_total_chars: 8000
# Web search tool (uses DuckDuckGo, no API key required)
- name: web_search
group: web

View File

@ -337,6 +337,25 @@ config: |
# top_k: 256
# max_chars_per_chunk: 800
# max_total_chars: 8000
# Optional tenant-shared, read-only LightRAG retrieval; alternative
# provider for the same knowledge_search tool (duplicate names keep the
# first entry, so configure exactly one). Requires LightRAG v1.4.9+. Put
# LIGHTRAG_API_KEY in `secrets` when the server enables API-key auth;
# omit `api_key` only for unauthenticated trusted-network deployments.
# Searches the single indexed workspace through the data-retrieval
# endpoint with no LLM generation inside LightRAG. Internal chunk and
# reference identifiers stay hidden from the Agent.
# - name: knowledge_search
# group: knowledge
# use: deerflow.community.lightrag.tools:knowledge_search_tool
# base_url: http://lightrag:9621
# api_key: $LIGHTRAG_API_KEY
# mode: mix
# timeout: 30
# top_k: 60
# chunk_top_k: 8 # Optional; omit to use the server default
# max_chars_per_chunk: 800
# max_total_chars: 8000
- name: web_search
group: web
use: deerflow.community.ddg_search.tools:web_search_tool