Files
deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/knowledge/llm_extractor.py
T
AdministratorandClaude Opus 4.8 d9fc7c9fdf chore(knowledge): 沉淀决策日志统一加 [knowledge] 前缀,跳过提到 WARNING 级
便于 docker logs 一条 grep 抓全部捕获决策(跳过/回退/失败):
  docker logs deerflow-vite-gateway 2>&1 | grep "\[knowledge\]"

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-08 22:55:44 +08:00

355 lines
14 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""LLM-based knowledge extraction (wiki-capture style).
Turns a conversation into a *declarative* knowledge note instead of dumping the
raw chat. Follows the obsidian-wiki ``wiki-capture`` skill: classify the note,
keep only high-value findings/decisions, drop greetings / intermediate
reasoning / verbatim transcripts, mark inferred content, and emit structured
JSON that we render into the Context / Finding / Reasoning / Implications /
Related body.
Falls back to the rule-based :func:`deerflow.knowledge.extractor.extract_thread`
when the LLM is unavailable or returns something unusable, so capture never
hard-fails.
"""
from __future__ import annotations
import json
import logging
import re
from typing import Any
from deerflow.knowledge import extractor
from deerflow.knowledge.schemas import NoteDraft, SourceDraft
from deerflow.knowledge.skill_templates import WIKI_CAPTURE_GUIDANCE
logger = logging.getLogger(__name__)
_VALID_CATEGORIES = {"synthesis", "concepts", "references", "decision", "journal"}
class LLMRejectedIngest(Exception):
"""The LLM judged the conversation as having no knowledge value.
Raised (instead of returning ``[]``) when the model explicitly sets
``should_ingest: false`` with no notes — so callers can *skip* ingestion
entirely rather than falling back to the rule-based verbatim dump.
"""
def _is_falsey(value: Any) -> bool:
if value is False:
return True
if isinstance(value, str):
return value.strip().lower() in ("false", "0", "no")
return False
_SYSTEM_PROMPT = """\
你是知识库整理专家,遵循 obsidian-wiki 的 wiki-capture 规范,把一次对话提炼成\
**陈述式知识(declarative knowledge)**,而不是聊天流水账。
要求:
- 只保留高价值内容:明确结论、方案、决策、关键事实、被引用的来源。
- 丢弃:寒暄、客套、中间推理过程、重复内容、无意义的原文堆砌。
- 不要写“用户问了什么、AI 回答了什么”,要写“结论是什么、为什么成立”。
- 推断内容用 ^[inferred] 标记,不确定/冲突用 ^[ambiguous] 标记。
- 分类 category 取值之一:synthesis(分析/方案/结论)、concepts(概念/框架)、\
references(外部来源汇总)、decision(架构/设计决策)、journal(会话纪要)。
- 用中文输出。
一次对话可能只讲一个主题,也可能覆盖多个相互独立的主题。请把它拆成 1~4 页知识:
- 默认只产出 **1 页**;只有当对话明确包含**多个彼此独立、可各自成文**的主题时才拆成多页。
- 每一页聚焦单一主题,标题与 summary 必须彼此区分,不要把同一结论拆到多页。
- 不值得沉淀时返回空的 notes 数组。
只输出一个 JSON 对象,不要任何额外文字或 Markdown 代码块,结构如下:
{
"should_ingest": true,
"reason": "为什么值得/不值得沉淀",
"notes": [
{
"title": "简洁的知识标题(不超过 40 字)",
"summary": "1-2 句话概括这页知识",
"category": "synthesis",
"context": "问题背景或触发场景,1-3 句",
"findings": ["结论/要点(陈述式,可含 ^[inferred])", "..."],
"reasoning": "为什么成立、取舍、不确定性",
"implications": ["后续建议/风险/下一步", "..."],
"tags": ["标签1", "标签2"],
"entities": [{"name": "实体名", "type": "technology|concept|person|org|product|other"}],
"relations": [{"from": "实体A", "to": "实体B", "type": "uses|relates_to|part_of|depends_on"}],
"confidence": 0.0
}
]
}"""
def _build_transcript(turns: list[dict[str, str]], sources: list[SourceDraft], *, max_chars: int) -> str:
parts: list[str] = []
for i, t in enumerate(turns, 1):
if t.get("user"):
parts.append(f"[用户 {i}] {t['user']}")
if t.get("ai"):
parts.append(f"[助手 {i}] {t['ai']}")
if sources:
parts.append("\n[来源]")
for s in sources[:20]:
label = s.title or s.url or s.tool_name or "来源"
parts.append(f"- {label}" + (f" ({s.url})" if s.url else ""))
text = "\n\n".join(parts)
if len(text) > max_chars:
text = text[:max_chars] + "\n…(已截断)"
return text
def _parse_json(raw: str) -> dict[str, Any] | None:
if not raw:
return None
text = raw.strip()
# Strip ```json fences if present.
fence = re.search(r"```(?:json)?\s*(.*?)```", text, re.DOTALL)
if fence:
text = fence.group(1).strip()
# Otherwise grab the outermost { ... }.
if not text.startswith("{"):
start = text.find("{")
end = text.rfind("}")
if start >= 0 and end > start:
text = text[start : end + 1]
try:
data = json.loads(text)
return data if isinstance(data, dict) else None
except json.JSONDecodeError:
logger.warning("Knowledge LLM returned non-JSON output; falling back to rule-based")
return None
def _render_body(data: dict[str, Any], title: str, related: list[str], sources: list[SourceDraft]) -> str:
def _bullets(items: Any) -> str:
if isinstance(items, str):
items = [items]
if not isinstance(items, list):
return ""
return "\n".join(f"- {str(x).strip()}" for x in items if str(x).strip())
context = str(data.get("context") or "").strip() or "- 来自一次对话沉淀。^[inferred]"
findings = _bullets(data.get("findings")) or "_暂无可提炼的结论。_"
reasoning = str(data.get("reasoning") or "").strip() or "(未提供推理说明)"
implications = _bullets(data.get("implications")) or "- 如内容过期或冲突,请编辑或归档。"
related_lines = [f"- {link}" for link in related]
source_lines = [f"- {s.title or s.url or '来源'}" + (f" — {s.url}" if s.url else "") for s in sources[:20]]
related_section = extractor._related_section_md(related_lines, source_lines)
return (
f"# {title}\n\n"
f"## Context\n\n{context}\n\n"
f"## Finding / Decision\n\n{findings}\n\n"
f"## Reasoning\n\n{reasoning}\n\n"
f"## Implications\n\n{implications}\n\n"
f"{related_section}"
)
def _notes_from_data(data: dict[str, Any]) -> list[dict[str, Any]]:
"""Normalize the LLM JSON into a list of per-note dicts.
Accepts the multi-note shape ``{"notes": [...]}`` as well as the legacy
single-note shape (``{"title": ..., "findings": ...}``) for backward
compatibility. Notes without any title/findings are dropped.
"""
raw_notes = data.get("notes")
if isinstance(raw_notes, list):
candidates = [n for n in raw_notes if isinstance(n, dict)]
elif data.get("title") or data.get("findings") or data.get("context"):
candidates = [data] # legacy single-object output
else:
candidates = []
return [n for n in candidates if str(n.get("title") or "").strip() or n.get("findings")]
def _build_note_draft(
note_data: dict[str, Any],
*,
thread_id: str,
thread_title: str | None,
turns: list[dict[str, str]],
title_override: str | None,
tags: list[str] | None,
project_name: str | None,
sources: list[SourceDraft],
source_key_suffix: str = "",
) -> NoteDraft:
"""Render one note dict into a :class:`NoteDraft`."""
category = str(note_data.get("category") or "synthesis").strip().lower()
if category not in _VALID_CATEGORIES:
category = "synthesis"
final_title = (title_override or "").strip() or str(note_data.get("title") or "").strip() or extractor._auto_title(thread_title, turns)
summary = str(note_data.get("summary") or "").strip()
related = extractor.related_links(project_name)
body = _render_body(note_data, final_title, related, sources)
base_tags = list(tags or [])
llm_tags = note_data.get("tags") if isinstance(note_data.get("tags"), list) else []
for tag in [*llm_tags, project_name or "zncm", "knowledge-base"]:
tag = str(tag).strip()
if tag and tag not in base_tags:
base_tags.append(tag)
try:
confidence = float(note_data.get("confidence") or 0.7)
except (TypeError, ValueError):
confidence = 0.7
entities = note_data.get("entities") if isinstance(note_data.get("entities"), list) else []
relations = note_data.get("relations") if isinstance(note_data.get("relations"), list) else []
return NoteDraft(
title=final_title,
summary=summary or final_title,
content_md=body,
category=category,
source_type="thread",
source_id=thread_id,
tags=base_tags,
confidence=max(0.0, min(confidence, 1.0)),
related=related,
sources=sources,
entities=[e for e in entities if isinstance(e, dict) and e.get("name")],
relations=[r for r in relations if isinstance(r, dict) and r.get("from") and r.get("to")],
source_key=f"thread:{thread_id}{source_key_suffix}",
)
async def extract_thread_llm_multi(
messages: list[dict[str, Any]],
*,
thread_id: str,
thread_title: str | None = None,
mode: str = "summary",
title: str | None = None,
tags: list[str] | None = None,
project_name: str | None = "zncm",
sources: list[SourceDraft] | None = None,
max_messages: int = 80,
model_name: str | None = None,
max_input_chars: int = 12000,
allow_multi: bool = True,
max_notes: int = 4,
) -> list[NoteDraft]:
"""Distill a thread into one *or several* cross-linkable knowledge notes.
The LLM may split a conversation that covers multiple independent topics into
several declarative notes. Returns ``[]`` when there is nothing to sediment
or the LLM path fails — callers should fall back to the rule-based extractor.
When ``allow_multi`` is false only the first note is kept.
"""
turns = extractor.split_turns(messages, max_messages=max_messages)
if not turns:
return []
sources = sources or []
try:
from langchain_core.messages import HumanMessage, SystemMessage
from deerflow.models import create_chat_model
model = create_chat_model(name=model_name or None, thinking_enabled=False)
transcript = _build_transcript(turns, sources, max_chars=max_input_chars)
prompt = [
SystemMessage(content=_SYSTEM_PROMPT + "\n\n" + WIKI_CAPTURE_GUIDANCE),
HumanMessage(content=f"以下是一次对话,请按规范提炼为知识 JSON:\n\n{transcript}"),
]
response = await model.ainvoke(prompt)
raw = response.content if isinstance(response.content, str) else str(response.content)
data = _parse_json(raw)
except Exception:
logger.exception("[knowledge] thread %s — LLM extraction call failed; using rule-based fallback", thread_id)
return []
if not data:
logger.warning("[knowledge] thread %s — LLM returned no parsable JSON; using rule-based fallback", thread_id)
return []
note_dicts = _notes_from_data(data)
if not note_dicts:
reason = str(data.get("reason") or "")[:200]
if _is_falsey(data.get("should_ingest")):
# The model deliberately judged this conversation worthless (greeting,
# meta chatter, no findings). Skip ingestion entirely — do NOT fall
# back to the rule-based verbatim dump.
logger.warning("[knowledge] thread %s SKIPPED — model judged not worth ingesting: %s", thread_id, reason)
raise LLMRejectedIngest(reason or "no knowledge value")
logger.warning(
"[knowledge] thread %s — LLM produced no notes (should_ingest=%s, reason=%r); using rule-based fallback",
thread_id,
data.get("should_ingest"),
reason,
)
return []
if not allow_multi:
note_dicts = note_dicts[:1]
else:
note_dicts = note_dicts[: max(1, max_notes)]
drafts: list[NoteDraft] = []
for i, nd in enumerate(note_dicts):
# The user-supplied title override only applies to a single-note result;
# multi-note splits keep each note's own LLM title.
override = title if (len(note_dicts) == 1) else None
suffix = "" if len(note_dicts) == 1 else f"#{i}"
drafts.append(
_build_note_draft(
nd,
thread_id=thread_id,
thread_title=thread_title,
turns=turns,
title_override=override,
tags=tags,
project_name=project_name,
sources=sources,
source_key_suffix=suffix,
)
)
return drafts
async def extract_thread_llm(
messages: list[dict[str, Any]],
*,
thread_id: str,
thread_title: str | None = None,
mode: str = "summary",
title: str | None = None,
tags: list[str] | None = None,
project_name: str | None = "zncm",
sources: list[SourceDraft] | None = None,
max_messages: int = 80,
model_name: str | None = None,
max_input_chars: int = 12000,
) -> NoteDraft | None:
"""Distill a thread into a single :class:`NoteDraft` (backward-compatible).
Thin wrapper over :func:`extract_thread_llm_multi` that returns the first
note, or ``None`` when nothing was produced.
"""
try:
drafts = await extract_thread_llm_multi(
messages,
thread_id=thread_id,
thread_title=thread_title,
mode=mode,
title=title,
tags=tags,
project_name=project_name,
sources=sources,
max_messages=max_messages,
model_name=model_name,
max_input_chars=max_input_chars,
allow_multi=False,
)
except LLMRejectedIngest:
return None
return drafts[0] if drafts else None