diff --git a/src/routes/ai.ts b/src/routes/ai.ts index fb191b7..28d5272 100644 --- a/src/routes/ai.ts +++ b/src/routes/ai.ts @@ -6,6 +6,7 @@ import type { FastifyInstance } from "fastify"; import { chat, isAvailable, loadConfig, LlmError } from "../llm/client.js"; import { aggregateReportPrompt, compareRecallPrompt, generateMindMapPrompt, type FragmentInput } from "../llm/prompts.js"; import { logStore } from "../admin/store.js"; +import { createTask, getTask, type TaskType } from "../task-queue.js"; interface AggregateReportBody { taskName: string; @@ -173,4 +174,31 @@ export async function aiRoutes(app: FastifyInstance): Promise { throw err; } }); + + // ============ 异步任务(提交 + 轮询)============ + + /** 提交 AI 任务,立即返回 taskId */ + app.post<{ Body: { type: string; params: Record } }>("/ai/tasks", async (request, reply) => { + const { type, params } = request.body ?? {}; + + if (!type || !["aggregate-report", "generate-mind-map", "compare-recall"].includes(type)) { + return reply.status(400).send({ error: "无效的 type,允许: aggregate-report, generate-mind-map, compare-recall" }); + } + if (!params || typeof params !== "object") { + return reply.status(400).send({ error: "params 必须为非空对象" }); + } + + const taskId = createTask(type as TaskType, params as Record); + return reply.status(201).send({ taskId }); + }); + + /** 查询任务状态与结果 */ + app.get<{ Params: { taskId: string } }>("/ai/tasks/:taskId", async (request, reply) => { + const { taskId } = request.params; + const task = getTask(taskId); + if (!task) { + return reply.status(404).send({ error: "任务不存在" }); + } + return task; + }); } diff --git a/src/task-queue.ts b/src/task-queue.ts new file mode 100644 index 0000000..03594ef --- /dev/null +++ b/src/task-queue.ts @@ -0,0 +1,196 @@ +/** + * 异步任务队列 —— 接收任务,后台 Worker 顺序处理,提供查询。 + * + * 设计要点: + * - 单线程 Worker(setImmediate 链),同一时刻只处理一个 LLM 调用 + * - TTL 清理:5 分钟一次,移除完成超过 1 小时的记录 + * - 异常隔离:单个任务失败不影响后续任务 + * - 日志复用 logStore,与管理面板日志统一 + */ + +import { randomUUID } from "node:crypto"; +import { chat, isAvailable, loadConfig, LlmError } from "./llm/client.js"; +import { aggregateReportPrompt, compareRecallPrompt, generateMindMapPrompt } from "./llm/prompts.js"; +import { logStore } from "./admin/store.js"; + +// ---- 类型 ---- + +export type TaskType = "aggregate-report" | "generate-mind-map" | "compare-recall"; + +export type TaskStatus = "pending" | "running" | "done" | "failed"; + +export interface TaskRecord { + id: string; + type: TaskType; + params: Record; + status: TaskStatus; + result?: unknown; + error?: string; + createdAt: string; // ISO 8601 + startedAt?: string; + completedAt?: string; +} + +// ---- 存储 ---- + +const records = new Map(); +const queue: TaskRecord[] = []; +let processing = false; + +// ---- 公开 API ---- + +export function createTask(type: TaskType, params: Record): string { + const id = randomUUID(); + const task: TaskRecord = { + id, + type, + params, + status: "pending", + createdAt: new Date().toISOString(), + }; + records.set(id, task); + queue.push(task); + setImmediate(() => tryProcessNext()); + return id; +} + +export function getTask(id: string): TaskRecord | undefined { + const t = records.get(id); + if (!t) return undefined; + // 返回浅拷贝,防止外部修改 + return { ...t }; +} + +// ---- 后台 Worker ---- + +async function tryProcessNext(): Promise { + if (processing || queue.length === 0) return; + processing = true; + + const task = queue.shift()!; + + try { + task.status = "running"; + task.startedAt = new Date().toISOString(); + + const config = loadConfig(); + + if (!isAvailable(config)) { + throw new LlmError("LLM API Key 未配置", 503); + } + + const messages = buildMessages(task); + let callOptions: { temperature?: number; maxTokens?: number } | undefined; + if (task.type === "compare-recall") { + callOptions = { temperature: 0.2, maxTokens: 4096 }; + } else if (task.type === "generate-mind-map") { + callOptions = { temperature: 0.3, maxTokens: 4096 }; + } else { + callOptions = { temperature: 0.3, maxTokens: 2048 }; + } + + const raw = await chat(config, messages, callOptions); + + let result: unknown; + if (task.type === "aggregate-report") { + result = { report: raw.trim(), model: config.model }; + } else if (task.type === "generate-mind-map") { + result = { outline: raw.trim(), model: config.model }; + } else { + // compare-recall: 清理可能的 ```json 包裹 + const clean = raw.trim().replace(/^```(?:json)?\s*/, "").replace(/\s*```$/, ""); + result = JSON.parse(clean); + } + + task.status = "done"; + task.result = result; + task.completedAt = new Date().toISOString(); + + // 记录到管理面板日志 + logStore.add({ + timestamp: new Date().toISOString(), + endpoint: `/ai/tasks#${task.type}`, + model: config.model, + durationMs: task.startedAt ? Date.parse(task.completedAt) - Date.parse(task.startedAt) : 0, + status: 200, + requestBody: task.params, + responseBody: result, + messages, + params: callOptions, + }); + + } catch (err: unknown) { + const errMsg = err instanceof Error ? err.message : String(err); + task.status = "failed"; + task.error = errMsg; + task.completedAt = new Date().toISOString(); + + const config = loadConfig(); + logStore.add({ + timestamp: new Date().toISOString(), + endpoint: `/ai/tasks#${task.type}`, + model: config.model, + durationMs: task.startedAt ? Date.parse(task.completedAt) - Date.parse(task.startedAt) : 0, + status: err instanceof LlmError ? err.statusCode : 500, + requestBody: task.params, + responseBody: {}, + error: errMsg, + }); + } finally { + processing = false; + // 处理下一个任务 + if (queue.length > 0) { + setImmediate(() => tryProcessNext()); + } + } +} + +// ---- Prompt 构建 ---- + +function buildMessages(task: TaskRecord) { + const p = task.params; + switch (task.type) { + case "aggregate-report": { + const fragments = p.fragments as Array<{ content: string }> | undefined; + return aggregateReportPrompt( + p.taskName as string, + fragments ?? [], + p.expectation as string | undefined, + ); + } + case "generate-mind-map": { + const reports = p.reports as string[] | undefined; + const fragments = p.fragments as string[] | undefined; + return generateMindMapPrompt( + p.taskName as string, + (p.taskDescription as string) ?? "", + reports ?? [], + fragments ?? [], + ); + } + case "compare-recall": { + return compareRecallPrompt( + p.taskName as string, + p.standardOutline as string, + p.recallOutline as string, + ); + } + } +} + +// ---- TTL 清理 ---- + +const TTL_MS = 60 * 60 * 1000; // 1 小时 +const CLEANUP_INTERVAL_MS = 5 * 60 * 1000; // 5 分钟 + +setInterval(() => { + const cutoff = Date.now() - TTL_MS; + for (const [id, task] of records) { + if (task.status === "done" || task.status === "failed") { + const completed = task.completedAt ? Date.parse(task.completedAt) : 0; + if (completed > 0 && completed < cutoff) { + records.delete(id); + } + } + } +}, CLEANUP_INTERVAL_MS);