Please address the comments from this code review:

## Overall Comments
- In `IndexRebuilder.sync` you build a new `EmbeddingStorage` and mutate it (delete/insert) but never call `save_index`/`close` on that instance before `FaissVecDB.switch_index` creates a *second* `EmbeddingStorage` on the same path, so the rebuilt index is never persisted to disk; consider either reusing the same storage in `switch_index` or explicitly saving/closing the temporary storage before switching.
- The structured upload path in `KBHelper._upload_document_structured` calls `upsert_doc_section` in a loop, each opening its own transaction; this could be significantly faster and reduce DB overhead if you batch-insert or upsert all sections in a single `session.begin()` block.
- In `IndexRebuilder.sync` you call `provider.get_embeddings` in a Python loop per batch instead of using the existing `get_embeddings_batch` API, which will limit throughput on large rebuilds; switching to the batched interface would align with the rest of the vector code and improve rebuild performance.

## Individual Comments

### Comment 1
<location path="astrbot/core/knowledge_base/kb_db_sqlite.py" line_range="177-186" />
<code_context>
     ) -> None:
         """导出 FAISS 索引文件"""
         try:
-            index_path = kb_helper.kb_dir / "index.faiss"
-            if index_path.exists():
</code_context>
<issue_to_address>
**suggestion (bug_risk):** v3 migration swallows all exceptions, which can hide real migration failures.

`migrate_to_v3` wraps each `ALTER`/`UPDATE` in `except Exception: pass`, which suppresses all failures (including syntax, permissions, or data issues) and can leave migrations partially applied with no signal.

Instead, only handle the expected idempotency cases explicitly (e.g. specific DB errors for duplicate/missing columns, like in `migrate_to_v2`), and log + re-raise any other exceptions so failed migrations are visible to operators.
</issue_to_address>

### Comment 2
<location path="astrbot/core/knowledge_base/kb_mgr.py" line_range="38" />
<code_context>
         self._session_deleted_callback_registered = False

         self.kb_insts: dict[str, KBHelper] = {}
+        self.rebuild_tasks: dict[str, dict] = {}
+        self.kb_rebuild_task_map: dict[str, str] = {}
+        self._running_rebuild_tasks: dict[str, asyncio.Task] = {}
</code_context>
<issue_to_address>
**issue (complexity):** Consider introducing a typed RebuildTask dataclass and central registries to replace ad‑hoc dicts and closures in the index rebuild flow while keeping the external behavior unchanged.

You can significantly reduce complexity in the rebuild flow by introducing a small typed abstraction instead of ad‑hoc dicts and nested closures, while keeping all behavior intact.

### 1. Replace loosely‑typed dicts with a `@dataclass`

Right now:

- `rebuild_tasks: dict[str, dict]`
- `kb_rebuild_task_map: dict[str, str]`
- `_running_rebuild_tasks: dict[str, asyncio.Task]`

all track aspects of the same concept using dicts with magic keys (`"status"`, `"stage"`, etc.). You can centralize this into a dataclass with explicit fields:

```python
# somewhere near top-level
from dataclasses import dataclass, field
from typing import Optional, Literal

RebuildStatus = Literal["processing", "completed", "failed"]
RebuildStage = Literal["prepare", "finished"]

@dataclass
class RebuildTask:
    task_id: str
    kb_id: str
    provider_id: str
    status: RebuildStatus = "processing"
    stage: RebuildStage = "prepare"
    current: int = 0
    total: int = 0
    error: Optional[str] = None
    task: Optional[asyncio.Task] = field(default=None, repr=False)
```

Then, in `__init__`, use a single registry keyed by `task_id` plus a thin kb→task mapping:

```python
self._rebuild_tasks: dict[str, RebuildTask] = {}
self._kb_to_task_id: dict[str, str] = {}
```

### 2. Simplify `start_rebuild_index` and progress updates

Refactor `start_rebuild_index` to work with `RebuildTask` instead of free‑form dicts and to avoid nested closures capturing outer state:

```python
async def start_rebuild_index(
    self,
    kb_id: str,
    new_provider_id: str,
    batch_size: int = 32,
) -> str:
    kb_helper = await self.get_kb(kb_id)
    if not kb_helper:
        raise ValueError("知识库不存在")

    existing_task_id = self._kb_to_task_id.get(kb_id)
    if existing_task_id:
        existing = self._rebuild_tasks.get(existing_task_id)
        if existing and existing.status == "processing":
            return existing_task_id

    task_id = str(uuid.uuid4())
    kb_helper.last_rebuild_task_id = task_id

    rebuild_task = RebuildTask(
        task_id=task_id,
        kb_id=kb_id,
        provider_id=new_provider_id,
    )
    self._kb_to_task_id[kb_id] = task_id
    self._rebuild_tasks[task_id] = rebuild_task

    async def progress(stage: str, current: int, total: int) -> None:
        t = self._rebuild_tasks.get(task_id)
        if not t:
            return
        t.stage = stage  # type: ignore[assignment]
        t.current = current
        t.total = total

    async def runner() -> None:
        try:
            await self.index_rebuilder.sync(
                kb_helper=kb_helper,
                new_provider_id=new_provider_id,
                batch_size=batch_size,
                progress_callback=progress,
            )
            t = self._rebuild_tasks.get(task_id)
            if t:
                t.status = "completed"
                t.stage = "finished"
        except Exception as e:
            logger.error(
                "Index rebuild failed for kb %s provider %s: %s",
                kb_id,
                new_provider_id,
                e,
            )
            logger.error(traceback.format_exc())
            t = self._rebuild_tasks.get(task_id)
            if t:
                t.status = "failed"
                t.error = str(e)
        finally:
            task = rebuild_task.task
            if task and not task.done():
                task.cancel()
            self._running_rebuild_tasks.pop(task_id, None)

    rebuild_task.task = asyncio.create_task(runner())
    self._running_rebuild_tasks[task_id] = rebuild_task.task
    return task_id
```

This keeps the same behavior but:

- Removes unstructured dicts and magic strings.
- Groups all state about a rebuild into one object.
- Makes the progress callback logic clearer and safer.

### 3. Simplify `get_rebuild_progress`

With the dataclass, `get_rebuild_progress` no longer needs to manipulate dicts directly:

```python
def get_rebuild_progress(
    self,
    kb_id: str | None = None,
    task_id: str | None = None,
) -> dict | None:
    target_task_id = task_id or (kb_id and self._kb_to_task_id.get(kb_id))
    if not target_task_id:
        return None

    t = self._rebuild_tasks.get(target_task_id)
    if not t:
        return None

    # Keep external API unchanged: return a dict
    return {
        "task_id": t.task_id,
        "kb_id": t.kb_id,
        "provider_id": t.provider_id,
        "status": t.status,
        "stage": t.stage,
        "current": t.current,
        "total": t.total,
        "error": t.error,
    }
```

This way:

- The internal model is strongly typed and centralized.
- The external API remains exactly as it was (still returns a dict).

### 4. Clean up `terminate` for rebuild tasks

`terminate` can cancel using the typed tasks:

```python
async def terminate(self) -> None:
    for t in list(self._rebuild_tasks.values()):
        if t.task and not t.task.done():
            t.task.cancel()
    self._running_rebuild_tasks.clear()

    # existing kb cleanup remains...
```

These focused changes keep all functionality (including the public dict-based progress interface and auto‑rebuild behavior) while consolidating job state and reducing the cognitive load around rebuild index management.
</issue_to_address>

### Comment 3
<location path="astrbot/core/db/vec_db/faiss_impl/vec_db.py" line_range="222" />
<code_context>
+                metadata_filters=metadata_filters,
+            )
+
+    async def switch_index(
+        self,
+        index_store_path: str,
</code_context>
<issue_to_address>
**issue (complexity):** Consider tightening the `_index_switch_lock` critical section in `switch_index` so it only guards the atomic swap of index-related fields instead of the whole storage creation and teardown process.

You can reduce both complexity and contention without changing behavior by tightening the critical section in `switch_index`.

Currently, `switch_index` does *everything* under `_index_switch_lock`: creating the new storage (which may be slow) and closing the old one. That’s exactly what the reviewer is pointing at: the lock is held across long-running I/O, serializing unrelated operations and making the lock’s purpose harder to reason about.

You can keep the same semantics but only hold the lock around the *field swap*; new storage is created before acquiring the lock, and old storage is closed after releasing it:

```python
async def switch_index(
    self,
    index_store_path: str,
    embedding_provider: EmbeddingProvider,
    rerank_provider: RerankProvider | None = None,
) -> None:
    """Hot-switch to a new FAISS index and embedding provider."""
    # 1. Create new storage *before* taking the lock.
    try:
        new_storage = EmbeddingStorage(
            embedding_provider.get_dim(),
            index_store_path,
        )
    except Exception:
        # If creation fails, keep old storage intact.
        raise

    # 2. Only keep the lock for the atomic swap of references.
    async with self._index_switch_lock:
        old_storage = self.embedding_storage

        self.index_store_path = index_store_path
        self.embedding_provider = embedding_provider
        self.rerank_provider = rerank_provider
        self.embedding_storage = new_storage

    # 3. Close the old storage after releasing the lock.
    if old_storage:
        await old_storage.close()
```

This keeps:

- All other methods still serialized vs. index switches (no functional change).
- The swap of `embedding_storage`/`embedding_provider` atomic and easy to reason about.

But it removes unnecessary lock contention and reduces conceptual complexity: `_index_switch_lock` now clearly protects just the “which index are we using?” pointers, not the entire lifecycle of storage creation and teardown.
</issue_to_address>

### Comment 4
<location path="astrbot/core/knowledge_base/kb_helper.py" line_range="387" />
<code_context>

             raise e

+    async def _upload_document_structured(
+        self,
+        file_name: str,
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the shared embedding and indexing workflow into a dedicated helper so both flat and structured upload paths reuse the same logic instead of duplicating it.

You can reduce the branching/duplication by extracting the common “embed + persist + stats/refresh” pipeline into a small helper that both flat and structured paths call.

Right now, both `upload_document` and `_upload_document_structured`:

- Generate `doc_id`
- Build `chunk_ids`, `contents`, `metadatas`
- Call `get_ep()` and do batched embeddings with progress
- Write to `vec_db.document_storage` and `vec_db.embedding_storage`
- Create and persist a `KBDocument`
- Update KB stats and refresh KB/doc

Only the “how to build `contents` + `metadatas` + sections” differs. You can encapsulate the common tail in a helper and keep the mode-specific logic small.

For example:

```python
async def _index_chunks(
    self,
    *,
    doc_id: str,
    contents: list[str],
    metadatas: list[dict],
    index_mode: str,
    file_name: str,
    file_type: str,
    file_size: int,
    batch_size: int,
    tasks_limit: int,
    max_retries: int,
    progress_callback=None,
) -> KBDocument:
    await self._ensure_vec_db()
    ep = await self.get_ep()

    async def embedding_progress(current: int, total: int) -> None:
        if progress_callback:
            await progress_callback("embedding", current, total)

    vectors = await ep.get_embeddings_batch(
        contents,
        batch_size=batch_size,
        tasks_limit=tasks_limit,
        max_retries=max_retries,
        progress_callback=embedding_progress if progress_callback else None,
    )

    chunk_ids = [str(uuid.uuid4()) for _ in contents]
    int_ids = await self.vec_db.document_storage.insert_documents_batch(
        chunk_ids, contents, metadatas,
    )
    await self.vec_db.embedding_storage.insert_batch(
        vectors=np.array(vectors, dtype=np.float32),
        ids=int_ids,
    )

    kb_doc = KBDocument(
        doc_id=doc_id,
        kb_id=self.kb.kb_id,
        doc_name=file_name,
        file_type=file_type,
        file_size=file_size,
        file_path="",
        index_mode=index_mode,
        chunk_count=len(contents),
        media_count=0,
    )
    async with self.kb_db.get_db() as session:
        async with session.begin():
            session.add(kb_doc)
        await session.refresh(kb_doc)

    await self.kb_db.update_kb_stats(kb_id=self.kb.kb_id, vec_db=self.vec_db)  # type: ignore[arg-type]
    await self.refresh_kb()
    await self.refresh_document(doc_id)
    return kb_doc
```

Then the two modes become mostly about preparing inputs:

**Structured:**

```python
async def _upload_document_structured(...):
    if file_content is None:
        raise ValueError("结构化上传需要 file_content")

    doc_id = str(uuid.uuid4())
    parser = await select_parser(f".{file_type}")
    if progress_callback:
        await progress_callback("parsing", 0, 100)
    parse_result = await parser.parse(file_content, file_name)
    if progress_callback:
        await progress_callback("parsing", 100, 100)

    structure_parser = StructureParser()
    if progress_callback:
        await progress_callback("chunking", 0, 100)
    nodes = await structure_parser.parse_structure(parse_result.text, file_type)
    sections = structure_parser.flatten(nodes)
    if not sections:
        logger.warning("Structured parse failed, fallback to flat mode for %s", file_name)
        return await self.upload_document(
            file_name=file_name,
            file_content=file_content,
            file_type=file_type,
            index_mode="flat",
            batch_size=batch_size,
            tasks_limit=tasks_limit,
            max_retries=max_retries,
            progress_callback=progress_callback,
        )

    contents = [section.path for section in sections]
    metadatas = [
        {
            "kb_id": self.kb.kb_id,
            "kb_doc_id": doc_id,
            "chunk_index": i,
            "index_mode": "structure",
            "section_path": section.path,
            "section_level": section.level,
        }
        for i, section in enumerate(sections)
    ]
    if progress_callback:
        await progress_callback("chunking", 100, 100)

    kb_doc = await self._index_chunks(
        doc_id=doc_id,
        contents=contents,
        metadatas=metadatas,
        index_mode="structure",
        file_name=file_name,
        file_type=file_type,
        file_size=len(file_content),
        batch_size=batch_size,
        tasks_limit=tasks_limit,
        max_retries=max_retries,
        progress_callback=progress_callback,
    )

    for i, section in enumerate(sections):
        await self.kb_db.upsert_doc_section(
            DocSection(
                doc_id=doc_id,
                kb_id=self.kb.kb_id,
                section_path=section.path,
                section_level=section.level,
                section_title=section.title,
                section_body=section.body,
                parent_section_id=None,
                sort_order=i,
            ),
        )
    return kb_doc
```

**Flat:**

Inside `upload_document`, after you’ve built `chunks_text` and flat metadatas:

```python
contents = chunks_text
metadatas = [
    {
        "kb_id": self.kb.kb_id,
        "kb_doc_id": doc_id,
        "chunk_index": i,
        "index_mode": "flat",
    }
    for i, _ in enumerate(chunks_text)
]

kb_doc = await self._index_chunks(
    doc_id=doc_id,
    contents=contents,
    metadatas=metadatas,
    index_mode="flat",
    file_name=file_name,
    file_type=file_type,
    file_size=file_size,
    batch_size=batch_size,
    tasks_limit=tasks_limit,
    max_retries=max_retries,
    progress_callback=progress_callback,
)
```

This keeps all current behavior (`index_mode`, structured sections, progress reporting) but concentrates the critical “indexing pipeline” into one place. Future changes (e.g., switching vector DB, changing metadata shape, altering stats/refresh behavior) will only need to be done in `_index_chunks`, reducing complexity and the chance of subtle drift between flat and structured paths.
</issue_to_address>