langgraph: 知识库同步治理——article 幂等重建、metadata 统一、initialize 去重、超时延长

- knowledge_sync.py: article 由增量追加改为全量幂等重建（先删 source=article 旧块再写），三源 metadata 键统一（id/source/post_id/title/category/tags/source_url/pub_date/chunk_index/chunk_total），显式传 Chroma doc id 保证幂等，单源失败隔离；增加分批重试与进度日志
- retriever.py: initialize() 不再写入粗粒度文章索引（消除 metadata.type=article 与 sync 的 metadata.source=article 双份重复数据）
- Dockerfile: gunicorn timeout 1800s → 7200s，防止长同步被杀

cfc-langgraph/Dockerfile

@@ -37,7 +37,7 @@ CMD ["python3", "-m", "gunicorn", "app.main:app", \
      "--worker-class", "uvicorn.workers.UvicornWorker", \
      "--bind", "0.0.0.0:9000", \
      "--workers", "1", \
-     "--timeout", "1800", \
+     "--timeout", "7200", \
      "--keep-alive", "10", \
      "--access-logfile", "-", \
      "--error-logfile", "-"]

cfc-langgraph/app/rag/retriever.py

@@ -22,46 +22,17 @@ class RagRetriever:
         self.java_client = JavaClient()
 
     async def initialize(self):
-        """从 Java 侧拉取知识库，更新到向量库"""
-        try:
-            articles = await self.java_client.get_published_articles()
-            if not articles:
-                logger.info("知识库初始化：无已发布文章")
-                return
-            # 防御式处理：容忍缺字段 / tags 为字符串或列表
-            texts = []
-            metadatas = []
-            for a in articles:
-                if not isinstance(a, dict):
-                    continue
-                title = a.get("title", "")
-                if not title:
-                    continue
-                tags = a.get("tags", []) or []
-                tags_str = tags if isinstance(tags, str) else " ".join(str(t) for t in tags)
-                texts.append(f"{title} {a.get('summary', '')} {tags_str}")
-                metadatas.append({
-                    "id": str(a["id"]),
-                    "title": title,
-                    "summary": a.get("summary", ""),
-                    "tags": tags_str,
-                    "type": "article",
-                })
-
-            if not texts:
-                logger.info("知识库初始化：无有效文章内容")
-                return
+        """初始化向量库连接（不写入任何内容）。
 
-            # upsert: 先删除旧的粗粒度文章索引(type=article), 再写入最新数据,
-            # 避免每次 worker 启动重复追加导致 embedding 无限累积
-            try:
-                self.vectorstore._collection.delete(where={"type": "article"})
-            except Exception as e:
-                logger.warning("知识库初始化: 删除旧文章索引失败: %s", e)
+        内容统一由 app.tasks.knowledge_sync.sync_knowledge_base 维护（article/microbiome/dan_knowledge 三源）。
 
-            # 添加最新文章索引
-            await self.vectorstore.aadd_texts(texts=texts, metadatas=metadatas)
-            logger.info("知识库向量化完成：%d 篇文章", len(metadatas))
+        旧实现在此额外写入一版粗粒度文章索引（metadata.type=article，标题+摘要+标签），
+        与同步任务写入的 metadata.source=article 分块索引（标题+摘要+正文）构成同一批文章的
+        双份向量，且每次容器重启都会追加，导致 HNSW 索引无限膨胀。治理后统一移除。
+        """
+        try:
+            count = self.vectorstore._collection.count()
+            logger.info("知识库初始化完成：collection 当前 %d 个文档（内容同步由 knowledge_sync 维护）", count)
         except Exception as e:
             logger.warning("知识库初始化失败：%s", e)
 
@@ -100,4 +71,4 @@ class RagRetriever:
             if len(results) >= k:
                 break
         
-        return results[:k]
+        return results[:k]

cfc-langgraph/app/tasks/knowledge_sync.py

@@ -1,17 +1,32 @@
-"""知识库同步定时任务: 定期从 Java 拉取文章, 更新 ChromaDB"""
-from app.rag.loader import KnowledgeLoader
-from app.rag.splitter import get_knowledge_splitter
-from app.rag.embeddings import get_embeddings
-from app.config import settings
-from langchain_chroma import Chroma
-from langchain_core.documents import Document
+"""知识库同步定时任务: 三源(article/microbiome/dan_knowledge)幂等同步到 ChromaDB
+
+治理改造（2026-10-19）：
+- article 由「只增量追加」改为「全量重建」：旧实现永不删除 article 旧块，
+  导致下架/删除文章残留在向量库、HNSW 索引无限膨胀（曾达 11.8GB）。
+- 三源 metadata 键统一（id/source/post_id/title/category/tags/source_url/pub_date/chunk_index/chunk_total），
+  消除旧实现 retriever.initialize 的 metadata.type=article 与同步任务 metadata.source=article 双份数据。
+- 显式传入 Chroma doc id（source_itemId_chunkIdx），写入幂等，可安全重试。
+- 每源独立 try/except，一处失败不影响其他源。
+"""
+import asyncio
+import datetime
+import json
 import logging
 import os
-import json
+
+from langchain_chroma import Chroma
+from langchain_core.documents import Document
+
+from app.config import settings
+from app.rag.embeddings import get_embeddings
+from app.rag.loader import KnowledgeLoader
+from app.rag.splitter import get_knowledge_splitter
 
 logger = logging.getLogger(__name__)
 
 SYNC_STATE_FILE = os.path.join(settings.chroma_db_path, ".sync_state")
+BATCH_SIZE = 100          # 每批写入的块数
+MAX_WRITE_RETRY = 3       # 单批写入失败重试次数（配合 embedding 429 退避）
 
 
 def _load_sync_state() -> dict:
@@ -50,35 +65,104 @@ def _delete_by_prefix(vectorstore, prefix: str):
         logger.warning("删除 source=%s 旧块失败: %s", prefix, e)
 
 
-def _build_chunks(splitter, items: list[dict], source: str) -> list[Document]:
-    chunks = []
+def _build_chunks(splitter, items: list[dict], source: str) -> tuple[list[Document], list[str]]:
+    """将条目分块，返回 (chunks, doc_ids)。
+
+    doc_id 形如 source_itemId_chunkIdx，作为 Chroma 文档 ID 显式传入，
+    保证同一知识条目的块可被幂等覆盖（而非每次生成随机 UUID 无限累积）。
+    兼容旧 metadata 字段映射（sourceUrl/pubDate）。
+    """
+    chunks: list[Document] = []
+    ids: list[str] = []
     for item in items:
         item_id = item.get("id")
         title = item.get("title", "")
         content = item.get("content", "")
         if item_id is None or not content:
             continue
-        full_text = f"{title}\n\n{content}"
+        # 若 content 已含 title 前缀则不重复拼接
+        full_text = f"{title}\n\n{content}" if title and title not in content else content
         split_texts = splitter.split_text(full_text)
         for i, text in enumerate(split_texts):
+            doc_id = f"{source}_{item_id}_{i}"
             metadata = {
-                "id": f"{source}_{item_id}_{i}",
+                "id": doc_id,
                 "source": source,
                 "post_id": str(item_id),
                 "title": title,
                 "category": item.get("category", ""),
                 "tags": item.get("tags", ""),
-                "source_url": item.get("sourceUrl", ""),
-                "pub_date": item.get("pubDate", ""),
+                "source_url": item.get("source_url", "") or item.get("sourceUrl", ""),
+                "pub_date": item.get("pub_date", "") or item.get("pubDate", ""),
                 "chunk_index": i,
                 "chunk_total": len(split_texts),
             }
             chunks.append(Document(page_content=text, metadata=metadata))
-    return chunks
+            ids.append(doc_id)
+    return chunks, ids
+
+
+async def _add_batch_with_retry(vectorstore, chunks: list, ids: list, label: str, start: int):
+    """单批写入，失败指数退避重试（embedding 网关限流时自动恢复）。"""
+    delay = 2.0
+    for attempt in range(MAX_WRITE_RETRY + 1):
+        try:
+            await vectorstore.aadd_documents(chunks, ids=ids)
+            return
+        except Exception as e:
+            if attempt == MAX_WRITE_RETRY:
+                raise
+            logger.warning("向量化写入失败[%s] %d-%d，%.0fs 后重试 (%d/%d): %s",
+                           label, start, start + len(chunks), delay, attempt + 1, MAX_WRITE_RETRY, e)
+            await asyncio.sleep(delay)
+            delay = min(delay * 2, 30.0)
+
+
+async def _write_chunks(vectorstore, chunks: list, ids: list, label: str):
+    """分批写入并输出进度日志。"""
+    total = len(chunks)
+    if total == 0:
+        logger.info("知识库同步[%s]: 无内容可写入", label)
+        return
+    for i in range(0, total, BATCH_SIZE):
+        batch_chunks = chunks[i:i + BATCH_SIZE]
+        batch_ids = ids[i:i + BATCH_SIZE]
+        await _add_batch_with_retry(vectorstore, batch_chunks, batch_ids, label, i)
+        done = min(i + BATCH_SIZE, total)
+        logger.info("向量化写入进度[%s]: %d/%d", label, done, total)
+
+
+async def _sync_source(vectorstore, splitter, source: str, fetch_func) -> int:
+    """全量重建一个 source：先删旧块，再全量拉取重建，返回写入块数。"""
+    try:
+        _delete_by_prefix(vectorstore, source)
+        items = await fetch_func()
+        chunks, ids = _build_chunks(splitter, items, source)
+        await _write_chunks(vectorstore, chunks, ids, source)
+        logger.info("知识库同步[%s]: %d 篇 -> %d 块", source, len(items), len(chunks))
+        return len(chunks)
+    except Exception as e:
+        logger.error("知识库同步[%s] 失败: %s", source, e)
+        return 0
+
+
+def _format_article_item(a: dict) -> dict:
+    """兼容 Java 文章结构，统一映射为题名/摘要/正文/标签/时间。"""
+    title = a.get("title", "")
+    return {
+        "id": a["id"],
+        "title": title,
+        "content": f"{a.get('summary', '')}\n\n{a.get('content', '')}".strip(),
+        "tags": a.get("tags", ""),
+        "pub_date": a.get("updatedAt", ""),
+    }
 
 
 async def sync_knowledge_base():
-    """同步三个数据源: articles 增量, microbiome/dan_knowledge 全量重建到 cfc_knowledge"""
+    """同步三个数据源: article/microbiome/dan_knowledge 全量重建到 cfc_knowledge。
+
+    每源独立失败隔离；全部成功后更新 .sync_state 记录 last_sync。
+    """
     loader = KnowledgeLoader()
     splitter = get_knowledge_splitter()
     embeddings = get_embeddings()
@@ -90,48 +174,22 @@ async def sync_knowledge_base():
     )
 
     state = _load_sync_state()
-    BATCH_SIZE = 100
-    stats = {}
-
-    articles = await loader.load_updated_since(state["last_sync"])
-    article_chunks = []
-    if articles:
-        docs = loader.format_for_indexing(articles)
-        for doc in docs:
-            if not doc.get("content"):
-                continue
-            split_texts = splitter.split_text(doc["content"])
-            for i, text in enumerate(split_texts):
-                metadata = dict(doc["metadata"])
-                metadata["chunk_index"] = i
-                article_chunks.append(Document(page_content=text, metadata=metadata))
-    stats["article"] = len(article_chunks)
-    logger.info("文章增量: %d 篇 -> %d 块", len(articles), len(article_chunks))
-
-    microbiome = await loader.java.get_microbiome_articles()
-    _delete_by_prefix(vectorstore, "microbiome")
-    microbiome_chunks = _build_chunks(splitter, microbiome, "microbiome")
-    stats["microbiome"] = len(microbiome_chunks)
-    logger.info("菌群知识库: %d 篇 -> %d 块", len(microbiome), len(microbiome_chunks))
-
-    dan_kb = await loader.java.get_dan_knowledge_base()
-    _delete_by_prefix(vectorstore, "dan_knowledge")
-    dan_chunks = _build_chunks(splitter, dan_kb, "dan_knowledge")
-    stats["dan_knowledge"] = len(dan_chunks)
-    logger.info("知识库(dan): %d 篇 -> %d 块", len(dan_kb), len(dan_chunks))
-
-    all_chunks = article_chunks + microbiome_chunks + dan_chunks
-    if not all_chunks:
-        logger.info("知识库同步: 无新增内容")
-        return
+    stats: dict = {}
 
-    for i in range(0, len(all_chunks), BATCH_SIZE):
-        batch = all_chunks[i:i + BATCH_SIZE]
-        await vectorstore.aadd_documents(batch)
-        logger.debug("知识库同步进度: %d/%d", i + len(batch), len(all_chunks))
+    async def fetch_articles():
+        articles = await loader.load_all_articles()
+        return [_format_article_item(a) for a in articles if a.get("id") is not None]
+
+    stats["article"] = await _sync_source(vectorstore, splitter, "article", fetch_articles)
+    stats["microbiome"] = await _sync_source(
+        vectorstore, splitter, "microbiome", loader.java.get_microbiome_articles)
+    stats["dan_knowledge"] = await _sync_source(
+        vectorstore, splitter, "dan_knowledge", loader.java.get_dan_knowledge_base)
+
+    if not any(stats.values()):
+        logger.warning("知识库同步: 全部数据源失败，不更新 last_sync")
+        return
 
-    import datetime
     state["last_sync"] = datetime.datetime.now().isoformat()
     _save_sync_state(state)
-
-    logger.info("知识库同步完成: 各源块数 %s", stats)
+    logger.info("知识库同步完成: 各源块数 %s", stats)