From 8e50ecef77e106656bac7dc9ffccdfb127c24f05 Mon Sep 17 00:00:00 2001 From: czj <13261895355@163.com> Date: Thu, 16 Jul 2026 12:19:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=85=A5=E6=96=87=E7=AB=A0=E4=BC=98?= =?UTF-8?q?=E5=8C=96LLM=E8=BF=BD=E8=B8=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/app/api/__tests__/jobs.test.ts | 76 +++++++++++++++++++---- src/app/api/jobs/optimize-stream/route.ts | 52 ++++++++++++++-- src/lib/llm/d1-trace-repository.ts | 1 - src/lib/llm/sqlite-trace-repository.ts | 1 - src/lib/workflow/article-optimizer.ts | 5 ++ src/lib/workflow/fact-extractor.ts | 8 ++- src/lib/workflow/quality-inspector.ts | 6 ++ src/lib/workflow/streaming-optimizer.ts | 15 ++++- src/lib/workflow/targeted-rewriter.ts | 8 +++ 9 files changed, 150 insertions(+), 22 deletions(-) diff --git a/src/app/api/__tests__/jobs.test.ts b/src/app/api/__tests__/jobs.test.ts index 3a042b3..46b5720 100644 --- a/src/app/api/__tests__/jobs.test.ts +++ b/src/app/api/__tests__/jobs.test.ts @@ -205,18 +205,17 @@ describe("job API routes", () => { }); it("streams a one-click optimization from body-only input", async () => { - llmMocks.generateValidatedJson - .mockResolvedValueOnce(validCandidateFactCard) - .mockResolvedValueOnce({ - title: "流式优化标题", - summary: "流式优化摘要。", - body_markdown: - "## 服务能力\nExample Technology Co., Ltd. 提供 GEO optimization 服务。", - image_suggestions: [], - changed_sections: ["title", "body"], - requires_user_confirmation: [], - }) - .mockResolvedValueOnce({ checks: [] }); + mockTrackedLlmResult(validCandidateFactCard, "llmcall_fact"); + mockTrackedLlmResult({ + title: "流式优化标题", + summary: "流式优化摘要。", + body_markdown: + "## 服务能力\nExample Technology Co., Ltd. 提供 GEO optimization 服务。", + image_suggestions: [], + changed_sections: ["title", "body"], + requires_user_confirmation: [], + }, "llmcall_draft"); + mockTrackedLlmResult({ checks: [] }, "llmcall_qa"); const response = await optimizeStream( request({ @@ -231,13 +230,25 @@ describe("job API routes", () => { expect(response.status).toBe(200); expect(events.map((event) => event.type)).toEqual([ "job_created", + "llm_call_started", + "llm_call_responded", + "llm_call_validated", "fact_card_ready", "draft_started", + "llm_call_started", + "llm_call_responded", + "llm_call_validated", "draft_ready", "qa_started", + "llm_call_started", + "llm_call_responded", + "llm_call_validated", "qa_ready", "final_ready", ]); + expect(JSON.stringify(events)).not.toContain("messages"); + expect(JSON.stringify(events)).not.toContain("choices"); + expect(JSON.stringify(events)).not.toContain("test-key"); expect( events.find((event) => event.type === "fact_card_ready")?.fact_card, ).toEqual(expect.objectContaining({ confirmed_by_user: false })); @@ -784,6 +795,47 @@ async function createJobFixture() { return response.json() as Promise<{ job: { id: string } }>; } +function mockTrackedLlmResult(value: unknown, callId: string) { + llmMocks.generateValidatedJson.mockImplementationOnce(async (input) => { + const onTraceEvent = input.onTraceEvent as + | ((event: Record) => void | Promise) + | undefined; + await onTraceEvent?.({ + type: "started", + call_id: callId, + task: input.task ?? "unknown", + context: { + workflow_stage: input.traceStage ?? "unknown", + rewrite_round: input.rewriteRound, + schema_name: input.schemaName, + }, + provider: "deepseek", + model: "deepseek-v4-pro", + request: { model: "deepseek-v4-pro", messages: [] }, + started_at: "2026-07-16T00:00:00.000Z", + }); + await onTraceEvent?.({ + type: "responded", + call_id: callId, + response: { + choices: [{ message: { content: JSON.stringify(value) } }], + usage: { prompt_tokens: 10, completion_tokens: 4, total_tokens: 14 }, + }, + duration_ms: 1200, + responded_at: "2026-07-16T00:00:01.200Z", + }); + await onTraceEvent?.({ + type: "validated", + call_id: callId, + schema_name: input.schemaName ?? "anonymousSchema", + schema_valid: true, + validation_issues: [], + validated_at: "2026-07-16T00:00:01.300Z", + }); + return value; + }); +} + function request(body: unknown, options: { apiKey?: string | null } = {}) { const headers: Record = { "content-type": "application/json" }; const apiKey = options.apiKey === undefined ? "test-key" : options.apiKey; diff --git a/src/app/api/jobs/optimize-stream/route.ts b/src/app/api/jobs/optimize-stream/route.ts index 66cac06..aebebfc 100644 --- a/src/app/api/jobs/optimize-stream/route.ts +++ b/src/app/api/jobs/optimize-stream/route.ts @@ -7,6 +7,13 @@ import { getRepositoryFromRuntime } from "../../../../lib/db/repository"; import { optimizationFactCardSchema } from "../../../../lib/domain/validation"; import type { LlmAuditSummary } from "../../../../lib/llm/audit"; import { LlmValidationError } from "../../../../lib/llm/client"; +import { getLlmTracePayloadStoreFromRuntime } from "../../../../lib/llm/trace-payload-store"; +import { + createLlmTraceRecorder, + createNoopLlmTraceRecorder, + safeTraceError, +} from "../../../../lib/llm/trace-recorder"; +import { getLlmTraceRepositoryFromRuntime } from "../../../../lib/llm/trace-repository"; import { getExportStoreFromRuntime } from "../../../../lib/workflow/export-store"; import { extractCandidateFactCard } from "../../../../lib/workflow/fact-extractor"; import { @@ -61,6 +68,7 @@ export async function POST(request: Request) { let jobId: string | undefined; let caseId: string | undefined; let stage: OptimizationStreamStage = "job"; + let traceRecorder = createNoopLlmTraceRecorder(); const llmAuditSummary: LlmAuditSummary[] = []; const processSummary: ProcessSummaryStep[] = []; const requestStartedAt = Date.now(); @@ -104,6 +112,23 @@ export async function POST(request: Request) { case: { id: optimizationCase.id, case_type: "article" }, }); + try { + traceRecorder = await createLlmTraceRecorder({ + jobId: job.id, + caseId: optimizationCase.id, + repository: getLlmTraceRepositoryFromRuntime(), + payloadStore: getLlmTracePayloadStoreFromRuntime(), + publish: (event) => send(event), + }); + } catch (error) { + send({ + type: "trace_warning", + job_id: job.id, + trace_completeness: "incomplete", + error_summary: safeTraceError(error), + }); + } + stage = "fact_card"; const factCardStartedAt = Date.now(); const factCard = optimizationFactCardSchema.parse( @@ -112,6 +137,7 @@ export async function POST(request: Request) { onAuditSummary: (summary) => { llmAuditSummary.push(summary); }, + onTraceEvent: traceRecorder.onLlmEvent, })), ); processSummary.push( @@ -124,20 +150,24 @@ export async function POST(request: Request) { }), ); const savedFactCard = await repository.saveFactCard(job.id, factCard); - send({ + const factCardReadyEvent: OptimizationStreamEvent = { type: "fact_card_ready", job_id: job.id, fact_card: savedFactCard, - }); + }; + await traceRecorder.onWorkflowEvent(factCardReadyEvent); + send(factCardReadyEvent); const result = await runStreamingOptimizationWorkflow({ jobId: job.id, input: normalized.articleInput, factCard: savedFactCard, - onEvent: (event) => { + onEvent: async (event) => { stage = stageForEvent(event, stage); + await traceRecorder.onWorkflowEvent(event); send(event); }, + onTraceEvent: traceRecorder.onLlmEvent, onAuditSummary: (summary) => { llmAuditSummary.push(summary); }, @@ -180,7 +210,7 @@ export async function POST(request: Request) { error_stage: null, error_summary: null, }); - send({ + const finalReadyEvent: OptimizationStreamEvent = { type: "final_ready", job_id: job.id, case: { id: optimizationCase.id, case_type: "article" }, @@ -188,7 +218,10 @@ export async function POST(request: Request) { optimized_article: optimizedArticle, qa_report: qaReport, export_paths: exportPaths, - }); + }; + await traceRecorder.onWorkflowEvent(finalReadyEvent); + await traceRecorder.finish({ status: "completed" }); + send(finalReadyEvent); } catch (error) { const message = error instanceof Error ? error.message : "优化失败"; let failedVersion: @@ -225,14 +258,21 @@ export async function POST(request: Request) { failedVersion = undefined; } } - send({ + const failedEvent: OptimizationStreamEvent = { type: "failed", job_id: jobId, case: caseId ? { id: caseId, case_type: "article" } : undefined, result_version: failedVersion, stage, error: message, + }; + await traceRecorder.onWorkflowEvent(failedEvent); + await traceRecorder.finish({ + status: "failed", + errorStage: stage, + errorSummary: message, }); + send(failedEvent); } finally { controller.close(); } diff --git a/src/lib/llm/d1-trace-repository.ts b/src/lib/llm/d1-trace-repository.ts index b6739ce..27025f9 100644 --- a/src/lib/llm/d1-trace-repository.ts +++ b/src/lib/llm/d1-trace-repository.ts @@ -7,7 +7,6 @@ import type { LlmTraceCallStatus, LlmTraceCompleteness, LlmTraceErrorType, - LlmTraceRun, LlmTraceRunStatus, LlmTraceWorkflowStage, } from "./trace-types"; diff --git a/src/lib/llm/sqlite-trace-repository.ts b/src/lib/llm/sqlite-trace-repository.ts index 6938faa..d489be0 100644 --- a/src/lib/llm/sqlite-trace-repository.ts +++ b/src/lib/llm/sqlite-trace-repository.ts @@ -9,7 +9,6 @@ import type { LlmTraceCallStatus, LlmTraceCompleteness, LlmTraceErrorType, - LlmTraceRun, LlmTraceRunStatus, LlmTraceWorkflowStage, } from "./trace-types"; diff --git a/src/lib/workflow/article-optimizer.ts b/src/lib/workflow/article-optimizer.ts index 29cf50a..4ee1ff9 100644 --- a/src/lib/workflow/article-optimizer.ts +++ b/src/lib/workflow/article-optimizer.ts @@ -13,12 +13,14 @@ import { export interface OptimizeArticleInput { input: ArticleInput; factCard: OptimizationFactCard; + onTraceEvent?: GenerateInput["onTraceEvent"]; onAuditSummary?: GenerateInput["onAuditSummary"]; } export async function optimizeArticle({ input, factCard, + onTraceEvent, onAuditSummary, }: OptimizeArticleInput): Promise { const llmArticle = await generateValidatedJson({ @@ -27,6 +29,9 @@ export async function optimizeArticle({ prompt: buildArticleOptimizerPrompt(input, factCard), temperature: 0.2, task: "article_optimizer", + schemaName: "optimizedArticleSchema", + traceStage: "draft", + onTraceEvent, onAuditSummary, }); diff --git a/src/lib/workflow/fact-extractor.ts b/src/lib/workflow/fact-extractor.ts index 3e941d3..39edcb7 100644 --- a/src/lib/workflow/fact-extractor.ts +++ b/src/lib/workflow/fact-extractor.ts @@ -8,7 +8,10 @@ import { export async function extractCandidateFactCard( input: ArticleInput, - options: { onAuditSummary?: GenerateInput["onAuditSummary"] } = {}, + options: { + onAuditSummary?: GenerateInput["onAuditSummary"]; + onTraceEvent?: GenerateInput["onTraceEvent"]; + } = {}, ): Promise { return generateValidatedJson({ schema: candidateFactCardSchema, @@ -16,6 +19,9 @@ export async function extractCandidateFactCard( prompt: buildFactExtractorPrompt(input), temperature: 0.1, task: "fact_extractor", + schemaName: "candidateFactCardSchema", + traceStage: "fact_card", + onTraceEvent: options.onTraceEvent, onAuditSummary: options.onAuditSummary, }); } diff --git a/src/lib/workflow/quality-inspector.ts b/src/lib/workflow/quality-inspector.ts index fc650c3..1d0760d 100644 --- a/src/lib/workflow/quality-inspector.ts +++ b/src/lib/workflow/quality-inspector.ts @@ -43,6 +43,8 @@ export interface InspectQualityInput { factCard: OptimizationFactCard; platform: PublishPlatform; sourceImages: ImageInput[]; + rewriteRound?: number; + onTraceEvent?: GenerateInput["onTraceEvent"]; onAuditSummary?: GenerateInput["onAuditSummary"]; } @@ -72,6 +74,10 @@ export async function inspectQualityWithLlm( }), temperature: 0.1, task: "quality_inspector", + schemaName: "llmQaPatchSchema", + traceStage: "qa", + rewriteRound: input.rewriteRound, + onTraceEvent: input.onTraceEvent, onAuditSummary: input.onAuditSummary, }); diff --git a/src/lib/workflow/streaming-optimizer.ts b/src/lib/workflow/streaming-optimizer.ts index a074347..755ee8f 100644 --- a/src/lib/workflow/streaming-optimizer.ts +++ b/src/lib/workflow/streaming-optimizer.ts @@ -13,6 +13,7 @@ export interface RunStreamingOptimizationWorkflowInput { input: ArticleInput; factCard: OptimizationFactCard; onEvent: (event: OptimizationStreamEvent) => void | Promise; + onTraceEvent?: GenerateInput["onTraceEvent"]; onAuditSummary?: GenerateInput["onAuditSummary"]; } @@ -21,6 +22,7 @@ export async function runStreamingOptimizationWorkflow({ input, factCard, onEvent, + onTraceEvent, onAuditSummary, }: RunStreamingOptimizationWorkflowInput) { const processSummary: ProcessSummaryStep[] = []; @@ -31,7 +33,12 @@ export async function runStreamingOptimizationWorkflow({ message: "正在生成优化草稿", }); let stageStartedAt = Date.now(); - let article = await optimizeArticle({ input, factCard, onAuditSummary }); + let article = await optimizeArticle({ + input, + factCard, + onTraceEvent, + onAuditSummary, + }); processSummary.push( createProcessStep({ stage: "draft", @@ -54,6 +61,8 @@ export async function runStreamingOptimizationWorkflow({ factCard, platform: input.platform, sourceImages: input.images, + rewriteRound: 0, + onTraceEvent, onAuditSummary, }); processSummary.push( @@ -81,6 +90,8 @@ export async function runStreamingOptimizationWorkflow({ article, factCard, failedChecks, + rewriteRound: nextRound, + onTraceEvent, onAuditSummary, }); rewriteRounds = nextRound; @@ -112,6 +123,8 @@ export async function runStreamingOptimizationWorkflow({ factCard, platform: input.platform, sourceImages: input.images, + rewriteRound: nextRound, + onTraceEvent, onAuditSummary, }); processSummary.push( diff --git a/src/lib/workflow/targeted-rewriter.ts b/src/lib/workflow/targeted-rewriter.ts index d4d6654..dfe896e 100644 --- a/src/lib/workflow/targeted-rewriter.ts +++ b/src/lib/workflow/targeted-rewriter.ts @@ -10,6 +10,8 @@ export interface RewriteFailedSectionsInput { article: OptimizedArticle; factCard: OptimizationFactCard; failedChecks: QaCheck[]; + rewriteRound?: number; + onTraceEvent?: GenerateInput["onTraceEvent"]; onAuditSummary?: GenerateInput["onAuditSummary"]; } @@ -17,6 +19,8 @@ export async function rewriteFailedSections({ article, factCard, failedChecks, + rewriteRound, + onTraceEvent, onAuditSummary, }: RewriteFailedSectionsInput): Promise { const llmArticle = await generateValidatedJson({ @@ -25,6 +29,10 @@ export async function rewriteFailedSections({ prompt: buildTargetedRewritePrompt({ article, factCard, failedChecks }), temperature: 0.15, task: "targeted_rewriter", + schemaName: "optimizedArticleSchema", + traceStage: "rewrite", + rewriteRound, + onTraceEvent, onAuditSummary, });