agent-Specialization/docs/analysis-task-recovery.md

693 lines
20 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# AI Agent 系统任务恢复与推送机制分析
## 概述
本文档详细分析 AI Agent 系统中**正在运行中的对话恢复机制和推送机制**,重点关注页面刷新后如何恢复任务状态、重放历史事件以及维护 UI 一致性。
---
## 1. 页面刷新后的任务恢复流程
### 1.1 恢复触发时机
任务恢复在页面挂载时立即触发:
**文件**: `/static/src/app/lifecycle.ts`
```typescript
export async function mounted() {
// ... 其他初始化代码
// 注册全局事件处理器(用于任务轮询)
(window as any).__taskEventHandler = (event: any) => {
if (typeof this.handleTaskEvent === 'function') {
this.handleTaskEvent(event);
}
};
// 立即尝试恢复运行中的任务(不延迟)
if (typeof this.restoreTaskState === 'function') {
this.restoreTaskState();
}
}
```
**关键点**:
-`mounted()` 生命周期钩子中立即调用 `restoreTaskState()`
- 不延迟执行,确保用户刷新后能快速看到任务恢复
- 先注册全局事件处理器 `__taskEventHandler`,用于后续轮询事件分发
### 1.2 查找运行中的任务
**文件**: `/static/src/stores/task.ts` - `loadRunningTask()`
```typescript
async loadRunningTask(conversationId: string | null = null) {
try {
debugLog('[Task] 查找运行中的任务');
// 1. 获取所有任务列表
const response = await fetch('/api/tasks');
const result = await response.json();
// 2. 查找运行中的任务可选按对话ID过滤
const runningTask = result.data.find((task: any) =>
task.status === 'running' &&
(!conversationId || task.conversation_id === conversationId)
);
if (runningTask) {
// 3. 更新任务状态
this.currentTaskId = runningTask.task_id;
this.taskStatus = runningTask.status;
this.taskCreatedAt = runningTask.created_at;
this.taskUpdatedAt = runningTask.updated_at;
// 4. 获取任务详情,计算已处理的事件数量
const detailResponse = await fetch(`/api/tasks/${runningTask.task_id}`);
if (detailResponse.ok) {
const detailResult = await detailResponse.json();
if (detailResult.success && detailResult.data.events) {
// 设置为当前事件数量,只获取新事件
this.lastEventIndex = detailResult.data.next_offset || detailResult.data.events.length;
debugLog('[Task] 设置起始偏移量:', this.lastEventIndex);
}
}
return runningTask;
}
return null;
} catch (error) {
console.error('[Task] 加载运行中任务失败:', error);
return null;
}
}
```
**关键逻辑**:
1. 调用 `/api/tasks` 获取用户的所有任务
2. 筛选 `status === 'running'` 且匹配当前对话ID的任务
3. 获取任务详情,计算 `lastEventIndex`(已处理的事件数量)
4. 返回运行中的任务信息
---
## 2. `_rebuildingFromScratch` 标志机制
### 2.1 标志的作用
`_rebuildingFromScratch` 是一个关键的布尔标志,用于区分两种恢复模式:
- **从头重建模式** (`true`): 历史消息不完整或缺失,需要从头重放所有事件来重建 assistant 响应
- **增量恢复模式** (`false`): 历史消息完整,只需恢复状态并继续轮询新事件
### 2.2 何时设置为 true
**文件**: `/static/src/app/methods/taskPolling.ts` - `restoreTaskState()`
```typescript
// 检查是否需要从头重建
const historyActionsCount = lastMessage?.actions?.length || 0;
const eventCount = allEvents.length;
const historyIncomplete = eventCount > historyActionsCount + 5; // 允许5个事件的误差
const needsRebuild = !isAssistantMessage ||
(isAssistantMessage && (!lastMessage.actions || lastMessage.actions.length === 0)) ||
historyIncomplete;
if (needsRebuild) {
debugLog('[TaskPolling] 需要从头重建 assistant 响应');
// 清空不完整的 assistant 消息
if (isAssistantMessage) {
this.messages.pop();
}
// 重置偏移量为 0从头获取所有事件
taskStore.lastEventIndex = 0;
// 标记正在从头重建
this._rebuildingFromScratch = true;
this._rebuildingEventCount = allEvents.length;
// 启动轮询
taskStore.startPolling((event: any) => {
this.handleTaskEvent(event);
});
// 延迟清除重建标记2秒后
setTimeout(() => {
this._rebuildingFromScratch = false;
this._rebuildingEventCount = 0;
}, 2000);
}
```
**触发条件**:
1. **最后一条消息不是 assistant 消息**: 说明历史加载有问题
2. **最后一条是空的 assistant 消息**: `actions` 数组为空或不存在
3. **历史不完整**: 事件数量远大于历史中的 actions 数量允许5个事件的误差
### 2.3 对事件处理的影响
`_rebuildingFromScratch` 标志会影响多个事件处理函数的行为:
#### 2.3.1 思考块处理
```typescript
handleThinkingStart(data: any, eventIdx: number) {
// ... 创建思考块
// 只在非历史恢复时展开思考块
if (!this._rebuildingFromScratch) {
this.chatExpandBlock(blockId);
}
}
handleThinkingEnd(data: any) {
// ... 完成思考块
// 只在非历史恢复时延迟折叠思考块
if (!this._rebuildingFromScratch) {
setTimeout(() => {
this.chatCollapseBlock(blockId);
}, 1000);
} else {
// 历史恢复时立即折叠,不需要动画
this.chatCollapseBlock(blockId);
}
}
```
**差异**:
- **从头重建**: 不展开思考块,立即折叠,跳过动画
- **正常流式**: 展开思考块延迟1秒后折叠有动画效果
#### 2.3.2 工具意图打字机效果
```typescript
handleToolIntent(data: any) {
const isHistoryRestore = this._rebuildingFromScratch;
if (isHistoryRestore) {
// 历史恢复,直接显示完整 intent
action.tool.intent_rendered = newIntent;
} else {
// 新工具块,逐字符显示(打字机效果)
action.tool.intent_rendered = '';
action.tool._intentTyping = true;
// 逐字符显示动画
const typeNextChar = () => {
if (charIndex < newIntent.length && action.tool._intentTyping) {
action.tool.intent_rendered += newIntent[charIndex];
charIndex++;
this.$forceUpdate();
action.tool._intentTimer = setTimeout(typeNextChar, charInterval);
}
};
action.tool._intentTimer = setTimeout(typeNextChar, 50);
}
}
```
**差异**:
- **从头重建**: 直接显示完整 intent无打字机效果
- **正常流式**: 逐字符显示,有打字机动画
#### 2.3.3 AI 消息开始处理
```typescript
handleAiMessageStart(data: any, eventIdx: number) {
// 如果是从头重建,标记消息为静默恢复
if (this._rebuildingFromScratch) {
const newMessage = this.messages[this.messages.length - 1];
if (newMessage && newMessage.role === 'assistant') {
newMessage.awaitingFirstContent = false;
newMessage.generatingLabel = '';
}
}
}
```
**差异**:
- **从头重建**: 清除"生成中"标签,静默恢复
- **正常流式**: 显示"生成中"动画
---
## 3. 历史事件的重放机制
### 3.1 事件获取
**后端**: `/server/tasks.py` - `get_task_api()`
```python
@tasks_bp.route("/api/tasks/<task_id>", methods=["GET"])
@api_login_required
def get_task_api(task_id: str):
username = get_current_username()
rec = task_manager.get_task(username, task_id)
if not rec:
return jsonify({"success": False, "error": "任务不存在"}), 404
# 获取偏移量参数
try:
offset = int(request.args.get("from", 0))
except Exception:
offset = 0
# 过滤事件(只返回 idx >= offset 的事件)
events = [e for e in rec.events if e["idx"] >= offset]
next_offset = events[-1]["idx"] + 1 if events else offset
return jsonify({
"success": True,
"data": {
"task_id": rec.task_id,
"status": rec.status,
"events": events,
"next_offset": next_offset,
}
})
```
**关键点**:
- 支持 `from` 参数,只返回 `idx >= from` 的事件
- 返回 `next_offset`,用于下次轮询
- 事件存储在 `deque(maxlen=1000)`最多保留1000个事件
### 3.2 事件重放顺序
**文件**: `/static/src/stores/task.ts` - `pollTaskEvents()`
```typescript
async pollTaskEvents(eventHandler: (event: any) => void) {
const response = await fetch(
`/api/tasks/${this.currentTaskId}?from=${this.lastEventIndex}`
);
const result = await response.json();
const data = result.data;
// 处理新事件(按顺序)
if (data.events && data.events.length > 0) {
for (const event of data.events) {
try {
eventHandler(event);
} catch (err) {
console.error('[Task] 处理事件失败:', err, event);
}
}
// 更新偏移量
this.lastEventIndex = data.next_offset;
}
}
```
**重放顺序**:
1. 按事件的 `idx` 顺序(从小到大)
2. 同步处理,确保顺序正确
3. 每个事件处理完后才处理下一个
### 3.3 重放时的特殊处理
**文件**: `/static/src/app/methods/taskPolling.ts` - `handleTaskEvent()`
```typescript
handleTaskEvent(event: any) {
// 1. 检查事件的 conversation_id 是否匹配
if (eventData.conversation_id && this.currentConversationId) {
if (eventData.conversation_id !== this.currentConversationId) {
debugLog(`忽略不匹配的事件`);
return;
}
}
// 2. 根据事件类型调用对应的处理方法
switch (eventType) {
case 'ai_message_start':
this.handleAiMessageStart(eventData, eventIdx);
break;
case 'thinking_start':
this.handleThinkingStart(eventData, eventIdx);
break;
// ... 其他事件类型
}
// 注意:不要在这里清除 _rebuildingFromScratch 标记
// 该标记应该在恢复完所有历史事件后才清除
}
```
**特殊处理**:
- **对话ID过滤**: 忽略不匹配当前对话的事件
- **跳过动画**: 通过 `_rebuildingFromScratch` 标志跳过动画效果
- **静默恢复**: 不显示"生成中"标签,不展开思考块
---
## 4. 恢复时的 UI 状态管理
### 4.1 思考块状态恢复
**文件**: `/static/src/app/methods/taskPolling.ts` - `restoreTaskState()`
```typescript
// 恢复思考块状态
if (lastMessage.actions) {
const thinkingActions = lastMessage.actions.filter(a => a.type === 'thinking');
if (inThinking && thinkingActions.length > 0) {
// 正在思考中,检查最后一个思考块是否正在流式输出
const lastThinking = thinkingActions[thinkingActions.length - 1];
// 只有当思考块正在流式输出时才设置锁定状态(但不展开)
if (lastThinking.streaming && lastThinking.blockId) {
this.$nextTick(() => {
this.chatSetThinkingLock(lastThinking.blockId, true);
});
}
}
// 确保所有思考块都是折叠状态
for (const thinking of thinkingActions) {
if (thinking.blockId) {
thinking.collapsed = true;
}
}
}
```
**状态恢复**:
- **折叠状态**: 所有思考块默认折叠
- **锁定状态**: 如果正在流式输出,设置锁定(防止用户手动展开)
- **不展开**: 即使正在流式输出,也不自动展开(避免干扰用户)
### 4.2 工具块状态恢复
```typescript
// 注册历史中的工具块到 toolActionIndex
const toolActions = lastMessage.actions.filter(a => a.type === 'tool');
for (const toolAction of toolActions) {
if (toolAction.tool && toolAction.tool.id) {
// 注册到 toolActionIndex
this.toolRegisterAction(toolAction, toolAction.tool.id);
// 如果有 executionId也注册
if (toolAction.tool.executionId) {
this.toolRegisterAction(toolAction, toolAction.tool.executionId);
}
// 追踪工具调用
if (toolAction.tool.name) {
this.toolTrackAction(toolAction.tool.name, toolAction);
}
}
}
```
**关键点**:
- **注册工具块**: 将历史中的工具块注册到 `toolActionIndex`
- **支持更新**: 后续的 `update_action` 事件可以找到对应的块进行状态更新
- **双重注册**: 同时注册 `id``executionId`,确保能找到
### 4.3 停止按钮状态
```typescript
// 标记状态为进行中
this.streamingMessage = true;
this.taskInProgress = true;
```
**状态设置**:
- `streamingMessage = true`: 显示停止按钮
- `taskInProgress = true`: 禁用输入框
- `stopRequested = false`: 重置停止请求标志
### 4.4 输入框状态
输入框状态由 `taskInProgress` 控制:
```typescript
// 在模板中
<input :disabled="taskInProgress" />
```
**状态**:
- `taskInProgress = true`: 输入框禁用
- `taskInProgress = false`: 输入框启用
---
## 5. 轮询的恢复机制
### 5.1 轮询启动
**文件**: `/static/src/stores/task.ts` - `startPolling()`
```typescript
startPolling(eventHandler?: (event: any) => void) {
if (this.isPolling) {
return;
}
this.isPolling = true;
// 获取事件处理器
const handler = eventHandler || ((window as any).__taskEventHandler);
// 立即执行一次
this.pollTaskEvents(handler);
// 设置定时轮询150ms 间隔)
this.pollingInterval = window.setInterval(() => {
this.pollTaskEvents(handler);
}, 150);
}
```
**关键点**:
- **立即执行**: 启动轮询后立即执行一次,不等待第一个间隔
- **150ms 间隔**: 接近流式输出效果
- **全局处理器**: 使用 `window.__taskEventHandler` 作为默认处理器
### 5.2 从哪个 offset 开始轮询
**两种模式**:
#### 5.2.1 从头重建模式
```typescript
// 重置偏移量为 0从头获取所有事件
taskStore.lastEventIndex = 0;
```
- **offset = 0**: 从第一个事件开始
- **目的**: 重建完整的 assistant 响应
#### 5.2.2 增量恢复模式
```typescript
// 获取任务详情,计算已处理的事件数量
const detailResponse = await fetch(`/api/tasks/${runningTask.task_id}`);
const detailResult = await detailResponse.json();
if (detailResult.success && detailResult.data.events) {
// 设置为当前事件数量,只获取新事件
this.lastEventIndex = detailResult.data.next_offset || detailResult.data.events.length;
}
```
- **offset = next_offset**: 从历史事件之后开始
- **目的**: 只获取新事件,避免重复处理
### 5.3 避免重复处理事件
**机制**:
1. **事件索引**: 每个事件有唯一的 `idx`
2. **偏移量过滤**: 后端只返回 `idx >= offset` 的事件
3. **偏移量更新**: 处理完事件后更新 `lastEventIndex = next_offset`
```typescript
// 处理新事件
if (data.events && data.events.length > 0) {
for (const event of data.events) {
eventHandler(event);
}
// 更新偏移量,下次轮询从这里开始
this.lastEventIndex = data.next_offset;
}
```
**保证**:
- 每个事件只处理一次
- 不会遗漏事件
- 不会重复处理
---
## 6. WebSocket vs 轮询的恢复差异
### 6.1 WebSocket 模式(已废弃)
**特点**:
- **实时推送**: 服务器主动推送事件
- **连接断开**: 刷新页面后 WebSocket 连接断开
- **无法恢复**: 断开期间的事件丢失
**问题**:
- 页面刷新后无法恢复任务状态
- 需要复杂的重连和事件补偿机制
### 6.2 轮询模式(当前实现)
**特点**:
- **主动拉取**: 客户端主动轮询事件
- **无连接状态**: 不依赖持久连接
- **完整恢复**: 可以从任意 offset 开始拉取事件
**优势**:
- 页面刷新后可以完整恢复任务状态
- 实现简单,无需处理重连逻辑
- 支持从头重建和增量恢复两种模式
### 6.3 恢复流程对比
| 特性 | WebSocket 模式 | 轮询模式 |
|------|---------------|---------|
| 连接状态 | 有状态(需要重连) | 无状态 |
| 事件恢复 | 困难(需要补偿机制) | 简单(从 offset 拉取) |
| 实时性 | 高(服务器推送) | 中150ms 轮询) |
| 可靠性 | 低(连接断开丢失事件) | 高(事件持久化) |
| 刷新恢复 | 不支持 | 完全支持 |
---
## 7. 完整恢复流程图
```
页面刷新
mounted() 生命周期
注册全局事件处理器 __taskEventHandler
调用 restoreTaskState()
检查是否已有任务在进行 ──→ 是 ──→ 跳过恢复
↓ 否
调用 taskStore.loadRunningTask()
查询 /api/tasks 获取任务列表
筛选 status === 'running' 的任务 ──→ 无 ──→ 恢复子智能体等待状态
↓ 有
获取任务详情 /api/tasks/{task_id}
计算 lastEventIndex已处理的事件数量
等待历史加载完成
检查历史完整性
├─→ 历史不完整 ──→ 从头重建模式
│ ↓
│ 删除不完整的 assistant 消息
│ ↓
│ 设置 _rebuildingFromScratch = true
│ ↓
│ 设置 lastEventIndex = 0
│ ↓
│ 启动轮询(从头重放所有事件)
│ ↓
│ 2秒后清除 _rebuildingFromScratch 标志
└─→ 历史完整 ──→ 增量恢复模式
恢复思考块状态(折叠、锁定)
恢复工具块状态(注册到 toolActionIndex
恢复文本块状态streaming 标志)
设置 streamingMessage = true
设置 taskInProgress = true
启动轮询(只获取新事件)
显示"任务恢复"提示
```
---
## 8. 关键代码文件索引
### 8.1 前端文件
| 文件路径 | 功能 |
|---------|------|
| `/static/src/app/lifecycle.ts` | 页面生命周期,触发恢复 |
| `/static/src/stores/task.ts` | 任务状态管理,轮询控制 |
| `/static/src/app/methods/taskPolling.ts` | 事件处理,状态恢复 |
| `/static/src/app/methods/history.ts` | 历史消息加载和渲染 |
| `/static/src/app/methods/conversation.ts` | 对话管理 |
### 8.2 后端文件
| 文件路径 | 功能 |
|---------|------|
| `/server/tasks.py` | 任务 API事件存储和查询 |
| `/server/chat_flow.py` | 聊天流程,事件生成 |
---
## 9. 最佳实践和注意事项
### 9.1 事件设计
1. **每个事件必须有唯一的 idx**: 用于偏移量过滤
2. **事件必须包含 conversation_id**: 用于过滤不匹配的事件
3. **事件必须按顺序生成**: 确保 idx 递增
4. **事件必须持久化**: 存储在 `deque(maxlen=1000)`
### 9.2 状态恢复
1. **先加载历史,再恢复任务**: 确保有完整的消息列表
2. **检查历史完整性**: 对比事件数量和 actions 数量
3. **注册工具块**: 确保后续的 `update_action` 事件能找到对应的块
4. **设置正确的偏移量**: 避免重复处理或遗漏事件
### 9.3 UI 体验
1. **跳过动画**: 从头重建时跳过所有动画效果
2. **静默恢复**: 不显示"生成中"标签
3. **折叠思考块**: 默认折叠,避免干扰用户
4. **显示提示**: 恢复成功后显示"任务恢复"提示
### 9.4 错误处理
1. **任务不存在**: 恢复子智能体等待状态
2. **历史加载失败**: 延迟重试最多5次
3. **事件处理失败**: 捕获异常,继续处理下一个事件
4. **对话切换**: 忽略不匹配的事件
---
## 10. 总结
AI Agent 系统的任务恢复机制通过以下关键技术实现:
1. **事件持久化**: 所有事件存储在后端,支持从任意 offset 查询
2. **轮询模式**: 客户端主动拉取事件,无需维护连接状态
3. **双模式恢复**: 支持从头重建和增量恢复两种模式
4. **智能判断**: 根据历史完整性自动选择恢复模式
5. **UI 优化**: 跳过动画、静默恢复,提升用户体验
6. **状态同步**: 恢复思考块、工具块、文本块的完整状态
这套机制确保了用户在页面刷新后能够无缝恢复任务,不会丢失任何进度,同时保持了良好的用户体验。