【从0到1构建一个ClaudeAgent】协作-自主Agent
·
- 工作-空闲循环：智能体在工作完成时自动进入空闲状态
- 任务自动认领：空闲时扫描任务板，自动认领无人认领的任务
- 身份重新注入：在上下文压缩后重新注入智能体身份信息
- 自动资源管理：空闲超时自动关机，释放资源
关键洞察：扫描看板，认领任务。队友自己扫描任务板并认领任务，无需主 Agent 逐个分配。
Java实现代码
java
public class AutonomousAgentsSystem {
// --- 新增配置 ---
private static final Path TASKS_DIR = WORKDIR.resolve(".tasks"); // 任务存储目录
private static final int POLL_INTERVAL = 5; // 空闲轮询间隔（秒）
private static final int IDLE_TIMEOUT = 60; // 空闲超时（秒）
private static final Object claimLock = new Object(); // 任务认领锁
// --- 新增：任务板扫描 ---
/**
* 扫描无人认领的任务
*/
private static List<Map<String, Object>> scanUnclaimedTasks() {
try {
Files.createDirectories(TASKS_DIR);
List<Map<String, Object>> unclaimed = new ArrayList<>();
Files.list(TASKS_DIR)
.filter(p -> p.getFileName().toString().startsWith("task_"))
.filter(p -> p.getFileName().toString().endsWith(".json"))
.sorted()
.forEach(p -> {
try {
String content = Files.readString(p);
Type type = new TypeToken<Map<String, Object>>(){}.getType();
Map<String, Object> task = gson.fromJson(content, type);
// 检查是否可认领
String status = (String) task.get("status");
String owner = (String) task.get("owner");
@SuppressWarnings("unchecked")
List<Integer> blockedBy = (List<Integer>) task.get("blockedBy");
if ("pending".equals(status) &&
(owner == null || owner.isEmpty()) &&
(blockedBy == null || blockedBy.isEmpty())) {
unclaimed.add(task);
}
// 可认领条件：状态为pending、无所有者、无阻塞依赖
// 有序扫描：按文件名排序，确保公平性
} catch (IOException e) {
// 忽略读取错误
}
});
return unclaimed;
} catch (IOException e) {
return new ArrayList<>();
}
}
/**
* 认领任务
*/
private static String claimTask(int taskId, String owner) {
synchronized (claimLock) { // 同步锁防止并发认领
Path taskPath = TASKS_DIR.resolve("task_" + taskId + ".json");
if (!Files.exists(taskPath)) {
return "Error: Task " + taskId + " not found";
}
try {
String content = Files.readString(taskPath);
Type type = new TypeToken<Map<String, Object>>(){}.getType();
Map<String, Object> task = gson.fromJson(content, type);
// 更新任务信息
task.put("owner", owner);
task.put("status", "in_progress");
// 原子性更新：确保任务不会被多个智能体同时认领
// 状态转换：pending → in_progress
Files.writeString(taskPath, gson.toJson(task));
return String.format("Claimed task #%d for %s", taskId, owner);
} catch (IOException e) {
return "Error: " + e.getMessage();
}
}
}
// --- 新增：身份重新注入 ---
/**
* 创建身份块（用于上下文压缩后重新注入身份）
*/
private static Map<String, Object> makeIdentityBlock(String name, String role, String teamName) {
return Map.of(
"role", "user",
"content", String.format(
"<identity>You are '%s', role: %s, team: %s. Continue your work.</identity>",
name, role, teamName
)
);
// 身份持久化：即使上下文被压缩，也能重新注入身份
// 结构化标记：<identity>标签明确标识身份信息
}
// --- 新增：团队成员工具 ---
private String executeTeammateTool(String sender, String toolName, Map<String, Object> args) {
try {
switch (toolName) {
// ... 原有的工具处理 ...
case "idle":
// idle 工具在工作循环中特殊处理
return "Entering idle phase. Will poll for new tasks.";
// 主动空闲：智能体主动表示完成当前工作
case "claim_task":
int taskId = ((Number) args.get("task_id")).intValue();
return claimTask(taskId, sender);
// 主动认领：智能体主动认领指定任务
}
} catch (Exception e) {
return "Error: " + e.getMessage();
}
}
// --- 新增：团队成员工具定义 ---
private List<Map<String, Object>> getTeammateTools() {
List<Map<String, Object>> tools = new ArrayList<>();
// 新增 idle 工具
tools.add(createToolSpec("idle",
"Signal that you have no more work. Enters idle polling phase.",
Map.of(),
List.of()));
// 无参数工具：简单的状态转换工具
// 新增 claim_task 工具
tools.add(createToolSpec("claim_task",
"Claim a task from the task board by ID.",
Map.of("task_id", Map.of("type", "integer")),
List.of("task_id")));
// 主动工作：智能体可以主动选择任务
return tools;
}
// --- 新增：团队成员主循环（工作-空闲循环）---
private void teammateLoop(String name, String role, String prompt, AtomicBoolean stopFlag) {
@SuppressWarnings("unchecked")
String teamName = (String) TEAM_MANAGER.config.get("team_name");
String systemPrompt = String.format(
"You are '%s', role: %s, team: %s, at %s. " +
"Use idle tool when you have no more work. You will auto-claim new tasks.",
name, role, teamName, WORKDIR
);
// 增强系统提示：包含空闲行为和自动认领机制
List<Map<String, Object>> messages = new ArrayList<>();
messages.add(Map.of("role", "user", "content", prompt));
while (!stopFlag.get()) { // 外层循环支持工作-空闲循环
// --- 工作阶段 ---
TEAM_MANAGER.setStatus(name, "working");
boolean completedWorkPhase = false;
for (int i = 0; i < 50 && !stopFlag.get(); i++) {
try {
// ... 检查邮箱、调用LLM、执行工具等逻辑 ...
boolean idleRequested = false;
for (Map<String, Object> block : content) {
if ("tool_use".equals(block.get("type"))) {
String toolName = (String) block.get("name");
if ("idle".equals(toolName)) {
idleRequested = true; // 标记空闲请求
break;
}
}
}
if (idleRequested) {
completedWorkPhase = true;
break; // 退出工作阶段，进入空闲阶段
}
} catch (Exception e) {
System.err.printf("[%s] Error: %s%n", name, e.getMessage());
break;
}
}
if (!completedWorkPhase) {
// 没有明确进入空闲，可能是因为错误
TEAM_MANAGER.setStatus(name, "idle");
return;
}
// --- 空闲阶段 ---
TEAM_MANAGER.setStatus(name, "idle");
boolean resume = false;
int maxPolls = IDLE_TIMEOUT / Math.max(POLL_INTERVAL, 1);
for (int pollCount = 0; pollCount < maxPolls && !stopFlag.get(); pollCount++) {
try {
Thread.sleep(POLL_INTERVAL * 1000L); // 周期性轮询
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
// 检查邮箱
List<Map<String, Object>> inbox = BUS.readInbox(name);
if (!inbox.isEmpty()) {
for (Map<String, Object> msg : inbox) {
if ("shutdown_request".equals(msg.get("type"))) {
TEAM_MANAGER.setStatus(name, "shutdown");
return;
}
messages.add(Map.of("role", "user", "content", gson.toJson(msg)));
}
resume = true; // 有消息，恢复工作
break;
}
// 扫描无人认领的任务
List<Map<String, Object>> unclaimed = scanUnclaimedTasks();
if (!unclaimed.isEmpty()) {
Map<String, Object> task = unclaimed.get(0);
int taskId = ((Number) task.get("id")).intValue();
// 自动认领任务
claimTask(taskId, name);
// 创建任务提示
String taskSubject = (String) task.get("subject");
String taskDesc = (String) task.get("description");
if (taskDesc == null) taskDesc = "";
String taskPrompt = String.format(
"<auto-claimed>Task #%d: %s%n%s</auto-claimed>",
taskId, taskSubject, taskDesc
);
// 如果消息历史太短，重新注入身份
if (messages.size() <= 3) {
messages.add(0, makeIdentityBlock(name, role, teamName));
messages.add(1, Map.of("role", "assistant",
"content", "I am " + name + ". Continuing."));
// 身份恢复：确保智能体知道自己的身份
}
messages.add(Map.of("role", "user", "content", taskPrompt));
messages.add(Map.of("role", "assistant",
"content", "Claimed task #" + taskId + ". Working on it."));
// 自动对话：模拟智能体认领任务并开始工作
resume = true; // 有任务，恢复工作
break;
}
}
if (!resume) {
// 空闲超时，自动关机
TEAM_MANAGER.setStatus(name, "shutdown");
return;
}
// 重新进入工作阶段
}
}
// --- 新增：主程序的特殊命令 ---
public static void main(String[] args) {
// ... 初始化代码 ...
while (true) {
System.out.print("\n\033[36ms11 >> \033[0m");
String userInput = scanner.nextLine().trim();
if (userInput.isEmpty() || "exit".equalsIgnoreCase(userInput) ||
"q".equalsIgnoreCase(userInput)) {
break;
}
// 特殊命令
if ("/team".equals(userInput)) {
System.out.println("\n" + TEAM_MANAGER.listAll());
continue;
}
if ("/inbox".equals(userInput)) {
List<Map<String, Object>> inbox = BUS.readInbox("lead");
if (inbox.isEmpty()) {
System.out.println("\nLeader's inbox is empty.");
} else {
System.out.println("\nLeader's inbox:");
System.out.println(gson.toJson(inbox));
}
continue;
}
if ("/tasks".equals(userInput)) {
try {
Files.createDirectories(TASKS_DIR);
Files.list(TASKS_DIR)
.filter(p -> p.getFileName().toString().startsWith("task_"))
.filter(p -> p.getFileName().toString().endsWith(".json"))
.sorted()
.forEach(p -> {
try {
String content = Files.readString(p);
Type type = new TypeToken<Map<String, Object>>(){}.getType();
Map<String, Object> task = gson.fromJson(content, type);
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
5
0- 0
已为社区贡献4条内容
所有评论(0)