Skip to content

P
Pangea-Agent
Commit cfa74117 authored by ligaowei's avatar ligaowei
Browse files
Options

refactor(agent): 重构事件数据构建器和Playwright管理模块 

将事件数据构建器从workpanel迁移到agent模块，统一管理事件数据构建
重构Playwright管理器接口位置，优化代码组织结构
移除冗余的workpanel相关代码，简化项目结构
调整ReactStepType枚举处理空格逻辑，增强健壮性
优化EventSplitter的事件分割逻辑，提高准确性
parent 55b91d26
Hide whitespace changes
Inline Side-by-side
Showing with 85 additions and 2136 deletions
backend/src/main/java/pangea/hiagent/workpanel/data/CompletionEventDataBuilder.java → backend/src/main/java/pangea/hiagent/agent/data/CompletionEventDataBuilder.java
View file @ cfa74117
package pangea.hiagent.workpanel.data;
package pangea.hiagent.agent.data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
......@@ -29,4 +29,4 @@ public class CompletionEventDataBuilder {
data.put("timestamp", System.currentTimeMillis());
return data;
}
}
\ No newline at end of file
}
backend/src/main/java/pangea/hiagent/workpanel/data/ErrorEventDataBuilder.java → backend/src/main/java/pangea/hiagent/agent/data/ErrorEventDataBuilder.java
View file @ cfa74117
package pangea.hiagent.workpanel.data;
package pangea.hiagent.agent.data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
......@@ -30,4 +30,5 @@ public class ErrorEventDataBuilder {
data.put("type", "error");
return data;
}
}
\ No newline at end of file
}
backend/src/main/java/pangea/hiagent/workpanel/data/MapPoolService.java → backend/src/main/java/pangea/hiagent/agent/data/MapPoolService.java
View file @ cfa74117
package pangea.hiagent.workpanel.data;
package pangea.hiagent.agent.data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import pangea.hiagent.common.utils.ObjectPool;
......@@ -58,4 +58,5 @@ public class MapPoolService {
public String getMapPoolStatistics() {
return mapPool.getStatistics();
}
}
\ No newline at end of file
}
backend/src/main/java/pangea/hiagent/agent/react/EventSplitter.java
View file @ cfa74117
......@@ -7,13 +7,18 @@ import java.util.regex.Pattern;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@Component
public class EventSplitter {
private final List<String> keywords = Arrays.asList(
"Thought", "Action", "Observation", "Iteration_Decision", "Final_Answer"
"Thought", "Action", "Observation", "Final_Answer"
);
private final Pattern keywordPattern = Pattern.compile(
String.format("(%s):", String.join("|", keywords))
String.format("(?i)(Thought|Action|Observation|Final[ _]Answer):", String.join("|", keywords)), Pattern.CASE_INSENSITIVE
);
private String currentType = null;
......@@ -30,8 +35,12 @@ public class EventSplitter {
// 每收到一个token/字符，调用此方法
public void feedToken(String token) {
buffer.append(token);
// log.debug("当前缓冲区: {}", buffer.toString());
Matcher matcher = keywordPattern.matcher(buffer);
if (matcher.find()) {
while (matcher.find()) {
log.debug("发现新事件关键词: {}", matcher.group(1));
// 发现新事件
if (currentType != null && currentContent.length() > 0) {
// 实时输出已分割事件
......@@ -40,12 +49,31 @@ public class EventSplitter {
// 更新事件类型
currentType = matcher.group(1);
currentContent.setLength(0);
// 移除关键词和冒号
buffer.delete(0, matcher.end());
// 累积匹配位置后的内容
currentContent.append(buffer.substring(matcher.end()));
// 重置buffer为剩余内容
buffer.setLength(0);
buffer.append(currentContent);
// 重新查找
matcher = keywordPattern.matcher(buffer);
}
// 检查是否有部分关键词在buffer末尾
if (buffer.length() > 0) {
// 检查是否可能是关键词的一部分
boolean isPartialKeyword = false;
String bufferStr = buffer.toString();
for (String keyword : keywords) {
if (keyword.startsWith(bufferStr) || bufferStr.startsWith(keyword)) {
isPartialKeyword = true;
break;
}
}
if (!isPartialKeyword) {
// 不是部分关键词，添加到内容中
currentContent.append(buffer);
buffer.setLength(0);
}
}
// 累积内容
currentContent.append(buffer);
buffer.setLength(0);
}
// 流式结束时，调用此方法输出最后一个事件
......
backend/src/main/java/pangea/hiagent/agent/react/ReactStepType.java
View file @ cfa74117
......@@ -25,6 +25,6 @@ public enum ReactStepType {
FINAL_ANSWER;
public static ReactStepType fromString(String currentType) {
return ReactStepType.valueOf(currentType.toUpperCase());
}
return ReactStepType.valueOf(currentType.toUpperCase().replace(" ", "_"));
}
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/agent/service/AgentChatService.java
View file @ cfa74117
......@@ -13,7 +13,6 @@ import pangea.hiagent.web.dto.ChatRequest;
import pangea.hiagent.model.Agent;
import pangea.hiagent.tool.AgentToolManager;
import pangea.hiagent.web.dto.AgentRequest;
import pangea.hiagent.workpanel.event.EventService;
import jakarta.servlet.http.HttpServletResponse;
/**
......@@ -31,7 +30,6 @@ public class AgentChatService {
private final pangea.hiagent.web.service.AgentService agentService;
public AgentChatService(
EventService eventService,
ErrorHandlerService errorHandlerService,
AgentProcessorFactory agentProcessorFactory,
AgentToolManager agentToolManager,
......
backend/src/main/java/pangea/hiagent/agent/service/SseTokenEmitter.java
View file @ cfa74117
......@@ -145,7 +145,7 @@ public class SseTokenEmitter implements TokenConsumerWithCompletion {
public void closeEmitter() {
try {
if (emitter != null && !userSseService.isEmitterCompleted(emitter)) {
emitter.complete();
// emitter.complete();
log.debug("SSE连接已关闭");
}
} catch (Exception ex) {
......
backend/src/main/java/pangea/hiagent/agent/service/UserSseService.java
View file @ cfa74117
package pangea.hiagent.agent.service;
import lombok.extern.slf4j.Slf4j;
import pangea.hiagent.agent.data.ErrorEventDataBuilder;
import pangea.hiagent.agent.data.MapPoolService;
import pangea.hiagent.agent.data.TokenEventDataBuilder;
import pangea.hiagent.web.dto.WorkPanelEvent;
import pangea.hiagent.workpanel.event.EventService;
import pangea.hiagent.workpanel.data.TokenEventDataBuilder;
import pangea.hiagent.workpanel.data.ErrorEventDataBuilder;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.Map;
......@@ -42,18 +43,19 @@ public class UserSseService {
// 心跳任务执行器 - 使用共享线程池以提高资源利用率
private final ScheduledExecutorService heartbeatExecutor;
// SSE超时时间（毫秒）
private static final long SSE_TIMEOUT = 0L; // 0表示不使用默认超时，由心跳机制管理连接
private final EventService eventService;
private final TokenEventDataBuilder tokenEventDataBuilder;
private final ErrorEventDataBuilder errorEventDataBuilder;
public UserSseService(EventService eventService, TokenEventDataBuilder tokenEventDataBuilder,
ErrorEventDataBuilder errorEventDataBuilder) {
this.eventService = eventService;
private final MapPoolService mapPoolService;
// SSE超时时间（毫秒）
private static final long SSE_TIMEOUT = 0L; // 0表示不使用默认超时，由心跳机制管理连接
public UserSseService(TokenEventDataBuilder tokenEventDataBuilder, ErrorEventDataBuilder errorEventDataBuilder, MapPoolService mapPoolService) {
this.tokenEventDataBuilder = tokenEventDataBuilder;
this.errorEventDataBuilder = errorEventDataBuilder;
this.mapPoolService = mapPoolService;
this.heartbeatExecutor = Executors.newScheduledThreadPool(2);
}
......@@ -291,7 +293,7 @@ public class UserSseService {
// 然后关闭SSE连接
try {
if (!isEmitterCompleted(emitter)) {
emitter.complete();
// emitter.complete();
log.debug("SSE连接已关闭");
}
} catch (Exception ex) {
......@@ -596,12 +598,19 @@ public class UserSseService {
}
// 从对象池获取Map，避免频繁创建对象
Map<String, Object> data =event.getMetadata();
Map<String, Object> data = mapPoolService.acquireMap();
// 设置基础属性
data.put("type", event.getType());
data.put("eventType", event.getType());
data.put("timestamp", event.getTimestamp());
data.put("title", event.getTitle());
data.put("content", event.getContent());
data.put("userId", event.getUserId());
if(event.getMetadata() != null) {
data.putAll(event.getMetadata());
}
return data;
}
......
backend/src/main/java/pangea/hiagent/common/config/DomSyncWebSocketConfig.java
View file @ cfa74117
......@@ -15,8 +15,8 @@ import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry
import org.springframework.web.socket.server.HandshakeInterceptor;
import org.springframework.web.util.UriComponentsBuilder;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
import pangea.hiagent.common.utils.JwtUtil;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import pangea.hiagent.websocket.DomSyncHandler;
import java.util.Map;
......
backend/src/main/java/pangea/hiagent/tool/aspect/ToolExecutionLoggerAspect.java deleted 100644 → 0
View file @ 55b91d26
// package pangea.hiagent.tool.aspect;
// import lombok.extern.slf4j.Slf4j;
// import org.aspectj.lang.ProceedingJoinPoint;
// import org.aspectj.lang.annotation.Around;
// import org.aspectj.lang.annotation.Aspect;
// import org.aspectj.lang.reflect.MethodSignature;
// import org.springframework.ai.tool.annotation.Tool;
// import org.springframework.beans.factory.annotation.Autowired;
// import org.springframework.stereotype.Component;
// import pangea.hiagent.workpanel.IWorkPanelDataCollector;
// import java.util.HashMap;
// import java.util.Map;
// /**
// * 工具执行日志记录切面类
// * 自动记录带有@Tool注解的方法执行信息，包括工具名称、方法名、输入参数、输出结果、运行时长等
// */
// @Slf4j
// @Aspect
// @Component
// public class ToolExecutionLoggerAspect {
// @Autowired
// private IWorkPanelDataCollector workPanelDataCollector;
// /**
// * 环绕通知，拦截所有带有@Tool注解的方法
// * @param joinPoint 连接点
// * @return 方法执行结果
// * @throws Throwable 异常
// */
// @Around("@annotation(tool)")
// public Object logToolExecution(ProceedingJoinPoint joinPoint, Tool tool) throws Throwable {
// // 获取方法签名
// MethodSignature signature = (MethodSignature) joinPoint.getSignature();
// String methodName = signature.getName();
// String className = signature.getDeclaringType().getSimpleName();
// String fullMethodName = className + "." + methodName;
// // 获取工具描述
// String toolDescription = tool.description();
// // 获取方法参数
// String[] paramNames = signature.getParameterNames();
// Object[] args = joinPoint.getArgs();
// // 构建输入参数映射
// Map<String, Object> inputParams = new HashMap<>();
// if (paramNames != null && args != null) {
// for (int i = 0; i < paramNames.length; i++) {
// if (i < args.length) {
// inputParams.put(paramNames[i], args[i]);
// }
// }
// }
// // 记录开始时间
// long startTime = System.currentTimeMillis();
// // 记录工具调用开始
// if (workPanelDataCollector != null) {
// try {
// workPanelDataCollector.recordToolCallAction(className, inputParams, null, "pending", null);
// } catch (Exception e) {
// log.warn("记录工具调用开始时发生错误: {}", e.getMessage(), e);
// }
// }
// log.info("开始执行工具方法: {}，描述: {}", fullMethodName, toolDescription);
// log.debug("工具方法参数: {}", inputParams);
// try {
// // 执行原方法
// Object result = joinPoint.proceed();
// // 记录结束时间
// long endTime = System.currentTimeMillis();
// long executionTime = endTime - startTime;
// // 记录工具调用完成
// if (workPanelDataCollector != null) {
// try {
// workPanelDataCollector.recordToolCallAction(className, inputParams, result, "success", executionTime);
// } catch (Exception e) {
// log.warn("记录工具调用完成时发生错误: {}", e.getMessage(), e);
// }
// }
// log.info("工具方法执行成功: {}，描述: {}，耗时: {}ms", fullMethodName, toolDescription, executionTime);
// // 精简日志记录，避免过多的debug级别日志
// if (log.isTraceEnabled()) {
// log.trace("工具方法执行结果类型: {}，结果: {}",
// result != null ? result.getClass().getSimpleName() : "null", result);
// }
// return result;
// } catch (Exception e) {
// // 记录结束时间
// long endTime = System.currentTimeMillis();
// long executionTime = endTime - startTime;
// // 记录工具调用错误
// if (workPanelDataCollector != null) {
// try {
// workPanelDataCollector.recordToolCallAction(className, inputParams, e, "error", executionTime);
// } catch (Exception ex) {
// log.warn("记录工具调用错误时发生错误: {}", ex.getMessage(), ex);
// }
// }
// log.error("工具方法执行失败: {}，描述: {}，耗时: {}ms，错误类型: {}",
// fullMethodName, toolDescription, executionTime, e.getClass().getSimpleName(), e);
// throw e;
// }
// }
// }
\ No newline at end of file
backend/src/main/java/pangea/hiagent/tool/impl/HisenseLbpmApprovalTool.java
View file @ cfa74117
......@@ -18,8 +18,8 @@ import java.util.HashMap;
import java.util.Map;
import pangea.hiagent.tool.BaseTool;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import pangea.hiagent.web.service.ToolConfigService;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
/**
* 海信LBPM流程审批工具类
......
backend/src/main/java/pangea/hiagent/tool/impl/HisensePerformanceApprovalTool.java
View file @ cfa74117
......@@ -18,8 +18,8 @@ import java.util.HashMap;
import java.util.Map;
import pangea.hiagent.tool.BaseTool;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import pangea.hiagent.web.service.ToolConfigService;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
/**
* 海信绩效系统流程审批工具类
......
backend/src/main/java/pangea/hiagent/tool/impl/HisenseSsoLoginTool.java
View file @ cfa74117
......@@ -16,8 +16,8 @@ import java.util.concurrent.ConcurrentMap;
import java.util.HashMap;
import java.util.Map;
import pangea.hiagent.tool.BaseTool;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import pangea.hiagent.web.service.ToolConfigService;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
/**
* 海信SSO认证工具类
......
backend/src/main/java/pangea/hiagent/tool/impl/PlaywrightWebTools.java
View file @ cfa74117
......@@ -4,10 +4,11 @@ import com.microsoft.playwright.*;
import com.microsoft.playwright.options.LoadState;
import com.microsoft.playwright.options.WaitUntilState;
import lombok.extern.slf4j.Slf4j;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import org.springframework.ai.tool.annotation.Tool;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
import java.util.Base64;
import java.util.List;
......
backend/src/main/java/pangea/hiagent/tool/impl/StorageFileAccessTool.java
View file @ cfa74117
......@@ -3,7 +3,6 @@ package pangea.hiagent.tool.impl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.tool.annotation.Tool;
import org.springframework.stereotype.Component;
import pangea.hiagent.workpanel.IWorkPanelDataCollector;
import java.io.BufferedReader;
import java.io.File;
......@@ -21,7 +20,6 @@ import java.util.List;
@Component
public class StorageFileAccessTool {
private final IWorkPanelDataCollector workPanelDataCollector;
// 支持的文件扩展名
private static final List<String> SUPPORTED_EXTENSIONS = Arrays.asList(
......@@ -31,10 +29,7 @@ public class StorageFileAccessTool {
// storage目录路径
private static final String STORAGE_DIR = "backend" + File.separator + "storage";
public StorageFileAccessTool(IWorkPanelDataCollector workPanelDataCollector) {
this.workPanelDataCollector = workPanelDataCollector;
}
/**
* 访问并预览storage目录下的文件
......@@ -87,9 +82,7 @@ public class StorageFileAccessTool {
log.info("成功读取文件: {}", fileName);
String result = "已成功在工作面板中预览文件: " + fileName;
// 发送embed事件到工作面板
workPanelDataCollector.recordEmbed(filePath, mimeType, title, content);
return result;
......
backend/src/main/java/pangea/hiagent/tool/impl/WebPageAccessTools.java
View file @ cfa74117
package pangea.hiagent.tool.impl;import lombok.extern.slf4j.Slf4j;
import pangea.hiagent.workpanel.IWorkPanelDataCollector;
import org.springframework.ai.tool.annotation.Tool;
import org.springframework.stereotype.Component;
......@@ -18,13 +17,6 @@ import java.net.URLConnection;
@Component
public class WebPageAccessTools {
// 通过构造器注入的方式引入IWorkPanelDataCollector依赖
private final IWorkPanelDataCollector workPanelDataCollector;
public WebPageAccessTools(IWorkPanelDataCollector workPanelDataCollector) {
this.workPanelDataCollector = workPanelDataCollector;
}
/**
* 根据网站名称访问网页并在工作面板中预览
* @param siteName 网站名称（如"百度"、"谷歌"等）
......@@ -84,9 +76,6 @@ public class WebPageAccessTools {
log.info("成功访问网页: {}", url);
String result = "已成功在工作面板中预览网页: " + url;
// 发送embed事件到工作面板
workPanelDataCollector.recordEmbed(url, "text/html", title, webContent);
return result;
} catch (Exception e) {
return handleError(e, "获取网页内容时发生错误");
......
backend/src/main/java/pangea/hiagent/workpanel/playwright/PlaywrightManager.java → backend/src/main/java/pangea/hiagent/tool/playwright/PlaywrightManager.java
View file @ cfa74117
package pangea.hiagent.workpanel.playwright;
package pangea.hiagent.tool.playwright;
import com.microsoft.playwright.Browser;
import com.microsoft.playwright.BrowserContext;
......
backend/src/main/java/pangea/hiagent/workpanel/playwright/PlaywrightManagerImpl.java → backend/src/main/java/pangea/hiagent/tool/playwright/PlaywrightManagerImpl.java
View file @ cfa74117
package pangea.hiagent.workpanel.playwright;
package pangea.hiagent.tool.playwright;
import com.microsoft.playwright.*;
import lombok.extern.slf4j.Slf4j;
......
backend/src/main/java/pangea/hiagent/websocket/DomSyncHandler.java
View file @ cfa74117
......@@ -2,9 +2,10 @@ package pangea.hiagent.websocket;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.socket.*;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
import java.util.HashMap;
import java.util.Map;
......
backend/src/main/java/pangea/hiagent/websocket/DomSyncService.java
View file @ cfa74117
......@@ -3,8 +3,8 @@ package pangea.hiagent.websocket;
import com.microsoft.playwright.*;
import com.microsoft.playwright.options.LoadState;
import lombok.extern.slf4j.Slf4j;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
import pangea.hiagent.common.utils.AsyncUserContextDecorator;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import java.util.concurrent.ConcurrentMap;
......
backend/src/main/java/pangea/hiagent/websocket/WebSocketConnectionManager.java
View file @ cfa74117
......@@ -2,8 +2,9 @@ package pangea.hiagent.websocket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.*;
import pangea.hiagent.workpanel.playwright.PlaywrightManager;
import pangea.hiagent.common.utils.UserUtils;
import pangea.hiagent.tool.playwright.PlaywrightManager;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
......
backend/src/main/java/pangea/hiagent/workpanel/IWorkPanelDataCollector.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel;
import pangea.hiagent.web.dto.WorkPanelEvent;
import java.util.function.Consumer;
import java.util.List;
/**
* 工作面板数据收集器接口
* 用于采集Agent执行过程中的各类数据（思考过程、工具调用等）
*/
public interface IWorkPanelDataCollector {
/**
* 记录思考过程
* @param content 思考内容
* @param thinkingType 思考类型（分析、规划、执行等）
*/
void recordThinking(String content, String thinkingType);
/**
* 记录工具调用Action
* @param toolName 工具名称
* @param input 工具输入参数
* @param output 工具输出结果
* @param status 执行状态（pending/success/failure/error）
* @param executionTime 执行时间（毫秒）
*/
void recordToolCallAction(String toolName, Object input, Object output, String status, Long executionTime);
/**
* 记录日志
* @param message 日志消息
* @param level 日志级别（info/warn/error/debug）
*/
void recordLog(String message, String level);
/**
* 记录embed嵌入事件
* @param url 嵌入资源URL（可选）
* @param type MIME类型
* @param title 嵌入标题
* @param htmlContent HTML内容（可选）
*/
void recordEmbed(String url, String type, String title, String htmlContent);
/**
* 获取所有收集的事件
*/
List<WorkPanelEvent> getEvents();
/**
* 订阅事件（用于实时推送）
* @param consumer 事件处理函数
*/
void subscribe(Consumer<WorkPanelEvent> consumer);
/**
* 清空所有事件
*/
void clear();
/**
* 获取最后一个工具调用事件
*/
WorkPanelEvent getLastToolCall();
/**
* 记录最终答案
* @param content 最终答案内容
*/
void recordFinalAnswer(String content);
/**
* 添加事件到收集器
* 统一的事件添加方法，用于避免重复实现
* @param event 工作面板事件
*/
void addEvent(WorkPanelEvent event);
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/workpanel/WorkPanelDataCollector.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import pangea.hiagent.workpanel.event.EventDeduplicationService;
import pangea.hiagent.workpanel.event.EventService;
import pangea.hiagent.agent.service.UserSseService;
import pangea.hiagent.web.dto.LogEvent;
import pangea.hiagent.web.dto.ResultEvent;
import pangea.hiagent.web.dto.ThoughtEvent;
import pangea.hiagent.web.dto.ToolEvent;
import pangea.hiagent.web.dto.WorkPanelEvent;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
/**
* 工作面板数据收集器实现
* 专门负责采集Agent执行过程中的各类数据，遵循单一职责原则
*/
@Slf4j
@Component
public class WorkPanelDataCollector implements IWorkPanelDataCollector {
/**
* 事件列表（线程安全）
*/
private final List<WorkPanelEvent> events = new CopyOnWriteArrayList<>();
/**
* 事件订阅者列表（线程安全）
*/
private final List<Consumer<WorkPanelEvent>> subscribers = new CopyOnWriteArrayList<>();
/**
* 用户ID到订阅者的映射，用于更好地管理订阅者
*/
private final Map<String, Consumer<WorkPanelEvent>> userSubscribers = new ConcurrentHashMap<>();
/**
* 事件去重服务
*/
@Autowired
private EventDeduplicationService eventDeduplicationService;
/**
* 统一事件服务
*/
@Autowired
private EventService eventService;
/**
* SSE服务
*/
@Autowired
private UserSseService unifiedSseService;
/**
* 最大事件数量，防止内存溢出
*/
private static final int MAX_EVENTS = 1000;
public WorkPanelDataCollector() {
// 默认构造函数
}
@Override
public void recordThinking(String content, String thinkingType) {
try {
WorkPanelEvent event = createThinkingEvent(content, thinkingType);
if (event != null && !eventDeduplicationService.isDuplicateEvent(event)) {
addEvent(event);
}
} catch (Exception e) {
logError("记录思考过程时发生错误", "content={}, type={}", content, thinkingType, e);
}
}
@Override
public void recordToolCallAction(String toolName, Object input, Object output, String status, Long executionTime) {
try {
WorkPanelEvent event = createToolCallActionEvent(toolName, input, output, status, executionTime);
if (event != null) {
handleToolCallAction(toolName, event, status, null);
}
} catch (Exception e) {
logError("记录工具调用Action时发生错误", "toolName={}", toolName, null, e);
}
}
@Override
public void recordLog(String message, String level) {
try {
WorkPanelEvent event = createLogEvent(message, level);
if (event != null) {
addEvent(event);
}
} catch (Exception e) {
logError("记录日志时发生错误", "message={}, level={}", message, level, e);
}
}
@Override
public void recordEmbed(String url, String type, String title, String htmlContent) {
try {
WorkPanelEvent event = createEmbedEvent(url, type, title, htmlContent);
if (event != null) {
addEvent(event);
}
} catch (Exception e) {
logError("recordEmbed时发生错误", "title={}, type={}", title, type, e);
}
}
@Override
public void recordFinalAnswer(String content) {
try {
WorkPanelEvent event = createFinalAnswerEvent(content);
if (event != null && !eventDeduplicationService.isDuplicateEvent(event)) {
addEvent(event);
}
} catch (Exception e) {
logError("记录最终答案时发生错误", "content={}", content, null, e);
}
}
@Override
public List<WorkPanelEvent> getEvents() {
return new ArrayList<>(events);
}
@Override
public void subscribe(Consumer<WorkPanelEvent> consumer) {
if (consumer != null) {
subscribers.add(consumer);
if (log.isTraceEnabled()) {
log.trace("已添加事件订阅者，当前订阅者数量: {}", subscribers.size());
}
}
}
/**
* 为指定用户订阅事件
*
* @param userId 用户ID
* @param consumer 事件消费者
*/
public void subscribe(String userId, Consumer<WorkPanelEvent> consumer) {
if (!isValidSubscription(userId, consumer)) {
return;
}
// 移除该用户之前的订阅者（如果存在）
removeExistingSubscriber(userId);
// 添加新的订阅者
addNewSubscriber(userId, consumer);
logSubscriptionDetails(userId);
}
/**
* 取消指定用户的事件订阅
*
* @param userId 用户ID
*/
public void unsubscribe(String userId) {
if (userId != null) {
Consumer<WorkPanelEvent> subscriber = userSubscribers.remove(userId);
if (subscriber != null) {
subscribers.remove(subscriber);
log.debug("已移除用户 {} 的事件订阅者", userId);
}
}
}
@Override
public void clear() {
try {
events.clear();
// 精简日志记录，避免过多的debug级别日志
if (log.isTraceEnabled()) {
log.trace("已清空工作面板事件");
}
} catch (Exception e) {
log.error("清空事件时发生错误", e);
}
}
@Override
public WorkPanelEvent getLastToolCall() {
// 从后往前查找最后一个工具调用事件
for (int i = events.size() - 1; i >= 0; i--) {
WorkPanelEvent event = events.get(i);
if (event instanceof ToolEvent) {
String type = event.getType();
if ("tool_call".equals(type) || "tool_result".equals(type)) {
return event;
}
}
}
return null;
}
@Override
public void addEvent(WorkPanelEvent event) {
// 直接复用已有的私有方法
if (event != null) {
// 调用内部的addEvent方法
addEventInternal(event);
}
}
/**
* 内部的添加事件方法，避免接口方法与私有方法的命名冲突
*/
private void addEventInternal(WorkPanelEvent event) {
try {
// 控制事件数量，防止内存溢出
if (events.size() >= MAX_EVENTS) {
events.remove(0); // 移除最老的事件
}
events.add(event);
String content = "";
if (event instanceof ThoughtEvent) {
content = ((ThoughtEvent) event).getContent();
} else if (event instanceof LogEvent) {
content = ((LogEvent) event).getContent();
} else if (event instanceof ResultEvent) {
content = ((ResultEvent) event).getContent();
}
log.debug("添加事件到列表: 类型={}, 内容={}", event.getType(), content);
// 更新最近事件缓存
eventDeduplicationService.updateRecentEventsCache(event);
// 通知所有订阅者
notifySubscribers(event);
} catch (Exception e) {
// 即使在addEvent方法中也增加异常保护，防止影响主流程
logDebug("添加事件失败: {}", e.getMessage());
}
}
// ==================== 私有方法 ====================
/**
* 创建思考事件
*/
private WorkPanelEvent createThinkingEvent(String content, String thinkingType) {
return eventService.recordThinking(content, thinkingType);
}
/**
* 创建工具调用Action事件
*/
private WorkPanelEvent createToolCallActionEvent(String toolName, Object input, Object output, String status, Long executionTime) {
return eventService.recordToolCallComplete(toolName, output, status, executionTime);
}
/**
* 处理工具调用Action
*/
private void handleToolCallAction(String toolName, WorkPanelEvent event, String status, String errorMessage) {
// 对于pending状态，直接添加事件
if ("pending".equals(status)) {
addEvent(event);
return;
}
// 查找最近的该工具的pending事件并更新它
WorkPanelEvent lastToolCall = getLastPendingToolCall(toolName);
if (lastToolCall != null) {
// 更新现有事件
if ("error".equals(status) && lastToolCall instanceof ToolEvent) {
updateEventAsError((ToolEvent) lastToolCall, errorMessage);
} else if (lastToolCall instanceof ToolEvent) {
ToolEvent toolEvent = (ToolEvent) lastToolCall;
toolEvent.setType(WorkPanelUtils.getEventTypeFromStatus(status));
toolEvent.setToolStatus(status);
// 注意：这里不设置toolOutput，因为event中可能包含完整的输出信息
toolEvent.setTimestamp(System.currentTimeMillis());
}
// 重新发布更新后的事件
notifySubscribers(lastToolCall);
} else {
addEvent(event);
}
}
/**
* 更新事件为错误状态
*/
private void updateEventAsError(WorkPanelEvent event, String errorMessage) {
event.setType("tool_error");
if (event instanceof ToolEvent) {
ToolEvent toolEvent = (ToolEvent) event;
// 使用反射设置content字段，因为ToolEvent没有setContent方法
try {
java.lang.reflect.Field contentField = WorkPanelEvent.class.getDeclaredField("content");
contentField.setAccessible(true);
contentField.set(toolEvent, errorMessage);
} catch (Exception e) {
// 如果反射失败，忽略错误
}
}
if (event instanceof ToolEvent) {
((ToolEvent) event).setToolStatus("failure");
}
event.setTimestamp(System.currentTimeMillis());
}
/**
* 创建日志事件
*/
private WorkPanelEvent createLogEvent(String message, String level) {
return eventService.recordLog(message, level);
}
/**
* 创建嵌入事件
*/
private WorkPanelEvent createEmbedEvent(String url, String type, String title, String htmlContent) {
return eventService.recordEmbed(url, type, title, htmlContent);
}
/**
* 创建最终答案事件
*/
private WorkPanelEvent createFinalAnswerEvent(String content) {
return eventService.recordFinalAnswer(content);
}
/**
* 通知所有订阅者
*/
private void notifySubscribers(WorkPanelEvent event) {
if (event == null) {
return;
}
try {
// 遍历所有订阅者并发送事件
for (Consumer<WorkPanelEvent> subscriber : subscribers) {
try {
if (subscriber != null) {
subscriber.accept(event);
}
} catch (Exception e) {
// 异常降级为debug日志，避免过度日志记录
// 异常通常由于SSE连接已断开导致，这是正常情况
if (e instanceof org.springframework.web.context.request.async.AsyncRequestNotUsableException) {
log.debug("订阅者处理事件失败：异步请求不可用（客户端已断开连接）");
} else if (e instanceof java.io.IOException) {
log.debug("订阅者处理事件失败：客户端连接已断开");
} else if (e.getMessage() != null && e.getMessage().contains("response has already been committed")) {
log.debug("订阅者处理事件失败：响应已提交");
} else {
// 其他异常也降级为debug，避免日志污染
log.debug("订阅者处理事件失败: {}", e.getMessage());
}
}
}
// 通过EventService发送事件到所有SSE连接
for (SseEmitter emitter : unifiedSseService.getEmitters()) {
try {
unifiedSseService.sendWorkPanelEvent(event);
} catch (Exception e) {
log.debug("通过EventService发送事件失败: {}", e.getMessage());
}
}
} catch (Exception e) {
log.debug("通知订阅者时发生错误: {}", e.getMessage());
}
}
/**
* 统一日志错误处理
*/
private void logError(String message, String format, Object arg1, Object arg2, Exception e) {
if (arg2 != null) {
log.error(message + ": " + format, arg1, arg2, e);
} else {
log.error(message + ": " + format, arg1, e);
}
}
/**
* 统一日志调试处理
*/
private void logDebug(String format, Object... arguments) {
// 即使在addEvent方法中也增加异常保护，防止影响主流程
log.debug(format, arguments);
}
/**
* 验证订阅参数是否有效
*/
private boolean isValidSubscription(String userId, Consumer<WorkPanelEvent> consumer) {
return userId != null && consumer != null;
}
/**
* 移除已存在的订阅者
*/
private void removeExistingSubscriber(String userId) {
Consumer<WorkPanelEvent> existingSubscriber = userSubscribers.get(userId);
if (existingSubscriber != null) {
subscribers.remove(existingSubscriber);
log.debug("已移除用户 {} 的旧订阅者", userId);
}
}
/**
* 添加新的订阅者
*/
private void addNewSubscriber(String userId, Consumer<WorkPanelEvent> consumer) {
subscribers.add(consumer);
userSubscribers.put(userId, consumer);
}
/**
* 记录订阅详情
*/
private void logSubscriptionDetails(String userId) {
// 精简日志记录，避免过多的debug级别日志
if (log.isTraceEnabled()) {
log.trace("已为用户 {} 添加事件订阅者，当前订阅者数量: {}", userId, subscribers.size());
}
}
/**
* 检查是否为指定工具的待处理调用事件
*/
private boolean isPendingToolCallEvent(WorkPanelEvent event, String toolName) {
// 检查事件是否为ToolEvent类型
if (!(event instanceof ToolEvent)) {
return false;
}
ToolEvent toolEvent = (ToolEvent) event;
return "tool_call".equals(event.getType()) &&
toolName != null && toolName.equals(toolEvent.getToolName()) &&
"pending".equals(toolEvent.getToolStatus());
}
/**
* 查找最近的指定工具的pending事件
*
* @param toolName 工具名称
* @return 最近的pending事件，如果不存在则返回null
*/
public WorkPanelEvent getLastPendingToolCall(String toolName) {
// 从后往前查找最近的该工具的pending事件
for (int i = events.size() - 1; i >= 0; i--) {
WorkPanelEvent event = events.get(i);
if (isPendingToolCallEvent(event, toolName)) {
return event;
}
}
return null;
}
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/workpanel/WorkPanelService.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import pangea.hiagent.web.dto.*;
import pangea.hiagent.web.service.AgentService;
import pangea.hiagent.model.Agent;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
/**
* 工作面板服务
* 负责处理工作面板相关的状态和事件
*/
@Slf4j
@Service
public class WorkPanelService {
@Autowired
private AgentService agentService;
// 用于跟踪已发送的事件ID，防止重复发送
private final Map<String, Set<String>> sentEventIds = new ConcurrentHashMap<>();
/**
* 获取工作面板当前状态
*/
public WorkPanelStatusDto getWorkPanelStatus(String agentId, String userId) {
try {
Agent agent = agentService.getAgent(agentId);
if (agent == null) {
throw new RuntimeException("Agent不存在");
}
if (!agent.getOwner().equals(userId)) {
throw new RuntimeException("无权访问该Agent");
}
log.info("获取Agent {} 的工作面板状态", agentId);
// 从工作面板收集器中读取数据
IWorkPanelDataCollector collector = null; // 移除了对ReActService的依赖
List<WorkPanelEvent> allEvents = collector != null ? collector.getEvents() : new ArrayList<>();
// 统计不同类型的事件
int totalEvents = allEvents.size();
int successfulCalls = (int) allEvents.stream()
.filter(e -> "tool_result".equals(e.getType()) &&
(e instanceof ToolEvent && "success".equals(((ToolEvent) e).getToolStatus())))
.count();
int failedCalls = (int) allEvents.stream()
.filter(e -> "tool_error".equals(e.getType()) ||
("tool_result".equals(e.getType()) &&
(e instanceof ToolEvent && "failure".equals(((ToolEvent) e).getToolStatus()))))
.count();
List<WorkPanelEvent> thinkingEvents = new ArrayList<>();
List<WorkPanelEvent> toolCallEvents = new ArrayList<>();
List<WorkPanelEvent> logEvents = new ArrayList<>();
for (WorkPanelEvent event : allEvents) {
switch (event.getType()) {
case "thought" -> thinkingEvents.add(event);
case "tool_call", "tool_result", "tool_error" -> toolCallEvents.add(event);
case "log" -> logEvents.add(event);
default -> {
}
}
}
WorkPanelStatusDto status = WorkPanelStatusDto.builder()
.id(agentId + "_workpanel")
.agentId(agentId)
.agentName(agent.getName())
.events(allEvents)
.thinkingEvents(thinkingEvents)
.toolCallEvents(toolCallEvents)
.logEvents(logEvents)
.totalEvents(totalEvents)
.successfulToolCalls(successfulCalls)
.failedToolCalls(failedCalls)
.updateTimestamp(System.currentTimeMillis())
.isProcessing(false)
.build();
return status;
} catch (Exception e) {
log.error("获取工作面板状态失败", e);
throw new RuntimeException("获取工作面板状态失败: " + e.getMessage(), e);
}
}
/**
* 清空工作面板数据
*/
public void clearWorkPanel(String agentId, String userId) {
try {
Agent agent = agentService.getAgent(agentId);
if (agent == null) {
throw new RuntimeException("Agent不存在");
}
if (!agent.getOwner().equals(userId)) {
throw new RuntimeException("无权访问该Agent");
}
log.info("清空Agent {} 的工作面板", agentId);
// 在实际应用中，这里应该从缓存中清除工作面板数据
// 清空已发送事件ID跟踪
sentEventIds.remove(agentId);
} catch (Exception e) {
log.error("清空工作面板失败", e);
throw new RuntimeException("清空工作面板失败: " + e.getMessage(), e);
}
}
/**
* 生成事件唯一标识
*/
public String generateEventId(WorkPanelEvent event) {
if (event == null) {
return "null_event_" + System.currentTimeMillis();
}
StringBuilder sb = new StringBuilder();
sb.append(event.getType()).append("_");
switch (event.getType()) {
case "thought":
if (event instanceof ThoughtEvent) {
ThoughtEvent thoughtEvent = (ThoughtEvent) event;
sb.append(thoughtEvent.getThinkingType() != null ? thoughtEvent.getThinkingType() : "default")
.append("_")
.append(thoughtEvent.getContent() != null ? thoughtEvent.getContent().hashCode() : 0);
} else {
sb.append("default").append("_").append(0);
}
break;
case "tool_call":
case "tool_result":
case "tool_error":
if (event instanceof ToolEvent) {
ToolEvent toolEvent = (ToolEvent) event;
sb.append(toolEvent.getToolName() != null ? toolEvent.getToolName() : "unknown")
.append("_")
.append(toolEvent.getToolAction() != null ? toolEvent.getToolAction() : "unknown")
.append("_")
.append(event.getTimestamp() != null ? event.getTimestamp() : System.currentTimeMillis());
} else {
sb.append("unknown").append("_").append("unknown").append("_").append(System.currentTimeMillis());
}
break;
case "log":
if (event instanceof LogEvent) {
LogEvent logEvent = (LogEvent) event;
sb.append(logEvent.getLogLevel() != null ? logEvent.getLogLevel() : "info")
.append("_")
.append(logEvent.getContent() != null ? logEvent.getContent().hashCode() : 0);
} else {
sb.append("info").append("_").append(0);
}
break;
case "embed":
if (event instanceof EmbedEvent) {
EmbedEvent embedEvent = (EmbedEvent) event;
sb.append(embedEvent.getEmbedTitle() != null ? embedEvent.getEmbedTitle() : "untitled")
.append("_")
.append(embedEvent.getEmbedUrl() != null ? embedEvent.getEmbedUrl().hashCode() : 0);
} else {
sb.append("untitled").append("_").append(0);
}
break;
default:
sb.append(event.getTimestamp() != null ? event.getTimestamp() : System.currentTimeMillis());
break;
}
return sb.toString();
}
/**
* 检查事件是否已发送
*/
public boolean isEventAlreadySent(String agentId, WorkPanelEvent event) {
String eventId = generateEventId(event);
Set<String> agentEventIds = sentEventIds.computeIfAbsent(agentId, k -> new HashSet<>());
return !agentEventIds.add(eventId); // 如果已存在，add返回false，表示已发送
}
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/workpanel/WorkPanelUtils.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel;
import lombok.extern.slf4j.Slf4j;
import pangea.hiagent.workpanel.event.EventTypeConverter;
import java.util.HashMap;
import java.util.Map;
/**
* 工作面板工具类
* 提供各种辅助方法用于工作面板事件处理
* 注意：此工具类已优化，移除了与EventTypeConverter重复的功能
*/
@Slf4j
public class WorkPanelUtils {
/**
* 将对象转换为Map（用于工具输入参数）
*/
public static Map<String, Object> convertToMap(Object input) {
if (input == null) {
return new HashMap<>();
}
if (input instanceof Map) {
// 安全地转换Map类型，确保键为String类型
Map<?, ?> rawMap = (Map<?, ?>) input;
Map<String, Object> resultMap = new HashMap<>();
for (Map.Entry<?, ?> entry : rawMap.entrySet()) {
// 将键转换为String类型
String key = entry.getKey() != null ? entry.getKey().toString() : "null";
resultMap.put(key, entry.getValue());
}
return resultMap;
}
// 简单对象转换为Map
Map<String, Object> result = new HashMap<>();
result.put("value", input);
return result;
}
/**
* 获取状态文本
* 已委托给EventTypeConverter处理
*/
public static String getStatusText(String status) {
EventTypeConverter converter = new EventTypeConverter();
return converter.getStatusText(status);
}
/**
* 根据状态确定事件类型
* 已委托给EventTypeConverter处理
*/
public static String getEventTypeFromStatus(String status) {
EventTypeConverter converter = new EventTypeConverter();
return converter.getEventTypeFromStatus(status);
}
/**
* 将对象转换为JSON字符串
*
* @param obj 要转换的对象
* @return JSON字符串表示
*/
public static String convertToJsonString(Object obj) {
if (obj == null) {
return "null";
}
try {
// 使用Jackson ObjectMapper进行序列化
com.fasterxml.jackson.databind.ObjectMapper mapper = new com.fasterxml.jackson.databind.ObjectMapper();
return mapper.writeValueAsString(obj);
} catch (Exception e) {
// 如果序列化失败，返回对象的toString()表示
return obj.toString();
}
}
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/workpanel/event/EventDeduplicationService.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel.event;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import pangea.hiagent.web.dto.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 事件去重服务
* 统一处理事件去重逻辑，避免在多个地方重复实现
*/
@Slf4j
@Component
public class EventDeduplicationService {
/**
* 最近事件的缓存，用于去重检查
*/
private final Map<String, WorkPanelEvent> recentEventsCache = new ConcurrentHashMap<>();
/**
* 缓存过期时间（毫秒），默认5秒
*/
private static final long CACHE_EXPIRY_TIME = 5000L;
/**
* 最大缓存大小，防止内存溢出
*/
private static final int MAX_CACHE_SIZE = 1000;
/**
* 上次清理缓存的时间
*/
private volatile long lastCleanupTime = 0;
/**
* 清理间隔（毫秒）
*/
private static final long CLEANUP_INTERVAL = 30000L;
/**
* 检查是否为重复事件
*/
public boolean isDuplicateEvent(WorkPanelEvent event) {
if (event == null) {
return false;
}
// 定期清理过期缓存
cleanupExpiredCache();
// 生成事件唯一标识
String eventKey = generateEventKey(event);
WorkPanelEvent cachedEvent = recentEventsCache.get(eventKey);
if (cachedEvent != null) {
// 检查缓存是否过期
long currentTime = System.currentTimeMillis();
if ((currentTime - cachedEvent.getTimestamp()) <= CACHE_EXPIRY_TIME) {
// 检查事件内容是否相同
return isEventContentEqual(event, cachedEvent);
} else {
// 缓存过期，移除
recentEventsCache.remove(eventKey);
}
}
return false;
}
/**
* 更新最近事件缓存
*/
public void updateRecentEventsCache(WorkPanelEvent event) {
if (event == null) {
return;
}
// 控制缓存大小，防止内存溢出
if (recentEventsCache.size() >= MAX_CACHE_SIZE) {
// 移除最老的条目（简单实现，实际可以根据LRU算法优化）
java.util.Iterator<Map.Entry<String, WorkPanelEvent>> iterator = recentEventsCache.entrySet().iterator();
if (iterator.hasNext()) {
iterator.next();
iterator.remove();
}
}
// 添加新条目
String eventKey = generateEventKey(event);
recentEventsCache.put(eventKey, event);
}
/**
* 生成事件唯一标识
*/
private String generateEventKey(WorkPanelEvent event) {
StringBuilder key = new StringBuilder();
key.append(event.getType() != null ? event.getType() : "");
key.append("_");
// 获取工具名称（如果是ToolEvent）
String toolName = "";
if (event instanceof ToolEvent) {
toolName = ((ToolEvent) event).getToolName();
}
key.append(toolName != null ? toolName : "");
key.append("_");
// 获取思考类型（如果是ThoughtEvent）
String thinkingType = "";
if (event instanceof ThoughtEvent) {
thinkingType = ((ThoughtEvent) event).getThinkingType();
}
key.append(thinkingType != null ? thinkingType : "");
// 对于思考事件，添加内容摘要
if ("thought".equals(event.getType()) && event instanceof ThoughtEvent) {
ThoughtEvent thoughtEvent = (ThoughtEvent) event;
if (thoughtEvent.getContent() != null) {
// 取内容的前50个字符作为摘要
String contentSummary = thoughtEvent.getContent().length() > 50 ?
thoughtEvent.getContent().substring(0, 50) : thoughtEvent.getContent();
key.append("_").append(contentSummary.hashCode());
}
}
return key.toString();
}
/**
* 检查两个事件的内容是否相等
*/
private boolean isEventContentEqual(WorkPanelEvent event1, WorkPanelEvent event2) {
if (event1 == event2) {
return true;
}
if (event1 == null || event2 == null) {
return false;
}
// 比较基本字段
if (!java.util.Objects.equals(event1.getType(), event2.getType())) {
return false;
}
// 比较工具名称（如果是ToolEvent）
String toolName1 = "";
String toolName2 = "";
if (event1 instanceof ToolEvent) {
toolName1 = ((ToolEvent) event1).getToolName();
}
if (event2 instanceof ToolEvent) {
toolName2 = ((ToolEvent) event2).getToolName();
}
if (!java.util.Objects.equals(toolName1, toolName2)) {
return false;
}
// 比较思考类型（如果是ThoughtEvent）
String thinkingType1 = "";
String thinkingType2 = "";
if (event1 instanceof ThoughtEvent) {
thinkingType1 = ((ThoughtEvent) event1).getThinkingType();
}
if (event2 instanceof ThoughtEvent) {
thinkingType2 = ((ThoughtEvent) event2).getThinkingType();
}
if (!java.util.Objects.equals(thinkingType1, thinkingType2)) {
return false;
}
// 比较内容字段（根据不同事件类型）
if (event1 instanceof ThoughtEvent && event2 instanceof ThoughtEvent) {
ThoughtEvent thought1 = (ThoughtEvent) event1;
ThoughtEvent thought2 = (ThoughtEvent) event2;
return java.util.Objects.equals(thought1.getContent(), thought2.getContent());
}
// 比较工具输入（确保都是ToolEvent）
if (event1 instanceof ToolEvent && event2 instanceof ToolEvent) {
if (!java.util.Objects.equals(((ToolEvent) event1).getToolInput(),
((ToolEvent) event2).getToolInput())) {
return false;
}
// 比较工具输出
if (!java.util.Objects.equals(((ToolEvent) event1).getToolOutput(),
((ToolEvent) event2).getToolOutput())) {
return false;
}
} else if (event1 instanceof ResultEvent && event2 instanceof ResultEvent) {
// 比较结果内容
String content1 = ((ResultEvent) event1).getContent();
String content2 = ((ResultEvent) event2).getContent();
if (!java.util.Objects.equals(content1, content2)) {
return false;
}
}
return true;
}
/**
* 清理过期缓存
*/
private void cleanupExpiredCache() {
long currentTime = System.currentTimeMillis();
// 检查是否需要执行清理
if (currentTime - lastCleanupTime < CLEANUP_INTERVAL) {
return;
}
// 更新上次清理时间
lastCleanupTime = currentTime;
// 清理过期条目
recentEventsCache.entrySet().removeIf(entry -> {
WorkPanelEvent event = entry.getValue();
return event == null || (currentTime - event.getTimestamp()) > CACHE_EXPIRY_TIME;
});
}
/**
* 清空缓存
*/
public void clearCache() {
recentEventsCache.clear();
}
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/workpanel/event/EventService.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel.event;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import pangea.hiagent.web.dto.EmbedEvent;
import pangea.hiagent.web.dto.LogEvent;
import pangea.hiagent.web.dto.ResultEvent;
import pangea.hiagent.web.dto.ThoughtEvent;
import pangea.hiagent.web.dto.ToolEvent;
import pangea.hiagent.web.dto.WorkPanelEvent;
import pangea.hiagent.workpanel.WorkPanelUtils;
import pangea.hiagent.workpanel.data.ErrorEventDataBuilder;
import pangea.hiagent.workpanel.data.MapPoolService;
import pangea.hiagent.workpanel.data.TokenEventDataBuilder;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.io.IOException;
/**
* 统一事件服务
* 整合事件创建、构建、发送等功能，事件处理架构
*/
@Slf4j
@Service
public class EventService {
@Autowired
private MapPoolService mapPoolService;
@Autowired
private ErrorEventDataBuilder errorEventDataBuilder;
@Autowired
private TokenEventDataBuilder tokenEventDataBuilder;
/**
* 工具调用状态跟踪映射
* key: toolName + timestamp
* value: pending状态的事件
*/
private final Map<String, ToolEvent> pendingToolCalls = new ConcurrentHashMap<>();
// 日期格式化对象，用于日志输出
private static final java.text.SimpleDateFormat DATE_FORMAT = new java.text.SimpleDateFormat("HH:mm:ss");
/**
* 记录工具调用开始事件
*
* @param toolName 工具名称
* @param input 工具输入参数
* @return 工具调用开始事件
*/
public ToolEvent recordToolCallStart(String toolName, Object input) {
try {
long currentTime = System.currentTimeMillis();
ToolEvent event = ToolEvent.builder()
.type("tool_call")
.timestamp(currentTime)
.toolName(toolName != null ? toolName : "未知工具")
.toolInput(WorkPanelUtils.convertToMap(input))
.toolStatus("pending")
.build();
// 跟踪pending状态的工具调用
String key = generateToolCallKey(toolName, currentTime);
pendingToolCalls.put(key, event);
// 输出工具调用日志
logToolCallStart(toolName, input, currentTime);
log.debug("已记录工具调用开始: 工具={}, 状态=pending", toolName);
return event;
} catch (Exception e) {
log.error("记录工具调用开始时发生错误: toolName={}", toolName, e);
return null;
}
}
/**
* 记录工具调用完成事件
*
* @param toolName 工具名称
* @param output 工具输出结果
* @param status 执行状态（success/failure）
* @param executionTime 执行时间（毫秒）
* @return 工具调用完成事件
*/
public ToolEvent recordToolCallComplete(String toolName, Object output, String status, Long executionTime) {
return handleToolCallCompletion(toolName, output, status, executionTime);
}
/**
* 记录工具调用错误事件
*
* @param toolName 工具名称
* @param errorMessage 错误信息
* @return 工具调用错误事件
*/
public ToolEvent recordToolCallError(String toolName, String errorMessage) {
return handleToolCallError(toolName, errorMessage);
}
/**
* 记录思考事件
*
* @param content 思考内容
* @param thinkingType 思考类型（分析、规划、执行等）
* @return 思考事件
*/
public ThoughtEvent recordThinking(String content, String thinkingType) {
try {
// 过滤掉空内容，避免记录过多无关信息
if (content == null || content.trim().isEmpty()) {
return null;
}
ThoughtEvent event = ThoughtEvent.builder()
.type("thought")
.timestamp(System.currentTimeMillis())
.content(content)
.thinkingType(thinkingType != null ? thinkingType : "reasoning")
.build();
log.debug("已记录思考过程: 类型={}, 内容={}", thinkingType, content);
return event;
} catch (Exception e) {
log.error("记录思考过程时发生错误: content={}, type={}", content, thinkingType, e);
return null;
}
}
/**
* 记录最终答案事件
*
* @param content 最终答案内容
* @return 最终答案事件
*/
public ResultEvent recordFinalAnswer(String content) {
try {
ResultEvent event = ResultEvent.builder()
.type("result")
.timestamp(System.currentTimeMillis())
.content(content)
.build();
log.debug("已记录最终答案: 内容={}", content);
return event;
} catch (Exception e) {
log.error("记录最终答案时发生错误: content={}", content, e);
return null;
}
}
/**
* 记录日志事件
*
* @param message 日志消息
* @param level 日志级别（info/warn/error/debug）
* @return 日志事件
*/
public LogEvent recordLog(String message, String level) {
try {
LogEvent event = LogEvent.builder()
.type("log")
.timestamp(System.currentTimeMillis())
.content(message)
.logLevel(level != null ? level : "info")
.build();
log.debug("已记录日志: 级别={}, 消息={}", level, message);
return event;
} catch (Exception e) {
log.error("记录日志时发生错误: message={}, level={}", message, level, e);
return null;
}
}
/**
* 记录嵌入事件
*
* @param url 嵌入资源URL（可选）
* @param type MIME类型
* @param title 嵌入标题
* @param htmlContent HTML内容（可选）
* @return 嵌入事件
*/
public EmbedEvent recordEmbed(String url, String type, String title, String htmlContent) {
try {
EmbedEvent event = EmbedEvent.builder()
.type("embed")
.timestamp(System.currentTimeMillis())
.embedUrl(url)
.embedType(type)
.embedTitle(title != null ? title : "网页预览")
.embedHtmlContent(htmlContent)
.build();
log.debug("已记录嵌入事件: 标题={}, URL={}", title, url);
return event;
} catch (Exception e) {
log.error("记录嵌入事件时发生错误: title={}, url={}", title, url, e);
return null;
}
}
/**
* 构建工作面板事件数据
*
* @param event 工作面板事件
* @return 事件数据
*/
public Map<String, Object> buildWorkPanelEventData(WorkPanelEvent event) {
if (event == null) {
return null;
}
// 从对象池获取Map，避免频繁创建对象
Map<String, Object> data = acquireMap();
// 设置基础属性
data.put("type", event.getType());
data.put("eventType", event.getType());
data.put("timestamp", event.getTimestamp());
// 根据事件类型设置特有属性
setEventSpecificProperties(event, data);
// 处理事件类型特定逻辑
processEventTypeSpecifics(event, data);
return data;
}
/**
* 发送工作面板事件给指定的SSE连接
*
* @param emitter SSE发射器
* @param event 工作面板事件
* @throws IOException IO异常
*/
public void sendWorkPanelEvent(SseEmitter emitter, WorkPanelEvent event) throws IOException {
if (event == null) {
log.warn("工作面板事件为空，无法发送事件");
return;
}
try {
// 检查emitter是否已经完成，避免向已完成的连接发送数据
java.lang.reflect.Field completedField = SseEmitter.class.getDeclaredField("completed");
completedField.setAccessible(true);
boolean isCompleted = completedField.getBoolean(emitter);
if (isCompleted) {
log.debug("SSE emitter已完成，跳过发送工作面板事件");
return;
}
// 构建事件数据
Map<String, Object> data = buildWorkPanelEventData(event);
if (data != null) {
log.debug("准备发送工作面板事件: 类型={}, 事件内容={}", event.getType(), event);
log.debug("事件数据: {}", data);
// 发送事件
emitter.send(org.springframework.web.servlet.mvc.method.annotation.SseEmitter.event().name("message").data(data));
log.debug("工作面板事件发送成功: 类型={}", event.getType());
} else {
log.warn("构建事件数据失败，无法发送事件: 类型={}", event.getType());
}
} catch (IllegalStateException e) {
// 处理 emitter 已关闭的情况
log.debug("无法发送工作面板事件，emitter已关闭或完成: {}", e.getMessage());
// 不重新抛出异常，避免影响主流程
} catch (Exception e) {
// 记录详细错误信息，但不中断主流程
if (!(e instanceof java.lang.reflect.InaccessibleObjectException) && !(e instanceof java.lang.NoSuchFieldException)) {
log.error("发送工作面板事件失败: 类型={}, 错误={}", event.getType(), e.getMessage(), e);
}
// 其他异常不重新抛出，避免影响主流程
}
}
/**
* 发送工作面板事件给指定用户
*
* @param emitter SSE发射器
* @param event 工作面板事件
*/
public void sendWorkPanelEventToUser(SseEmitter emitter, WorkPanelEvent event) {
log.debug("开始发送工作面板事件");
if (emitter != null) {
try {
// 直接向当前 emitter 发送事件
sendWorkPanelEvent(emitter, event);
log.debug("已发送工作面板事件到客户端: {}", event.getType());
} catch (IOException e) {
log.error("发送工作面板事件失败: {}", e.getMessage(), e);
}
} else {
log.debug("连接已失效，跳过发送事件: {}", event.getType());
}
}
/**
* 发送错误事件
*
* @param emitter SSE发射器
* @param errorMessage 错误信息
* @throws IOException IO异常
*/
public void sendErrorEvent(SseEmitter emitter, String errorMessage) throws IOException {
if (emitter == null) {
log.warn("SSE发射器为空，无法发送错误事件");
return;
}
try {
// 检查emitter是否已经完成，避免向已完成的连接发送数据
java.lang.reflect.Field completedField = SseEmitter.class.getDeclaredField("completed");
completedField.setAccessible(true);
boolean isCompleted = completedField.getBoolean(emitter);
if (isCompleted) {
log.debug("SSE emitter已完成，跳过发送错误事件");
return;
}
// 构建错误事件数据
Map<String, Object> data = errorEventDataBuilder.createErrorEventData(errorMessage);
if (data != null) {
log.debug("准备发送错误事件: 错误信息={}", errorMessage);
log.debug("错误事件数据: {}", data);
// 发送事件
emitter.send(SseEmitter.event().name("error").data(data));
log.debug("错误事件发送成功");
} else {
log.warn("构建错误事件数据失败，无法发送事件");
}
} catch (IllegalStateException e) {
// 处理 emitter 已关闭的情况
log.debug("无法发送错误事件，emitter已关闭或完成: {}", e.getMessage());
// 不重新抛出异常，避免影响主流程
} catch (Exception e) {
// 记录详细错误信息，但不中断主流程
if (!(e instanceof java.lang.reflect.InaccessibleObjectException) && !(e instanceof java.lang.NoSuchFieldException)) {
log.error("发送错误事件失败: 错误信息={}, 错误={}", errorMessage, e.getMessage(), e);
}
// 其他异常不重新抛出，避免影响主流程
}
}
/**
* 发送Token事件
*
* @param emitter SSE发射器
* @param token Token内容
* @throws IOException IO异常
*/
public void sendTokenEvent(SseEmitter emitter, String token) throws IOException {
if (emitter == null) {
log.warn("SSE发射器为空，无法发送token事件");
return;
}
try {
// 检查emitter是否已经完成，避免向已完成的连接发送数据
java.lang.reflect.Field completedField = SseEmitter.class.getDeclaredField("completed");
completedField.setAccessible(true);
boolean isCompleted = completedField.getBoolean(emitter);
if (isCompleted) {
log.debug("SSE emitter已完成，跳过发送token事件");
return;
}
// 构建token事件数据
Map<String, Object> data = tokenEventDataBuilder.createOptimizedTokenEventData(token);
if (data != null) {
// log.debug("准备发送token事件: token长度={}", token != null ? token.length() : 0);
// 发送事件
emitter.send(SseEmitter.event().name("token").data(data));
// log.debug("token事件发送成功");
} else {
log.warn("构建token事件数据失败，无法发送事件");
}
} catch (IllegalStateException e) {
// 处理 emitter 已关闭的情况
log.debug("无法发送token事件，emitter已关闭或完成: {}", e.getMessage());
// 不重新抛出异常，避免影响主流程
} catch (Exception e) {
// 记录详细错误信息，但不中断主流程
if (!(e instanceof java.lang.reflect.InaccessibleObjectException) && !(e instanceof java.lang.NoSuchFieldException)) {
log.error("发送token事件失败: token长度={}, 错误={}", token != null ? token.length() : 0, e.getMessage(), e);
}
// 其他异常不重新抛出，避免影响主流程
}
}
// ==================== 私有方法 ====================
/**
* 处理工具调用完成
*
* @param toolName 工具名称
* @param output 工具输出结果
* @param status 执行状态
* @param executionTime 执行时间
* @return 工作面板事件
*/
private ToolEvent handleToolCallCompletion(String toolName, Object output, String status, Long executionTime) {
long currentTime = System.currentTimeMillis();
// 查找对应的pending事件
ToolEvent pendingEvent = findPendingToolCall(toolName);
if (pendingEvent != null) {
// 移除跟踪记录
String key = generateToolCallKey(toolName, pendingEvent.getTimestamp());
pendingToolCalls.remove(key);
// 更新现有事件
updatePendingEventForCompletion(pendingEvent, output, status, executionTime, currentTime);
// 输出工具调用完成日志
logToolCallCompletion(toolName, output, status, executionTime, currentTime);
log.debug("已记录工具调用完成: 工具={}, 状态={}, 执行时间={}ms", toolName, status, executionTime);
return pendingEvent;
} else {
// 如果没有对应的pending事件，创建一个新事件
ToolEvent event = createNewCompletionEvent(toolName, output, status, executionTime, currentTime);
log.debug("已记录工具调用完成（无对应pending事件）: 工具={}, 状态={}, 执行时间={}ms", toolName, status, executionTime);
return event;
}
}
/**
* 处理工具调用错误
*
* @param toolName 工具名称
* @param errorMessage 错误信息
* @return 工作面板事件
*/
private ToolEvent handleToolCallError(String toolName, String errorMessage) {
// 查找对应的pending事件
ToolEvent pendingEvent = findPendingToolCall(toolName);
if (pendingEvent != null) {
// 移除跟踪记录
String key = generateToolCallKey(toolName, pendingEvent.getTimestamp());
pendingToolCalls.remove(key);
// 更新现有事件为错误状态
pendingEvent.setType("tool_error");
// 使用反射设置content字段
try {
java.lang.reflect.Field contentField = WorkPanelEvent.class.getDeclaredField("content");
contentField.setAccessible(true);
contentField.set(pendingEvent, errorMessage);
} catch (Exception e) {
// 如果反射失败，忽略错误
}
pendingEvent.setToolStatus("failure");
pendingEvent.setTimestamp(System.currentTimeMillis());
log.debug("已记录工具调用错误: 工具={}, 错误={}", toolName, errorMessage);
return pendingEvent;
} else {
// 如果没有对应的pending事件，创建一个新事件
ToolEvent event = ToolEvent.builder()
.type("tool_error")
.timestamp(System.currentTimeMillis())
.toolName(toolName)
.toolStatus("failure")
.build();
// 使用反射设置content字段
try {
java.lang.reflect.Field contentField = WorkPanelEvent.class.getDeclaredField("content");
contentField.setAccessible(true);
contentField.set(event, errorMessage);
} catch (Exception e) {
// 如果反射失败，忽略错误
}
log.debug("已记录工具调用错误（无对应pending事件）: 工具={}, 错误={}", toolName, errorMessage);
return event;
}
}
/**
* 生成工具调用键
*
* @param toolName 工具名称
* @param timestamp 时间戳
* @return 工具调用键
*/
private String generateToolCallKey(String toolName, long timestamp) {
return (toolName != null ? toolName : "unknown") + "_" + timestamp;
}
/**
* 查找pending状态的工具调用事件
*
* @param toolName 工具名称
* @return pending状态的事件，如果未找到则返回null
*/
private ToolEvent findPendingToolCall(String toolName) {
if (toolName == null) {
return null;
}
// 查找最近的该工具的pending事件
for (Map.Entry<String, ToolEvent> entry : pendingToolCalls.entrySet()) {
ToolEvent event = entry.getValue();
if (toolName.equals(event.getToolName()) && "pending".equals(event.getToolStatus())) {
return event;
}
}
return null;
}
/**
* 更新pending事件为完成状态
*
* @param pendingEvent pending状态的事件
* @param output 工具输出结果
* @param status 执行状态
* @param executionTime 执行时间
* @param currentTime 当前时间
*/
private void updatePendingEventForCompletion(ToolEvent pendingEvent, Object output, String status, Long executionTime, long currentTime) {
pendingEvent.setType(WorkPanelUtils.getEventTypeFromStatus(status));
pendingEvent.setToolOutput(output);
pendingEvent.setToolStatus(status != null ? status : "unknown");
pendingEvent.setExecutionTime(executionTime);
pendingEvent.setTimestamp(currentTime);
}
/**
* 创建新的完成事件
*
* @param toolName 工具名称
* @param output 工具输出结果
* @param status 执行状态
* @param executionTime 执行时间
* @param currentTime 当前时间
* @return 新的完成事件
*/
private ToolEvent createNewCompletionEvent(String toolName, Object output, String status, Long executionTime, long currentTime) {
return ToolEvent.builder()
.type(WorkPanelUtils.getEventTypeFromStatus(status))
.timestamp(currentTime)
.toolName(toolName != null ? toolName : "未知工具")
.toolOutput(output)
.toolStatus(status != null ? status : "unknown")
.executionTime(executionTime)
.build();
}
/**
* 输出工具调用开始日志
*
* @param toolName 工具名称
* @param input 工具输入参数
* @param currentTime 当前时间
*/
private void logToolCallStart(String toolName, Object input, long currentTime) {
String formattedTime = DATE_FORMAT.format(new java.util.Date(currentTime));
log.info("\n🔧 工具调用: [{}]\n⏰ 时间: {}\n📥 输入: {}\n📊 状态: 处理中",
toolName != null ? toolName : "未知工具",
formattedTime,
WorkPanelUtils.convertToJsonString(input));
}
/**
* 输出工具调用完成日志
*
* @param toolName 工具名称
* @param output 工具输出结果
* @param status 执行状态
* @param executionTime 执行时间
* @param currentTime 当前时间
*/
private void logToolCallCompletion(String toolName, Object output, String status, Long executionTime, long currentTime) {
String formattedTime = DATE_FORMAT.format(new java.util.Date(currentTime));
if ("success".equals(status)) {
log.info("\n🔧 工具调用: [{}]\n⏰ 时间: {}\n✅ 状态: 成功\n📤 输出: {}{}",
toolName != null ? toolName : "未知工具",
formattedTime,
WorkPanelUtils.convertToJsonString(output),
executionTime != null ? "\n⏱️ 耗时: " + executionTime + "ms" : "");
} else if ("failure".equals(status) || "error".equals(status)) {
log.info("\n🔧 工具调用: [{}]\n⏰ 时间: {}\n❌ 状态: 失败\n💬 错误: {}{}",
toolName != null ? toolName : "未知工具",
formattedTime,
WorkPanelUtils.convertToJsonString(output),
executionTime != null ? "\n⏱️ 耗时: " + executionTime + "ms" : "");
}
}
/**
* 根据事件类型设置特有属性
*
* @param event 工作面板事件
* @param data 事件数据
*/
private void setEventSpecificProperties(WorkPanelEvent event, Map<String, Object> data) {
String eventType = event.getType();
// 设置通用属性
if (event instanceof ThoughtEvent) {
data.put("content", ((ThoughtEvent) event).getContent());
} else if (event instanceof LogEvent) {
data.put("content", ((LogEvent) event).getContent());
} else if (event instanceof ResultEvent) {
data.put("content", ((ResultEvent) event).getContent());
}
switch (eventType) {
case "thought":
// 思考事件特有属性
if (event instanceof ThoughtEvent) {
ThoughtEvent thoughtEvent = (ThoughtEvent) event;
data.put("thinkingType", thoughtEvent.getThinkingType());
}
data.put("title", generateTitleForThought(event));
break;
case "tool_call":
case "tool_result":
case "tool_error":
// 工具事件特有属性
if (event instanceof ToolEvent) {
ToolEvent toolEvent = (ToolEvent) event;
data.put("toolName", toolEvent.getToolName());
data.put("toolInput", toolEvent.getToolInput());
data.put("toolOutput", toolEvent.getToolOutput());
data.put("toolStatus", toolEvent.getToolStatus());
data.put("executionTime", toolEvent.getExecutionTime());
// 根据状态设置错误信息
if ("tool_error".equals(eventType)) {
data.put("errorMessage", toolEvent.getErrorMessage());
data.put("errorCode", toolEvent.getErrorCode());
}
}
data.put("title", event instanceof ToolEvent && ((ToolEvent) event).getToolName() != null ? ((ToolEvent) event).getToolName() : "未知工具");
break;
case "log":
// 日志事件特有属性
if (event instanceof LogEvent) {
LogEvent logEvent = (LogEvent) event;
data.put("logLevel", logEvent.getLogLevel());
}
data.put("title", "日志");
break;
case "embed":
// 嵌入事件特有属性
if (event instanceof EmbedEvent) {
EmbedEvent embedEvent = (EmbedEvent) event;
data.put("embedUrl", embedEvent.getEmbedUrl());
data.put("embedType", embedEvent.getEmbedType());
data.put("embedTitle", embedEvent.getEmbedTitle());
data.put("embedHtmlContent", embedEvent.getEmbedHtmlContent());
}
data.put("title", event instanceof EmbedEvent && ((EmbedEvent) event).getEmbedTitle() != null ? ((EmbedEvent) event).getEmbedTitle() : "网页预览");
break;
case "result":
// 结果事件特有属性
data.put("title", "最终答案");
break;
default:
// 默认标题
data.put("title", generateDefaultTitle(event));
break;
}
}
/**
* 生成思考事件的标题
*
* @param event 思考事件
* @return 标题
*/
private String generateTitleForThought(WorkPanelEvent event) {
String content = null;
if (event instanceof ThoughtEvent) {
content = ((ThoughtEvent) event).getContent();
} else if (event instanceof LogEvent) {
content = ((LogEvent) event).getContent();
} else if (event instanceof ResultEvent) {
content = ((ResultEvent) event).getContent();
}
if (content != null && !content.isEmpty()) {
return content.substring(0, Math.min(content.length(), 50)) + (content.length() > 50 ? "..." : "");
}
return "思考过程";
}
/**
* 生成默认标题
*
* @param event 事件
* @return 标题
*/
private String generateDefaultTitle(WorkPanelEvent event) {
String content = null;
if (event instanceof ThoughtEvent) {
content = ((ThoughtEvent) event).getContent();
} else if (event instanceof LogEvent) {
content = ((LogEvent) event).getContent();
} else if (event instanceof ResultEvent) {
content = ((ResultEvent) event).getContent();
}
if (content != null && !content.isEmpty()) {
return content.substring(0, Math.min(content.length(), 50)) + (content.length() > 50 ? "..." : "");
}
return "事件";
}
/**
* 处理事件类型特定逻辑
*
* @param event 工作面板事件
* @param data 事件数据
*/
private void processEventTypeSpecifics(WorkPanelEvent event, Map<String, Object> data) {
String eventType = event.getType();
if (eventType != null) {
switch (eventType) {
case "tool_call":
case "tool_result":
case "tool_error":
// 处理工具事件
if (event instanceof ToolEvent) {
processToolEvent((ToolEvent) event, data);
}
break;
case "thought":
// 处理思考事件
if (event instanceof ThoughtEvent) {
processThinkingEvent((ThoughtEvent) event, data);
}
break;
// 其他事件类型可以根据需要添加处理逻辑
}
}
}
/**
* 处理工具事件
*
* @param event 工具事件
* @param data 事件数据
*/
private void processToolEvent(ToolEvent event, Map<String, Object> data) {
if (event == null || data == null) {
return;
}
// 添加工具事件特定的日志
if (log.isInfoEnabled()) {
log.info("[工具事件] 类型={}, 工具={}, 有toolInput={}, 有toolOutput={}",
event.getType(),
event.getToolName(),
event.getToolInput() != null,
event.getToolOutput() != null);
}
if (log.isDebugEnabled()) {
log.debug("[工具事件详情] toolInput={}, toolOutput={}",
event.getToolInput(),
event.getToolOutput());
}
}
/**
* 处理思考事件
*
* @param event 思考事件
* @param data 事件数据
*/
private void processThinkingEvent(ThoughtEvent event, Map<String, Object> data) {
if (event == null || data == null) {
return;
}
// 对于最终答案类型的思考事件，添加特殊标记
if ("final_answer".equals(event.getThinkingType())) {
data.put("isFinalAnswer", true);
}
}
/**
* 从对象池获取HashMap实例
*/
private Map<String, Object> acquireMap() {
return mapPoolService.acquireMap();
}
/**
* 获取对象池统计信息
*/
public String getMapPoolStatistics() {
return mapPoolService.getMapPoolStatistics();
}
/**
* 将HashMap实例归还到对象池
*/
public void releaseMap(Map<String, Object> map) {
mapPoolService.releaseMap(map);
}
}
\ No newline at end of file
backend/src/main/java/pangea/hiagent/workpanel/event/EventTypeConverter.java deleted 100644 → 0
View file @ 55b91d26
package pangea.hiagent.workpanel.event;
import org.springframework.stereotype.Component;
/**
* 事件类型转换工具类
* 统一处理事件类型转换逻辑，避免在多个地方重复实现
*/
@Component
public class EventTypeConverter {
/**
* 根据状态确定事件类型
*/
public String getEventTypeFromStatus(String status) {
if (status == null) {
return "tool_result";
}
switch (status.toLowerCase()) {
case "success":
return "tool_result";
case "error":
case "failure":
return "tool_error";
default:
return "tool_result";
}
}
/**
* 获取状态文本
*/
public String getStatusText(String status) {
if (status == null) {
return "未知";
}
switch (status.toLowerCase()) {
case "success": return "成功";
case "pending": return "处理中";
case "error": return "错误";
case "failure": return "失败";
default: return status;
}
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment