feat(agent): 实现流式输出取消功能
- 添加 AbortController 管理,支持取消正在进行的请求 - Agent.chat() 新增 abortSignal 参数,传递给 streamText/generateText - 取消时保存用户消息和 AI 已输出的部分内容 - cancelProcessing 实际调用 abort() 中止流式请求
This commit is contained in:
@@ -67,13 +67,21 @@ interface SessionManagerConstructor {
|
||||
new (): SessionManagerInstance;
|
||||
}
|
||||
|
||||
/**
|
||||
* Chat 选项接口
|
||||
*/
|
||||
interface ChatOptions {
|
||||
onStream?: (chunk: string) => void;
|
||||
abortSignal?: AbortSignal;
|
||||
}
|
||||
|
||||
/**
|
||||
* Agent 实例接口
|
||||
*/
|
||||
interface AgentInstance {
|
||||
setRegistry(registry: unknown): void;
|
||||
setSessionManager(manager: SessionManagerInstance): void;
|
||||
chat(message: string, onStream?: (chunk: string) => void): Promise<ChatResult>;
|
||||
chat(message: string, options?: ChatOptions): Promise<ChatResult>;
|
||||
getToolCount(): { core: number; discovered: number; total: number };
|
||||
getContextUsageFormatted(): string;
|
||||
getContextUsage(): TokenUsage;
|
||||
@@ -149,6 +157,9 @@ const agentCache: Map<string, AgentInstance> = new Map();
|
||||
// SessionManager 实例缓存(每个 session 一个)
|
||||
const sessionManagerCache: Map<string, SessionManagerInstance> = new Map();
|
||||
|
||||
// AbortController 缓存(每个 session 一个,用于取消正在进行的请求)
|
||||
const abortControllerCache: Map<string, AbortController> = new Map();
|
||||
|
||||
// 配置错误缓存(用于向客户端返回友好错误)
|
||||
let lastConfigError: { provider: string; message: string } | null = null;
|
||||
|
||||
@@ -313,6 +324,16 @@ export async function destroyAgent(sessionId: string): Promise<void> {
|
||||
export async function processMessage(sessionId: string, content: string): Promise<void> {
|
||||
const sessionManager = getSessionManager();
|
||||
|
||||
// 取消之前可能存在的请求
|
||||
const existingController = abortControllerCache.get(sessionId);
|
||||
if (existingController) {
|
||||
existingController.abort();
|
||||
}
|
||||
|
||||
// 创建新的 AbortController
|
||||
const abortController = new AbortController();
|
||||
abortControllerCache.set(sessionId, abortController);
|
||||
|
||||
// 更新状态
|
||||
sessionManager.updateStatus(sessionId, 'busy' as SessionStatus);
|
||||
emitStatusEvent(sessionId, 'processing', { message: '正在处理...' });
|
||||
@@ -338,6 +359,7 @@ export async function processMessage(sessionId: string, content: string): Promis
|
||||
emitLogEvent(sessionId, 'error', `配置错误: ${lastConfigError.message}`);
|
||||
sessionManager.updateStatus(sessionId, 'idle' as SessionStatus);
|
||||
emitStatusEvent(sessionId, 'idle');
|
||||
abortControllerCache.delete(sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -357,26 +379,33 @@ export async function processMessage(sessionId: string, content: string): Promis
|
||||
|
||||
sessionManager.updateStatus(sessionId, 'idle' as SessionStatus);
|
||||
emitStatusEvent(sessionId, 'idle');
|
||||
abortControllerCache.delete(sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// 调用 Agent 的 chat 方法,使用流式回调
|
||||
const result = await agent.chat(content, (chunk: string) => {
|
||||
// 推送流式内容
|
||||
broadcastToSession(sessionId, {
|
||||
type: 'chunk',
|
||||
sessionId,
|
||||
payload: { content: chunk },
|
||||
});
|
||||
// 调用 Agent 的 chat 方法,使用流式回调和 AbortSignal
|
||||
const result = await agent.chat(content, {
|
||||
onStream: (chunk: string) => {
|
||||
// 检查是否已取消
|
||||
if (abortController.signal.aborted) return;
|
||||
|
||||
// 检测工具调用
|
||||
if (chunk.includes('[调用工具:')) {
|
||||
const match = chunk.match(/\[调用工具: (.+?)\]/);
|
||||
if (match) {
|
||||
emitLogEvent(sessionId, 'info', `调用工具: ${match[1]}`);
|
||||
// 推送流式内容
|
||||
broadcastToSession(sessionId, {
|
||||
type: 'chunk',
|
||||
sessionId,
|
||||
payload: { content: chunk },
|
||||
});
|
||||
|
||||
// 检测工具调用
|
||||
if (chunk.includes('[调用工具:')) {
|
||||
const match = chunk.match(/\[调用工具: (.+?)\]/);
|
||||
if (match) {
|
||||
emitLogEvent(sessionId, 'info', `调用工具: ${match[1]}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
abortSignal: abortController.signal,
|
||||
});
|
||||
|
||||
// 消息已由 Core Agent 自动持久化,这里只更新 Server 端的会话计数
|
||||
@@ -429,6 +458,8 @@ export async function processMessage(sessionId: string, content: string): Promis
|
||||
|
||||
emitLogEvent(sessionId, 'error', errorMessage);
|
||||
} finally {
|
||||
// 清理 AbortController
|
||||
abortControllerCache.delete(sessionId);
|
||||
sessionManager.updateStatus(sessionId, 'idle' as SessionStatus);
|
||||
}
|
||||
}
|
||||
@@ -437,11 +468,23 @@ export async function processMessage(sessionId: string, content: string): Promis
|
||||
* 取消正在进行的处理
|
||||
*/
|
||||
export function cancelProcessing(sessionId: string): void {
|
||||
// TODO: 实现取消逻辑
|
||||
// 目前 AI SDK 的 streamText 不支持取消
|
||||
// 获取并中止 AbortController
|
||||
const controller = abortControllerCache.get(sessionId);
|
||||
if (controller) {
|
||||
controller.abort();
|
||||
abortControllerCache.delete(sessionId);
|
||||
}
|
||||
|
||||
const sessionManager = getSessionManager();
|
||||
sessionManager.updateStatus(sessionId, 'idle' as SessionStatus);
|
||||
emitStatusEvent(sessionId, 'cancelled');
|
||||
|
||||
// 通知客户端已取消
|
||||
broadcastToSession(sessionId, {
|
||||
type: 'cancelled',
|
||||
sessionId,
|
||||
payload: { message: '用户取消了请求' },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user