From fe1e3065d60629ceacadb67ee041156ba83e8ade Mon Sep 17 00:00:00 2001 From: czj <13261895355@163.com> Date: Thu, 16 Jul 2026 12:09:09 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9ELLM=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E8=BF=BD=E8=B8=AA=E6=94=B6=E9=9B=86=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/lib/llm/__tests__/trace-recorder.test.ts | 226 +++++++++++ src/lib/llm/trace-recorder.ts | 404 +++++++++++++++++++ 2 files changed, 630 insertions(+) create mode 100644 src/lib/llm/__tests__/trace-recorder.test.ts create mode 100644 src/lib/llm/trace-recorder.ts diff --git a/src/lib/llm/__tests__/trace-recorder.test.ts b/src/lib/llm/__tests__/trace-recorder.test.ts new file mode 100644 index 0000000..e9c07e9 --- /dev/null +++ b/src/lib/llm/__tests__/trace-recorder.test.ts @@ -0,0 +1,226 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import type { LlmTracePayloadStore } from "../trace-payload-store"; +import { + createLlmTraceRecorder, + type LlmTraceRecorder, +} from "../trace-recorder"; +import type { LlmTraceRepository } from "../trace-repository"; +import type { + LlmTraceCall, + LlmTraceRun, + LlmTraceStreamEvent, +} from "../trace-types"; + +class MemoryTraceRepository implements LlmTraceRepository { + runs = new Map(); + calls = new Map(); + deletedRuns: string[] = []; + failWrites = false; + + async putRun(run: LlmTraceRun) { + if (this.failWrites) throw new Error("index unavailable"); + this.runs.set(run.job_id, structuredClone(run)); + } + + async putCall(call: LlmTraceCall) { + if (this.failWrites) throw new Error("index unavailable"); + this.calls.set(call.call_id, structuredClone(call)); + } + + async getRun(jobId: string) { + return this.runs.get(jobId) ?? null; + } + + async getLatestRun() { + return [...this.runs.values()][0] ?? null; + } + + async listCalls(jobId: string) { + return [...this.calls.values()] + .filter((call) => call.job_id === jobId) + .sort((left, right) => left.sequence - right.sequence); + } + + async listTerminalRunsExcept(jobId: string) { + return [...this.runs.values()].filter( + (run) => run.job_id !== jobId && run.status !== "running", + ); + } + + async deleteRun(jobId: string) { + this.deletedRuns.push(jobId); + this.runs.delete(jobId); + for (const call of this.calls.values()) { + if (call.job_id === jobId) this.calls.delete(call.call_id); + } + } +} + +class MemoryPayloadStore implements LlmTracePayloadStore { + values = new Map(); + deletedJobs: string[] = []; + failPuts = false; + + async putJson(key: string, value: unknown) { + if (this.failPuts) throw new Error("payload unavailable"); + this.values.set(key, structuredClone(value)); + } + + async getJson(key: string) { + return this.values.get(key) ?? null; + } + + async deleteJob(jobId: string) { + this.deletedJobs.push(jobId); + } +} + +describe("createLlmTraceRecorder", () => { + let repository: MemoryTraceRepository; + let payloadStore: MemoryPayloadStore; + let published: LlmTraceStreamEvent[]; + let recorder: LlmTraceRecorder; + + beforeEach(async () => { + repository = new MemoryTraceRepository(); + payloadStore = new MemoryPayloadStore(); + published = []; + recorder = await createLlmTraceRecorder({ + jobId: "job_1", + caseId: "case_1", + repository, + payloadStore, + publish: (event) => { + published.push(event); + }, + }); + }); + + it("persists exact bodies before publishing public metadata", async () => { + await recorder.onLlmEvent({ + type: "started", + call_id: "llmcall_1", + task: "quality_inspector", + context: { workflow_stage: "qa", schema_name: "llmQaPatchSchema" }, + provider: "deepseek", + model: "deepseek-v4-pro", + request: { model: "deepseek-v4-pro", messages: [] }, + started_at: "2026-07-16T00:00:00.000Z", + }); + await recorder.onLlmEvent({ + type: "responded", + call_id: "llmcall_1", + response: { + choices: [{ message: { content: "{}" } }], + usage: { prompt_tokens: 10, completion_tokens: 4, total_tokens: 14 }, + }, + duration_ms: 1200, + responded_at: "2026-07-16T00:00:01.200Z", + }); + + expect(payloadStore.values.get( + "llm-traces/job_1/llmcall_1/request.json", + )).toEqual({ model: "deepseek-v4-pro", messages: [] }); + expect(payloadStore.values.get( + "llm-traces/job_1/llmcall_1/response.json", + )).toEqual({ + choices: [{ message: { content: "{}" } }], + usage: { prompt_tokens: 10, completion_tokens: 4, total_tokens: 14 }, + }); + expect(published[0]).toMatchObject({ + type: "llm_call_started", + request_available: true, + }); + expect(published[1]).toMatchObject({ + type: "llm_call_responded", + token_usage: { prompt_tokens: 10, completion_tokens: 4, total_tokens: 14 }, + response_available: true, + }); + expect(JSON.stringify(published)).not.toContain("messages"); + expect(JSON.stringify(published)).not.toContain("choices"); + }); + + it("stores QA business failure separately from schema success", async () => { + await recorder.onLlmEvent({ + type: "started", + call_id: "llmcall_qa", + task: "quality_inspector", + context: { workflow_stage: "qa", schema_name: "llmQaPatchSchema" }, + provider: "deepseek", + model: "deepseek-v4-pro", + request: { messages: [] }, + started_at: "2026-07-16T00:00:00.000Z", + }); + await recorder.onLlmEvent({ + type: "validated", + call_id: "llmcall_qa", + schema_name: "llmQaPatchSchema", + schema_valid: true, + validation_issues: [], + validated_at: "2026-07-16T00:00:01.000Z", + }); + + await recorder.onWorkflowEvent({ + type: "qa_ready", + job_id: "job_1", + qa_report: { overall_status: "fail", checks: [] }, + }); + + await expect(repository.listCalls("job_1")).resolves.toEqual([ + expect.objectContaining({ + task: "quality_inspector", + schema_valid: true, + business_status: "fail", + }), + ]); + }); + + it("keeps running runs and only the newest terminal full trace", async () => { + repository.runs.set("job_old_completed", { + ...repository.runs.get("job_1")!, + job_id: "job_old_completed", + status: "completed", + }); + repository.runs.set("job_other_running", { + ...repository.runs.get("job_1")!, + job_id: "job_other_running", + status: "running", + }); + + await recorder.finish({ status: "completed" }); + + expect(payloadStore.deletedJobs).toEqual(["job_old_completed"]); + expect(repository.deletedRuns).toEqual(["job_old_completed"]); + expect(repository.deletedRuns).not.toContain("job_other_running"); + }); + + it("marks the trace incomplete without throwing when payload storage fails", async () => { + payloadStore.failPuts = true; + + await expect(recorder.onLlmEvent({ + type: "started", + call_id: "llmcall_failed_storage", + task: "fact_extractor", + context: { workflow_stage: "fact_card" }, + provider: "deepseek", + model: "deepseek-v4-pro", + request: { messages: [{ role: "user", content: "原文" }] }, + started_at: "2026-07-16T00:00:00.000Z", + })).resolves.toBeUndefined(); + + expect(published).toEqual([ + expect.objectContaining({ + type: "llm_call_started", + request_available: false, + }), + expect.objectContaining({ + type: "trace_warning", + trace_completeness: "incomplete", + }), + ]); + await expect(repository.getRun("job_1")).resolves.toMatchObject({ + trace_completeness: "incomplete", + }); + }); +}); diff --git a/src/lib/llm/trace-recorder.ts b/src/lib/llm/trace-recorder.ts new file mode 100644 index 0000000..9fdefaf --- /dev/null +++ b/src/lib/llm/trace-recorder.ts @@ -0,0 +1,404 @@ +import type { OptimizationStreamEvent } from "../workflow/stream-events"; +import { + tracePayloadKey, + type LlmTracePayloadStore, +} from "./trace-payload-store"; +import type { LlmTraceRepository } from "./trace-repository"; +import type { + LlmClientTraceEvent, + LlmClientTraceHandler, + LlmTraceCall, + LlmTraceRun, + LlmTraceRunStatus, + LlmTraceStreamEvent, + LlmTraceWorkflowStage, +} from "./trace-types"; + +export interface LlmTraceRecorder { + onLlmEvent: LlmClientTraceHandler; + onWorkflowEvent(event: OptimizationStreamEvent): Promise; + finish(input: { + status: Exclude; + errorStage?: string; + errorSummary?: string; + }): Promise; +} + +interface CreateLlmTraceRecorderInput { + jobId: string; + caseId: string | null; + repository: LlmTraceRepository; + payloadStore: LlmTracePayloadStore; + publish: (event: LlmTraceStreamEvent) => void | Promise; +} + +function nowIso() { + return new Date().toISOString(); +} + +function createRunningRun(jobId: string, caseId: string | null): LlmTraceRun { + const timestamp = nowIso(); + return { + job_id: jobId, + case_id: caseId, + status: "running", + current_stage: "input", + trace_completeness: "complete", + error_stage: null, + error_summary: null, + started_at: timestamp, + finished_at: null, + updated_at: timestamp, + }; +} + +export function safeTraceError(error: unknown) { + if (error instanceof Error) return `${error.name}: ${error.message}`; + return String(error); +} + +export function createNoopLlmTraceRecorder(): LlmTraceRecorder { + return { + onLlmEvent: async () => undefined, + onWorkflowEvent: async () => undefined, + finish: async () => undefined, + }; +} + +function tokenUsageFromResponse(response: unknown) { + if (!response || typeof response !== "object") return null; + const usage = (response as { usage?: unknown }).usage; + if (!usage || typeof usage !== "object") return null; + return Object.fromEntries( + Object.entries(usage).filter( + (entry): entry is [string, number] => typeof entry[1] === "number", + ), + ); +} + +function stageForWorkflowEvent( + event: OptimizationStreamEvent, +): LlmTraceWorkflowStage | null { + switch (event.type) { + case "job_created": + return "input"; + case "fact_card_ready": + return "fact_card"; + case "draft_started": + case "draft_ready": + return "draft"; + case "qa_started": + case "qa_ready": + return "qa"; + case "rewrite_started": + case "rewrite_ready": + return "rewrite"; + case "final_ready": + return "final"; + case "failed": + return event.stage === "job" ? "input" : event.stage; + default: + return null; + } +} + +function isTraceStage(value: string): value is LlmTraceWorkflowStage { + return [ + "unknown", + "input", + "fact_card", + "draft", + "qa", + "rewrite", + "final", + ].includes(value); +} + +export async function createLlmTraceRecorder({ + jobId, + caseId, + repository, + payloadStore, + publish, +}: CreateLlmTraceRecorderInput): Promise { + const calls = new Map(); + let sequence = 0; + let run = createRunningRun(jobId, caseId); + await repository.putRun(run); + + async function publishSafely(event: LlmTraceStreamEvent) { + try { + await publish(event); + } catch { + // A disconnected observer must never fail the article workflow. + } + } + + async function warn(error: unknown) { + run = { + ...run, + trace_completeness: "incomplete", + updated_at: nowIso(), + }; + try { + await repository.putRun(run); + } catch { + // Keep the in-memory incomplete state even when the index is unavailable. + } + await publishSafely({ + type: "trace_warning", + job_id: jobId, + trace_completeness: "incomplete", + error_summary: safeTraceError(error), + }); + } + + async function saveCall(call: LlmTraceCall) { + calls.set(call.call_id, call); + await repository.putCall(call); + } + + async function applyStartedEvent( + event: Extract, + ) { + const requestKey = tracePayloadKey(jobId, event.call_id, "request"); + let storedRequestKey: string | null = null; + let traceError: unknown; + try { + await payloadStore.putJson(requestKey, event.request); + storedRequestKey = requestKey; + } catch (error) { + traceError = error; + } + + const call: LlmTraceCall = { + call_id: event.call_id, + job_id: jobId, + sequence: ++sequence, + task: event.task, + workflow_stage: event.context.workflow_stage, + rewrite_round: event.context.rewrite_round ?? null, + provider: event.provider, + model: event.model, + status: "started", + request_object_key: storedRequestKey, + response_object_key: null, + token_usage: null, + schema_name: event.context.schema_name ?? null, + schema_valid: null, + validation_issues: [], + business_status: null, + duration_ms: null, + started_at: event.started_at, + responded_at: null, + validated_at: null, + failed_at: null, + error_type: null, + error_summary: null, + }; + + try { + await saveCall(call); + } catch (error) { + traceError ??= error; + calls.set(call.call_id, call); + } + await publishSafely({ + type: "llm_call_started", + job_id: jobId, + call_id: call.call_id, + sequence: call.sequence, + task: call.task, + workflow_stage: call.workflow_stage, + rewrite_round: call.rewrite_round, + provider: call.provider, + model: call.model, + started_at: call.started_at, + request_available: storedRequestKey !== null, + }); + if (traceError) await warn(traceError); + } + + async function applyRespondedEvent( + event: Extract, + ) { + const existing = calls.get(event.call_id); + if (!existing) throw new Error(`Unknown LLM trace call: ${event.call_id}`); + + const responseKey = tracePayloadKey(jobId, event.call_id, "response"); + let storedResponseKey: string | null = null; + let traceError: unknown; + try { + await payloadStore.putJson(responseKey, event.response); + storedResponseKey = responseKey; + } catch (error) { + traceError = error; + } + + const call: LlmTraceCall = { + ...existing, + status: "responded", + response_object_key: storedResponseKey, + token_usage: tokenUsageFromResponse(event.response), + duration_ms: event.duration_ms, + responded_at: event.responded_at, + }; + try { + await saveCall(call); + } catch (error) { + traceError ??= error; + calls.set(call.call_id, call); + } + await publishSafely({ + type: "llm_call_responded", + job_id: jobId, + call_id: call.call_id, + duration_ms: event.duration_ms, + token_usage: call.token_usage, + responded_at: event.responded_at, + response_available: storedResponseKey !== null, + }); + if (traceError) await warn(traceError); + } + + async function applyValidatedEvent( + event: Extract, + ) { + const existing = calls.get(event.call_id); + if (!existing) throw new Error(`Unknown LLM trace call: ${event.call_id}`); + const call: LlmTraceCall = { + ...existing, + status: "validated", + schema_name: event.schema_name, + schema_valid: event.schema_valid, + validation_issues: event.validation_issues, + validated_at: event.validated_at, + }; + await saveCall(call); + await publishSafely({ + type: "llm_call_validated", + job_id: jobId, + call_id: call.call_id, + schema_name: event.schema_name, + schema_valid: event.schema_valid, + validation_issues: event.validation_issues, + validated_at: event.validated_at, + }); + } + + async function applyFailedEvent( + event: Extract, + ) { + const existing = calls.get(event.call_id); + if (!existing) throw new Error(`Unknown LLM trace call: ${event.call_id}`); + const call: LlmTraceCall = { + ...existing, + status: "failed", + duration_ms: event.duration_ms, + failed_at: event.failed_at, + error_type: event.error_type, + error_summary: event.error_summary, + }; + await saveCall(call); + await publishSafely({ + type: "llm_call_failed", + job_id: jobId, + call_id: call.call_id, + error_type: event.error_type, + error_summary: event.error_summary, + failed_at: event.failed_at, + }); + } + + async function applyLlmEvent(event: LlmClientTraceEvent) { + switch (event.type) { + case "started": + await applyStartedEvent(event); + return; + case "responded": + await applyRespondedEvent(event); + return; + case "validated": + await applyValidatedEvent(event); + return; + case "failed": + await applyFailedEvent(event); + } + } + + async function applyWorkflowEvent(event: OptimizationStreamEvent) { + const stage = stageForWorkflowEvent(event); + if (stage) { + run = { ...run, current_stage: stage, updated_at: nowIso() }; + await repository.putRun(run); + } + + if (event.type === "qa_ready") { + const qualityCall = [...calls.values()] + .filter((call) => call.task === "quality_inspector") + .sort((left, right) => right.sequence - left.sequence)[0]; + if (qualityCall) { + await saveCall({ + ...qualityCall, + business_status: event.qa_report.overall_status, + }); + } + } + } + + async function finishRun(input: { + status: Exclude; + errorStage?: string; + errorSummary?: string; + }) { + const finishedAt = nowIso(); + run = { + ...run, + status: input.status, + current_stage: input.status === "completed" + ? "final" + : input.errorStage && isTraceStage(input.errorStage) + ? input.errorStage + : run.current_stage, + error_stage: input.errorStage ?? null, + error_summary: input.errorSummary ?? null, + finished_at: finishedAt, + updated_at: finishedAt, + }; + await repository.putRun(run); + + const expired = await repository.listTerminalRunsExcept(jobId); + for (const oldRun of expired) { + try { + await payloadStore.deleteJob(oldRun.job_id); + await repository.deleteRun(oldRun.job_id); + } catch (error) { + await warn(error); + } + } + } + + return { + onLlmEvent: async (event) => { + try { + await applyLlmEvent(event); + } catch (error) { + await warn(error); + } + }, + onWorkflowEvent: async (event) => { + try { + await applyWorkflowEvent(event); + } catch (error) { + await warn(error); + } + }, + finish: async (input) => { + try { + await finishRun(input); + } catch (error) { + await warn(error); + } + }, + }; +}