语义核心 Python 向量存储迁移指南

概述

本指南介绍语义内核版本 1.34 中引入的主要矢量存储更新，这表示对矢量存储实现进行重大改革，以与 .NET SDK 保持一致，并提供更统一、更直观的 API。 这些更改合并了连接器体系结构下 semantic_kernel.data.vector 的所有内容，并改进了连接器体系结构。

关键改进摘要

• 统一字段模型：单个 VectorStoreField 类替换多个字段类型
• 整合嵌入：在矢量场规范中直接生成嵌入
• 简化的搜索：直接在集合上轻松创建搜索函数
• 合并结构：semantic_kernel.data.vector 和 semantic_kernel.connectors 下面的所有内容
• 增强型文本搜索：使用简化的连接器改进了文本搜索功能
• 弃用：旧memory_stores版本已弃用，推荐使用新的矢量存储体系结构

1. 集成嵌入和矢量存储模型/字段更新

定义向量存储模型的方式有很多变化，最大的是我们现在支持直接在向量存储字段定义中集成嵌入。 这意味着，当指定字段作为向量时，该字段的内容将自动嵌入使用指定的嵌入生成器，例如 OpenAI 的文本嵌入模型。 这简化了创建和管理矢量字段的过程。

定义该字段时，需要确保三件事，尤其是在使用 Pydantic 模型时：

1. 键入：对于嵌入生成器的输入，字段可能具有三种类型，即 list[float]、str 或其他类型，以及表示取消设置字段的时间的 None。
2. 默认值：字段必须具有默认值 None 或其他值，这样在从 get 或 search 获取记录时就不会出错，当前 include_vectors=False 是默认值。

此处有两个问题，第一个是当使用vectorstoremodel修饰一个类时，字段的第一个类型批注会用于填充VectorStoreField类的type参数，因此需要确保第一个类型批注是用于创建向量存储集合的正确类型，通常是list[float]。 默认情况下，getsearch方法的结果中不包括include_vectors，因此该字段需要一个默认值，键入需要与之相匹配，因此经常允许None，并将默认值设置为None。 创建字段时，需要嵌入的值通常为字符串，因此 str 还需要包含这些值。 此更改的原因是允许在嵌入内容和实际存储在数据字段中的内容方面具有更大的灵活性，这是一种常见的设置：

from semantic_kernel.data.vector import VectorStoreField, vectorstoremodel
from typing import Annotated
from dataclasses import dataclass

@vectorstoremodel
@dataclass
class MyRecord:
    content: Annotated[str, VectorStoreField('data', is_indexed=True, is_full_text_indexed=True)]
    title: Annotated[str, VectorStoreField('data', is_indexed=True, is_full_text_indexed=True)]
    id: Annotated[str, VectorStoreField('key')]
    vector: Annotated[list[float] | str | None, VectorStoreField(
        'vector', 
        dimensions=1536, 
        distance_function="cosine",
        embedding_generator=OpenAITextEmbedding(ai_model_id="text-embedding-3-small"),
    )] = None

    def __post_init__(self):
        if self.vector is None:
            self.vector = f"Title: {self.title}, Content: {self.content}"

请注意 post_init 方法，这会创建一个嵌入的值，该值比单个字段更复杂。 这三种类型也存在。

之前：独立字段类

from semantic_kernel.data import (
    VectorStoreRecordKeyField,
    VectorStoreRecordDataField, 
    VectorStoreRecordVectorField
)

# Old approach with separate field classes
fields = [
    VectorStoreRecordKeyField(name="id"),
    VectorStoreRecordDataField(name="text", is_filterable=True, is_full_text_searchable=True),
    VectorStoreRecordVectorField(name="vector", dimensions=1536, distance_function="cosine")
]

之后：含集成嵌入的统一 VectorStoreField

from semantic_kernel.data.vector import VectorStoreField
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding

# New unified approach with integrated embeddings
embedding_service = OpenAITextEmbedding(
    ai_model_id="text-embedding-3-small"
)

fields = [
    VectorStoreField(
        "key",
        name="id",
    ),
    VectorStoreField(
        "data",
        name="text",
        is_indexed=True,  # Previously is_filterable
        is_full_text_indexed=True  # Previously is_full_text_searchable
    ),
    VectorStoreField(
        "vector",
        name="vector",
        dimensions=1536,
        distance_function="cosine",
        embedding_generator=embedding_service  # Integrated embedding generation
    )
]

字段定义中的主要更改

1. 单个字段类： VectorStoreField 替换所有以前的字段类型
2. 字段类型规范：使用 field_type: Literal["key", "data", "vector"] 参数，可以是位置参数，因此 VectorStoreField("key") 有效。
3. 增强的属性：
   • storage_name 已添加，设置后可用作矢量存储中的字段名称，否则使用 name 参数。
   • dimensions 现在是矢量字段的必需参数。
   • distance_function和index_kind都是可选的，如果未指定，这两个字段将分别被设置为DistanceFunction.DEFAULT和IndexKind.DEFAULT，此更改仅适用于矢量字段。每个向量存储的实现都有相应的逻辑来选择该存储的默认值。
4. 属性重命名：
   • property_type → type_ 作为构造函数中的属性和 type
   • is_filterable → is_indexed
   • is_full_text_searchable → is_full_text_indexed
5. 集成嵌入：直接添加到 embedding_generator 向量字段，或者可以设置 embedding_generator 矢量存储集合本身，该集合本身将用于该存储中的所有向量字段，此值优先于集合级别嵌入生成器。

2. 存储和集合上的新方法

增强的商店界面

from semantic_kernel.connectors.in_memory import InMemoryStore

# Before: Limited collection methods
collection = InMemoryStore.get_collection("my_collection", record_type=MyRecord)

# After: Slimmer collection interface with new methods
collection = InMemoryStore.get_collection(MyRecord)
# if the record type has the `vectorstoremodel` decorator it can contain both the collection_name and the definition for the collection.

# New methods for collection management
await store.collection_exists("my_collection")
await store.ensure_collection_deleted("my_collection")
# both of these methods, create a simple model to streamline doing collection management tasks.
# they both call the underlying `VectorStoreCollection` methods, see below.

增强的集合接口

from semantic_kernel.connectors.in_memory import InMemoryCollection

collection = InMemoryCollection(
    record_type=MyRecord,
    embedding_generator=OpenAITextEmbedding(ai_model_id="text-embedding-3-small")  # Optional, if there is no embedding generator set on the record type
)
# If both the collection and the record type have an embedding generator set, the record type's embedding generator will be used for the collection. If neither is set, it is assumed the vector store itself can create embeddings, or that vectors are included in the records already, if that is not the case, it will likely raise.

# Enhanced collection operations
await collection.collection_exists()
await collection.ensure_collection_exists()
await collection.ensure_collection_deleted()

# CRUD methods
# Removed batch operations, all CRUD operations can now take both a single record or a list of records
records = [
    MyRecord(id="1", text="First record"),
    MyRecord(id="2", text="Second record")
]
ids = ["1", "2"]
# this method adds vectors automatically
await collection.upsert(records)

# You can do get with one or more ids, and it will return a list of records
await collection.get(ids)  # Returns a list of records
# you can also do a get without ids, with top, skip and order_by parameters
await collection.get(top=10, skip=0, order_by='id')
# the order_by parameter can be a string or a dict, with the key being the field name and the value being True for ascending or False for descending order.
# At this time, not all vector stores support this method.

# Delete also allows for single or multiple ids
await collection.delete(ids)

query = "search term"
# New search methods, these use the built-in embedding generator to take the value and create a vector
results = await collection.search(query, top=10)
results = await collection.hybrid_search(query, top=10)

# You can also supply a vector directly
query_vector = [0.1, 0.2, 0.3]  # Example vector
results = await collection.search(vector=query_vector, top=10)
results = await collection.hybrid_search(query, vector=query_vector, top=10)

3. 用于搜索的增强筛选器

新的矢量存储实现从基于字符串的 FilterClause 对象移动到更强大且类型安全的 lambda 表达式或可调用的筛选器。

之前：FilterClause 对象

from semantic_kernel.data.text_search import SearchFilter, EqualTo, AnyTagsEqualTo
from semantic_kernel.data.vector_search import VectorSearchFilter

# Creating filters using FilterClause objects
text_filter = SearchFilter()
text_filter.equal_to("category", "AI")
text_filter.equal_to("status", "active")

# Vector search filters
vector_filter = VectorSearchFilter()
vector_filter.equal_to("category", "AI")
vector_filter.any_tag_equal_to("tags", "important")

# Using in search
results = await collection.search(
    "query text",
    options=VectorSearchOptions(filter=vector_filter)
)

之后：Lambda 表达式筛选器

# When defining the collection with the generic type hints, most IDE's will be able to infer the type of the record, so you can use the record type directly in the lambda expressions.
collection = InMemoryCollection[str, MyRecord](MyRecord)

# Using lambda expressions for more powerful and type-safe filtering
# The code snippets below work on a data model with more fields then defined earlier.

# Direct lambda expressions
results = await collection.search(
    "query text", 
    filter=lambda record: record.category == "AI" and record.status == "active"
)

# Complex filtering with multiple conditions
results = await collection.search(
    "query text",
    filter=lambda record: (
        record.category == "AI" and 
        record.score > 0.8 and
        "important" in record.tags
    )
)

# Combining conditions with boolean operators
results = await collection.search(
    "query text",
    filter=lambda record: (
        record.category == "AI" or record.category == "ML"
    ) and record.published_date >= datetime(2024, 1, 1)
)

# Range filtering (now possible with lambda expressions)
results = await collection.search(
    "query text",
    filter=lambda record: 0.5 <= record.confidence_score <= 0.9
)

筛选器的迁移提示

1. 简单相等： filter.equal_to("field", "value") 变为 lambda r: r.field == "value"
2. 多个条件：链式使用 and/or 运算符，而不是调用多个筛选器
3. 标记/数组包含： filter.any_tag_equal_to("tags", "value") 变为 lambda r: "value" in r.tags
4. 增强功能：支持范围查询、复杂的布尔逻辑和自定义谓词

4. 改进了创建搜索函数的简便性

之前：使用 VectorStoreTextSearch 创建搜索函数

from semantic_kernel.connectors.in_memory import InMemoryCollection
from semantic_kernel.data import VectorStoreTextSearch

collection = InMemoryCollection(collection_name='collection', record_type=MyRecord)
search = VectorStoreTextSearch.from_vectorized_search(vectorized_search=collection, embedding_generator=OpenAITextEmbedding(ai_model_id="text-embedding-3-small"))

search_function = search.create_search(
    function_name='search',
    ...
)

之后：直接搜索函数创建

collection = InMemoryCollection(MyRecord)
# Create search function directly on collection
search_function = collection.create_search_function(
    function_name="search",
    search_type="vector",  # or "keyword_hybrid"
    top=10,
    vector_property_name="vector",  # Name of the vector field
)

# Add to kernel directly
kernel.add_function(plugin_name="memory", function=search_function)

5. 连接器重命名和导入更改

导入路径合并

# Before: Scattered imports
from semantic_kernel.connectors.memory.azure_cognitive_search import AzureCognitiveSearchMemoryStore
from semantic_kernel.connectors.memory.chroma import ChromaMemoryStore
from semantic_kernel.connectors.memory.pinecone import PineconeMemoryStore
from semantic_kernel.connectors.memory.qdrant import QdrantMemoryStore

# After: Consolidated under connectors
from semantic_kernel.connectors.azure_ai_search import AzureAISearchStore
from semantic_kernel.connectors.chroma import ChromaVectorStore
from semantic_kernel.connectors.pinecone import PineconeVectorStore
from semantic_kernel.connectors.qdrant import QdrantVectorStore

# Alternative after: Consolidated with lazy loading:
from semantic_kernel.connectors.memory import (
    AzureAISearchStore,
    ChromaVectorStore,
    PineconeVectorStore,
    QdrantVectorStore,
    WeaviateVectorStore,
    RedisVectorStore
)

连接器类重命名

旧名称	新名称
AzureCosmosDBforMongoDB*	CosmosMongo*
AzureCosmosDBForNoSQL*	CosmosNoSql*

6. 文本搜索改进并删除了必应连接器

必应连接器已被移除，同时文本搜索界面得到了增强。

必应文本搜索连接器已删除。 迁移到备用搜索提供程序：

# Before: Bing Connector (REMOVED)
from semantic_kernel.connectors.search.bing import BingConnector

bing_search = BingConnector(api_key="your-bing-key")

# After: Use Brave Search or other providers
from semantic_kernel.connectors.brave import BraveSearch
# or
from semantic_kernel.connectors.search import BraveSearch

brave_search = BraveSearch()

# Create text search function
text_search_function = brave_search.create_search_function(
    function_name="web_search",
    query_parameter_name="query",
    description="Search the web for information"
)

kernel.add_function(plugin_name="search", function=text_search_function)

改进了搜索方法

之前：具有不同返回类型的三种不同的搜索方法

from semantic_kernel.connectors.brave import BraveSearch
brave_search = BraveSearch()
# Before: Separate search methods
search_results: KernelSearchResult[str] = await brave_search.search(
    query="semantic kernel python",
    top=5,
)

search_results: KernelSearchResult[TextSearchResult] = await brave_search.get_text_search_results(
    query="semantic kernel python",
    top=5,
)

search_results: KernelSearchResult[BraveWebPage] = await brave_search.get_search_results(
    query="semantic kernel python",
    top=5,
)

之后：具有输出类型参数的统一搜索方法

from semantic_kernel.data.text_search import SearchOptions
# Enhanced search results with metadata
search_results: KernelSearchResult[str] = await brave_search.search(
    query="semantic kernel python",
    output_type=str, # can also be TextSearchResult or anything else for search engine specific results, default is `str`
    top=5,
    filter=lambda result: result.country == "NL",  # Example filter
)

async for result in search_results.results:
    assert isinstance(result, str)  # or TextSearchResult if using that type
    print(f"Result: {result}")
    print(f"Metadata: {search_results.metadata}")

7. 停用旧内存存储

所有基于 MemoryStoreBase 的旧内存存储已被移入 semantic_kernel.connectors.memory_stores ，并且现在被标记为弃用。 其中大多数都具有基于 VectorStore 和 VectorStoreCollection 的等效新实现，可在其中 semantic_kernel.connectors.memory找到。

这些连接器将完全删除：

• AstraDB
• Milvus
• Usearch

如果仍然需要上述任一，请确保从已弃用的模块和semantic_kernel.memory文件夹中获取代码，或者基于新的VectorStoreCollection 类实现您自己的向量存储集合。

如果根据 GitHub 上的反馈存在大量需求，我们将考虑恢复它们。 但目前，它们尚未维护，并将在未来被删除。

从 SemanticTextMemory 迁移

# Before: SemanticTextMemory (DEPRECATED)
from semantic_kernel.memory import SemanticTextMemory
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbeddingGenerationService

embedding_service = OpenAITextEmbeddingGenerationService(ai_model_id="text-embedding-3-small")
memory = SemanticTextMemory(storage=vector_store, embeddings_generator=embedding_service)

# Store memory
await memory.save_information(collection="docs", text="Important information", id="doc1")

# Search memory  
results = await memory.search(collection="docs", query="important", limit=5)

# After: Direct Vector Store Usage
from semantic_kernel.data.vector import VectorStoreField, vectorstoremodel
from semantic_kernel.connectors.in_memory import InMemoryCollection

# Define data model
@vectorstoremodel
@dataclass
class MemoryRecord:
    id: Annotated[str, VectorStoreField('key')]
    text: Annotated[str, VectorStoreField('data', is_full_text_indexed=True)]
    embedding: Annotated[list[float] | str | None, VectorStoreField('vector', dimensions=1536, distance_function="cosine", embedding_generator=OpenAITextEmbedding(ai_model_id="text-embedding-3-small"))] = None

# Create vector store with integrated embeddings
collection = InMemoryCollection(
    record_type=MemoryRecord,
    embedding_generator=OpenAITextEmbedding(ai_model_id="text-embedding-3-small")  # Optional, if not set on the record type
)

# Store with automatic embedding generation
record = MemoryRecord(id="doc1", text="Important information", embedding='Important information')
await collection.upsert(record)

# Search with built-in function
search_function = collection.create_search_function(
    function_name="search_docs",
    search_type="vector"
)

内存插件迁移

如果想要有一个还可以保存信息的插件，则可以轻松创建如下所示的插件：

# Before: TextMemoryPlugin (DEPRECATED)
from semantic_kernel.core_plugins import TextMemoryPlugin

memory_plugin = TextMemoryPlugin(memory)
kernel.add_plugin(memory_plugin, "memory")

# After: Custom plugin using vector store search functions
from semantic_kernel.functions import kernel_function

class VectorMemoryPlugin:
    def __init__(self, collection: VectorStoreCollection):
        self.collection = collection
    
    @kernel_function(name="save")
    async def save_memory(self, text: str, key: str) -> str:
        record = MemoryRecord(id=key, text=text, embedding=text)
        await self.collection.upsert(record)
        return f"Saved to {self.collection.collection_name}"
    
    @kernel_function(name="search") 
    async def search_memory(self, query: str, limit: int = 5) -> str:
        results = await self.collection.search(
            query, top=limit, vector_property_name="embedding"
        )        
        return "\n".join([r.record.text async for r in results.results])

# Register the new plugin
memory_plugin = VectorMemoryPlugin(collection)
kernel.add_plugin(memory_plugin, "memory")

矢量搜索的迁移清单

步骤 1：更新导入

• [ ] 将内存存储导入替换为矢量存储等效项
• [ ] 更新字段导入，以使用 VectorStoreField
• [ ] 删除必应连接器导入

步骤 2：更新字段定义

• [ ] 转换为统一 VectorStoreField 类
• [ ] 更新属性名称 （is_filterable → is_indexed）
• [ ] 向向量字段添加集成的嵌入生成器

步骤 3：更新集合使用情况

• [ ] 将内存操作替换为矢量存储方法
• [ ] 在适用情况下使用新的批处理操作
• [ ] 实现新的搜索函数创建

步骤 4：更新搜索实现

• [ ] 将手动搜索函数替换为 create_search_function
• [ ] 更新文本搜索以使用新提供程序
• [ ] 实现有用的混合搜索
• [ ] 从 FilterClause 表达式迁移到 lambda 用于筛选的表达式

步骤 5：删除已弃用的代码

• [ ] 删除 SemanticTextMemory 的使用
• [ ] 删除 TextMemoryPlugin 依赖项

性能和功能优势

性能改进

• 批处理作：新的批处理插入/删除方法可提高吞吐量
• 集成嵌入：消除单独的嵌入生成步骤
• 优化搜索：针对每种商店类型优化内置搜索函数

功能增强

• 混合搜索：合并矢量和文本搜索以获取更好的结果
• 高级筛选：增强的筛选器表达式和索引

开发人员体验

• 简化的 API：减少学习的类和方法
• 一致接口：跨所有矢量存储的统一方法
• 更好的文档：清除示例和迁移路径
• 具备前瞻性：与 .NET SDK 适配以实现一致的跨平台开发

结论

上面讨论的矢量存储更新表示语义内核 Python SDK 中的显著改进。 新的统一体系结构提供更好的性能、增强的功能和更直观的开发人员体验。 虽然迁移需要更新导入和重构现有代码，但可维护性和功能的优势使此升级非常推荐。

有关迁移的其他帮助，请参阅目录中的更新示例 samples/concepts/memory/ 和全面的 API 文档。