feat: Add analyze module for LLM integration
This commit introduces the `analyze` module, which orchestrates communication with Large Language Models (LLMs) for text summarization and analysis. Key components include: - `llm.rs`: Contains the `LlmError` enum and the `chat_once` function for interacting with OpenAI-compatible LLM APIs. - `prompt.rs`: Defines the system prompt and logic for rendering user content from analysis inputs. - `recovery.rs`: Implements logic to scan for and re-enqueue pending analysis jobs upon server startup. - `worker.rs`: The core worker that consumes analysis jobs, processes them with the LLM, and writes the output. - `mod.rs`: Defines `AnalyzeJob` and associated channel types for inter-component communication. This module enables the server to process dictated recordings, summarize them using an LLM, and store the results.
This commit is contained in:
@@ -0,0 +1,126 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::debug;
|
||||
|
||||
/// Error variants returned by the LLM client (OpenAI-compatible Chat
|
||||
/// Completions API — Ionos, Azure OpenAI, OpenAI proper, Together.ai, etc.).
|
||||
///
|
||||
/// **Security:** `Display` and `Debug` MUST NOT contain the HTTP response body.
|
||||
/// Provider error bodies can echo back details that include the API key or
|
||||
/// other sensitive context (e.g. `{"error": "invalid api_key sk-…"}`). Only
|
||||
/// the status and a short category string are safe to log.
|
||||
#[derive(Debug)]
|
||||
pub enum LlmError {
|
||||
/// Transport-level failure (DNS, TLS, connect, timeout). `reqwest::Error`
|
||||
/// is intentionally included — its `Display` only reveals URL/method,
|
||||
/// not request headers, so the API key stays out of logs.
|
||||
Http(reqwest::Error),
|
||||
/// Provider returned a non-2xx status. Body is deliberately discarded.
|
||||
Status { status: u16 },
|
||||
/// Response was 2xx but the expected `choices[0].message.content`
|
||||
/// shape was absent or empty.
|
||||
Parse(&'static str),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for LlmError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Http(e) => write!(f, "llm http error: {e}"),
|
||||
Self::Status { status } => write!(f, "llm returned status {status}"),
|
||||
Self::Parse(msg) => write!(f, "llm response parse error: {msg}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for LlmError {}
|
||||
|
||||
/// Grouped provider configuration. Passed by reference to `chat_once`.
|
||||
pub struct LlmSettings<'a> {
|
||||
pub base_url: &'a str,
|
||||
pub api_key: &'a str,
|
||||
pub model: &'a str,
|
||||
pub temperature: f32,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ChatRequest<'a> {
|
||||
model: &'a str,
|
||||
temperature: f32,
|
||||
stream: bool,
|
||||
messages: Vec<Message<'a>>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct Message<'a> {
|
||||
role: &'a str,
|
||||
content: &'a str,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ChatResponse {
|
||||
choices: Vec<Choice>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct Choice {
|
||||
message: ResponseMessage,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ResponseMessage {
|
||||
content: Option<String>,
|
||||
}
|
||||
|
||||
/// Single-shot chat completion against an OpenAI-compatible endpoint.
|
||||
/// The `api_key` in `settings` is used as a Bearer token; do not log it.
|
||||
pub async fn chat_once(
|
||||
client: &reqwest::Client,
|
||||
settings: &LlmSettings<'_>,
|
||||
system_prompt: &str,
|
||||
user_content: &str,
|
||||
timeout: Duration,
|
||||
) -> Result<String, LlmError> {
|
||||
let url = format!("{}/v1/chat/completions", settings.base_url.trim_end_matches('/'));
|
||||
debug!(%url, model = settings.model, "calling llm chat completions");
|
||||
|
||||
let body = ChatRequest {
|
||||
model: settings.model,
|
||||
temperature: settings.temperature,
|
||||
stream: false,
|
||||
messages: vec![
|
||||
Message { role: "system", content: system_prompt },
|
||||
Message { role: "user", content: user_content },
|
||||
],
|
||||
};
|
||||
|
||||
let response = client
|
||||
.post(&url)
|
||||
.bearer_auth(settings.api_key)
|
||||
.timeout(timeout)
|
||||
.json(&body)
|
||||
.send()
|
||||
.await
|
||||
.map_err(LlmError::Http)?;
|
||||
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
// Body deliberately dropped — see LlmError docstring.
|
||||
drop(response);
|
||||
return Err(LlmError::Status { status: status.as_u16() });
|
||||
}
|
||||
|
||||
let parsed: ChatResponse = response.json().await.map_err(LlmError::Http)?;
|
||||
let content = parsed
|
||||
.choices
|
||||
.into_iter()
|
||||
.next()
|
||||
.and_then(|c| c.message.content)
|
||||
.ok_or(LlmError::Parse("missing choices[0].message.content"))?;
|
||||
|
||||
let trimmed = content.trim().to_string();
|
||||
if trimmed.is_empty() {
|
||||
return Err(LlmError::Parse("empty content after trim"));
|
||||
}
|
||||
Ok(trimmed)
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
use std::path::PathBuf;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
pub mod llm;
|
||||
pub mod prompt;
|
||||
pub mod recovery;
|
||||
pub mod worker;
|
||||
|
||||
/// A single request to analyze a case. Payload is tiny (path + u32), so the
|
||||
/// channel is unbounded — persistence lives in `analysis_input_v{N}.json` on
|
||||
/// disk, not in the queue. Recovery re-enqueues any inputs the server may
|
||||
/// have held at crash time.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AnalyzeJob {
|
||||
pub case_dir: PathBuf,
|
||||
pub version: u32,
|
||||
}
|
||||
|
||||
pub type AnalyzeSender = mpsc::UnboundedSender<AnalyzeJob>;
|
||||
pub type AnalyzeReceiver = mpsc::UnboundedReceiver<AnalyzeJob>;
|
||||
|
||||
pub fn channel() -> (AnalyzeSender, AnalyzeReceiver) {
|
||||
mpsc::unbounded_channel()
|
||||
}
|
||||
|
||||
/// Persistent input for a single analysis run. Written by the close-case
|
||||
/// handler, consumed by the analyze worker, and left on disk indefinitely as
|
||||
/// an audit trail of what was sent to the LLM.
|
||||
///
|
||||
/// Timestamps are stored as ISO-8601 strings (lexicographic order matches
|
||||
/// chronological order, so string comparison is sufficient for the
|
||||
/// "Nachtrag" detection the UI needs later).
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct AnalysisInput {
|
||||
pub version: u32,
|
||||
/// Latest filesystem mtime (server clock) of any `.m4a` in the case at
|
||||
/// the moment the close button was clicked. Used later to detect late
|
||||
/// arrivals that should trigger a re-analysis prompt.
|
||||
pub last_recording_mtime: String,
|
||||
pub recordings: Vec<RecordingInput>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct RecordingInput {
|
||||
/// Recording timestamp from the watch (filename-derived, ISO-8601).
|
||||
pub recorded_at: String,
|
||||
pub text: String,
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
use super::AnalysisInput;
|
||||
|
||||
/// System prompt for the consolidation LLM. From docs/projektplan.md:365-369.
|
||||
/// Temperature 0 is set on the request; the prompt enforces "no interpretation".
|
||||
pub const SYSTEM_PROMPT: &str = "Du bist ein medizinischer Assistent. Fasse die folgenden diktierten Abschnitte zu einem zusammenhängenden Text zusammen. Bereinigen und strukturieren, nichts hinzufügen, nichts interpretieren, keine Diagnose, keine Arztbrief-Struktur. Entferne Redundanzen. Spätere Aufnahmen haben Vorrang — Korrekturen, Nachträge und Widersprüche zugunsten der chronologisch letzten Aussage auflösen. Antworte auf Deutsch. Nur der zusammengefasste Text, keine Einleitung, kein Abschlusssatz.";
|
||||
|
||||
/// Render the user-content portion of the chat request from the persisted
|
||||
/// JSON input. Recordings with blank text are skipped — they carry no signal
|
||||
/// for the LLM and only add noise. Chronological order is taken as given;
|
||||
/// callers must sort `recordings` before passing in (the handler does).
|
||||
pub fn render_prompt(input: &AnalysisInput) -> String {
|
||||
let mut out = String::new();
|
||||
let mut first = true;
|
||||
for r in &input.recordings {
|
||||
if r.text.trim().is_empty() {
|
||||
continue;
|
||||
}
|
||||
if !first {
|
||||
out.push_str("\n\n---\n\n");
|
||||
}
|
||||
first = false;
|
||||
out.push_str("## ");
|
||||
out.push_str(&r.recorded_at);
|
||||
out.push_str("\n\n");
|
||||
out.push_str(r.text.trim());
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::analyze::RecordingInput;
|
||||
|
||||
fn mk(input: Vec<(&str, &str)>) -> AnalysisInput {
|
||||
AnalysisInput {
|
||||
version: 1,
|
||||
last_recording_mtime: "2026-04-15T10:47:00Z".into(),
|
||||
recordings: input
|
||||
.into_iter()
|
||||
.map(|(ts, text)| RecordingInput {
|
||||
recorded_at: ts.into(),
|
||||
text: text.into(),
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn renders_single_recording() {
|
||||
let out = render_prompt(&mk(vec![("2026-04-15T10:32:00Z", "Patient klagt über Knie.")]));
|
||||
assert_eq!(out, "## 2026-04-15T10:32:00Z\n\nPatient klagt über Knie.");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn renders_multiple_with_separator() {
|
||||
let out = render_prompt(&mk(vec![
|
||||
("2026-04-15T10:32:00Z", "Erste Aufnahme."),
|
||||
("2026-04-15T10:38:15Z", "Zweite Aufnahme."),
|
||||
]));
|
||||
assert!(out.contains("\n\n---\n\n"));
|
||||
assert!(out.starts_with("## 2026-04-15T10:32:00Z"));
|
||||
assert!(out.contains("## 2026-04-15T10:38:15Z"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn skips_blank_recordings() {
|
||||
let out = render_prompt(&mk(vec![
|
||||
("2026-04-15T10:32:00Z", "Nur Inhalt."),
|
||||
("2026-04-15T10:33:00Z", " "),
|
||||
("2026-04-15T10:34:00Z", ""),
|
||||
]));
|
||||
assert_eq!(out.matches("---").count(), 0);
|
||||
assert!(!out.contains("10:33:00"));
|
||||
assert!(!out.contains("10:34:00"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn handles_empty_input() {
|
||||
assert_eq!(render_prompt(&mk(vec![])), "");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trims_per_recording_text() {
|
||||
let out = render_prompt(&mk(vec![("2026-04-15T10:32:00Z", " hallo\n\n")]));
|
||||
assert!(out.ends_with("hallo"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use tracing::{info, warn};
|
||||
|
||||
use super::{AnalyzeJob, AnalyzeSender};
|
||||
|
||||
/// Walk `data_path/*/open/*/analysis_input_v*.json` and enqueue every input
|
||||
/// that does not yet have a matching `document_v*.md` sibling.
|
||||
///
|
||||
/// Intended to run once at startup. Send is synchronous on the unbounded
|
||||
/// channel — only fails if the worker is gone, which is a programming error.
|
||||
pub async fn scan_and_enqueue(data_path: &Path, tx: &AnalyzeSender) {
|
||||
let pending = collect_pending(data_path).await;
|
||||
let total = pending.len();
|
||||
info!(count = total, "Analyze recovery scan found pending inputs");
|
||||
|
||||
for (case_dir, version) in pending {
|
||||
if tx.send(AnalyzeJob { case_dir: case_dir.clone(), version }).is_err() {
|
||||
warn!(case = %case_dir.display(), "Recovery send failed (worker gone)");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns `(case_dir, version)` tuples for every `analysis_input_v{N}.json`
|
||||
/// whose `document_v{N}.md` is missing.
|
||||
async fn collect_pending(data_path: &Path) -> Vec<(PathBuf, u32)> {
|
||||
let mut out: Vec<(PathBuf, u32)> = Vec::new();
|
||||
|
||||
let mut users = match tokio::fs::read_dir(data_path).await {
|
||||
Ok(r) => r,
|
||||
Err(_) => return out,
|
||||
};
|
||||
|
||||
while let Ok(Some(user_entry)) = users.next_entry().await {
|
||||
let open_dir = user_entry.path().join("open");
|
||||
let mut cases = match tokio::fs::read_dir(&open_dir).await {
|
||||
Ok(r) => r,
|
||||
Err(_) => continue,
|
||||
};
|
||||
|
||||
while let Ok(Some(case_entry)) = cases.next_entry().await {
|
||||
let case_dir = case_entry.path();
|
||||
if !case_dir.is_dir() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let mut files = match tokio::fs::read_dir(&case_dir).await {
|
||||
Ok(r) => r,
|
||||
Err(_) => continue,
|
||||
};
|
||||
|
||||
while let Ok(Some(file)) = files.next_entry().await {
|
||||
let path = file.path();
|
||||
let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
|
||||
continue;
|
||||
};
|
||||
let Some(version) = parse_input_version(name) else {
|
||||
continue;
|
||||
};
|
||||
let document = case_dir.join(format!("document_v{version}.md"));
|
||||
if !document.exists() {
|
||||
out.push((case_dir.clone(), version));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Oldest case directories first — not strictly required for correctness,
|
||||
// but gives deterministic recovery order. Cases are UUIDs so sorting is
|
||||
// purely lexicographic, which is fine.
|
||||
out.sort();
|
||||
out
|
||||
}
|
||||
|
||||
/// Parse `analysis_input_v{N}.json` and return `N` (as u32) if the pattern
|
||||
/// matches exactly. Rejects malformed names, leading zeros, and overflow.
|
||||
fn parse_input_version(filename: &str) -> Option<u32> {
|
||||
let rest = filename.strip_prefix("analysis_input_v")?;
|
||||
let num = rest.strip_suffix(".json")?;
|
||||
if num.is_empty() || (num.starts_with('0') && num.len() > 1) {
|
||||
return None;
|
||||
}
|
||||
num.parse::<u32>().ok()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_valid_names() {
|
||||
assert_eq!(parse_input_version("analysis_input_v1.json"), Some(1));
|
||||
assert_eq!(parse_input_version("analysis_input_v42.json"), Some(42));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_invalid_names() {
|
||||
assert_eq!(parse_input_version("analysis_input_v.json"), None);
|
||||
assert_eq!(parse_input_version("analysis_input_v01.json"), None);
|
||||
assert_eq!(parse_input_version("analysis_input_vx.json"), None);
|
||||
assert_eq!(parse_input_version("document_v1.md"), None);
|
||||
assert_eq!(parse_input_version("analysis_input_v1.txt"), None);
|
||||
assert_eq!(parse_input_version(""), None);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
use super::{llm, prompt, AnalysisInput, AnalyzeReceiver};
|
||||
use crate::config::Config;
|
||||
|
||||
/// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
|
||||
/// but sequential processing keeps the architecture simple and shields the
|
||||
/// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
|
||||
pub async fn run(mut rx: AnalyzeReceiver, config: Arc<Config>, client: reqwest::Client) {
|
||||
info!("Analyze worker started");
|
||||
let timeout = Duration::from_secs(config.llm_timeout_seconds);
|
||||
|
||||
while let Some(job) = rx.recv().await {
|
||||
process(&job.case_dir, job.version, &config, &client, timeout).await;
|
||||
}
|
||||
|
||||
warn!("Analyze worker stopped (channel closed)");
|
||||
}
|
||||
|
||||
async fn process(
|
||||
case_dir: &Path,
|
||||
version: u32,
|
||||
config: &Config,
|
||||
client: &reqwest::Client,
|
||||
timeout: Duration,
|
||||
) {
|
||||
let input_path = case_dir.join(format!("analysis_input_v{version}.json"));
|
||||
let document_path = case_dir.join(format!("document_v{version}.md"));
|
||||
|
||||
// If the document already exists, the job is obsolete (e.g. recovery
|
||||
// re-enqueued a finished case during a race). Skip silently.
|
||||
if document_path.exists() {
|
||||
return;
|
||||
}
|
||||
|
||||
let input: AnalysisInput = match tokio::fs::read(&input_path).await {
|
||||
Ok(bytes) => match serde_json::from_slice(&bytes) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
error!(path = %input_path.display(), error = %e, "analysis input parse failed");
|
||||
return;
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
error!(path = %input_path.display(), error = %e, "analysis input read failed");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let user_content = prompt::render_prompt(&input);
|
||||
if user_content.is_empty() {
|
||||
error!(case = %case_dir.display(), "analysis input has no usable recordings");
|
||||
return;
|
||||
}
|
||||
|
||||
let total_chars = user_content.chars().count();
|
||||
info!(
|
||||
case = %case_dir.display(),
|
||||
version,
|
||||
recording_count = input.recordings.len(),
|
||||
total_chars,
|
||||
"sending to llm"
|
||||
);
|
||||
|
||||
// NOTE: If the server crashes between a successful LLM response and the
|
||||
// document write below, the recovery scan will re-enqueue this job and
|
||||
// the call will be made again. Accepted cost — rare event, small
|
||||
// per-call price. Response-caching in a `.tmp` file would prevent it.
|
||||
let settings = llm::LlmSettings {
|
||||
base_url: &config.llm_url,
|
||||
api_key: &config.llm_api_key,
|
||||
model: &config.llm_model,
|
||||
temperature: config.llm_temperature,
|
||||
};
|
||||
let document = match llm::chat_once(
|
||||
client,
|
||||
&settings,
|
||||
prompt::SYSTEM_PROMPT,
|
||||
&user_content,
|
||||
timeout,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(text) => text,
|
||||
Err(e) => {
|
||||
// Do not log the response body — LlmError::Display is already
|
||||
// redacted, but reinforce the rule here for future readers.
|
||||
error!(case = %case_dir.display(), version, error = %e, "llm call failed");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = write_atomic(&document_path, document.as_bytes()).await {
|
||||
error!(path = %document_path.display(), error = %e, "writing document failed");
|
||||
return;
|
||||
}
|
||||
|
||||
info!(
|
||||
case = %case_dir.display(),
|
||||
version,
|
||||
bytes = document.len(),
|
||||
"analysis done"
|
||||
);
|
||||
}
|
||||
|
||||
/// Write `bytes` to `path` atomically: write to `<path>.tmp`, fsync, rename.
|
||||
/// A crash between write and rename leaves no half-written target file — the
|
||||
/// state machine simply sees the job as still pending and retries.
|
||||
async fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
|
||||
let tmp = path.with_extension({
|
||||
let mut ext = path.extension().map(|s| s.to_os_string()).unwrap_or_default();
|
||||
ext.push(".tmp");
|
||||
ext
|
||||
});
|
||||
|
||||
let mut file = tokio::fs::File::create(&tmp).await?;
|
||||
file.write_all(bytes).await?;
|
||||
file.sync_all().await?;
|
||||
drop(file);
|
||||
tokio::fs::rename(&tmp, path).await?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
pub mod analyze;
|
||||
pub mod auth;
|
||||
pub mod config;
|
||||
pub mod error;
|
||||
|
||||
Reference in New Issue
Block a user