[Bug] Qdrant MemoryDb seems to recreate collection with every request #1056
Comments
|
is the memory configured to persist documents? |
|
Unsure what you're referring to - isnt that what memory is - persistent storage? We've been using KM with AI Search for a while and are looking at supporting Qdrant in addition to AI Search. Our AI Search extension method for setting up KM looks like this: private static IKernelMemory CreateKernelMemoryInstance(
IServiceProvider serviceProvider,
ServiceConfigurationOptions serviceConfigurationOptions,
string blobContainerName,
string indexName = "")
{
var azureCredentialHelper = serviceProvider.GetRequiredService<AzureCredentialHelper>();
var configuration = serviceProvider.GetRequiredService<IConfiguration>();
var baseSearchClient = serviceProvider.GetRequiredService<SearchIndexClient>();
// Get OpenAI connection string
var openAiConnString = configuration.GetConnectionString("openai-planner") ?? throw new InvalidOperationException("OpenAI connection string not found.");
// Extract Endpoint and Key
var openAiEndpoint = openAiConnString.Split(";").FirstOrDefault(x => x.Contains("Endpoint="))?.Split("=")[1]
?? throw new ArgumentException("OpenAI endpoint must be provided in the configuration.");
var openAiKey = openAiConnString.Split(";").FirstOrDefault(x => x.Contains("Key="))?.Split("=")[1]
?? throw new ArgumentException("OpenAI key must be provided in the configuration.");
var openAiEmbeddingConfig = new AzureOpenAIConfig()
{
Auth = AzureOpenAIConfig.AuthTypes.APIKey,
Endpoint = openAiEndpoint,
APIKey = openAiKey,
Deployment = serviceConfigurationOptions.OpenAi.EmbeddingModelDeploymentName,
APIType = AzureOpenAIConfig.APITypes.EmbeddingGeneration
};
var openAiChatCompletionConfig = new AzureOpenAIConfig()
{
Auth = AzureOpenAIConfig.AuthTypes.APIKey,
Endpoint = openAiEndpoint,
APIKey = openAiKey,
Deployment = serviceConfigurationOptions.OpenAi.Gpt4o_Or_Gpt4128KDeploymentName,
APIType = AzureOpenAIConfig.APITypes.ChatCompletion
};
var azureAiSearchConfig = new AzureAISearchConfig()
{
Endpoint = baseSearchClient.Endpoint.AbsoluteUri,
Auth = AzureAISearchConfig.AuthTypes.ManualTokenCredential
};
azureAiSearchConfig.SetCredential(azureCredentialHelper.GetAzureCredential());
// Blob Connection String
var blobConnection = configuration.GetConnectionString("blob-docing") ?? throw new ArgumentException("Blob Connection String must be provided");
// Extract Blob account name
var blobAccountName = blobConnection.Split(".")[0].Split("//")[1];
var azureBlobsConfig = new AzureBlobsConfig()
{
Account = blobAccountName,
ConnectionString = blobConnection,
Container = blobContainerName,
Auth = AzureBlobsConfig.AuthTypes.ManualTokenCredential
};
azureBlobsConfig.SetCredential(azureCredentialHelper.GetAzureCredential());
var textPartitioningOptions = new TextPartitioningOptions
{
MaxTokensPerParagraph = PartitionSize,
OverlappingTokens = 0
};
// Get or create KernelMemoryConfig
var kernelMemoryConfig = serviceProvider.GetService<KernelMemoryConfig>() ?? configuration.GetSection("KernelMemory").Get<KernelMemoryConfig>() ?? new KernelMemoryConfig();
if (kernelMemoryConfig.DataIngestion.MemoryDbUpsertBatchSize < 20)
{
kernelMemoryConfig.DataIngestion.MemoryDbUpsertBatchSize = 20;
}
if (!string.IsNullOrEmpty(indexName))
{
kernelMemoryConfig.DefaultIndexName = indexName;
}
var kernelMemoryBuilder = new KernelMemoryBuilder();
kernelMemoryBuilder.Services.AddSingleton(kernelMemoryConfig);
kernelMemoryBuilder
.WithAzureOpenAITextEmbeddingGeneration(openAiEmbeddingConfig)
.WithAzureOpenAITextGeneration(openAiChatCompletionConfig)
.WithCustomTextPartitioningOptions(textPartitioningOptions)
.WithAzureBlobsDocumentStorage(azureBlobsConfig)
.WithAzureAISearchMemoryDb(azureAiSearchConfig)
;
// Add Logging
kernelMemoryBuilder.Services.AddLogging(l =>
{
l.AddConsole().SetMinimumLevel(LogLevel.Information);
l.AddConfiguration(configuration);
});
var kernelMemory = kernelMemoryBuilder.Build();
return kernelMemory;
}And this is the Qdrant version: private static IKernelMemory CreateKernelMemoryInstance(
IServiceProvider serviceProvider,
ServiceConfigurationOptions serviceConfigurationOptions,
string blobContainerName,
string indexName = "")
{
var azureCredentialHelper = serviceProvider.GetRequiredService<AzureCredentialHelper>();
var configuration = serviceProvider.GetRequiredService<IConfiguration>();
//var baseSearchClient = serviceProvider.GetRequiredService<SearchIndexClient>();
// Get OpenAI connection string
var openAiConnString = configuration.GetConnectionString("openai-planner") ?? throw new InvalidOperationException("OpenAI connection string not found.");
// Extract Endpoint and Key
var openAiEndpoint = openAiConnString.Split(";").FirstOrDefault(x => x.Contains("Endpoint="))?.Split("=")[1]
?? throw new ArgumentException("OpenAI endpoint must be provided in the configuration.");
var openAiKey = openAiConnString.Split(";").FirstOrDefault(x => x.Contains("Key="))?.Split("=")[1]
?? throw new ArgumentException("OpenAI key must be provided in the configuration.");
var openAiEmbeddingConfig = new AzureOpenAIConfig()
{
Auth = AzureOpenAIConfig.AuthTypes.APIKey,
Endpoint = openAiEndpoint,
APIKey = openAiKey,
Deployment = serviceConfigurationOptions.OpenAi.EmbeddingModelDeploymentName,
APIType = AzureOpenAIConfig.APITypes.EmbeddingGeneration
};
var openAiChatCompletionConfig = new AzureOpenAIConfig()
{
Auth = AzureOpenAIConfig.AuthTypes.APIKey,
Endpoint = openAiEndpoint,
APIKey = openAiKey,
Deployment = serviceConfigurationOptions.OpenAi.Gpt4o_Or_Gpt4128KDeploymentName,
APIType = AzureOpenAIConfig.APITypes.ChatCompletion
};
var qdrantConnectionString = configuration.GetConnectionString("qdrant") ?? throw new ArgumentException("Qdrant connection string must be provided");
// Parse the connection string to extract the Qdrant URL and API key
// Connection string format: "Endpoint=http://localhost:6334;Key=123456!@#$%"
var qdrantUrl = qdrantConnectionString.Split(";").FirstOrDefault(x => x.Contains("Endpoint="))?.Split("=")[1]
?? throw new ArgumentException("Qdrant endpoint must be provided in the configuration.");
// If the URL points to localhost and a port number - this is the grpc port number. Increase the port number by one so we get
// the http port number.
if (qdrantUrl.Contains("localhost") && qdrantUrl.Contains(":"))
{
var portIndex = qdrantUrl.LastIndexOf(":", StringComparison.Ordinal);
var portNumber = int.Parse(qdrantUrl.Substring(portIndex + 1));
qdrantUrl = qdrantUrl.Substring(0, portIndex + 1) + (portNumber + 1);
}
var qdrantApiKey = qdrantConnectionString.Split(";").FirstOrDefault(x => x.Contains("Key="))?.Split("=")[1]
?? throw new ArgumentException("Qdrant API key must be provided in the configuration.");
var qdrantConfig = new QdrantConfig()
{
Endpoint = qdrantUrl,
APIKey = qdrantApiKey,
};
// Blob Connection String
var blobConnection = configuration.GetConnectionString("blob-docing") ?? throw new ArgumentException("Blob Connection String must be provided");
// Extract Blob account name
var blobAccountName = blobConnection.Split(".")[0].Split("//")[1];
var azureBlobsConfig = new AzureBlobsConfig()
{
Account = blobAccountName,
ConnectionString = blobConnection,
Container = blobContainerName,
Auth = AzureBlobsConfig.AuthTypes.ManualTokenCredential
};
azureBlobsConfig.SetCredential(azureCredentialHelper.GetAzureCredential());
var textPartitioningOptions = new TextPartitioningOptions
{
MaxTokensPerParagraph = PartitionSize,
OverlappingTokens = 0
};
// Get or create KernelMemoryConfig
var kernelMemoryConfig = serviceProvider.GetService<KernelMemoryConfig>() ?? configuration.GetSection("KernelMemory").Get<KernelMemoryConfig>() ?? new KernelMemoryConfig();
if (kernelMemoryConfig.DataIngestion.MemoryDbUpsertBatchSize < 20)
{
kernelMemoryConfig.DataIngestion.MemoryDbUpsertBatchSize = 20;
}
if (!string.IsNullOrEmpty(indexName))
{
kernelMemoryConfig.DefaultIndexName = indexName;
}
var kernelMemoryBuilder = new KernelMemoryBuilder();
kernelMemoryBuilder.Services.AddSingleton(kernelMemoryConfig);
kernelMemoryBuilder
.WithAzureOpenAITextEmbeddingGeneration(openAiEmbeddingConfig)
.WithAzureOpenAITextGeneration(openAiChatCompletionConfig)
.WithCustomTextPartitioningOptions(textPartitioningOptions)
.WithAzureBlobsDocumentStorage(azureBlobsConfig)
.WithQdrantMemoryDb(qdrantConfig)
;
// Add Logging
kernelMemoryBuilder.Services.AddLogging(l =>
{
l.AddConsole().SetMinimumLevel(LogLevel.Information);
l.AddConfiguration(configuration);
});
var kernelMemory = kernelMemoryBuilder.Build();
return kernelMemory;
}As you can see, very little difference except for the MemorDb provider being swapped out. This is the method that stores content - working today with AI Search - an index is created on AI search if it doesn't exist, and then content is additively stored. With Qdrant, each time we import a document, the collection is re-created and the last piece of content imported is the only one available. public async Task StoreContentAsync(string documentLibraryName, string indexName, Stream fileStream,
string fileName, string? documentUrl, string? userId = null, Dictionary<string, string>? additionalTags = null)
{
var memory = await GetKernelMemoryForDocumentLibrary(documentLibraryName);
if (memory == null)
{
_logger.LogError("Kernel Memory service not found for Document Library {DocumentLibraryName}", documentLibraryName);
throw new Exception("Kernel Memory service not found for Document Library " + documentLibraryName);
}
// URL encode the documentUrl if it is not null
if (documentUrl != null)
{
documentUrl = WebUtility.UrlEncode(documentUrl);
}
// Decode the fileName
fileName = WebUtility.UrlDecode(fileName);
// Sanitize the fileName - replace spaces with underscores, pluses with underscores, tildes with underscores, and slashes with underscores
fileName = fileName.Replace(" ", "_").Replace("+", "_").Replace("~", "_").Replace("/", "_");
var documentRequest = new DocumentUploadRequest()
{
DocumentId = fileName,
Files = [new DocumentUploadRequest.UploadedFile(fileName, fileStream)],
Index = indexName
};
var isDocumentLibraryDocument = documentLibraryName.StartsWith("Additional-").ToString().ToLowerInvariant();
var tags = new Dictionary<string, string>
{
{"DocumentProcessName", documentLibraryName},
{"IsDocumentLibraryDocument", isDocumentLibraryDocument},
{"OriginalDocumentUrl", documentUrl ?? string.Empty},
{"UploadedByUserOid", userId ?? string.Empty}
};
foreach (var (key, value) in tags)
{
documentRequest.Tags.Add(key, value);
}
if (additionalTags != null)
{
// Check the keys in the additionalParameters dictionary. If any of them look like URLs (if they start with https: or http:, URL encode the value
foreach (var (key, value) in additionalTags)
{
if (key.StartsWith("http:") || key.StartsWith("https:"))
{
additionalTags[key] = WebUtility.UrlEncode(value);
}
}
foreach (var (key, value) in additionalTags)
{
documentRequest.Tags.Add(key, value);
}
}
await memory.ImportDocumentAsync(documentRequest);
} |
|
Oh, and here's how Qdrant is set up in the Aspire AppHost, you'll notice it's persistent and with a data volume, if that's what you were referring to. qdrant = builder.AddQdrant("qdrant")
.WithExternalHttpEndpoints()
.WithDataVolume("pvico-qdrant-vol")
.WithLifetime(ContainerLifetime.Persistent); |
I've opened #1068 to discuss. |
# for free
to join this conversation on GitHub.
Already have an account?
# to comment
Context / Scenario
Every time I upload a new file with ImportDocumentAsync, the collection is recreated and everything in it is cleared out, making it kind of useless. I've tried creating the collection ahead of time using the C# QdrantClient, but this behavior persist.
I understand that this is the default behavior in Qdrant when using "recreate_collection" - but the memorydb here uses the HTTP REST interface and not the .Net QdrantClient, and I don't know how this maps to SDK methods (it also interferes a bit with Aspire, btw - where the default connection string points to the GRPC endpoint and not the HTTP endpoint for Qdrant).
I'd rewrite the memorydb to use the QdrantClient as a PR, but don't really know enough about Qdrant to do it safely...
What happened?
The collection for a Kernel Memory instance using Qdrant is recreated with every upload.
Importance
a fix would make my life easier
Platform, Language, Versions
Qdrant running in Docker locally using the Aspire Host package extensions for it.
Relevant log output
The text was updated successfully, but these errors were encountered: