【从0到1构建一个ClaudeAgent】协作-Agent团队 _

一个 Agent 干不完怎么办Java实现代码javapublic class AgentTeamsSystem { // --- 配置 --- private static final Path WORKDIR Paths.get(System.getProperty(user.dir)); private static final Path TEAM_DIR WORKDIR.resolve(.team); private static final Path INBOX_DIR TEAM_DIR.resolve(inbox); private static final Gson gson new GsonBuilder().setPrettyPrinting().create(); // 有效消息类型 private static final SetString VALID_MSG_TYPES Set.of( message, broadcast, shutdown_request, shutdown_response, plan_approval_response ); // --- 消息系统MessageBus--- static class MessageBus { private final Path inboxDir; public MessageBus(Path inboxDir) { this.inboxDir inboxDir; try { Files.createDirectories(inboxDir); } catch (IOException e) { throw new RuntimeException(Failed to create inbox directory, e); } } /** * 发送消息到指定智能体 */ public String send(String sender, String to, String content, String msgType, MapString, Object extra) { if (!VALID_MSG_TYPES.contains(msgType)) { return String.format(Error: Invalid type %s. Valid: %s, msgType, String.join(, , VALID_MSG_TYPES)); } MapString, Object message new LinkedHashMap(); message.put(type, msgType); message.put(from, sender); message.put(content, content); message.put(timestamp, System.currentTimeMillis() / 1000.0); if (extra ! null) { message.putAll(extra); } Path inboxPath inboxDir.resolve(to .jsonl); try { String jsonLine gson.toJson(message) \n; Files.writeString(inboxPath, jsonLine, StandardOpenOption.CREATE, StandardOpenOption.APPEND); return String.format(Sent %s to %s, msgType, to); } catch (IOException e) { return Error: e.getMessage(); } } /** * 读取并清空邮箱 */ public ListMapString, Object readInbox(String name) { Path inboxPath inboxDir.resolve(name .jsonl); if (!Files.exists(inboxPath)) { return new ArrayList(); } try { ListMapString, Object messages new ArrayList(); ListString lines Files.readAllLines(inboxPath); for (String line : lines) { if (!line.trim().isEmpty()) { Type type new TypeTokenMapString, Object(){}.getType(); MapString, Object message gson.fromJson(line, type); messages.add(message); } } // 清空邮箱消费模式 Files.writeString(inboxPath, ); return messages; } catch (IOException e) { return new ArrayList(); } } /** * 广播消息到所有队友 */ public String broadcast(String sender, String content, ListString teammates) { int count 0; for (String name : teammates) { if (!name.equals(sender)) { send(sender, name, content, broadcast); count; } } return String.format(Broadcast to %d teammates, count); } } // 初始化消息总线 private static final MessageBus BUS new MessageBus(INBOX_DIR); // --- 智能体管理器TeammateManager--- static class TeammateManager { private final Path teamDir; private final Path configPath; private MapString, Object config; private final MapString, Thread threads new ConcurrentHashMap(); private final MapString, AtomicBoolean stopFlags new ConcurrentHashMap(); public TeammateManager(Path teamDir) { this.teamDir teamDir; this.configPath teamDir.resolve(config.json); loadConfig(); } SuppressWarnings(unchecked) private void loadConfig() { try { if (Files.exists(configPath)) { String content Files.readString(configPath); Type type new TypeTokenMapString, Object(){}.getType(); this.config gson.fromJson(content, type); } else { this.config new HashMap(); config.put(team_name, default); config.put(members, new ArrayListMapString, Object()); saveConfig(); } } catch (IOException e) { throw new RuntimeException(Failed to load team config, e); } } SuppressWarnings(unchecked) public String spawn(String name, String role, String prompt) { MapString, Object member findMember(name); if (member ! null) { String status (String) member.get(status); if (!idle.equals(status) !shutdown.equals(status)) { return String.format(Error: %s is currently %s, name, status); } member.put(status, working); member.put(role, role); } else { member new LinkedHashMap(); member.put(name, name); member.put(role, role); member.put(status, working); ((ListMapString, Object) config.get(members)).add(member); } saveConfig(); // 停止之前的线程如果存在 if (threads.containsKey(name)) { stopFlags.get(name).set(true); try { threads.get(name).join(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } // 创建新的停止标志 AtomicBoolean stopFlag new AtomicBoolean(false); stopFlags.put(name, stopFlag); // 创建并启动新线程 Thread thread new Thread(() - teammateLoop(name, role, prompt, stopFlag), Teammate- name); thread.setDaemon(true); threads.put(name, thread); thread.start(); return String.format(Spawned %s (role: %s), name, role); } private void teammateLoop(String name, String role, String prompt, AtomicBoolean stopFlag) { String systemPrompt String.format( You are %s, role: %s, at %s. Use send_message to communicate. Complete your task., name, role, WORKDIR ); ListMapString, Object messages new ArrayList(); messages.add(Map.of(role, user, content, prompt)); // 最大迭代次数限制 for (int i 0; i 50 !stopFlag.get(); i) { try { // 检查邮箱 ListMapString, Object inbox BUS.readInbox(name); for (MapString, Object msg : inbox) { messages.add(Map.of(role, user, content, gson.toJson(msg))); } // 模拟调用 LLM MapString, Object response simulateTeammateLLMCall(systemPrompt, messages, name); if (response null || end_turn.equals(response.get(stop_reason))) { break; } // ... 执行工具调用 // 短暂休眠避免 CPU 过度使用 Thread.sleep(100); } catch (Exception e) { System.err.printf([%s] Error: %s%n, name, e.getMessage()); break; } } // 更新状态 SuppressWarnings(unchecked) MapString, Object member findMember(name); if (member ! null !shutdown.equals(member.get(status))) { member.put(status, idle); saveConfig(); } threads.remove(name); stopFlags.remove(name); } SuppressWarnings(unchecked) public String listAll() { ListMapString, Object members (ListMapString, Object) config.get(members); if (members.isEmpty()) { return No teammates.; } StringBuilder sb new StringBuilder(); sb.append(Team: ).append(config.get(team_name)).append(\n); for (MapString, Object member : members) { sb.append(String.format( %s (%s): %s%n, member.get(name), member.get(role), member.get(status) )); } return sb.toString().trim(); } SuppressWarnings(unchecked) public ListString memberNames() { ListMapString, Object members (ListMapString, Object) config.get(members); return members.stream() .map(m - (String) m.get(name)) .collect(Collectors.toList()); } /** * 获取活动成员数量 */ SuppressWarnings(unchecked) public int getActiveCount() { ListMapString, Object members (ListMapString, Object) config.get(members); int count 0; for (MapString, Object member : members) { if (working.equals(member.get(status))) { count; } } return count; } } // 初始化智能体管理器 private static final TeammateManager TEAM_MANAGER new TeammateManager(TEAM_DIR); // --- 工具枚举 --- public enum ToolType { BASH(bash, Run a shell command.), READ_FILE(read_file, Read file contents.), WRITE_FILE(write_file, Write content to file.), EDIT_FILE(edit_file, Replace exact text in file.), SPAWN_TEAMMATE(spawn_teammate, Spawn a persistent teammate that runs in its own thread.), // 新增 LIST_TEAMMATES(list_teammates, List all teammates with name, role, status.), // 新增 SEND_MESSAGE(send_message, Send a message to a teammates inbox.), // 新增 READ_INBOX(read_inbox, Read and drain the leads inbox.), // 新增 BROADCAST(broadcast, Send a message to all teammates.); // 新增 public final String name; public final String description; ToolType(String name, String description) { this.name name; this.description description; } } // --- 工具处理器映射 --- private static final MapString, ToolExecutor TOOL_HANDLERS new HashMap(); static { // ... 省略基础工具注册 // 团队管理工具 TOOL_HANDLERS.put(ToolType.SPAWN_TEAMMATE.name, args - { String name (String) args.get(name); String role (String) args.get(role); String prompt (String) args.get(prompt); return TEAM_MANAGER.spawn(name, role, prompt); }); TOOL_HANDLERS.put(ToolType.LIST_TEAMMATES.name, args - { return TEAM_MANAGER.listAll(); }); TOOL_HANDLERS.put(ToolType.SEND_MESSAGE.name, args - { String to (String) args.get(to); String content (String) args.get(content); String msgType (String) args.get(msg_type); if (msgType null) msgType message; return BUS.send(lead, to, content, msgType); }); TOOL_HANDLERS.put(ToolType.READ_INBOX.name, args - { ListMapString, Object inbox BUS.readInbox(lead); return gson.toJson(inbox); }); TOOL_HANDLERS.put(ToolType.BROADCAST.name, args - { String content (String) args.get(content); return BUS.broadcast(lead, content, TEAM_MANAGER.memberNames()); }); } // --- Agent 主循环领导智能体--- public static void agentLoop(ListMapString, Object messages) { while (true) { try { // 检查领导邮箱 ListMapString, Object inbox BUS.readInbox(lead); if (!inbox.isEmpty()) { String inboxJson gson.toJson(inbox); messages.add(Map.of( role, user, content, inbox inboxJson /inbox )); messages.add(Map.of( role, assistant, content, Noted inbox messages. )); // 邮箱自动注入自动检查并注入收到的消息 // 结构化格式用XML标签包裹便于LLM解析 } // 显示团队状态 int activeCount TEAM_MANAGER.getActiveCount(); if (activeCount 0) { System.out.printf([Active teammates: %d]%n, activeCount); } // ... 省略相同的 LLM 调用和工具执行逻辑 } catch (Exception e) { System.err.println(Error in agent loop: e.getMessage()); e.printStackTrace(); return; } } } }这段代码引入了智能体团队系统实现了多个 Agent 之间的协作核心思想持久队友 异步邮箱。什么是持久队友Persistent Teammates它们是长期存活、有身份意识的 Agent通过基于文件的邮箱JSONL 格式异步通信能够处理跨越单个执行周期的复杂任务委托。多智能体系统架构核心思想从单智能体系统升级为多智能体协作系统引入分布式、角色化、可通信的智能体团队实现复杂的协同工作流和分布式问题解决。java// 系统配置 private static final Path TEAM_DIR WORKDIR.resolve(.team); private static final Path INBOX_DIR TEAM_DIR.resolve(inbox); // 团队持久化.team目录存储团队配置 // 消息传递inbox子目录实现智能体间通信 // 文件系统基础通过文件系统实现简单的分布式通信多智能体协同多个智能体可以并行工作协同解决问题角色化分工不同智能体担任不同角色专业化分工持久化团队团队配置和状态可以持久化保存去中心化通信基于文件系统的轻量级消息传递消息总线系统MessageBusjava// 消息总线 - 智能体间通信基础设施 static class MessageBus { private final Path inboxDir; // 文件系统邮箱每个智能体一个jsonl文件 // 异步通信发送方不阻塞接收方主动拉取 // 松耦合智能体间通过邮箱解耦 /** * 发送消息到指定智能体 */ public String send(String sender, String to, String content, String msgType, MapString, Object extra) { if (!VALID_MSG_TYPES.contains(msgType)) { return String.format(Error: Invalid type %s. Valid: %s, msgType, String.join(, , VALID_MSG_TYPES)); } // 消息类型验证确保消息结构符合协议 MapString, Object message new LinkedHashMap(); message.put(type, msgType); message.put(from, sender); message.put(content, content); message.put(timestamp, System.currentTimeMillis() / 1000.0); // 结构化消息类型、发送方、内容、时间戳 // 可扩展支持额外字段 Path inboxPath inboxDir.resolve(to .jsonl); try { String jsonLine gson.toJson(message) \n; Files.writeString(inboxPath, jsonLine, StandardOpenOption.CREATE, StandardOpenOption.APPEND); // 追加写入支持多条消息 // JSONL格式每行一个JSON对象便于处理 } } /** * 读取并清空邮箱 */ public ListMapString, Object readInbox(String name) { Path inboxPath inboxDir.resolve(name .jsonl); if (!Files.exists(inboxPath)) { return new ArrayList(); } try { ListMapString, Object messages new ArrayList(); ListString lines Files.readAllLines(inboxPath); for (String line : lines) { if (!line.trim().isEmpty()) { Type type new TypeTokenMapString, Object(){}.getType(); MapString, Object message gson.fromJson(line, type); messages.add(message); } } // 清空邮箱消费模式 Files.writeString(inboxPath, ); // 消费一次消息被读取后清空避免重复处理 // 确保每个消息只被处理一次 return messages; } } }异步通信发送和接收解耦不阻塞发送方文件系统存储简单可靠支持进程间通信结构化消息明确的消息格式支持多种消息类型消费模式读取后清空避免消息重复处理可扩展协议通过msgType支持不同的通信语义智能体管理器TeammateManagerjava// 智能体管理器 - 多智能体生命周期管理 static class TeammateManager { private final Path teamDir; private final Path configPath; private MapString, Object config; private final MapString, Thread threads new ConcurrentHashMap(); private final MapString, AtomicBoolean stopFlags new ConcurrentHashMap(); // 配置管理团队配置持久化到文件 // 线程管理每个智能体在自己的线程中运行 // 停止控制支持优雅停止智能体 public String spawn(String name, String role, String prompt) { MapString, Object member findMember(name); if (member ! null) { String status (String) member.get(status); if (!idle.equals(status) !shutdown.equals(status)) { return String.format(Error: %s is currently %s, name, status); } member.put(status, working); member.put(role, role); } else { member new LinkedHashMap(); member.put(name, name); member.put(role, role); member.put(status, working); ((ListMapString, Object) config.get(members)).add(member); } // 状态管理智能体有明确的状态机 // 重用支持可以重用已有的智能体 // 角色配置为智能体分配特定角色 saveConfig(); // 创建新的停止标志 AtomicBoolean stopFlag new AtomicBoolean(false); stopFlags.put(name, stopFlag); // 创建并启动新线程 Thread thread new Thread(() - teammateLoop(name, role, prompt, stopFlag), Teammate- name); thread.setDaemon(true); threads.put(name, thread); thread.start(); // 独立线程每个智能体在独立线程中运行 // 守护线程不会阻止JVM退出 // 命名线程便于调试和监控 return String.format(Spawned %s (role: %s), name, role); } private void teammateLoop(String name, String role, String prompt, AtomicBoolean stopFlag) { String systemPrompt String.format( You are %s, role: %s, at %s. Use send_message to communicate. Complete your task., name, role, WORKDIR ); // 个性化系统提示为每个智能体定制角色 // 明确角色让智能体知道自己的身份和职责 ListMapString, Object messages new ArrayList(); messages.add(Map.of(role, user, content, prompt)); // 初始化消息从传入的prompt开始 // 最大迭代次数限制 for (int i 0; i 50 !stopFlag.get(); i) { try { // 检查邮箱 ListMapString, Object inbox BUS.readInbox(name); for (MapString, Object msg : inbox) { messages.add(Map.of(role, user, content, gson.toJson(msg))); } // 邮箱检查每次迭代前检查新消息 // 消息注入将收到的消息加入上下文 // 持续通信支持动态的任务调整 // 短暂休眠避免 CPU 过度使用 Thread.sleep(100); // 节能设计避免忙等待 } } // 更新状态 MapString, Object member findMember(name); if (member ! null !shutdown.equals(member.get(status))) { member.put(status, idle); saveConfig(); } // 状态恢复完成后状态恢复为idle // 配置持久化状态变化立即保存 } }生命周期管理智能体的创建、运行、停止、销毁状态持久化智能体状态保存到文件重启可恢复独立执行每个智能体在自己的线程中独立运行通信集成自动检查邮箱支持动态通信优雅停止支持安全的停止机制多智能体通信工具集java// 团队管理工具集 public enum ToolType { SPAWN_TEAMMATE(spawn_teammate, Spawn a persistent teammate that runs in its own thread.), LIST_TEAMMATES(list_teammates, List all teammates with name, role, status.), SEND_MESSAGE(send_message, Send a message to a teammates inbox.), READ_INBOX(read_inbox, Read and drain the leads inbox.), BROADCAST(broadcast, Send a message to all teammates.); // 团队创建动态生成新的智能体 // 状态查询查看所有智能体状态 // 点对点通信向特定智能体发送消息 // 广播通信向所有智能体发送消息 // 邮箱读取获取收到的消息 } // 工具处理器 TOOL_HANDLERS.put(ToolType.SEND_MESSAGE.name, args - { String to (String) args.get(to); String content (String) args.get(content); String msgType (String) args.get(msg_type); if (msgType null) msgType message; return BUS.send(lead, to, content, msgType); // 领导身份所有消息都以lead身份发送 // 灵活消息类型支持不同类型的消息 }); TOOL_HANDLERS.put(ToolType.BROADCAST.name, args - { String content (String) args.get(content); return BUS.broadcast(lead, content, TEAM_MANAGER.memberNames()); // 批量发送向所有团队成员发送消息 // 排除自己广播不包含发送者自己 });完整的通信API提供完整的智能体间通信能力领导-成员模式明确的领导智能体控制整个团队灵活的通信模式支持点对点、广播、邮箱读取与现有系统集成与基础工具无缝集成领导智能体主循环java// Agent 主循环领导智能体 public static void agentLoop(ListMapString, Object messages) { while (true) { try { // 检查领导邮箱 ListMapString, Object inbox BUS.readInbox(lead); if (!inbox.isEmpty()) { String inboxJson gson.toJson(inbox); messages.add(Map.of( role, user, content, inbox inboxJson /inbox )); messages.add(Map.of( role, assistant, content, Noted inbox messages. )); // 自动邮箱检查每次迭代前检查新消息 // 结构化注入用XML标签包裹便于解析 // 对话完整添加assistant响应保持结构 } // 显示团队状态 int activeCount TEAM_MANAGER.getActiveCount(); if (activeCount 0) { System.out.printf([Active teammates: %d]%n, activeCount); } // 状态监控实时显示活跃智能体数量 } } }自动通信领导智能体自动接收和处理消息状态感知实时了解团队状态决策依据基于团队反馈做出更好的决策领导协调领导智能体负责协调整个团队架构演进与价值从 BackgroundTasksSystem 到 AgentTeamsSystem 的升级维度BackgroundTasksSystemAgentTeamsSystem架构模式主从异步任务多智能体协作智能水平被动执行任务主动协作解决通信方式结果通知结构化消息传递角色分工无明确的角色化分工