diff --git a/packages/summarizer/index.ts b/packages/summarizer/index.ts index 40530ff..d502f8f 100644 --- a/packages/summarizer/index.ts +++ b/packages/summarizer/index.ts @@ -1,4 +1,5 @@ import { basename, join } from "node:path"; +import { mkdir } from "node:fs/promises"; import { listCompanies, summarizeCompany, writeSummary, type LlmCallStatus, type LlmCallTrace } from "./summarizer"; import { dataPath } from "../../deployPaths"; import { allowedModelFromEnv, isSafeStem, jsonError, validateRequestedModel, withAuth } from "../../security"; @@ -10,6 +11,9 @@ const PAGE_PATH = join(import.meta.dir, "index.html"); const MAX_PARALLEL_SUMMARIES = 2; const MAX_ACTIVE_JOBS = Number(process.env.BMP_MAX_ACTIVE_SUMMARY_JOBS ?? 10); const EXPECTED_LLM_CALLS_PER_COMPANY = 13; +const OBSERVABILITY_DIR = + process.env.BMP_OBSERVABILITY_DIR ?? + dataPath(join(import.meta.dir, "outputs", "_observability"), "summarizer", "outputs", "_observability"); type JobStatus = "queued" | "running" | "done" | "error" | "cancelled"; @@ -25,6 +29,8 @@ interface SummaryJob { downloads: SummaryDownload[]; errors: string[]; llmCalls: SafeLlmCallTrace[]; + observability: ObservabilityLog; + observabilityWrite: Promise; usage: UsageSummary; statusMessage?: string; createdAt: string; @@ -35,9 +41,47 @@ type SafeLlmCallTrace = Pick< LlmCallTrace, "id" | "operation" | "label" | "model" | "status" | "startedAt" | "finishedAt" | "durationMs" | "error" > & { + attempt: number; + maxAttempts: number; usage?: unknown; }; +interface ObservabilityEvent { + at: string; + type: string; + message?: string; + data?: unknown; +} + +interface ObservabilityLog { + version: 1; + jobId: string; + stem: string; + model: string; + status: JobStatus; + createdAt: string; + finishedAt?: string; + total: number; + completed: number; + current?: string; + companies: string[]; + files: string[]; + downloads: SummaryDownload[]; + errors: SerializedError[]; + inputs: Record; + outputs: Record; + llmCalls: LlmCallTrace[]; + events: ObservabilityEvent[]; + logPath: string; +} + +interface SerializedError { + name?: string; + message: string; + stack?: string; + cause?: unknown; +} + interface SummaryDownload { company: string; label: string; @@ -77,6 +121,8 @@ function sanitizeTrace(trace: LlmCallTrace): SafeLlmCallTrace { operation: trace.operation, label: trace.label, model: trace.model, + attempt: trace.attempt, + maxAttempts: trace.maxAttempts, status: trace.status, startedAt: trace.startedAt, finishedAt: trace.finishedAt, @@ -209,9 +255,83 @@ function isCancelled(job: SummaryJob): boolean { const jobs = new Map(); +function serializeError(error: unknown): SerializedError { + if (error instanceof Error) { + return { + name: error.name, + message: error.message, + stack: error.stack, + cause: error.cause, + }; + } + return { message: String(error) }; +} + +function createObservabilityLog(jobId: string, stem: string, model: string, total: number, createdAt: string): ObservabilityLog { + return { + version: 1, + jobId, + stem, + model, + status: "queued", + createdAt, + total, + completed: 0, + companies: [], + files: [], + downloads: [], + errors: [], + inputs: {}, + outputs: {}, + llmCalls: [], + events: [ + { + at: createdAt, + type: "job.created", + message: "Summary job was queued.", + }, + ], + logPath: join(OBSERVABILITY_DIR, `${createdAt.replaceAll(":", "-")}-${jobId}.json`), + }; +} + +function recordEvent(job: SummaryJob, type: string, message?: string, data?: unknown): void { + job.observability.events.push({ + at: new Date().toISOString(), + type, + message, + data, + }); +} + +function syncObservabilityState(job: SummaryJob): void { + job.observability.status = job.status; + job.observability.finishedAt = job.finishedAt; + job.observability.completed = job.completed; + job.observability.current = job.current; + job.observability.files = [...job.files]; + job.observability.downloads = [...job.downloads]; +} + +function persistObservability(job: SummaryJob): Promise { + syncObservabilityState(job); + job.observabilityWrite = job.observabilityWrite + .catch(() => undefined) + .then(async () => { + await mkdir(OBSERVABILITY_DIR, { recursive: true }); + await Bun.write(job.observability.logPath, JSON.stringify(job.observability, null, 2)); + }) + .catch((error) => { + console.error("Failed to write summarizer observability log:", error); + }); + return job.observabilityWrite; +} + function createJob(stem: string, model: string, total: number): SummaryJob { + const id = crypto.randomUUID(); + const createdAt = new Date().toISOString(); const job: SummaryJob = { - id: crypto.randomUUID(), + id, status: "queued", stem, model, @@ -221,10 +341,13 @@ function createJob(stem: string, model: string, total: number): SummaryJob { downloads: [], errors: [], llmCalls: [], + observability: createObservabilityLog(id, stem, model, total, createdAt), + observabilityWrite: Promise.resolve(), usage: emptyUsageSummary(), - createdAt: new Date().toISOString(), + createdAt, }; jobs.set(job.id, job); + void persistObservability(job); return job; } @@ -269,14 +392,31 @@ async function executeJob(job: SummaryJob): Promise { if (isCancelled(job)) return; job.status = "running"; job.statusMessage = "Auswertung wird vorbereitet."; + recordEvent(job, "job.started", job.statusMessage); + await persistObservability(job); const companies = job.stem === "__all__" ? await listCompanies() : [job.stem]; + job.observability.companies = [...companies]; + recordEvent(job, "companies.selected", undefined, { companies }); + await persistObservability(job); try { await runWithConcurrency(companies, MAX_PARALLEL_SUMMARIES, async (company) => { if (isCancelled(job)) return; job.current = company; job.statusMessage = `LLM-Auswertung für ${company} läuft.`; + recordEvent(job, "company.started", job.statusMessage, { company }); + await persistObservability(job); const summary = await summarizeCompany(company, job.model, { + onCompanyInput: (companyStem, input) => { + job.observability.inputs[companyStem] = input; + recordEvent(job, "company.input.loaded", undefined, { company: companyStem }); + void persistObservability(job); + }, + onCompanyOutput: (companyStem, output) => { + job.observability.outputs[companyStem] = output; + recordEvent(job, "company.output.created", undefined, { company: companyStem }); + void persistObservability(job); + }, onLlmCall: (trace) => { const existingIndex = job.llmCalls.findIndex((call) => call.id === trace.id); const safeTrace = sanitizeTrace(trace); @@ -285,24 +425,48 @@ async function executeJob(job: SummaryJob): Promise { } else { job.llmCalls.push(safeTrace); } + const fullTraceIndex = job.observability.llmCalls.findIndex((call) => call.id === trace.id); + if (fullTraceIndex >= 0) { + job.observability.llmCalls[fullTraceIndex] = trace; + } else { + job.observability.llmCalls.push(trace); + } job.usage = summarizeUsage(job.llmCalls); job.statusMessage = trace.status === "running" - ? `${trace.label} wird verarbeitet.` - : `${trace.label} abgeschlossen.`; + ? `${trace.label} wird verarbeitet (Versuch ${trace.attempt}/${trace.maxAttempts}).` + : trace.status === "error" + ? `${trace.label} fehlgeschlagen (Versuch ${trace.attempt}/${trace.maxAttempts}).` + : `${trace.label} abgeschlossen.`; + recordEvent(job, `llm.${trace.status}`, job.statusMessage, { + id: trace.id, + operation: trace.operation, + label: trace.label, + attempt: trace.attempt, + maxAttempts: trace.maxAttempts, + }); + void persistObservability(job); }, }); job.statusMessage = `Downloads für ${company} werden vorbereitet.`; + recordEvent(job, "company.summary.created", job.statusMessage, { company }); if (isCancelled(job)) return; const outputPaths = await writeSummary(company, summary, OUTPUTS_DIR); job.files.push(...outputPaths.map((path) => basename(path))); job.downloads.push(...outputPaths.map((path) => buildDownload(company, path))); job.completed += 1; + recordEvent(job, "company.finished", `Auswertung für ${company} fertig.`, { + company, + outputFiles: outputPaths.map((path) => basename(path)), + }); + await persistObservability(job); }); if (isCancelled(job)) { job.current = undefined; job.statusMessage = "Auswertung abgebrochen."; job.finishedAt = new Date().toISOString(); + recordEvent(job, "job.cancelled", job.statusMessage); + await persistObservability(job); return; } @@ -310,17 +474,25 @@ async function executeJob(job: SummaryJob): Promise { job.current = undefined; job.statusMessage = "Auswertung fertig."; job.finishedAt = new Date().toISOString(); + recordEvent(job, "job.done", job.statusMessage); + await persistObservability(job); } catch (error) { if (isCancelled(job)) { job.current = undefined; job.statusMessage = "Auswertung abgebrochen."; job.finishedAt = new Date().toISOString(); + recordEvent(job, "job.cancelled", job.statusMessage); + await persistObservability(job); return; } + const serializedError = serializeError(error); job.status = "error"; - job.errors.push(String(error)); - job.statusMessage = "Auswertung fehlgeschlagen."; + job.errors.push(serializedError.message); + job.observability.errors.push(serializedError); + job.statusMessage = `Auswertung fehlgeschlagen: ${serializedError.message}`; job.finishedAt = new Date().toISOString(); + recordEvent(job, "job.error", job.statusMessage, serializedError); + await persistObservability(job); } } @@ -432,7 +604,10 @@ export const routes = { return Response.json(publicJob(job)); }), DELETE: withAuth(async (req: Request) => { - const id = (req as Request & { params: Record }).params.id; + const id = (req as Request & { params: Record }).params.id; + if (!id) { + return Response.json({ error: "Job not found" }, { status: 404 }); + } const job = jobs.get(id); if (!job) { return Response.json({ error: "Job not found" }, { status: 404 }); diff --git a/packages/summarizer/summarizer.ts b/packages/summarizer/summarizer.ts index 1363457..4f1072c 100644 --- a/packages/summarizer/summarizer.ts +++ b/packages/summarizer/summarizer.ts @@ -90,6 +90,8 @@ export interface LlmCallTrace { operation: "segmentierung" | "zusammenfassung" | "scoring" | "swot"; label: string; model: string; + attempt: number; + maxAttempts: number; status: LlmCallStatus; startedAt: string; finishedAt?: string; @@ -101,6 +103,8 @@ export interface LlmCallTrace { export interface SummarizeCompanyOptions { onLlmCall?: (trace: LlmCallTrace) => void; + onCompanyInput?: (stem: string, input: ExtractedData) => void; + onCompanyOutput?: (stem: string, output: SummaryData) => void; } const OUTPUTS_DIR = @@ -111,6 +115,8 @@ const OPENROUTER_PRIVACY_PROVIDER = { data_collection: "deny", zdr: true, } as const; +const MAX_LLM_RETRY_ATTEMPTS = Math.max(1, Number(process.env.BMP_LLM_RETRY_ATTEMPTS ?? 3)); +const LLM_RETRY_BASE_DELAY_MS = Math.max(0, Number(process.env.BMP_LLM_RETRY_BASE_DELAY_MS ?? 750)); function createClient(): OpenAI { const apiKey = process.env.OPENROUTER_API_KEY; @@ -169,12 +175,16 @@ function beginLlmTrace( label: string, model: string, request: Record, + attempt: number, + maxAttempts: number, ): LlmCallTrace { const trace: LlmCallTrace = { id: crypto.randomUUID(), operation, label, model, + attempt, + maxAttempts, status: "running", startedAt: new Date().toISOString(), request, @@ -200,6 +210,66 @@ function finishLlmTrace( }); } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function retryDelayMs(attempt: number): number { + if (LLM_RETRY_BASE_DELAY_MS <= 0) return 0; + const jitter = Math.floor(Math.random() * Math.min(250, LLM_RETRY_BASE_DELAY_MS)); + return LLM_RETRY_BASE_DELAY_MS * 2 ** Math.max(0, attempt - 1) + jitter; +} + +function shouldRetryLlmError(error: unknown): boolean { + const status = (error as { status?: unknown; code?: unknown })?.status; + if (typeof status === "number") { + return status === 408 || status === 409 || status === 425 || status === 429 || status >= 500; + } + const message = String(error).toLowerCase(); + return [ + "timeout", + "timed out", + "rate limit", + "temporarily", + "econnreset", + "fetch failed", + "socket", + "overloaded", + "provider returned error", + ].some((needle) => message.includes(needle)); +} + +async function createCompletionWithRetry( + client: OpenAI, + request: Record, + trace: { + operation: LlmCallTrace["operation"]; + label: string; + model: string; + onLlmCall?: SummarizeCompanyOptions["onLlmCall"]; + } | undefined, +) { + let lastError: unknown; + for (let attempt = 1; attempt <= MAX_LLM_RETRY_ATTEMPTS; attempt += 1) { + const llmTrace = trace + ? beginLlmTrace(trace.onLlmCall, trace.operation, trace.label, trace.model, request, attempt, MAX_LLM_RETRY_ATTEMPTS) + : undefined; + try { + const completion = await client.chat.completions.create(request as never); + if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { response: completion }); + return completion; + } catch (error) { + lastError = error; + if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { error }); + if (attempt >= MAX_LLM_RETRY_ATTEMPTS || !shouldRetryLlmError(error)) { + throw error; + } + await sleep(retryDelayMs(attempt)); + } + } + throw lastError; +} + async function createStructuredCompletion( client: OpenAI, model: string, @@ -222,19 +292,16 @@ async function createStructuredCompletion( session_id: sessionId, }; const primaryTrace = trace - ? beginLlmTrace(trace.onLlmCall, trace.operation, trace.label, model, structuredRequest as Record) + ? { ...trace, model } : undefined; try { - const completion = await client.chat.completions.create(structuredRequest as never); - if (primaryTrace) finishLlmTrace(trace?.onLlmCall, primaryTrace, { response: completion }); + const completion = await createCompletionWithRetry(client, structuredRequest as Record, primaryTrace); return completion; } catch (error) { const message = String(error); if (!message.includes("400") && !message.toLowerCase().includes("provider returned error")) { - if (primaryTrace) finishLlmTrace(trace?.onLlmCall, primaryTrace, { error }); throw error; } - if (primaryTrace) finishLlmTrace(trace?.onLlmCall, primaryTrace, { error }); const fallbackRequest = { model, @@ -245,16 +312,9 @@ async function createStructuredCompletion( session_id: sessionId, }; const fallbackTrace = trace - ? beginLlmTrace(trace.onLlmCall, trace.operation, `${trace.label} (JSON fallback)`, model, fallbackRequest as Record) + ? { ...trace, label: `${trace.label} (JSON fallback)`, model } : undefined; - try { - const completion = await client.chat.completions.create(fallbackRequest as never); - if (fallbackTrace) finishLlmTrace(trace?.onLlmCall, fallbackTrace, { response: completion }); - return completion; - } catch (fallbackError) { - if (fallbackTrace) finishLlmTrace(trace?.onLlmCall, fallbackTrace, { error: fallbackError }); - throw fallbackError; - } + return createCompletionWithRetry(client, fallbackRequest as Record, fallbackTrace); } } @@ -287,16 +347,9 @@ async function summarizeText( session_id: sessionId, }; const llmTrace = trace - ? beginLlmTrace(trace.onLlmCall, "zusammenfassung", trace.label, model, request as Record) + ? { operation: "zusammenfassung" as const, label: trace.label, model, onLlmCall: trace.onLlmCall } : undefined; - let completion: Awaited>; - try { - completion = await client.chat.completions.create(request as never); - if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { response: completion }); - } catch (error) { - if (llmTrace) finishLlmTrace(trace?.onLlmCall, llmTrace, { error }); - throw error; - } + const completion = await createCompletionWithRetry(client, request as Record, llmTrace); const msg = completion.choices[0]?.message as Record | undefined; const content = (msg?.content ?? msg?.reasoning) as string | null | undefined; @@ -1142,6 +1195,7 @@ export async function summarizeCompany( } const data: ExtractedData = await file.json(); + options.onCompanyInput?.(stem, data); const client = createClient(); const output: SummaryData = structuredClone(data); const privacy: Pseudonymizer = createPseudonymizer(data); @@ -1292,6 +1346,7 @@ export async function summarizeCompany( output._privacy = privacy.audit(); output._schemaVersion = 3; output._summarizedAt = new Date().toISOString(); + options.onCompanyOutput?.(stem, output); return output; }