0.1.02思维导图全屏/lightrag修复/luckysheet修复

2026-01-06 05:21:53 +08:00
parent fb6832c30d
commit 9884ea0ea6
12 changed files with 570 additions and 199 deletions
@@ -1,10 +1,12 @@
 import os
 from functools import lru_cache
-from pydantic import BaseSettings, Field
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict


 class Settings(BaseSettings):
    """FastAPI 配置，统一读取 .env 或系统环境变量"""
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore")

    app_name: str = Field("AIMNOTE Ingest Service", env="INGEST_APP_NAME")
    environment: str = Field("development", env="INGEST_ENV")
@@ -12,16 +14,13 @@ class Settings(BaseSettings):
    supabase_url: str = Field("http://127.0.0.1:54321", env="SUPABASE_URL")
    supabase_key: str = Field("", env="SUPABASE_SERVICE_ROLE_KEY")
    lightrag_url: str = Field("http://127.0.0.1:7777", env="LIGHTRAG_URL")
+    lightrag_api_key: str = Field("", env="LIGHTRAG_API_KEY")
    ollama_base_url: str = Field("http://127.0.0.1:11434/v1", env="OLLAMA_BASE_URL")
    ollama_api_key: str = Field("ollama", env="OLLAMA_API_KEY")
    embeddings_model: str = Field("qwen3-embedding:8b", env="EMBEDDING_MODEL")
    llm_model: str = Field("qwen3:32b", env="DEFAULT_LLM_MODEL")
    rerank_model: str = Field("qwen3-reranker-4b", env="RERANK_MODEL")

-    class Config:
-        env_file = ".env"
-        case_sensitive = False
-

@lru_cache()
 def get_settings() -> Settings:
@@ -1,42 +1,13 @@
-import sys
-from pathlib import Path
-
 from app.core.config import get_settings
-from app.models.job import IngestJob
-from app.services.embeddings import EmbeddingGenerator
 from app.services.job_store import InMemoryJobStore
 from app.services.scheduler import SchedulerManager
 from app.services.tasks import IngestWorker
 from app.services.webhooks import LightRAGWebhook

-ROOT = Path(__file__).resolve().parents[3]
-sys.path.append(str(ROOT / "packages" / "siyuan_ingest" / "src"))
-from siyuan_ingest import SiYuanClient, SupabaseSync, SupabaseWriter  # type: ignore  # noqa: E402
-
 settings = get_settings()

 job_store = InMemoryJobStore()
 scheduler = SchedulerManager()
-siyuan_client = SiYuanClient()
 webhook = LightRAGWebhook()
-embedding_generator = EmbeddingGenerator(
-    api_base=settings.ollama_base_url,
-    api_key=settings.ollama_api_key,
-    model=settings.embeddings_model,
-)
-supabase_sync = SupabaseSync(
-    supabase_url=settings.supabase_url + "/rest/v1",
-    supabase_key=settings.supabase_key,
-)
-supabase_writer = SupabaseWriter(
-    supabase_url=settings.supabase_url + "/rest/v1",
-    supabase_key=settings.supabase_key,
-)
-worker = IngestWorker(
-    job_store=job_store,
-    siyuan_client=siyuan_client,
-    webhook=webhook,
-    supabase_sync=supabase_sync,
-    supabase_writer=supabase_writer,
-    embedding_generator=embedding_generator,
-)
+worker = IngestWorker(job_store=job_store, webhook=webhook)
+
@@ -1,49 +1,26 @@
 import asyncio
 import contextlib
 import logging
-import sys
-from pathlib import Path
-from typing import Dict, List, Optional
+from typing import List, Optional
 from uuid import uuid4

 from app.models.job import IngestJob
-from app.services.embeddings import EmbeddingGenerator
 from app.services.job_store import InMemoryJobStore
 from app.services.webhooks import LightRAGWebhook

-ROOT = Path(__file__).resolve().parents[3]
-sys.path.append(str(ROOT / "packages" / "siyuan_ingest" / "src"))
-from siyuan_ingest import (  # type: ignore  # noqa: E402
-    IngestState,
-    SiYuanClient,
-    SiYuanContentExtractor,
-    SupabaseSync,
-    SupabaseWriter,
-    chunk_blocks,
-)
-
 logger = logging.getLogger(__name__)


 class IngestWorker:
-    """处理 ingest 队列，串联 SiYuan → Supabase → LightRAG"""
+    """处理 ingest 队列：将外部文本推送到 LightRAG。

-    def __init__(
-        self,
-        job_store: InMemoryJobStore,
-        siyuan_client: SiYuanClient,
-        webhook: LightRAGWebhook,
-        supabase_sync: Optional[SupabaseSync] = None,
-        supabase_writer: Optional[SupabaseWriter] = None,
-        embedding_generator: Optional[EmbeddingGenerator] = None,
-    ) -> None:
+    说明：历史版本依赖 `siyuan_ingest` 从思源笔记拉取块数据，但该包当前不在仓库内，
+    回档后会导致服务无法启动。此处改为“由调用方直接提供 blocks 文本列表”。
+    """
+
+    def __init__(self, job_store: InMemoryJobStore, webhook: LightRAGWebhook) -> None:
        self.job_store = job_store
-        self.siyuan_client = siyuan_client
-        self.extractor = SiYuanContentExtractor(siyuan_client)
        self.webhook = webhook
-        self.supabase_sync = supabase_sync
-        self.supabase_writer = supabase_writer
-        self.embedding_generator = embedding_generator
        self.queue: asyncio.Queue[IngestJob] = asyncio.Queue()
        self._consumer_task: Optional[asyncio.Task[None]] = None

@@ -57,13 +34,19 @@ class IngestWorker:
            with contextlib.suppress(asyncio.CancelledError):
                await self._consumer_task

-    async def enqueue(self, notebook_id: Optional[str], doc_id: Optional[str], source: str, blocks: List[str]) -> IngestJob:
+    async def enqueue(
+        self,
+        notebook_id: Optional[str],
+        doc_id: Optional[str],
+        source: str,
+        blocks: List[str],
+    ) -> IngestJob:
        job = IngestJob(
            id=str(uuid4()),
            notebook_id=notebook_id,
            doc_id=doc_id,
            source=source,
-            blocks=blocks,
+            blocks=blocks or [],
        )
        await self.job_store.save(job)
        await self.queue.put(job)
@@ -80,12 +63,19 @@ class IngestWorker:
        try:
            job.mark_running()
            await self.job_store.save(job)
-            blocks = await self.extractor.list_blocks(job.notebook_id or "")
-            chunks = chunk_blocks(blocks)
-            logger.info("任务 %s 产生 %s 个 chunk", job.id, len(chunks))
-            await self._maybe_embed(chunks)
-            await self._persist_to_supabase(job, chunks)
-            await self.webhook.notify_ingest(job.doc_id, [c["block_id"] for c in chunks if c.get("block_id")])
+
+            texts = [b.strip() for b in job.blocks if b and b.strip()]
+            if not texts:
+                raise ValueError("blocks 为空：当前 ingest_service 仅支持由调用方提供文本列表")
+
+            # 约定：doc_id 存在时使用 doc:// 作为 file_source，便于后续统一引用
+            file_source = f"doc://{job.doc_id}" if job.doc_id else f"ingest://{job.id}"
+            merged_text = "\n\n".join(texts)
+            track_id = await self.webhook.ingest_text(file_source=file_source, text=merged_text)
+            job.payload["file_source"] = file_source
+            if track_id:
+                job.payload["lightrag_track_id"] = track_id
+
            job.mark_done()
        except Exception as exc:  # noqa: BLE001
            logger.exception("任务 %s 失败: %s", job.id, exc)
@@ -93,33 +83,3 @@ class IngestWorker:
        finally:
            await self.job_store.save(job)

-    async def _persist_to_supabase(self, job: IngestJob, chunks: List[Dict[str, str]]) -> None:
-        await self._maybe_sync_state(job, chunks)
-        if not self.supabase_writer:
-            logger.debug("未配置 SupabaseWriter，跳过落库")
-            return
-        await self.supabase_writer.ingest_chunks(job.notebook_id, chunks)
-
-    async def _maybe_sync_state(self, job: IngestJob, chunks: List[Dict[str, str]]) -> None:
-        if not self.supabase_sync:
-            return
-        if not chunks:
-            return
-        merged_checksum = chunks[-1]["checksum"]
-        state = await self.supabase_sync.fetch_state(job.notebook_id, job.doc_id)
-        if state and not state.should_ingest(merged_checksum):
-            logger.info("任务 %s 未发生变化，跳过写入 Supabase", job.id)
-            return
-        state = state or IngestState(notebook_id=job.notebook_id, doc_id=job.doc_id)
-        state.update(merged_checksum)
-        await self.supabase_sync.upsert_state(state)
-
-    async def _maybe_embed(self, chunks: List[Dict[str, str]]) -> None:
-        if not self.embedding_generator:
-            return
-        texts = [c.get("text") or "" for c in chunks]
-        if not texts:
-            return
-        embeddings = await self.embedding_generator.embed_texts(texts)
-        for chunk, embedding in zip(chunks, embeddings):
-            chunk["embedding"] = embedding
@@ -1,5 +1,5 @@
 import logging
-from typing import List, Optional
+from typing import Optional

 import httpx

@@ -9,18 +9,36 @@ logger = logging.getLogger(__name__)


 class LightRAGWebhook:
-    """封装与 LightRAG 的交互，默认调用 /ingest"""
+    """封装与 LightRAG 的交互。
+
+    说明：当前 LightRAG(>=1.4.x) 的“文本入库”接口为 `/documents/text` / `/documents/texts`，
+    旧版项目里使用的 `/ingest` 已不存在（回档后代码仍在调用旧接口）。
+    """

    def __init__(self, base_url: Optional[str] = None):
        settings = get_settings()
        self.base_url = base_url or settings.lightrag_url.rstrip("/")
+        self.api_key = settings.lightrag_api_key

-    async def notify_ingest(self, doc_id: Optional[str], block_ids: List[str]) -> None:
-        if not block_ids:
-            logger.info("无可用 block 推送给 LightRAG")
-            return
-        payload = {"doc_id": doc_id, "block_ids": block_ids}
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            resp = await client.post(f"{self.base_url}/ingest", json=payload)
+    def _headers(self) -> dict:
+        headers: dict = {}
+        if self.api_key:
+            headers["X-API-Key"] = self.api_key
+        return headers
+
+    async def ingest_text(self, file_source: str, text: str) -> Optional[str]:
+        """向 LightRAG 提交一段文本，触发后台索引。
+
+        返回 track_id（如果服务端返回），便于后续排查。
+        """
+        if not text.strip():
+            logger.info("空文本，跳过推送到 LightRAG（file_source=%s）", file_source)
+            return None
+        payload = {"text": text, "file_source": file_source}
+        async with httpx.AsyncClient(timeout=60.0, headers=self._headers()) as client:
+            resp = await client.post(f"{self.base_url}/documents/text", json=payload)
            resp.raise_for_status()
-            logger.info("LightRAG ingest 完成，返回 %s", resp.status_code)
+            data = resp.json()
+            track_id = data.get("track_id") if isinstance(data, dict) else None
+            logger.info("LightRAG 入库请求已提交（file_source=%s, track_id=%s）", file_source, track_id)
+            return track_id