|
|
@@ -30,8 +30,46 @@ def _save_sync_state(state: dict):
|
|
|
json.dump(state, f)
|
|
|
|
|
|
|
|
|
+def _delete_by_prefix(vectorstore, prefix: str):
|
|
|
+ try:
|
|
|
+ existing = vectorstore.get()
|
|
|
+ ids = [i for i in existing.get("ids", []) if i.startswith(prefix)]
|
|
|
+ if ids:
|
|
|
+ vectorstore.delete(ids)
|
|
|
+ logger.info("已删除前缀 %s 的 %d 个旧块", prefix, len(ids))
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning("删除前缀 %s 旧块失败: %s", prefix, e)
|
|
|
+
|
|
|
+
|
|
|
+def _build_chunks(splitter, items: list[dict], source: str) -> list[Document]:
|
|
|
+ chunks = []
|
|
|
+ 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}"
|
|
|
+ split_texts = splitter.split_text(full_text)
|
|
|
+ for i, text in enumerate(split_texts):
|
|
|
+ metadata = {
|
|
|
+ "id": f"{source}_{item_id}_{i}",
|
|
|
+ "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", ""),
|
|
|
+ "chunk_index": i,
|
|
|
+ "chunk_total": len(split_texts),
|
|
|
+ }
|
|
|
+ chunks.append(Document(page_content=text, metadata=metadata))
|
|
|
+ return chunks
|
|
|
+
|
|
|
+
|
|
|
async def sync_knowledge_base():
|
|
|
- """执行知识库同步 (全量+增量)"""
|
|
|
+ """同步三个数据源: articles 增量, microbiome/dan_knowledge 全量重建到 cfc_knowledge"""
|
|
|
loader = KnowledgeLoader()
|
|
|
splitter = get_knowledge_splitter()
|
|
|
embeddings = get_embeddings()
|
|
|
@@ -43,35 +81,49 @@ async def sync_knowledge_base():
|
|
|
)
|
|
|
|
|
|
state = _load_sync_state()
|
|
|
+ BATCH_SIZE = 100
|
|
|
+ stats = {}
|
|
|
|
|
|
articles = await loader.load_updated_since(state["last_sync"])
|
|
|
- if not articles:
|
|
|
- logger.info("知识库同步: 无更新内容")
|
|
|
+ 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
|
|
|
|
|
|
- docs = loader.format_for_indexing(articles)
|
|
|
-
|
|
|
- chunks = []
|
|
|
- for doc in docs:
|
|
|
- split_texts = splitter.split_text(doc["content"])
|
|
|
- for i, text in enumerate(split_texts):
|
|
|
- metadata = dict(doc["metadata"])
|
|
|
- metadata["chunk_index"] = i
|
|
|
- chunks.append(Document(page_content=text, metadata=metadata))
|
|
|
-
|
|
|
- if not chunks:
|
|
|
- logger.info("知识库同步: 无新增块")
|
|
|
- return
|
|
|
-
|
|
|
- BATCH_SIZE = 100
|
|
|
- for i in range(0, len(chunks), BATCH_SIZE):
|
|
|
- batch = chunks[i:i + BATCH_SIZE]
|
|
|
+ for i in range(0, len(all_chunks), BATCH_SIZE):
|
|
|
+ batch = all_chunks[i:i + BATCH_SIZE]
|
|
|
await vectorstore.aadd_documents(batch)
|
|
|
vectorstore.persist()
|
|
|
- logger.debug("知识库同步进度: %d/%d", i + len(batch), len(chunks))
|
|
|
+ logger.debug("知识库同步进度: %d/%d", i + len(batch), len(all_chunks))
|
|
|
|
|
|
import datetime
|
|
|
state["last_sync"] = datetime.datetime.now().isoformat()
|
|
|
_save_sync_state(state)
|
|
|
|
|
|
- logger.info("知识库同步完成: 新增 %d 篇文章, %d 个块", len(articles), len(chunks))
|
|
|
+ logger.info("知识库同步完成: 各源块数 %s", stats)
|