feat(knowledge): 阶段三 知识治理 + 阶段四 图谱增强

阶段三（治理）:
- sensitive.py：写入前脱敏 token/密钥/手机号/身份证/邮箱（sensitive_redaction_enabled）
- 版本快照：更新前写 knowledge_note_versions；GET /notes/{id}/versions
- dedup.py：标题/摘要 Jaccard + 内容哈希查重；GET /duplicates、POST /merge（来源并入主条目、重复项归档）
- auto_ingest.py：独立后台队列/线程沉淀，不阻塞问答；中间件 after_agent 在 auto_ingest_enabled 时入队；价值判断（最小字符）+ skip_existing 去重；自动沉淀落 draft 待审

阶段四（图谱）:
- LLM 抽取 entities/relations；service 存 knowledge_entities/knowledge_relations 并连边
- exporter.build_graph 支持实体节点 + note→entity/entity→entity 边；graph.html/json 增强
- GET /api/knowledge/graph 返回产品图谱数据

前端：知识库页「查重合并」对话框 + 相关 api/hooks
配置：sensitive_redaction_enabled / dedup_similarity_threshold / entity_extraction_enabled
测试：tests/test_knowledge.py 20 passed

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

frontend-web/src/core/knowledge/api.ts

@@ -131,3 +131,61 @@ export async function exportKnowledgeGraph() {
 export function knowledgeExportFileUrl(filename: string) {
   return `${baseUrl()}/export/files/${encodeURIComponent(filename)}`;
 }
+
+// --- phase 2/3/4 ---
+
+export interface KnowledgeVersion {
+  id: string;
+  note_id: string;
+  title: string;
+  summary?: string | null;
+  content_md: string;
+  status: string;
+  tags: string[];
+  edited_by?: string | null;
+  created_at?: string | null;
+}
+
+export async function listNoteVersions(noteId: string) {
+  const res = await parseResponse<{ items: KnowledgeVersion[] }>(
+    await apiFetch(`${baseUrl()}/notes/${encodeURIComponent(noteId)}/versions`),
+  );
+  return res.items;
+}
+
+export interface DuplicateGroup {
+  primary: { id: string; title: string | null };
+  duplicates: Array<{ id: string; title: string | null; score: number }>;
+}
+
+export async function listDuplicates() {
+  const res = await parseResponse<{ groups: DuplicateGroup[] }>(
+    await apiFetch(`${baseUrl()}/duplicates`),
+  );
+  return res.groups;
+}
+
+export async function mergeKnowledgeNotes(primaryId: string, duplicateIds: string[]) {
+  return parseResponse<{ note: KnowledgeNote }>(
+    await apiFetch(`${baseUrl()}/merge`, {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ primary_id: primaryId, duplicate_ids: duplicateIds }),
+    }),
+  );
+}
+
+export async function reindexKnowledge() {
+  return parseResponse<{ reindexed: number; embedding_enabled: number }>(
+    await apiFetch(`${baseUrl()}/reindex`, { method: "POST" }),
+  );
+}
+
+export interface KnowledgeGraph {
+  nodes: Array<{ id: string; type: string; label: string; category?: string | null }>;
+  edges: Array<{ id: string; source: string; target: string; label: string; weight: number }>;
+}
+
+export async function getKnowledgeGraph() {
+  return parseResponse<KnowledgeGraph>(await apiFetch(`${baseUrl()}/graph`));
+}

frontend-web/src/core/knowledge/hooks.ts

@@ -12,7 +12,9 @@ import {
   getKnowledgeNote,
   ingestThread,
   ingestThreadsBatch,
+  listDuplicates,
   listKnowledgeNotes,
+  mergeKnowledgeNotes,
   searchKnowledge,
   updateKnowledgeNote,
 } from "./api";
@@ -85,5 +87,19 @@ export function useKnowledgeActions() {
       onSuccess: invalidate,
     }),
     exportGraph: useMutation({ mutationFn: exportKnowledgeGraph }),
+    merge: useMutation({
+      mutationFn: ({ primaryId, duplicateIds }: { primaryId: string; duplicateIds: string[] }) =>
+        mergeKnowledgeNotes(primaryId, duplicateIds),
+      onSuccess: invalidate,
+    }),
   };
 }
+
+export function useKnowledgeDuplicates(enabled: boolean) {
+  return useQuery({
+    queryKey: [...ROOT_KEY, "duplicates"],
+    queryFn: listDuplicates,
+    enabled,
+    refetchOnWindowFocus: false,
+  });
+}

frontend-web/src/pages/KnowledgeBasePage.tsx

@@ -1,5 +1,6 @@
 import {
   BookOpen,
+  Copy,
   ExternalLink,
   Loader2,
   Network,
@@ -37,6 +38,7 @@ import {
 import { Textarea } from "@/components/ui/textarea";
 import {
   useKnowledgeActions,
+  useKnowledgeDuplicates,
   useKnowledgeNote,
   useKnowledgeNotes,
 } from "@/core/knowledge/hooks";
@@ -270,6 +272,83 @@ function BatchImportDialog({
   );
 }
 
+function DedupDialog({
+  open,
+  onOpenChange,
+}: {
+  open: boolean;
+  onOpenChange: (v: boolean) => void;
+}) {
+  const { data: groups = [], isLoading } = useKnowledgeDuplicates(open);
+  const { merge } = useKnowledgeActions();
+
+  const handleMerge = async (primaryId: string, duplicateIds: string[]) => {
+    try {
+      await merge.mutateAsync({ primaryId, duplicateIds });
+      toast.success("已合并重复知识");
+    } catch (error) {
+      toast.error(error instanceof Error ? error.message : "合并失败");
+    }
+  };
+
+  return (
+    <Dialog open={open} onOpenChange={onOpenChange}>
+      <DialogContent className="max-w-2xl">
+        <DialogHeader>
+          <DialogTitle>查重与合并</DialogTitle>
+          <DialogDescription>
+            基于标题/摘要相似度与内容哈希检测重复知识；合并会把重复项归档并把来源并入主条目。
+          </DialogDescription>
+        </DialogHeader>
+        <div className="max-h-[60vh] overflow-auto">
+          {isLoading ? (
+            <div className="text-muted-foreground p-4 text-sm">检测中...</div>
+          ) : groups.length === 0 ? (
+            <div className="text-muted-foreground rounded-md border border-dashed p-6 text-center text-sm">
+              未发现明显重复的知识。
+            </div>
+          ) : (
+            <ul className="flex flex-col gap-3">
+              {groups.map((g) => (
+                <li key={g.primary.id} className="rounded-md border p-3">
+                  <div className="mb-2 text-sm font-medium">
+                    保留：{g.primary.title || g.primary.id}
+                  </div>
+                  <ul className="mb-2 flex flex-col gap-1">
+                    {g.duplicates.map((d) => (
+                      <li key={d.id} className="text-muted-foreground text-xs">
+                        合并：{d.title || d.id}（相似度 {Math.round(d.score * 100)}%）
+                      </li>
+                    ))}
+                  </ul>
+                  <Button
+                    size="sm"
+                    variant="outline"
+                    onClick={() =>
+                      void handleMerge(
+                        g.primary.id,
+                        g.duplicates.map((d) => d.id),
+                      )
+                    }
+                    disabled={merge.isPending}
+                  >
+                    合并到主条目
+                  </Button>
+                </li>
+              ))}
+            </ul>
+          )}
+        </div>
+        <DialogFooter>
+          <Button variant="outline" onClick={() => onOpenChange(false)}>
+            关闭
+          </Button>
+        </DialogFooter>
+      </DialogContent>
+    </Dialog>
+  );
+}
+
 function NoteDetail({ noteId }: { noteId: string }) {
   const navigate = useNavigate();
   const { data: note, isLoading } = useKnowledgeNote(noteId);
@@ -479,6 +558,7 @@ export default function KnowledgeBasePage() {
   const [status, setStatus] = useState("all");
   const [createOpen, setCreateOpen] = useState(false);
   const [batchOpen, setBatchOpen] = useState(false);
+  const [dedupOpen, setDedupOpen] = useState(false);
 
   const params = useMemo(
     () => ({
@@ -535,6 +615,9 @@ export default function KnowledgeBasePage() {
           <Button variant="outline" size="sm" onClick={() => setBatchOpen(true)}>
             <RefreshCw className="size-4" /> 批量导入
           </Button>
+          <Button variant="outline" size="sm" onClick={() => setDedupOpen(true)}>
+            <Copy className="size-4" /> 查重
+          </Button>
           <Button
             variant="outline"
             size="sm"
@@ -592,6 +675,7 @@ export default function KnowledgeBasePage() {
 
       <CreateNoteDialog open={createOpen} onOpenChange={setCreateOpen} />
       <BatchImportDialog open={batchOpen} onOpenChange={setBatchOpen} />
+      <DedupDialog open={dedupOpen} onOpenChange={setDedupOpen} />
     </div>
   );
 }

offline-backend-20260512/backend/app/gateway/deps.py

@@ -143,6 +143,8 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         from deerflow.persistence.ai_writing_sessions import (
             make_ai_writing_session_store,
+        )
+        from deerflow.persistence.ai_writing_sessions import (
             register_default_store as _register_ai_writing_default_store,
         )
 
@@ -163,6 +165,21 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.knowledge_service = make_knowledge_service(sf, config.knowledge)
 
+        # Background auto-sedimentation queue (phase 3) — drains finished
+        # conversations off the chat path. Worker only does work when
+        # ``knowledge.auto_ingest_enabled`` is set; the queue is always wired so
+        # the middleware can enqueue without a hard dependency.
+        from deerflow.knowledge.auto_ingest import AutoIngestQueue, set_auto_ingest_queue
+
+        if app.state.knowledge_service is not None:
+            _auto_ingest_queue = AutoIngestQueue(
+                lambda: getattr(app.state, "knowledge_service", None),
+                min_chars=config.knowledge.auto_ingest_min_chars,
+            )
+            _auto_ingest_queue.start()
+            set_auto_ingest_queue(_auto_ingest_queue)
+            app.state.knowledge_auto_ingest_queue = _auto_ingest_queue
+
         from deerflow.persistence.user_prompts import make_user_prompt_store
 
         app.state.user_prompt_store = make_user_prompt_store(sf)
@@ -185,6 +202,8 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         from deerflow.persistence.roundtable_jobs import (
             make_roundtable_job_store,
+        )
+        from deerflow.persistence.roundtable_jobs import (
             register_default_store as _register_roundtable_job_default_store,
         )
 
@@ -242,6 +261,12 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
             from deerflow.persistence.notifications import set_notification_store
 
             set_notification_store(None)
+            queue = getattr(app.state, "knowledge_auto_ingest_queue", None)
+            if queue is not None:
+                from deerflow.knowledge.auto_ingest import set_auto_ingest_queue
+
+                await queue.stop()
+                set_auto_ingest_queue(None)
             await close_engine()
 
 

offline-backend-20260512/backend/app/gateway/routers/knowledge.py

@@ -304,6 +304,30 @@ async def search(body: SearchRequest, request: Request) -> dict[str, Any]:
     return {"items": items}
 
 
+class MergeRequest(BaseModel):
+    primary_id: str
+    duplicate_ids: list[str] = Field(default_factory=list)
+
+
+@router.get("/duplicates")
+@require_auth
+async def list_duplicates(request: Request) -> dict[str, Any]:
+    """Return suggested duplicate groups (phase 3, heuristic)."""
+    service = get_knowledge_service(request)
+    return {"groups": await service.find_duplicates()}
+
+
+@router.post("/merge")
+@require_auth
+async def merge_notes(body: MergeRequest, request: Request) -> dict[str, Any]:
+    """Merge duplicate notes into a primary note (sources moved, dups archived)."""
+    service = get_knowledge_service(request)
+    note = await service.merge_notes(body.primary_id, body.duplicate_ids, updated_by=get_effective_user_id())
+    if note is None:
+        raise HTTPException(status_code=404, detail="Primary note not found")
+    return {"note": note}
+
+
 @router.post("/reindex")
 @require_auth
 async def reindex_embeddings(request: Request) -> dict[str, Any]:
@@ -320,6 +344,14 @@ async def list_note_versions(note_id: str, request: Request) -> dict[str, Any]:
     return {"items": await service.list_versions(note_id)}
 
 
+@router.get("/graph")
+@require_auth
+async def get_graph(request: Request) -> dict[str, Any]:
+    """Return product knowledge-graph data (notes + entities + relations)."""
+    service = get_knowledge_service(request)
+    return await service.get_graph()
+
+
 @router.post("/export/graph")
 @require_auth
 async def export_graph(request: Request) -> dict[str, Any]:

offline-backend-20260512/backend/config.yaml

@@ -363,6 +363,9 @@ knowledge:
   max_source_items: 20
   rag_enabled: false
   rag_limit: 5
+  # 阶段三：写入前脱敏（token/密钥/手机号/身份证/邮箱）；重复检测相似度阈值。
+  sensitive_redaction_enabled: true
+  dedup_similarity_threshold: 0.82
   export:
     enabled: true
     graph_html_enabled: true

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/knowledge_rag_middleware.py

@@ -21,6 +21,7 @@ from typing import Any, override
 from langchain.agents import AgentState
 from langchain.agents.middleware import AgentMiddleware
 from langchain_core.messages import HumanMessage
+from langgraph.runtime import Runtime
 
 logger = logging.getLogger(__name__)
 
@@ -132,3 +133,58 @@ class KnowledgeRagMiddleware(AgentMiddleware[AgentState]):
         if msg is not None:
             request = request.override(messages=[*request.messages, msg])
         return handler(request)
+
+    # ---- after_agent: enqueue background auto-sedimentation -------------
+    def _resolve_thread_id(self, runtime: Runtime | None) -> str | None:
+        if runtime is not None and getattr(runtime, "context", None):
+            tid = runtime.context.get("thread_id")
+            if tid:
+                return str(tid)
+        try:
+            from langgraph.config import get_config
+
+            return get_config().get("configurable", {}).get("thread_id")
+        except Exception:
+            return None
+
+    @override
+    def after_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
+        """When auto-ingest is on, hand the finished turn to the background queue."""
+        try:
+            from deerflow.config import get_app_config
+
+            knowledge_cfg = getattr(get_app_config(), "knowledge", None)
+            if knowledge_cfg is None or not knowledge_cfg.enabled or not knowledge_cfg.auto_ingest_enabled:
+                return None
+
+            from deerflow.knowledge.auto_ingest import AutoIngestItem, get_auto_ingest_queue
+
+            queue = get_auto_ingest_queue()
+            if queue is None:
+                return None
+
+            messages = state.get("messages") if isinstance(state, dict) else getattr(state, "messages", None)
+            if not messages:
+                return None
+
+            from deerflow.runtime.references import build_reference_batches
+            from deerflow.runtime.serialization import serialize_lc_object
+            from deerflow.runtime.user_context import get_effective_user_id
+
+            serialized = [serialize_lc_object(m) for m in messages]
+            thread_id = self._resolve_thread_id(runtime)
+            if not thread_id:
+                return None
+            title = state.get("title") if isinstance(state, dict) else getattr(state, "title", None)
+            queue.enqueue(
+                AutoIngestItem(
+                    thread_id=thread_id,
+                    user_id=get_effective_user_id(),
+                    messages=serialized,
+                    thread_title=title if isinstance(title, str) else None,
+                    reference_batches=build_reference_batches(serialized),
+                )
+            )
+        except Exception:
+            logger.exception("KnowledgeRagMiddleware.after_agent enqueue failed (non-fatal)")
+        return None

offline-backend-20260512/backend/packages/harness/deerflow/config/knowledge_config.py

@@ -64,6 +64,9 @@ class KnowledgeConfig(BaseModel):
     max_source_items: int = Field(default=20, description="Maximum source items recorded per note.")
     rag_enabled: bool = Field(default=False, description="Enable RAG injection before agent runs (phase 2).")
     rag_limit: int = Field(default=5, description="Maximum knowledge snippets injected per query.")
+    sensitive_redaction_enabled: bool = Field(default=True, description="Redact secrets/PII (tokens, keys, phone, id-card, email) before persisting notes (phase 3).")
+    dedup_similarity_threshold: float = Field(default=0.82, description="Title/summary similarity above which two notes are flagged as duplicates (phase 3).")
+    entity_extraction_enabled: bool = Field(default=True, description="Extract entities/relations during LLM distillation and feed them into the graph (phase 4).")
     export: KnowledgeExportConfig = Field(default_factory=KnowledgeExportConfig, description="Graph export configuration.")
     query: KnowledgeQueryConfig = Field(default_factory=KnowledgeQueryConfig, description="Layered query configuration.")
     embedding: KnowledgeEmbeddingConfig = Field(default_factory=KnowledgeEmbeddingConfig, description="Embedding configuration.")

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/auto_ingest.py

@@ -0,0 +1,125 @@
+"""Background auto-sediment queue (phase 3).
+
+A single asyncio worker drains a bounded queue of finished conversations and
+sediments the worthwhile ones into the knowledge base — entirely off the chat
+request path, so it never adds latency to a reply. Enqueue happens from
+``KnowledgeRagMiddleware.after_agent`` when ``knowledge.auto_ingest_enabled``.
+
+Value judgment is intentionally cheap (min assistant chars + dedup via
+``skip_existing``); the LLM still distills the kept conversations via the normal
+``capture_thread`` path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_QUEUE: AutoIngestQueue | None = None
+
+
+@dataclass
+class _QueueUser:
+    """Minimal CurrentUser stand-in (only ``.id`` is needed for context)."""
+
+    id: str
+
+
+@dataclass
+class AutoIngestItem:
+    thread_id: str
+    user_id: str
+    messages: list[dict[str, Any]]
+    thread_title: str | None = None
+    reference_batches: list[dict[str, Any]] = field(default_factory=list)
+
+
+class AutoIngestQueue:
+    """Bounded background queue + single worker for auto-sedimentation."""
+
+    def __init__(self, service_provider: Callable[[], Any], *, min_chars: int = 300, maxsize: int = 256) -> None:
+        self._provider = service_provider
+        self._min_chars = min_chars
+        self._queue: asyncio.Queue[AutoIngestItem] = asyncio.Queue(maxsize=maxsize)
+        self._task: asyncio.Task | None = None
+
+    def start(self) -> None:
+        if self._task is None or self._task.done():
+            self._task = asyncio.create_task(self._worker(), name="knowledge-auto-ingest")
+            logger.info("Knowledge auto-ingest worker started")
+
+    async def stop(self) -> None:
+        if self._task is not None:
+            self._task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._task
+            self._task = None
+
+    def enqueue(self, item: AutoIngestItem) -> None:
+        try:
+            self._queue.put_nowait(item)
+        except asyncio.QueueFull:
+            logger.warning("Knowledge auto-ingest queue full; dropping thread %s", item.thread_id)
+
+    def _worth_ingesting(self, item: AutoIngestItem) -> bool:
+        chars = 0
+        for m in item.messages:
+            if str(m.get("type") or m.get("role") or "").lower() == "ai":
+                content = m.get("content")
+                chars += len(content) if isinstance(content, str) else len(str(content or ""))
+        return chars >= self._min_chars
+
+    async def _process(self, item: AutoIngestItem) -> None:
+        service = self._provider()
+        if service is None:
+            return
+        if not self._worth_ingesting(item):
+            logger.debug("Auto-ingest skipped (below min chars) thread=%s", item.thread_id)
+            return
+        from deerflow.runtime.user_context import reset_current_user, set_current_user
+
+        token = set_current_user(_QueueUser(id=item.user_id))
+        try:
+            result = await service.capture_thread(
+                thread_id=item.thread_id,
+                messages=item.messages,
+                thread_title=item.thread_title,
+                status="draft",  # auto-sedimented notes land as drafts for review
+                include_sources=True,
+                reference_batches=item.reference_batches,
+                created_by=item.user_id,
+                skip_existing=True,
+            )
+            if result.get("created"):
+                logger.info("Auto-ingested thread %s -> note %s", item.thread_id, result["note"]["id"])
+        except Exception:
+            logger.exception("Auto-ingest failed for thread %s", item.thread_id)
+        finally:
+            reset_current_user(token)
+
+    async def _worker(self) -> None:
+        while True:
+            item = await self._queue.get()
+            try:
+                await self._process(item)
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("Auto-ingest worker error")
+            finally:
+                self._queue.task_done()
+
+
+def set_auto_ingest_queue(queue: AutoIngestQueue | None) -> None:
+    global _QUEUE
+    _QUEUE = queue
+
+
+def get_auto_ingest_queue() -> AutoIngestQueue | None:
+    return _QUEUE

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/dedup.py

@@ -0,0 +1,75 @@
+"""Duplicate detection for knowledge notes (phase 3).
+
+First-release heuristic: token Jaccard over ``title + summary + tags`` plus an
+exact ``content_md`` hash match. Cheap, deterministic and embedding-free so it
+works on offline deployments. Returns *suggestions*; merging is an explicit,
+user-driven action.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from deerflow.knowledge.manifest import content_hash
+
+_TOKEN_RE = re.compile(r"[\w一-鿿]+", re.UNICODE)
+
+
+def _tokens(*texts: str) -> set[str]:
+    out: set[str] = set()
+    for t in texts:
+        out.update(m.lower() for m in _TOKEN_RE.findall(t or ""))
+    return out
+
+
+def _jaccard(a: set[str], b: set[str]) -> float:
+    if not a or not b:
+        return 0.0
+    inter = len(a & b)
+    union = len(a | b)
+    return inter / union if union else 0.0
+
+
+def find_duplicate_groups(notes: list[dict[str, Any]], *, threshold: float = 0.82, max_notes: int = 800) -> list[dict[str, Any]]:
+    """Return groups ``{primary, duplicates:[{id,title,score,reason}]}``.
+
+    ``primary`` is the most recently updated note in each cluster.
+    """
+    notes = notes[:max_notes]
+    feats: list[tuple[dict[str, Any], set[str], str]] = []
+    for n in notes:
+        toks = _tokens(n.get("title") or "", n.get("summary") or "", " ".join(n.get("tags") or []))
+        feats.append((n, toks, content_hash(n.get("content_md") or "")))
+
+    used: set[str] = set()
+    groups: list[dict[str, Any]] = []
+    for i in range(len(feats)):
+        ni, ti, hi = feats[i]
+        if ni["id"] in used:
+            continue
+        dups: list[dict[str, Any]] = []
+        for j in range(i + 1, len(feats)):
+            nj, tj, hj = feats[j]
+            if nj["id"] in used:
+                continue
+            if hi and hi == hj:
+                score, reason = 1.0, "identical-content"
+            else:
+                score = _jaccard(ti, tj)
+                reason = "similar-title-summary"
+            if score >= threshold:
+                dups.append({"id": nj["id"], "title": nj.get("title"), "score": round(score, 4), "reason": reason})
+                used.add(nj["id"])
+        if dups:
+            used.add(ni["id"])
+            # Primary = newest by updated_at.
+            cluster = [ni, *[next(f[0] for f in feats if f[0]["id"] == d["id"]) for d in dups]]
+            primary = max(cluster, key=lambda x: x.get("updated_at") or "")
+            others = [
+                {"id": c["id"], "title": c.get("title"), "score": next((d["score"] for d in dups if d["id"] == c["id"]), 1.0)}
+                for c in cluster
+                if c["id"] != primary["id"]
+            ]
+            groups.append({"primary": {"id": primary["id"], "title": primary.get("title")}, "duplicates": others})
+    return groups

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/exporter.py

@@ -25,16 +25,33 @@ def _vault_key(vault_path: str | None) -> str | None:
     return key
 
 
-def build_graph(notes: list[dict[str, Any]]) -> dict[str, Any]:
+def build_graph(
-    """Return ``{nodes, edges}`` from note dicts (id, title, tags, vault_path, content_md)."""
+    notes: list[dict[str, Any]],
+    entities: list[dict[str, Any]] | None = None,
+    relations: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Return ``{nodes, edges}`` from notes (+ optional entities/relations).
+
+    Note nodes come from ``notes`` (id, title, tags, vault_path, content_md);
+    when ``entities``/``relations`` are supplied (phase 4), entity nodes and
+    note→entity / entity→entity edges are added on top of the wikilink and
+    shared-tag edges.
+    """
     nodes = []
     by_key: dict[str, str] = {}
+    note_ids: set[str] = set()
     for n in notes:
         nodes.append({"id": n["id"], "type": "note", "label": n.get("title") or n["id"], "category": n.get("source_type")})
+        note_ids.add(n["id"])
         key = _vault_key(n.get("vault_path"))
         if key:
             by_key[key] = n["id"]
 
+    entity_ids: set[str] = set()
+    for ent in entities or []:
+        nodes.append({"id": ent["id"], "type": "entity", "label": ent.get("name") or ent["id"], "category": ent.get("entity_type")})
+        entity_ids.add(ent["id"])
+
     edges: list[dict[str, Any]] = []
     edge_seen: set[tuple[str, str, str]] = set()
 
@@ -66,6 +83,14 @@ def build_graph(notes: list[dict[str, Any]]) -> dict[str, Any]:
             for j in range(i + 1, len(ids)):
                 add_edge(ids[i], ids[j], "related", 1)
 
+    # Entity / relation edges (phase 4) — only when both endpoints exist as nodes.
+    valid = note_ids | entity_ids
+    for rel in relations or []:
+        src = rel.get("from_note_id") or rel.get("from_entity_id")
+        tgt = rel.get("to_note_id") or rel.get("to_entity_id")
+        if src in valid and tgt in valid:
+            add_edge(src, tgt, rel.get("relation_type") or "related", float(rel.get("weight") or 1))
+
     return {"nodes": nodes, "edges": edges}
 
 
@@ -109,13 +134,15 @@ def export_all(
     vault_root: Path | str,
     notes: list[dict[str, Any]],
     *,
+    entities: list[dict[str, Any]] | None = None,
+    relations: list[dict[str, Any]] | None = None,
     write_html: bool = True,
     write_json: bool = True,
 ) -> dict[str, Any]:
     """Generate graph artifacts under ``<vault>/wiki-export/`` and return stats."""
     out_dir = Path(vault_root).resolve() / "wiki-export"
     out_dir.mkdir(parents=True, exist_ok=True)
-    graph = build_graph(notes)
+    graph = build_graph(notes, entities, relations)
     written: list[str] = []
     if write_json:
         (out_dir / "graph.json").write_text(json.dumps(graph, ensure_ascii=False, indent=2), encoding="utf-8")

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/llm_extractor.py

@@ -52,6 +52,8 @@ references（外部来源汇总）、decision（架构/设计决策）、journal
   "reasoning": "为什么成立、取舍、不确定性",
   "implications": ["后续建议/风险/下一步", "..."],
   "tags": ["标签1", "标签2"],
+  "entities": [{"name": "实体名", "type": "technology|concept|person|org|product|other"}],
+  "relations": [{"from": "实体A", "to": "实体B", "type": "uses|relates_to|part_of|depends_on"}],
   "confidence": 0.0
 }"""
 
@@ -192,6 +194,9 @@ async def extract_thread_llm(
     except (TypeError, ValueError):
         confidence = 0.7
 
+    entities = data.get("entities") if isinstance(data.get("entities"), list) else []
+    relations = data.get("relations") if isinstance(data.get("relations"), list) else []
+
     return NoteDraft(
         title=final_title,
         summary=summary or final_title,
@@ -203,5 +208,7 @@ async def extract_thread_llm(
         confidence=max(0.0, min(confidence, 1.0)),
         related=related,
         sources=sources,
+        entities=[e for e in entities if isinstance(e, dict) and e.get("name")],
+        relations=[r for r in relations if isinstance(r, dict) and r.get("from") and r.get("to")],
         source_key=f"thread:{thread_id}",
     )

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/obsidian_adapter.py

@@ -143,9 +143,23 @@ class ObsidianWikiAdapter:
         self._remove_from_index(note_id)
         self._append_log("ARCHIVE", note_id, "", updated)
 
-    def export_graph(self, notes: list[dict[str, Any]], *, updated: str) -> dict[str, Any]:
+    def export_graph(
+        self,
+        notes: list[dict[str, Any]],
+        *,
+        entities: list[dict[str, Any]] | None = None,
+        relations: list[dict[str, Any]] | None = None,
+        updated: str,
+    ) -> dict[str, Any]:
         self.ensure_ready()
-        result = exporter.export_all(self.root, notes, write_html=self.export_html, write_json=self.export_json)
+        result = exporter.export_all(
+            self.root,
+            notes,
+            entities=entities,
+            relations=relations,
+            write_html=self.export_html,
+            write_json=self.export_json,
+        )
         self._append_log("EXPORT", "graph", f"nodes={result['stats']['nodes']} edges={result['stats']['edges']}", updated)
         return result
 

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/schemas.py

@@ -53,6 +53,10 @@ class NoteDraft:
     actions: list[str] = field(default_factory=list)
     related: list[str] = field(default_factory=list)
     sources: list[SourceDraft] = field(default_factory=list)
+    # Phase 4: extracted graph signal. entities=[{name,type}],
+    # relations=[{from,to,type}] (names refer to entities or this note's title).
+    entities: list[dict] = field(default_factory=list)
+    relations: list[dict] = field(default_factory=list)
     # Stable identity hint used for manifest dedup (e.g. ``thread:thread_id``).
     source_key: str | None = None
 

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/sensitive.py

@@ -0,0 +1,54 @@
+"""Sensitive-information redaction for knowledge content (phase 3).
+
+Best-effort masking of secrets / PII before a note is persisted: API keys,
+bearer tokens, generic ``key=secret`` assignments, Chinese mobile numbers,
+ID-card numbers and email addresses. Conservative by design — it favours not
+mangling legitimate prose over catching every possible secret.
+"""
+
+from __future__ import annotations
+
+import re
+
+_MASK = "[REDACTED]"
+
+# Order matters: more specific patterns first.
+_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
+    ("openai_key", re.compile(r"\bsk-[A-Za-z0-9]{20,}\b")),
+    ("aws_key", re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b")),
+    ("github_token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b")),
+    ("bearer", re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._\-]{16,}")),
+    # key/secret/password/token assignment: keep the field name, mask the value.
+    ("assignment", re.compile(r"(?i)\b(api[_-]?key|secret|password|passwd|token|access[_-]?token)\b\s*[:=]\s*[\"']?[^\s\"']{6,}[\"']?")),
+    ("cn_id_card", re.compile(r"\b\d{17}[\dXx]\b")),
+    ("cn_mobile", re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")),
+    ("email", re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b")),
+]
+
+
+def redact(text: str | None) -> tuple[str, list[str]]:
+    """Return ``(redacted_text, hit_kinds)``.
+
+    ``hit_kinds`` lists which detectors fired (for logging/audit). When ``text``
+    is falsy it is returned unchanged with no hits.
+    """
+    if not text:
+        return text or "", []
+    hits: list[str] = []
+    out = text
+    for kind, pattern in _PATTERNS:
+        def _sub(m: re.Match[str], _kind: str = kind) -> str:
+            # For assignments keep the field name + delimiter, mask only the value.
+            if _kind == "assignment":
+                head = re.split(r"[:=]", m.group(0), maxsplit=1)[0]
+                sep = ":" if ":" in m.group(0) else "="
+                return f"{head}{sep} {_MASK}"
+            if _kind == "bearer":
+                return f"Bearer {_MASK}"
+            return _MASK
+
+        new_out, n = pattern.subn(_sub, out)
+        if n:
+            hits.append(kind)
+            out = new_out
+    return out, hits

offline-backend-20260512/backend/packages/harness/deerflow/knowledge/service.py

@@ -21,7 +21,7 @@ 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
-from deerflow.knowledge.schemas import NoteDraft, normalize_source_type, normalize_status
+from deerflow.knowledge.schemas import NoteDraft, SourceDraft, normalize_source_type, normalize_status
 from deerflow.knowledge.search import search_notes
 from deerflow.knowledge.sources import from_reference_batches, from_search_results
 
@@ -60,8 +60,24 @@ class KnowledgeService:
             logger.exception("Failed to initialize knowledge vault at %s", vault_root)
 
     # -- internal helpers -------------------------------------------------
+    def _redact_draft(self, draft: NoteDraft) -> NoteDraft:
+        """Mask secrets/PII in a draft before persistence (phase 3)."""
+        if not self.config.sensitive_redaction_enabled:
+            return draft
+        from deerflow.knowledge.sensitive import redact
+
+        draft.content_md, body_hits = redact(draft.content_md)
+        draft.summary, _ = redact(draft.summary)
+        for src in draft.sources:
+            if src.snippet:
+                src.snippet, _ = redact(src.snippet)
+        if body_hits:
+            logger.info("Knowledge redaction applied to note '%s': %s", draft.title, ",".join(sorted(set(body_hits))))
+        return draft
+
     async def _persist_draft(self, draft: NoteDraft, *, created_by: str | None) -> dict[str, Any]:
         """Create a DB row, write the vault file, attach sources, return note."""
+        draft = self._redact_draft(draft)
         note = await self.repo.create_note(
             title=draft.title,
             summary=draft.summary,
@@ -83,8 +99,32 @@ class KnowledgeService:
         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)
+        await self._store_entities(note_id, draft)
         return await self.repo.get_note(note_id) or note
 
+    async def _store_entities(self, note_id: str, draft: NoteDraft) -> None:
+        """Persist extracted entities/relations and link them to the note (phase 4)."""
+        if not self.config.entity_extraction_enabled or not draft.entities:
+            return
+        try:
+            await self.repo.clear_note_relations(note_id)
+            name_to_id: dict[str, str] = {}
+            for ent in draft.entities:
+                name = str(ent.get("name") or "").strip()
+                if not name:
+                    continue
+                row = await self.repo.get_or_create_entity(name, ent.get("type"))
+                name_to_id[name.lower()] = row["id"]
+                # note -> entity "mentions"
+                await self.repo.add_relation(from_note_id=note_id, to_entity_id=row["id"], relation_type="mentions")
+            for rel in draft.relations:
+                a = name_to_id.get(str(rel.get("from") or "").strip().lower())
+                b = name_to_id.get(str(rel.get("to") or "").strip().lower())
+                if a and b and a != b:
+                    await self.repo.add_relation(from_entity_id=a, to_entity_id=b, relation_type=str(rel.get("type") or "relates_to"))
+        except Exception:
+            logger.exception("Failed to store entities/relations for note %s", note_id)
+
     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():
@@ -343,6 +383,56 @@ class KnowledgeService:
             lines.append("")
         return {"text": "\n".join(lines).strip(), "items": items}
 
+    # -- dedup & merge (phase 3) -----------------------------------------
+    async def find_duplicates(self) -> list[dict[str, Any]]:
+        from deerflow.knowledge.dedup import find_duplicate_groups
+
+        notes = await self.repo.list_all()
+        return find_duplicate_groups(notes, threshold=self.config.dedup_similarity_threshold)
+
+    async def merge_notes(self, primary_id: str, duplicate_ids: list[str], *, updated_by: str | None = None) -> dict[str, Any] | None:
+        """Merge ``duplicate_ids`` into ``primary_id``.
+
+        Sources from each duplicate are re-attached to the primary, a wikilink
+        + "Merged from" note is appended to the primary body, the duplicates are
+        archived (soft delete), and a ``merged_into`` relation is recorded.
+        """
+        primary = await self.repo.get_note(primary_id)
+        if primary is None:
+            return None
+        merged_titles: list[str] = []
+        for dup_id in duplicate_ids:
+            if dup_id == primary_id:
+                continue
+            dup = await self.repo.get_note(dup_id)
+            if dup is None:
+                continue
+            # Move sources.
+            for src in dup.get("sources") or []:
+                await self.repo.add_sources(
+                    primary_id,
+                    [
+                        SourceDraft(
+                            source_type=src.get("source_type") or "tool",
+                            thread_id=src.get("thread_id"),
+                            message_id=src.get("message_id"),
+                            tool_name=src.get("tool_name"),
+                            title=src.get("title"),
+                            url=src.get("url"),
+                            snippet=src.get("snippet"),
+                        )
+                    ],
+                    created_by=updated_by,
+                )
+            await self.repo.add_relation(from_note_id=dup_id, to_note_id=primary_id, relation_type="merged_into")
+            await self.delete_note(dup_id, updated_by=updated_by)
+            merged_titles.append(dup.get("title") or dup_id)
+
+        if merged_titles:
+            appended = primary["content_md"].rstrip() + "\n\n## Merged\n\n" + "\n".join(f"- {t}" for t in merged_titles) + "\n"
+            await self.update_note(primary_id, content_md=appended, updated_by=updated_by)
+        return await self.repo.get_note(primary_id)
+
     async def reindex_all(self) -> dict[str, int]:
         """Rebuild embeddings for every non-archived note (admin/maintenance)."""
         notes = await self.repo.list_all()
@@ -356,8 +446,19 @@ class KnowledgeService:
         from deerflow.utils.time import now_iso
 
         notes = await self.repo.list_all()
-        result = self.adapter.export_graph(notes, updated=now_iso())
+        entities = await self.repo.list_entities() if self.config.entity_extraction_enabled else []
+        relations = await self.repo.list_relations() if self.config.entity_extraction_enabled else []
+        result = self.adapter.export_graph(notes, entities=entities, relations=relations, updated=now_iso())
         return {"files": result["files"], "stats": result["stats"]}
 
+    async def get_graph(self) -> dict[str, Any]:
+        """Return product graph data ``{nodes, edges}`` (notes + entities)."""
+        from deerflow.knowledge import exporter
+
+        notes = await self.repo.list_all()
+        entities = await self.repo.list_entities() if self.config.entity_extraction_enabled else []
+        relations = await self.repo.list_relations() if self.config.entity_extraction_enabled else []
+        return exporter.build_graph(notes, entities, relations)
+
     def export_file_path(self, filename: str) -> Path | None:
         return self.adapter.export_file_path(filename)