新增LLM任务追踪收集器
This commit is contained in:
@@ -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<string, LlmTraceRun>();
|
||||
calls = new Map<string, LlmTraceCall>();
|
||||
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<string, unknown>();
|
||||
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",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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<void>;
|
||||
finish(input: {
|
||||
status: Exclude<LlmTraceRunStatus, "running">;
|
||||
errorStage?: string;
|
||||
errorSummary?: string;
|
||||
}): Promise<void>;
|
||||
}
|
||||
|
||||
interface CreateLlmTraceRecorderInput {
|
||||
jobId: string;
|
||||
caseId: string | null;
|
||||
repository: LlmTraceRepository;
|
||||
payloadStore: LlmTracePayloadStore;
|
||||
publish: (event: LlmTraceStreamEvent) => void | Promise<void>;
|
||||
}
|
||||
|
||||
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<LlmTraceRecorder> {
|
||||
const calls = new Map<string, LlmTraceCall>();
|
||||
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<LlmClientTraceEvent, { type: "started" }>,
|
||||
) {
|
||||
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<LlmClientTraceEvent, { type: "responded" }>,
|
||||
) {
|
||||
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<LlmClientTraceEvent, { type: "validated" }>,
|
||||
) {
|
||||
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<LlmClientTraceEvent, { type: "failed" }>,
|
||||
) {
|
||||
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<LlmTraceRunStatus, "running">;
|
||||
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);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user