陈子默
588e810c51
- 编译知识库英文运行名、描述、检索配置和独立 Registration - 统一最终分数阈值并保持模型上下文、检索事件与引用一致 - 完善 AG-UI 知识库检索运行态与完成态展示
308 lines
9.3 KiB
TypeScript
308 lines
9.3 KiB
TypeScript
import type { Message } from '@ag-ui/client';
|
||
|
||
import { EventSchemas, EventType, randomUUID } from '@ag-ui/client';
|
||
import { events } from 'fetch-event-stream';
|
||
|
||
import { createEventStreamHeaders, resolveApiUrl } from '#/api/request';
|
||
|
||
import { easyFlowAguiCustomEvent } from './custom-events';
|
||
|
||
export interface EasyFlowAguiRunOptions {
|
||
forwardedProps?: Record<string, unknown>;
|
||
onCursor?: (cursor: number) => void;
|
||
onEvent: (event: AguiEvent) => Promise<void> | void;
|
||
runId?: string;
|
||
threadId: string;
|
||
url: string;
|
||
userMessage: Message;
|
||
}
|
||
|
||
export interface EasyFlowAguiResumeOptions {
|
||
afterCursor?: number;
|
||
onCursor?: (cursor: number) => void;
|
||
onEvent: (event: AguiEvent) => Promise<void> | void;
|
||
runId: string;
|
||
}
|
||
|
||
export type AguiEvent = ReturnType<(typeof EventSchemas)['parse']>;
|
||
|
||
interface ActiveAguiRun {
|
||
aborted: boolean;
|
||
abort: () => void;
|
||
}
|
||
|
||
const TOOL_START_PAINT_FALLBACK_MS = 50;
|
||
|
||
/** 带 HTTP 状态码的 AG-UI 传输异常。 */
|
||
export class EasyFlowAguiHttpError extends Error {
|
||
constructor(
|
||
message: string,
|
||
public readonly status: number,
|
||
) {
|
||
super(message);
|
||
this.name = 'EasyFlowAguiHttpError';
|
||
}
|
||
}
|
||
|
||
/** 页面事件投影失败，重放同一游标无法自行恢复。 */
|
||
export class EasyFlowAguiProjectionError extends Error {
|
||
constructor(public readonly projectionCause: unknown) {
|
||
super(
|
||
projectionCause instanceof Error
|
||
? projectionCause.message
|
||
: 'Agent 事件投影失败',
|
||
);
|
||
this.name = 'EasyFlowAguiProjectionError';
|
||
}
|
||
}
|
||
|
||
async function projectEvent(
|
||
onEvent: (event: AguiEvent) => Promise<void> | void,
|
||
event: AguiEvent,
|
||
) {
|
||
try {
|
||
await onEvent(event);
|
||
} catch (error) {
|
||
throw new EasyFlowAguiProjectionError(error);
|
||
}
|
||
}
|
||
|
||
function toTransportJson<T>(value: T): T {
|
||
// AG-UI SDK 会在请求前 structuredClone；先以真实传输格式解除 Vue Proxy，避免草稿对象克隆失败。
|
||
// eslint-disable-next-line unicorn/prefer-structured-clone -- structuredClone 无法复制 Proxy
|
||
return JSON.parse(JSON.stringify(value)) as T;
|
||
}
|
||
|
||
/**
|
||
* 在工具开始事件后让出一次页面绘制机会。
|
||
*
|
||
* <p>AG-UI 可能在同一批响应数据中连续送达工具开始和结果事件。Vue 虽已收到开始态，
|
||
* 浏览器仍可能在首次绘制前就将其覆盖为完成态。这里使用一帧回调并设置有界兜底，
|
||
* 既确保前台页面能展示“调用中”，也避免后台标签页暂停动画帧时阻塞事件流。</p>
|
||
*
|
||
* @returns 页面获得绘制机会后完成的 Promise
|
||
*/
|
||
function waitForToolStartPaint(): Promise<void> {
|
||
return new Promise((resolve) => {
|
||
let settled = false;
|
||
const complete = () => {
|
||
if (settled) return;
|
||
settled = true;
|
||
clearTimeout(fallbackTimer);
|
||
resolve();
|
||
};
|
||
const fallbackTimer = setTimeout(complete, TOOL_START_PAINT_FALLBACK_MS);
|
||
if (typeof globalThis.requestAnimationFrame !== 'function') {
|
||
setTimeout(complete, 0);
|
||
return;
|
||
}
|
||
globalThis.requestAnimationFrame(() => {
|
||
// requestAnimationFrame 回调发生在绘制前，再让出一个任务周期后继续消费后续事件。
|
||
setTimeout(complete, 0);
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 判断事件是否开启了需要即时呈现的工具执行状态。
|
||
*
|
||
* @param event AG-UI 事件
|
||
* @returns 标准工具开始或知识库检索开始时为 true
|
||
*/
|
||
function startsVisibleToolExecution(event: AguiEvent) {
|
||
if (event.type === EventType.TOOL_CALL_START) {
|
||
return true;
|
||
}
|
||
if (
|
||
event.type !== EventType.CUSTOM ||
|
||
event.name !== easyFlowAguiCustomEvent.knowledgeRetrievalStatus
|
||
) {
|
||
return false;
|
||
}
|
||
const value =
|
||
event.value &&
|
||
typeof event.value === 'object' &&
|
||
!Array.isArray(event.value)
|
||
? (event.value as Record<string, unknown>)
|
||
: {};
|
||
return String(value.status || '').toLowerCase() === 'running';
|
||
}
|
||
|
||
/**
|
||
* EasyFlow 的无头 AG-UI 运行客户端。
|
||
*
|
||
* <p>客户端以最小标准 AG-UI 请求体启动运行，页面只消费投影回调。出站边界固定裁剪
|
||
* tools、context、state 和历史消息，服务端仍会独立执行同样的安全校验。</p>
|
||
*/
|
||
export class EasyFlowAguiClient {
|
||
private activeRun?: ActiveAguiRun;
|
||
|
||
abort() {
|
||
this.detach();
|
||
}
|
||
|
||
async cancel(runId: string) {
|
||
const requestUrl = `/api/v1/agent/agui/run/${encodeURIComponent(runId)}/cancel`;
|
||
const response = await fetch(resolveApiUrl(requestUrl), {
|
||
headers: createEventStreamHeaders(requestUrl),
|
||
method: 'POST',
|
||
});
|
||
if (!response.ok) {
|
||
throw new EasyFlowAguiHttpError(
|
||
await responseError(response),
|
||
response.status,
|
||
);
|
||
}
|
||
const body = (await response.json()) as {
|
||
errorCode?: number;
|
||
message?: string;
|
||
};
|
||
if (body.errorCode !== undefined && body.errorCode !== 0) {
|
||
throw new Error(body.message || '停止 Agent 运行失败');
|
||
}
|
||
this.detach();
|
||
}
|
||
|
||
detach() {
|
||
if (!this.activeRun) return;
|
||
this.activeRun.aborted = true;
|
||
this.activeRun.abort();
|
||
this.activeRun = undefined;
|
||
}
|
||
|
||
async resume(options: EasyFlowAguiResumeOptions) {
|
||
this.detach();
|
||
const controller = new AbortController();
|
||
const activeRun: ActiveAguiRun = {
|
||
abort: () => controller.abort(),
|
||
aborted: false,
|
||
};
|
||
this.activeRun = activeRun;
|
||
let terminalReceived = false;
|
||
try {
|
||
const requestUrl = `/api/v1/agent/agui/run/${encodeURIComponent(options.runId)}/events?after=${Math.max(0, options.afterCursor || 0)}`;
|
||
const response = await fetch(resolveApiUrl(requestUrl), {
|
||
headers: createEventStreamHeaders(requestUrl),
|
||
method: 'GET',
|
||
signal: controller.signal,
|
||
});
|
||
await assertSseResponse(response);
|
||
for await (const message of events(response, controller.signal)) {
|
||
if (activeRun.aborted || !message.data) continue;
|
||
const event = EventSchemas.parse(JSON.parse(message.data));
|
||
await projectEvent(options.onEvent, event as AguiEvent);
|
||
const cursor = Number(message.id || 0);
|
||
if (Number.isSafeInteger(cursor) && cursor > 0) {
|
||
options.onCursor?.(cursor);
|
||
}
|
||
if (
|
||
event.type === EventType.RUN_FINISHED ||
|
||
event.type === EventType.RUN_ERROR
|
||
) {
|
||
terminalReceived = true;
|
||
}
|
||
if (startsVisibleToolExecution(event as AguiEvent)) {
|
||
await waitForToolStartPaint();
|
||
}
|
||
}
|
||
if (!activeRun.aborted && !terminalReceived) {
|
||
throw new Error('Agent 事件流缺少终态，请重试');
|
||
}
|
||
} catch (error) {
|
||
if (activeRun.aborted) return;
|
||
throw error;
|
||
} finally {
|
||
if (this.activeRun === activeRun) {
|
||
this.activeRun = undefined;
|
||
}
|
||
}
|
||
}
|
||
|
||
async run(options: EasyFlowAguiRunOptions) {
|
||
this.detach();
|
||
const requestUrl = options.url;
|
||
const controller = new AbortController();
|
||
const activeRun: ActiveAguiRun = {
|
||
abort: () => controller.abort(),
|
||
aborted: false,
|
||
};
|
||
this.activeRun = activeRun;
|
||
let terminalReceived = false;
|
||
try {
|
||
const response = await fetch(resolveApiUrl(requestUrl), {
|
||
body: JSON.stringify({
|
||
context: [],
|
||
forwardedProps: options.forwardedProps
|
||
? toTransportJson(options.forwardedProps)
|
||
: undefined,
|
||
messages: [toTransportJson(options.userMessage)],
|
||
runId: options.runId || createAguiRunId(),
|
||
state: {},
|
||
threadId: options.threadId,
|
||
tools: [],
|
||
}),
|
||
headers: createEventStreamHeaders(requestUrl),
|
||
method: 'POST',
|
||
signal: controller.signal,
|
||
});
|
||
await assertSseResponse(response);
|
||
for await (const message of events(response, controller.signal)) {
|
||
if (activeRun.aborted || !message.data) continue;
|
||
const event = EventSchemas.parse(JSON.parse(message.data));
|
||
if (
|
||
event.type === EventType.RUN_FINISHED ||
|
||
event.type === EventType.RUN_ERROR
|
||
) {
|
||
terminalReceived = true;
|
||
}
|
||
await projectEvent(options.onEvent, event as AguiEvent);
|
||
const cursor = Number(message.id || 0);
|
||
if (Number.isSafeInteger(cursor) && cursor > 0) {
|
||
options.onCursor?.(cursor);
|
||
}
|
||
if (startsVisibleToolExecution(event as AguiEvent)) {
|
||
await waitForToolStartPaint();
|
||
}
|
||
}
|
||
if (!terminalReceived) {
|
||
if (activeRun.aborted) return;
|
||
throw new Error('Agent 事件流缺少终态，请重试');
|
||
}
|
||
} catch (error) {
|
||
if (activeRun.aborted) return;
|
||
if (terminalReceived && !(error instanceof EasyFlowAguiProjectionError)) {
|
||
return;
|
||
}
|
||
throw error;
|
||
} finally {
|
||
if (this.activeRun === activeRun) {
|
||
this.activeRun = undefined;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
export function createAguiRunId() {
|
||
return `run_${randomUUID()}`;
|
||
}
|
||
|
||
async function assertSseResponse(response: Response) {
|
||
const contentType = response.headers.get('content-type') || '';
|
||
if (!response.ok || !contentType.includes('text/event-stream')) {
|
||
throw new EasyFlowAguiHttpError(
|
||
await responseError(response),
|
||
response.status,
|
||
);
|
||
}
|
||
}
|
||
|
||
async function responseError(response: Response) {
|
||
try {
|
||
const body = await response.clone().json();
|
||
return body?.message || body?.error || `HTTP ${response.status}`;
|
||
} catch {
|
||
const text = await response.text();
|
||
return text.trim() || `HTTP ${response.status}`;
|
||
}
|
||
}
|