diff --git a/README.md b/README.md index e1cf4c1d55..98c732f2e8 100644 --- a/README.md +++ b/README.md @@ -1352,6 +1352,27 @@ establish that a package is valid or readable. The scope note remains visible even when there are no YAML warnings; check Gateway logs for other load failures. Parsing remains strict and invalid skills stay unavailable to agents. +#### Read-only knowledge sources + +Configure `knowledge_search` with either +`deerflow.community.ragflow.tools:knowledge_search_tool` or +`deerflow.community.lightrag.tools:knowledge_search_tool` in `config.yaml`. +Both support attributable retrieval evidence. LightRAG and RAGFlow responses +without complete provider locators use the shared DeerFlow source contract; +complete RAGFlow responses retain their existing citation snapshots and scope +controls. Knowledge citations show the +provider, document title, and retrieved excerpt in chat, including after reload +and when a subagent performed the retrieval. Missing page numbers and public +URLs remain absent; sources without a URL are readable as local evidence. +Inline citations remain readable after Markdown sanitization, and supplied page +numbers use the selected interface language. New providers use the shared +envelope; existing RAGFlow citation histories remain supported. +Expand a completed subtask card to inspect its report's forwarded knowledge +sources, including when reopening a saved conversation. +Existing web citations and the text-only Python `knowledge_search(query)` APIs +remain supported. No knowledge management service or page is required. +See [the knowledge contract guide](backend/docs/KNOWLEDGE.md). + Skills are what make DeerFlow do *almost anything*. A standard Agent Skill is a structured capability module — a Markdown file that defines a workflow, best practices, and references to supporting resources. DeerFlow ships with built-in skills for research, report generation, slide creation, web pages, image and video generation, and more. But the real power is extensibility: add your own skills, replace the built-in ones, or combine them into compound workflows. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index b53fe95583..af2eddabcb 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -168,10 +168,12 @@ environment with `uv run --no-sync --project backend` from the repository root. `--project` preserves the caller's working directory and relative runtime/config selectors; `--no-sync` retains installed extras such as PostgreSQL drivers. -More specific `AGENTS.md` files in backend code directories contain the subsystem sections split from this file. Follow the nearest file in the directory tree. +Read the nearest scoped `AGENTS.md` for subsystem ownership. ## Architecture +Read-only knowledge: [contract](docs/KNOWLEDGE.md). + ### Harness / App Split The backend is split into two layers with a strict dependency direction: diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index e959aaf1e4..47aa629a5a 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -393,11 +393,18 @@ def _strip_external_message_metadata(message: Any) -> Any: additional_kwargs.pop(ORIGINAL_USER_CONTENT_KEY, None) for key in _SERVER_OWNED_MESSAGE_METADATA_KEYS: additional_kwargs.pop(key, None) + from deerflow.knowledge.sources import strip_source_artifact + + artifact = getattr(message, "artifact", None) + clean_artifact = strip_source_artifact(artifact) if _is_human_message_like(message): additional_kwargs = _mark_untrusted_framework_markers(additional_kwargs, message.name) - if additional_kwargs == message.additional_kwargs: + if additional_kwargs == message.additional_kwargs and clean_artifact is artifact: return message - return message.model_copy(update={"additional_kwargs": additional_kwargs}) + update = {"additional_kwargs": additional_kwargs} + if clean_artifact is not artifact: + update["artifact"] = clean_artifact + return message.model_copy(update=update) def _strip_external_metadata_from_message_like(item: Any) -> Any: @@ -413,6 +420,12 @@ def _strip_external_metadata_from_message_like(item: Any) -> Any: return _strip_external_message_metadata(item) if not isinstance(item, dict): return item + from deerflow.knowledge.sources import strip_source_artifact + + artifact = item.get("artifact") + clean_artifact = strip_source_artifact(artifact) + if clean_artifact is not artifact: + item = {**item, "artifact": clean_artifact} # A missing (or non-dict) ``additional_kwargs`` is the most natural request # shape, and it still needs the mark: the messages reducer coerces the dict # with ``convert_to_messages``, which supplies ``additional_kwargs={}``, so an diff --git a/backend/docs/KNOWLEDGE.md b/backend/docs/KNOWLEDGE.md new file mode 100644 index 0000000000..169cc84be7 --- /dev/null +++ b/backend/docs/KNOWLEDGE.md @@ -0,0 +1,132 @@ +# Read-only knowledge contract (#6269, phase 1) + +`deerflow.knowledge` owns `RetrievalResult`, `RetrievalHit`, `KnowledgeSource`, +`KnowledgeCapabilities`, and the search-only `RetrievalProvider` protocol. +These depend on no knowledge management service, resource store, or catalog. +Existing RAGFlow scope admission and selection UI remain authoritative. +The shared contract does not add scope selection or WeKnora support to LightRAG. + +## Sources and citations + +A source requires `provider`, a real document `title`, and retrieved `excerpt`. +Optional `dataset_name`, HTTP(S) `url`, and positive integer `page` must come +from actual provider evidence. Never derive pages from chunk ordinals or URLs +from local file paths. Unattributed chunks remain usable text without citations. + +Opaque IDs hash the provider, title, excerpt, and supplied locators. Provider +dataset UUIDs, chunk IDs, and response-local reference IDs are not exposed in +citations. The model uses `[citation:Title](#knowledge-source-ID)` to refer to +the exact evidence. Both StructuredTools use `content_and_artifact`, with this +envelope on `ToolMessage.artifact`: + +```json +{ + "knowledge_sources_version": 1, + "knowledge_sources": [ + { + "id": "<32 lowercase hex characters>", + "provider": "lightrag", + "title": "docs/guide.md", + "excerpt": "Actual retrieved document text" + } + ] +} +``` + +Existing streaming, checkpoints, and run-event serialization preserve that +artifact. Readers require the envelope version, valid source fields, the source +handle, and actual excerpt in tool content. Backend readers recompute source IDs. +Assistant labels do not establish evidence. Gateway run input and checkpoint +writes strip the reserved source keys from caller artifacts. + +The reserved envelope keys are the shared reader's trust boundary, not the +tool name. Server-side producers that write these keys become evidence +authorities and must supply validated retrieval evidence; naming a tool +`knowledge_search` grants no authority to caller artifacts. This allows new +provider tools without extending a name allowlist. Native RAGFlow readers +continue to apply their existing tool-name allowlist. +The wire schema lives at `contracts/knowledge_sources.v1.json`; backend and +frontend tests pin the version, field bounds, and source IDs to that schema. + +Complete RAGFlow responses retain the native version-one +`knowledge_sources: {version, sources}` snapshots and `#knowledge-ID` links +already supported by the Gateway, frontend, ordinary tasks, and durable batch +exports. RAGFlow responses without complete provider locators fall back to the +shared contract above; LightRAG uses it directly. Both formats remain readable +after replay. When a task cites both formats, native RAGFlow snapshots stay in +`knowledge_sources`, while shared records live in `retrieval_sources` with +`knowledge_sources_version: 1`. Readers validate the same evidence and version; +Gateway strips all three keys from untrusted writes. Budget pruning must not +overwrite a native snapshot with shared records. + +The shared envelope is the intended default for new provider integrations. +Convergence should happen at adapter output boundaries: migrate newly produced +RAGFlow evidence to the shared form once native scope, budgets, delegation and +durable batch forwarding have equivalent coverage. Keep native readers for +persisted citations and exports; do not rewrite historical IDs or fabricate +missing locators. Removing native writers/readers requires a separate migration +proposal and compatibility tests, not a full migration in phase 1. + +The frontend resolves citations against preceding tool messages in loaded thread +history, displaying provider, title, excerpt, and only supplied locators. Absent +evidence stays unresolved until loaded. Historical web citations retain their +existing display. Old numbered knowledge text lacks structured evidence, so +replay never invents metadata for it. Direct Python text APIs keep their format. +Inline shared references accept both the raw `#knowledge-source-ID` and the +sanitized `#user-content-knowledge-source-ID`; the sanitizer's DOM-clobbering +protection stays enabled. Missing or malformed references render inert labels. + +`format_result` budgets chunk text and source evidence together: a cited hit +that cannot fit is omitted as a whole. Results display at most 100 hits, matching +the bounded artifact readers. Only displayed excerpts enter artifacts. +The global tool-output budget prunes evidence removed by truncation or +externalization and replaces complete orphan citations with their labels. +Subagents accumulate at most 100 validated sources across values frames before +terminal publication, surviving compaction. Ordinary `task` forwards only +sources whose handles appear in the child report, with a bounded evidence +appendix. Existing native RAGFlow durable batch forwarding is retained; +shared-contract durable batch forwarding is outside this phase. +The accumulator scans only the changed history suffix, restarting at changed +prefixes after compaction; null messages are treated as empty frames. The +frontend caches validation by immutable message object in a WeakMap, so repeated +stream updates do not rescan unchanged tool excerpts. + +## Adapters and capabilities + +Both tools advertise `search` and `sources` in +`metadata.knowledge_capabilities.operations`. Optional `catalog`, `documents`, +`dataset_filter`, and `document_filter` describe separate operations. Search +support implies none of them. These metadata fields do not replace existing +RAGFlow scope admission or change operator-configured retrieval scope. + +RAGFlow maps normalized chunk content, document title, dataset name, and +`doc_aggs` title fallback. Dataset resolution, embedding-group ordering, access +errors, and credential redaction retain their existing behavior. + +LightRAG consumes `/query/data` document chunks, joining missing `file_path` +through `chunks[].reference_id` and `references[].reference_id`. Entities and +relationships alone cannot produce document citations. Missing attribution and +`unknown_source` remain unattributed. The response fields were checked against +LightRAG v1.5.7, commit `28ff1b05f2ac3f3e6fa14dd2cd33656579bd0c9c`, in +[`convert_to_user_format` and `generate_reference_list_from_chunks`](https://github.com/HKUDS/LightRAG/blob/28ff1b05f2ac3f3e6fa14dd2cd33656579bd0c9c/lightrag/utils.py). +Neither adapter fabricates page numbers or public document URLs. + +## Validation + +Offline tests use explicitly synthetic API-shaped responses for real +StructuredTool execution, budgets, credential redaction, delegation, replay, +caller forgery, and frontend display. These are not live provider responses. + +Verify actual retrieved evidence from an existing indexed LightRAG workspace +with the read-only opt-in check from `backend/`: + +```bash +DEER_FLOW_RUN_LIVE_TESTS=1 \ +LIGHTRAG_BASE_URL=http://localhost:9621 \ +LIGHTRAG_EVIDENCE_QUERY='a question matching an indexed document' \ +uv run pytest tests/test_knowledge_contract_live.py -m live -q +``` + +Set `LIGHTRAG_API_KEY` in the environment if required. This test performs no +upload or resource changes. It requires attributable hits and checks each +citation against the real returned chunk evidence. diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py index e330c7646b..3b9082e770 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py @@ -719,7 +719,7 @@ def _patch_tool_message( trigger = _effective_trigger(tool_name, config) citation_result = None if tool_name in {"knowledge_search", "task"} and trigger > 0 and len(text) > trigger: - citation_result = budget_source_artifact(text, msg.artifact, trigger, summary=(budgeted[0] if budgeted else text) if tool_name == "task" else "") + citation_result = budget_source_artifact(text, msg.artifact, trigger, summary=(budgeted.replacement if budgeted else text) if tool_name == "task" else "") if citation_result is not None: replacement, update["artifact"] = citation_result transform_kind = "truncated" @@ -728,6 +728,11 @@ def _patch_tool_message( transform_kind = budgeted.transform_kind else: return msg + from deerflow.knowledge.sources import prune_sources + + replacement, artifact = prune_sources(replacement, update.get("artifact", msg.artifact)) + if artifact is not msg.artifact or "artifact" in update: + update["artifact"] = artifact update["content"] = [{"type": "text", "text": replacement}, *retained_blocks] if retained_blocks else replacement if getattr(msg, "response_metadata", None): update["response_metadata"] = dict(msg.response_metadata) diff --git a/backend/packages/harness/deerflow/community/lightrag/formatting.py b/backend/packages/harness/deerflow/community/lightrag/formatting.py index f6083074d8..cb6e5d4ea4 100644 --- a/backend/packages/harness/deerflow/community/lightrag/formatting.py +++ b/backend/packages/harness/deerflow/community/lightrag/formatting.py @@ -5,13 +5,29 @@ from __future__ import annotations from collections.abc import Mapping from typing import Any +from deerflow.knowledge import KnowledgeCapabilities, KnowledgeSource, RetrievalHit, RetrievalResult +from deerflow.knowledge.sources import truncate as _truncate -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}" +CAPABILITIES = KnowledgeCapabilities(operations=frozenset({"search", "sources"})) + + +def normalize_retrieval_result(result: Mapping[str, Any]) -> RetrievalResult: + """Use /query/data chunks and joined references, never graph summaries.""" + references = _reference_file_paths(result.get("references")) + hits = [] + for chunk in _chunks(result.get("chunks")): + content = chunk.get("content") + if not isinstance(content, str) or not content.strip(): + continue + content = content.strip()[:100_000] + title = chunk.get("file_path") + if not isinstance(title, str) or not title.strip(): + title = references.get(str(chunk.get("reference_id"))) + source = None + if isinstance(title, str) and title.strip() and title.strip().lower() not in {"unknown", "unknown_source", "unknown document"}: + source = KnowledgeSource(provider="lightrag", title=title.strip()[:500], excerpt=content) + hits.append(RetrievalHit(content=content, source=source)) + return RetrievalResult(hits=hits) def _chunks(value: object) -> list[Mapping[str, Any]]: diff --git a/backend/packages/harness/deerflow/community/lightrag/tools.py b/backend/packages/harness/deerflow/community/lightrag/tools.py index 0772ab2513..b980a41f20 100644 --- a/backend/packages/harness/deerflow/community/lightrag/tools.py +++ b/backend/packages/harness/deerflow/community/lightrag/tools.py @@ -11,9 +11,11 @@ from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, SecretStr, Valida from deerflow.community.provider_credentials import redact_secret, reject_url_userinfo, resolve_api_key from deerflow.config import get_app_config +from deerflow.knowledge import format_result +from deerflow.knowledge.sources import redact_strings, source_artifact from .client import LightRAGAPIError, LightRAGClient, LightRAGConnectionError, LightRAGProtocolError -from .formatting import format_retrieval_result +from .formatting import CAPABILITIES, format_retrieval_result, normalize_retrieval_result logger = logging.getLogger(__name__) @@ -94,7 +96,7 @@ def _tool_error(exc: Exception, settings: _LightRAGRetrievalSettings) -> str: return "Error: An unexpected LightRAG retrieval error occurred; try again later." -async def knowledge_search(query: str) -> str: +async def _search(query: str, *, with_sources: bool = False) -> str | tuple[str, dict]: """Search the operator-configured LightRAG instance. LightRAG has no dataset catalog to scope: the deployment's single indexed @@ -117,6 +119,9 @@ async def knowledge_search(query: str) -> str: top_k=settings.top_k, chunk_top_k=settings.chunk_top_k, ) + if with_sources: + normalized = normalize_retrieval_result(redact_strings(result, _api_key(settings))) + return format_result(normalized, max_chars_per_chunk=settings.max_chars_per_chunk, max_total_chars=settings.max_total_chars) formatted = format_retrieval_result( result, max_chars_per_chunk=settings.max_chars_per_chunk, @@ -129,18 +134,27 @@ async def knowledge_search(query: str) -> str: return _tool_error(exc, settings) +async def knowledge_search(query: str) -> str: + """Legacy text-only API; graph tool calls additionally carry source artifacts.""" + result = await _search(query) + assert isinstance(result, str) + return result + + 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." + base += " Cite only returned evidence using the exact [citation:Title](#knowledge-source-ID) links supplied with each chunk; never invent source locators." return f"{base} Internal identifiers are never shown to the model." -async def _knowledge_search_entrypoint(query: str) -> str: +async def _knowledge_search_entrypoint(query: str) -> tuple[str, dict]: """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) + result = await _search(query, with_sources=True) + return result if isinstance(result, tuple) else (result, source_artifact([])) knowledge_search_tool = StructuredTool.from_function( @@ -148,4 +162,6 @@ knowledge_search_tool = StructuredTool.from_function( name="knowledge_search", description=_tool_description(), parse_docstring=True, + response_format="content_and_artifact", + metadata={"knowledge_capabilities": CAPABILITIES.model_dump(mode="json")}, ) diff --git a/backend/packages/harness/deerflow/community/ragflow/formatting.py b/backend/packages/harness/deerflow/community/ragflow/formatting.py index ce2bf20614..80a09844a6 100644 --- a/backend/packages/harness/deerflow/community/ragflow/formatting.py +++ b/backend/packages/harness/deerflow/community/ragflow/formatting.py @@ -2,16 +2,35 @@ from __future__ import annotations +import math from collections.abc import Callable, Mapping from typing import Any +from deerflow.knowledge import KnowledgeCapabilities, KnowledgeSource, RetrievalHit, RetrievalResult +from deerflow.knowledge.sources import truncate as _truncate -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}" +CAPABILITIES = KnowledgeCapabilities(operations=frozenset({"search", "sources"})) + + +def normalize_retrieval_result(result: Mapping[str, Any], *, dataset_names_by_id: Mapping[str, str]) -> RetrievalResult: + """Map public retrieval fields to attributable evidence, without locators.""" + aggregates = _document_aggregates(result.get("doc_aggs")) + names = {str(item["doc_id"]): item["doc_name"] for item in aggregates if item.get("doc_id") and isinstance(item.get("doc_name"), str)} + hits = [] + chunks = result.get("chunks") + for chunk in chunks if isinstance(chunks, list) else []: + if not isinstance(chunk, Mapping) or not isinstance(chunk.get("content"), str): + continue + content = chunk["content"].strip() + if not content: + continue + title = chunk.get("document_keyword") or names.get(str(chunk.get("document_id"))) + source = None + if isinstance(title, str) and title.strip(): + source = KnowledgeSource(provider="ragflow", title=title.strip()[:500], excerpt=content[:100_000], dataset_name=dataset_names_by_id.get(str(chunk.get("dataset_id")))) + score = _score(chunk.get("similarity")) + hits.append(RetrievalHit(content=content[:100_000], source=source, score=score if score is not None and math.isfinite(score) else None)) + return RetrievalResult(hits=hits) def _document_aggregates(value: object) -> list[Mapping[str, Any]]: diff --git a/backend/packages/harness/deerflow/community/ragflow/tools.py b/backend/packages/harness/deerflow/community/ragflow/tools.py index 2ce4e65cf3..5aa6784059 100644 --- a/backend/packages/harness/deerflow/community/ragflow/tools.py +++ b/backend/packages/harness/deerflow/community/ragflow/tools.py @@ -14,6 +14,8 @@ from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, SecretStr, Valida from deerflow.community.provider_credentials import redact_secret, reject_url_userinfo, resolve_api_key from deerflow.config import get_app_config +from deerflow.knowledge import format_result +from deerflow.knowledge.sources import redact_strings from deerflow.knowledge_scope import ( KNOWLEDGE_SCOPE_RUNTIME_KEY, canonicalize_knowledge_scope, @@ -22,7 +24,7 @@ from deerflow.knowledge_scope import ( from deerflow.tools.types import Runtime from .client import RAGFlowAPIError, RAGFlowClient, RAGFlowConnectionError, RAGFlowProtocolError -from .formatting import format_retrieval_result, format_retrieval_sources +from .formatting import CAPABILITIES, format_retrieval_result, format_retrieval_sources, normalize_retrieval_result logger = logging.getLogger(__name__) @@ -565,6 +567,9 @@ async def knowledge_search( max_total_chars=settings.max_total_chars, redact=lambda value: redact_secret(value, _api_key(settings)), ) + if artifact is None: + normalized = normalize_retrieval_result(redact_strings(result, _api_key(settings)), dataset_names_by_id=redact_strings(names_by_id, _api_key(settings))) + content, artifact = format_result(normalized, max_chars_per_chunk=settings.max_chars_per_chunk, max_total_chars=settings.max_total_chars) if artifact is not None: _source_artifact.update(artifact) return content @@ -628,6 +633,7 @@ knowledge_search_tool = StructuredTool.from_function( response_format="content_and_artifact", description=_tool_description(), parse_docstring=True, + metadata={"knowledge_capabilities": CAPABILITIES.model_dump(mode="json")}, ) diff --git a/backend/packages/harness/deerflow/knowledge/__init__.py b/backend/packages/harness/deerflow/knowledge/__init__.py new file mode 100644 index 0000000000..9904987770 --- /dev/null +++ b/backend/packages/harness/deerflow/knowledge/__init__.py @@ -0,0 +1,6 @@ +"""Read-only knowledge contracts; no catalog or management dependency.""" + +from .contract import KnowledgeCapabilities, KnowledgeSource, RetrievalHit, RetrievalProvider, RetrievalResult +from .sources import format_result, sources_from_messages + +__all__ = ["KnowledgeCapabilities", "KnowledgeSource", "RetrievalHit", "RetrievalProvider", "RetrievalResult", "format_result", "sources_from_messages"] diff --git a/backend/packages/harness/deerflow/knowledge/contract.py b/backend/packages/harness/deerflow/knowledge/contract.py new file mode 100644 index 0000000000..bd176084e1 --- /dev/null +++ b/backend/packages/harness/deerflow/knowledge/contract.py @@ -0,0 +1,68 @@ +"""Provider-neutral retrieval evidence and operation-level capabilities.""" + +from __future__ import annotations + +import hashlib +import json +from typing import Literal, Protocol + +from pydantic import BaseModel, ConfigDict, Field, HttpUrl, computed_field, field_validator + + +class KnowledgeCapabilities(BaseModel): + """An operation is supported only when explicitly advertised. + + Search-only providers need no catalog, scope, or management implementation. + Scope admission and adapters for these optional operations belong to phase 2. + """ + + operations: frozenset[Literal["search", "sources", "catalog", "documents", "dataset_filter", "document_filter"]] = frozenset({"search"}) + + def supports(self, operation: str) -> bool: + return operation in self.operations + + +class KnowledgeSource(BaseModel): + """One attributable excerpt actually returned by a retrieval tool. + + Titles and optional locators are provider evidence, never inferred links, + page numbers, or local reference ordinals. IDs expose no provider internals. + """ + + model_config = ConfigDict(frozen=True, extra="ignore", str_strip_whitespace=True) + + provider: str = Field(min_length=1, max_length=100) + title: str = Field(min_length=1, max_length=500) + excerpt: str = Field(min_length=1, max_length=100_000) + dataset_name: str | None = Field(default=None, max_length=500) + url: HttpUrl | None = None + page: int | None = Field(default=None, gt=0, le=9_007_199_254_740_991, strict=True) + + @field_validator("url") + @classmethod + def reject_credentials(cls, value: HttpUrl | None) -> HttpUrl | None: + if value is not None and (value.username is not None or value.password is not None): + raise ValueError("source URL must not contain credentials") + return value + + @computed_field + @property + def id(self) -> str: + identity = [self.provider, self.title, self.excerpt, self.dataset_name, str(self.url) if self.url else None, self.page] + return hashlib.sha256(json.dumps(identity, ensure_ascii=False).encode("utf-8")).hexdigest()[:32] + + +class RetrievalHit(BaseModel): + content: str + source: KnowledgeSource | None = None + score: float | None = Field(default=None, allow_inf_nan=False) + + +class RetrievalResult(BaseModel): + hits: list[RetrievalHit] = Field(default_factory=list) + + +class RetrievalProvider(Protocol): + capabilities: KnowledgeCapabilities + + async def search(self, query: str) -> RetrievalResult: ... diff --git a/backend/packages/harness/deerflow/knowledge/sources.py b/backend/packages/harness/deerflow/knowledge/sources.py new file mode 100644 index 0000000000..e35d3a8730 --- /dev/null +++ b/backend/packages/harness/deerflow/knowledge/sources.py @@ -0,0 +1,155 @@ +"""Shared validation, evidence-preserving budgets, and delegation forwarding.""" + +from __future__ import annotations + +import re +from collections.abc import Iterable, Mapping +from typing import Any + +from pydantic import ValidationError + +from .contract import KnowledgeSource, RetrievalResult + +ARTIFACT_VERSION = 1 +TRUNCATION_MARKER = "… (response truncated)" + + +def redact_strings(value: Any, secret: str | None) -> Any: + """Redact before normalization so text and artifact never expose secrets.""" + if isinstance(value, str): + return value.replace(secret, "[REDACTED]") if secret else value + if isinstance(value, Mapping): + return {key: redact_strings(item, secret) for key, item in value.items()} + if isinstance(value, list): + return [redact_strings(item, secret) for item in value] + return value + + +def strip_source_artifact(artifact: Any) -> Any: + """Remove only server-owned evidence keys from untrusted writes.""" + if isinstance(artifact, Mapping) and ("knowledge_sources" in artifact or "knowledge_sources_version" in artifact or "retrieval_sources" in artifact): + return {key: value for key, value in artifact.items() if key not in {"knowledge_sources", "knowledge_sources_version", "retrieval_sources"}} + return artifact + + +def prune_sources(content: str, artifact: Any) -> tuple[str, Any]: + """Budget transformations retain only evidence still visible to the model.""" + if not isinstance(artifact, Mapping) or artifact.get("knowledge_sources_version") != ARTIFACT_VERSION: + return content, artifact + sources = sources_from_messages([{"type": "tool", "content": content, "artifact": artifact}]) + allowed = {source.id for source in sources} + pattern = r"\[citation:((?:\\.|[^\]\\])*)\]\(#knowledge-source-([a-f0-9]{32})\)" + content = re.sub(pattern, lambda match: match[0] if match[2] in allowed else match[1], content) + if "retrieval_sources" in artifact: + return content, {**artifact, "retrieval_sources": [source.model_dump(mode="json", exclude_none=True) for source in sources]} + return content, {**artifact, **source_artifact(sources)} + + +def truncate(value: str, max_chars: int, *, marker: str = "…") -> str: + max_chars = max(0, max_chars) + 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 source_artifact(sources: Iterable[KnowledgeSource]) -> dict[str, Any]: + return {"knowledge_sources_version": ARTIFACT_VERSION, "knowledge_sources": [source.model_dump(mode="json", exclude_none=True) for source in sources]} + + +def citation(source: KnowledgeSource) -> str: + # Escape labels before inserting provider-controlled titles into markdown. + title = source.title.replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]").replace("\n", " ") + return f"[citation:{title}](#knowledge-source-{source.id})" + + +def format_result(result: RetrievalResult, *, max_chars_per_chunk: int = 800, max_total_chars: int = 8000) -> tuple[str, dict[str, Any]]: + """Budget text and its evidence together; never cut a citation in half.""" + entries: list[str] = [] + sources: list[KnowledgeSource] = [] + used_chars = 0 + budget = max(0, max_total_chars) + truncated = False + for index, hit in enumerate(result.hits, 1): + if index > 100: + truncated = True + break # Match the bounded artifact readers and delegation ledger. + content = truncate(hit.content.strip(), max_chars_per_chunk) + if not content: + continue + source = hit.source + # Evidence must be from this hit. Adapters cannot attach unrelated text. + if source is not None and source.excerpt != hit.content.strip(): + source = None + if content == "…" and len(hit.content.strip()) > len(content): + source = None # A truncation marker alone is not retrieved evidence. + if source is not None: + source = KnowledgeSource.model_validate({**source.model_dump(exclude={"id"}), "excerpt": content}) + header = f"[{index}] {citation(source) if source else 'Unattributed content'}\n" + separator = "\n\n" if entries else "" + remaining = budget - used_chars - len(separator) + entry = header + content + if len(entry) > remaining: + # Drop the entire hit when it cannot fit with real evidence. This + # keeps the stable source identity independent of total budget. + truncated = True + break + entries.append(separator + entry) + used_chars += len(separator) + len(entry) + if source is not None: + sources.append(source) + text = "".join(entries) + if truncated: + remaining = budget - len(text) + text += truncate(TRUNCATION_MARKER, remaining, marker="") + elif not text: + text = truncate("No relevant content found.", budget, marker="") + return text, source_artifact(sources) + + +def sources_from_messages(messages: Iterable[Any]) -> list[KnowledgeSource]: + """Accept native and serialized ToolMessages, never assistant claims. + + Child values accumulation scans changed message suffixes and retains bounded + evidence across compaction. Replay deduplicates by evidence identity. + """ + sources: dict[str, KnowledgeSource] = {} + for message in messages: + if isinstance(message, Mapping): + kind, artifact, content = message.get("type"), message.get("artifact"), message.get("content") + else: + kind, artifact, content = getattr(message, "type", None), getattr(message, "artifact", None), getattr(message, "content", None) + if kind != "tool" or not isinstance(content, str) or not isinstance(artifact, Mapping) or type(artifact.get("knowledge_sources_version")) is not int or artifact.get("knowledge_sources_version") != ARTIFACT_VERSION: + continue + raw_sources = artifact.get("knowledge_sources") if isinstance(artifact.get("knowledge_sources"), list) else artifact.get("retrieval_sources") + if not isinstance(raw_sources, list): + continue + for raw in raw_sources[:100]: + try: + source = KnowledgeSource.model_validate(raw) + except (ValidationError, TypeError): + continue + if not isinstance(raw, Mapping) or raw.get("id") != source.id: + continue + if source.excerpt in content and f"#knowledge-source-{source.id}" in content: + sources[source.id] = source + return list(sources.values())[-100:] + + +def forward_sources(report: str, raw_sources: Iterable[Any], *, max_total_chars: int = 8000) -> tuple[str, dict[str, Any]]: + """Forward only citations backed by child tool evidence, with excerpts.""" + hits = [] + from .contract import RetrievalHit + + for raw in raw_sources: + try: + source = KnowledgeSource.model_validate(raw) + except (ValidationError, TypeError): + continue + if isinstance(raw, Mapping) and raw.get("id") == source.id and f"#knowledge-source-{source.id}" in report: + hits.append(RetrievalHit(content=source.excerpt, source=source)) + if not hits: + return prune_sources(report, source_artifact([])) + evidence, artifact = format_result(RetrievalResult(hits=hits), max_chars_per_chunk=100_000, max_total_chars=max_total_chars) + return prune_sources(f"{report}\n\nKnowledge evidence:\n{evidence}", artifact) diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 541334d49f..3befa99dea 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -1,5 +1,14 @@ ### Subagent System (`packages/harness/deerflow/subagents/`) +Validated ToolMessage knowledge evidence accumulates across values frames in +`SubagentResult.knowledge_sources` (100 sources, deduplicated by evidence ID). +Publication precedes terminal state and survives compaction. Ordinary `task` +forwards reported handles with a bounded evidence appendix and source artifact. +Mixed task results keep native RAGFlow snapshots in `knowledge_sources` and +shared evidence in `retrieval_sources`; pruning must preserve the native snapshot. +Shared evidence scans start at the changed message suffix; compaction or prefix +replacement resets that boundary. Null frames must not crash the accumulator. + Result-reader acceptance runs on SQLite and, with TEST_POSTGRES_URI, an isolated PostgreSQL schema; both exercise submission, completion, reopen and ToolNode continuation without paid model work. Always clean up the temporary schema. diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 765007a699..efd680bdfb 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -173,6 +173,8 @@ class SubagentResult: admission_failure: bool = False tool_receipts: list[dict[str, Any]] | None = field(default=None, kw_only=True) bash_executions: list[dict[str, Any]] | None = field(default=None, kw_only=True) + knowledge_sources: list[dict[str, Any]] = field(default_factory=list, kw_only=True) + _knowledge_messages: list[Any] = field(default_factory=list, init=False, repr=False) cancel_event: threading.Event = field(default_factory=threading.Event, repr=False) _state_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) @@ -224,6 +226,29 @@ class SubagentResult: return None return [dict(receipt) for receipt in self.tool_receipts] + def update_knowledge_sources(self, messages: list[Any] | None) -> None: + """Accumulate validated retrieval evidence before terminal publication.""" + from deerflow.knowledge import sources_from_messages + + with self._state_lock: + if self.status.is_terminal: + return + if messages is None: + return + # Values histories grow cumulatively, but compaction/replacements + # can change the prefix. Rescan from the first changed message. + start = 0 + for previous, current in zip(self._knowledge_messages, messages): + if previous != current: + break + start += 1 + sources = sources_from_messages(messages[start:]) if start < len(messages) else [] + self._knowledge_messages = list(messages) + merged = {source["id"]: source for source in self.knowledge_sources} + for source in sources: + merged[source.id] = source.model_dump(mode="json", exclude_none=True) + self.knowledge_sources = list(merged.values())[-100:] + def try_set_terminal( self, status: SubagentStatus, @@ -1826,6 +1851,7 @@ class SubagentExecutor: # harvesting includes a tool result that completed while the # cancellation request was in flight. final_state = chunk + result.update_knowledge_sources(chunk.get("messages") or []) result.update_tool_receipts(terminal_receipts()) result.update_bash_executions(current_bash_executions()) @@ -1845,7 +1871,7 @@ class SubagentExecutor: # several ToolMessages when the model emits multiple tool calls in # one turn, so capturing only messages[-1] would drop all but the # last output (#3779). Dedup/serialization live in capture_step_message. - messages = chunk.get("messages", []) + messages = chunk.get("messages") or [] previous_count = len(ai_messages) processed_message_count = capture_new_step_messages(messages, ai_messages, seen_message_ids, processed_message_count) if len(ai_messages) > previous_count: diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index a3e38998a6..242ee12da3 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -635,12 +635,23 @@ def _task_result_command( source_messages: list[dict] | None = None, receipt_verdict: dict | None = None, acceptance_verdict: dict | None = None, + knowledge_sources: list[dict] | None = None, ) -> Command: content, metadata_error = format_subagent_result_message(status, result=result, error=error, stop_reason=stop_reason) if acceptance_verdict is not None: # RFC #4651 PR4: the rendered checklist rides the model-visible result # text; metadata carries the structured verdict for the ledger/judge. content = f"{content}\n\n{render_acceptance_section(acceptance_verdict)}" + artifact = cited_source_artifact(source_messages or [], content, include_omissions=True) + if knowledge_sources: + from deerflow.knowledge.sources import forward_sources + + content, unified_artifact = forward_sources(content, knowledge_sources) + if artifact is None: + artifact = unified_artifact + else: + # Preserve the native RAGFlow envelope for existing history and batch readers. + artifact = {**artifact, "retrieval_sources": unified_artifact["knowledge_sources"], "knowledge_sources_version": 1} return Command( update={ "messages": [ @@ -648,7 +659,7 @@ def _task_result_command( content=content, tool_call_id=tool_call_id, name="task", - artifact=cited_source_artifact(source_messages or [], content, include_omissions=True), + artifact=artifact, additional_kwargs=make_subagent_additional_kwargs( status, result=result, @@ -1129,6 +1140,7 @@ async def task_tool( source_messages=getattr(result, "ai_messages", None), receipt_verdict=receipt_verdict, acceptance_verdict=acceptance_verdict, + knowledge_sources=getattr(result, "knowledge_sources", None), ) elif result.status == SubagentStatus.FAILED: _report_subagent_usage(runtime, result) diff --git a/backend/tests/test_knowledge_contract.py b/backend/tests/test_knowledge_contract.py new file mode 100644 index 0000000000..6c7344325b --- /dev/null +++ b/backend/tests/test_knowledge_contract.py @@ -0,0 +1,196 @@ +"""Provider-independent evidence and budget regressions (offline).""" + +import json +from pathlib import Path + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from pydantic import ValidationError + +from deerflow.knowledge import KnowledgeSource, RetrievalHit, RetrievalResult, format_result, sources_from_messages + + +def test_source_schema_matches_cross_component_contract(): + from jsonschema import Draft202012Validator + + from deerflow.knowledge.sources import ARTIFACT_VERSION + + schema = json.loads((Path(__file__).parents[2] / "contracts/knowledge_sources.v1.json").read_text(encoding="utf-8")) + Draft202012Validator.check_schema(schema) + properties = schema["$defs"]["source"]["properties"] + model = KnowledgeSource.model_json_schema(mode="serialization")["properties"] + for name in ("provider", "title", "excerpt"): + assert model[name]["maxLength"] == properties[name]["maxLength"] + assert model[name]["minLength"] == properties[name]["minLength"] + assert model["dataset_name"]["anyOf"][0]["maxLength"] == properties["dataset_name"]["maxLength"] + assert model["page"]["anyOf"][0]["maximum"] == properties["page"]["maximum"] + assert ARTIFACT_VERSION == schema["properties"]["knowledge_sources_version"]["const"] + source = KnowledgeSource(provider="demo", title="Guide", excerpt="evidence") + _, artifact = format_result(RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)])) + Draft202012Validator(schema).validate(artifact) + + +def test_sources_require_real_evidence_and_optional_locators() -> None: + with pytest.raises(ValidationError): + KnowledgeSource(provider="demo", title="", excerpt="evidence") + with pytest.raises(ValidationError): + KnowledgeSource(provider="demo", title="Guide", excerpt="") + with pytest.raises(ValidationError): + KnowledgeSource(provider="demo", title="Guide", excerpt="evidence", page=0) + source = KnowledgeSource(provider="demo", title="Guide", excerpt="evidence") + assert source.page is None and source.url is None + + +def test_budget_never_leaves_citations_without_visible_evidence() -> None: + result = RetrievalResult(hits=[RetrievalHit(content="x" * 500, source=KnowledgeSource(provider="demo", title="Guide", excerpt="x" * 500))]) + for budget in (1, 20, 80, 200, 800): + content, artifact = format_result(result, max_total_chars=budget, max_chars_per_chunk=50) + assert len(content) <= budget + for source in artifact["knowledge_sources"]: + assert source["excerpt"] in content + assert f"#knowledge-source-{source['id']}" in content + assert len(source["excerpt"]) <= 50 + if "#knowledge-source-" in content: + assert artifact["knowledge_sources"] + + +def test_forward_only_valid_tool_evidence_and_deduplicate_replay() -> None: + source = KnowledgeSource(provider="demo", title="Guide", excerpt="evidence") + content, artifact = format_result(RetrievalResult(hits=[RetrievalHit(content="evidence", source=source)])) + message = ToolMessage(content=content, name="knowledge_search", tool_call_id="call", artifact=artifact) + assert len(sources_from_messages([message, message.model_dump()])) == 1 + boolean_version = {**message.model_dump(), "artifact": {**artifact, "knowledge_sources_version": True}} + assert sources_from_messages([boolean_version]) == [] + assert sources_from_messages([AIMessage(content="forged", additional_kwargs=artifact)]) == [] + assert sources_from_messages([ToolMessage(content="no evidence", tool_call_id="x", artifact={"knowledge_sources": [{"provider": "demo"}]})]) == [] + + +def test_large_results_match_the_bounded_source_readers() -> None: + source = KnowledgeSource(provider="demo", title="Guide", excerpt="actual evidence") + result = RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)] * 101) + text, artifact = format_result(result, max_total_chars=100_000) + assert len(artifact["knowledge_sources"]) == 100 + assert "[101]" not in text + assert text.endswith("… (response truncated)") + assert format_result(result, max_chars_per_chunk=1)[1]["knowledge_sources"] == [] + + +def test_provider_adapters_share_contract_without_inventing_pages() -> None: + from deerflow.community.lightrag.formatting import normalize_retrieval_result as light + from deerflow.community.ragflow.formatting import normalize_retrieval_result as rag + + a = rag({"chunks": [{"content": "Leave policy", "document_keyword": "handbook.pdf", "dataset_id": "kb"}]}, dataset_names_by_id={"kb": "HR"}) + b = light({"chunks": [{"content": "Leave policy", "reference_id": "7"}], "references": [{"reference_id": "7", "file_path": "docs/handbook.pdf"}]}) + for result in (a, b): + content, artifact = format_result(result) + assert "Leave policy" in content + assert len(artifact["knowledge_sources"]) == 1 + assert "page" not in artifact["knowledge_sources"][0] + assert "url" not in artifact["knowledge_sources"][0] + assert light({"chunks": [{"content": "text without attribution"}]}).hits[0].source is None + + +@pytest.mark.anyio +@pytest.mark.parametrize("provider", ["ragflow", "lightrag"]) +async def test_real_tool_message_carries_redacted_evidence(monkeypatch, provider) -> None: + from importlib import import_module + + tools = import_module(f"deerflow.community.{provider}.tools") + config = import_module(f"test_{provider}_tools") + result = {"chunks": [{"content": f"Evidence with {provider}-secret", "document_keyword": "guide.pdf", "file_path": "docs/guide.md", "dataset_id": config.DATASET_ID_1 if provider == "ragflow" else "unused"}]} + if provider == "ragflow": + fake = config.FakeRAGFlowClient(datasets_by_id={config.DATASET_ID_1: [config._dataset(config.DATASET_ID_1, "HR")]}, retrieval=result) + else: + fake = config.FakeLightRAGClient(data=result) + config._install(monkeypatch, fake) + from langgraph.graph import END, START, MessagesState, StateGraph + from langgraph.prebuilt import ToolNode + + builder = StateGraph(MessagesState) + builder.add_node("tools", ToolNode([tools.knowledge_search_tool])) + builder.add_edge(START, "tools") + builder.add_edge("tools", END) + state = await builder.compile().ainvoke({"messages": [AIMessage(content="", tool_calls=[{"id": "call", "name": "knowledge_search", "args": {"query": "evidence"}}])]}) + message = state["messages"][-1] + assert isinstance(message, ToolMessage) + sources = sources_from_messages([message]) + assert len(sources) == 1 + assert sources[0].provider == provider + assert f"{provider}-secret" not in str(message.model_dump()) + + +def test_forwarding_only_includes_reported_sources() -> None: + from deerflow.tools.builtins.task_tool import _task_result_command + + source = KnowledgeSource(provider="demo", title="Guide", excerpt="real evidence") + content, artifact = format_result(RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)])) + message = ToolMessage(content=content, name="knowledge_search", tool_call_id="call", artifact=artifact) + sources = [source.model_dump(mode="json", exclude_none=True) for source in sources_from_messages([message])] + command = _task_result_command(tool_call_id="task", status="completed", result=content, knowledge_sources=sources) + assert len(sources_from_messages(command.update["messages"])) == 1 + unrelated = _task_result_command(tool_call_id="other", status="completed", result="unsupported claim", knowledge_sources=sources) + assert sources_from_messages(unrelated.update["messages"]) == [] + + +def test_gateway_drops_caller_forged_source_artifacts() -> None: + from app.gateway.services import normalize_input, strip_server_owned_state_metadata + + artifact = {"knowledge_sources_version": 1, "knowledge_sources": [{"title": "forged"}], "other": "preserved"} + raw = {"type": "tool", "content": "claim", "tool_call_id": "call", "artifact": artifact} + assert normalize_input({"messages": [raw]})["messages"][0].artifact == {"other": "preserved"} + assert strip_server_owned_state_metadata({"messages": [raw]})["messages"][0].artifact == {"other": "preserved"} + assert artifact["knowledge_sources"] # no mutation of caller data + + +def test_global_budget_prunes_hidden_evidence_and_orphan_citations() -> None: + from deerflow.agents.middlewares.tool_output_budget_middleware import _patch_tool_message + from deerflow.config.tool_output_config import ToolOutputConfig + + source = KnowledgeSource(provider="demo", title="Guide", excerpt="x" * 500) + content, artifact = format_result(RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)])) + message = ToolMessage(content=content, tool_call_id="call", name="knowledge_search", artifact=artifact) + patched = _patch_tool_message(message, ToolOutputConfig(externalize_min_chars=0, fallback_max_chars=100), None) + assert len(patched.content) <= 100 + assert sources_from_messages([patched]) == [] + assert patched.artifact["knowledge_sources"] == [] + + +def test_mixed_delegation_preserves_both_source_contracts() -> None: + from deerflow.knowledge.sources import prune_sources, strip_source_artifact + from deerflow.tools.builtins.task_tool import _task_result_command + + unified = KnowledgeSource(provider="lightrag", title="Guide", excerpt="LightRAG evidence") + unified_text, unified_artifact = format_result(RetrievalResult(hits=[RetrievalHit(content=unified.excerpt, source=unified)])) + unified_sources = [item.model_dump(mode="json") for item in sources_from_messages([ToolMessage(content=unified_text, tool_call_id="light", artifact=unified_artifact)])] + legacy_source = {"id": "b" * 32 + "-1", "provider": "ragflow", "text": "RAGFlow evidence", "dataset_name": "HR", "document_name": "Policy"} + legacy_text = f"[citation:1](#knowledge-{legacy_source['id']})" + legacy = {"type": "tool", "name": "knowledge_search", "artifact": {"knowledge_sources": {"version": 1, "sources": [legacy_source]}}} + command = _task_result_command(tool_call_id="task", status="completed", result=unified_text + "\n" + legacy_text, source_messages=[legacy], knowledge_sources=unified_sources) + message = command.update["messages"][0] + assert message.artifact["knowledge_sources"]["sources"] == [legacy_source] + assert sources_from_messages([message])[0].provider == "lightrag" + _, pruned = prune_sources(legacy_text, message.artifact) + assert pruned["retrieval_sources"] == [] + assert pruned["knowledge_sources"]["sources"] == [legacy_source] + alternate = {key: value for key, value in message.artifact.items() if key != "knowledge_sources"} + assert sources_from_messages([message.model_copy(update={"artifact": alternate})])[0].provider == "lightrag" + _, pruned_alternate = prune_sources(legacy_text, alternate) + assert pruned_alternate["retrieval_sources"] == [] + assert strip_source_artifact({**message.artifact, "other": "kept"}) == {"other": "kept"} + + +@pytest.mark.anyio +async def test_ragflow_fallback_honors_disabled_runtime_scope(monkeypatch) -> None: + from types import SimpleNamespace + + from deerflow.community.ragflow import tools + from deerflow.knowledge_scope import KNOWLEDGE_SCOPE_RUNTIME_KEY + + def forbidden(): + pytest.fail("disabled knowledge scope must not resolve settings or access the provider") + + monkeypatch.setattr(tools, "_settings_or_error", forbidden) + runtime = SimpleNamespace(context={KNOWLEDGE_SCOPE_RUNTIME_KEY: {"version": 1, "mode": "disabled"}}) + text, artifact = await tools._knowledge_search_entrypoint("policy", runtime) + assert "disabled" in text + assert artifact is None diff --git a/backend/tests/test_knowledge_contract_live.py b/backend/tests/test_knowledge_contract_live.py new file mode 100644 index 0000000000..07e9bbbffc --- /dev/null +++ b/backend/tests/test_knowledge_contract_live.py @@ -0,0 +1,29 @@ +"""Opt-in validation against real LightRAG /query/data evidence; no uploads.""" + +import os + +import pytest + +from deerflow.community.lightrag.client import LightRAGClient +from deerflow.community.lightrag.formatting import normalize_retrieval_result +from deerflow.knowledge import format_result, sources_from_messages + + +@pytest.mark.live +@pytest.mark.anyio +async def test_lightrag_live_evidence_contract() -> None: + if os.getenv("DEER_FLOW_RUN_LIVE_TESTS") != "1": + pytest.skip("live knowledge checks are opt-in") + base_url, query = os.getenv("LIGHTRAG_BASE_URL"), os.getenv("LIGHTRAG_EVIDENCE_QUERY") + if not base_url or not query: + pytest.skip("set LIGHTRAG_BASE_URL and LIGHTRAG_EVIDENCE_QUERY for an indexed workspace") + result = await LightRAGClient(base_url=base_url, api_key=os.getenv("LIGHTRAG_API_KEY")).query_data(query, mode="mix", chunk_top_k=8) + normalized = normalize_retrieval_result(result) + assert any(hit.source for hit in normalized.hits), "Live query must retrieve attributable document text" + text, artifact = format_result(normalized) + sources = sources_from_messages([{"type": "tool", "content": text, "artifact": artifact}]) + assert sources, "At least one real retrieved source must fit the default budget" + for source in sources: + assert any(hit.source and hit.source.title == source.title and source.excerpt.rstrip("…") in hit.content for hit in normalized.hits) + assert source.provider == "lightrag" + assert source.page is None # /query/data does not supply page evidence diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 98bb6e8b80..70e13120ea 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -54,6 +54,39 @@ _MOCKED_MODULE_NAMES = [ _LANGGRAPH_HAS_ROOT_LINEAGE_STREAM_REGRESSION = Version(package_version("langgraph")) >= Version("1.2.6") +def test_knowledge_sources_survive_compaction_and_terminal_publication(_setup_executor_classes): + from langchain_core.messages import AIMessage, ToolMessage + + from deerflow.knowledge import KnowledgeSource, RetrievalHit, RetrievalResult, format_result + + classes = _setup_executor_classes + source = KnowledgeSource(provider="demo", title="Guide", excerpt="real evidence") + content, artifact = format_result(RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)])) + result = classes["SubagentResult"](task_id="task", trace_id="trace", status=classes["SubagentStatus"].RUNNING) + result.update_knowledge_sources([ToolMessage(content=content, tool_call_id="call", artifact=artifact)]) + result.update_knowledge_sources([AIMessage(content="compacted")]) + assert len(result.knowledge_sources) == 1 + result.try_set_terminal(classes["SubagentStatus"].COMPLETED, result=content) + result.update_knowledge_sources([]) + assert len(result.knowledge_sources) == 1 + + +def test_knowledge_accumulator_scans_only_changed_tail(_setup_executor_classes, monkeypatch): + import deerflow.knowledge as knowledge + + classes = _setup_executor_classes + scanned = [] + monkeypatch.setattr(knowledge, "sources_from_messages", lambda messages: scanned.append(list(messages)) or []) + result = classes["SubagentResult"](task_id="task", trace_id="trace", status=classes["SubagentStatus"].RUNNING) + first, second, compacted = {"type": "ai", "content": "first"}, {"type": "ai", "content": "second"}, {"type": "ai", "content": "summary"} + result.update_knowledge_sources([first]) + result.update_knowledge_sources([first, second]) + result.update_knowledge_sources([first, second]) + result.update_knowledge_sources([compacted]) + result.update_knowledge_sources(None) + assert scanned == [[first], [second], [compacted]] + + def _default_app_config(): return SimpleNamespace( tool_search=SimpleNamespace(enabled=False), diff --git a/contracts/knowledge_sources.v1.json b/contracts/knowledge_sources.v1.json new file mode 100644 index 0000000000..4c5ad0eec3 --- /dev/null +++ b/contracts/knowledge_sources.v1.json @@ -0,0 +1,35 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "Shared retrieval source envelope v1", + "description": "Server-owned tool evidence; readers also require the handle and excerpt in tool content. Mixed native RAGFlow artifacts carry this array under retrieval_sources.", + "type": "object", + "required": ["knowledge_sources_version", "knowledge_sources"], + "properties": { + "knowledge_sources_version": { "const": 1, "type": "integer" }, + "knowledge_sources": { + "type": "array", + "maxItems": 100, + "items": { "$ref": "#/$defs/source" } + } + }, + "$defs": { + "source": { + "type": "object", + "required": ["id", "provider", "title", "excerpt"], + "properties": { + "id": { "type": "string", "pattern": "^[a-f0-9]{32}$" }, + "provider": { "type": "string", "minLength": 1, "maxLength": 100 }, + "title": { "type": "string", "minLength": 1, "maxLength": 500 }, + "excerpt": { "type": "string", "minLength": 1, "maxLength": 100000 }, + "dataset_name": { "type": "string", "maxLength": 500 }, + "url": { + "type": "string", + "format": "uri", + "pattern": "^https?://", + "description": "No username or password allowed." + }, + "page": { "type": "integer", "minimum": 1, "maximum": 9007199254740991 } + } + } + } +} diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index c713fac54e..e90df8d473 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -76,6 +76,23 @@ Keep IM notification preferences viewer-fenced and separate from browser alerts; ## Architecture +Knowledge evidence is parsed in `core/citations/sources.ts` from version-1 +ToolMessage artifacts. Message rows receive sources from preceding loaded tool +messages. `#knowledge-source-ID` links resolve to `KnowledgeCitation` and an +expandable source panel with actual excerpts. Missing URLs/pages remain absent. +Web citations and old history remain compatible. Never infer evidence from +assistant labels or numbered legacy text. +Shared records use the `RetrievalSource` type (native RAGFlow retains +`KnowledgeSource`). Accept raw and `#user-content-knowledge-source-ID` +fragments without disabling sanitizer clobber protection. Validate immutable +tool messages through `createKnowledgeSourceReader` to reuse per-object parses. +Cross-component field bounds live in `contracts/knowledge_sources.v1.json`. +Completed `SubtaskCard` reports read shared evidence only from the matching +`task` ToolMessage after its invocation in the current subagent group, selected +by `findSubtaskResultMessage`. Prompts/earlier steps do not borrow final evidence. +Remount report Markdown when validated source IDs change: Streamdown caches +unchanged text blocks independently of renderer components. + ``` Frontend (Next.js) ──▶ LangGraph SDK ──▶ LangGraph Backend (lead_agent) ├── Sub-Agents @@ -86,22 +103,16 @@ Users create **threads** (conversations), send messages, set thread-scoped `/goa ### Source Layout (`src/`) -- **`app/`** — Next.js App Router. Routes include `/` (landing), `/showcase/[thread_id]` (allowlisted public read-only demos), `/workspace/chats/[thread_id]` (authenticated chat), `/workspace/agents/[agent_name]` and `/workspace/agents/new` (custom agents), `/artifacts/view` (chrome-free window that renders Markdown or CSV/TSV artifacts with the panel's own renderer), `/blog/…`, the `(auth)/{login,setup,auth/callback}` flow, `/[lang]/docs/…`, and `/api/…` route handlers (e.g. `/api/memory`). -- **`components/`** — React components: - - `ui/` — Shadcn UI primitives (auto-generated, ESLint-ignored) - - `ai-elements/` — Vercel AI SDK elements (auto-generated, ESLint-ignored) - - `workspace/` — Chat page components (messages, artifacts, settings) - - `landing/` — Landing page sections - - `docs/` — Docs / MDX rendering components -- **`core/`** — Business logic, the heart of the app. Domains include `threads/` (creation, streaming, state), `api/` (LangGraph client singleton), `agents/` (custom agents), `subagents/` (runtime worker catalog and administrator mutations), `auth/` (authentication), `artifacts/`, `channels/` (IM connections), `integrations/` (managed third-party integration status/install clients such as Lark CLI), `i18n/` (en-US, zh-CN), `settings/`, `memory/`, `skills/`, `messages/`, `mcp/`, `models/`, `input-polish/` (pre-send draft rewrite API), `voice-input/` (browser speech-recognition helpers), `suggestions/`, `tasks/`, `todos/`, `tools/`, `workspace-changes/` (run-scoped changed-file summaries and diff fetching), `config/`, `notification/`, `blog/`, plus rendering helpers (`rehype/`, `streamdown/`) and `utils/`. -- **`hooks/`** — Shared React hooks -- **`lib/`** — Utilities (`cn()` from clsx + tailwind-merge) -- **`content/`** — MDX content (blog posts, docs) rendered by the app -- **`styles/`** — Global CSS with Tailwind v4 `@import` syntax and CSS variables for theming -- **`typings/`** — Ambient TypeScript declarations -- Root files: `env.js` (env validation), `mdx-components.ts` (MDX component map) +- `app/` — Next.js routes: chat/workspace, agents, auth, showcase, artifact views, + localized docs/blog, and API handlers. +- `components/` — generated `ui/` and `ai-elements/`; application components under + `workspace/`, `landing/`, and `docs/`. +- `core/` — domain logic, API clients, state/hooks, i18n, and Markdown renderers. +- `hooks/`, `lib/`, `typings/` — shared hooks, utilities, and ambient types. +- `content/`, `styles/` — MDX content and Tailwind styles. +- `env.js`, `mdx-components.ts` — environment validation and MDX component map. -More specific `AGENTS.md` files under `src/` contain the frontend sections split from this file. +Read [src/AGENTS.md](src/AGENTS.md) for component and domain ownership. ## Code Style diff --git a/frontend/src/components/workspace/citations/citation-sources-panel.tsx b/frontend/src/components/workspace/citations/citation-sources-panel.tsx index 8acc70d5bf..907d7e2c3a 100644 --- a/frontend/src/components/workspace/citations/citation-sources-panel.tsx +++ b/frontend/src/components/workspace/citations/citation-sources-panel.tsx @@ -17,6 +17,8 @@ import { writeTextToClipboard } from "@/core/clipboard"; import { useI18n } from "@/core/i18n/hooks"; import { cn } from "@/lib/utils"; +import { KnowledgeEvidence } from "./knowledge-citation"; + export function CitationSourcesPanel({ className, sources, @@ -44,33 +46,47 @@ export function CitationSourcesPanel({
{source.title}
++ {source.provider} + {source.dataset_name ? ` / ${source.dataset_name}` : ""} + {source.page + ? ` · ${t.citations.sourcePages(String(source.page))}` + : ""} +
++ {source.excerpt} ++ {source.url && ( + + {source.url} + + )} +