AdministratorandClaude Opus 4.8
e7ff07debd
- project_name 默认 deerflow → zncm（config.yaml + 配置默认 + 模块内字面量）
- related_links() 返回空：笔记 Related 不再注入 index/_meta/taxonomy/hub 结构链接，
只保留真实相关笔记的可点 [[标题]]；无相关笔记且无来源时不再渲染 ## Related 段
- 移除随之多余的项目 hub 页脚手架（index.md/_meta/taxonomy.md 仍由 obsidian 脚手架保留，可解析）
- GET /notes/{id} 用 auth provider 把 created_by/updated_by(UUID) 解析为邮箱
(created_by_name/updated_by_name)，前端 footer 显示人名、解析不到才显示短 id
- 仅对新沉淀笔记生效，现有笔记不动
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
630 lines
28 KiB
Python
630 lines
28 KiB
Python
"""KnowledgeService — high-level orchestration for the knowledge base.
|
|
|
|
Ties together the repository (DB mirror), the :class:`ObsidianWikiAdapter`
|
|
(Markdown vault + index/log/hot/manifest + graph export) and the extractor
|
|
(thread/search → :class:`NoteDraft`). The Gateway router only talks to this
|
|
service; everything obsidian-wiki specific stays inside the ``knowledge``
|
|
module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
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, VaultPathError, extract_wikilinks, normalize_wikilink_target
|
|
from deerflow.knowledge.obsidian_adapter import ObsidianWikiAdapter
|
|
from deerflow.knowledge.repository import KnowledgeRepository
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EmptyThreadError(ValueError):
|
|
"""Raised when a thread has no sediment-able content."""
|
|
|
|
|
|
def _resolve_vault_dir(vault_dir: str) -> Path:
|
|
path = Path(vault_dir)
|
|
if path.is_absolute():
|
|
return path.resolve()
|
|
return (runtime_home() / vault_dir).resolve()
|
|
|
|
|
|
def _inject_related_notes(body: str, links: list[str]) -> str:
|
|
"""Insert real related-note ``[[wikilinks]]`` into the Related block.
|
|
|
|
Notes carry only content-specific related notes now (the structural meta-links
|
|
were dropped), so this is the sole writer of ``## Related`` links. When the
|
|
body has no ``## Related`` section yet, one is appended.
|
|
"""
|
|
if not links:
|
|
return body
|
|
bullets = "\n".join(f"- {link}" for link in links)
|
|
marker = "## Related\n"
|
|
idx = body.find(marker)
|
|
if idx == -1:
|
|
return body.rstrip() + "\n\n## Related\n\n" + bullets + "\n"
|
|
insert_at = idx + len(marker)
|
|
return body[:insert_at] + "\n" + bullets + "\n" + body[insert_at:]
|
|
|
|
|
|
class KnowledgeService:
|
|
"""Application-facing knowledge base API."""
|
|
|
|
def __init__(self, repo: KnowledgeRepository, config: KnowledgeConfig) -> None:
|
|
self.repo = repo
|
|
self.config = config
|
|
self.project_name = config.project_name
|
|
vault_root = _resolve_vault_dir(config.vault_dir)
|
|
self.adapter = ObsidianWikiAdapter(
|
|
vault_root,
|
|
project_name=config.project_name,
|
|
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
|
|
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 _find_related_notes(self, draft: NoteDraft) -> list[dict[str, Any]]:
|
|
"""Find existing notes on the same topic to cross-link under ## Related.
|
|
|
|
Uses the existing keyword/hybrid search over the title (+ summary). The
|
|
draft's own note does not exist yet, so we exclude same-title hits to
|
|
avoid self-links when re-capturing or deduping. Best-effort: any failure
|
|
yields no links rather than blocking the capture.
|
|
"""
|
|
if not self.config.related_notes_enabled:
|
|
return []
|
|
# Build a topical query from the title + summary + the note's *distinctive*
|
|
# tags (the per-note project/knowledge-base defaults are dropped, else
|
|
# every note would match every other). CJK runs tokenize as a single
|
|
# token, so shared distinctive tags are what make cross-topic matches work.
|
|
default_tags = {t.lower() for t in (self.project_name or "", "knowledge-base", "zncm") if t}
|
|
distinctive = [t for t in (draft.tags or []) if t and t.lower() not in default_tags]
|
|
query = " ".join(p for p in [draft.title or "", draft.summary or "", *distinctive] if p).strip()
|
|
if not query:
|
|
return []
|
|
limit = max(1, self.config.related_notes_limit)
|
|
mode = "hybrid" if self.embedding_client.is_enabled() else "keyword"
|
|
try:
|
|
items = await search_notes(self.repo, query, mode=mode, limit=limit + 3, embedding_client=self.embedding_client)
|
|
except Exception:
|
|
logger.exception("Failed to find related notes for '%s'", draft.title)
|
|
return []
|
|
own_title = (draft.title or "").strip().lower()
|
|
out: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for it in items:
|
|
nid = it.get("note_id")
|
|
title = (it.get("title") or "").strip()
|
|
if not nid or nid in seen or not title:
|
|
continue
|
|
if title.lower() == own_title: # the draft itself / a duplicate
|
|
continue
|
|
seen.add(nid)
|
|
out.append({"id": nid, "title": title})
|
|
if len(out) >= limit:
|
|
break
|
|
return out
|
|
|
|
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)
|
|
related_notes = await self._find_related_notes(draft)
|
|
if related_notes:
|
|
real_links = [f"[[{n['title']}]]" for n in related_notes]
|
|
draft.content_md = _inject_related_notes(draft.content_md, real_links)
|
|
draft.related = list(dict.fromkeys([*real_links, *draft.related]))
|
|
note = await self.repo.create_note(
|
|
title=draft.title,
|
|
summary=draft.summary,
|
|
content_md=draft.content_md,
|
|
source_type=draft.source_type,
|
|
source_id=draft.source_id,
|
|
status=draft.status,
|
|
confidence=draft.confidence,
|
|
tags=draft.tags,
|
|
created_by=created_by,
|
|
)
|
|
note_id = note["id"]
|
|
try:
|
|
rel = self.adapter.write_note(note_id, draft, created=note["created_at"], updated=note["updated_at"], status=draft.status)
|
|
await self.repo.set_vault_path(note_id, rel)
|
|
note["vault_path"] = rel
|
|
except Exception:
|
|
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)
|
|
await self._store_entities(note_id, draft)
|
|
# Record note→note "related" edges (after _store_entities, which clears
|
|
# this note's relations) so the graph reflects the cross-links too.
|
|
for rel_note in related_notes:
|
|
try:
|
|
await self.repo.add_relation(from_note_id=note_id, to_note_id=rel_note["id"], relation_type="related")
|
|
except Exception:
|
|
logger.exception("Failed to link related note %s -> %s", note_id, rel_note["id"])
|
|
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():
|
|
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,
|
|
*,
|
|
thread_id: str,
|
|
messages: list[dict[str, Any]],
|
|
thread_title: str | None = None,
|
|
mode: str = "summary",
|
|
title: str | None = None,
|
|
status: str = "approved",
|
|
tags: list[str] | None = None,
|
|
include_sources: bool = True,
|
|
reference_batches: list[dict[str, Any]] | None = None,
|
|
created_by: str | None = None,
|
|
skip_existing: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Sediment a thread into a knowledge note.
|
|
|
|
Returns ``{"note": <note>, "created": bool}``. Raises
|
|
:class:`EmptyThreadError` when there is nothing worth keeping.
|
|
"""
|
|
if skip_existing:
|
|
existing = await self.repo.find_by_source("thread", thread_id)
|
|
if existing is None and self.adapter.manifest.has_source(f"thread:{thread_id}"):
|
|
existing = await self.repo.find_by_source("thread", thread_id)
|
|
if existing:
|
|
return {"note": existing, "created": False}
|
|
|
|
sources = []
|
|
if include_sources and reference_batches:
|
|
sources = from_reference_batches(reference_batches, thread_id=thread_id, max_items=self.config.max_source_items)
|
|
|
|
drafts: list[NoteDraft] = []
|
|
# Prefer LLM distillation (wiki-capture style) so notes are declarative
|
|
# knowledge, not raw chat. A single conversation may split into several
|
|
# cross-linked notes. Falls back to the rule-based extractor (always one
|
|
# note) when the LLM is disabled/unavailable or returns nothing usable.
|
|
if self.config.llm_extract_enabled:
|
|
drafts = await llm_extractor.extract_thread_llm_multi(
|
|
messages,
|
|
thread_id=thread_id,
|
|
thread_title=thread_title,
|
|
mode=mode,
|
|
title=title,
|
|
tags=tags,
|
|
project_name=self.project_name,
|
|
sources=sources,
|
|
max_messages=self.config.max_thread_messages,
|
|
model_name=self.config.extract_model_name,
|
|
max_input_chars=self.config.extract_max_input_chars,
|
|
allow_multi=self.config.multi_note_enabled,
|
|
)
|
|
if not drafts:
|
|
single = extractor.extract_thread(
|
|
messages,
|
|
thread_id=thread_id,
|
|
thread_title=thread_title,
|
|
mode=mode,
|
|
title=title,
|
|
tags=tags,
|
|
project_name=self.project_name,
|
|
sources=sources,
|
|
max_messages=self.config.max_thread_messages,
|
|
)
|
|
drafts = [single] if single else []
|
|
if not drafts:
|
|
raise EmptyThreadError(f"Thread {thread_id} has no sediment-able content")
|
|
|
|
normalized = normalize_status(status)
|
|
for draft in drafts:
|
|
draft.status = normalized
|
|
# Cross-link sibling notes from the same conversation by title, so the
|
|
# split pages reference each other (resolved once all are persisted).
|
|
if len(drafts) > 1:
|
|
for i, draft in enumerate(drafts):
|
|
siblings = [f"[[{o.title}]]" for j, o in enumerate(drafts) if j != i and o.title]
|
|
if siblings:
|
|
draft.content_md = _inject_related_notes(draft.content_md, siblings)
|
|
draft.related = list(dict.fromkeys([*siblings, *draft.related]))
|
|
|
|
notes = [await self._persist_draft(draft, created_by=created_by) for draft in drafts]
|
|
return {"note": notes[0], "created": True, "notes": notes}
|
|
|
|
async def capture_search(
|
|
self,
|
|
*,
|
|
query: str,
|
|
results: list[dict[str, Any]] | None,
|
|
thread_id: str | None = None,
|
|
message_id: str | None = None,
|
|
title: str | None = None,
|
|
tags: list[str] | None = None,
|
|
status: str = "approved",
|
|
created_by: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Sediment search/tool results into a reference note."""
|
|
sources = from_search_results(query, results, thread_id=thread_id, message_id=message_id, max_items=self.config.max_source_items)
|
|
draft = extractor.extract_search(query, sources, thread_id=thread_id, project_name=self.project_name, title=title, tags=tags)
|
|
draft.status = normalize_status(status)
|
|
return await self._persist_draft(draft, created_by=created_by)
|
|
|
|
async def create_manual_note(
|
|
self,
|
|
*,
|
|
title: str,
|
|
content_md: str,
|
|
tags: list[str] | None = None,
|
|
status: str = "approved",
|
|
summary: str | None = None,
|
|
created_by: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a note directly from product-edited Markdown."""
|
|
# Strip any frontmatter the user may have pasted; vault writer adds its own.
|
|
_, body = MarkdownVault.parse(content_md)
|
|
body = body or content_md
|
|
base_tags = list(tags or [])
|
|
for default in (self.project_name, "knowledge-base"):
|
|
if default and default not in base_tags:
|
|
base_tags.append(default)
|
|
draft = NoteDraft(
|
|
title=title.strip() or "知识笔记",
|
|
summary=(summary or body.strip().splitlines()[0] if body.strip() else "")[:280],
|
|
content_md=body,
|
|
category="synthesis",
|
|
source_type="manual",
|
|
source_id=None,
|
|
status=normalize_status(status),
|
|
tags=base_tags,
|
|
confidence=0.5,
|
|
related=extractor.related_links(self.project_name),
|
|
)
|
|
return await self._persist_draft(draft, created_by=created_by)
|
|
|
|
# -- CRUD -------------------------------------------------------------
|
|
async def list_notes(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
tag: str | None = None,
|
|
source_type: str | None = None,
|
|
status: str | None = None,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
items, total = await self.repo.list_notes(
|
|
keyword=keyword,
|
|
tag=tag,
|
|
source_type=normalize_source_type(source_type) if source_type else None,
|
|
status=status,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
return {"items": items, "total": total}
|
|
|
|
async def get_note(self, note_id: str) -> dict[str, Any] | None:
|
|
return await self.repo.get_note(note_id)
|
|
|
|
async def update_note(
|
|
self,
|
|
note_id: str,
|
|
*,
|
|
title: str | None = None,
|
|
content_md: str | None = None,
|
|
tags: list[str] | None = None,
|
|
status: str | None = None,
|
|
summary: str | None = None,
|
|
updated_by: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
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)
|
|
body = parsed or body
|
|
note = await self.repo.update_note(
|
|
note_id,
|
|
title=title,
|
|
summary=summary,
|
|
content_md=body,
|
|
status=normalize_status(status) if status else None,
|
|
tags=tags,
|
|
updated_by=updated_by,
|
|
)
|
|
if note is None:
|
|
return None
|
|
try:
|
|
rel = self.adapter.update_note_file(
|
|
note_id,
|
|
current.get("vault_path"),
|
|
title=note["title"],
|
|
content_md=note["content_md"],
|
|
tags=note.get("tags") or [],
|
|
status=note["status"],
|
|
summary=note.get("summary") or "",
|
|
category="synthesis",
|
|
created=note["created_at"],
|
|
updated=note["updated_at"],
|
|
)
|
|
if rel and rel != current.get("vault_path"):
|
|
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)
|
|
if current is None:
|
|
return False
|
|
note = await self.repo.set_status(note_id, "archived", updated_by=updated_by)
|
|
try:
|
|
self.adapter.archive_note(note_id, current.get("vault_path"), updated=(note or current)["updated_at"])
|
|
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
|
|
|
|
# -- wikilinks --------------------------------------------------------
|
|
async def resolve_wikilink(self, target: str) -> dict[str, Any]:
|
|
"""Resolve a single ``[[wikilink]]`` target to a navigable destination.
|
|
|
|
Returns one of:
|
|
- ``{"kind": "note", "note_id", "title", "exists": True}`` — a knowledge note
|
|
- ``{"kind": "vault", "vault_path", "title", "exists": True}`` — a vault meta/hub page
|
|
- ``{"kind": "missing", "exists": False}`` — nothing matches
|
|
|
|
Resolution order: note id → note vault_path → note title → vault file.
|
|
"""
|
|
norm = normalize_wikilink_target(target)
|
|
base = {"target": target, "normalized": norm}
|
|
if not norm:
|
|
return {**base, "kind": "missing", "exists": False}
|
|
|
|
# 1) Direct note id (e.g. "kb_abc...").
|
|
if norm.startswith("kb_"):
|
|
note = await self.repo.get_note(norm, with_sources=False)
|
|
if note and note.get("status") != "archived":
|
|
return {**base, "kind": "note", "note_id": note["id"], "title": note["title"], "exists": True}
|
|
|
|
# 2) Note by vault path (links from index.md / hot.md / structural hub).
|
|
note = await self.repo.find_by_vault_path(norm)
|
|
if note:
|
|
return {**base, "kind": "note", "note_id": note["id"], "title": note["title"], "exists": True}
|
|
|
|
# 3) Note by title (the real related-note cross-links use [[Title]]).
|
|
note = await self.repo.find_by_title(norm) or await self.repo.find_by_title(norm.split("/")[-1])
|
|
if note:
|
|
return {**base, "kind": "note", "note_id": note["id"], "title": note["title"], "exists": True}
|
|
|
|
# 4) A vault meta/hub page that has no DB note (index, _meta/taxonomy, hub…).
|
|
page = self.read_vault_page(norm)
|
|
if page is not None:
|
|
return {**base, "kind": "vault", "vault_path": page["path"], "title": page["title"], "exists": True}
|
|
|
|
return {**base, "kind": "missing", "exists": False}
|
|
|
|
async def resolve_wikilinks(self, targets: list[str]) -> dict[str, dict[str, Any]]:
|
|
"""Resolve many targets at once; returns ``{original_target: resolution}``."""
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for target in dict.fromkeys(t for t in targets if t and t.strip()):
|
|
out[target] = await self.resolve_wikilink(target)
|
|
return out
|
|
|
|
async def resolve_note_wikilinks(self, note_id: str) -> dict[str, dict[str, Any]] | None:
|
|
"""Extract every ``[[wikilink]]`` in a note's body and resolve them.
|
|
|
|
Returns ``{target: resolution}`` or ``None`` when the note is missing.
|
|
Numeric ``[[123]]`` citation badges are skipped (handled by the renderer).
|
|
"""
|
|
note = await self.repo.get_note(note_id, with_sources=False)
|
|
if note is None:
|
|
return None
|
|
targets = [t for t in extract_wikilinks(note.get("content_md") or "") if not t.isdigit()]
|
|
return await self.resolve_wikilinks(targets)
|
|
|
|
def read_vault_page(self, path: str) -> dict[str, Any] | None:
|
|
"""Read a raw vault Markdown page (meta/hub pages without a DB note).
|
|
|
|
Returns ``{"path", "title", "content_md"}`` or ``None`` if the page does
|
|
not exist or the path escapes the vault root.
|
|
"""
|
|
norm = normalize_wikilink_target(path)
|
|
if not norm:
|
|
return None
|
|
rel = f"{norm}.md"
|
|
try:
|
|
parsed = self.adapter.vault.read_note(rel)
|
|
except VaultPathError:
|
|
return None
|
|
if parsed is None:
|
|
return None
|
|
fm, body = parsed
|
|
title = str(fm.get("title") or "").strip() or norm.split("/")[-1]
|
|
return {"path": norm, "title": title, "content_md": body}
|
|
|
|
# -- 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, 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}
|
|
|
|
# -- 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()
|
|
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
|
|
|
|
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 []
|
|
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)
|