diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java index ae4b9999..f54232b6 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolExecutionExecutor.java @@ -177,6 +177,19 @@ public class ToolExecutionExecutor { events.add(GraphEventPublisher.toolStart(toolName, arguments)); + // 0. 子会话工具拦截：委派上下文中的子 Agent 禁止调用特定工具 + if (vip.mate.tool.builtin.DelegationContext.currentDepth() > 0) { + java.util.Set denied = vip.mate.tool.builtin.DelegationContext.childDeniedTools(); + if (denied.contains(toolName)) { + String msg = "[安全限制] 子 Agent 不允许使用工具: " + toolName; + log.info("[ToolExecutor] Child agent blocked from using tool: {}", toolName); + events.add(GraphEventPublisher.toolComplete(toolName, msg, false)); + allResponses.add(new org.springframework.ai.chat.messages.ToolResponseMessage.ToolResponse( + toolCall.id(), toolName, msg)); + continue; + } + } + // 1. JSON 校验 if (arguments != null && !arguments.isBlank()) { try { diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index c6d28658..44ec809b 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -101,6 +101,30 @@ public class ChatStreamTracker { private final ConcurrentHashMap runs = new ConcurrentHashMap<>(); + /** 事件 relay：子会话事件转发到父会话（用于 Agent 委派进度可见性） */ + private final ConcurrentHashMap>> eventRelays = new ConcurrentHashMap<>(); + + /** + * 注册事件 relay：将 sourceConversationId 的广播事件同时转发给 listener。 + * 返回一个 Runnable，调用后取消注册。 + */ + public Runnable addEventRelay(String sourceConversationId, + java.util.function.BiConsumer listener) { + eventRelays.computeIfAbsent(sourceConversationId, k -> new java.util.concurrent.CopyOnWriteArrayList<>()) + .add(listener); + log.debug("Event relay registered for conversation {}", sourceConversationId); + return () -> { + List> listeners = eventRelays.get(sourceConversationId); + if (listeners != null) { + listeners.remove(listener); + if (listeners.isEmpty()) { + eventRelays.remove(sourceConversationId); + } + } + log.debug("Event relay removed for conversation {}", sourceConversationId); + }; + } + /** 心跳调度线程池（守护线程） */ private final ScheduledExecutorService heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> { @@ -214,6 +238,18 @@ public class ChatStreamTracker { } } } + + // 事件 relay：转发给注册的监听器（用于子会话→父会话进度传递） + List> relays = eventRelays.get(conversationId); + if (relays != null) { + for (var relay : relays) { + try { + relay.accept(eventName, jsonData); + } catch (Exception e) { + log.debug("Event relay error for {}: {}", conversationId, e.getMessage()); + } + } + } } /** diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java index 1b2ac084..c2e12e30 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java @@ -1,6 +1,8 @@ package vip.mate.tool.builtin; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.tool.annotation.Tool; @@ -9,16 +11,24 @@ import org.springframework.stereotype.Component; import vip.mate.agent.AgentService; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.repository.AgentMapper; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.workspace.conversation.ConversationService; -import java.util.List; -import java.util.UUID; +import java.util.*; +import java.util.concurrent.*; import java.util.stream.Collectors; /** - * 内置工具：Agent 委派 + * 内置工具：Agent 委派（多 Agent 协作） *

- * 允许当前 Agent 将子任务委派给另一个 Agent 执行，实现多 Agent 协作。 - * 被委派的 Agent 在独立会话中运行，结果作为工具观察返回给调用方。 + * 支持两种模式： + *

    + *
  • {@link #delegateToAgent} — 单任务委派（串行）
    • + *
  • {@link #delegateParallel} — 多任务并行委派（最多 3 个子 Agent 同时执行）
    • + *
+ * 被委派的 Agent 在独立子会话中运行（记录父子关系）， + * 执行期间通过 SSE 事件 relay 向父会话实时推送进度。 + * 子 Agent 的工具集自动收窄——禁止递归委派和 Agent 发现工具。 * * @author MateClaw Team */ @@ -29,19 +39,37 @@ public class DelegateAgentTool { private static final int MAX_DELEGATION_DEPTH = 3; private static final int MAX_RESULT_LENGTH = 4000; + private static final int MAX_PARALLEL_CHILDREN = 3; + private static final int PARALLEL_TIMEOUT_SECONDS = 300; // 5 分钟 + + /** 子 Agent 禁用的工具：防递归 + 防副作用 */ + private static final Set CHILD_DENIED_TOOLS = Set.of( + "delegateToAgent", // 禁止递归委派 + "delegateParallel", // 禁止并行递归 + "listAvailableAgents" // 子 Agent 不需要发现其他 Agent + ); + + /** 并行委派执行器：JDK 21 虚拟线程，每个子 Agent 一个轻量级虚拟线程 */ + private static final ExecutorService DELEGATION_EXECUTOR = + Executors.newVirtualThreadPerTaskExecutor(); private final AgentService agentService; private final AgentMapper agentMapper; + private final ChatStreamTracker streamTracker; + private final ConversationService conversationService; + private final ObjectMapper objectMapper; + + // ==================== 单任务委派 ==================== @Tool(description = """ Delegate a task to another Agent for multi-agent collaboration. \ Target Agent executes in an independent session and returns its final reply. \ - Provide complete task context.""") + Parent receives real-time progress updates during execution. \ + For multiple parallel tasks, use delegateParallel instead.""") public String delegateToAgent( @ToolParam(description = "Target Agent name (exact match)") String agentName, @ToolParam(description = "Task description with complete context information") String task) { - // 1. 参数校验 if (agentName == null || agentName.isBlank()) { return "[错误] 请指定目标 Agent 名称。" + availableAgentsHint(); } @@ -49,52 +77,244 @@ public class DelegateAgentTool { return "[错误] 请提供任务描述。"; } - // 2. 递归深度检查 int depth = DelegationContext.currentDepth(); if (depth >= MAX_DELEGATION_DEPTH) { - return "[错误] 委派层级已达上限（" + MAX_DELEGATION_DEPTH + " 层），无法继续委派，请直接处理任务。"; + return "[错误] 委派层级已达上限（" + MAX_DELEGATION_DEPTH + " 层），请直接处理任务。"; } - // 3. 按名称查找目标 Agent - AgentEntity target = agentMapper.selectOne( - new LambdaQueryWrapper() - .eq(AgentEntity::getName, agentName.trim()) - .eq(AgentEntity::getEnabled, true)); - + AgentEntity target = findAgent(agentName); if (target == null) { return "[错误] 未找到名为「" + agentName + "」的已启用 Agent。" + availableAgentsHint(); } - // 4. 在独立会话中执行 - String tempConversationId = "delegate-" + UUID.randomUUID(); - log.info("Agent 委派: depth={}, target={}({}), task={}", - depth + 1, target.getName(), target.getId(), - task.length() > 100 ? task.substring(0, 100) + "..." : task); + String parentConversationId = resolveParentConversationId(); + String childConversationId = createChildConv(target, parentConversationId); - DelegationContext.enter(); + log.info("Agent 委派: depth={}, target={}({}), childConv={}, parentConv={}", + depth + 1, target.getName(), target.getId(), childConversationId, parentConversationId); + + // SSE 广播 + relay + boolean hasParent = parentConversationId != null && streamTracker.isRunning(parentConversationId); + if (hasParent) { + streamTracker.broadcastObject(parentConversationId, "delegation_start", Map.of( + "childConversationId", childConversationId, + "childAgentName", target.getName(), + "task", truncate(task, 200))); + } + Runnable stopRelay = hasParent ? registerRelay(childConversationId, parentConversationId, target.getName()) : null; + + // 执行 + ChildResult result = runSingleChild(0, target, task, parentConversationId, childConversationId); + + // 清理 + 广播结果 + if (stopRelay != null) stopRelay.run(); + if (hasParent) { + broadcastEnd(parentConversationId, childConversationId, target.getName(), result); + } + + return result.toToolResponse(target.getName()); + } + + // ==================== 并行委派 ==================== + + @Tool(description = """ + Delegate multiple tasks to different Agents in parallel (max 3). \ + Each task runs concurrently in an independent child session. \ + Use this when you have multiple independent sub-tasks that can run simultaneously. \ + Input is a JSON array: [{"agentName":"Agent名称","task":"任务描述"}, ...]""") + public String delegateParallel( + @ToolParam(description = "JSON array of tasks: [{\"agentName\":\"X\",\"task\":\"Y\"}, ...]") + String tasksJson) { + + // 1. 解析任务列表 + List> tasks; try { - String result = agentService.chat(target.getId(), task, tempConversationId); - String truncated = truncate(result, MAX_RESULT_LENGTH); - return "[Agent「" + target.getName() + "」的回复]\n\n" + truncated; + tasks = objectMapper.readValue(tasksJson, new TypeReference<>() {}); } catch (Exception e) { - log.error("Agent 委派执行失败: target={}, error={}", target.getName(), e.getMessage(), e); - return "[错误] Agent「" + target.getName() + "」执行失败: " + e.getMessage(); + return "[错误] 无法解析任务 JSON：" + e.getMessage() + "\n格式: [{\"agentName\":\"X\",\"task\":\"Y\"}]"; + } + + if (tasks == null || tasks.isEmpty()) { + return "[错误] 任务列表为空。"; + } + if (tasks.size() > MAX_PARALLEL_CHILDREN) { + return "[错误] 最多支持 " + MAX_PARALLEL_CHILDREN + " 个并行任务，当前 " + tasks.size() + " 个。"; + } + + int depth = DelegationContext.currentDepth(); + if (depth >= MAX_DELEGATION_DEPTH) { + return "[错误] 委派层级已达上限（" + MAX_DELEGATION_DEPTH + " 层），请直接处理任务。"; + } + + String parentConversationId = resolveParentConversationId(); + boolean hasParent = parentConversationId != null && streamTracker.isRunning(parentConversationId); + + // 2. 主线程：校验所有 Agent + 创建子会话 + 注册 relay + record PreparedChild(int index, AgentEntity agent, String task, String childConvId, Runnable stopRelay) {} + List prepared = new ArrayList<>(); + List errors = new ArrayList<>(); + + for (int i = 0; i < tasks.size(); i++) { + Map t = tasks.get(i); + String agentName = t.get("agentName"); + String task = t.get("task"); + + if (agentName == null || agentName.isBlank() || task == null || task.isBlank()) { + errors.add("[任务 " + (i + 1) + "] agentName 或 task 为空"); + continue; + } + + AgentEntity agent = findAgent(agentName); + if (agent == null) { + errors.add("[任务 " + (i + 1) + "] 未找到 Agent「" + agentName + "」"); + continue; + } + + String childConvId = createChildConv(agent, parentConversationId); + Runnable stopRelay = hasParent ? registerRelay(childConvId, parentConversationId, agent.getName()) : null; + prepared.add(new PreparedChild(i, agent, task, childConvId, stopRelay)); + } + + if (prepared.isEmpty()) { + return "[错误] 所有任务校验失败：\n" + String.join("\n", errors); + } + + log.info("并行委派: {} 个任务, parentConv={}", prepared.size(), parentConversationId); + + // 3. 广播 delegation_start（并行模式） + if (hasParent) { + List> childrenInfo = prepared.stream().map(p -> Map.of( + "childConversationId", p.childConvId, + "childAgentName", p.agent.getName(), + "task", truncate(p.task, 100) + )).toList(); + streamTracker.broadcastObject(parentConversationId, "delegation_start", Map.of( + "parallel", true, + "children", childrenInfo)); + } + + // 4. 并行执行 + long startTime = System.currentTimeMillis(); + Map> futures = new LinkedHashMap<>(); + + for (PreparedChild p : prepared) { + CompletableFuture future = CompletableFuture.supplyAsync( + () -> runSingleChild(p.index, p.agent, p.task, parentConversationId, p.childConvId), + DELEGATION_EXECUTOR); + futures.put(p.index, future); + } + + // 5. 等待全部完成（带超时） + List results = new ArrayList<>(); + try { + CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0])) + .get(PARALLEL_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (TimeoutException e) { + log.warn("并行委派超时 ({}s)，收集已完成的结果", PARALLEL_TIMEOUT_SECONDS); + } catch (Exception e) { + log.error("并行委派异常: {}", e.getMessage()); + } + + // 收集结果（已完成的 + 超时的） + for (var entry : futures.entrySet()) { + int idx = entry.getKey(); + CompletableFuture f = entry.getValue(); + PreparedChild p = prepared.stream().filter(pp -> pp.index == idx).findFirst().orElse(null); + String agentName = p != null ? p.agent.getName() : "Unknown"; + + if (f.isDone() && !f.isCompletedExceptionally()) { + try { + results.add(f.get()); + } catch (Exception ex) { + results.add(ChildResult.error(idx, agentName, ex.getMessage())); + } + } else { + f.cancel(true); + results.add(ChildResult.error(idx, agentName, "超时 (" + PARALLEL_TIMEOUT_SECONDS + "s)")); + } + } + + long totalDurationMs = System.currentTimeMillis() - startTime; + + // 6. 清理 relay + for (PreparedChild p : prepared) { + if (p.stopRelay != null) p.stopRelay.run(); + } + + // 7. 广播 delegation_end + if (hasParent) { + streamTracker.broadcastObject(parentConversationId, "delegation_end", Map.of( + "parallel", true, + "totalDurationMs", totalDurationMs, + "success", results.stream().allMatch(r -> r.success), + "completedCount", results.stream().filter(r -> r.success).count(), + "totalCount", results.size())); + } + + // 8. 构建返回结果 + results.sort(Comparator.comparingInt(r -> r.taskIndex)); + StringBuilder sb = new StringBuilder(); + if (!errors.isEmpty()) { + sb.append("⚠️ 部分任务未执行：\n"); + errors.forEach(e -> sb.append(" ").append(e).append("\n")); + sb.append("\n"); + } + sb.append("并行执行 ").append(results.size()).append(" 个任务（总耗时 ") + .append(totalDurationMs / 1000).append("s）：\n\n"); + for (ChildResult r : results) { + sb.append("---\n### [任务 ").append(r.taskIndex + 1).append("] Agent「").append(r.agentName).append("」"); + sb.append(r.success ? " ✓" : " ✗").append(" (").append(r.durationMs / 1000).append("s)\n\n"); + sb.append(r.success ? r.result : "[错误] " + r.error).append("\n\n"); + } + return truncate(sb.toString(), MAX_RESULT_LENGTH * 2); // 并行结果允许更长 + } + + // ==================== 子 Agent 执行（单/并行共用） ==================== + + /** + * 执行单个子 Agent。在子线程内独立设置 DelegationContext，解决 ThreadLocal 并行问题。 + */ + private ChildResult runSingleChild(int taskIndex, AgentEntity target, String task, + String parentConversationId, String childConversationId) { + DelegationContext.enter(parentConversationId, CHILD_DENIED_TOOLS); + try { + long startTime = System.currentTimeMillis(); + String result = agentService.chat(target.getId(), task, childConversationId); + long durationMs = System.currentTimeMillis() - startTime; + return ChildResult.success(taskIndex, target.getName(), truncate(result, MAX_RESULT_LENGTH), durationMs); + } catch (Exception e) { + log.error("子 Agent 执行失败: taskIndex={}, agent={}, error={}", + taskIndex, target.getName(), e.getMessage()); + return ChildResult.error(taskIndex, target.getName(), e.getMessage()); } finally { DelegationContext.exit(); } } + /** 子 Agent 执行结果 */ + private record ChildResult(int taskIndex, String agentName, boolean success, + String result, String error, long durationMs) { + static ChildResult success(int idx, String name, String result, long ms) { + return new ChildResult(idx, name, true, result, null, ms); + } + static ChildResult error(int idx, String name, String err) { + return new ChildResult(idx, name, false, null, err != null ? err : "Unknown error", 0); + } + String toToolResponse(String agentName) { + if (success) return "[Agent「" + agentName + "」的回复]\n\n" + result; + return "[错误] Agent「" + agentName + "」执行失败: " + error; + } + } + + // ==================== 辅助方法 ==================== + @Tool(description = "List all available Agents (enabled), including name, type, and description.") public String listAvailableAgents() { List agents = agentMapper.selectList( new LambdaQueryWrapper() .eq(AgentEntity::getEnabled, true) .orderByAsc(AgentEntity::getName)); - - if (agents.isEmpty()) { - return "当前没有可用的 Agent。"; - } - + if (agents.isEmpty()) return "当前没有可用的 Agent。"; StringBuilder sb = new StringBuilder("可用 Agent 列表：\n\n"); for (AgentEntity agent : agents) { sb.append("- **").append(agent.getName()).append("**"); @@ -107,23 +327,67 @@ public class DelegateAgentTool { return sb.toString(); } - private String availableAgentsHint() { - List agents = agentMapper.selectList( - new LambdaQueryWrapper() - .eq(AgentEntity::getEnabled, true) - .select(AgentEntity::getName)); - if (agents.isEmpty()) { - return ""; + private AgentEntity findAgent(String name) { + return agentMapper.selectOne(new LambdaQueryWrapper() + .eq(AgentEntity::getName, name.trim()) + .eq(AgentEntity::getEnabled, true)); + } + + private String createChildConv(AgentEntity target, String parentConversationId) { + String childConvId = "child-" + UUID.randomUUID().toString().substring(0, 12); + try { + conversationService.createChildConversation(childConvId, target.getId(), "system", + target.getWorkspaceId() != null ? target.getWorkspaceId() : 1L, parentConversationId); + } catch (Exception e) { + log.warn("Failed to create child conversation: {}", e.getMessage()); } - String names = agents.stream() - .map(AgentEntity::getName) - .collect(Collectors.joining("、")); - return "\n可用 Agent: " + names; + return childConvId; + } + + private Runnable registerRelay(String childConvId, String parentConvId, String childAgentName) { + return streamTracker.addEventRelay(childConvId, (eventName, jsonData) -> { + if ("tool_call_started".equals(eventName) || "tool_call_completed".equals(eventName) || "phase".equals(eventName)) { + try { + streamTracker.broadcastObject(parentConvId, "delegation_progress", Map.of( + "childConversationId", childConvId, + "childAgentName", childAgentName, + "originalEvent", eventName, + "data", jsonData)); + } catch (Exception e) { + log.debug("Relay error: {}", e.getMessage()); + } + } + }); + } + + private void broadcastEnd(String parentConvId, String childConvId, String agentName, ChildResult result) { + streamTracker.broadcastObject(parentConvId, "delegation_end", Map.of( + "childConversationId", childConvId, + "childAgentName", agentName, + "success", result.success, + "durationMs", result.durationMs, + "resultPreview", result.success ? truncate(result.result, 200) : (result.error != null ? result.error : "") + )); + } + + private String resolveParentConversationId() { + try { + String ctxConvId = ToolExecutionContext.conversationId(); + if (ctxConvId != null && !ctxConvId.isBlank()) return ctxConvId; + } catch (Exception ignored) {} + return DelegationContext.parentConversationId(); + } + + private String availableAgentsHint() { + List agents = agentMapper.selectList(new LambdaQueryWrapper() + .eq(AgentEntity::getEnabled, true).select(AgentEntity::getName)); + if (agents.isEmpty()) return ""; + return "\n可用 Agent: " + agents.stream().map(AgentEntity::getName).collect(Collectors.joining("、")); } private static String truncate(String text, int maxLength) { if (text == null) return ""; if (text.length() <= maxLength) return text; - return text.substring(0, maxLength) + "\n\n... [结果已截断，原文共 " + text.length() + " 字符]"; + return text.substring(0, maxLength) + "\n... [截断，原文 " + text.length() + " 字符]"; } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegationContext.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegationContext.java index f466e3fb..840b2fa6 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegationContext.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegationContext.java @@ -1,16 +1,20 @@ package vip.mate.tool.builtin; +import java.util.Set; + /** - * 跟踪 Agent 委派调用深度，防止无限递归。 + * 跟踪 Agent 委派调用的上下文信息，防止无限递归并传递父会话信息。 *

- * 使用 ThreadLocal 存储当前线程的委派层级。 - * 每次 {@link DelegateAgentTool} 发起委派时 +1，返回后 -1。 + * 使用 ThreadLocal 存储当前线程的委派层级、父会话 ID 和子 Agent 禁用工具集。 + * 每次 {@link DelegateAgentTool} 发起委派时调用 enter()，返回后调用 exit()。 * * @author MateClaw Team */ public final class DelegationContext { private static final ThreadLocal DEPTH = ThreadLocal.withInitial(() -> 0); + private static final ThreadLocal PARENT_CONVERSATION_ID = new ThreadLocal<>(); + private static final ThreadLocal> CHILD_DENIED_TOOLS = new ThreadLocal<>(); private DelegationContext() {} @@ -19,9 +23,29 @@ public final class DelegationContext { return DEPTH.get(); } - /** 进入下一层委派 */ - public static void enter() { + /** 获取父会话 ID（用于事件 relay） */ + public static String parentConversationId() { + return PARENT_CONVERSATION_ID.get(); + } + + /** 获取子 Agent 禁用的工具集 */ + public static Set childDeniedTools() { + Set denied = CHILD_DENIED_TOOLS.get(); + return denied != null ? denied : Set.of(); + } + + /** 进入下一层委派（带父会话 ID 和子 Agent 工具限制） */ + public static void enter(String parentConversationId, Set deniedTools) { DEPTH.set(DEPTH.get() + 1); + PARENT_CONVERSATION_ID.set(parentConversationId); + if (deniedTools != null) { + CHILD_DENIED_TOOLS.set(deniedTools); + } + } + + /** 进入下一层委派（兼容旧调用） */ + public static void enter() { + enter(null, null); } /** 退出当前委派层 */ @@ -29,6 +53,8 @@ public final class DelegationContext { int current = DEPTH.get(); if (current <= 1) { DEPTH.remove(); + PARENT_CONVERSATION_ID.remove(); + CHILD_DENIED_TOOLS.remove(); } else { DEPTH.set(current - 1); } diff --git a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java index c7248109..5112b53d 100644 --- a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java +++ b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java @@ -60,8 +60,10 @@ public class ConversationService { */ public List listConversations(String username, Long workspaceId) { // 同时返回当前用户的会话 和 定时任务（system）产生的会话 + // 排除子会话（委派产生的子会话不在侧边栏显示） LambdaQueryWrapper wrapper = new LambdaQueryWrapper() .in(ConversationEntity::getUsername, username, SYSTEM_USER) + .isNull(ConversationEntity::getParentConversationId) .orderByDesc(ConversationEntity::getLastActiveTime); if (workspaceId != null) { wrapper.eq(ConversationEntity::getWorkspaceId, workspaceId); @@ -129,6 +131,20 @@ public class ConversationService { return conv; } + /** + * 创建子会话（委派场景），关联父会话 ID。 + */ + @Transactional + public ConversationEntity createChildConversation(String childConversationId, Long agentId, + String username, Long workspaceId, + String parentConversationId) { + ConversationEntity conv = getOrCreateConversation(childConversationId, agentId, username, workspaceId); + conv.setParentConversationId(parentConversationId); + conv.setTitle("子任务"); + conversationMapper.updateById(conv); + return conv; + } + /** * 获取或创建共享渠道会话。 *

diff --git a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/model/ConversationEntity.java b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/model/ConversationEntity.java index 698c8700..e002927e 100644 --- a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/model/ConversationEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/model/ConversationEntity.java @@ -45,6 +45,9 @@ public class ConversationEntity { /** 所属工作区 ID（默认 1 = default） */ private Long workspaceId; + /** 父会话 ID（委派场景下，子会话记录其父会话的 conversationId） */ + private String parentConversationId; + @TableField(fill = FieldFill.INSERT) private LocalDateTime createTime; diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V5__conversation_parent.sql b/mateclaw-server/src/main/resources/db/migration/h2/V5__conversation_parent.sql new file mode 100644 index 00000000..8d4d6e68 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V5__conversation_parent.sql @@ -0,0 +1,3 @@ +-- V5: Add parent_conversation_id to mate_conversation for multi-agent delegation tracking +ALTER TABLE mate_conversation ADD COLUMN IF NOT EXISTS parent_conversation_id VARCHAR(64) DEFAULT NULL; +CREATE INDEX IF NOT EXISTS idx_conversation_parent ON mate_conversation(parent_conversation_id); diff --git a/mateclaw-server/src/main/resources/db/schema.sql b/mateclaw-server/src/main/resources/db/schema.sql index 4c9aa35d..8e2db168 100644 --- a/mateclaw-server/src/main/resources/db/schema.sql +++ b/mateclaw-server/src/main/resources/db/schema.sql @@ -158,6 +158,7 @@ CREATE TABLE IF NOT EXISTS mate_conversation ( last_active_time DATETIME, stream_status VARCHAR(16) NOT NULL DEFAULT 'idle', workspace_id BIGINT NOT NULL DEFAULT 1, + parent_conversation_id VARCHAR(64) DEFAULT NULL, create_time DATETIME NOT NULL, update_time DATETIME NOT NULL, deleted INT NOT NULL DEFAULT 0 diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index 30415583..aa13d4f6 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -597,6 +597,86 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) + // ===== Agent 委派事件 ===== + stream.on('delegation_start', (data) => { + if (isStaleEvent(data)) return + streamPhase.value = 'executing_tool' + if (currentAssistantId.value) { + const segs = currentSegments.value + // 关闭之前的 thinking/content segment + const runningSeg = segs.findLast((s: MessageSegment) => s.status === 'running') + if (runningSeg) runningSeg.status = 'completed' + + if (data.parallel && Array.isArray(data.children)) { + // 并行模式：为每个子任务创建一个 delegation segment + for (const child of data.children) { + segs.push({ + id: genSegId(), + type: 'tool_call', + status: 'running', + toolName: `→ ${child.childAgentName || 'Agent'}`, + toolArgs: child.task || '', + timestamp: Date.now() + }) + } + } else { + // 单任务模式 + segs.push({ + id: genSegId(), + type: 'tool_call', + status: 'running', + toolName: `→ ${data.childAgentName || 'Agent'}`, + toolArgs: data.task || '', + timestamp: Date.now() + }) + } + flushSegmentsToMessage() + } + }) + + stream.on('delegation_progress', (data) => { + if (isStaleEvent(data)) return + if (currentAssistantId.value && data.originalEvent === 'tool_call_started') { + const segs = currentSegments.value + // 按 childAgentName 匹配对应的 delegation segment（并行时多个） + const childName = data.childAgentName || '' + const delegSeg = segs.findLast((s: MessageSegment) => + s.type === 'tool_call' && s.status === 'running' && s.toolName === `→ ${childName}`) + || segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) + if (delegSeg) { + const childData = typeof data.data === 'string' ? data.data : JSON.stringify(data.data) + delegSeg.toolArgs = (delegSeg.toolArgs || '') + '\n [子任务] ' + childData + } + } + }) + + stream.on('delegation_end', (data) => { + if (isStaleEvent(data)) return + if (currentAssistantId.value) { + const segs = currentSegments.value + if (data.parallel) { + // 并行模式：关闭所有 running 的 delegation segments + const totalMs = data.totalDurationMs ? Math.round(data.totalDurationMs / 1000) : 0 + segs.filter((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) + .forEach((s: MessageSegment) => { + s.status = 'completed' + s.toolName = (s.toolName || '') + (data.success ? ' ✓' : ' ✗') + }) + } else { + // 单任务模式 + const delegSeg = segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) + if (delegSeg) { + delegSeg.status = 'completed' + delegSeg.toolName = (delegSeg.toolName || '') + (data.success ? ' ✓' : ' ✗') + if (data.durationMs) { + delegSeg.toolArgs = (delegSeg.toolArgs || '') + `\n 耗时: ${Math.round(data.durationMs / 1000)}s` + } + } + } + flushSegmentsToMessage() + } + }) + stream.on('plan_created', (data) => { if (isStaleEvent(data)) return if (currentAssistantId.value) { diff --git a/mateclaw-ui/src/composables/chat/useStream.ts b/mateclaw-ui/src/composables/chat/useStream.ts index efaa4ecf..0c908491 100644 --- a/mateclaw-ui/src/composables/chat/useStream.ts +++ b/mateclaw-ui/src/composables/chat/useStream.ts @@ -39,6 +39,10 @@ export type SSEEventType = | 'tts_ready' // 浏览器执行事件 | 'browser_action' + // Agent 委派事件 + | 'delegation_start' + | 'delegation_progress' + | 'delegation_end' export interface SSEEvent { type: SSEEventType