tessl/pypi-semantic-kernel
Semantic Kernel Python SDK - comprehensive AI development framework for building AI agents and multi-agent systems
—
Pending
memory-stores.mddocs/
Memory and Vector Stores
Vector database integrations for semantic memory, embeddings storage, and retrieval-augmented generation. Supports popular vector stores including Chroma, Pinecone, Qdrant, Redis, Azure Cognitive Search, and many others.
Capabilities
Memory Store Base Interface
Common interface for all vector store implementations.
class MemoryStore:
"""
Base interface for vector memory stores.
"""
async def create_collection(
self,
collection_name: str,
**kwargs
) -> None:
"""
Create a new collection in the memory store.
Parameters:
- collection_name: Name of the collection to create
- **kwargs: Store-specific configuration options
"""
async def get_collections(self) -> list[str]:
"""
Get all collection names in the memory store.
Returns:
List of collection names
"""
async def delete_collection(self, collection_name: str) -> None:
"""
Delete a collection from the memory store.
Parameters:
- collection_name: Name of the collection to delete
"""
async def does_collection_exist(self, collection_name: str) -> bool:
"""
Check if a collection exists.
Parameters:
- collection_name: Name of the collection to check
Returns:
True if collection exists, False otherwise
"""
async def upsert(
self,
collection_name: str,
record: MemoryRecord,
**kwargs
) -> str:
"""
Insert or update a memory record.
Parameters:
- collection_name: Target collection name
- record: Memory record to upsert
- **kwargs: Store-specific options
Returns:
ID of the upserted record
"""
async def upsert_batch(
self,
collection_name: str,
records: list[MemoryRecord],
**kwargs
) -> list[str]:
"""
Insert or update multiple memory records.
Parameters:
- collection_name: Target collection name
- records: List of memory records to upsert
- **kwargs: Store-specific options
Returns:
List of IDs for the upserted records
"""
async def get(
self,
collection_name: str,
key: str,
with_embedding: bool = False,
**kwargs
) -> MemoryRecord | None:
"""
Get a memory record by key.
Parameters:
- collection_name: Collection to search in
- key: Record key/ID
- with_embedding: Whether to include embedding vector
- **kwargs: Store-specific options
Returns:
MemoryRecord if found, None otherwise
"""
async def get_batch(
self,
collection_name: str,
keys: list[str],
with_embeddings: bool = False,
**kwargs
) -> list[MemoryRecord]:
"""
Get multiple memory records by keys.
Parameters:
- collection_name: Collection to search in
- keys: List of record keys/IDs
- with_embeddings: Whether to include embedding vectors
- **kwargs: Store-specific options
Returns:
List of found MemoryRecord objects
"""
async def remove(self, collection_name: str, key: str, **kwargs) -> None:
"""
Remove a memory record by key.
Parameters:
- collection_name: Collection containing the record
- key: Record key/ID to remove
- **kwargs: Store-specific options
"""
async def remove_batch(
self,
collection_name: str,
keys: list[str],
**kwargs
) -> None:
"""
Remove multiple memory records by keys.
Parameters:
- collection_name: Collection containing the records
- keys: List of record keys/IDs to remove
- **kwargs: Store-specific options
"""
async def get_nearest_matches(
self,
collection_name: str,
embedding: list[float],
limit: int,
min_relevance_score: float = 0.0,
with_embeddings: bool = False,
**kwargs
) -> list[tuple[MemoryRecord, float]]:
"""
Find the nearest matching records to a query embedding.
Parameters:
- collection_name: Collection to search in
- embedding: Query embedding vector
- limit: Maximum number of results to return
- min_relevance_score: Minimum relevance score threshold
- with_embeddings: Whether to include embedding vectors in results
- **kwargs: Store-specific options
Returns:
List of tuples containing (MemoryRecord, relevance_score)
"""
async def get_nearest_match(
self,
collection_name: str,
embedding: list[float],
min_relevance_score: float = 0.0,
with_embedding: bool = False,
**kwargs
) -> tuple[MemoryRecord, float] | None:
"""
Find the single nearest matching record to a query embedding.
Parameters:
- collection_name: Collection to search in
- embedding: Query embedding vector
- min_relevance_score: Minimum relevance score threshold
- with_embedding: Whether to include embedding vector in result
- **kwargs: Store-specific options
Returns:
Tuple of (MemoryRecord, relevance_score) if found, None otherwise
"""
class MemoryRecord:
"""
Represents a memory record in a vector store.
"""
def __init__(
self,
id: str,
text: str,
is_reference: bool = False,
external_source_name: str | None = None,
description: str | None = None,
additional_metadata: str | None = None,
embedding: list[float] | None = None,
key: str | None = None,
timestamp: datetime | None = None
):
"""
Initialize a memory record.
Parameters:
- id: Unique identifier for the record
- text: Text content of the memory
- is_reference: Whether this is a reference to external content
- external_source_name: Name of external source if reference
- description: Description of the memory content
- additional_metadata: Additional metadata as string
- embedding: Vector embedding for the text
- key: Alternative key for the record
- timestamp: Timestamp when record was created
"""
@property
def id(self) -> str:
"""Get the record ID."""
@property
def text(self) -> str:
"""Get the text content."""
@property
def embedding(self) -> list[float] | None:
"""Get the embedding vector."""
@property
def metadata(self) -> dict[str, Any]:
"""Get all metadata as a dictionary."""Vector Store Connectors
Specific implementations for popular vector databases.
# Chroma Vector Store
from semantic_kernel.connectors.memory_stores.chroma import ChromaMemoryStore
class ChromaMemoryStore(MemoryStore):
"""Chroma vector database connector."""
def __init__(self, persist_directory: str | None = None):
"""
Initialize Chroma memory store.
Parameters:
- persist_directory: Directory to persist Chroma data
"""
# Pinecone Vector Store
from semantic_kernel.connectors.memory_stores.pinecone import PineconeMemoryStore
class PineconeMemoryStore(MemoryStore):
"""Pinecone vector database connector."""
def __init__(self, api_key: str, environment: str):
"""
Initialize Pinecone memory store.
Parameters:
- api_key: Pinecone API key
- environment: Pinecone environment
"""
# Qdrant Vector Store
from semantic_kernel.connectors.memory_stores.qdrant import QdrantMemoryStore
class QdrantMemoryStore(MemoryStore):
"""Qdrant vector database connector."""
def __init__(
self,
host: str = "localhost",
port: int = 6333,
api_key: str | None = None
):
"""
Initialize Qdrant memory store.
Parameters:
- host: Qdrant server host
- port: Qdrant server port
- api_key: API key for authentication
"""
# Redis Vector Store
from semantic_kernel.connectors.memory_stores.redis import RedisMemoryStore
class RedisMemoryStore(MemoryStore):
"""Redis vector database connector."""
def __init__(
self,
connection_string: str,
vector_size: int = 1536
):
"""
Initialize Redis memory store.
Parameters:
- connection_string: Redis connection string
- vector_size: Dimension of embedding vectors
"""
# Azure Cognitive Search
from semantic_kernel.connectors.memory_stores.azure_cognitive_search import AzureCognitiveSearchMemoryStore
class AzureCognitiveSearchMemoryStore(MemoryStore):
"""Azure Cognitive Search vector store connector."""
def __init__(
self,
search_endpoint: str,
admin_key: str,
vector_size: int = 1536
):
"""
Initialize Azure Cognitive Search memory store.
Parameters:
- search_endpoint: Azure Search service endpoint
- admin_key: Admin API key
- vector_size: Dimension of embedding vectors
"""
# Weaviate Vector Store
from semantic_kernel.connectors.memory_stores.weaviate import WeaviateMemoryStore
class WeaviateMemoryStore(MemoryStore):
"""Weaviate vector database connector."""
def __init__(
self,
url: str = "http://localhost:8080",
api_key: str | None = None,
use_embed: bool = True
):
"""
Initialize Weaviate memory store.
Parameters:
- url: Weaviate instance URL
- api_key: API key for authentication
- use_embed: Whether to use Weaviate's embedding service
"""
# Additional Vector Stores Available:
# - PostgreSQL with pgvector: PostgresMemoryStore
# - MongoDB Atlas: MongoDBAtlasMemoryStore
# - Milvus: MilvusMemoryStore
# - AstraDB: AstraDBMemoryStore
# - Azure Cosmos DB: AzureCosmosDBMemoryStore, AzureCosmosDBNoSQLMemoryStore
# - USearch: USearchMemoryStoreMemory Integration with Kernel
Helper classes for integrating memory with kernel operations.
class SemanticTextMemory:
"""
High-level interface for semantic text memory operations.
"""
def __init__(
self,
storage: MemoryStore,
embeddings_generator: EmbeddingGeneratorBase
):
"""
Initialize semantic text memory.
Parameters:
- storage: Vector store for memory persistence
- embeddings_generator: Service for generating embeddings
"""
async def save_information(
self,
collection: str,
text: str,
id: str,
description: str | None = None,
additional_metadata: str | None = None
) -> None:
"""
Save information to memory with automatic embedding generation.
Parameters:
- collection: Collection to save to
- text: Text content to save
- id: Unique identifier for the information
- description: Optional description
- additional_metadata: Optional additional metadata
"""
async def save_reference(
self,
collection: str,
text: str,
external_id: str,
external_source_name: str,
description: str | None = None,
additional_metadata: str | None = None
) -> None:
"""
Save a reference to external information.
Parameters:
- collection: Collection to save to
- text: Text content of the reference
- external_id: ID in the external system
- external_source_name: Name of the external source
- description: Optional description
- additional_metadata: Optional additional metadata
"""
async def get(
self,
collection: str,
key: str
) -> MemoryQueryResult | None:
"""
Get information from memory by key.
Parameters:
- collection: Collection to search in
- key: Key/ID of the information
Returns:
MemoryQueryResult if found, None otherwise
"""
async def search(
self,
collection: str,
query: str,
limit: int = 1,
min_relevance_score: float = 0.0
) -> list[MemoryQueryResult]:
"""
Search for information using natural language query.
Parameters:
- collection: Collection to search in
- query: Natural language search query
- limit: Maximum number of results
- min_relevance_score: Minimum relevance threshold
Returns:
List of MemoryQueryResult objects ordered by relevance
"""
class MemoryQueryResult:
"""
Result from a memory query operation.
"""
def __init__(
self,
metadata: MemoryRecordMetadata,
relevance: float,
embedding: list[float] | None = None
):
"""
Initialize memory query result.
Parameters:
- metadata: Metadata for the memory record
- relevance: Relevance score (0.0 to 1.0)
- embedding: Vector embedding (optional)
"""
@property
def metadata(self) -> MemoryRecordMetadata:
"""Get the memory record metadata."""
@property
def relevance(self) -> float:
"""Get the relevance score."""
@property
def embedding(self) -> list[float] | None:
"""Get the embedding vector."""
class MemoryRecordMetadata:
"""
Metadata for a memory record.
"""
def __init__(
self,
is_reference: bool,
id: str,
text: str,
description: str,
external_source_name: str | None = None,
additional_metadata: str | None = None
):
"""
Initialize memory record metadata.
Parameters:
- is_reference: Whether this is a reference to external content
- id: Unique identifier
- text: Text content
- description: Description of the content
- external_source_name: External source name if reference
- additional_metadata: Additional metadata string
"""
@property
def id(self) -> str:
"""Get the record ID."""
@property
def text(self) -> str:
"""Get the text content."""
@property
def description(self) -> str:
"""Get the description."""
@property
def is_reference(self) -> bool:
"""Check if this is a reference record."""Usage Examples
Basic Memory Operations
from semantic_kernel.connectors.memory_stores.chroma import ChromaMemoryStore
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding
from semantic_kernel.memory import SemanticTextMemory
# Setup vector store and embeddings
vector_store = ChromaMemoryStore(persist_directory="./chroma_db")
embedding_service = OpenAITextEmbedding(
ai_model_id="text-embedding-ada-002",
api_key="your-api-key"
)
# Create semantic memory
memory = SemanticTextMemory(
storage=vector_store,
embeddings_generator=embedding_service
)
# Save information
await memory.save_information(
collection="documents",
text="Semantic Kernel is an AI orchestration framework",
id="sk_intro",
description="Introduction to Semantic Kernel"
)
# Search for information
results = await memory.search(
collection="documents",
query="What is Semantic Kernel?",
limit=3,
min_relevance_score=0.7
)
for result in results:
print(f"Relevance: {result.relevance}")
print(f"Text: {result.metadata.text}")Integration with Kernel
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function
# Create kernel with memory
kernel = Kernel()
kernel.add_service(embedding_service)
# Add memory plugin
class MemoryPlugin:
def __init__(self, memory: SemanticTextMemory):
self._memory = memory
@kernel_function(description="Search memory for relevant information")
async def search_memory(self, query: str, collection: str = "documents") -> str:
results = await self._memory.search(collection, query, limit=3)
if results:
return results[0].metadata.text
return "No relevant information found."
@kernel_function(description="Save information to memory")
async def save_to_memory(
self,
text: str,
id: str,
collection: str = "documents"
) -> str:
await self._memory.save_information(collection, text, id)
return f"Saved information with ID: {id}"
# Add memory plugin to kernel
memory_plugin = MemoryPlugin(memory)
kernel.add_plugin(memory_plugin, plugin_name="memory")
# Use memory in functions
result = await kernel.invoke("memory", "search_memory", query="AI frameworks")
print(result.value)Install with Tessl CLI
npx tessl i tessl/pypi-semantic-kernel