fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: 改进知识库的初始化错误处理 (#7243)

Browse Source 
* fix: 改进 KnowledgeBaseManager 和 KBHelper 中的初始化错误处理

* fix: 改进知识库初始化和重排序错误处理，增强日志记录

* fix: 改进知识库模块初始化和检索错误处理

* fix(ui): handle kb init errors in list cards

display a dedicated error state for knowledge base cards that fail
initialization, including a visible badge and error details

prevent navigation and edit actions for failed cards while keeping
delete available, and hide normal stats/description for error items

add list.initError locale strings for en-US, ru-RU, and zh-CN

* fix(kb): avoid replacing helper on init failure

Initialize a new KB helper before swapping instances so a failed re-init
does not break the active knowledge base service.

If initialization fails, restore in-memory KB settings and keep the
existing helper and previous init error state.

Also clear stale init_error after successful vector DB initialization to
prevent outdated error reporting.

* test(kb): add kb manager resilience tests

cover initialization failure and recovery scenarios to guard
against regressions in kb error handling

include reference assets under refs for test validation

---------

Co-authored-by: RC-CHN <1051989940@qq.com>
This commit is contained in:
氕氙
2026-04-02 16:16:22 +08:00
committed by GitHub
parent e8d6938d31
commit 9d4472cb2d
9 changed files with 498 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
astrbot/core/knowledge_base/kb_helper.py
View File

@@ -108,6 +108,7 @@ Text chunk to process:
class KBHelper:
vec_db: BaseVecDB
kb: KnowledgeBase
init_error: str | None
def __init__(
self,
@@ -122,6 +123,7 @@ class KBHelper:
self.prov_mgr = provider_manager
self.kb_root_dir = kb_root_dir
self.chunker = chunker
self.init_error = None
self.kb_dir = Path(self.kb_root_dir) / self.kb.kb_id
self.kb_medias_dir = Path(self.kb_dir) / "medias" / self.kb.kb_id
@@ -148,13 +150,14 @@ class KBHelper:
async def get_rp(self) -> RerankProvider | None:
if not self.kb.rerank_provider_id:
return None
rp: RerankProvider = await self.prov_mgr.get_provider_by_id(
rp: RerankProvider | None = await self.prov_mgr.get_provider_by_id(
self.kb.rerank_provider_id,
) # type: ignore
if not rp:
raise ValueError(
f"无法找到 ID 为 {self.kb.rerank_provider_id} 的 Rerank Provider",
logger.warning(
f"知识库 {self.kb.kb_name}({self.kb.kb_id}) 的 Rerank Provider({self.kb.rerank_provider_id}) 不可用，将跳过重排序。",
)
return None
return rp
async def _ensure_vec_db(self) -> FaissVecDB:
@@ -162,7 +165,13 @@ class KBHelper:
raise ValueError(f"知识库 {self.kb.kb_name} 未配置 Embedding Provider")
ep = await self.get_ep()
rp = await self.get_rp()
rp: RerankProvider | None = None
try:
rp = await self.get_rp()
except Exception as e:
logger.warning(
f"知识库 {self.kb.kb_name}({self.kb.kb_id}) 初始化重排序能力失败，将跳过重排序: {e}",
)
vec_db = FaissVecDB(
doc_store_path=str(self.kb_dir / "doc.db"),
@@ -172,6 +181,8 @@ class KBHelper:
)
await vec_db.initialize()
self.vec_db = vec_db
# Clear stale init_error once initialization succeeds.
self.init_error = None
return vec_db
async def delete_vec_db(self) -> None:
@@ -183,7 +194,7 @@ class KBHelper:
shutil.rmtree(self.kb_dir)
async def terminate(self) -> None:
if self.vec_db:
if hasattr(self, "vec_db") and self.vec_db:
await self.vec_db.close()
async def upload_document(

74
astrbot/core/knowledge_base/kb_mgr.py
View File

@@ -1,4 +1,3 @@
import traceback
from pathlib import Path
from astrbot.core import logger
@@ -56,8 +55,7 @@ class KnowledgeBaseManager:
logger.error(f"知识库模块导入失败: {e}")
logger.warning("请确保已安装所需依赖: pypdf, aiofiles, Pillow, rank-bm25")
except Exception as e:
logger.error(f"知识库模块初始化失败: {e}")
logger.error(traceback.format_exc())
logger.error(f"知识库模块初始化失败: {e}", exc_info=True)
async def _init_kb_database(self) -> None:
self.kb_db = KBSQLiteDatabase(DB_PATH.as_posix())
@@ -76,7 +74,14 @@ class KnowledgeBaseManager:
kb_root_dir=FILES_PATH,
chunker=CHUNKER,
)
await kb_helper.initialize()
try:
await kb_helper.initialize()
except Exception as e:
kb_helper.init_error = str(e)
logger.error(
f"知识库 {record.kb_name}({record.kb_id}) 初始化失败: {e}",
exc_info=True,
)
self.kb_insts[record.kb_id] = kb_helper
async def create_kb(
@@ -179,6 +184,20 @@ class KnowledgeBaseManager:
return None
kb = kb_helper.kb
previous_state = {
"kb_name": kb.kb_name,
"description": kb.description,
"emoji": kb.emoji,
"embedding_provider_id": kb.embedding_provider_id,
"rerank_provider_id": kb.rerank_provider_id,
"chunk_size": kb.chunk_size,
"chunk_overlap": kb.chunk_overlap,
"top_k_dense": kb.top_k_dense,
"top_k_sparse": kb.top_k_sparse,
"top_m_final": kb.top_m_final,
}
previous_init_error = kb_helper.init_error
if kb_name is not None:
kb.kb_name = kb_name
if description is not None:
@@ -198,12 +217,47 @@ class KnowledgeBaseManager:
kb.top_k_sparse = top_k_sparse
if top_m_final is not None:
kb.top_m_final = top_m_final
# Build a new helper first. Keep current vec_db alive until new init succeeds.
new_helper = KBHelper(
kb_db=self.kb_db,
kb=kb,
provider_manager=self.provider_manager,
kb_root_dir=FILES_PATH,
chunker=CHUNKER,
)
try:
await new_helper.initialize()
except Exception as e:
# Roll back in-memory settings and keep current helper available.
kb.kb_name = previous_state["kb_name"]
kb.description = previous_state["description"]
kb.emoji = previous_state["emoji"]
kb.embedding_provider_id = previous_state["embedding_provider_id"]
kb.rerank_provider_id = previous_state["rerank_provider_id"]
kb.chunk_size = previous_state["chunk_size"]
kb.chunk_overlap = previous_state["chunk_overlap"]
kb.top_k_dense = previous_state["top_k_dense"]
kb.top_k_sparse = previous_state["top_k_sparse"]
kb.top_m_final = previous_state["top_m_final"]
kb_helper.init_error = previous_init_error
logger.error(
f"知识库 {kb.kb_name}({kb.kb_id}) 重新初始化失败，继续使用旧实例: {e}",
exc_info=True,
)
return kb_helper
async with self.kb_db.get_db() as session:
session.add(kb)
await session.commit()
await session.refresh(kb)
return kb_helper
old_helper = kb_helper
self.kb_insts[kb_id] = new_helper
await old_helper.terminate()
new_helper.init_error = None
return new_helper
async def retrieve(
self,
@@ -215,11 +269,21 @@ class KnowledgeBaseManager:
"""从指定知识库中检索相关内容"""
kb_ids = []
kb_id_helper_map = {}
unavailable_kbs = []
for kb_name in kb_names:
if kb_helper := await self.get_kb_by_name(kb_name):
if kb_helper.init_error:
unavailable_kbs.append((kb_name, kb_helper.init_error))
logger.warning(f"知识库 {kb_name} 不可用: {kb_helper.init_error}")
continue
kb_ids.append(kb_helper.kb.kb_id)
kb_id_helper_map[kb_helper.kb.kb_id] = kb_helper
# all requested KBs are unavailable
if not kb_ids and unavailable_kbs:
errors = "; ".join(f"{n}: {e}" for n, e in unavailable_kbs)
raise ValueError(f"所有请求的知识库均不可用: {errors}")
if not kb_ids:
return {}

23
astrbot/core/knowledge_base/retrieval/manager.py
View File

@@ -184,12 +184,15 @@ class RetrievalManager:
first_rerank = vec_db.rerank_provider
break
if first_rerank and retrieval_results:
retrieval_results = await self._rerank(
query=query,
results=retrieval_results,
top_k=top_m_final,
rerank_provider=first_rerank,
)
try:
retrieval_results = await self._rerank(
query=query,
results=retrieval_results,
top_k=top_m_final,
rerank_provider=first_rerank,
)
except Exception as e:
logger.warning(f"Rerank 执行失败，已跳过重排序并使用融合结果: {e}")
return retrieval_results[:top_m_final]
@@ -229,10 +232,10 @@ class RetrievalManager:
all_results.extend(vec_results)
except Exception as e:
from astrbot.core import logger
logger.warning(f"知识库 {kb_id} 稠密检索失败: {e}")
continue
logger.error(f"知识库 {kb_id} 稠密检索失败: {e}", exc_info=True)
if len(kb_ids) == 1:
raise RuntimeError(f"知识库 {kb_id} 稠密检索失败: {e}") from e
# multi-KB: skip the faulty KB and continue
# 按相似度排序并返回 top_k
all_results.sort(key=lambda x: x.similarity, reverse=True)

7
astrbot/dashboard/routes/knowledge_base.py
View File

@@ -314,7 +314,12 @@ class KnowledgeBaseRoute(Route):
# 转换为字典列表
kb_list = []
for kb in kbs:
kb_list.append(kb.model_dump())
kb_dict = kb.model_dump()
# include init_error from KBHelper if present
kb_helper = await kb_manager.get_kb(kb.kb_id)
if kb_helper and kb_helper.init_error:
kb_dict["init_error"] = kb_helper.init_error
kb_list.append(kb_dict)
return (
Response()

3
dashboard/src/i18n/locales/en-US/features/knowledge-base/index.json
View File

@@ -10,7 +10,8 @@
"loading": "Loading...",
"documents": "Documents",
"chunks": "Chunks",
"sessionConfig": "Session Config"
"sessionConfig": "Session Config",
"initError": "Initialization Failed"
},
"card": {
"edit": "Edit",

3
dashboard/src/i18n/locales/ru-RU/features/knowledge-base/index.json
View File

@@ -10,7 +10,8 @@
"loading": "Загрузка...",
"documents": "док.",
"chunks": "фрагм.",
"sessionConfig": "Профиль"
"sessionConfig": "Профиль",
"initError": "Ошибка инициализации"
},
"card": {
"edit": "Изменить",

3
dashboard/src/i18n/locales/zh-CN/features/knowledge-base/index.json
View File

@@ -10,7 +10,8 @@
"loading": "正在加载...",
"documents": "文档",
"chunks": "分块",
"sessionConfig": "会话配置"
"sessionConfig": "会话配置",
"initError": "初始化失败"
},
"card": {
"edit": "编辑",

90
dashboard/src/views/knowledge-base/KBList.vue
View File

@@ -27,14 +27,27 @@
</div>
<div v-else-if="kbList.length > 0" class="kb-grid">
<v-card v-for="kb in kbList" :key="kb.kb_id" class="kb-card" elevation="2" hover
@click="navigateToDetail(kb.kb_id)">
<div class="kb-card-content">
<v-card v-for="kb in kbList" :key="kb.kb_id" class="kb-card" elevation="2" :hover="!kb.init_error"
:class="{ 'kb-card-error': kb.init_error }"
@click="!kb.init_error && navigateToDetail(kb.kb_id)">
<!-- Error badge -->
<v-badge v-if="kb.init_error" color="error" icon="mdi-alert-circle"
class="kb-error-badge position-absolute" style="top: 0; right: 0; transform: translate(34%, -34%);" />
<div class="kb-card-content" :class="{ 'kb-card-content-error': kb.init_error }">
<div class="kb-emoji">{{ kb.emoji || '📚' }}</div>
<h3 class="kb-name">{{ kb.kb_name }}</h3>
<p class="kb-description text-medium-emphasis">{{ kb.description || '暂无描述' }}</p>
<p v-if="!kb.init_error" class="kb-description text-medium-emphasis">{{ kb.description || '暂无描述' }}</p>
<div class="kb-stats mt-4">
<!-- Error message display -->
<div v-if="kb.init_error" class="kb-error-panel mt-3 mb-2">
<div class="kb-error-title">
<v-icon size="16" color="error">mdi-close-circle</v-icon>
<span>{{ t('list.initError') }}</span>
</div>
<div class="kb-error-detail" :title="kb.init_error">{{ kb.init_error }}</div>
</div>
<div class="kb-stats mt-4" v-if="!kb.init_error">
<div class="stat-item">
<v-icon size="small" color="primary">mdi-file-document</v-icon>
<span>{{ kb.doc_count || 0 }} {{ t('list.documents') }}</span>
@@ -45,8 +58,8 @@
</div>
</div>
<div class="kb-actions">
<v-btn icon="mdi-pencil" size="small" variant="text" color="info" @click.stop="editKB(kb)" />
<div class="kb-actions" :class="{ 'error-actions': kb.init_error }">
<v-btn v-if="!kb.init_error" icon="mdi-pencil" size="small" variant="text" color="info" @click.stop="editKB(kb)" />
<v-btn icon="mdi-delete" size="small" variant="text" color="error" @click.stop="confirmDelete(kb)" />
</div>
</div>
@@ -478,6 +491,34 @@ onMounted(() => {
box-shadow: 0 12px 24px rgba(0, 0, 0, 0.15) !important;
}
/* Error state card styles */
.kb-card-error {
cursor: not-allowed;
border: 1px solid rgba(var(--v-theme-error), 0.3);
background-color: rgba(var(--v-theme-error), 0.02) !important;
overflow: visible; /* Allow badge to overflow */
}
.kb-card-error:hover {
transform: none;
box-shadow: 0 4px 12px rgba(var(--v-theme-error), 0.1) !important;
border-color: rgba(var(--v-theme-error), 0.5);
}
.kb-card-error .kb-emoji {
opacity: 0.7;
filter: grayscale(0.5);
}
.kb-card-error .kb-name {
color: rgba(var(--v-theme-on-surface), 0.7);
}
.kb-error-badge {
z-index: 10;
opacity: 0.9;
}
.kb-card-content {
padding: 24px;
display: flex;
@@ -488,6 +529,11 @@ onMounted(() => {
position: relative;
}
.kb-card-content-error {
justify-content: center;
gap: 8px;
}
.kb-emoji {
font-size: 56px;
margin-bottom: 8px;
@@ -518,6 +564,36 @@ onMounted(() => {
justify-content: center;
}
.kb-error-panel {
width: 100%;
text-align: left;
background: rgba(var(--v-theme-error), 0.08);
border: 1px solid rgba(var(--v-theme-error), 0.18);
border-radius: 10px;
padding: 10px 12px;
}
.kb-error-title {
display: flex;
align-items: center;
gap: 8px;
font-size: 0.8rem;
font-weight: 600;
color: rgb(var(--v-theme-error));
margin-bottom: 4px;
}
.kb-error-detail {
font-size: 0.78rem;
line-height: 1.35;
color: rgba(var(--v-theme-on-surface), 0.82);
word-break: break-word;
display: -webkit-box;
-webkit-line-clamp: 3;
-webkit-box-orient: vertical;
overflow: hidden;
}
.stat-item {
display: flex;
align-items: center;

305
tests/unit/test_kb_manager_resilience.py Normal file
View File

@@ -0,0 +1,305 @@
"""
Unit tests for knowledge base manager resilience behavior.
Tests the following scenarios:
1. update_kb preserves old instance when re-initialization fails
2. update_kb switches instance only after new instance initializes successfully
3. _ensure_vec_db clears stale init_error after successful initialization
These tests use lazy imports and mocks to avoid circular import issues
in the astrbot core module chain.
"""
import sys
import types
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@pytest.fixture
def stub_provider_manager_module():
"""Stub provider manager module to avoid circular imports in unit tests."""
original_module = sys.modules.get("astrbot.core.provider.manager")
stub_module = types.ModuleType("astrbot.core.provider.manager")
class ProviderManager: ...
setattr(stub_module, "ProviderManager", ProviderManager)
sys.modules["astrbot.core.provider.manager"] = stub_module
try:
yield
finally:
if original_module is not None:
sys.modules["astrbot.core.provider.manager"] = original_module
else:
sys.modules.pop("astrbot.core.provider.manager", None)
@pytest.fixture
def mock_provider_manager():
"""Create a mock ProviderManager."""
manager = MagicMock()
manager.get_provider_by_id = AsyncMock()
manager.acm = MagicMock()
manager.acm.default_conf = {}
return manager
@pytest.fixture
def mock_kb_db():
"""Create a mock KBSQLiteDatabase."""
db = MagicMock()
db.get_db = MagicMock()
db.list_kbs = AsyncMock(return_value=[])
db.get_kb_by_id = AsyncMock()
return db
@pytest.fixture
def mock_knowledge_base():
"""Create a mock KnowledgeBase instance."""
# Use lazy import to avoid circular import
from astrbot.core.knowledge_base.models import KnowledgeBase
kb = KnowledgeBase(
kb_name="test_kb",
description="Test knowledge base",
emoji="📚",
embedding_provider_id="test-embedding-provider",
rerank_provider_id=None,
chunk_size=512,
chunk_overlap=50,
top_k_dense=50,
top_k_sparse=50,
top_m_final=5,
)
return kb
@pytest.fixture
def mock_embedding_provider():
"""Create a mock EmbeddingProvider."""
provider = MagicMock()
provider.get_embeddings_batch = AsyncMock(return_value=[[0.1, 0.2, 0.3]])
return provider
@pytest.mark.asyncio
async def test_update_kb_preserves_old_instance_when_reinit_fails(
stub_provider_manager_module,
mock_provider_manager,
mock_kb_db,
mock_knowledge_base,
mock_embedding_provider,
):
"""
Test that update_kb preserves the old KBHelper instance when
re-initialization fails, ensuring the knowledge base remains available.
"""
# Lazy import to avoid circular import
from astrbot.core.knowledge_base.kb_helper import KBHelper
from astrbot.core.knowledge_base.kb_mgr import KnowledgeBaseManager
# Setup: create an existing KBHelper with working vec_db
mock_provider_manager.get_provider_by_id.return_value = mock_embedding_provider
# Create KBHelper using __new__ to avoid __init__ side effects
old_helper = KBHelper.__new__(KBHelper)
old_helper.kb = mock_knowledge_base
old_helper.prov_mgr = mock_provider_manager
old_helper.kb_db = mock_kb_db
old_helper.kb_root_dir = "/tmp/test_kb"
old_helper.chunker = MagicMock()
old_helper.init_error = None
old_helper.vec_db = MagicMock() # Simulate existing working vec_db
old_helper.terminate = AsyncMock()
# Create KBManager and inject the existing helper
kb_mgr = KnowledgeBaseManager.__new__(KnowledgeBaseManager)
kb_mgr.provider_manager = mock_provider_manager
kb_mgr.kb_db = mock_kb_db
kb_mgr.kb_insts = {mock_knowledge_base.kb_id: old_helper}
kb_mgr.retrieval_manager = MagicMock()
# Mock KBHelper creation to simulate initialization failure
with patch.object(KBHelper, "initialize", new_callable=AsyncMock) as mock_init:
# First call (for new_helper) should fail
mock_init.side_effect = Exception("Embedding provider unavailable")
# Execute update_kb with a different embedding provider
result = await kb_mgr.update_kb(
kb_id=mock_knowledge_base.kb_id,
kb_name="updated_kb",
embedding_provider_id="new-embedding-provider",
)
# Verify: the old helper should be returned, not a new one
assert result is not None
assert result is old_helper
assert kb_mgr.kb_insts[mock_knowledge_base.kb_id] is old_helper
# Verify: old helper's vec_db should still be available
assert hasattr(result, "vec_db")
assert result.vec_db is not None
# Verify: failure does not replace the existing helper state
assert result.init_error is None
assert result.kb.kb_name == "test_kb"
assert result.kb.embedding_provider_id == "test-embedding-provider"
@pytest.mark.asyncio
async def test_update_kb_switches_instance_only_after_new_reinit_success(
stub_provider_manager_module,
mock_provider_manager,
mock_kb_db,
mock_knowledge_base,
mock_embedding_provider,
):
"""
Test that update_kb only switches to the new KBHelper instance
after the new instance successfully initializes.
"""
# Lazy import to avoid circular import
from astrbot.core.knowledge_base.kb_helper import KBHelper
from astrbot.core.knowledge_base.kb_mgr import KnowledgeBaseManager
# Setup: create an existing KBHelper
mock_provider_manager.get_provider_by_id.return_value = mock_embedding_provider
old_helper = KBHelper.__new__(KBHelper)
old_helper.kb = mock_knowledge_base
old_helper.prov_mgr = mock_provider_manager
old_helper.kb_db = mock_kb_db
old_helper.kb_root_dir = "/tmp/test_kb"
old_helper.chunker = MagicMock()
old_helper.init_error = None
old_helper.vec_db = MagicMock()
old_helper.terminate = AsyncMock()
kb_mgr = KnowledgeBaseManager.__new__(KnowledgeBaseManager)
kb_mgr.provider_manager = mock_provider_manager
kb_mgr.kb_db = mock_kb_db
kb_mgr.kb_insts = {mock_knowledge_base.kb_id: old_helper}
kb_mgr.retrieval_manager = MagicMock()
# Mock session context for database operations
mock_session = MagicMock()
mock_session.add = MagicMock()
mock_session.commit = AsyncMock()
mock_session.refresh = AsyncMock()
mock_db_context = MagicMock()
mock_db_context.__aenter__ = AsyncMock(return_value=mock_session)
mock_db_context.__aexit__ = AsyncMock()
mock_kb_db.get_db.return_value = mock_db_context
# Mock KBHelper.initialize to succeed
with patch.object(KBHelper, "initialize", new_callable=AsyncMock) as mock_init:
mock_init.return_value = None
# Execute update_kb
result = await kb_mgr.update_kb(
kb_id=mock_knowledge_base.kb_id,
kb_name="updated_kb",
embedding_provider_id="new-embedding-provider",
)
# Verify: a new helper should be returned
assert result is not None
assert result is not old_helper
assert result.init_error is None
assert kb_mgr.kb_insts[mock_knowledge_base.kb_id] is result
# Verify: old helper should be terminated
old_helper.terminate.assert_called_once()
@pytest.mark.asyncio
async def test_ensure_vec_db_clears_stale_init_error(
stub_provider_manager_module,
mock_provider_manager,
mock_kb_db,
mock_knowledge_base,
mock_embedding_provider,
):
"""
Test that _ensure_vec_db clears the init_error attribute
after successful initialization, removing stale error state.
"""
# Lazy import to avoid circular import
from astrbot.core.knowledge_base.kb_helper import KBHelper
# Setup: create KBHelper with stale init_error
mock_provider_manager.get_provider_by_id.return_value = mock_embedding_provider
helper = KBHelper.__new__(KBHelper)
helper.kb = mock_knowledge_base
helper.prov_mgr = mock_provider_manager
helper.kb_db = mock_kb_db
helper.kb_root_dir = "/tmp/test_kb"
helper.chunker = MagicMock()
helper.init_error = "Previous initialization failed"
helper.kb_dir = Path("/tmp/test_kb") / mock_knowledge_base.kb_id
helper.kb_medias_dir = helper.kb_dir / "medias" / mock_knowledge_base.kb_id
helper.kb_files_dir = helper.kb_dir / "files" / mock_knowledge_base.kb_id
# Mock FaissVecDB initialization
mock_vec_db = MagicMock()
mock_vec_db.initialize = AsyncMock()
mock_vec_db.close = AsyncMock()
with patch(
"astrbot.core.knowledge_base.kb_helper.FaissVecDB",
return_value=mock_vec_db,
):
# Execute _ensure_vec_db
await helper._ensure_vec_db()
# Verify: init_error should be cleared
assert helper.init_error is None
assert helper.vec_db is mock_vec_db
@pytest.mark.asyncio
async def test_ensure_vec_db_sets_init_error_on_failure(
stub_provider_manager_module,
mock_provider_manager,
mock_kb_db,
mock_knowledge_base,
):
"""
Test that _ensure_vec_db does NOT clear init_error when
initialization fails, preserving the error state.
"""
# Lazy import to avoid circular import
from astrbot.core.knowledge_base.kb_helper import KBHelper
# Setup: provider unavailable
mock_provider_manager.get_provider_by_id.return_value = None
helper = KBHelper.__new__(KBHelper)
helper.kb = mock_knowledge_base
helper.prov_mgr = mock_provider_manager
helper.kb_db = mock_kb_db
helper.kb_root_dir = "/tmp/test_kb"
helper.chunker = MagicMock()
helper.init_error = "Previous initialization failed"
helper.kb_dir = Path("/tmp/test_kb") / mock_knowledge_base.kb_id
helper.kb_medias_dir = helper.kb_dir / "medias" / mock_knowledge_base.kb_id
helper.kb_files_dir = helper.kb_dir / "files" / mock_knowledge_base.kb_id
# Execute _ensure_vec_db - should raise exception
try:
await helper._ensure_vec_db()
pytest.fail("Expected exception but none was raised")
except ValueError as e:
# Verify: exception should be raised
assert "无法找到" in str(e) or "未配置" in str(e)
# Verify: init_error should NOT be cleared (still has previous error)
# Note: _ensure_vec_db doesn't set init_error; that's done by the caller
assert helper.init_error is not None