emit completion tokens to frontend · graham.systems/petite@dbf098e

+26 -4

src-tauri/src/letta/commands.rs

··· 1 + use tauri::ipc::Channel; 2 + 1 3 use crate::{ 2 - letta::{LettaAgentInfo, LettaConfigKey}, 4 + letta::{types::LettaCompletionMessage, LettaAgentInfo, LettaConfigKey}, 3 5 state::AppState, 4 6 }; 5 7 6 8 #[tauri::command] 7 9 pub async fn list_agents(state: tauri::State<'_, AppState>) -> Result<Vec<LettaAgentInfo>, ()> { 8 - Ok(state.letta_manager.list_agents().await) 10 + match state.letta_manager.list_agents().await { 11 + Ok(res) => Ok(res), 12 + Err(err) => { 13 + eprintln!("failed to list agents: {}", err); 14 + Ok(Vec::new()) 15 + } 16 + } 9 17 } 10 18 11 19 #[tauri::command] ··· 58 66 pub async fn start_llm_completion( 59 67 state: tauri::State<'_, AppState>, 60 68 message: String, 69 + on_event: Channel<LettaCompletionMessage>, 61 70 ) -> Result<(), ()> { 62 - state.letta_manager.start_completion(message).await; 71 + match state.letta_manager.start_completion(message).await { 72 + Ok(mut rx) => { 73 + while let Some(ev) = rx.recv().await { 74 + on_event 75 + .send(ev) 76 + .expect("failed to forward event to channel"); 77 + } 63 78 64 - Ok(()) 79 + Ok(()) 80 + } 81 + Err(err) => { 82 + eprintln!("failed to start completion: {}", err); 83 + 84 + Ok(()) 85 + } 86 + } 65 87 }

+75 -74

src-tauri/src/letta/mod.rs

··· 4 4 use reqwest::Client; 5 5 use reqwest_eventsource::{Event, EventSource}; 6 6 use serde_json::{from_str, json}; 7 - use tauri::async_runtime::Mutex; 7 + use tauri::async_runtime::{channel, spawn, Mutex, Receiver, Sender}; 8 8 use tauri::Wry; 9 9 use tauri_plugin_store::Store; 10 10 11 11 use crate::secrets::SecretsManager; 12 - use types::{LettaAgentInfo, LettaCompletionMessage, LettaConfigKey}; 12 + use types::{LettaAgentInfo, LettaCompletionMessage, LettaConfigKey, LettaError}; 13 13 14 14 pub mod commands; 15 15 pub mod types; ··· 57 57 } 58 58 } 59 59 60 - pub async fn list_agents(&self) -> Vec<LettaAgentInfo> { 61 - match self 60 + pub async fn list_agents(&self) -> Result<Vec<LettaAgentInfo>, LettaError> { 61 + let api_key = self 62 62 .secrets_manager 63 - .get_secret(crate::secrets::SecretName::LettaApiKey) 64 - { 65 - Ok(api_key) => { 66 - let base_url = self.base_url.lock().await.to_owned(); 67 - let req = self 68 - .http_client 69 - .get(format!("{base_url}/v1/agents/")) 70 - .header("Authorization", format!("Bearer {api_key}")); 63 + .get_secret(crate::secrets::SecretName::LettaApiKey)?; 64 + let base_url = self.base_url.lock().await.to_owned(); 65 + let req = self 66 + .http_client 67 + .get(format!("{base_url}/v1/agents/")) 68 + .header("Authorization", format!("Bearer {api_key}")); 71 69 72 - let res = req.send().await.expect("failed to list Letta agents"); 73 - let out = res 74 - .json::<Vec<LettaAgentInfo>>() 75 - .await 76 - .expect("failed to deserialize agent info list"); 70 + let res = req.send().await.expect("failed to list Letta agents"); 71 + let out = res.json::<Vec<LettaAgentInfo>>().await?; 77 72 78 - out 79 - } 80 - Err(_) => Vec::new(), 81 - } 73 + Ok(out) 82 74 } 83 75 84 - pub async fn start_completion(&self, msg: String) { 85 - match self 76 + pub async fn start_completion( 77 + &self, 78 + msg: String, 79 + ) -> Result<Receiver<LettaCompletionMessage>, LettaError> { 80 + let api_key = self 86 81 .secrets_manager 87 - .get_secret(crate::secrets::SecretName::LettaApiKey) 88 - { 89 - Ok(api_key) => { 90 - let base_url = self.base_url.lock().await.to_owned(); 91 - let agent_id = self.agent_id.lock().await.to_owned(); 92 - let body = &json!({ 93 - "messages": [ 94 - { 95 - "role": "user", 96 - "content": [{ 97 - "type": "text", 98 - "text": msg, 99 - }] 100 - } 101 - ], 102 - "stream_tokens": true 103 - }); 82 + .get_secret(crate::secrets::SecretName::LettaApiKey)?; 83 + let base_url = self.base_url.lock().await.to_owned(); 84 + let agent_id = self.agent_id.lock().await.to_owned(); 85 + let body = &json!({ 86 + "messages": [ 87 + { 88 + "role": "user", 89 + "content": [{ 90 + "type": "text", 91 + "text": msg, 92 + }] 93 + } 94 + ], 95 + "stream_tokens": true 96 + }); 97 + 98 + println!("body: {:?}", body); 99 + 100 + let req = self 101 + .http_client 102 + .post(format!("{base_url}/v1/agents/{agent_id}/messages/stream")) 103 + .header("Authorization", format!("Bearer {api_key}")) 104 + .header("Content-Type", "application/json") 105 + .json(body); 106 + 107 + let source = EventSource::new(req).expect("failed to clone request"); 104 108 105 - println!("body: {:?}", body); 109 + let (tx, rx) = channel::<LettaCompletionMessage>(100); 110 + let handler = handle_completion_messages(source, tx); 106 111 107 - let req = self 108 - .http_client 109 - .post(format!("{base_url}/v1/agents/{agent_id}/messages/stream")) 110 - .header("Authorization", format!("Bearer {api_key}")) 111 - .header("Content-Type", "application/json") 112 - .json(body); 112 + spawn(handler); 113 113 114 - let mut source = 115 - EventSource::new(req).expect("could not convert request to event source"); 114 + return Ok(rx); 115 + } 116 + } 116 117 117 - while let Some(event) = source.next().await { 118 - match event { 119 - Ok(Event::Open) => println!("stream opened"), 120 - Ok(Event::Message(msg)) => { 121 - if msg.data == "[DONE]" { 122 - continue; 123 - }; 118 + async fn handle_completion_messages( 119 + mut source: EventSource, 120 + sender: Sender<LettaCompletionMessage>, 121 + ) { 122 + while let Some(event) = source.next().await { 123 + match event { 124 + Ok(Event::Open) => println!("stream opened"), 125 + Ok(Event::Message(msg)) => { 126 + if msg.data == "[DONE]" { 127 + continue; 128 + }; 124 129 125 - match from_str::<LettaCompletionMessage>(&msg.data) { 126 - Ok(content) => { 127 - println!("parsed content: {:?}", content) 128 - } 129 - Err(err) => { 130 - eprintln!( 131 - "failed to parse message: {:?}, err: {}", 132 - msg.data.clone(), 133 - err 134 - ) 135 - } 136 - } 137 - } 138 - Err(err) => { 139 - eprintln!("got stream error: {}", err); 140 - source.close(); 141 - } 130 + match from_str::<LettaCompletionMessage>(&msg.data) { 131 + Ok(content) => { 132 + sender.send(content).await.expect("failed to forward event"); 133 + } 134 + Err(err) => { 135 + eprintln!( 136 + "failed to parse message: {:?}, err: {}", 137 + msg.data.clone(), 138 + err 139 + ) 142 140 } 143 141 } 144 142 } 145 - Err(err) => eprintln!("could not fetch Letta API key: {}", err), 143 + Err(err) => { 144 + eprintln!("got stream error: {}", err); 145 + source.close(); 146 + } 146 147 } 147 148 } 148 149 }

+39

src-tauri/src/letta/types.rs

··· 1 + use core::fmt; 2 + 1 3 use serde::{Deserialize, Serialize}; 2 4 use serde_string_enum::{DeserializeLabeledStringEnum, SerializeLabeledStringEnum}; 3 5 use strum::{Display, EnumString}; 4 6 use ts_rs::TS; 7 + 8 + #[derive(Debug)] 9 + pub enum LettaError { 10 + ConfigurationError(String), 11 + ApiKeyError(keyring::Error), 12 + RequestError(reqwest::Error), 13 + EventSourceError(reqwest_eventsource::Error), 14 + } 15 + 16 + impl fmt::Display for LettaError { 17 + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 18 + match self { 19 + LettaError::ConfigurationError(msg) => write!(f, "Configuration error: {}", msg), 20 + LettaError::ApiKeyError(err) => write!(f, "API key error: {}", err), 21 + LettaError::RequestError(err) => write!(f, "Letta request error: {}", err), 22 + LettaError::EventSourceError(err) => write!(f, "Letta event source error: {}", err), 23 + } 24 + } 25 + } 26 + 27 + impl From<keyring::Error> for LettaError { 28 + fn from(err: keyring::Error) -> Self { 29 + LettaError::ApiKeyError(err) 30 + } 31 + } 32 + 33 + impl From<reqwest::Error> for LettaError { 34 + fn from(err: reqwest::Error) -> Self { 35 + LettaError::RequestError(err) 36 + } 37 + } 38 + 39 + impl From<reqwest_eventsource::Error> for LettaError { 40 + fn from(err: reqwest_eventsource::Error) -> Self { 41 + LettaError::EventSourceError(err) 42 + } 43 + } 5 44 6 45 #[derive(Serialize, Deserialize, Clone, Display, EnumString)] 7 46 #[strum(prefix = "letta")]

+134 -1

src/lib/rust/LettaCompletionMessage.ts

··· 5 5 import type { LettaToolCall } from "./LettaToolCall"; 6 6 import type { LettaToolReturnStatus } from "./LettaToolReturnStatus"; 7 7 8 - export type LettaCompletionMessage = { "message_type": "system_message", id: string, date: string, content: string, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, } | { "message_type": "user_message", id: string, date: string, content: Array<LettaMessageContent>, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, } | { "message_type": "reasoning_message", id: string, date: string, reasoning: string, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, source: LettaReasoningSource | null, signature: string | null, } | { "message_type": "hidden_reasoning_message", id: string, date: string, state: LettaHiddenReasoningState, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, hidden_reasoning: string | null, } | { "message_type": "tool_call_message", id: string, date: string, tool_call: LettaToolCall, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, } | { "message_type": "tool_return_message", id: string, date: string, tool_return: string, status: LettaToolReturnStatus, tool_call_id: string, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, stdout: Array<string> | null, stderr: Array<string> | null, } | { "message_type": "assistant_message", id: string, date: string, content: Array<LettaMessageContent>, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, } | { "message_type": "approval_request_message", id: string, date: string, tool_call: LettaToolCall, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, } | { "message_type": "approval_response_message", id: string, date: string, approve: boolean, approval_request_id: string, name: string | null, otid: string | null, sender_id: string | null, step_id: string | null, is_err: boolean | null, seq_id: bigint | null, run_id: string | null, reason: string | null, } | { "message_type": "stop_reason", stop_reason: string, } | { "message_type": "usage_statistics", completion_tokens: bigint, prompt_tokens: bigint, step_count: bigint, }; 8 + export type LettaCompletionMessage = 9 + | { 10 + message_type: "system_message"; 11 + id: string; 12 + date: string; 13 + content: string; 14 + name: string | null; 15 + otid: string | null; 16 + sender_id: string | null; 17 + step_id: string | null; 18 + is_err: boolean | null; 19 + seq_id: bigint | null; 20 + run_id: string | null; 21 + } 22 + | { 23 + message_type: "user_message"; 24 + id: string; 25 + date: string; 26 + content: Array<LettaMessageContent>; 27 + name: string | null; 28 + otid: string | null; 29 + sender_id: string | null; 30 + step_id: string | null; 31 + is_err: boolean | null; 32 + seq_id: bigint | null; 33 + run_id: string | null; 34 + } 35 + | { 36 + message_type: "reasoning_message"; 37 + id: string; 38 + date: string; 39 + reasoning: string; 40 + name: string | null; 41 + otid: string | null; 42 + sender_id: string | null; 43 + step_id: string | null; 44 + is_err: boolean | null; 45 + seq_id: bigint | null; 46 + run_id: string | null; 47 + source: LettaReasoningSource | null; 48 + signature: string | null; 49 + } 50 + | { 51 + message_type: "hidden_reasoning_message"; 52 + id: string; 53 + date: string; 54 + state: LettaHiddenReasoningState; 55 + name: string | null; 56 + otid: string | null; 57 + sender_id: string | null; 58 + step_id: string | null; 59 + is_err: boolean | null; 60 + seq_id: bigint | null; 61 + run_id: string | null; 62 + hidden_reasoning: string | null; 63 + } 64 + | { 65 + message_type: "tool_call_message"; 66 + id: string; 67 + date: string; 68 + tool_call: LettaToolCall; 69 + name: string | null; 70 + otid: string | null; 71 + sender_id: string | null; 72 + step_id: string | null; 73 + is_err: boolean | null; 74 + seq_id: bigint | null; 75 + run_id: string | null; 76 + } 77 + | { 78 + message_type: "tool_return_message"; 79 + id: string; 80 + date: string; 81 + tool_return: string; 82 + status: LettaToolReturnStatus; 83 + tool_call_id: string; 84 + name: string | null; 85 + otid: string | null; 86 + sender_id: string | null; 87 + step_id: string | null; 88 + is_err: boolean | null; 89 + seq_id: bigint | null; 90 + run_id: string | null; 91 + stdout: Array<string> | null; 92 + stderr: Array<string> | null; 93 + } 94 + | { 95 + message_type: "assistant_message"; 96 + id: string; 97 + date: string; 98 + content: Array<LettaMessageContent>; 99 + name: string | null; 100 + otid: string | null; 101 + sender_id: string | null; 102 + step_id: string | null; 103 + is_err: boolean | null; 104 + seq_id: bigint | null; 105 + run_id: string | null; 106 + } 107 + | { 108 + message_type: "approval_request_message"; 109 + id: string; 110 + date: string; 111 + tool_call: LettaToolCall; 112 + name: string | null; 113 + otid: string | null; 114 + sender_id: string | null; 115 + step_id: string | null; 116 + is_err: boolean | null; 117 + seq_id: bigint | null; 118 + run_id: string | null; 119 + } 120 + | { 121 + message_type: "approval_response_message"; 122 + id: string; 123 + date: string; 124 + approve: boolean; 125 + approval_request_id: string; 126 + name: string | null; 127 + otid: string | null; 128 + sender_id: string | null; 129 + step_id: string | null; 130 + is_err: boolean | null; 131 + seq_id: bigint | null; 132 + run_id: string | null; 133 + reason: string | null; 134 + } 135 + | { message_type: "stop_reason"; stop_reason: string } 136 + | { 137 + message_type: "usage_statistics"; 138 + completion_tokens: bigint; 139 + prompt_tokens: bigint; 140 + step_count: bigint; 141 + };

+60 -5

src/routes/+page.svelte

··· 1 1 <script lang="ts"> 2 - import { Channel, invoke } from "@tauri-apps/api/core"; 2 + import type { LettaCompletionMessage } from "$lib/rust/LettaCompletionMessage"; 3 + import { Channel, invoke } from "@tauri-apps/api/core"; 3 4 import { flip } from "svelte/animate"; 4 5 import { fade, fly } from "svelte/transition"; 5 6 ··· 36 37 } 37 38 ] 38 39 40 + let placeholder = $state(textPlaceholder); 39 41 let isRecording = $state(false); 40 42 let isDrafting = $state(false); 41 - let isCompleting = $state(false) 42 - let placeholder = $state(textPlaceholder); 43 43 let draft = $state(""); 44 + let isCompleting = $state(false) 45 + let agentResponse = $state("") 44 46 let history = $state(new Array<Message>()); 45 47 46 48 async function handleCompose() { ··· 66 68 isRecording = false; 67 69 } 68 70 71 + const onEvent = new Channel<LettaCompletionMessage>() 72 + 73 + let lastType = "" 74 + onEvent.onmessage = (e) => { 75 + if (e.message_type !== lastType && agentResponse) { 76 + let content = agentResponse 77 + 78 + agentResponse = "" 79 + lastType = e.message_type; 80 + 81 + history.push({ 82 + id: crypto.randomUUID(), 83 + from: "letta", 84 + content 85 + }) 86 + } else if (!lastType) { 87 + lastType = e.message_type 88 + } 89 + 90 + switch (e.message_type) { 91 + case "assistant_message": 92 + agentResponse += e.content.map(c => c.type === "text" ? c.text : "").join("") 93 + break; 94 + case "reasoning_message": 95 + agentResponse += e.reasoning 96 + break; 97 + default: 98 + console.log("unsupported message:", e) 99 + } 100 + } 101 + 102 + (async function () { 103 + isCompleting = true; 104 + await invoke("start_llm_completion", { message: draft, onEvent }) 105 + isCompleting = false; 106 + })(); 107 + 69 108 const id = crypto.randomUUID(); 70 - invoke("start_llm_completion", { message: draft }) 71 - 72 109 history.push({ id, from: "user", content: draft }) 73 110 74 111 draft = "" ··· 209 246 </div> 210 247 </div> 211 248 {/each} 249 + 250 + {#if agentResponse && history.length > 0} 251 + <div 252 + in:fly={{x: -20 }} 253 + class={[ 254 + "flex", 255 + history[history.length - 1].from === "letta" ? "mt-1" : "mt-3", 256 + "pr-6 justify-start", 257 + ]}> 258 + <div class={[ 259 + "py-3 px-4 rounded-t-xl leading-snug border border-rose-pine-muted/20", 260 + "bg-rose-pine-base rounded-br-xl" 261 + ]} 262 + > 263 + {agentResponse} 264 + </div> 265 + </div> 266 + {/if} 212 267 213 268 {#if isCompleting} 214 269 <div class="flex">

Configure Feed

Configure Feed