937 lines
44 KiB
Python
937 lines
44 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 to cross-link under ## Related — by **shared tags**.
|
||
|
||
Matching is strictly tag-based: a candidate must share at least one of the
|
||
draft's *distinctive* tags (the per-note project / ``knowledge-base``
|
||
defaults are dropped, else everything matches everything). Free-text
|
||
title/summary matching was removed because CJK tokenization made generic
|
||
words (助手 / 知识图谱 / 页面) link unrelated notes. No distinctive tags →
|
||
no related notes (better empty than noisy). Best-effort.
|
||
"""
|
||
if not self.config.related_notes_enabled:
|
||
return []
|
||
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]
|
||
if not distinctive:
|
||
return []
|
||
own_title = (draft.title or "").strip().lower()
|
||
# Rank candidates by how many distinctive tags they share with the draft.
|
||
shared: dict[str, dict[str, Any]] = {}
|
||
for tag in distinctive:
|
||
try:
|
||
items, _total = await self.repo.list_notes(tag=tag, limit=50, offset=0)
|
||
except Exception:
|
||
logger.exception("Related-notes tag lookup failed for tag '%s'", tag)
|
||
continue
|
||
for it in items:
|
||
nid = it.get("id")
|
||
title = (it.get("title") or "").strip()
|
||
if not nid or not title or title.lower() == own_title:
|
||
continue
|
||
entry = shared.setdefault(nid, {"id": nid, "title": title, "count": 0})
|
||
entry["count"] += 1
|
||
ranked = sorted(shared.values(), key=lambda e: e["count"], reverse=True)
|
||
limit = max(1, self.config.related_notes_limit)
|
||
return [{"id": e["id"], "title": e["title"]} for e in ranked[:limit]]
|
||
|
||
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,
|
||
folder=draft.folder,
|
||
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,
|
||
model_name: str | None = None,
|
||
template_id: str | None = None,
|
||
folder: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Sediment a thread into a knowledge note.
|
||
|
||
``model_name`` optionally overrides the configured extraction model for
|
||
this one capture (falls back to ``config.extract_model_name``).
|
||
``template_id`` selects an operator-edited extraction prompt template
|
||
(falls back to the ``thread`` default, then the built-in prompt).
|
||
``folder`` files the resulting note(s) under a user directory.
|
||
|
||
Returns ``{"note": <note>, "created": bool}``. Raises
|
||
:class:`EmptyThreadError` when there is nothing worth keeping.
|
||
"""
|
||
extract_prompt = await self._resolve_extract_prompt(template_id, "thread")
|
||
# Always-on entry log: proves a capture was triggered for this thread,
|
||
# regardless of how it ends (success / skip / fallback). Lets you tell
|
||
# "capture never ran" apart from "logs not visible" via one grep.
|
||
logger.info(
|
||
"[knowledge] capturing thread %s (msgs=%d, llm_extract=%s, skip_existing=%s)",
|
||
thread_id,
|
||
len(messages or []),
|
||
self.config.llm_extract_enabled,
|
||
skip_existing,
|
||
)
|
||
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:
|
||
logger.info("[knowledge] thread %s already captured (note %s); skipping", thread_id, existing.get("id"))
|
||
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.
|
||
llm_attempted = self.config.llm_extract_enabled
|
||
llm_fallback = False
|
||
if llm_attempted:
|
||
try:
|
||
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=model_name or self.config.extract_model_name,
|
||
max_input_chars=self.config.extract_max_input_chars,
|
||
allow_multi=self.config.multi_note_enabled,
|
||
system_prompt=extract_prompt,
|
||
)
|
||
except llm_extractor.LLMRejectedIngest as exc:
|
||
# The model judged this conversation worthless (greeting / meta
|
||
# chatter). Skip ingestion entirely — do NOT dump the raw chat.
|
||
raise EmptyThreadError(f"Thread {thread_id} judged not worth ingesting: {exc}") from exc
|
||
if not drafts:
|
||
# The LLM path was supposed to run but produced nothing usable
|
||
# (API error / non-JSON). We will fall back to the rule-based
|
||
# extractor, which yields a low-quality verbatim dump with no
|
||
# entities/tags — flag it so it lands as a draft.
|
||
llm_fallback = True
|
||
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:
|
||
logger.info("[knowledge] thread %s has no sediment-able content; skipping", thread_id)
|
||
raise EmptyThreadError(f"Thread {thread_id} has no sediment-able content")
|
||
|
||
# When the LLM was enabled but silently fell back to rule-based extraction,
|
||
# the note is a verbatim dump (no distillation / entities / tags). Land it
|
||
# as a *draft* for review instead of silently approving it, and log loudly
|
||
# so the extraction failure is diagnosable (see also the WARNING/EXCEPTION
|
||
# logged inside llm_extractor.extract_thread_llm_multi).
|
||
effective_status = status
|
||
if llm_fallback:
|
||
logger.warning(
|
||
"[knowledge] thread %s — capture fell back to rule-based extraction "
|
||
"(LLM distillation produced nothing usable); landing as DRAFT. "
|
||
"Check extraction model '%s' / earlier [knowledge] logs for the cause.",
|
||
thread_id,
|
||
self.config.extract_model_name or "models[0]",
|
||
)
|
||
if normalize_status(status) == "approved":
|
||
effective_status = "draft"
|
||
|
||
normalized = normalize_status(effective_status)
|
||
for draft in drafts:
|
||
draft.status = normalized
|
||
if folder:
|
||
draft.folder = folder
|
||
# 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]
|
||
logger.info(
|
||
"[knowledge] thread %s captured → %d note(s) [%s], status=%s, llm_fallback=%s",
|
||
thread_id,
|
||
len(notes),
|
||
", ".join(n.get("title", "") for n in notes),
|
||
normalized,
|
||
llm_fallback,
|
||
)
|
||
return {"note": notes[0], "created": True, "notes": notes, "llm_fallback": llm_fallback}
|
||
|
||
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,
|
||
folder: 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),
|
||
folder=(folder or None),
|
||
tags=base_tags,
|
||
confidence=0.5,
|
||
related=extractor.related_links(self.project_name),
|
||
)
|
||
return await self._persist_draft(draft, created_by=created_by)
|
||
|
||
def _verbatim_file_draft(self, body: str, filename: str, *, title: str | None, tags: list[str] | None, doc_id: str) -> NoteDraft:
|
||
"""Build a verbatim fallback note from a file when LLM distillation fails.
|
||
|
||
Lands as a ``draft`` (like the thread fallback) since it is the raw
|
||
document, not distilled knowledge.
|
||
"""
|
||
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)
|
||
name = (title or Path(filename).stem or "文档知识").strip() or "文档知识"
|
||
summary = next((ln.strip() for ln in body.splitlines() if ln.strip()), name)[:280]
|
||
return NoteDraft(
|
||
title=name,
|
||
summary=summary,
|
||
content_md=body,
|
||
category="references",
|
||
source_type="file",
|
||
source_id=doc_id,
|
||
status="draft",
|
||
tags=base_tags,
|
||
confidence=0.4,
|
||
related=extractor.related_links(self.project_name),
|
||
source_key=f"file:{doc_id}",
|
||
)
|
||
|
||
async def capture_file(
|
||
self,
|
||
*,
|
||
content_md: str,
|
||
filename: str,
|
||
title: str | None = None,
|
||
tags: list[str] | None = None,
|
||
status: str = "approved",
|
||
created_by: str | None = None,
|
||
model_name: str | None = None,
|
||
skip_existing: bool = False,
|
||
template_id: str | None = None,
|
||
folder: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Import a file (already converted to Markdown/text) into knowledge notes.
|
||
|
||
Runs the same wiki-capture LLM distillation as thread capture, so an
|
||
uploaded document becomes one or more declarative wiki notes in the vault.
|
||
Falls back to a verbatim ``draft`` note when the LLM is disabled or returns
|
||
nothing usable. Returns ``{"note", "created", "notes", "llm_fallback"}``;
|
||
raises :class:`EmptyThreadError` when there is nothing to keep.
|
||
"""
|
||
import hashlib
|
||
|
||
text = (content_md or "").strip()
|
||
if not text:
|
||
raise EmptyThreadError(f"File {filename} is empty")
|
||
# Strip any frontmatter the document carries; the vault writer adds its own.
|
||
_, parsed = MarkdownVault.parse(text)
|
||
body = (parsed or text).strip()
|
||
if not body:
|
||
raise EmptyThreadError(f"File {filename} has no body content")
|
||
# Stable id from name + size so re-importing the same file can be skipped.
|
||
doc_id = hashlib.sha1(f"{filename}:{len(body)}".encode("utf-8")).hexdigest()[:16]
|
||
|
||
logger.info(
|
||
"[knowledge] importing file '%s' (chars=%d, doc_id=%s, llm_extract=%s, skip_existing=%s)",
|
||
filename,
|
||
len(body),
|
||
doc_id,
|
||
self.config.llm_extract_enabled,
|
||
skip_existing,
|
||
)
|
||
|
||
if skip_existing:
|
||
existing = await self.repo.find_by_source("file", doc_id)
|
||
if existing:
|
||
logger.info("[knowledge] file '%s' already imported (note %s); skipping", filename, existing.get("id"))
|
||
return {"note": existing, "created": False, "notes": [existing], "llm_fallback": False}
|
||
|
||
extract_prompt = await self._resolve_extract_prompt(template_id, "document")
|
||
drafts: list[NoteDraft] = []
|
||
llm_fallback = False
|
||
if self.config.llm_extract_enabled:
|
||
try:
|
||
drafts = await llm_extractor.extract_document_llm_multi(
|
||
body,
|
||
doc_id=doc_id,
|
||
doc_title=title or Path(filename).stem,
|
||
title=title,
|
||
tags=tags,
|
||
project_name=self.project_name,
|
||
model_name=model_name or self.config.extract_model_name,
|
||
max_input_chars=self.config.extract_max_input_chars,
|
||
allow_multi=self.config.multi_note_enabled,
|
||
system_prompt=extract_prompt,
|
||
)
|
||
except llm_extractor.LLMRejectedIngest as exc:
|
||
raise EmptyThreadError(f"File {filename} judged not worth ingesting: {exc}") from exc
|
||
if not drafts:
|
||
llm_fallback = True
|
||
if not drafts:
|
||
drafts = [self._verbatim_file_draft(body, filename, title=title, tags=tags, doc_id=doc_id)]
|
||
|
||
effective_status = status
|
||
if llm_fallback and normalize_status(status) == "approved":
|
||
effective_status = "draft"
|
||
normalized = normalize_status(effective_status)
|
||
for draft in drafts:
|
||
draft.status = normalized
|
||
if folder:
|
||
draft.folder = folder
|
||
|
||
# Cross-link sibling notes split from the same document (resolved by title
|
||
# once all are persisted), mirroring multi-note thread capture.
|
||
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]
|
||
logger.info(
|
||
"[knowledge] file '%s' imported → %d note(s) [%s], status=%s, llm_fallback=%s",
|
||
filename,
|
||
len(notes),
|
||
", ".join(n.get("title", "") for n in notes),
|
||
normalized,
|
||
llm_fallback,
|
||
)
|
||
return {"note": notes[0], "created": True, "notes": notes, "llm_fallback": llm_fallback}
|
||
|
||
# -- CRUD -------------------------------------------------------------
|
||
async def list_notes(
|
||
self,
|
||
*,
|
||
keyword: str | None = None,
|
||
tag: str | None = None,
|
||
entity: str | None = None,
|
||
source_type: str | None = None,
|
||
status: str | None = None,
|
||
folder: str | None = None,
|
||
limit: int = 20,
|
||
offset: int = 0,
|
||
) -> dict[str, Any]:
|
||
items, total = await self.repo.list_notes(
|
||
keyword=keyword,
|
||
tag=tag,
|
||
entity=entity,
|
||
source_type=normalize_source_type(source_type) if source_type else None,
|
||
status=status,
|
||
folder=folder,
|
||
limit=limit,
|
||
offset=offset,
|
||
)
|
||
return {"items": items, "total": total}
|
||
|
||
async def get_note(self, note_id: str) -> dict[str, Any] | None:
|
||
note = await self.repo.get_note(note_id)
|
||
if note is not None:
|
||
note["entities"] = await self.repo.list_note_entities(note_id)
|
||
return note
|
||
|
||
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,
|
||
folder: str | None = None,
|
||
set_folder: bool = False,
|
||
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,
|
||
folder=folder,
|
||
set_folder=set_folder,
|
||
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)
|
||
|
||
# -- folders ----------------------------------------------------------
|
||
async def list_folders(self) -> list[dict[str, Any]]:
|
||
return await self.repo.list_folders()
|
||
|
||
async def create_folder(self, path: str, *, created_by: str | None = None) -> dict[str, Any]:
|
||
return await self.repo.create_folder(path, created_by=created_by)
|
||
|
||
async def rename_folder(self, folder_id: str, new_path: str) -> dict[str, Any] | None:
|
||
return await self.repo.rename_folder(folder_id, new_path)
|
||
|
||
async def delete_folder(self, folder_id: str, *, reassign_to: str | None = None) -> bool:
|
||
return await self.repo.delete_folder(folder_id, reassign_to=reassign_to)
|
||
|
||
# -- extraction templates ---------------------------------------------
|
||
async def _resolve_extract_prompt(self, template_id: str | None, scope: str) -> str | None:
|
||
"""Return the system prompt for a capture, or ``None`` for the built-in default.
|
||
|
||
Resolution: explicit ``template_id`` → the scope's default template →
|
||
``None`` (the extractor then uses its hard-coded prompt). Best-effort: any
|
||
lookup failure degrades to the built-in prompt so capture never breaks.
|
||
"""
|
||
try:
|
||
if template_id:
|
||
tpl = await self.repo.get_template(template_id)
|
||
if tpl and tpl.get("enabled") and (tpl.get("system_prompt") or "").strip():
|
||
return tpl["system_prompt"]
|
||
default = await self.repo.get_default_template(scope)
|
||
if default and (default.get("system_prompt") or "").strip():
|
||
return default["system_prompt"]
|
||
except Exception:
|
||
logger.exception("Failed to resolve knowledge extraction template (id=%s, scope=%s)", template_id, scope)
|
||
return None
|
||
|
||
async def list_extract_templates(self, *, scope: str | None = None, enabled_only: bool = False) -> list[dict[str, Any]]:
|
||
return await self.repo.list_templates(scope=scope, enabled_only=enabled_only)
|
||
|
||
async def get_extract_template(self, template_id: str) -> dict[str, Any] | None:
|
||
return await self.repo.get_template(template_id)
|
||
|
||
async def create_extract_template(self, **kwargs: Any) -> dict[str, Any]:
|
||
return await self.repo.create_template(**kwargs)
|
||
|
||
async def update_extract_template(self, template_id: str, **kwargs: Any) -> dict[str, Any] | None:
|
||
return await self.repo.update_template(template_id, **kwargs)
|
||
|
||
async def delete_extract_template(self, template_id: str) -> bool:
|
||
return await self.repo.delete_template(template_id)
|
||
|
||
async def template_name_taken(self, name: str, *, exclude_id: str | None = None) -> bool:
|
||
return await self.repo.template_name_taken(name, exclude_id=exclude_id)
|
||
|
||
async def seed_default_templates(self, *, created_by: str | None = None) -> int:
|
||
"""Seed the two built-in distillation prompts as editable templates (idempotent).
|
||
|
||
Runs once on startup so an operator immediately sees — and can edit — the
|
||
prompts that drive 对话 / 文档 extraction. Existing templates with the same
|
||
name are left untouched. Returns the number of templates created.
|
||
"""
|
||
from deerflow.knowledge import llm_extractor
|
||
|
||
guidance = llm_extractor.WIKI_CAPTURE_GUIDANCE
|
||
seeds = [
|
||
("默认 · 对话提炼", "thread", llm_extractor._SYSTEM_PROMPT + "\n\n" + guidance, "对话沉淀的默认提炼提示词(wiki-capture 规范)。"),
|
||
("默认 · 文档提炼", "document", llm_extractor._DOC_SYSTEM_PROMPT + "\n\n" + guidance, "文档导入的默认提炼提示词(wiki-capture 规范)。"),
|
||
]
|
||
created = 0
|
||
for name, scope, prompt, desc in seeds:
|
||
try:
|
||
if await self.repo.template_name_taken(name):
|
||
continue
|
||
await self.repo.create_template(name=name, system_prompt=prompt, scope=scope, description=desc, is_default=True, enabled=True, created_by=created_by)
|
||
created += 1
|
||
except Exception:
|
||
logger.exception("Failed to seed knowledge extraction template '%s'", name)
|
||
if created:
|
||
logger.info("[knowledge] seeded %d default extraction template(s)", created)
|
||
return created
|
||
|
||
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)
|
||
# Drop this note's entity mentions + any now-orphaned entities so the graph
|
||
# follows the deletion (no lingering nodes from removed notes).
|
||
try:
|
||
await self.repo.clear_note_relations(note_id)
|
||
pruned = await self.repo.prune_orphan_entities()
|
||
if pruned:
|
||
logger.info("[knowledge] deleted note %s → pruned %d orphan entities", note_id, pruned)
|
||
except Exception:
|
||
logger.exception("Failed to prune relations/entities for deleted note %s", 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 []
|
||
blacklist = await self._graph_blacklist_set()
|
||
result = self.adapter.export_graph(notes, entities=entities, relations=relations, updated=now_iso(), blacklist=blacklist)
|
||
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 []
|
||
blacklist = await self._graph_blacklist_set()
|
||
return exporter.build_graph(notes, entities, relations, blacklist=blacklist)
|
||
|
||
# -- graph blacklist ---------------------------------------------------
|
||
async def _graph_blacklist_set(self) -> set[str]:
|
||
from deerflow.knowledge import exporter
|
||
|
||
items = await self.repo.list_graph_blacklist()
|
||
return exporter.normalize_blacklist([item["label"] for item in items])
|
||
|
||
async def list_graph_blacklist(self) -> list[dict[str, Any]]:
|
||
"""Return blacklist entries (node labels hidden from the graph)."""
|
||
return await self.repo.list_graph_blacklist()
|
||
|
||
async def add_graph_blacklist(self, labels: list[str], *, created_by: str | None = None) -> int:
|
||
"""Blacklist node labels (entity names / note titles). Returns the count added."""
|
||
return await self.repo.add_graph_blacklist(labels, created_by=created_by)
|
||
|
||
async def remove_graph_blacklist(self, entry_id: str) -> bool:
|
||
return await self.repo.remove_graph_blacklist(entry_id)
|
||
|
||
def export_file_path(self, filename: str) -> Path | None:
|
||
return self.adapter.export_file_path(filename)
|