feat(knowledge): 阶段二 向量 RAG 与召回注入

- embeddings.py：OpenAI 兼容 embedding 客户端，未配置时关键词兜底
- chunking.py + knowledge_embeddings 表：Markdown 分块与向量索引（Python 余弦）
- search.py：keyword/vector/hybrid 三种模式，向量不可用自动降级
- 笔记创建/更新自动重建向量索引；新增 /reindex 维护接口
- KnowledgeRagMiddleware：按对话级开关 knowledge_rag_enabled 注入 <knowledge-context>，接入 lead agent 中间件链（默认随 config.knowledge.rag_enabled）
- service.build_rag_context（wiki-query 风格分层召回）
- 笔记版本快照（knowledge_note_versions）+ GET /notes/{id}/versions
- 前端：聊天页「知识库」对话级 RAG 开关 + 类型字段
- 新增 entities/relations 表（阶段四预置）

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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 (
+    <Tooltip content={enabled ? "知识库召回：已开启" : "知识库召回：已关闭"}>
+      <Button
+        variant="ghost"
+        onClick={toggle}
+        className={cn(
+          "text-muted-foreground hover:text-foreground",
+          enabled && "text-primary hover:text-primary",
+        )}
+      >
+        <Database />
+        知识库
+      </Button>
+    </Tooltip>
+  );
+}

frontend-web/src/core/threads/types.ts

@@ -18,9 +18,11 @@ export interface AgentThreadContext extends Record<string, unknown> {
   reasoning_effort?: "minimal" | "low" | "medium" | "high";
   writing_mode?: boolean;
   writing_artifact_path?: string;
-  memory_injection_enabled?: boolean;
+  memory_injection_enabled?: boolean;
-  rag_mode_enabled?: boolean;
+  rag_mode_enabled?: boolean;
-  agent_id?: string;
+  /** 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;
 }

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}
               />
+              <KnowledgeRagToggle threadId={threadId} />
               <ExportTrigger threadId={threadId} />
               <KnowledgeSaveTrigger threadId={threadId} />
               <ArtifactTrigger />

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]:

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

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
+``<knowledge-context>`` 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"</{_TAG}>"
+        )
+        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)

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.")
+    provider: str = Field(default="openai", description="Embedding provider (openai-compatible).")
-    model: str = Field(default="", description="Embedding model name.")
+    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):

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()]

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 []

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))

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:

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);
+- ``keyword`` — title>tags>summary>body scoring over the DB mirror (always works).
-for now ``vector`` and ``hybrid`` transparently fall back to keyword search so
+- ``vector``  — cosine over per-chunk embeddings (requires a configured
-the API contract is stable. Scoring weights title > tags > summary > body.
+  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(
+async def _keyword_scores(repo: KnowledgeRepository, query: str, *, candidate_cap: int) -> dict[str, tuple[float, dict[str, Any]]]:
-    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."""
     terms = _tokens(query)
     if not terms:
-        return []
+        return {}
-
-    # Pull a candidate window (non-archived). Phase 2 will replace this with
-    # an index; for the first release in-Python scoring is sufficient.
     candidates, _total = await repo.list_notes(limit=candidate_cap, offset=0)
-
+    scored: dict[str, tuple[float, dict[str, Any]]] = {}
-    scored: list[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:
+        if score > 0:
-            continue
+            scored[note["id"]] = (score, note)
-        scored.append((score, note))
+    return scored
 
-    scored.sort(key=lambda x: x[0], reverse=True)
+
-    max_score = scored[0][0] if scored else 1.0
+async def _vector_scores(
-    results = []
+    repo: KnowledgeRepository,
-    for score, note in scored[:limit]:
+    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),
+                "snippet": _snippet(snippet_src, terms),
-                "score": round(score / max_score, 4),
+                "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"),

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": <system context or "">, "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

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