新增LLM追踪索引存储

This commit is contained in:
czj
2026-07-16 12:03:12 +08:00
parent a22844d2dd
commit 602c0b99da
7 changed files with 721 additions and 0 deletions
@@ -0,0 +1,48 @@
CREATE TABLE IF NOT EXISTS llm_trace_runs (
job_id TEXT PRIMARY KEY,
case_id TEXT,
status TEXT NOT NULL,
current_stage TEXT NOT NULL,
trace_completeness TEXT NOT NULL,
error_stage TEXT,
error_summary TEXT,
started_at TEXT NOT NULL,
finished_at TEXT,
updated_at TEXT NOT NULL,
FOREIGN KEY (job_id) REFERENCES article_jobs(id) ON DELETE CASCADE,
FOREIGN KEY (case_id) REFERENCES optimization_cases(id) ON DELETE SET NULL
);
CREATE TABLE IF NOT EXISTS llm_trace_calls (
call_id TEXT PRIMARY KEY,
job_id TEXT NOT NULL,
sequence INTEGER NOT NULL,
task TEXT NOT NULL,
workflow_stage TEXT NOT NULL,
rewrite_round INTEGER,
provider TEXT NOT NULL,
model TEXT NOT NULL,
status TEXT NOT NULL,
request_object_key TEXT,
response_object_key TEXT,
token_usage TEXT,
schema_name TEXT,
schema_valid INTEGER,
validation_issues TEXT NOT NULL DEFAULT '[]',
business_status TEXT,
duration_ms INTEGER,
started_at TEXT NOT NULL,
responded_at TEXT,
validated_at TEXT,
failed_at TEXT,
error_type TEXT,
error_summary TEXT,
UNIQUE (job_id, sequence),
FOREIGN KEY (job_id) REFERENCES llm_trace_runs(job_id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_llm_trace_runs_status_updated
ON llm_trace_runs(status, updated_at DESC);
CREATE INDEX IF NOT EXISTS idx_llm_trace_calls_job_sequence
ON llm_trace_calls(job_id, sequence);
+49
View File
@@ -115,6 +115,49 @@ export function initializeSchema(db: Database.Database) {
foreign key (article_job_id) references article_jobs(id) on delete set null
);
create table if not exists llm_trace_runs (
job_id text primary key,
case_id text,
status text not null,
current_stage text not null,
trace_completeness text not null,
error_stage text,
error_summary text,
started_at text not null,
finished_at text,
updated_at text not null,
foreign key (job_id) references article_jobs(id) on delete cascade,
foreign key (case_id) references optimization_cases(id) on delete set null
);
create table if not exists llm_trace_calls (
call_id text primary key,
job_id text not null,
sequence integer not null,
task text not null,
workflow_stage text not null,
rewrite_round integer,
provider text not null,
model text not null,
status text not null,
request_object_key text,
response_object_key text,
token_usage text,
schema_name text,
schema_valid integer,
validation_issues text not null default '[]',
business_status text,
duration_ms integer,
started_at text not null,
responded_at text,
validated_at text,
failed_at text,
error_type text,
error_summary text,
unique (job_id, sequence),
foreign key (job_id) references llm_trace_runs(job_id) on delete cascade
);
create table if not exists rubric_versions (
id text primary key,
version text not null,
@@ -204,6 +247,12 @@ export function initializeSchema(db: Database.Database) {
create index if not exists idx_performance_snapshots_publication
on performance_snapshots(publication_id);
create index if not exists idx_llm_trace_runs_status_updated
on llm_trace_runs(status, updated_at desc);
create index if not exists idx_llm_trace_calls_job_sequence
on llm_trace_calls(job_id, sequence);
`);
ensureColumn(db, "article_jobs", "case_id", "text");
@@ -0,0 +1,114 @@
import { describe, expect, it, vi } from "vitest";
import { createD1TraceRepository } from "../d1-trace-repository";
import type { LlmTraceCall, LlmTraceRun } from "../trace-types";
const run: LlmTraceRun = {
job_id: "job_1",
case_id: "case_1",
status: "running",
current_stage: "draft",
trace_completeness: "complete",
error_stage: null,
error_summary: null,
started_at: "2026-07-16T00:00:00.000Z",
finished_at: null,
updated_at: "2026-07-16T00:00:01.000Z",
};
const call: LlmTraceCall = {
call_id: "llmcall_1",
job_id: "job_1",
sequence: 1,
task: "article_optimizer",
workflow_stage: "draft",
rewrite_round: null,
provider: "deepseek",
model: "deepseek-v4-pro",
status: "validated",
request_object_key: "llm-traces/job_1/llmcall_1/request.json",
response_object_key: "llm-traces/job_1/llmcall_1/response.json",
token_usage: { prompt_tokens: 10, completion_tokens: 4, total_tokens: 14 },
schema_name: "optimizedArticleSchema",
schema_valid: true,
validation_issues: ["title: required"],
business_status: null,
duration_ms: 1200,
started_at: "2026-07-16T00:00:00.000Z",
responded_at: "2026-07-16T00:00:01.000Z",
validated_at: "2026-07-16T00:00:01.200Z",
failed_at: null,
error_type: null,
error_summary: null,
};
describe("createD1TraceRepository", () => {
it("binds run and call snapshots with JSON fields serialized", async () => {
const runStatement = vi.fn().mockResolvedValue({ success: true });
const bind = vi.fn().mockReturnValue({ run: runStatement });
const prepare = vi.fn().mockReturnValue({ bind });
const repository = createD1TraceRepository(
{ prepare } as unknown as D1Database,
);
await repository.putRun(run);
await repository.putCall(call);
expect(prepare).toHaveBeenNthCalledWith(
1,
expect.stringContaining("insert into llm_trace_runs"),
);
expect(prepare).toHaveBeenNthCalledWith(
2,
expect.stringContaining("insert into llm_trace_calls"),
);
expect(bind).toHaveBeenNthCalledWith(
2,
call.call_id,
call.job_id,
call.sequence,
call.task,
call.workflow_stage,
null,
call.provider,
call.model,
call.status,
call.request_object_key,
call.response_object_key,
JSON.stringify(call.token_usage),
call.schema_name,
1,
JSON.stringify(call.validation_issues),
null,
call.duration_ms,
call.started_at,
call.responded_at,
call.validated_at,
null,
null,
null,
);
});
it("reads calls in sequence order and parses JSON fields", async () => {
const all = vi.fn().mockResolvedValue({
results: [{
...call,
token_usage: JSON.stringify(call.token_usage),
schema_valid: 1,
validation_issues: JSON.stringify(call.validation_issues),
}],
});
const bind = vi.fn().mockReturnValue({ all });
const prepare = vi.fn().mockReturnValue({ bind });
const repository = createD1TraceRepository(
{ prepare } as unknown as D1Database,
);
await expect(repository.listCalls("job_1")).resolves.toEqual([call]);
expect(prepare).toHaveBeenCalledWith(
expect.stringContaining("order by sequence"),
);
expect(bind).toHaveBeenCalledWith("job_1");
});
});
@@ -0,0 +1,99 @@
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { createSqliteRepository } from "../../db/sqlite-repository";
import type { LlmTraceCall, LlmTraceRun } from "../trace-types";
import { createSqliteTraceRepository } from "../sqlite-trace-repository";
function runFixture(jobId: string): LlmTraceRun {
return {
job_id: jobId,
case_id: null,
status: "running",
current_stage: "fact_card",
trace_completeness: "complete",
error_stage: null,
error_summary: null,
started_at: "2026-07-16T00:00:00.000Z",
finished_at: null,
updated_at: "2026-07-16T00:00:00.000Z",
};
}
function callFixture(
jobId: string,
callId: string,
sequence: number,
): LlmTraceCall {
return {
call_id: callId,
job_id: jobId,
sequence,
task: sequence === 1 ? "fact_extractor" : "article_optimizer",
workflow_stage: sequence === 1 ? "fact_card" : "draft",
rewrite_round: null,
provider: "deepseek",
model: "deepseek-v4-pro",
status: "started",
request_object_key: `llm-traces/${jobId}/${callId}/request.json`,
response_object_key: null,
token_usage: null,
schema_name: null,
schema_valid: null,
validation_issues: [],
business_status: null,
duration_ms: null,
started_at: `2026-07-16T00:00:0${sequence}.000Z`,
responded_at: null,
validated_at: null,
failed_at: null,
error_type: null,
error_summary: null,
};
}
describe("createSqliteTraceRepository", () => {
let tempDir: string;
let dbPath: string;
beforeEach(() => {
tempDir = mkdtempSync(join(tmpdir(), "geo-llm-traces-"));
dbPath = join(tempDir, "app.db");
});
afterEach(() => {
rmSync(tempDir, { recursive: true, force: true });
});
it("stores a run and ordered calls, then deletes only trace rows", async () => {
const appRepository = createSqliteRepository(dbPath);
const job = await appRepository.createArticleJob({
source_title: "Title",
source_body: "Body",
image_inputs: [],
publish_platform: "official_site",
user_instructions: "",
});
const repository = createSqliteTraceRepository(dbPath);
await repository.putRun(runFixture(job.id));
await repository.putCall(callFixture(job.id, "llmcall_1", 1));
await repository.putCall(callFixture(job.id, "llmcall_2", 2));
await expect(repository.getLatestRun()).resolves.toMatchObject({
job_id: job.id,
status: "running",
});
await expect(repository.listCalls(job.id)).resolves.toEqual([
expect.objectContaining({ call_id: "llmcall_1", sequence: 1 }),
expect.objectContaining({ call_id: "llmcall_2", sequence: 2 }),
]);
await repository.deleteRun(job.id);
await expect(appRepository.getArticleJob(job.id)).resolves.not.toBeNull();
await expect(repository.getRun(job.id)).resolves.toBeNull();
});
});
+192
View File
@@ -0,0 +1,192 @@
import type { LlmTraceRepository } from "./trace-repository";
import type {
LlmBusinessStatus,
LlmProviderName,
LlmTaskName,
LlmTraceCall,
LlmTraceCallStatus,
LlmTraceCompleteness,
LlmTraceErrorType,
LlmTraceRun,
LlmTraceRunStatus,
LlmTraceWorkflowStage,
} from "./trace-types";
interface TraceRunRow {
job_id: string;
case_id: string | null;
status: LlmTraceRunStatus;
current_stage: LlmTraceWorkflowStage;
trace_completeness: LlmTraceCompleteness;
error_stage: string | null;
error_summary: string | null;
started_at: string;
finished_at: string | null;
updated_at: string;
}
interface TraceCallRow {
call_id: string;
job_id: string;
sequence: number;
task: LlmTaskName;
workflow_stage: LlmTraceWorkflowStage;
rewrite_round: number | null;
provider: LlmProviderName;
model: string;
status: LlmTraceCallStatus;
request_object_key: string | null;
response_object_key: string | null;
token_usage: string | null;
schema_name: string | null;
schema_valid: 0 | 1 | null;
validation_issues: string;
business_status: LlmBusinessStatus | null;
duration_ms: number | null;
started_at: string;
responded_at: string | null;
validated_at: string | null;
failed_at: string | null;
error_type: LlmTraceErrorType | null;
error_summary: string | null;
}
const putRunSql = `
insert into llm_trace_runs (
job_id, case_id, status, current_stage, trace_completeness,
error_stage, error_summary, started_at, finished_at, updated_at
) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
on conflict(job_id) do update set
case_id = excluded.case_id,
status = excluded.status,
current_stage = excluded.current_stage,
trace_completeness = excluded.trace_completeness,
error_stage = excluded.error_stage,
error_summary = excluded.error_summary,
started_at = excluded.started_at,
finished_at = excluded.finished_at,
updated_at = excluded.updated_at
`;
const putCallSql = `
insert into llm_trace_calls (
call_id, job_id, sequence, task, workflow_stage, rewrite_round,
provider, model, status, request_object_key, response_object_key,
token_usage, schema_name, schema_valid, validation_issues,
business_status, duration_ms, started_at, responded_at, validated_at,
failed_at, error_type, error_summary
) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
on conflict(call_id) do update set
sequence = excluded.sequence,
task = excluded.task,
workflow_stage = excluded.workflow_stage,
rewrite_round = excluded.rewrite_round,
provider = excluded.provider,
model = excluded.model,
status = excluded.status,
request_object_key = excluded.request_object_key,
response_object_key = excluded.response_object_key,
token_usage = excluded.token_usage,
schema_name = excluded.schema_name,
schema_valid = excluded.schema_valid,
validation_issues = excluded.validation_issues,
business_status = excluded.business_status,
duration_ms = excluded.duration_ms,
started_at = excluded.started_at,
responded_at = excluded.responded_at,
validated_at = excluded.validated_at,
failed_at = excluded.failed_at,
error_type = excluded.error_type,
error_summary = excluded.error_summary
`;
function toTraceCall(row: TraceCallRow): LlmTraceCall {
return {
...row,
token_usage: row.token_usage == null
? null
: JSON.parse(row.token_usage) as Record<string, number>,
schema_valid: row.schema_valid == null ? null : row.schema_valid === 1,
validation_issues: JSON.parse(row.validation_issues) as string[],
};
}
export function createD1TraceRepository(db: D1Database): LlmTraceRepository {
return {
async putRun(run) {
await db.prepare(putRunSql).bind(
run.job_id,
run.case_id,
run.status,
run.current_stage,
run.trace_completeness,
run.error_stage,
run.error_summary,
run.started_at,
run.finished_at,
run.updated_at,
).run();
},
async putCall(call) {
await db.prepare(putCallSql).bind(
call.call_id,
call.job_id,
call.sequence,
call.task,
call.workflow_stage,
call.rewrite_round,
call.provider,
call.model,
call.status,
call.request_object_key,
call.response_object_key,
call.token_usage == null ? null : JSON.stringify(call.token_usage),
call.schema_name,
call.schema_valid == null ? null : Number(call.schema_valid),
JSON.stringify(call.validation_issues),
call.business_status,
call.duration_ms,
call.started_at,
call.responded_at,
call.validated_at,
call.failed_at,
call.error_type,
call.error_summary,
).run();
},
async getRun(jobId) {
return await db
.prepare("select * from llm_trace_runs where job_id = ?")
.bind(jobId)
.first<TraceRunRow>() ?? null;
},
async getLatestRun() {
return await db.prepare(`
select * from llm_trace_runs
order by case when status = 'running' then 0 else 1 end, updated_at desc
limit 1
`).first<TraceRunRow>() ?? null;
},
async listCalls(jobId) {
const result = await db
.prepare("select * from llm_trace_calls where job_id = ? order by sequence")
.bind(jobId)
.all<TraceCallRow>();
return result.results.map(toTraceCall);
},
async listTerminalRunsExcept(jobId) {
const result = await db.prepare(`
select * from llm_trace_runs
where status <> 'running' and job_id <> ?
order by finished_at desc, updated_at desc
`).bind(jobId).all<TraceRunRow>();
return result.results;
},
async deleteRun(jobId) {
await db
.prepare("delete from llm_trace_runs where job_id = ?")
.bind(jobId)
.run();
},
};
}
+195
View File
@@ -0,0 +1,195 @@
import { createDatabase, getDefaultDatabasePath } from "../db/connection";
import { initializeSchema } from "../db/schema";
import type { LlmTraceRepository } from "./trace-repository";
import type {
LlmBusinessStatus,
LlmProviderName,
LlmTaskName,
LlmTraceCall,
LlmTraceCallStatus,
LlmTraceCompleteness,
LlmTraceErrorType,
LlmTraceRun,
LlmTraceRunStatus,
LlmTraceWorkflowStage,
} from "./trace-types";
interface TraceRunRow {
job_id: string;
case_id: string | null;
status: LlmTraceRunStatus;
current_stage: LlmTraceWorkflowStage;
trace_completeness: LlmTraceCompleteness;
error_stage: string | null;
error_summary: string | null;
started_at: string;
finished_at: string | null;
updated_at: string;
}
interface TraceCallRow {
call_id: string;
job_id: string;
sequence: number;
task: LlmTaskName;
workflow_stage: LlmTraceWorkflowStage;
rewrite_round: number | null;
provider: LlmProviderName;
model: string;
status: LlmTraceCallStatus;
request_object_key: string | null;
response_object_key: string | null;
token_usage: string | null;
schema_name: string | null;
schema_valid: 0 | 1 | null;
validation_issues: string;
business_status: LlmBusinessStatus | null;
duration_ms: number | null;
started_at: string;
responded_at: string | null;
validated_at: string | null;
failed_at: string | null;
error_type: LlmTraceErrorType | null;
error_summary: string | null;
}
const putRunSql = `
insert into llm_trace_runs (
job_id, case_id, status, current_stage, trace_completeness,
error_stage, error_summary, started_at, finished_at, updated_at
) values (
@job_id, @case_id, @status, @current_stage, @trace_completeness,
@error_stage, @error_summary, @started_at, @finished_at, @updated_at
)
on conflict(job_id) do update set
case_id = excluded.case_id,
status = excluded.status,
current_stage = excluded.current_stage,
trace_completeness = excluded.trace_completeness,
error_stage = excluded.error_stage,
error_summary = excluded.error_summary,
started_at = excluded.started_at,
finished_at = excluded.finished_at,
updated_at = excluded.updated_at
`;
const putCallSql = `
insert into llm_trace_calls (
call_id, job_id, sequence, task, workflow_stage, rewrite_round,
provider, model, status, request_object_key, response_object_key,
token_usage, schema_name, schema_valid, validation_issues,
business_status, duration_ms, started_at, responded_at, validated_at,
failed_at, error_type, error_summary
) values (
@call_id, @job_id, @sequence, @task, @workflow_stage, @rewrite_round,
@provider, @model, @status, @request_object_key, @response_object_key,
@token_usage, @schema_name, @schema_valid, @validation_issues,
@business_status, @duration_ms, @started_at, @responded_at, @validated_at,
@failed_at, @error_type, @error_summary
)
on conflict(call_id) do update set
sequence = excluded.sequence,
task = excluded.task,
workflow_stage = excluded.workflow_stage,
rewrite_round = excluded.rewrite_round,
provider = excluded.provider,
model = excluded.model,
status = excluded.status,
request_object_key = excluded.request_object_key,
response_object_key = excluded.response_object_key,
token_usage = excluded.token_usage,
schema_name = excluded.schema_name,
schema_valid = excluded.schema_valid,
validation_issues = excluded.validation_issues,
business_status = excluded.business_status,
duration_ms = excluded.duration_ms,
started_at = excluded.started_at,
responded_at = excluded.responded_at,
validated_at = excluded.validated_at,
failed_at = excluded.failed_at,
error_type = excluded.error_type,
error_summary = excluded.error_summary
`;
function toCallParams(call: LlmTraceCall) {
return {
...call,
token_usage: call.token_usage == null ? null : JSON.stringify(call.token_usage),
schema_valid: call.schema_valid == null ? null : Number(call.schema_valid),
validation_issues: JSON.stringify(call.validation_issues),
};
}
function toTraceCall(row: TraceCallRow): LlmTraceCall {
return {
...row,
token_usage: row.token_usage == null
? null
: JSON.parse(row.token_usage) as Record<string, number>,
schema_valid: row.schema_valid == null ? null : row.schema_valid === 1,
validation_issues: JSON.parse(row.validation_issues) as string[],
};
}
export function createSqliteTraceRepository(
dbPath = getDefaultDatabasePath(),
): LlmTraceRepository {
function withDb<T>(action: (db: ReturnType<typeof createDatabase>) => T) {
const db = createDatabase(dbPath);
initializeSchema(db);
try {
return action(db);
} finally {
db.close();
}
}
return {
async putRun(run) {
withDb((db) => db.prepare(putRunSql).run(run));
},
async putCall(call) {
withDb((db) => db.prepare(putCallSql).run(toCallParams(call)));
},
async getRun(jobId) {
return withDb((db) => {
const row = db
.prepare("select * from llm_trace_runs where job_id = ?")
.get(jobId) as TraceRunRow | undefined;
return row ?? null;
});
},
async getLatestRun() {
return withDb((db) => {
const row = db.prepare(`
select * from llm_trace_runs
order by case when status = 'running' then 0 else 1 end, updated_at desc
limit 1
`).get() as TraceRunRow | undefined;
return row ?? null;
});
},
async listCalls(jobId) {
return withDb((db) => {
const rows = db
.prepare("select * from llm_trace_calls where job_id = ? order by sequence")
.all(jobId) as TraceCallRow[];
return rows.map(toTraceCall);
});
},
async listTerminalRunsExcept(jobId) {
return withDb((db) => db
.prepare(`
select * from llm_trace_runs
where status <> 'running' and job_id <> ?
order by finished_at desc, updated_at desc
`)
.all(jobId) as TraceRunRow[]);
},
async deleteRun(jobId) {
withDb((db) => db
.prepare("delete from llm_trace_runs where job_id = ?")
.run(jobId));
},
};
}
+24
View File
@@ -0,0 +1,24 @@
import { getDefaultDatabasePath } from "../db/connection";
import { getAppCloudflareEnv } from "../runtime/cloudflare";
import { createD1TraceRepository } from "./d1-trace-repository";
import { createSqliteTraceRepository } from "./sqlite-trace-repository";
import type { LlmTraceCall, LlmTraceRun } from "./trace-types";
export interface LlmTraceRepository {
putRun(run: LlmTraceRun): Promise<void>;
putCall(call: LlmTraceCall): Promise<void>;
getRun(jobId: string): Promise<LlmTraceRun | null>;
getLatestRun(): Promise<LlmTraceRun | null>;
listCalls(jobId: string): Promise<LlmTraceCall[]>;
listTerminalRunsExcept(jobId: string): Promise<LlmTraceRun[]>;
deleteRun(jobId: string): Promise<void>;
}
export function getLlmTraceRepositoryFromRuntime(): LlmTraceRepository {
if (process.env.APP_RUNTIME === "cloudflare") {
const env = getAppCloudflareEnv();
if (!env?.DB) throw new Error("Cloudflare D1 binding DB is required");
return createD1TraceRepository(env.DB);
}
return createSqliteTraceRepository(getDefaultDatabasePath());
}