streaming json rendering in responder dev route

core/assets/js/app.js

@@ -27,6 +27,7 @@ import * as Components from "../svelte/**/*.svelte";
 import { format } from "timeago.js";
 import ClipboardJS from "clipboard";
 import Chart from "chart.js/auto";
+import { parse } from 'best-effort-json-parser'
 
 import "@getcanary/web/components/canary-root.js";
 import "@getcanary/web/components/canary-provider-cloud.js";
@@ -151,6 +152,18 @@ let hooks = {
       });
     },
   },
+  PartialJSON: {
+    mounted() {
+      this.fn();
+    },
+    updated() {
+      this.fn();
+    },
+    fn() {
+      const parsed = parse(this.el.textContent);
+      this.el.textContent = JSON.stringify(parsed, null, 2);
+    },
+  },
 };
 let liveSocket = new LiveSocket("/live", Socket, {
   params: { _csrf_token: csrfToken },

core/assets/package.json

@@ -18,6 +18,7 @@
   },
   "dependencies": {
     "@getcanary/web": "^1.0.0-rc.4",
+    "best-effort-json-parser": "^1.1.2",
     "chart.js": "^4.4.4",
     "clipboard": "^2.0.11",
     "clsx": "^2.1.1",

core/lib/canary/interactions/responder.ex

@@ -19,15 +19,18 @@ defmodule Canary.Interactions.Responder.Default do
   @behaviour Canary.Interactions.Responder
   require Ash.Query
 
+  alias Canary.Sources.Document
+
   def run(sources, query, handle_delta, opts) do
     {:ok, results} = Canary.Searcher.run(sources, query, cache: opts[:cache])
 
     docs =
       results
       |> search_results_to_docs()
-      |> then(
-        &Canary.Reranker.run!(query, &1, threshold: 0.05, renderer: fn doc -> doc.content end)
-      )
+      |> then(fn docs ->
+        opts = [threshold: 0.01, renderer: fn doc -> doc.content end]
+        Canary.Reranker.run!(query, docs, opts) |> Enum.take(3)
+      end)
 
     messages = [
       %{
@@ -47,9 +50,10 @@ defmodule Canary.Interactions.Responder.Default do
         %{
           model: Application.fetch_env!(:canary, :chat_completion_model),
           messages: messages,
-          temperature: 0.2,
+          temperature: 0,
           frequency_penalty: 0.02,
           max_tokens: 5000,
+          response_format: %{type: "json_object"},
           stream: handle_delta != nil
         },
         callback: fn data ->
@@ -68,7 +72,7 @@ defmodule Canary.Interactions.Responder.Default do
     completion = if completion == "", do: Agent.get(pid, & &1), else: completion
     safe(handle_delta, %{type: :complete, content: completion})
 
-    {:ok, %{response: completion, references: []}}
+    {:ok, %{response: completion, references: [], docs: docs}}
   end
 
   defp search_results_to_docs(results) do
@@ -80,8 +84,9 @@ defmodule Canary.Interactions.Responder.Default do
     Canary.Sources.Document
     |> Ash.Query.filter(id in ^doc_ids)
     |> Ash.read!()
-    |> Enum.flat_map(fn %{chunks: chunks} -> chunks end)
-    |> Enum.map(fn chunk -> %{title: chunk.value.title, content: chunk.value.content} end)
+    |> Enum.map(fn %Document{meta: %Ash.Union{value: meta}, chunks: chunks} ->
+      %{title: meta.title, content: chunks |> Enum.map(& &1.value.content) |> Enum.join("\n")}
+    end)
   end
 
   defp safe(func, arg) do

core/lib/canary/prompts/responder_assistant_schema.json

@@ -1,6 +1,9 @@
 {
     "type": "object",
     "properties": {
+      "thinking": {
+        "type": "string"
+      },
       "response": {
         "type": "array",
         "items": {
@@ -45,5 +48,5 @@
         }
       }
     },
-    "required": ["response"]
+    "required": ["thinking", "response"]
   }

core/lib/canary/prompts/responder_system.eex

@@ -13,8 +13,19 @@ You are a world-class technical support engineer. Your job is to answer user's q
 
 ## Response Format
 
-In any case, you must respond in markdown format. List, Link, Inline Code, Block Code, and Bold are supported.
-In addition, you can use `<canary-reference>` tag to reference the given documents.
+You MUST output JSON that follows the schema below:
+
+```json
+<%= @schema %>
+```
+
+Information about each field:
+- thinking: this is invisible to the user. use it to plan on how will you respond.
+- response: this is the actual response. There are two types of response:
+  - text: this is a string. You can use it to respond with a single sentence.
+  - reference: this is a JSON object. You can use it to reference the given documents.
+
+## Guidelines
 
 You should always start with **<IMMEDIATE_ANSWER>**, and then add more details.
 
@@ -27,7 +38,6 @@ You should always start with **<IMMEDIATE_ANSWER>**, and then add more details.
 - Not sure what you mean.
 - etc
 
-## Guidelines
 
 - Always stick to the question asked, and do not add any extra information.
 - You can use your existing knowledge to understand the user's query and given documents, but you should NOT directly use it for answering the question.

core/lib/canary/query/understander.ex

@@ -47,7 +47,7 @@ defmodule Canary.Query.Understander.LLM do
       }
     ]
 
-    args = %{model: chat_model, messages: messages}
+    args = %{model: chat_model, messages: messages, temperature: 0}
 
     case Canary.AI.chat(args, timeout: 2_000) do
       {:ok, completion} ->

core/lib/canary/reranker.ex

@@ -35,9 +35,9 @@ defmodule Canary.Reranker.Cohere do
       end
 
     case result do
-      {:ok, %{status: 200, body: body}} ->
+      {:ok, %{status: 200, body: %{"results" => results}}} ->
         reranked =
-          body["results"]
+          results
           |> Enum.sort_by(& &1["relevance_score"], :asc)
           |> Enum.filter(fn %{"relevance_score" => score} -> score > threshold end)
           |> Enum.map(fn %{"index" => i} -> i end)
@@ -79,18 +79,18 @@ defmodule Canary.Reranker.Jina do
   use Retry
 
   def run(query, docs, opts) do
-    threshold = opts[:threshold] || 0
     renderer = opts[:renderer] || fn doc -> doc end
+    threshold = opts[:threshold] || 0
 
     result =
       retry with: exponential_backoff() |> randomize |> expiry(4_000) do
         request(query, docs, renderer)
       end
 
     case result do
-      {:ok, %{status: 200, body: body}} ->
+      {:ok, %{status: 200, body: %{"results" => results}}} ->
         reranked =
-          body["results"]
+          results
           |> Enum.sort_by(& &1["relevance_score"], :asc)
           |> Enum.filter(fn %{"relevance_score" => score} -> score > threshold end)
           |> Enum.map(fn %{"index" => i} -> i end)

core/lib/canary_web/live/dev/responder_live.ex

@@ -39,7 +39,7 @@ defmodule CanaryWeb.Dev.ResponderLive do
         >
           <Primer.octicon name="paste-16" />
         </Primer.button>
-        <pre class=" text-lg"><%= @response %></pre>
+        <pre id="response-json" phx-hook="PartialJSON" class="text-lg"><%= @response %></pre>
 
         <Primer.button
           is_small
@@ -89,20 +89,53 @@ defmodule CanaryWeb.Dev.ResponderLive do
 
   @impl true
   def handle_event("submit", %{"query" => query}, socket) do
-    now = System.monotonic_time()
-
-    {:ok, %{response: response, docs: docs}} =
-      Canary.Interactions.Responder.run(socket.assigns.selected, query, fn _ -> :ok end)
-
-    delta = System.monotonic_time() - now
+    self = self()
+    selected = socket.assigns.selected
 
     socket =
       socket
       |> assign(query: query)
-      |> assign(response: response)
-      |> assign(docs: docs)
-      |> assign(latency: System.convert_time_unit(delta, :native, :millisecond))
+      |> assign(response: "")
+      |> assign(docs: [])
+      |> assign(latency: 0)
+      |> assign(started_at: System.monotonic_time())
+      |> start_async(:task, fn ->
+        {:ok, %{docs: docs}} = Canary.Interactions.Responder.run(selected, query, &send(self, &1))
+        docs
+      end)
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_info(%{type: :progress, content: content}, socket) do
+    socket = socket |> assign(response: socket.assigns.response <> content)
+
+    socket =
+      if socket.assigns.latency == 0 do
+        delta = System.monotonic_time() - socket.assigns.started_at
+        socket |> assign(latency: System.convert_time_unit(delta, :native, :millisecond))
+      else
+        socket
+      end
 
     {:noreply, socket}
   end
+
+  @impl true
+  def handle_info(%{type: :complete, content: content}, socket) do
+    socket = socket |> assign(response: content)
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_async(:task, {:ok, docs}, socket) do
+    socket = socket |> assign(docs: docs)
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_async(:task, _result, socket) do
+    {:noreply, socket}
+  end
 end