Implement new streaming ask endpoint (WIP)

clients/dotnet/WebClient/MemoryWebClient.cs

@@ -7,6 +7,8 @@
 using System.Linq;
 using System.Net;
 using System.Net.Http;
+using System.Net.Http.Json;
+using System.Runtime.CompilerServices;
 using System.Text;
 using System.Text.Json;
 using System.Threading;
@@ -358,6 +360,48 @@ public async Task<MemoryAnswer> AskAsync(
         return JsonSerializer.Deserialize<MemoryAnswer>(json, s_caseInsensitiveJsonOptions) ?? new MemoryAnswer();
     }
 
+    /// <inheritdoc />
+    public async IAsyncEnumerable<MemoryAnswer> AskStreamingAsync(
+        string question,
+        string? index = null,
+        MemoryFilter? filter = null,
+        ICollection<MemoryFilter>? filters = null,
+        double minRelevance = 0,
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        if (filter != null)
+        {
+            if (filters == null) { filters = new List<MemoryFilter>(); }
+
+            filters.Add(filter);
+        }
+
+        MemoryQuery request = new()
+        {
+            Index = index,
+            Question = question,
+            Filters = (filters is { Count: > 0 }) ? filters.ToList() : new(),
+            MinRelevance = minRelevance
+        };
+        using StringContent content = new(JsonSerializer.Serialize(request), Encoding.UTF8, "application/json");
+
+        using var httpRequest = new HttpRequestMessage(HttpMethod.Post, Constants.HttpAskStreamEndpoint);
+        httpRequest.Content = content;
+
+        using HttpResponseMessage response = await this._client.SendAsync(httpRequest, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
+        response.EnsureSuccessStatusCode();
+
+        await foreach (var responsePart in response.Content.ReadFromJsonAsAsyncEnumerable<MemoryAnswer>(cancellationToken))
+        {
+            if (responsePart is null)
+            {
+                continue;
+            }
+
+            yield return responsePart;
+        }
+    }
+
     #region private
 
     private static (string contentType, long contentLength, DateTimeOffset lastModified) GetFileDetails(HttpResponseMessage response)

examples/001-dotnet-WebClient/Program.cs

@@ -55,6 +55,7 @@ public static async Task Main()
         // =======================
 
         await AskSimpleQuestion();
+        await AskSimpleQuestionWithStreamingAndShowSources();
         await AskSimpleQuestionAndShowSources();
         await AskQuestionAboutImageContent();
         await AskQuestionUsingFilter();
@@ -272,6 +273,53 @@ due to the speed of light being a very large number when squared. This concept i
         */
     }
 
+    private static async Task AskSimpleQuestionWithStreamingAndShowSources()
+    {
+        var question = "Any news from NASA about Orion?";
+        Console.WriteLine($"Question: {question}");
+        var answer = s_memory.AskStreamingAsync(question, filter: MemoryFilters.ByTag("user", "Taylor"));
+        Console.WriteLine("\nAnswer:\n");
+
+        List<Citation>? citations = [];
+        bool isFirstPart = true;
+        await foreach (var answerPart in answer)
+        {
+            if (isFirstPart)
+            {
+                citations = answerPart.RelevantSources;
+                isFirstPart = false;
+            }
+
+            Console.Write(answerPart.Result);
+        }
+
+        Console.WriteLine("\n\nSources:\n");
+        foreach (var x in citations)
+        {
+            Console.WriteLine(x.SourceUrl != null
+                ? $"  - {x.SourceUrl} [{x.Partitions.First().LastUpdate:D}]"
+                : $"  - {x.SourceName}  - {x.Link} [{x.Partitions.First().LastUpdate:D}]");
+        }
+
+        Console.WriteLine("\n====================================\n");
+
+        /* OUTPUT
+
+        Question: Any news from NASA about Orion?
+
+        Answer:
+        Yes, NASA has invited media to see the new test version of the Orion spacecraft and the hardware teams will use to recover the capsule and astronauts upon their return from space during the Artemis II mission.
+        The event will take place at Naval Base San Diego on August 2.
+        Personnel involved in recovery operations from NASA, the U.S. Navy, and the U.S. Air Force will be available to speak with media.
+        Teams are currently conducting tests in the Pacific Ocean to demonstrate and evaluate the processes, procedures, and hardware for recovery operations for crewed Artemis missions.
+        The tests will help prepare the team for Artemis II, NASA's first crewed mission under Artemis that will send four astronauts in Orion around the Moon to checkout systems ahead of future lunar missions.
+        The Artemis II crew will participate in recovery testing at sea next year.
+
+        Sources:
+            - /download?index=default&documentId=doc003&filename=file5-NASA-news.pdf [Friday, 17 May 2024]
+        */
+    }
+
     // Another question without filters and show sources
     private static async Task AskSimpleQuestionAndShowSources()
     {

examples/002-dotnet-Serverless/Program.cs

@@ -91,6 +91,7 @@ public static async Task Main()
         // =======================
 
         await AskSimpleQuestion();
+        await AskSimpleQuestionWithStreamingAndShowSources();
         await AskSimpleQuestionAndShowSources();
         await AskQuestionAboutImageContent();
         await AskQuestionUsingFilter();
@@ -310,6 +311,55 @@ due to the speed of light being a very large number when squared. This concept i
         */
     }
 
+    // Question without filters and show sources, with streaming
+    private static async Task AskSimpleQuestionWithStreamingAndShowSources()
+    {
+        var question = "What's E = m*c^2?";
+        Console.WriteLine($"Question: {question}");
+
+        var answer = s_memory.AskStreamingAsync(question, minRelevance: 0.76);
+        Console.WriteLine("\nAnswer:\n");
+
+        List<Citation>? citations = [];
+        bool isFirstPart = true;
+        await foreach (var answerPart in answer)
+        {
+            if (isFirstPart)
+            {
+                citations = answerPart.RelevantSources;
+                isFirstPart = false;
+            }
+
+            Console.Write(answerPart.Result);
+        }
+
+        Console.WriteLine("\n\nSources:\n");
+        foreach (var x in citations)
+        {
+            Console.WriteLine(x.SourceUrl != null
+                ? $"  - {x.SourceUrl} [{x.Partitions.First().LastUpdate:D}]"
+                : $"  - {x.SourceName}  - {x.Link} [{x.Partitions.First().LastUpdate:D}]");
+        }
+
+        Console.WriteLine("\n====================================\n");
+
+        /* OUTPUT
+
+        Question: What's E = m*c^2?
+
+        Answer: E = m*c^2 is the formula representing the principle of mass-energy equivalence, which was introduced by Albert Einstein. In this equation,
+        E stands for energy, m represents mass, and c is the speed of light in a vacuum, which is approximately 299,792,458 meters per second (m/s).
+        The equation states that the energy (E) of a system in its rest frame is equal to its mass (m) multiplied by the square of the speed of light (c^2).
+        This implies that mass and energy are interchangeable; a small amount of mass can be converted into a large amount of energy and vice versa,
+        due to the speed of light being a very large number when squared. This concept is a fundamental principle in physics and has important implications
+        in various fields, including nuclear physics and cosmology.
+
+        Sources:
+            - /download?index=default&documentId=doc003&filename=file5-NASA-news.pdf [Friday, 17 May 2024]
+
+        */
+    }
+
     // Another question without filters and show sources
     private static async Task AskSimpleQuestionAndShowSources()
     {

service/Abstractions/Constants.cs

@@ -52,6 +52,7 @@ public static class Constants
 
     // Endpoints
     public const string HttpAskEndpoint = "/ask";
+    public const string HttpAskStreamEndpoint = "/ask/stream";
     public const string HttpSearchEndpoint = "/search";
     public const string HttpDownloadEndpoint = "/download";
     public const string HttpUploadEndpoint = "/upload";

service/Abstractions/IKernelMemory.cs

@@ -211,4 +211,22 @@ public Task<MemoryAnswer> AskAsync(
         ICollection<MemoryFilter>? filters = null,
         double minRelevance = 0,
         CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Search the given index for an answer to the given query.
+    /// </summary>
+    /// <param name="question">Question to answer</param>
+    /// <param name="index">Optional index name</param>
+    /// <param name="filter">Filter to match</param>
+    /// <param name="filters">Filters to match (using inclusive OR logic). If 'filter' is provided too, the value is merged into this list.</param>
+    /// <param name="minRelevance">Minimum Cosine Similarity required</param>
+    /// <param name="cancellationToken">Async task cancellation token</param>
+    /// <returns>A stream that contains an answer to the query, or an empty list</returns>
+    public IAsyncEnumerable<MemoryAnswer> AskStreamingAsync(
+        string question,
+        string? index = null,
+        MemoryFilter? filter = null,
+        ICollection<MemoryFilter>? filters = null,
+        double minRelevance = 0,
+        CancellationToken cancellationToken = default);
 }

service/Abstractions/Search/ISearchClient.cs

@@ -45,6 +45,23 @@ Task<MemoryAnswer> AskAsync(
         double minRelevance = 0,
         CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Answer the given question, if possible, grounding the response with relevant memories matching the given criteria.
+    /// First result in the stream contains metadata about the result, subsequent results only contain answer tokens.
+    /// </summary>
+    /// <param name="index">Index (aka collection) to search for grounding information</param>
+    /// <param name="question">Question to answer</param>
+    /// <param name="filters">Filtering criteria to select memories to consider</param>
+    /// <param name="minRelevance">Minimum relevance of the memories considered</param>
+    /// <param name="cancellationToken">Async task cancellation token</param>
+    /// <returns>Answer to the given question</returns>
+    IAsyncEnumerable<MemoryAnswer> AskStreamingAsync(
+        string index,
+        string question,
+        ICollection<MemoryFilter>? filters = null,
+        double minRelevance = 0,
+        CancellationToken cancellationToken = default);
+
     /// <summary>
     /// List the available memory indexes (aka collections).
     /// </summary>

service/Core/MemoryServerless.cs

@@ -257,4 +257,29 @@ public Task<MemoryAnswer> AskAsync(
             minRelevance: minRelevance,
             cancellationToken: cancellationToken);
     }
+
+    /// <inheritdoc />
+    public IAsyncEnumerable<MemoryAnswer> AskStreamingAsync(
+        string question,
+        string? index = null,
+        MemoryFilter? filter = null,
+        ICollection<MemoryFilter>? filters = null,
+        double minRelevance = 0,
+        CancellationToken cancellationToken = default)
+    {
+        if (filter != null)
+        {
+            if (filters == null) { filters = new List<MemoryFilter>(); }
+
+            filters.Add(filter);
+        }
+
+        index = IndexName.CleanName(index, this._defaultIndexName);
+        return this._searchClient.AskStreamingAsync(
+            index: index,
+            question: question,
+            filters: filters,
+            minRelevance: minRelevance,
+            cancellationToken: cancellationToken);
+    }
 }

service/Core/MemoryService.cs

@@ -234,4 +234,29 @@ public Task<MemoryAnswer> AskAsync(
             minRelevance: minRelevance,
             cancellationToken: cancellationToken);
     }
+
+    /// <inheritdoc />
+    public IAsyncEnumerable<MemoryAnswer> AskStreamingAsync(
+        string question,
+        string? index = null,
+        MemoryFilter? filter = null,
+        ICollection<MemoryFilter>? filters = null,
+        double minRelevance = 0,
+        CancellationToken cancellationToken = default)
+    {
+        if (filter != null)
+        {
+            if (filters == null) { filters = new List<MemoryFilter>(); }
+
+            filters.Add(filter);
+        }
+
+        index = IndexName.CleanName(index, this._defaultIndexName);
+        return this._searchClient.AskStreamingAsync(
+            index: index,
+            question: question,
+            filters: filters,
+            minRelevance: minRelevance,
+            cancellationToken: cancellationToken);
+    }
 }