AdministratorandClaude Opus 4.8
81e4fd92ce
阶段三（治理）:
- sensitive.py：写入前脱敏 token/密钥/手机号/身份证/邮箱（sensitive_redaction_enabled）
- 版本快照：更新前写 knowledge_note_versions；GET /notes/{id}/versions
- dedup.py：标题/摘要 Jaccard + 内容哈希查重；GET /duplicates、POST /merge（来源并入主条目、重复项归档）
- auto_ingest.py：独立后台队列/线程沉淀，不阻塞问答；中间件 after_agent 在 auto_ingest_enabled 时入队；价值判断（最小字符）+ skip_existing 去重；自动沉淀落 draft 待审
阶段四（图谱）:
- LLM 抽取 entities/relations；service 存 knowledge_entities/knowledge_relations 并连边
- exporter.build_graph 支持实体节点 + note→entity/entity→entity 边；graph.html/json 增强
- GET /api/knowledge/graph 返回产品图谱数据
前端：知识库页「查重合并」对话框 + 相关 api/hooks
配置：sensitive_redaction_enabled / dedup_similarity_threshold / entity_extraction_enabled
测试：tests/test_knowledge.py 20 passed
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""Background auto-sediment queue (phase 3).
|
|
|
|
A single asyncio worker drains a bounded queue of finished conversations and
|
|
sediments the worthwhile ones into the knowledge base — entirely off the chat
|
|
request path, so it never adds latency to a reply. Enqueue happens from
|
|
``KnowledgeRagMiddleware.after_agent`` when ``knowledge.auto_ingest_enabled``.
|
|
|
|
Value judgment is intentionally cheap (min assistant chars + dedup via
|
|
``skip_existing``); the LLM still distills the kept conversations via the normal
|
|
``capture_thread`` path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_QUEUE: AutoIngestQueue | None = None
|
|
|
|
|
|
@dataclass
|
|
class _QueueUser:
|
|
"""Minimal CurrentUser stand-in (only ``.id`` is needed for context)."""
|
|
|
|
id: str
|
|
|
|
|
|
@dataclass
|
|
class AutoIngestItem:
|
|
thread_id: str
|
|
user_id: str
|
|
messages: list[dict[str, Any]]
|
|
thread_title: str | None = None
|
|
reference_batches: list[dict[str, Any]] = field(default_factory=list)
|
|
|
|
|
|
class AutoIngestQueue:
|
|
"""Bounded background queue + single worker for auto-sedimentation."""
|
|
|
|
def __init__(self, service_provider: Callable[[], Any], *, min_chars: int = 300, maxsize: int = 256) -> None:
|
|
self._provider = service_provider
|
|
self._min_chars = min_chars
|
|
self._queue: asyncio.Queue[AutoIngestItem] = asyncio.Queue(maxsize=maxsize)
|
|
self._task: asyncio.Task | None = None
|
|
|
|
def start(self) -> None:
|
|
if self._task is None or self._task.done():
|
|
self._task = asyncio.create_task(self._worker(), name="knowledge-auto-ingest")
|
|
logger.info("Knowledge auto-ingest worker started")
|
|
|
|
async def stop(self) -> None:
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._task
|
|
self._task = None
|
|
|
|
def enqueue(self, item: AutoIngestItem) -> None:
|
|
try:
|
|
self._queue.put_nowait(item)
|
|
except asyncio.QueueFull:
|
|
logger.warning("Knowledge auto-ingest queue full; dropping thread %s", item.thread_id)
|
|
|
|
def _worth_ingesting(self, item: AutoIngestItem) -> bool:
|
|
chars = 0
|
|
for m in item.messages:
|
|
if str(m.get("type") or m.get("role") or "").lower() == "ai":
|
|
content = m.get("content")
|
|
chars += len(content) if isinstance(content, str) else len(str(content or ""))
|
|
return chars >= self._min_chars
|
|
|
|
async def _process(self, item: AutoIngestItem) -> None:
|
|
service = self._provider()
|
|
if service is None:
|
|
return
|
|
if not self._worth_ingesting(item):
|
|
logger.debug("Auto-ingest skipped (below min chars) thread=%s", item.thread_id)
|
|
return
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
|
|
token = set_current_user(_QueueUser(id=item.user_id))
|
|
try:
|
|
result = await service.capture_thread(
|
|
thread_id=item.thread_id,
|
|
messages=item.messages,
|
|
thread_title=item.thread_title,
|
|
status="draft", # auto-sedimented notes land as drafts for review
|
|
include_sources=True,
|
|
reference_batches=item.reference_batches,
|
|
created_by=item.user_id,
|
|
skip_existing=True,
|
|
)
|
|
if result.get("created"):
|
|
logger.info("Auto-ingested thread %s -> note %s", item.thread_id, result["note"]["id"])
|
|
except Exception:
|
|
logger.exception("Auto-ingest failed for thread %s", item.thread_id)
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def _worker(self) -> None:
|
|
while True:
|
|
item = await self._queue.get()
|
|
try:
|
|
await self._process(item)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("Auto-ingest worker error")
|
|
finally:
|
|
self._queue.task_done()
|
|
|
|
|
|
def set_auto_ingest_queue(queue: AutoIngestQueue | None) -> None:
|
|
global _QUEUE
|
|
_QUEUE = queue
|
|
|
|
|
|
def get_auto_ingest_queue() -> AutoIngestQueue | None:
|
|
return _QUEUE
|