diff --git a/frontend-web/src/components/workspace/knowledge-rag-toggle.tsx b/frontend-web/src/components/workspace/knowledge-rag-toggle.tsx new file mode 100644 index 00000000..42c12c40 --- /dev/null +++ b/frontend-web/src/components/workspace/knowledge-rag-toggle.tsx @@ -0,0 +1,44 @@ +"use client"; + +import { Database } from "lucide-react"; + +import { Button } from "@/components/ui/button"; +import { useThreadSettings } from "@/core/settings"; +import { cn } from "@/lib/utils"; + +import { Tooltip } from "./tooltip"; + +/** + * Per-conversation toggle for knowledge-base RAG injection (phase 2). + * + * Writes ``context.knowledge_rag_enabled`` into the thread settings, which the + * run forwards as ``runtime.context`` so ``KnowledgeRagMiddleware`` injects + * recalled knowledge before each model call for this conversation only. + */ +export function KnowledgeRagToggle({ threadId }: { threadId: string }) { + const [settings, setSettings] = useThreadSettings(threadId); + const enabled = settings.context.knowledge_rag_enabled === true; + + const toggle = () => { + setSettings("context", { + ...settings.context, + knowledge_rag_enabled: !enabled, + }); + }; + + return ( + + + + ); +} diff --git a/frontend-web/src/core/threads/types.ts b/frontend-web/src/core/threads/types.ts index 57a566a6..475269a7 100644 --- a/frontend-web/src/core/threads/types.ts +++ b/frontend-web/src/core/threads/types.ts @@ -18,9 +18,11 @@ export interface AgentThreadContext extends Record { reasoning_effort?: "minimal" | "low" | "medium" | "high"; writing_mode?: boolean; writing_artifact_path?: string; - memory_injection_enabled?: boolean; - rag_mode_enabled?: boolean; - agent_id?: string; + memory_injection_enabled?: boolean; + rag_mode_enabled?: boolean; + /** Per-conversation switch for knowledge-base RAG injection (phase 2). */ + knowledge_rag_enabled?: boolean; + agent_id?: string; /** @deprecated kept for reading legacy thread metadata; prefer agent_id */ agent_name?: string; } diff --git a/frontend-web/src/pages/ChatPage.tsx b/frontend-web/src/pages/ChatPage.tsx index 8199283c..dacfbe6b 100644 --- a/frontend-web/src/pages/ChatPage.tsx +++ b/frontend-web/src/pages/ChatPage.tsx @@ -16,6 +16,7 @@ import { } from "@/components/workspace/chats"; import { ExportTrigger } from "@/components/workspace/export-trigger"; import { KnowledgeSaveTrigger } from "@/components/workspace/knowledge-save-trigger"; +import { KnowledgeRagToggle } from "@/components/workspace/knowledge-rag-toggle"; import { InputBox } from "@/components/workspace/input-box"; import { MessageList, @@ -409,6 +410,7 @@ export default function ChatPage() { enabled={tokenUsageEnabled} messages={thread.messages} /> + diff --git a/offline-backend-20260512/backend/app/gateway/routers/knowledge.py b/offline-backend-20260512/backend/app/gateway/routers/knowledge.py index 8a023c85..88d6dded 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/knowledge.py +++ b/offline-backend-20260512/backend/app/gateway/routers/knowledge.py @@ -304,6 +304,22 @@ async def search(body: SearchRequest, request: Request) -> dict[str, Any]: return {"items": items} +@router.post("/reindex") +@require_auth +async def reindex_embeddings(request: Request) -> dict[str, Any]: + """Rebuild the vector index for all notes (maintenance; no-op if embeddings off).""" + service = get_knowledge_service(request) + return await service.reindex_all() + + +@router.get("/notes/{note_id}/versions") +@require_auth +async def list_note_versions(note_id: str, request: Request) -> dict[str, Any]: + """Return the edit history (pre-update snapshots) for a note.""" + service = get_knowledge_service(request) + return {"items": await service.list_versions(note_id)} + + @router.post("/export/graph") @require_auth async def export_graph(request: Request) -> dict[str, Any]: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py index ae1398fe..833091e2 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -299,6 +299,16 @@ def _build_middlewares( if not is_bootstrap and not is_writing_setup: middlewares.append(MemoryMiddleware(agent_name=agent_name, memory_config=resolved_app_config.memory, injection_enabled=memory_injection_enabled)) + # KnowledgeRagMiddleware — inject knowledge-base recall (phase 2). Gated + # per-conversation by the ``knowledge_rag_enabled`` runtime flag, defaulting + # to the global ``knowledge.rag_enabled`` config. No-op when knowledge is + # disabled. Skipped in bootstrap/writing-setup like MemoryMiddleware. + knowledge_cfg = getattr(resolved_app_config, "knowledge", None) + if not is_bootstrap and not is_writing_setup and knowledge_cfg is not None and knowledge_cfg.enabled: + from deerflow.agents.middlewares.knowledge_rag_middleware import KnowledgeRagMiddleware + + middlewares.append(KnowledgeRagMiddleware(limit=knowledge_cfg.rag_limit, default_enabled=knowledge_cfg.rag_enabled)) + # Add ViewImageMiddleware only for the NATIVE-vision path (main model # ingests raw image data). In delegated mode the view_image tool already # returns a text description and stores nothing in `viewed_images`, so the diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/knowledge_rag_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/knowledge_rag_middleware.py new file mode 100644 index 00000000..49f403cc --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/knowledge_rag_middleware.py @@ -0,0 +1,134 @@ +"""KnowledgeRagMiddleware — inject knowledge-base recall before the model call. + +Phase 2 of the product knowledge base. When enabled (per-conversation runtime +flag ``knowledge_rag_enabled``, falling back to the global +``knowledge.rag_enabled`` config), the latest user question is used to retrieve +the most relevant knowledge notes, which are injected as a transient +```` system message — not persisted to state, so it never +leaks to the frontend transcript. + +Retrieval is hybrid (vector + keyword) when an embedding endpoint is +configured, else keyword/wiki-query layered search. Failures are swallowed: RAG +must never break a conversation. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any, override + +from langchain.agents import AgentState +from langchain.agents.middleware import AgentMiddleware +from langchain_core.messages import HumanMessage + +logger = logging.getLogger(__name__) + +_TAG = "knowledge-context" + + +def _coerce_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for block in content: + if isinstance(block, str): + parts.append(block) + elif isinstance(block, dict) and block.get("type") == "text": + parts.append(str(block.get("text", ""))) + return " ".join(parts) + return str(content or "") + + +def _latest_human_text(messages: list) -> str: + for msg in reversed(messages or []): + if isinstance(msg, HumanMessage): + return _coerce_text(msg.content).strip() + return "" + + +class KnowledgeRagMiddleware(AgentMiddleware[AgentState]): + """Inject knowledge-base recall as transient system context.""" + + state_schema = AgentState + + def __init__(self, *, limit: int = 5, default_enabled: bool = False) -> None: + super().__init__() + self._limit = limit + self._default_enabled = default_enabled + self._service = None # lazy + + def _get_service(self): + if self._service is not None: + return self._service + try: + from deerflow.config import get_app_config + from deerflow.knowledge import make_knowledge_service + from deerflow.persistence.engine import get_session_factory + + config = get_app_config() + self._service = make_knowledge_service(get_session_factory(), config.knowledge) + except Exception: + logger.exception("KnowledgeRagMiddleware: failed to build knowledge service") + self._service = None + return self._service + + def _enabled_for(self, request: Any) -> bool: + runtime = getattr(request, "runtime", None) + if runtime is not None and getattr(runtime, "context", None): + flag = runtime.context.get("knowledge_rag_enabled") + if flag is not None: + return bool(flag) + return self._default_enabled + + async def _build_context_message(self, request: Any) -> HumanMessage | None: + if not self._enabled_for(request): + return None + service = self._get_service() + if service is None: + return None + query = _latest_human_text(getattr(request, "messages", None) or []) + if not query: + return None + try: + result = await service.build_rag_context(query, limit=self._limit) + except Exception: + logger.exception("KnowledgeRagMiddleware: recall failed (non-fatal)") + return None + text = result.get("text") if isinstance(result, dict) else "" + if not text: + return None + fenced = ( + f"<{_TAG}>\n" + "[系统提示:以下是知识库召回的背景资料,不是用户的新输入。回答时优先参考," + "并在使用到时标注来源 note_id。]\n\n" + f"{text}\n" + f"" + ) + return HumanMessage(content=fenced) + + @override + async def awrap_model_call(self, request: Any, handler: Any) -> Any: + msg = await self._build_context_message(request) + if msg is not None: + request = request.override(messages=[*request.messages, msg]) + return await handler(request) + + @override + def wrap_model_call(self, request: Any, handler: Any) -> Any: + # Sync path: best-effort. The gateway runtime is async, so this is rarely + # hit; run the async recall in a private loop when there is none. + if not self._enabled_for(request): + return handler(request) + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + # Cannot block a running loop from sync code; skip injection. + return handler(request) + msg = loop.run_until_complete(self._build_context_message(request)) + except Exception: + msg = None + if msg is not None: + request = request.override(messages=[*request.messages, msg]) + return handler(request) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/knowledge_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/knowledge_config.py index 18900103..10de6e53 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/knowledge_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/knowledge_config.py @@ -20,11 +20,22 @@ class KnowledgeQueryConfig(BaseModel): class KnowledgeEmbeddingConfig(BaseModel): - """Vector embedding config (phase 2 — disabled in the first release).""" + """Vector embedding config (phase 2). + + Vector RAG is opt-in: set ``enabled: true`` and point ``base_url`` at an + OpenAI-compatible ``/embeddings`` endpoint. When disabled or unreachable, + search transparently falls back to keyword scoring, so the feature degrades + gracefully on offline deployments without an embedding model. + """ enabled: bool = Field(default=False, description="Enable vector embeddings.") - provider: str = Field(default="local", description="Embedding provider.") - model: str = Field(default="", description="Embedding model name.") + provider: str = Field(default="openai", description="Embedding provider (openai-compatible).") + model: str = Field(default="", description="Embedding model name, e.g. text-embedding-3-small / bge-m3.") + base_url: str = Field(default="", description="OpenAI-compatible base URL (…/v1). Empty disables vector mode.") + api_key: str = Field(default="", description="API key for the embedding endpoint ($ENV resolved like other config).") + dimensions: int | None = Field(default=None, description="Optional embedding dimensions hint sent to the endpoint.") + batch_size: int = Field(default=32, description="Texts per embedding request.") + timeout_seconds: int = Field(default=30, description="HTTP timeout per embedding request.") class KnowledgeConfig(BaseModel): diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/chunking.py b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/chunking.py new file mode 100644 index 00000000..7897d8c1 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/chunking.py @@ -0,0 +1,60 @@ +"""Markdown chunking for embedding/index (phase 2). + +Splits a note body into overlapping chunks along Markdown structure (headings → +paragraphs) so each embedded chunk is semantically coherent and bounded in +size. Frontmatter is assumed already stripped by the caller. +""" + +from __future__ import annotations + +import re + +_HEADING_RE = re.compile(r"^#{1,6}\s+", re.MULTILINE) + + +def _split_paragraphs(text: str) -> list[str]: + return [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] + + +def chunk_markdown(content: str, *, max_chars: int = 800, overlap: int = 120) -> list[str]: + """Return a list of text chunks for ``content``. + + Sections are cut at Markdown headings; long sections are further split into + ~``max_chars`` windows with ``overlap`` characters of carry-over. + """ + text = (content or "").strip() + if not text: + return [] + + # Split into heading-delimited sections, keeping the heading with its body. + sections: list[str] = [] + last = 0 + for m in _HEADING_RE.finditer(text): + if m.start() > last: + sections.append(text[last : m.start()].strip()) + last = m.start() + sections.append(text[last:].strip()) + sections = [s for s in sections if s] + + chunks: list[str] = [] + for section in sections: + if len(section) <= max_chars: + chunks.append(section) + continue + # Pack paragraphs into windows, then hard-split anything still too long. + buf = "" + for para in _split_paragraphs(section): + if len(buf) + len(para) + 2 <= max_chars: + buf = f"{buf}\n\n{para}".strip() + else: + if buf: + chunks.append(buf) + if len(para) <= max_chars: + buf = para + else: + for i in range(0, len(para), max_chars - overlap): + chunks.append(para[i : i + max_chars]) + buf = "" + if buf: + chunks.append(buf) + return [c for c in chunks if c.strip()] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/embeddings.py b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/embeddings.py new file mode 100644 index 00000000..b28843e4 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/embeddings.py @@ -0,0 +1,79 @@ +"""Embedding generation for vector RAG (phase 2). + +Talks to an OpenAI-compatible ``/embeddings`` endpoint configured under +``knowledge.embedding``. Everything here is **best-effort**: if embeddings are +disabled, unconfigured, or the endpoint errors, callers get an empty result and +the search layer falls back to keyword scoring — so vector RAG never breaks a +deployment that has no embedding model. +""" + +from __future__ import annotations + +import logging +import math +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from deerflow.config.knowledge_config import KnowledgeEmbeddingConfig + +logger = logging.getLogger(__name__) + + +def cosine_similarity(a: list[float], b: list[float]) -> float: + """Cosine similarity of two equal-length vectors (0 when degenerate).""" + if not a or not b or len(a) != len(b): + return 0.0 + dot = sum(x * y for x, y in zip(a, b, strict=False)) + na = math.sqrt(sum(x * x for x in a)) + nb = math.sqrt(sum(y * y for y in b)) + if na == 0 or nb == 0: + return 0.0 + return dot / (na * nb) + + +class EmbeddingClient: + """Thin async client over an OpenAI-compatible embeddings endpoint.""" + + def __init__(self, config: KnowledgeEmbeddingConfig) -> None: + self.config = config + + def is_enabled(self) -> bool: + return bool(self.config.enabled and self.config.base_url and self.config.model) + + async def embed_texts(self, texts: list[str]) -> list[list[float]]: + """Return one vector per input text; ``[]`` on any failure/disabled.""" + if not self.is_enabled() or not texts: + return [] + try: + import httpx + except Exception: # pragma: no cover - httpx is a core dep + logger.warning("httpx unavailable; knowledge embeddings disabled") + return [] + + url = self.config.base_url.rstrip("/") + "/embeddings" + headers = {"Content-Type": "application/json"} + if self.config.api_key: + headers["Authorization"] = f"Bearer {self.config.api_key}" + + vectors: list[list[float]] = [] + batch = max(1, self.config.batch_size) + try: + async with httpx.AsyncClient(timeout=self.config.timeout_seconds) as client: + for start in range(0, len(texts), batch): + chunk = texts[start : start + batch] + payload: dict = {"model": self.config.model, "input": chunk} + if self.config.dimensions: + payload["dimensions"] = self.config.dimensions + resp = await client.post(url, headers=headers, json=payload) + resp.raise_for_status() + data = resp.json() + items = sorted(data.get("data", []), key=lambda d: d.get("index", 0)) + vectors.extend([item.get("embedding", []) for item in items]) + except Exception: + logger.exception("Knowledge embedding request failed; falling back to keyword search") + return [] + return vectors + + async def embed_query(self, text: str) -> list[float]: + vecs = await self.embed_texts([text]) + return vecs[0] if vecs else [] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/models.py b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/models.py index 7c254faf..07a4e338 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/models.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/models.py @@ -13,11 +13,11 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import Float, ForeignKey, String, Text, UniqueConstraint +from sqlalchemy import Float, ForeignKey, Integer, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base -from deerflow.persistence.types import BeijingDateTime, PortableLongText +from deerflow.persistence.types import BeijingDateTime, PortableJSON, PortableLongText class KnowledgeNoteRow(Base): @@ -77,3 +77,65 @@ class KnowledgeNoteTagRow(Base): note_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_notes.id", ondelete="CASCADE"), primary_key=True) tag_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_tags.id", ondelete="CASCADE"), primary_key=True) + + +class KnowledgeEmbeddingRow(Base): + """Per-chunk embedding index (phase 2, mirrors ``knowledge_embeddings``). + + The vector is stored as a JSON float array; first-release similarity is + computed in Python (no external vector DB). ``vector_id`` is reserved for a + future external store mapping. + """ + + __tablename__ = "knowledge_embeddings" + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + note_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_notes.id", ondelete="CASCADE"), index=True) + chunk_index: Mapped[int] = mapped_column(Integer, default=0) + content: Mapped[str] = mapped_column(Text) + vector: Mapped[list] = mapped_column(PortableJSON(), default=list) + vector_id: Mapped[str | None] = mapped_column(String(256)) + created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC)) + + +class KnowledgeNoteVersionRow(Base): + """Immutable snapshot of a note before each update (phase 3 versioning).""" + + __tablename__ = "knowledge_note_versions" + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + note_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_notes.id", ondelete="CASCADE"), index=True) + title: Mapped[str] = mapped_column(String(512)) + summary: Mapped[str | None] = mapped_column(Text) + content_md: Mapped[str] = mapped_column(PortableLongText()) + status: Mapped[str] = mapped_column(String(32)) + tags_json: Mapped[list] = mapped_column(PortableJSON(), default=list) + edited_by: Mapped[str | None] = mapped_column(String(128)) + created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True) + + +class KnowledgeEntityRow(Base): + """Knowledge entity (phase 4, mirrors ``knowledge_entities``).""" + + __tablename__ = "knowledge_entities" + __table_args__ = (UniqueConstraint("name", "entity_type", name="uq_knowledge_entities_name_type"),) + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + name: Mapped[str] = mapped_column(String(256), index=True) + entity_type: Mapped[str | None] = mapped_column(String(64)) + created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC)) + + +class KnowledgeRelationRow(Base): + """Relation between notes/entities (phase 4, mirrors ``knowledge_relations``).""" + + __tablename__ = "knowledge_relations" + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + from_note_id: Mapped[str | None] = mapped_column(String(64), index=True) + to_note_id: Mapped[str | None] = mapped_column(String(64), index=True) + from_entity_id: Mapped[str | None] = mapped_column(String(64), index=True) + to_entity_id: Mapped[str | None] = mapped_column(String(64), index=True) + relation_type: Mapped[str] = mapped_column(String(128)) + weight: Mapped[float] = mapped_column(Float, default=1.0) + created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/repository.py b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/repository.py index 93c56573..a87692e4 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/repository.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/repository.py @@ -16,8 +16,12 @@ from sqlalchemy import delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.knowledge.models import ( + KnowledgeEmbeddingRow, + KnowledgeEntityRow, KnowledgeNoteRow, KnowledgeNoteTagRow, + KnowledgeNoteVersionRow, + KnowledgeRelationRow, KnowledgeSourceRow, KnowledgeTagRow, ) @@ -288,6 +292,140 @@ class KnowledgeRepository: out.append(d) return out + # -- embeddings (phase 2) -------------------------------------------- + async def replace_embeddings(self, note_id: str, chunks: list[tuple[int, str, list[float]]]) -> None: + """Replace all embedding rows for a note with ``(index, content, vector)``.""" + async with self._sf() as session: + await session.execute(delete(KnowledgeEmbeddingRow).where(KnowledgeEmbeddingRow.note_id == note_id)) + for idx, content, vector in chunks: + session.add( + KnowledgeEmbeddingRow( + id=_new_id("kbemb"), + note_id=note_id, + chunk_index=idx, + content=content, + vector=list(vector), + created_at=datetime.now(UTC), + ) + ) + await session.commit() + + async def delete_embeddings(self, note_id: str) -> None: + async with self._sf() as session: + await session.execute(delete(KnowledgeEmbeddingRow).where(KnowledgeEmbeddingRow.note_id == note_id)) + await session.commit() + + async def all_embeddings(self) -> list[dict[str, Any]]: + """Return every embedding chunk joined to its (non-archived) note.""" + async with self._sf() as session: + stmt = ( + select(KnowledgeEmbeddingRow, KnowledgeNoteRow.title, KnowledgeNoteRow.status) + .join(KnowledgeNoteRow, KnowledgeNoteRow.id == KnowledgeEmbeddingRow.note_id) + .where(KnowledgeNoteRow.status != "archived") + ) + out = [] + for emb, title, _status in (await session.execute(stmt)).all(): + out.append({"note_id": emb.note_id, "chunk_index": emb.chunk_index, "content": emb.content, "vector": emb.vector or [], "title": title}) + return out + + async def has_embeddings(self) -> bool: + async with self._sf() as session: + row = (await session.execute(select(KnowledgeEmbeddingRow.id).limit(1))).first() + return row is not None + + # -- versions (phase 3) ---------------------------------------------- + async def add_version(self, note: dict[str, Any], *, edited_by: str | None) -> None: + async with self._sf() as session: + session.add( + KnowledgeNoteVersionRow( + id=_new_id("kbver"), + note_id=note["id"], + title=note.get("title") or "", + summary=note.get("summary"), + content_md=note.get("content_md") or "", + status=note.get("status") or "approved", + tags_json=note.get("tags") or [], + edited_by=edited_by, + created_at=datetime.now(UTC), + ) + ) + await session.commit() + + async def list_versions(self, note_id: str) -> list[dict[str, Any]]: + async with self._sf() as session: + stmt = select(KnowledgeNoteVersionRow).where(KnowledgeNoteVersionRow.note_id == note_id).order_by(KnowledgeNoteVersionRow.created_at.desc()) + out = [] + for r in (await session.execute(stmt)).scalars(): + d = r.to_dict() + d["tags"] = d.pop("tags_json", []) + d["created_at"] = _iso(d.get("created_at")) + out.append(d) + return out + + # -- entities & relations (phase 4) ---------------------------------- + async def get_or_create_entity(self, name: str, entity_type: str | None) -> dict[str, Any]: + name = (name or "").strip() + async with self._sf() as session: + stmt = select(KnowledgeEntityRow).where(func.lower(KnowledgeEntityRow.name) == name.lower()) + row = (await session.execute(stmt)).scalars().first() + if row is None: + row = KnowledgeEntityRow(id=_new_id("kbent"), name=name, entity_type=entity_type, created_at=datetime.now(UTC)) + session.add(row) + await session.commit() + await session.refresh(row) + return {"id": row.id, "name": row.name, "entity_type": row.entity_type} + + async def add_relation( + self, + *, + from_note_id: str | None = None, + to_note_id: str | None = None, + from_entity_id: str | None = None, + to_entity_id: str | None = None, + relation_type: str, + weight: float = 1.0, + ) -> None: + async with self._sf() as session: + session.add( + KnowledgeRelationRow( + id=_new_id("kbrel"), + from_note_id=from_note_id, + to_note_id=to_note_id, + from_entity_id=from_entity_id, + to_entity_id=to_entity_id, + relation_type=relation_type, + weight=weight, + created_at=datetime.now(UTC), + ) + ) + await session.commit() + + async def clear_note_relations(self, note_id: str) -> None: + async with self._sf() as session: + await session.execute(delete(KnowledgeRelationRow).where(KnowledgeRelationRow.from_note_id == note_id)) + await session.commit() + + async def list_entities(self) -> list[dict[str, Any]]: + async with self._sf() as session: + rows = (await session.execute(select(KnowledgeEntityRow))).scalars() + return [{"id": r.id, "name": r.name, "entity_type": r.entity_type} for r in rows] + + async def list_relations(self) -> list[dict[str, Any]]: + async with self._sf() as session: + rows = (await session.execute(select(KnowledgeRelationRow))).scalars() + return [ + { + "id": r.id, + "from_note_id": r.from_note_id, + "to_note_id": r.to_note_id, + "from_entity_id": r.from_entity_id, + "to_entity_id": r.to_entity_id, + "relation_type": r.relation_type, + "weight": r.weight, + } + for r in rows + ] + # -- sources ---------------------------------------------------------- async def add_sources(self, note_id: str, sources: list[SourceDraft], *, created_by: str | None = None) -> None: if not sources: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/search.py b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/search.py index d0d34f8e..f3430e17 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/search.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/search.py @@ -1,17 +1,22 @@ -"""Keyword / frontmatter / Markdown-content search (first release). +"""Knowledge search: keyword / vector / hybrid (phase 1 + 2). -Vector and hybrid modes are reserved for phase 2 (see the dev doc §9.9 / §13); -for now ``vector`` and ``hybrid`` transparently fall back to keyword search so -the API contract is stable. Scoring weights title > tags > summary > body. +- ``keyword`` — title>tags>summary>body scoring over the DB mirror (always works). +- ``vector`` — cosine over per-chunk embeddings (requires a configured + embedding endpoint; otherwise transparently falls back to ``keyword``). +- ``hybrid`` — min-max-normalized blend of keyword + vector. """ from __future__ import annotations import re -from typing import Any +from typing import TYPE_CHECKING, Any +from deerflow.knowledge.embeddings import cosine_similarity from deerflow.knowledge.repository import KnowledgeRepository +if TYPE_CHECKING: + from deerflow.knowledge.embeddings import EmbeddingClient + _TOKEN_RE = re.compile(r"[\w一-鿿]+", re.UNICODE) @@ -33,24 +38,12 @@ def _snippet(content: str, terms: list[str], *, width: int = 160) -> str: return body[:width] + ("…" if len(body) > width else "") -async def search_notes( - repo: KnowledgeRepository, - query: str, - *, - mode: str = "keyword", - limit: int = 8, - candidate_cap: int = 500, -) -> list[dict[str, Any]]: - """Score notes against ``query`` and return the top ``limit`` matches.""" +async def _keyword_scores(repo: KnowledgeRepository, query: str, *, candidate_cap: int) -> dict[str, tuple[float, dict[str, Any]]]: terms = _tokens(query) if not terms: - return [] - - # Pull a candidate window (non-archived). Phase 2 will replace this with - # an index; for the first release in-Python scoring is sufficient. + return {} candidates, _total = await repo.list_notes(limit=candidate_cap, offset=0) - - scored: list[tuple[float, dict[str, Any]]] = [] + scored: dict[str, tuple[float, dict[str, Any]]] = {} for note in candidates: title = (note.get("title") or "").lower() summary = (note.get("summary") or "").lower() @@ -65,22 +58,99 @@ async def search_notes( if term in summary: score += 2.0 score += min(content.count(term), 5) * 0.5 - if score <= 0: - continue - scored.append((score, note)) + if score > 0: + scored[note["id"]] = (score, note) + return scored - scored.sort(key=lambda x: x[0], reverse=True) - max_score = scored[0][0] if scored else 1.0 - results = [] - for score, note in scored[:limit]: + +async def _vector_scores( + repo: KnowledgeRepository, + query: str, + embedding_client: EmbeddingClient, +) -> dict[str, tuple[float, dict[str, Any]]]: + qvec = await embedding_client.embed_query(query) + if not qvec: + return {} + embeddings = await repo.all_embeddings() + best: dict[str, tuple[float, str, str]] = {} # note_id -> (score, title, snippet) + for emb in embeddings: + sim = cosine_similarity(qvec, emb.get("vector") or []) + if sim <= 0: + continue + nid = emb["note_id"] + if nid not in best or sim > best[nid][0]: + best[nid] = (sim, emb.get("title") or "", emb.get("content") or "") + return {nid: (score, {"id": nid, "title": title, "_chunk": chunk}) for nid, (score, title, chunk) in best.items()} + + +def _normalize(scores: dict[str, float]) -> dict[str, float]: + if not scores: + return {} + mx = max(scores.values()) + if mx <= 0: + return {k: 0.0 for k in scores} + return {k: v / mx for k, v in scores.items()} + + +async def search_notes( + repo: KnowledgeRepository, + query: str, + *, + mode: str = "keyword", + limit: int = 8, + candidate_cap: int = 500, + embedding_client: EmbeddingClient | None = None, +) -> list[dict[str, Any]]: + """Score notes against ``query`` and return the top ``limit`` matches.""" + terms = _tokens(query) + if not terms: + return [] + + want_vector = mode in ("vector", "hybrid") and embedding_client is not None and embedding_client.is_enabled() + keyword = await _keyword_scores(repo, query, candidate_cap=candidate_cap) if mode != "vector" or not want_vector else {} + vector = await _vector_scores(repo, query, embedding_client) if want_vector else {} + + # Pure vector (when available); else keyword; hybrid blends both. + if mode == "vector" and want_vector: + combined = {nid: (sc, meta) for nid, (sc, meta) in vector.items()} + ranked = sorted(combined.items(), key=lambda kv: kv[1][0], reverse=True)[:limit] + return await _materialize(repo, ranked, terms) + + if mode == "hybrid" and want_vector: + kw_norm = _normalize({k: v[0] for k, v in keyword.items()}) + vec_norm = _normalize({k: v[0] for k, v in vector.items()}) + ids = set(kw_norm) | set(vec_norm) + blended: dict[str, tuple[float, dict[str, Any]]] = {} + for nid in ids: + score = 0.5 * kw_norm.get(nid, 0.0) + 0.5 * vec_norm.get(nid, 0.0) + meta = keyword.get(nid, (0, None))[1] or vector.get(nid, (0, {}))[1] + blended[nid] = (score, meta) + ranked = sorted(blended.items(), key=lambda kv: kv[1][0], reverse=True)[:limit] + return await _materialize(repo, ranked, terms) + + # keyword (default, and fallback when vector unavailable) + ranked = sorted(keyword.items(), key=lambda kv: kv[1][0], reverse=True)[:limit] + if not ranked and vector: # mode=vector requested but keyword empty: use vector + ranked = sorted(vector.items(), key=lambda kv: kv[1][0], reverse=True)[:limit] + return await _materialize(repo, ranked, terms) + + +async def _materialize(repo: KnowledgeRepository, ranked: list[tuple[str, tuple[float, dict[str, Any]]]], terms: list[str]) -> list[dict[str, Any]]: + max_score = ranked[0][1][0] if ranked else 1.0 + results: list[dict[str, Any]] = [] + for nid, (score, meta) in ranked: + note = meta if (meta and meta.get("content_md")) else await repo.get_note(nid, with_sources=False) + if note is None: + continue + snippet_src = note.get("content_md") or meta.get("_chunk") or note.get("summary") or "" results.append( { - "note_id": note["id"], + "note_id": nid, "chunk_id": None, - "title": note.get("title"), + "title": note.get("title") or meta.get("title"), "summary": note.get("summary"), - "snippet": _snippet(note.get("content_md") or note.get("summary") or "", terms), - "score": round(score / max_score, 4), + "snippet": _snippet(snippet_src, terms), + "score": round(score / max_score, 4) if max_score else 0.0, "tags": note.get("tags") or [], "source_type": note.get("source_type"), "updated_at": note.get("updated_at"), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/service.py b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/service.py index de258ab8..f32d97e8 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/service.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/service.py @@ -16,6 +16,8 @@ from typing import Any from deerflow.config.knowledge_config import KnowledgeConfig from deerflow.config.runtime_paths import runtime_home from deerflow.knowledge import extractor, llm_extractor +from deerflow.knowledge.chunking import chunk_markdown +from deerflow.knowledge.embeddings import EmbeddingClient from deerflow.knowledge.markdown_vault import MarkdownVault from deerflow.knowledge.obsidian_adapter import ObsidianWikiAdapter from deerflow.knowledge.repository import KnowledgeRepository @@ -51,6 +53,7 @@ class KnowledgeService: export_html=config.export.graph_html_enabled, export_json=config.export.graph_json_enabled, ) + self.embedding_client = EmbeddingClient(config.embedding) try: self.adapter.ensure_ready() except Exception: # pragma: no cover - vault init best-effort @@ -79,8 +82,25 @@ class KnowledgeService: logger.exception("Failed to write knowledge note %s to vault", note_id) if draft.sources: await self.repo.add_sources(note_id, draft.sources, created_by=created_by) + await self._reindex_embeddings(note_id, draft.content_md) return await self.repo.get_note(note_id) or note + async def _reindex_embeddings(self, note_id: str, content_md: str) -> None: + """Best-effort: (re)build the per-chunk embedding index for a note.""" + if not self.embedding_client.is_enabled(): + return + try: + chunks = chunk_markdown(content_md) + if not chunks: + await self.repo.delete_embeddings(note_id) + return + vectors = await self.embedding_client.embed_texts(chunks) + if not vectors or len(vectors) != len(chunks): + return + await self.repo.replace_embeddings(note_id, [(i, c, v) for i, (c, v) in enumerate(zip(chunks, vectors, strict=False))]) + except Exception: + logger.exception("Failed to (re)index embeddings for note %s", note_id) + # -- capture ---------------------------------------------------------- async def capture_thread( self, @@ -237,6 +257,11 @@ class KnowledgeService: current = await self.repo.get_note(note_id, with_sources=False) if current is None: return None + # Snapshot the pre-edit state for version history (phase 3). + try: + await self.repo.add_version(current, edited_by=updated_by) + except Exception: + logger.exception("Failed to snapshot version for note %s", note_id) body = content_md if body is not None: _, parsed = MarkdownVault.parse(body) @@ -269,8 +294,13 @@ class KnowledgeService: await self.repo.set_vault_path(note_id, rel) except Exception: logger.exception("Failed to update knowledge note %s in vault", note_id) + if note.get("content_md") is not None: + await self._reindex_embeddings(note_id, note["content_md"]) return await self.repo.get_note(note_id) + async def list_versions(self, note_id: str) -> list[dict[str, Any]]: + return await self.repo.list_versions(note_id) + async def delete_note(self, note_id: str, *, updated_by: str | None = None, hard: bool = False) -> bool: """Soft delete (archive) by default; hard delete removes the vault file.""" current = await self.repo.get_note(note_id, with_sources=False) @@ -282,13 +312,45 @@ class KnowledgeService: if hard and current.get("vault_path"): self.adapter.vault.delete_note(current["vault_path"]) self.adapter.manifest.remove_page(note_id) + await self.repo.delete_embeddings(note_id) except Exception: logger.exception("Failed to archive knowledge note %s in vault", note_id) return True # -- search & graph --------------------------------------------------- async def search(self, query: str, *, mode: str = "keyword", limit: int = 8) -> list[dict[str, Any]]: - return await search_notes(self.repo, query, mode=mode, limit=limit) + return await search_notes(self.repo, query, mode=mode, limit=limit, embedding_client=self.embedding_client) + + async def build_rag_context(self, query: str, *, limit: int = 5) -> dict[str, Any]: + """Retrieve top knowledge for ``query`` and format an injectable block. + + Returns ``{"text": , "items": [...]}``. Uses hybrid + search when embeddings are available, else keyword (wiki-query style). + """ + mode = "hybrid" if self.embedding_client.is_enabled() else "keyword" + items = await search_notes(self.repo, query, mode=mode, limit=limit, embedding_client=self.embedding_client) + if not items: + return {"text": "", "items": []} + lines = [ + "以下是产品知识库中与本问题相关的资料。回答时优先参考;若使用其中信息,请在回答中标注来源(note_id)。", + "", + ] + for i, it in enumerate(items, 1): + lines.append(f"[知识 {i}] {it.get('title') or ''}") + if it.get("snippet"): + lines.append(it["snippet"]) + lines.append(f"来源:{it['note_id']}") + lines.append("") + return {"text": "\n".join(lines).strip(), "items": items} + + async def reindex_all(self) -> dict[str, int]: + """Rebuild embeddings for every non-archived note (admin/maintenance).""" + notes = await self.repo.list_all() + count = 0 + for note in notes: + await self._reindex_embeddings(note["id"], note.get("content_md") or "") + count += 1 + return {"reindexed": count, "embedding_enabled": int(self.embedding_client.is_enabled())} async def export_graph(self) -> dict[str, Any]: from deerflow.utils.time import now_iso diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py index 88e8759c..429b7545 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -25,8 +25,12 @@ from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow from deerflow.persistence.memory_config.model import UserMemoryConfigRow from deerflow.persistence.feedback.model import FeedbackRow from deerflow.knowledge.models import ( + KnowledgeEmbeddingRow, + KnowledgeEntityRow, KnowledgeNoteRow, KnowledgeNoteTagRow, + KnowledgeNoteVersionRow, + KnowledgeRelationRow, KnowledgeSourceRow, KnowledgeTagRow, ) @@ -66,8 +70,12 @@ __all__ = [ "AgentRow", "AgentSkillRow", "AdminLeaderboardDailyStatRow", + "KnowledgeEmbeddingRow", + "KnowledgeEntityRow", "KnowledgeNoteRow", "KnowledgeNoteTagRow", + "KnowledgeNoteVersionRow", + "KnowledgeRelationRow", "KnowledgeSourceRow", "KnowledgeTagRow", "LlmCallMetricRow",