feat(lpt-ai):异步任务队列 + 提交/轮询路由

lpt-ai 新增异步任务处理能力,LLM 调用改为后台 Worker 顺序执行:

新增 src/task-queue.ts:
- createTask(type, params) → taskId(randomUUID)
- getTask(taskId) → { status, result?, error? }
- 单线程 Worker(setImmediate 链),同一时刻只处理一个 LLM 调用
- 按类型分派到对应的 prompt 函数
- TTL 清理:5 分钟/次,移除完成超过 1 小时的记录
- 异常隔离 + logStore 集成

修改 src/routes/ai.ts:
- POST /ai/tasks:接收 { type, params },返回 201 { taskId }
- GET /ai/tasks/:taskId:返回任务状态与结果(pending/running/done/failed)
- 旧同步端点完全不动,向后兼容
This commit is contained in:
2026-07-04 16:38:48 +08:00
parent e549395e99
commit 438045b57a
2 changed files with 224 additions and 0 deletions
+28
View File
@@ -6,6 +6,7 @@ import type { FastifyInstance } from "fastify";
import { chat, isAvailable, loadConfig, LlmError } from "../llm/client.js"; import { chat, isAvailable, loadConfig, LlmError } from "../llm/client.js";
import { aggregateReportPrompt, compareRecallPrompt, generateMindMapPrompt, type FragmentInput } from "../llm/prompts.js"; import { aggregateReportPrompt, compareRecallPrompt, generateMindMapPrompt, type FragmentInput } from "../llm/prompts.js";
import { logStore } from "../admin/store.js"; import { logStore } from "../admin/store.js";
import { createTask, getTask, type TaskType } from "../task-queue.js";
interface AggregateReportBody { interface AggregateReportBody {
taskName: string; taskName: string;
@@ -173,4 +174,31 @@ export async function aiRoutes(app: FastifyInstance): Promise<void> {
throw err; throw err;
} }
}); });
// ============ 异步任务(提交 + 轮询)============
/** 提交 AI 任务,立即返回 taskId */
app.post<{ Body: { type: string; params: Record<string, unknown> } }>("/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<string, unknown>);
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;
});
} }
+196
View File
@@ -0,0 +1,196 @@
/**
* 异步任务队列 —— 接收任务,后台 Worker 顺序处理,提供查询。
*
* 设计要点:
* - 单线程 WorkersetImmediate 链),同一时刻只处理一个 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<string, unknown>;
status: TaskStatus;
result?: unknown;
error?: string;
createdAt: string; // ISO 8601
startedAt?: string;
completedAt?: string;
}
// ---- 存储 ----
const records = new Map<string, TaskRecord>();
const queue: TaskRecord[] = [];
let processing = false;
// ---- 公开 API ----
export function createTask(type: TaskType, params: Record<string, unknown>): 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<void> {
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);