Agent向量数据库选型与工程实践

向量数据库是Agent记忆系统和RAG检索系统的基础设施。Agent需要存储和检索大量的语义向量——记忆嵌入、知识分块嵌入、用户画像向量——这些向量的存储效率与检索速度直接决定Agent的响应性能。本文将深入对比主流向量数据库方案，剖析索引算法原理，讨论性能调优策略，并给出可落地的向量存储层代码实现。

一、向量数据库在Agent中的角色

Agent系统中的向量数据管理面临几个独特挑战。首先是写入模式的不均匀性：知识库导入时是批量写入，而记忆系统运行时是持续的单条写入。其次是查询模式的多样性：有时需要精确的top-K检索，有时需要范围查询，有时需要结合元数据过滤。最后是一致性要求：Agent的记忆更新需要即时可见，而知识库的批量导入可以容忍最终一致性。

不同的向量数据库在这些维度上各有取舍。选择合适的向量数据库不是简单的性能对比，而是需要结合Agent的具体使用场景、数据规模、部署环境和团队能力进行综合评估。

二、主流向量数据库对比

FAISS

FAISS（Facebook AI Similarity Search）是Meta开源的向量相似度搜索库，严格来说它不是一个数据库，而是一个搜索库。FAISS的核心优势在于极致的检索性能和丰富的索引类型支持。它提供了从暴力搜索（Flat）到近似搜索（IVF、PQ、HNSW）的完整索引方案，支持GPU加速，单机可处理十亿级向量。

FAISS的局限在于缺乏数据管理能力。它没有持久化存储（数据在内存中，进程退出即丢失）、没有事务支持、没有内置的分布式方案、不支持元数据过滤。在Agent系统中，FAISS适合作为嵌入式检索引擎，由上层应用负责数据持久化和元数据管理。

import numpy as np
from typing import List, Tuple, Optional
import pickle
import os

class FAISSVectorStore:
    """基于FAISS的向量存储封装"""
    
    def __init__(self, dim: int, index_type: str = "ivf"):
        self.dim = dim
        self.index_type = index_type
        self.index = None
        self.id_map = {}  # FAISS内部ID到业务ID的映射
        self.reverse_map = {}  # 业务ID到FAISS内部ID
        self.metadata = {}  # 业务ID到元数据
        self._next_id = 0
        self._init_index()
    
    def _init_index(self):
        """初始化FAISS索引"""
        import faiss
        
        if self.index_type == "flat":
            # 精确搜索，适合小数据集
            self.index = faiss.IndexFlatIP(self.dim)
        elif self.index_type == "ivf":
            # IVF近似搜索
            nlist = 100  # 聚类中心数
            quantizer = faiss.IndexFlatIP(self.dim)
            self.index = faiss.IndexIVFFlat(quantizer, self.dim, nlist, 
                                            faiss.METRIC_INNER_PRODUCT)
            self.index.nprobe = 10  # 搜索时探测的聚类数
        elif self.index_type == "hnsw":
            # HNSW图索引
            self.index = faiss.IndexHNSWFlat(self.dim, 32)
            self.index.hnsw.efConstruction = 40
            self.index.hnsw.efSearch = 16
        elif self.index_type == "ivf_pq":
            # IVF + 乘积量化，压缩存储
            nlist = 100
            m = 16  # 子向量数
            quantizer = faiss.IndexFlatIP(self.dim)
            self.index = faiss.IndexIVFPQ(quantizer, self.dim, nlist, m, 8)
    
    def add(self, item_id: str, vector: np.ndarray, 
            metadata: Optional[dict] = None):
        """添加向量"""
        vec = vector.reshape(1, -1).astype(np.float32)
        
        # IVF索引需要先训练
        if self.index_type in ("ivf", "ivf_pq") and not self.index.is_trained:
            # 需要足够的数据进行训练
            pass  # 实际应用中需要先收集训练数据
        
        internal_id = self._next_id
        self._next_id += 1
        
        self.index.add(vec)
        self.id_map[internal_id] = item_id
        self.reverse_map[item_id] = internal_id
        if metadata:
            self.metadata[item_id] = metadata
    
    def batch_add(self, items: List[Tuple[str, np.ndarray, dict]]):
        """批量添加"""
        vectors = []
        for item_id, vec, meta in items:
            internal_id = self._next_id
            self._next_id += 1
            self.id_map[internal_id] = item_id
            self.reverse_map[item_id] = internal_id
            if meta:
                self.metadata[item_id] = meta
            vectors.append(vec)
        
        matrix = np.stack(vectors).astype(np.float32)
        
        # IVF/PQ需要训练
        if self.index_type in ("ivf", "ivf_pq") and not self.index.is_trained:
            self.index.train(matrix)
        
        self.index.add(matrix)
    
    def search(self, query_vec: np.ndarray, top_k: int = 10,
               filter_fn=None) -> List[Tuple[str, float, dict]]:
        """检索"""
        vec = query_vec.reshape(1, -1).astype(np.float32)
        scores, indices = self.index.search(vec, top_k * 2)
        
        results = []
        for i, (score, idx) in enumerate(zip(scores[0], indices[0])):
            if idx == -1:
                continue
            item_id = self.id_map.get(int(idx))
            if item_id is None:
                continue
            meta = self.metadata.get(item_id, {})
            # 元数据过滤
            if filter_fn and not filter_fn(meta):
                continue
            results.append((item_id, float(score), meta))
            if len(results) >= top_k:
                break
        
        return results
    
    def save(self, path: str):
        """持久化"""
        import faiss
        faiss.write_index(self.index, f"{path}.index")
        with open(f"{path}.meta", "wb") as f:
            pickle.dump({
                "id_map": self.id_map,
                "reverse_map": self.reverse_map,
                "metadata": self.metadata,
                "next_id": self._next_id,
                "dim": self.dim,
                "index_type": self.index_type
            }, f)
    
    def load(self, path: str):
        """加载"""
        import faiss
        self.index = faiss.read_index(f"{path}.index")
        with open(f"{path}.meta", "rb") as f:
            data = pickle.load(f)
            self.id_map = data["id_map"]
            self.reverse_map = data["reverse_map"]
            self.metadata = data["metadata"]
            self._next_id = data["next_id"]

Milvus

Milvus是Zilliz开源的云原生向量数据库，专为大规模向量搜索设计。与FAISS不同，Milvus是一个完整的数据库系统，提供持久化存储、分布式架构、事务支持、元数据过滤和高可用能力。Milvus支持多种索引类型（IVF_FLAT、IVF_SQ8、IVF_PQ、HNSW、ANNOY），支持标量字段过滤，提供Python、Java、Go等多语言SDK。

Milvus的架构分为四个层次：接入层（处理客户端请求）、协调服务层（管理集群元数据和任务调度）、工作节点层（执行具体的数据操作）、存储层（使用MinIO或S3作为对象存储）。这种分层架构使得Milvus可以水平扩展，支持十亿级向量的存储和检索。

Milvus适合中大型Agent系统的生产部署，特别是需要多Agent共享向量数据、需要高可用和水平扩展的场景。其部署复杂度较高，需要Docker/Kubernetes环境，对小团队或原型验证阶段可能过重。

Chroma

Chroma是专为AI应用设计的轻量级向量数据库。它的设计理念是"开发者优先"，强调易用性和快速集成。Chroma内置了嵌入模型接口，支持自动嵌入计算，开发者只需传入文本即可自动完成向量化。Chroma支持Python和JavaScript SDK，数据可以存储在内存中或持久化到本地DuckDB。

Chroma的优势在于极低的使用门槛。几行代码就能完成向量存储和检索的完整流程，无需关心索引构建和参数调优。它适合Agent原型开发、小规模应用和开发测试环境。Chroma的局限在于性能和扩展性：它不支持分布式部署，单机性能在百万级向量时开始下降，不适合大规模生产环境。

Pinecone

Pinecone是全托管的云向量数据库服务。用户无需部署和维护任何基础设施，通过API即可使用向量存储和检索服务。Pinecone支持元数据过滤、向量更新、命名空间隔离等功能，自动处理索引构建、分片和副本。

Pinecone的优势在于零运维成本和弹性扩展能力。用户按使用量付费，无需预估容量。它适合快速上线的Agent应用和对运维能力要求低的团队。Pinecone的局限在于数据需要发送到第三方云服务，对数据隐私敏感的场景不适用；且长期大规模使用的成本可能高于自建方案。

三、索引算法深度解析

向量检索的核心挑战是高维空间中的最近邻搜索。暴力搜索需要计算查询向量与所有存储向量的距离，复杂度为O(n)，在百万级以上数据集上无法接受。近似最近邻（ANN）索引通过牺牲少量精度换取大幅性能提升，是大规模向量检索的必然选择。

HNSW索引

HNSW（Hierarchical Navigable Small World）是目前最主流的向量索引算法。它构建一个多层图结构：最底层包含所有向量节点，上层是下层的稀疏子集，顶层只有少数节点。检索时从顶层开始，在每层中通过贪心搜索找到局部最近邻，然后下降到下一层继续搜索，直到最底层。

HNSW的核心参数有两个：M是每个节点的最大邻居数（通常16-48），efConstruction是构建时的搜索宽度（通常200-500），efSearch是检索时的搜索宽度（通常16-100）。M越大索引质量越好但内存占用越高；efSearch越大检索精度越高但延迟越高。

HNSW的优势在于检索速度快（亚毫秒级）、召回率高（95%以上）、支持动态插入。它的局限是内存占用较大（需要存储图结构），且不支持高效删除（删除需要标记后批量重建）。

IVF索引

IVF（Inverted File）通过K-Means聚类将向量空间划分为nlist个桶。每个向量被分配到最近的聚类中心所在的桶。检索时，查询向量先找到最近的nprobe个聚类中心，然后只在这些桶中进行暴力搜索。

IVF的关键参数是nlist（聚类中心数）和nprobe（搜索时探测的桶数）。nlist通常设置为数据量的平方根，nprobe越大召回率越高但延迟越高。IVF的优势在于内存效率高（可以通过PQ压缩进一步降低），构建速度快。局限是召回率不如HNSW，且需要训练阶段（K-Means聚类）。

class VectorIndexBenchmark:
    """向量索引性能基准测试"""
    
    def __init__(self, dim: int = 768, n_vectors: int = 100000):
        self.dim = dim
        self.n_vectors = n_vectors
        self.data = self._generate_data()
        self.queries = self._generate_queries()
        self.ground_truth = self._compute_ground_truth()
    
    def _generate_data(self):
        """生成测试数据"""
        import numpy as np
        return np.random.randn(self.n_vectors, self.dim).astype(np.float32)
    
    def _generate_queries(self, n_queries: int = 100):
        """生成查询向量"""
        import numpy as np
        return np.random.randn(n_queries, self.dim).astype(np.float32)
    
    def _compute_ground_truth(self, top_k: int = 10):
        """计算精确搜索的ground truth"""
        import numpy as np
        ground_truth = []
        for query in self.queries:
            # 归一化后计算内积（等价于余弦相似度）
            data_norm = self.data / (np.linalg.norm(self.data, axis=1, keepdims=True) + 1e-8)
            query_norm = query / (np.linalg.norm(query) + 1e-8)
            scores = data_norm @ query_norm
            top_indices = np.argsort(-scores)[:top_k]
            ground_truth.append(set(top_indices.tolist()))
        return ground_truth
    
    def benchmark_hnsw(self, m_values: List[int] = [16, 32, 48],
                       ef_values: List[int] = [16, 32, 64, 128]):
        """测试HNSW不同参数的性能"""
        import faiss
        import time
        
        results = []
        for m in m_values:
            for ef in ef_values:
                # 构建索引
                index = faiss.IndexHNSWFlat(self.dim, m)
                index.hnsw.efConstruction = 200
                index.hnsw.efSearch = ef
                
                # 构建时间
                build_start = time.time()
                index.add(self.data)
                build_time = time.time() - build_start
                
                # 检索延迟和召回率
                latencies = []
                recall_hits = 0
                total_queries = len(self.queries)
                
                for i, query in enumerate(self.queries):
                    search_start = time.time()
                    scores, indices = index.search(
                        query.reshape(1, -1), 10
                    )
                    latencies.append(time.time() - search_start)
                    
                    # 计算召回率
                    retrieved = set(indices[0][indices[0] >= 0].tolist())
                    recall_hits += len(retrieved & self.ground_truth[i])
                
                avg_latency = sum(latencies) / len(latencies) * 1000  # ms
                recall = recall_hits / (total_queries * 10)
                
                results.append({
                    "index": "HNSW",
                    "M": m,
                    "efSearch": ef,
                    "build_time": build_time,
                    "avg_latency_ms": avg_latency,
                    "recall": recall
                })
        
        return results
    
    def benchmark_ivf(self, nlist_values: List[int] = [100, 1000, 10000],
                      nprobe_values: List[int] = [1, 10, 50, 100]):
        """测试IVF不同参数的性能"""
        import faiss
        import time
        
        results = []
        for nlist in nlist_values:
            if nlist > self.n_vectors:
                continue
            for nprobe in nprobe_values:
                if nprobe > nlist:
                    continue
                
                quantizer = faiss.IndexFlatIP(self.dim)
                index = faiss.IndexIVFFlat(quantizer, self.dim, nlist,
                                           faiss.METRIC_INNER_PRODUCT)
                index.nprobe = nprobe
                
                # 训练
                train_start = time.time()
                index.train(self.data)
                train_time = time.time() - train_start
                
                # 构建
                build_start = time.time()
                index.add(self.data)
                build_time = time.time() - build_start
                
                # 检索
                latencies = []
                recall_hits = 0
                
                for i, query in enumerate(self.queries):
                    search_start = time.time()
                    scores, indices = index.search(
                        query.reshape(1, -1), 10
                    )
                    latencies.append(time.time() - search_start)
                    
                    retrieved = set(indices[0][indices[0] >= 0].tolist())
                    recall_hits += len(retrieved & self.ground_truth[i])
                
                avg_latency = sum(latencies) / len(latencies) * 1000
                recall = recall_hits / (len(self.queries) * 10)
                
                results.append({
                    "index": "IVF",
                    "nlist": nlist,
                    "nprobe": nprobe,
                    "train_time": train_time,
                    "build_time": build_time,
                    "avg_latency_ms": avg_latency,
                    "recall": recall
                })
        
        return results

四、性能调优策略

向量数据库的性能调优是一个系统工程，涉及索引参数、硬件配置、数据布局和查询优化多个层面。

索引参数调优是最直接的优化手段。对于HNSW，efSearch是延迟与召回率的调节旋钮：生产环境建议从16开始，逐步增大直到召回率达到目标值（通常95%以上），记录此时的efSearch作为生产参数。对于IVF，nprobe起类似作用。参数选择应基于实际数据和工作负载的基准测试，而非默认值。

数据布局优化关注向量在存储中的排列方式。将同一类型或同一命名空间的向量存储在一起，可以提高缓存命中率。对于需要元数据过滤的场景，将过滤字段与向量存储在同一节点上，避免跨节点查询。如果向量维度很高（如1536维），可以考虑使用PQ（Product Quantization）压缩，将向量从1536维压缩到64字节，内存占用降低90%以上，代价是少量精度损失。

查询优化包括批量查询、异步查询和结果缓存。批量查询将多个查询合并为一次索引访问，减少索引加载开销。异步查询利用多线程或协程并行处理多个查询，提高吞吐量。结果缓存对重复或相似查询返回缓存结果，避免重复计算。

五、数据分片与分布式部署

当单机无法容纳全部向量数据时，需要进行数据分片。向量数据的分片策略与关系型数据库不同，不能简单地按ID范围分片，因为向量检索需要扫描所有分片的结果并合并。

常见的分片策略有三种。基于聚类的分片将语义相近的向量分配到同一分片，检索时可以只查询最相关的几个分片，减少跨分片查询。基于哈希的分片将向量按ID哈希均匀分布到各分片，保证负载均衡但每次查询需要扫描所有分片。基于元数据的分片按业务属性（如用户ID、时间范围）分片，支持按属性路由查询。

from typing import List, Dict, Optional
import numpy as np

class ShardedVectorStore:
    """分片向量存储"""
    
    def __init__(self, shard_strategy: str = "hash", 
                 n_shards: int = 4,
                 dim: int = 768):
        self.shard_strategy = shard_strategy
        self.n_shards = n_shards
        self.dim = dim
        self.shards: List[FAISSVectorStore] = [
            FAISSVectorStore(dim, "hnsw") for _ in range(n_shards)
        ]
        self.shard_routing: Dict[str, int] = {}  # item_id -> shard_idx
    
    def add(self, item_id: str, vector: np.ndarray, 
            metadata: Optional[dict] = None):
        """添加向量到分片"""
        shard_idx = self._get_shard(item_id, metadata)
        self.shards[shard_idx].add(item_id, vector, metadata)
        self.shard_routing[item_id] = shard_idx
    
    def batch_add(self, items: List[Tuple[str, np.ndarray, dict]]):
        """批量添加"""
        # 按分片分组
        shard_groups: Dict[int, List] = {i: [] for i in range(self.n_shards)}
        for item_id, vec, meta in items:
            shard_idx = self._get_shard(item_id, meta)
            shard_groups[shard_idx].append((item_id, vec, meta))
            self.shard_routing[item_id] = shard_idx
        
        # 并行添加到各分片
        for shard_idx, group in shard_groups.items():
            if group:
                self.shards[shard_idx].batch_add(group)
    
    def search(self, query_vec: np.ndarray, top_k: int = 10,
               filter_fn=None,
               search_shards: Optional[List[int]] = None) -> List[Tuple[str, float, dict]]:
        """跨分片检索"""
        # 确定搜索的分片
        if search_shards is None:
            shards_to_search = list(range(self.n_shards))
        else:
            shards_to_search = search_shards
        
        # 并行搜索各分片
        all_results = []
        per_shard_k = top_k * 2  # 每个分片多取一些用于合并
        
        for shard_idx in shards_to_search:
            results = self.shards[shard_idx].search(
                query_vec, per_shard_k, filter_fn
            )
            all_results.extend(results)
        
        # 全局排序并取top_k
        all_results.sort(key=lambda x: x[1], reverse=True)
        return all_results[:top_k]
    
    def _get_shard(self, item_id: str, metadata: Optional[dict]) -> int:
        """确定分片"""
        if self.shard_strategy == "hash":
            return hash(item_id) % self.n_shards
        elif self.shard_strategy == "metadata" and metadata:
            # 按元数据字段分片
            key = metadata.get("namespace", "default")
            return hash(key) % self.n_shards
        elif self.shard_strategy == "cluster":
            # 基于聚类分片（需要预计算聚类中心）
            return 0  # 简化实现
        else:
            return hash(item_id) % self.n_shards
    
    def get_shard_stats(self) -> List[Dict]:
        """获取各分片统计信息"""
        stats = []
        for i, shard in enumerate(self.shards):
            stats.append({
                "shard_id": i,
                "vector_count": shard._next_id,
                "memory_mb": shard._next_id * self.dim * 4 / (1024 * 1024)
            })
        return stats

分布式部署还需要考虑副本和容错。每个分片可以维护一个或多个副本，当主分片不可用时自动切换到副本。副本同步策略分为同步复制（写入时等待所有副本确认，一致性高但延迟大）和异步复制（写入主分片后立即返回，后台同步副本，延迟低但可能短暂不一致）。Agent记忆系统通常选择异步复制以优先保证写入延迟，而知识库系统可以选择同步复制以保证数据安全。

六、统一向量存储层设计

在Agent系统中，不同的子系统（记忆、RAG、用户画像）可能使用不同的向量数据库后端。设计统一的向量存储层抽象，使得上层逻辑与具体存储后端解耦，是工程实践中的重要架构决策。

from abc import ABC, abstractmethod
from typing import List, Tuple, Dict, Optional
import numpy as np
import time
import json

class VectorStoreBackend(ABC):
    """向量存储后端抽象接口"""
    
    @abstractmethod
    def add(self, item_id: str, vector: np.ndarray, 
            metadata: Optional[dict] = None) -> bool:
        """添加单个向量"""
        pass
    
    @abstractmethod
    def batch_add(self, items: List[Tuple[str, np.ndarray, dict]]) -> int:
        """批量添加向量"""
        pass
    
    @abstractmethod
    def search(self, query_vec: np.ndarray, top_k: int = 10,
               filter_dict: Optional[dict] = None) -> List[Tuple[str, float, dict]]:
        """检索向量"""
        pass
    
    @abstractmethod
    def delete(self, item_id: str) -> bool:
        """删除向量"""
        pass
    
    @abstractmethod
    def update(self, item_id: str, vector: Optional[np.ndarray] = None,
               metadata: Optional[dict] = None) -> bool:
        """更新向量或元数据"""
        pass
    
    @abstractmethod
    def count(self) -> int:
        """返回存储的向量总数"""
        pass
    
    @abstractmethod
    def save(self, path: str) -> bool:
        """持久化"""
        pass
    
    @abstractmethod
    def load(self, path: str) -> bool:
        """加载"""
        pass


class InMemoryVectorBackend(VectorStoreBackend):
    """内存向量存储后端"""
    
    def __init__(self, dim: int):
        self.dim = dim
        self._vectors: Dict[str, np.ndarray] = {}
        self._metadata: Dict[str, dict] = {}
        self._timestamps: Dict[str, float] = {}
    
    def add(self, item_id: str, vector: np.ndarray,
            metadata: Optional[dict] = None) -> bool:
        self._vectors[item_id] = vector.astype(np.float32)
        if metadata:
            self._metadata[item_id] = metadata
        self._timestamps[item_id] = time.time()
        return True
    
    def batch_add(self, items: List[Tuple[str, np.ndarray, dict]]) -> int:
        count = 0
        for item_id, vec, meta in items:
            self.add(item_id, vec, meta)
            count += 1
        return count
    
    def search(self, query_vec: np.ndarray, top_k: int = 10,
               filter_dict: Optional[dict] = None) -> List[Tuple[str, float, dict]]:
        if not self._vectors:
            return []
        
        query = query_vec.astype(np.float32)
        query_norm = query / (np.linalg.norm(query) + 1e-8)
        
        scores = []
        for item_id, vec in self._vectors.items():
            # 元数据过滤
            if filter_dict:
                meta = self._metadata.get(item_id, {})
                if not all(meta.get(k) == v for k, v in filter_dict.items()):
                    continue
            
            vec_norm = vec / (np.linalg.norm(vec) + 1e-8)
            sim = float(np.dot(query_norm, vec_norm))
            scores.append((item_id, sim, self._metadata.get(item_id, {})))
        
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:top_k]
    
    def delete(self, item_id: str) -> bool:
        self._vectors.pop(item_id, None)
        self._metadata.pop(item_id, None)
        self._timestamps.pop(item_id, None)
        return True
    
    def update(self, item_id: str, vector: Optional[np.ndarray] = None,
               metadata: Optional[dict] = None) -> bool:
        if vector is not None:
            self._vectors[item_id] = vector.astype(np.float32)
        if metadata is not None:
            existing = self._metadata.get(item_id, {})
            existing.update(metadata)
            self._metadata[item_id] = existing
        self._timestamps[item_id] = time.time()
        return True
    
    def count(self) -> int:
        return len(self._vectors)
    
    def save(self, path: str) -> bool:
        data = {
            "dim": self.dim,
            "vectors": {k: v.tolist() for k, v in self._vectors.items()},
            "metadata": self._metadata,
            "timestamps": self._timestamps
        }
        with open(path, 'w') as f:
            json.dump(data, f)
        return True
    
    def load(self, path: str) -> bool:
        with open(path, 'r') as f:
            data = json.load(f)
        self.dim = data["dim"]
        self._vectors = {k: np.array(v, dtype=np.float32) 
                        for k, v in data["vectors"].items()}
        self._metadata = data["metadata"]
        self._timestamps = data["timestamps"]
        return True


class VectorStoreManager:
    """向量存储管理器 - 统一管理多个向量空间"""
    
    def __init__(self, default_dim: int = 768):
        self.default_dim = default_dim
        self._stores: Dict[str, VectorStoreBackend] = {}
        self._configs: Dict[str, dict] = {}
    
    def create_collection(self, name: str, dim: Optional[int] = None,
                          backend: str = "memory",
                          config: Optional[dict] = None) -> VectorStoreBackend:
        """创建向量集合"""
        dim = dim or self.default_dim
        
        if backend == "memory":
            store = InMemoryVectorBackend(dim)
        elif backend == "faiss":
            store = FAISSVectorStore(dim, 
                                    index_type=config.get("index_type", "hnsw"))
        else:
            raise ValueError(f"不支持的后端类型: {backend}")
        
        self._stores[name] = store
        self._configs[name] = config or {}
        return store
    
    def get_collection(self, name: str) -> Optional[VectorStoreBackend]:
        """获取向量集合"""
        return self._stores.get(name)
    
    def delete_collection(self, name: str) -> bool:
        """删除向量集合"""
        if name in self._stores:
            del self._stores[name]
            del self._configs[name]
            return True
        return False
    
    def list_collections(self) -> List[Dict]:
        """列出所有集合"""
        return [
            {
                "name": name,
                "count": store.count(),
                "config": self._configs[name]
            }
            for name, store in self._stores.items()
        ]
    
    def save_all(self, base_path: str):
        """保存所有集合"""
        import os
        os.makedirs(base_path, exist_ok=True)
        for name, store in self._stores.items():
            store.save(os.path.join(base_path, f"{name}.json"))
    
    def load_all(self, base_path: str):
        """加载所有集合"""
        import os
        if not os.path.exists(base_path):
            return
        for filename in os.listdir(base_path):
            if filename.endswith(".json"):
                name = filename[:-5]
                store = InMemoryVectorBackend(self.default_dim)
                store.load(os.path.join(base_path, filename))
                self._stores[name] = store


# 使用示例
if __name__ == "__main__":
    manager = VectorStoreManager(default_dim=256)
    
    # 创建Agent记忆向量集合
    memory_store = manager.create_collection(
        "agent_memory", dim=256, backend="memory"
    )
    
    # 创建RAG知识库向量集合
    knowledge_store = manager.create_collection(
        "knowledge_base", dim=256, backend="memory"
    )
    
    # 添加记忆向量
    import numpy as np
    memory_store.add("mem_1", np.random.randn(256), 
                     {"type": "fact", "timestamp": time.time()})
    memory_store.add("mem_2", np.random.randn(256),
                     {"type": "preference", "timestamp": time.time()})
    
    # 检索
    query = np.random.randn(256)
    results = memory_store.search(query, top_k=5)
    print(f"记忆检索结果: {len(results)} 条")
    for item_id, score, meta in results:
        print(f"  {item_id}: score={score:.4f}, type={meta.get('type')}")
    
    # 集合统计
    for col in manager.list_collections():
        print(f"集合: {col['name']}, 向量数: {col['count']}")

七、向量数据库选型决策框架

基于以上分析，可以构建一个向量数据库选型决策框架。

数据规模是首要考量因素。万级以下向量选择Chroma或内存方案即可；十万到百万级选择FAISS单机部署；百万到千万级选择Milvus单机或Pinecone；亿级以上需要Milvus分布式集群。

部署环境约束影响选型范围。如果要求完全离线部署，排除Pinecone等云服务；如果缺乏Kubernetes运维能力，排除Milvus分布式部署；如果开发资源有限，优先选择Chroma或FAISS等轻量方案。

性能要求决定索引选择。延迟要求亚毫秒级选择HNSW；可以接受10毫秒延迟且需要高压缩比选择IVF_PQ；需要精确结果选择Flat。召回率要求99%以上可能需要HNSW配合较大的efSearch，或使用Flat精确搜索。

数据特性影响分片和索引策略。如果向量维度很高（如2048维），PQ压缩的收益更大；如果需要频繁删除，HNSW的删除成本较高，IVF更合适；如果需要元数据过滤，选择支持原生过滤的Milvus或Chroma。

八、总结

向量数据库是Agent系统的核心基础设施，选型需要综合考虑数据规模、性能要求、部署环境和团队能力。FAISS适合作为高性能嵌入式检索引擎，Milvus适合大规模生产部署，Chroma适合快速原型开发，Pinecone适合零运维的云原生场景。

索引算法的选择是性能调优的关键。HNSW在大多数场景下提供最佳的延迟-召回率平衡，IVF适合内存受限场景，PQ压缩适合超大规模数据。参数调优应基于实际数据的基准测试，而非默认值。

统一向量存储层抽象使得Agent系统可以在不同后端之间灵活切换，保护上层逻辑投资。通过分片、副本和缓存策略，向量存储层可以水平扩展到亿级向量，满足大型Agent系统的需求。在实际工程中，建议从简单方案起步，随着数据规模和性能需求的增长逐步演进到更复杂的架构。