吉鑫升andClaude Opus 4.8
5ebbb82d66
- 知识库新增文件夹与模板支持（模型/仓储/服务/迁移），前端知识管理组件 - 岗位管理模型与接口完善 - 技能新增压缩设置页与相关存储/工具能力 - 系统设置扩展配置项 - 摘要中间件 keep 窗口改用 token 预算（16000），减少重复压缩 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
937 lines
43 KiB
Python
937 lines
43 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)
|