initial commit
This commit is contained in:
syntaxbullet
406
packages/summarizer/index.ts
Normal file
406
packages/summarizer/index.ts
Normal file
@@ -0,0 +1,406 @@
|
||||
import { basename, join } from "node:path";
|
||||
import { listCompanies, summarizeCompany, writeSummary, type LlmCallStatus, type LlmCallTrace } from "./summarizer";
|
||||
import { dataPath } from "../../deployPaths";
|
||||
import { allowedModelFromEnv, isSafeStem, jsonError, validateRequestedModel, withAuth } from "../../security";
|
||||
|
||||
const OUTPUTS_DIR =
|
||||
process.env.BMP_SUMMARIZER_OUTPUTS_DIR ??
|
||||
dataPath(join(import.meta.dir, "outputs"), "summarizer", "outputs");
|
||||
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;
|
||||
|
||||
type JobStatus = "queued" | "running" | "done" | "error";
|
||||
|
||||
interface SummaryJob {
|
||||
id: string;
|
||||
status: JobStatus;
|
||||
stem: string;
|
||||
model: string;
|
||||
total: number;
|
||||
completed: number;
|
||||
current?: string;
|
||||
files: string[];
|
||||
downloads: SummaryDownload[];
|
||||
errors: string[];
|
||||
llmCalls: SafeLlmCallTrace[];
|
||||
usage: UsageSummary;
|
||||
statusMessage?: string;
|
||||
createdAt: string;
|
||||
finishedAt?: string;
|
||||
}
|
||||
|
||||
type SafeLlmCallTrace = Pick<
|
||||
LlmCallTrace,
|
||||
"id" | "operation" | "label" | "model" | "status" | "startedAt" | "finishedAt" | "durationMs" | "error"
|
||||
> & {
|
||||
usage?: unknown;
|
||||
};
|
||||
|
||||
interface SummaryDownload {
|
||||
company: string;
|
||||
label: string;
|
||||
url: string;
|
||||
}
|
||||
|
||||
interface UsageSummary {
|
||||
cost: number;
|
||||
upstreamInferenceCost: number;
|
||||
promptTokens: number;
|
||||
completionTokens: number;
|
||||
totalTokens: number;
|
||||
reasoningTokens: number;
|
||||
cachedTokens: number;
|
||||
calls: number;
|
||||
}
|
||||
|
||||
interface LlmProgress {
|
||||
total: number;
|
||||
running: number;
|
||||
done: number;
|
||||
error: number;
|
||||
latest?: {
|
||||
label: string;
|
||||
operation: LlmCallTrace["operation"];
|
||||
status: LlmCallStatus;
|
||||
startedAt: string;
|
||||
finishedAt?: string;
|
||||
durationMs?: number;
|
||||
};
|
||||
}
|
||||
|
||||
function sanitizeTrace(trace: LlmCallTrace): SafeLlmCallTrace {
|
||||
const response = trace.response as { usage?: unknown } | undefined;
|
||||
return {
|
||||
id: trace.id,
|
||||
operation: trace.operation,
|
||||
label: trace.label,
|
||||
model: trace.model,
|
||||
status: trace.status,
|
||||
startedAt: trace.startedAt,
|
||||
finishedAt: trace.finishedAt,
|
||||
durationMs: trace.durationMs,
|
||||
usage: response?.usage,
|
||||
error: trace.error,
|
||||
};
|
||||
}
|
||||
|
||||
function emptyUsageSummary(): UsageSummary {
|
||||
return {
|
||||
cost: 0,
|
||||
upstreamInferenceCost: 0,
|
||||
promptTokens: 0,
|
||||
completionTokens: 0,
|
||||
totalTokens: 0,
|
||||
reasoningTokens: 0,
|
||||
cachedTokens: 0,
|
||||
calls: 0,
|
||||
};
|
||||
}
|
||||
|
||||
function summarizeUsage(calls: SafeLlmCallTrace[]): UsageSummary {
|
||||
const summary = emptyUsageSummary();
|
||||
for (const call of calls) {
|
||||
const usage = readUsage(call.usage);
|
||||
if (!usage) continue;
|
||||
summary.calls += 1;
|
||||
summary.cost += usage.cost;
|
||||
summary.upstreamInferenceCost += usage.upstreamInferenceCost;
|
||||
summary.promptTokens += usage.promptTokens;
|
||||
summary.completionTokens += usage.completionTokens;
|
||||
summary.totalTokens += usage.totalTokens;
|
||||
summary.reasoningTokens += usage.reasoningTokens;
|
||||
summary.cachedTokens += usage.cachedTokens;
|
||||
}
|
||||
return {
|
||||
...summary,
|
||||
cost: Number(summary.cost.toFixed(8)),
|
||||
upstreamInferenceCost: Number(summary.upstreamInferenceCost.toFixed(8)),
|
||||
};
|
||||
}
|
||||
|
||||
function summarizeLlmProgress(job: SummaryJob): LlmProgress {
|
||||
const calls = job.llmCalls;
|
||||
const progress: LlmProgress = {
|
||||
total: Math.max(1, job.total) * EXPECTED_LLM_CALLS_PER_COMPANY,
|
||||
running: 0,
|
||||
done: 0,
|
||||
error: 0,
|
||||
};
|
||||
|
||||
for (const call of calls) {
|
||||
progress[call.status] += 1;
|
||||
}
|
||||
|
||||
const latest = [...calls].sort((a, b) => {
|
||||
const aTime = Date.parse(a.finishedAt ?? a.startedAt);
|
||||
const bTime = Date.parse(b.finishedAt ?? b.startedAt);
|
||||
return bTime - aTime;
|
||||
})[0];
|
||||
|
||||
if (latest) {
|
||||
progress.latest = {
|
||||
label: latest.label,
|
||||
operation: latest.operation,
|
||||
status: latest.status,
|
||||
startedAt: latest.startedAt,
|
||||
finishedAt: latest.finishedAt,
|
||||
durationMs: latest.durationMs,
|
||||
};
|
||||
}
|
||||
|
||||
return progress;
|
||||
}
|
||||
|
||||
function readUsage(value: unknown): UsageSummary | undefined {
|
||||
if (typeof value !== "object" || value === null) return undefined;
|
||||
const usage = value as Record<string, unknown>;
|
||||
const promptDetails = usage.prompt_tokens_details as Record<string, unknown> | undefined;
|
||||
const completionDetails = usage.completion_tokens_details as Record<string, unknown> | undefined;
|
||||
const costDetails = usage.cost_details as Record<string, unknown> | undefined;
|
||||
return {
|
||||
cost: readNumber(usage.cost ?? usage.total_cost),
|
||||
upstreamInferenceCost: readNumber(costDetails?.upstream_inference_cost ?? costDetails?.total_cost),
|
||||
promptTokens: readNumber(usage.prompt_tokens),
|
||||
completionTokens: readNumber(usage.completion_tokens),
|
||||
totalTokens: readNumber(usage.total_tokens),
|
||||
reasoningTokens: readNumber(completionDetails?.reasoning_tokens),
|
||||
cachedTokens: readNumber(promptDetails?.cached_tokens),
|
||||
calls: 1,
|
||||
};
|
||||
}
|
||||
|
||||
function readNumber(value: unknown): number {
|
||||
if (typeof value === "number" && Number.isFinite(value)) return value;
|
||||
if (typeof value === "string") {
|
||||
const parsed = Number(value);
|
||||
return Number.isFinite(parsed) ? parsed : 0;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
function publicJob(job: SummaryJob) {
|
||||
return {
|
||||
id: job.id,
|
||||
status: job.status,
|
||||
stem: job.stem,
|
||||
model: job.model,
|
||||
total: job.total,
|
||||
completed: job.completed,
|
||||
current: job.current,
|
||||
downloads: job.downloads,
|
||||
errors: job.errors,
|
||||
usage: job.usage,
|
||||
llmProgress: summarizeLlmProgress(job),
|
||||
statusMessage: job.statusMessage,
|
||||
createdAt: job.createdAt,
|
||||
finishedAt: job.finishedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function activeJobCount(): number {
|
||||
return [...jobs.values()].filter((job) => job.status === "queued" || job.status === "running").length;
|
||||
}
|
||||
|
||||
const jobs = new Map<string, SummaryJob>();
|
||||
|
||||
function createJob(stem: string, model: string, total: number): SummaryJob {
|
||||
const job: SummaryJob = {
|
||||
id: crypto.randomUUID(),
|
||||
status: "queued",
|
||||
stem,
|
||||
model,
|
||||
total,
|
||||
completed: 0,
|
||||
files: [],
|
||||
downloads: [],
|
||||
errors: [],
|
||||
llmCalls: [],
|
||||
usage: emptyUsageSummary(),
|
||||
createdAt: new Date().toISOString(),
|
||||
};
|
||||
jobs.set(job.id, job);
|
||||
return job;
|
||||
}
|
||||
|
||||
function labelForOutputFile(file: string): string {
|
||||
if (file.endsWith(".json")) return "JSON herunterladen";
|
||||
if (file.endsWith(".xlsx")) return "XLSX herunterladen";
|
||||
if (file.endsWith(".questions.pdf")) return "Fragen-PDF herunterladen";
|
||||
if (file.endsWith(".questions.html")) return "Fragen-HTML herunterladen";
|
||||
if (file.endsWith(".pdf")) return "PDF herunterladen";
|
||||
if (file.endsWith(".report.html")) return "Report-HTML herunterladen";
|
||||
return basename(file);
|
||||
}
|
||||
|
||||
function buildDownload(company: string, filePath: string): SummaryDownload {
|
||||
const file = basename(filePath);
|
||||
return {
|
||||
company,
|
||||
label: labelForOutputFile(file),
|
||||
url: `/downloads/summarizer/${encodeURIComponent(company)}/${encodeURIComponent(file)}`,
|
||||
};
|
||||
}
|
||||
|
||||
async function runWithConcurrency<T>(
|
||||
items: T[],
|
||||
limit: number,
|
||||
worker: (item: T) => Promise<void>,
|
||||
): Promise<void> {
|
||||
let nextIndex = 0;
|
||||
|
||||
async function runner() {
|
||||
while (nextIndex < items.length) {
|
||||
const index = nextIndex++;
|
||||
await worker(items[index]!);
|
||||
}
|
||||
}
|
||||
|
||||
const count = Math.min(limit, items.length);
|
||||
await Promise.all(Array.from({ length: count }, () => runner()));
|
||||
}
|
||||
|
||||
async function executeJob(job: SummaryJob): Promise<void> {
|
||||
job.status = "running";
|
||||
job.statusMessage = "Auswertung wird vorbereitet.";
|
||||
const companies = job.stem === "__all__" ? await listCompanies() : [job.stem];
|
||||
|
||||
try {
|
||||
await runWithConcurrency(companies, MAX_PARALLEL_SUMMARIES, async (company) => {
|
||||
job.current = company;
|
||||
job.statusMessage = `LLM-Auswertung für ${company} läuft.`;
|
||||
const summary = await summarizeCompany(company, job.model, {
|
||||
onLlmCall: (trace) => {
|
||||
const existingIndex = job.llmCalls.findIndex((call) => call.id === trace.id);
|
||||
const safeTrace = sanitizeTrace(trace);
|
||||
if (existingIndex >= 0) {
|
||||
job.llmCalls[existingIndex] = safeTrace;
|
||||
} else {
|
||||
job.llmCalls.push(safeTrace);
|
||||
}
|
||||
job.usage = summarizeUsage(job.llmCalls);
|
||||
job.statusMessage = trace.status === "running"
|
||||
? `${trace.label} wird verarbeitet.`
|
||||
: `${trace.label} abgeschlossen.`;
|
||||
},
|
||||
});
|
||||
job.statusMessage = `Downloads für ${company} werden vorbereitet.`;
|
||||
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;
|
||||
});
|
||||
|
||||
job.status = "done";
|
||||
job.current = undefined;
|
||||
job.statusMessage = "Auswertung fertig.";
|
||||
job.finishedAt = new Date().toISOString();
|
||||
} catch (error) {
|
||||
job.status = "error";
|
||||
job.errors.push(String(error));
|
||||
job.statusMessage = "Auswertung fehlgeschlagen.";
|
||||
job.finishedAt = new Date().toISOString();
|
||||
}
|
||||
}
|
||||
|
||||
export const routes = {
|
||||
"/summarizer": {
|
||||
GET: withAuth(async () => new Response(Bun.file(PAGE_PATH), {
|
||||
headers: { "Content-Type": "text/html; charset=utf-8" },
|
||||
})),
|
||||
},
|
||||
|
||||
"/api/summarizer/companies": {
|
||||
GET: withAuth(async () => {
|
||||
const companies = await listCompanies();
|
||||
return Response.json({ companies });
|
||||
}),
|
||||
},
|
||||
|
||||
"/api/summarizer/summarize": {
|
||||
POST: withAuth(async (req: Request) => {
|
||||
let body: { stem?: string; model?: string };
|
||||
try {
|
||||
body = await req.json();
|
||||
} catch {
|
||||
return Response.json({ error: "Expected JSON body" }, { status: 400 });
|
||||
}
|
||||
|
||||
const { stem } = body;
|
||||
const model = validateRequestedModel(body.model);
|
||||
|
||||
if (!stem || (stem !== "__all__" && !isSafeStem(stem))) {
|
||||
return Response.json({ error: 'Missing or invalid "stem" field' }, { status: 400 });
|
||||
}
|
||||
if (!model) {
|
||||
return Response.json(
|
||||
{ error: `Model is not allowed for this demo. Use ${allowedModelFromEnv()}.` },
|
||||
{ status: 400 },
|
||||
);
|
||||
}
|
||||
|
||||
if (!process.env.OPENROUTER_API_KEY) {
|
||||
return Response.json(
|
||||
{ error: "OPENROUTER_API_KEY is not set in environment" },
|
||||
{ status: 500 },
|
||||
);
|
||||
}
|
||||
|
||||
if (activeJobCount() >= MAX_ACTIVE_JOBS) {
|
||||
return jsonError(`Es laufen bereits ${MAX_ACTIVE_JOBS} Auswertungen. Bitte warten Sie, bis eine davon fertig ist.`, 429);
|
||||
}
|
||||
|
||||
const companies = await listCompanies();
|
||||
if (stem !== "__all__" && !companies.includes(stem)) {
|
||||
return Response.json({ error: "Company not found" }, { status: 404 });
|
||||
}
|
||||
const total = stem === "__all__" ? (await listCompanies()).length : 1;
|
||||
const job = createJob(stem, model, total);
|
||||
setTimeout(() => {
|
||||
void executeJob(job);
|
||||
}, 0);
|
||||
|
||||
return Response.json({ ok: true, jobId: job.id });
|
||||
}, {
|
||||
csrf: true,
|
||||
limit: { key: "summarize", max: Math.max(20, MAX_ACTIVE_JOBS * 2), windowMs: 60_000 },
|
||||
}),
|
||||
},
|
||||
|
||||
"/api/summarizer/jobs/:id/llm-calls": {
|
||||
GET: withAuth(async (req: Request) => {
|
||||
const id = (req as Request & { params: Record<string, string> }).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 });
|
||||
}
|
||||
return Response.json({
|
||||
jobId: job.id,
|
||||
status: job.status,
|
||||
total: job.total,
|
||||
completed: job.completed,
|
||||
current: job.current,
|
||||
usage: job.usage,
|
||||
llmCalls: job.llmCalls,
|
||||
});
|
||||
}),
|
||||
},
|
||||
|
||||
"/api/summarizer/jobs/:id": {
|
||||
GET: withAuth(async (req: Request) => {
|
||||
const id = (req as Request & { params: Record<string, string> }).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 });
|
||||
}
|
||||
return Response.json(publicJob(job));
|
||||
}),
|
||||
},
|
||||
} as const;
|
||||
Reference in New Issue
Block a user