解析 ‘Dynamic Index Selection’：Agent 如何根据问题领域（医疗/法律/通用）自主切换底层向量库？

各位编程专家、AI架构师和对智能系统充满热情的听众们，大家好！

今天，我们将深入探讨一个在构建高性能、高精度智能代理时至关重要的技术主题——动态索引选择 (Dynamic Index Selection)。在当今信息爆炸的时代，智能代理不再满足于一知半解，它们被期望能像领域的专家一样，在医疗、法律、金融等特定领域提供深度、准确的答案。然而，一个包罗万象的“通用”向量库，往往难以承载如此高阶的期望。

我们的核心议题是：一个智能代理如何根据用户查询的问题领域（例如医疗、法律、通用知识），自主且智能地切换到底层最合适的向量数据库或索引？ 这不仅仅是工程上的挑战，更是提升代理智能水平、降低运营成本、优化用户体验的关键一步。

I. 引言：智能代理的挑战与动态索引选择的必要性

设想一下，你正在开发一个面向大众的AI助手。用户可能问：“我最近感到胸闷气短，这可能是什么症状？”——这是一个医疗问题。紧接着，他可能又问：“合同违约的赔偿标准是什么？”——这又是一个法律问题。如果你的代理底层只有一个庞大的、混合了所有领域知识的向量库，会发生什么？

1. 信息噪音与相关性下降：当查询一个特定领域的知识时，一个通用的大型向量库会返回大量来自不相关领域的结果。例如，搜索“症状”可能会返回关于“经济症状”或“社会症状”的文档，导致检索到的信息充满噪音，降低了核心信息的召回率和精度。
2. 性能瓶颈：随着数据量的增长，单一向量库的规模会变得极其庞大。检索操作的时间复杂度通常与索引大小相关，导致查询延迟增加，影响用户体验。
3. 维护与更新困难：不同领域的知识更新频率和维护策略可能大相径庭。将所有数据混合在一起，使得针对特定领域的增量更新、数据清洗和模型调优变得异常复杂。
4. 成本高昂：巨大的单一向量库需要更多的存储资源、计算资源（特别是对于高性能的近似最近邻搜索ANN算法）以及更高的嵌入模型推理成本，这直接转化为更高的运营开销。
5. 领域专业性缺失：特定领域的知识往往需要特定的语境理解和术语识别。一个泛化的嵌入模型和向量库，难以捕捉这些细微之处，导致在专业领域的表现力不足。

动态索引选择正是为了解决这些痛点而生。其核心思想是：将不同领域的知识存储在独立的、专门优化的向量索引中，并通过智能机制在运行时根据查询内容，自动选择最合适的索引进行检索。 这就像一个图书馆管理员，根据你提出的问题，直接把你带到医学区、法律区或历史区，而不是让你漫无目的地在大厅里寻找。

II. 核心概念回顾

在深入探讨架构之前，我们先快速回顾几个核心概念：

1. 向量数据库（Vector Databases）与向量索引（Vector Indexes）

   • 向量数据库：专门设计用于存储、管理和查询高维向量数据的数据库。它们通常内置了高效的近似最近邻（ANN）搜索算法，如HNSW、IVF等，以便在海量向量中快速找到与给定查询向量最相似的Top-K个向量。常见的向量数据库有Pinecone、Weaviate、Chroma、Qdrant等。
   • 向量索引：是向量数据库内部用于组织和加速向量搜索的数据结构。一个向量数据库可以包含多个逻辑上的向量索引或集合（collections）。在我们的语境中，“向量库”和“向量索引”可以互换使用，指的是承载特定领域知识的向量存储单元。

2. 嵌入（Embeddings）

   • 嵌入模型 (Embedding Model)：将文本、图像、音频等非结构化数据转换成高维实数向量（即“嵌入”或“向量表示”）的模型。这些向量在语义上是紧密的：含义相似的文本会映射到向量空间中距离较近的点，含义不同的则距离较远。
   • 作用：是向量检索的基础。无论是查询还是被索引的文档，都需要先通过嵌入模型转换成向量，才能进行相似度计算。

3. 智能代理（AI Agent）架构概述

   • 一个典型的智能代理通常包含以下核心组件：
     • 感知模块：接收用户输入（文本、语音等）。
     • 理解模块：解析用户意图、提取关键信息。
     • 记忆模块：短期记忆（对话历史）、长期记忆（知识库/向量库）。
     • 规划模块：根据意图和记忆制定行动计划。
     • 行动模块：执行计划（如调用外部工具、检索信息、生成响应）。
     • 反思模块：评估行动结果，进行自我修正。
   • 在动态索引选择的场景中，我们的关注点主要集中在理解模块如何识别领域，以及行动模块如何利用领域信息进行智能检索。

4. 领域分类（Domain Classification）

   • 定义：判断一段文本（如用户查询）所属的预定义类别（如医疗、法律、通用）。
   • 重要性：它是实现动态索引选择的“大脑”，决定了代理会将查询路由到哪个专业知识库。准确的领域分类是整个系统有效运行的前提。

III. 动态索引选择的架构设计

为了实现智能代理的动态索引选择能力，我们需要一个精心设计的模块化架构。以下是其核心组件及其交互方式：

整体架构概览

我们可以将整个流程想象成一个智能的问答流水线：

1. 用户查询：用户提出问题。
2. 查询接收与预处理：标准化查询格式。
3. 领域分类器：分析查询内容，识别其所属领域。
4. 索引注册与管理服务：根据分类结果，查找并提供对应领域的向量索引信息。
5. 向量库适配器接口：提供统一的API，屏蔽不同向量数据库的具体实现细节。
6. 查询路由与执行器：将查询发送到选定的领域向量库进行检索。
7. 结果聚合与后处理：收集检索结果，可能结合大语言模型（LLM）生成最终答案。

组件详解

1. 查询接收与预处理层 (Query Ingestion & Preprocessing)

这是系统的入口点。它负责接收来自用户或上游系统的原始查询，并进行初步的标准化和清洗，例如：

• 去除多余空格、标点符号。
• 统一大小写（如果语言不敏感）。
• 语言检测（如果支持多语言）。
• 对查询进行初步分词或文本规范化。

2. 领域分类器 (Domain Classifier Module)

这是动态索引选择的核心智能部分。它的任务是准确地判断用户查询的领域。

a) 基于规则/关键词的方法

优点：实现简单，易于理解和调试。
缺点：覆盖范围有限，容易误判，难以处理语义复杂的查询，维护成本随规则增加而提高。

适用场景：领域边界清晰、关键词明确、领域数量较少、对分类精度要求不极致的场景。

b) 基于机器学习的方法 (Text Classification)

优点：能够学习更复杂的语义模式，泛化能力强，对模糊查询有更好的处理能力，分类精度更高。
缺点：需要标注数据进行训练，模型训练和部署成本较高，可解释性相对较差。

适用场景：领域数量多、领域边界模糊、查询语义复杂、对分类精度要求高的生产环境。

实现方式：

• 传统ML：如支持向量机（SVM）、朴素贝叶斯、随机森林等，结合TF-IDF、Word2Vec等特征工程。
• 深度学习：如TextCNN、BERT、RoBERTa、DeBERTa等预训练语言模型，通常通过微调（fine-tuning）进行文本分类。这是目前最主流且效果最好的方法。
• 零样本/少样本学习：利用大型语言模型的few-shot或zero-shot能力进行分类，减少标注数据需求。

3. 索引注册与管理服务 (Index Registry & Management Service)

这个服务扮演着“领域到索引”的映射表角色。它维护所有可用向量索引的元数据，包括：

• 领域名称 (Domain Name)：如“medical”, “legal”, “general”。
• 向量库类型 (Vector Store Type)：如“Chroma”, “Pinecone”, “Weaviate”, “FAISS”。
• 连接配置 (Connection Config)：如API密钥、索引名称、集合名称、主机地址等。
• 嵌入模型信息 (Embedding Model Info)：该索引对应的文档是使用哪个嵌入模型生成的，确保查询向量和索引向量在同一嵌入空间。
• 描述 (Description)：对该索引内容和用途的简要说明。

作用：解耦领域分类器与具体的向量库实现，提供统一的索引管理接口。

4. 向量库适配器接口 (Vector Store Adapter Interface)

不同向量数据库的API和操作方式千差万别。为了让上层逻辑无需关心这些细节，我们需要一个抽象的适配器接口。所有的具体向量库实现都必须遵循这个接口，从而实现多态性。

核心方法：

• retrieve(query_vector, top_k, filters)：根据查询向量进行相似度搜索。
• add_documents(documents)：向索引中添加文档。
• delete_documents(ids)：从索引中删除文档。
• update_documents(documents)：更新索引中的文档。

5. 查询路由与执行器 (Query Router & Executor)

这是将领域分类结果转化为实际检索行动的组件。

• 它接收领域分类器返回的领域标签。
• 通过索引注册服务获取对应领域的向量库适配器实例。
• 使用预设的嵌入模型将用户查询转换为向量。
• 调用适配器的retrieve方法，执行向量搜索。

6. 结果聚合与后处理 (Result Aggregation & Post-processing)

从选定向量库中检索到的原始文档片段或ID，可能需要进一步的处理：

• 排序与去重：对检索结果进行二次排序，去除重复或低质量的文档。
• 结合LLM生成答案：最常见的做法是将检索到的相关文档作为上下文（Context）输入给大型语言模型（LLM），由LLM综合这些信息生成一个连贯、准确的答案。这便是RAG（Retrieval-Augmented Generation）架构的核心。
• 格式化输出：将最终答案以用户友好的格式呈现。

IV. 详细实现与代码示例 (Python)

现在，让我们通过Python代码来逐步构建这个动态索引选择系统。我们将使用一些流行的库，如transformers用于领域分类，sentence-transformers用于生成嵌入向量。对于向量数据库，我们将抽象其接口，并以Chroma和Pinecone为例说明其适配器。

1. 向量库适配器接口设计

首先定义一个抽象基类（Abstract Base Class, ABC）作为所有向量库适配器的接口。

import abc
from typing import List, Dict, Any, Optional

# 假设文档结构
class Document:
    def __init__(self, page_content: str, metadata: Optional[Dict[str, Any]] = None, doc_id: Optional[str] = None):
        self.page_content = page_content
        self.metadata = metadata if metadata is not None else {}
        self.doc_id = doc_id

    def to_dict(self):
        return {"page_content": self.page_content, "metadata": self.metadata, "doc_id": self.doc_id}

    @classmethod
    def from_dict(cls, data: Dict[str, Any]):
        return cls(data.get("page_content", ""), data.get("metadata"), data.get("doc_id"))

# 抽象向量库适配器接口
class VectorStoreAdapter(abc.ABC):
    """
    抽象向量库适配器接口，定义了与不同向量数据库交互的通用方法。
    """
    def __init__(self, config: Dict[str, Any]):
        self.config = config

    @abc.abstractmethod
    def retrieve(self, query_vector: List[float], top_k: int = 5, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
        """
        根据查询向量检索最相似的文档。
        :param query_vector: 查询文本的嵌入向量。
        :param top_k: 返回最相似文档的数量。
        :param filters: 用于元数据过滤的字典。
        :return: 相似文档列表。
        """
        pass

    @abc.abstractmethod
    def add_documents(self, documents: List[Document]) -> None:
        """
        向向量库中添加文档。
        :param documents: 待添加的文档列表。
        """
        pass

    @abc.abstractmethod
    def delete_documents(self, doc_ids: List[str]) -> None:
        """
        根据文档ID从向量库中删除文档。
        :param doc_ids: 待删除文档的ID列表。
        """
        pass

    @abc.abstractmethod
    def update_documents(self, documents: List[Document]) -> None:
        """
        更新向量库中的文档。
        :param documents: 待更新的文档列表（包含doc_id）。
        """
        pass

    @property
    @abc.abstractmethod
    def embedding_dimension(self) -> int:
        """
        返回向量库期望的嵌入向量维度。
        """
        pass

# 示例：Chroma向量库适配器（简化版，仅作示意）
# 实际使用需要安装 chromadb 库
# from chromadb import Client as ChromaClient
# from chromadb.utils import embedding_functions

class ChromaAdapter(VectorStoreAdapter):
    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        # 实际这里会初始化 ChromaClient 并获取 collection
        # self.client = ChromaClient(...)
        # self.collection_name = config.get("collection_name", "default_collection")
        # self.collection = self.client.get_or_create_collection(self.collection_name,
        #                                                       embedding_function=embedding_functions.SentenceTransformerEmbeddingFunction(model_name="paraphrase-multilingual-MiniLM-L12-v2"))
        print(f"初始化 ChromaAdapter, 集合: {config.get('collection_name', 'default')}")
        self._dummy_data: Dict[str, Document] = {} # 模拟存储
        self._embedding_dim = 384 # 示例维度

    def retrieve(self, query_vector: List[float], top_k: int = 5, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
        print(f"ChromaAdapter: 检索 {top_k} 个文档 (模拟)")
        # 实际这里会调用 Chroma 的查询API
        # results = self.collection.query(query_embeddings=[query_vector], n_results=top_k, where=filters)
        # return [Document(page_content=doc['document'], metadata=doc['metadata']) for doc in results['documents']]

        # 模拟返回与查询向量“最相似”的文档
        # 实际会进行向量相似度计算，这里只是简单返回
        simulated_results = []
        for doc_id, doc in self._dummy_data.items():
            simulated_results.append(doc)
            if len(simulated_results) >= top_k:
                break
        return simulated_results

    def add_documents(self, documents: List[Document]) -> None:
        print(f"ChromaAdapter: 添加 {len(documents)} 个文档 (模拟)")
        # 实际这里会生成文档嵌入并添加到 Chroma
        # self.collection.add(documents=[d.page_content for d in documents],
        #                    metadatas=[d.metadata for d in documents],
        #                    ids=[d.doc_id for d in documents if d.doc_id])
        for doc in documents:
            self._dummy_data[doc.doc_id if doc.doc_id else str(hash(doc.page_content))] = doc

    def delete_documents(self, doc_ids: List[str]) -> None:
        print(f"ChromaAdapter: 删除 {len(doc_ids)} 个文档 (模拟)")
        for doc_id in doc_ids:
            self._dummy_data.pop(doc_id, None)

    def update_documents(self, documents: List[Document]) -> None:
        print(f"ChromaAdapter: 更新 {len(documents)} 个文档 (模拟)")
        self.add_documents(documents) # 简化，直接覆盖

    @property
    def embedding_dimension(self) -> int:
        return self._embedding_dim

# 示例：Pinecone向量库适配器（简化版，仅作示意）
# 实际使用需要安装 pinecone-client 库
# from pinecone import init, Index
class PineconeAdapter(VectorStoreAdapter):
    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        # 实际这里会初始化 Pinecone 并连接到 index
        # init(api_key=config["api_key"], environment=config["environment"])
        # self.index_name = config.get("index_name", "default_index")
        # self.index = Index(self.index_name)
        print(f"初始化 PineconeAdapter, 索引: {config.get('index_name', 'default')}")
        self._dummy_data: Dict[str, Document] = {} # 模拟存储
        self._embedding_dim = 768 # 示例维度

    def retrieve(self, query_vector: List[float], top_k: int = 5, filters: Optional[Dict[str, Any]] = None) -> List[Document]:
        print(f"PineconeAdapter: 检索 {top_k} 个文档 (模拟)")
        # 实际这里会调用 Pinecone 的查询API
        # query_results = self.index.query(vector=query_vector, top_k=top_k, filter=filters, include_metadata=True)
        # return [Document(page_content=match.metadata.get('text', ''), metadata=match.metadata, doc_id=match.id)
        #         for match in query_results.matches]

        simulated_results = []
        for doc_id, doc in self._dummy_data.items():
            simulated_results.append(doc)
            if len(simulated_results) >= top_k:
                break
        return simulated_results

    def add_documents(self, documents: List[Document]) -> None:
        print(f"PineconeAdapter: 添加 {len(documents)} 个文档 (模拟)")
        # 实际这里会生成文档嵌入并上传到 Pinecone
        # vectors = []
        # for doc in documents:
        #    embedding = self.embedding_model.encode(doc.page_content).tolist()
        #    vectors.append({"id": doc.doc_id, "values": embedding, "metadata": doc.metadata})
        # self.index.upsert(vectors=vectors)
        for doc in documents:
            self._dummy_data[doc.doc_id if doc.doc_id else str(hash(doc.page_content))] = doc

    def delete_documents(self, doc_ids: List[str]) -> None:
        print(f"PineconeAdapter: 删除 {len(doc_ids)} 个文档 (模拟)")
        for doc_id in doc_ids:
            self._dummy_data.pop(doc_id, None)

    def update_documents(self, documents: List[Document]) -> None:
        print(f"PineconeAdapter: 更新 {len(documents)} 个文档 (模拟)")
        self.add_documents(documents) # 简化，直接覆盖

    @property
    def embedding_dimension(self) -> int:
        return self._embedding_dim

说明：为了避免真实DB连接和凭证问题，这里的ChromaAdapter和PineconeAdapter是高度简化的“模拟”版本，它们只打印操作信息并使用一个字典_dummy_data来模拟存储。在实际项目中，你需要替换为真正的SDK调用。

2. 索引注册与管理服务

我们将创建一个IndexRegistry类来管理不同领域的向量索引。

class IndexRegistry:
    def __init__(self):
        self._registry: Dict[str, VectorStoreAdapter] = {}
        self._domain_configs: Dict[str, Dict[str, Any]] = {}

    def register_index(self, domain_name: str, adapter: VectorStoreAdapter, config: Dict[str, Any]):
        """
        注册一个向量索引到特定领域。
        :param domain_name: 领域名称（如 "medical", "legal"）。
        :param adapter: 向量库适配器实例。
        :param config: 该索引的配置信息。
        """
        if domain_name in self._registry:
            print(f"警告: 领域 '{domain_name}' 已存在，将被覆盖。")
        self._registry[domain_name] = adapter
        self._domain_configs[domain_name] = config
        print(f"注册领域 '{domain_name}' 成功。")

    def get_index_adapter(self, domain_name: str) -> VectorStoreAdapter:
        """
        根据领域名称获取对应的向量库适配器。
        :param domain_name: 领域名称。
        :return: 向量库适配器实例。
        :raises ValueError: 如果领域未注册。
        """
        if domain_name not in self._registry:
            raise ValueError(f"错误: 没有为领域 '{domain_name}' 注册任何索引。")
        return self._registry[domain_name]

    def get_index_config(self, domain_name: str) -> Dict[str, Any]:
        """
        获取某个领域的索引配置。
        :param domain_name: 领域名称。
        :return: 索引配置字典。
        """
        return self._domain_configs.get(domain_name, {})

    def list_domains(self) -> List[str]:
        """列出所有已注册的领域。"""
        return list(self._registry.keys())

# 初始化并填充注册表
registry = IndexRegistry()

# 假设我们有以下配置
medical_config = {"collection_name": "medical_docs_collection", "type": "chroma", "description": "医疗疾病、症状、治疗方案知识"}
legal_config = {"index_name": "legal_cases_index", "type": "pinecone", "api_key": "YOUR_PINECONE_API_KEY", "environment": "YOUR_PINECONE_ENV", "description": "法律法规、判例、合同模板知识"}
general_config = {"collection_name": "general_knowledge_collection", "type": "chroma", "description": "通用百科、历史、科学知识"}

# 实例化适配器并注册
medical_adapter = ChromaAdapter(medical_config)
legal_adapter = PineconeAdapter(legal_config)
general_adapter = ChromaAdapter(general_config) # 假设通用知识也用Chroma

registry.register_index("medical", medical_adapter, medical_config)
registry.register_index("legal", legal_adapter, legal_config)
registry.register_index("general", general_adapter, general_config)

print("n已注册的领域:", registry.list_domains())

# 填充一些模拟数据
medical_adapter.add_documents([
    Document("胸闷气短可能是心血管疾病的症状，也可能是呼吸系统问题。", metadata={"source": "medical_handbook", "tag": "symptom"}, doc_id="med_001"),
    Document("高血压患者应定期测量血压，并遵医嘱服用降压药。", metadata={"source": "medical_guide", "tag": "treatment"}, doc_id="med_002")
])
legal_adapter.add_documents([
    Document("根据《中华人民共和国合同法》规定，违约方应承担赔偿责任。", metadata={"source": "law_text", "tag": "contract"}, doc_id="leg_001"),
    Document("知识产权侵权纠纷通常涉及专利、商标和著作权。", metadata={"source": "legal_faq", "tag": "ip_law"}, doc_id="leg_002")
])
general_adapter.add_documents([
    Document("黑洞是宇宙中引力极强的区域，任何物质都无法逃脱。", metadata={"source": "science_wiki", "tag": "astronomy"}, doc_id="gen_001"),
    Document("人工智能（AI）正在改变我们的生活和工作方式。", metadata={"source": "tech_blog", "tag": "technology"}, doc_id="gen_002")
])
print("n模拟数据已填充到各索引。")

索引配置表格示例

领域名称	向量库类型	索引/集合名称	嵌入维度	描述
medical	Chroma	medical_docs_collection	384	医疗疾病、症状、治疗方案知识
legal	Pinecone	legal_cases_index	768	法律法规、判例、合同模板知识
general	Chroma	general_knowledge_collection	384	通用百科、历史、科学知识

3. 领域分类器

我们将展示两种分类器：基于关键词的简单分类器和基于机器学习的分类器。

a) 基于规则/关键词的简单分类器

class KeywordDomainClassifier:
    def __init__(self):
        # 规则字典：键为领域名，值为关键词列表
        self.rules = {
            "medical": ["症状", "诊断", "药物", "疾病", "医院", "医生", "治疗", "疫苗", "健康", "药店"],
            "legal": ["法律", "法规", "判例", "律师", "合同", "诉讼", "法庭", "权利", "纠纷", "起诉", "条例"],
            "general": [] # 作为默认或回退领域
        }

    def classify(self, query: str) -> str:
        """
        根据查询中的关键词进行领域分类。
        :param query: 用户查询字符串。
        :return: 识别出的领域名称，如果未匹配则返回 "general"。
        """
        query_lower = query.lower()
        for domain, keywords in self.rules.items():
            if domain == "general": # 'general'是回退，不参与关键词匹配
                continue
            if any(kw in query_lower for kw in keywords):
                return domain
        return "general" # 如果没有任何特定领域关键词，则归类为通用领域

# 实例化简单分类器
keyword_classifier = KeywordDomainClassifier()

b) 基于机器学习的分类器

这里我们将使用transformers库中的pipeline功能来模拟一个文本分类器。在实际应用中，你需要一个针对你的领域数据进行过微调的模型。

from transformers import pipeline
import os

class MLDomainClassifier:
    def __init__(self, model_name: str = "uer/roberta-base-chinese-ext-domain-classification"):
        # 实际生产中，model_name 应该指向你自己的微调模型路径或 Hugging Face Hub上的模型。
        # 注意：首次运行时会自动下载模型，可能需要时间。
        # 为了演示，如果下载失败，我们使用一个简单的回退机制。
        try:
            # 尝试加载预训练/微调的文本分类模型
            # 这里的模型名称是示意性的，需要替换为实际可用的中文分类模型
            self.classifier = pipeline("text-classification", model=model_name, tokenizer=model_name)
            print(f"成功加载机器学习分类模型: {model_name}")
            # 假设模型的输出标签与我们的领域名称匹配，或者需要一个映射
            self._label_map = {
                "LABEL_0": "medical",  # 假设模型输出的标签
                "LABEL_1": "legal",
                "LABEL_2": "general",
                "医疗": "medical", # 如果模型直接输出中文标签
                "法律": "legal",
                "通用": "general"
            }
        except Exception as e:
            print(f"警告: 无法加载机器学习模型 '{model_name}': {e}。将回退到基于关键词的分类。")
            self.classifier = None
            self._keyword_fallback = KeywordDomainClassifier() # 作为回退

    def classify(self, query: str) -> str:
        """
        使用机器学习模型对查询进行领域分类。
        :param query: 用户查询字符串。
        :return: 识别出的领域名称。
        """
        if self.classifier:
            try:
                # 机器学习分类
                results = self.classifier(query, top_k=1) # 获取置信度最高的分类结果
                if results and len(results) > 0:
                    predicted_label = results[0]['label']
                    # 尝试从映射中获取领域，如果不存在则尝试直接使用
                    domain = self._label_map.get(predicted_label, predicted_label.lower())
                    if domain in registry.list_domains(): # 确保是已注册的领域
                        return domain
                # 如果模型分类结果不明确或不在注册表中，则回退
                print(f"机器学习分类结果 '{predicted_label}' 不明确或未注册，回退到关键词分类。")
                return self._keyword_fallback.classify(query)
            except Exception as e:
                print(f"机器学习分类过程中发生错误: {e}。回退到关键词分类。")
                return self._keyword_fallback.classify(query)
        else:
            # 如果机器学习模型未加载成功，则使用关键词回退
            return self._keyword_fallback.classify(query)

# 实例化机器学习分类器
# 注意：首次运行可能需要下载模型，请确保网络连接。
# 为避免下载问题，此处暂时注释掉真实模型名，用一个假的模型名触发回退。
# 实际应使用如 "bert-base-chinese-text-classification-medical" 或 "textattack/bert-base-uncased-SST-2" (英文示例)
ml_classifier = MLDomainClassifier(model_name="dummy/model-for-domain-classification") # 此处传入一个不存在的模型名，强制走关键词回退
# 或者，如果你有一个可用的中文分类模型，请替换为它的名称，例如：
# ml_classifier = MLDomainClassifier(model_name="path/to/your/fine_tuned_model")

说明：在生产环境中，你通常会先离线训练并微调一个基于BERT或RoBERTa的文本分类模型，然后将其部署。这里的MLDomainClassifier展示了如何集成这样的模型。如果模型加载失败，它会优雅地回退到关键词分类器，保证系统的鲁棒性。

4. 智能代理核心编排器 (Agent Orchestrator)

这是将所有组件整合在一起的“大脑”，负责接收查询、分类、路由和执行检索。

from sentence_transformers import SentenceTransformer

class DynamicIndexAgent:
    def __init__(self, domain_classifier, index_registry, embedding_model_name: str = "paraphrase-multilingual-MiniLM-L12-v2"):
        """
        初始化动态索引代理。
        :param domain_classifier: 领域分类器实例。
        :param index_registry: 索引注册表实例。
        :param embedding_model_name: 用于生成查询嵌入的 Sentence Transformer 模型名称。
        """
        self.domain_classifier = domain_classifier
        self.index_registry = index_registry
        # 加载嵌入模型，用于将查询转换为向量
        try:
            self.embedding_model = SentenceTransformer(embedding_model_name)
            print(f"成功加载嵌入模型: {embedding_model_name}")
        except Exception as e:
            print(f"错误: 无法加载嵌入模型 '{embedding_model_name}': {e}。请检查模型名称和网络连接。")
            raise

    def process_query(self, query: str, top_k: int = 5) -> List[Document]:
        """
        处理用户查询，动态选择索引并执行检索。
        :param query: 用户查询字符串。
        :param top_k: 检索最相似文档的数量。
        :return: 检索到的文档列表。
        """
        print(f"n--- 处理查询: '{query}' ---")

        # 1. 领域分类
        domain = self.domain_classifier.classify(query)
        print(f"分类器将查询 '{query}' 归类到领域: {domain}")

        # 2. 获取对应向量库适配器
        adapter: VectorStoreAdapter
        try:
            adapter = self.index_registry.get_index_adapter(domain)
        except ValueError as e:
            print(f"错误: {e}. 回退到通用索引 ('general')。")
            domain = "general" # 强制回退到通用领域
            adapter = self.index_registry.get_index_adapter(domain) # 重新获取通用适配器

        # 检查嵌入维度是否匹配
        # 实际场景中，注册表应存储每个索引的embedding model信息，确保一致性
        if self.embedding_model.get_sentence_embedding_dimension() != adapter.embedding_dimension:
            print(f"警告: 当前嵌入模型维度 ({self.embedding_model.get_sentence_embedding_dimension()}) 与 '{domain}' 索引期望维度 ({adapter.embedding_dimension}) 不匹配。这可能导致检索结果不准确！")
            # 真实场景中，这里可能需要加载对应索引的嵌入模型，或者抛出错误。
            # 为了演示，我们继续使用当前模型，但给出警告。

        # 3. 生成查询向量
        query_vector = self.embedding_model.encode(query).tolist()
        print(f"查询 '{query}' 的嵌入向量已生成。")

        # 4. 执行向量检索
        results = adapter.retrieve(query_vector, top_k=top_k)
        print(f"从 '{domain}' 索引检索到 {len(results)} 个结果。")

        # 5. 后处理 (可选: 结合LLM生成答案等)
        # 在这个讲座中，我们直接返回原始检索结果。
        # 实际应用中，这里会集成LLM进行RAG。
        # 例如：
        # context = "n".join([doc.page_content for doc in results])
        # final_answer = self.llm.generate(prompt=f"根据以下信息回答问题：{query}nn信息：{context}")
        # return final_answer

        return results

# --- 实例化并运行示例 ---

# 选择使用哪个分类器
# current_classifier = keyword_classifier # 使用关键词分类器
current_classifier = ml_classifier # 使用机器学习分类器 (如果模型加载失败会回退到关键词)

# 初始化代理
agent = DynamicIndexAgent(current_classifier, registry)

# 医疗查询
query_medical = "我最近感到胸闷气短，这可能是什么症状？"
results_medical = agent.process_query(query_medical)
print("n--- 医疗查询结果 ---")
for i, doc in enumerate(results_medical):
    print(f"{i+1}. 内容: {doc.page_content[:50]}..., 元数据: {doc.metadata}")

# 法律查询
query_legal = "合同违约的赔偿标准是什么？"
results_legal = agent.process_query(query_legal)
print("n--- 法律查询结果 ---")
for i, doc in enumerate(results_legal):
    print(f"{i+1}. 内容: {doc.page_content[:50]}..., 元数据: {doc.metadata}")

# 通用查询
query_general = "什么是黑洞？"
results_general = agent.process_query(query_general)
print("n--- 通用查询结果 ---")
for i, doc in enumerate(results_general):
    print(f"{i+1}. 内容: {doc.page_content[:50]}..., 元数据: {doc.metadata}")

# 一个未被特定关键词命中的查询，应回退到通用
query_unclassified = "给我讲个笑话。"
results_unclassified = agent.process_query(query_unclassified)
print("n--- 未分类查询结果 (回退到通用) ---")
for i, doc in enumerate(results_unclassified):
    print(f"{i+1}. 内容: {doc.page_content[:50]}..., 元数据: {doc.metadata}")

运行结果（部分模拟输出）：

初始化 ChromaAdapter, 集合: medical_docs_collection
初始化 PineconeAdapter, 索引: legal_cases_index
初始化 ChromaAdapter, 集合: general_knowledge_collection
注册领域 'medical' 成功。
注册领域 'legal' 成功。
注册领域 'general' 成功。

已注册的领域: ['medical', 'legal', 'general']
ChromaAdapter: 添加 2 个文档 (模拟)
PineconeAdapter: 添加 2 个文档 (模拟)
ChromaAdapter: 添加 2 个文档 (模拟)

模拟数据已填充到各索引。
警告: 无法加载机器学习模型 'dummy/model-for-domain-classification': No such model named dummy/model-for-domain-classification is found in the models repository. 。将回退到基于关键词的分类。
成功加载嵌入模型: paraphrase-multilingual-MiniLM-L12-v2

--- 处理查询: '我最近感到胸闷气短，这可能是什么症状？' ---
分类器将查询 '我最近感到胸闷气短，这可能是什么症状？' 归类到领域: medical
查询 '我最近感到胸闷气短，这可能是什么症状？' 的嵌入向量已生成。
ChromaAdapter: 检索 5 个文档 (模拟)
从 'medical' 索引检索到 2 个结果。

--- 医疗查询结果 ---
1. 内容: 胸闷气短可能是心血管疾病的症状，也可能是呼吸系统问题。, 元数据: {'source': 'medical_handbook', 'tag': 'symptom'}
2. 内容: 高血压患者应定期测量血压，并遵医嘱服用降压药。, 元数据: {'source': 'medical_guide', 'tag': 'treatment'}

--- 处理查询: '合同违约的赔偿标准是什么？' ---
分类器将查询 '合同违约的赔偿标准是什么？' 归类到领域: legal
查询 '合同违约的赔偿标准是什么？' 的嵌入向量已生成。
PineconeAdapter: 检索 5 个文档 (模拟)
从 'legal' 索引检索到 2 个结果。

--- 法律查询结果 ---
1. 内容: 根据《中华人民共和国合同法》规定，违约方应承担赔偿责任。, 元数据: {'source': 'law_text', 'tag': 'contract'}
2. 内容: 知识产权侵权纠纷通常涉及专利、商标和著作权。, 元数据: {'source': 'legal_faq', 'tag': 'ip_law'}

--- 处理查询: '什么是黑洞？' ---
分类器将查询 '什么是黑洞？' 归类到领域: general
查询 '什么是黑洞？' 的嵌入向量已生成。
ChromaAdapter: 检索 5 个文档 (模拟)
从 'general' 索引检索到 2 个结果。

--- 通用查询结果 ---
1. 内容: 黑洞是宇宙中引力极强的区域，任何物质都无法逃脱。, 元数据: {'source': 'science_wiki', 'tag': 'astronomy'}
2. 内容: 人工智能（AI）正在改变我们的生活和工作方式。, 元数据: {'source': 'tech_blog', 'tag': 'technology'}

--- 处理查询: '给我讲个笑话。' ---
分类器将查询 '给我讲个笑话。' 归类到领域: general
查询 '给我讲个笑话。' 的嵌入向量已生成。
ChromaAdapter: 检索 5 个文档 (模拟)
从 'general' 索引检索到 2 个结果。

--- 未分类查询结果 (回退到通用) ---
1. 内容: 黑洞是宇宙中引力极强的区域，任何物质都无法逃脱。, 元数据: {'source': 'science_wiki', 'tag': 'astronomy'}
2. 内容: 人工智能（AI）正在改变我们的生活和工作方式。, 元数据: {'source': 'tech_blog', 'tag': 'technology'}

从模拟输出可以看出，系统成功地根据查询内容将请求路由到了不同的“向量库”（即不同的Adapter实例），即使这些Adapter只是模拟了行为。这正是动态索引选择的核心效果。

V. 高级考量与未来展望

动态索引选择并非一劳永逸的解决方案，在实际部署中还需要考虑诸多高级问题：

1. 多领域查询处理：

   • 问题：有些查询可能涉及多个领域，例如“医疗纠纷如何通过法律途径解决？”。单一分类器可能难以准确归类。
   • 解决方案：
     • 多标签分类：训练一个能够为查询分配多个领域标签的分类器。
     • 并行检索与结果融合：同时向多个被识别的领域索引发送查询，然后将多个索引的检索结果进行智能融合、去重和排序。
     • 分层分类：先粗粒度分类（如“专业领域” vs “通用”），再细粒度分类。

2. 性能与成本优化：

   • 懒加载 (Lazy Loading)：只在需要时才初始化或连接特定的向量库适配器，减少启动时间和资源占用。
   • 缓存 (Caching)：缓存领域分类结果和热门查询的检索结果，避免重复计算。
   • 异步检索 (Asynchronous Retrieval)：对于并行检索场景，使用异步I/O可以显著提高响应速度。
   • 资源调度：根据负载动态调整向量数据库实例的规模。

3. 索引的动态更新与维护：

   • 数据新鲜度：不同领域的知识更新速度不同（如新闻、股票信息需要实时更新，法律法规更新周期较长）。需要有机制定期或实时地更新特定领域的索引。
   • 版本控制：对索引进行版本管理，方便回滚或A/B测试。
   • 监控与告警：监控索引的健康状况、查询性能和数据质量。

4. 安全与合规：

   • 数据隔离：不同领域的敏感数据（如医疗隐私、法律机密）必须严格隔离存储和访问。动态索引选择天然支持这种隔离。
   • 访问控制：为不同的领域索引配置独立的访问权限，确保只有授权的组件或用户才能访问。
   • 合规性：确保数据存储和处理符合GDPR、HIPAA等相关法规。

5. 反馈机制与自适应学习：

   • 用户反馈：收集用户对答案满意度的反馈，用于评估和优化领域分类器和检索模型。
   • 模型迭代：定期重新训练和更新领域分类模型，以适应新的查询模式和领域发展。
   • 不确定性处理：当领域分类器对某个查询的置信度较低时，可以触发人工审核、转向通用索引或寻求用户澄清。

6. 可解释性：

   • 当代理给出答案时，如果能说明“我根据您的提问，判断这是一个医疗问题，并在医疗知识库中找到了这些信息”，将大大增强用户的信任感。

7. 混合检索策略：

   • 动态索引选择可以与关键词搜索、知识图谱、传统数据库查询等其他检索技术结合，形成更强大的混合检索系统，以应对更复杂的查询场景。

VI. 智能代理的演进之路，动态索引选择是其关键一步

通过本场讲座，我们深入探讨了智能代理如何通过动态索引选择，根据问题领域自主切换底层向量库。我们从单一向量库的局限性出发，逐步构建了一个模块化、可扩展的架构，并辅以Python代码示例，演示了领域分类器、索引注册表和向量库适配器如何协同工作。

这种机制不仅显著提升了智能代理在特定领域的检索精度和答案相关性，有效降低了信息噪音，同时也优化了系统的性能和可扩展性，使得对海量、多源异构知识的管理变得更加高效和经济。动态索引选择是构建真正智能、上下文感知且领域专业化AI代理的关键组成部分，它使代理能够更精准地理解用户意图，更有效地利用其知识储备，最终提供更优质的用户体验。随着AI技术的不断发展，这种智能路由能力将成为未来复杂智能系统不可或缺的一部分。