【从0到1构建一个ClaudeAgent】协作-团队协议
<p>多个 Agent 怎么协商？</p><p>这个在智能体团队系统的基础上，增加了<strong>两种协议</strong>：</p>
<ol>
<li><strong>关机协议（Shutdown Protocol）</strong>：
<ul>
<li>领导智能体请求团队成员关机</li>
<li>团队成员可以批准或拒绝关机请求</li>
<li>基于 request_id 的请求-响应模式</li>
</ul>
</li>
<li><strong>计划审批协议（Plan Approval Protocol）</strong>：
<ul>
<li>团队成员提交计划给领导审批</li>
<li>领导批准或拒绝计划</li>
<li>同样基于 request_id 的请求-响应模式</li>
</ul>
</li>
</ol>
<p><strong>关键洞察</strong>：相同的 request_id 关联模式，应用于两个不同的协作领域。</p>
<h2 id="java代码">Java代码</h2>
<pre><code class="language-java">public class TeamProtocolsSystem {
// --- 配置 ---
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 Set<String> VALID_MSG_TYPES = Set.of(
"message", "broadcast", "shutdown_request",
"shutdown_response", "plan_approval_response"
);
// --- 请求追踪器 ---
private static final Map<String, ShutdownRequest> shutdownRequests = new ConcurrentHashMap<>();
private static final Map<String, PlanRequest> planRequests = new ConcurrentHashMap<>();
private static final Object trackerLock = new Object();
static class ShutdownRequest {
String requestId;
String target;// 目标智能体
String status;// pending, approved, rejected
long timestamp;
}
static class PlanRequest {
String requestId;
String from; // 提交者
String plan; // 计划内容
String status;// pending, approved, rejected
long timestamp;
}
// --- 消息系统（MessageBus）---
static class MessageBus {
private final Path inboxDir;
private final AtomicInteger requestIdCounter = new AtomicInteger(1);
public MessageBus(Path inboxDir) {
this.inboxDir = inboxDir;
try {
Files.createDirectories(inboxDir);
} catch (IOException e) {
throw new RuntimeException("Failed to create inbox directory", e);
}
}
/**
* 生成唯一的请求ID
*/
public String generateRequestId() {
return "req_" + requestIdCounter.getAndIncrement() + "_" +
System.currentTimeMillis() % 10000;
}
/**
* 发送消息到指定智能体
*/
public String send(String sender, String to, String content,
String msgType, Map<String, Object> extra) {
if (!VALID_MSG_TYPES.contains(msgType)) {
return String.format("Error: Invalid type '%s'. Valid: %s",
msgType, String.join(", ", VALID_MSG_TYPES));
}
Map<String, 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);
}
// ... 省略相同的发送逻辑
}
}
// 初始化消息总线
private static final MessageBus BUS = new MessageBus(INBOX_DIR);
// --- 智能体管理器（TeammateManager）---
static class TeammateManager {
// ... 省略相同的配置加载、保存逻辑
private void teammateLoop(String name, String role, String prompt, AtomicBoolean stopFlag) {
String systemPrompt = String.format(
"You are '%s', role: %s, at %s. " +
"Submit plans via plan_approval before major work. " +
"Respond to shutdown_request with shutdown_response.",
name, role, WORKDIR
);
// 增强系统提示：明确协议要求
// 计划审批：重要工作前需要提交计划
// 关机响应：需要响应关机请求
List<Map<String, Object>> messages = new ArrayList<>();
messages.add(Map.of("role", "user", "content", prompt));
boolean shouldExit = false;
// 最大迭代次数限制
for (int i = 0; i < 50 && !stopFlag.get(); i++) {
try {
// 检查邮箱
List<Map<String, Object>> inbox = BUS.readInbox(name);
for (Map<String, Object> msg : inbox) {
messages.add(Map.of("role", "user", "content", gson.toJson(msg)));
}
if (shouldExit) {
break;
}
// ... 省略相同的LLM调用和执行逻辑
for (Map<String, Object> block : content) {
if ("tool_use".equals(block.get("type"))) {
String toolName = (String) block.get("name");
String toolId = (String) block.get("id");
@SuppressWarnings("unchecked")
Map<String, Object> args = (Map<String, Object>) block.get("input");
String output = executeTeammateTool(name, toolName, args);
// 如果批准了关机，设置退出标志
if ("shutdown_response".equals(toolName) &&
Boolean.TRUE.equals(args.get("approve"))) {
shouldExit = true;
}
}
}
}
}
// 更新状态
Map<String, Object> member = findMember(name);
if (member != null) {
member.put("status", shouldExit ? "shutdown" : "idle");
saveConfig();
}
// 状态更新：根据退出原因设置不同状态
}
private String executeTeammateTool(String sender, String toolName, Map<String, Object> args) {
try {
switch (toolName) {
// ... 省略基础工具
case "shutdown_response":
String reqId = (String) args.get("request_id");
Boolean approve = (Boolean) args.get("approve");
if (approve == null) approve = false;
synchronized (trackerLock) {
ShutdownRequest req = shutdownRequests.get(reqId);
if (req != null) {
req.status = approve ? "approved" : "rejected";
}
}
// 状态更新：在追踪器中更新请求状态
String reason = (String) args.get("reason");
if (reason == null) reason = "";
BUS.send(
sender, "lead", reason, "shutdown_response",
Map.of("request_id", reqId, "approve", approve)
);
// 回复领导：通知领导审批结果
return String.format("Shutdown %s", approve ? "approved" : "rejected");
case "plan_approval":
String planText = (String) args.get("plan");
String planReqId = BUS.generateRequestId();
synchronized (trackerLock) {
planRequests.put(planReqId, new PlanRequest(planReqId, sender, planText));
}
// 计划提交：创建新的计划请求
BUS.send(
sender, "lead", planText, "plan_approval_response",
Map.of("request_id", planReqId, "plan", planText)
);
// 通知领导：发送计划审批请求
return String.format("Plan submitted (request_id=%s). Waiting for lead approval.", planReqId);
}
}
}
}
// --- 领导特定的协议处理器 ---
/**
* 处理关机请求
*/
private static String handleShutdownRequest(String teammate) {
String reqId = BUS.generateRequestId();
synchronized (trackerLock) {
shutdownRequests.put(reqId, new ShutdownRequest(reqId, teammate));
}
// 创建请求：在追踪器中记录关机请求
BUS.send(
"lead", teammate, "Please shut down gracefully.",
"shutdown_request", Map.of("request_id", reqId)
);
// 发送请求：向目标智能体发送关机请求
return String.format("Shutdown request %s sent to '%s' (status: pending)", reqId, teammate);
}
/**
* 处理计划审批
*/
private static String handlePlanReview(String requestId, boolean approve, String feedback) {
PlanRequest req;
synchronized (trackerLock) {
req = planRequests.get(requestId);
}
if (req == null) {
return String.format("Error: Unknown plan request_id '%s'", requestId);
}
synchronized (trackerLock) {
req.status = approve ? "approved" : "rejected";
}
// 状态更新：更新计划审批状态
BUS.send(
"lead", req.from, feedback, "plan_approval_response",
Map.of("request_id", requestId, "approve", approve, "feedback", feedback)
);
// 回复提交者：发送审批结果和反馈
return String.format("Plan %s for '%s'", approve ? "approved" : "rejected", req.from);
}
/**
* 检查关机请求状态
*/
private static String checkShutdownStatus(String requestId) {
synchronized (trackerLock) {
ShutdownRequest req = shutdownRequests.get(requestId);
if (req == null) {
return gson.toJson(Map.of("error", "not found"));
}
return gson.toJson(Map.of(
"request_id", req.requestId,
"target", req.target,
"status", req.status,
"timestamp", req.timestamp
));
}
// 状态查询：返回请求的详细信息
}
// --- 工具枚举 ---
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 teammate's inbox."),
READ_INBOX("read_inbox", "Read and drain the lead's inbox."),
BROADCAST("broadcast", "Send a message to all teammates."),
SHUTDOWN_REQUEST("shutdown_request", "Request a teammate to shut down gracefully. Returns a request_id for tracking."),// 新增
SHUTDOWN_RESPONSE("shutdown_response", "Check the status of a shutdown request by request_id."),// 新增
PLAN_APPROVAL("plan_approval", "Approve or reject a teammate's plan. Provide request_id + approve + optional feedback.");// 新增
public final String name;
public final String description;
ToolType(String name, String description) { this.name = name; this.description = description; }
}
// --- 工具处理器映射 ---
private static final Map<String, ToolExecutor> TOOL_HANDLERS = new HashMap<>();
static {
// ... 省略基础工具和团队管理工具注册
// 协议工具
TOOL_HANDLERS.put(ToolType.SHUTDOWN_REQUEST.name, args -> {
String teammate = (String) args.get("teammate");
return handleShutdownRequest(teammate);
});
TOOL_HANDLERS.put(ToolType.SHUTDOWN_RESPONSE.name, args -> {
String requestId = (String) args.get("request_id");
return checkShutdownStatus(requestId);
});
TOOL_HANDLERS.put(ToolType.PLAN_APPROVAL.name, args -> {
String requestId = (String) args.get("request_id");
Boolean approve = (Boolean) args.get("approve");
String feedback = (String) args.get("feedback");
if (feedback == null) feedback = "";
return handlePlanReview(requestId, approve, feedback);
});
}
// --- Agent 主循环（领导智能体）---
public static void agentLoop(List<Map<String, Object>> messages) {
while (true) {
try {
// ... 省略邮箱检查逻辑
// 显示待处理的请求
synchronized (trackerLock) {
long pendingShutdowns = shutdownRequests.values().stream()
.filter(r -> "pending".equals(r.status))
.count();
long pendingPlans = planRequests.values().stream()
.filter(r -> "pending".equals(r.status))
.count();
if (pendingShutdowns > 0 || pendingPlans > 0) {
System.out.printf("%n",
pendingShutdowns, pendingPlans);
}
}
// 状态监控：实时显示待处理的协议请求
// ... 省略相同的 LLM 调用和执行逻辑
}
}
}
}
</code></pre>
<h2 id="协议化团队协作系统">协议化团队协作系统</h2>
<p><strong>核心思想</strong>：从松散的智能体协作升级为<strong>结构化、协议化、有状态的团队协作系统</strong>，引入<strong>正式的请求-响应协议、状态追踪、审批流程</strong>，实现<strong>企业级团队管理的标准化和可控性</strong>。</p>
<pre><code class="language-java">// 请求追踪器 - 协议状态管理
private static final Map<String, ShutdownRequest> shutdownRequests = new ConcurrentHashMap<>();
private static final Map<String, PlanRequest> planRequests = new ConcurrentHashMap<>();
private static final Object trackerLock = new Object();
// 全局状态：追踪所有协议请求的状态
// 并发安全：支持多线程安全访问
// 生命周期：从创建、处理到完成的全周期追踪
</code></pre>
<ul>
<li><strong>状态驱动</strong>：所有协议请求都有明确的状态生命周期</li>
<li><strong>集中管理</strong>：全局追踪所有请求状态</li>
<li><strong>并发安全</strong>：支持多智能体并发访问</li>
<li><strong>可追溯性</strong>：所有请求都有时间戳和唯一标识</li>
</ul>
<h2 id="正式的协议类型定义">正式的协议类型定义</h2>
<pre><code class="language-java">// 关机请求协议
static class ShutdownRequest {
String requestId; // 唯一请求ID
String target; // 目标智能体
String status; // 状态：pending, approved, rejected
long timestamp; // 创建时间戳
// 结构化请求：明确的目标和状态
// 唯一标识：支持多个并发请求
// 时间追踪：支持超时和时序分析
}
// 计划审批协议
static class PlanRequest {
String requestId; // 唯一请求ID
String from; // 提交者
String plan; // 计划内容
String status; // 状态：pending, approved, rejected
long timestamp; // 创建时间戳
// 计划提交流程：从提交到审批的完整生命周期
// 内容关联：保存计划的具体内容
// 来源追踪：明确计划提交者
}
</code></pre>
<ul>
<li><strong>协议实体</strong>：每个协议类型都有明确的数据结构</li>
<li><strong>状态机</strong>：明确的pending → approved/rejected状态转换</li>
<li><strong>审计跟踪</strong>：时间戳支持审计和时序分析</li>
<li><strong>关系管理</strong>：明确的请求-响应关系</li>
</ul>
<h2 id="协议工具生态系统">协议工具生态系统</h2>
<pre><code class="language-java">// 协议工具集
public enum ToolType {
SHUTDOWN_REQUEST("shutdown_request", "Request a teammate to shut down gracefully. Returns a request_id for tracking."),
SHUTDOWN_RESPONSE("shutdown_response", "Check the status of a shutdown request by request_id."),
PLAN_APPROVAL("plan_approval", "Approve or reject a teammate's plan. Provide request_id + approve + optional feedback.");
// 正式协议：明确的工具命名和描述
// 请求-响应模式：领导发起请求，成员响应
// 状态查询：支持请求状态检查
// 反馈机制：支持审批反馈
}
</code></pre>
<ul>
<li><strong>标准化接口</strong>：所有协议都有标准化的工具接口</li>
<li><strong>完整工作流</strong>：创建、查询、处理的完整工作流</li>
<li><strong>可追踪性</strong>：通过request_id追踪所有请求</li>
<li><strong>反馈机制</strong>：支持审批反馈，促进改进</li>
</ul>
<h2 id="关机协议实现">关机协议实现</h2>
<pre><code class="language-java">/**
* 处理关机请求
*/
private static String handleShutdownRequest(String teammate) {
String reqId = BUS.generateRequestId();// 生成唯一请求ID
synchronized (trackerLock) {
shutdownRequests.put(reqId, new ShutdownRequest(reqId, teammate));
}
// 状态记录：创建新的关机请求，状态为pending
BUS.send(
"lead", teammate, "Please shut down gracefully.",
"shutdown_request", Map.of("request_id", reqId)
);
// 消息发送：向目标智能体发送正式的关机请求
// 包含请求ID：便于目标智能体识别和响应
return String.format("Shutdown request %s sent to '%s' (status: pending)", reqId, teammate);
// 确认反馈：向领导智能体返回请求详情
}
</code></pre>
<pre><code class="language-java">// 团队成员响应关机请求
case "shutdown_response":
String reqId = (String) args.get("request_id");
Boolean approve = (Boolean) args.get("approve");
if (approve == null) approve = false;
synchronized (trackerLock) {
ShutdownRequest req = shutdownRequests.get(reqId);
if (req != null) {
req.status = approve ? "approved" : "rejected";
}
}
// 状态更新：根据响应更新请求状态
String reason = (String) args.get("reason");
if (reason == null) reason = "";
BUS.send(
sender, "lead", reason, "shutdown_response",
Map.of("request_id", reqId, "approve", approve)
);
// 回复领导：发送正式的关机响应
// 包含原因：提供批准/拒绝的理由
return String.format("Shutdown %s", approve ? "approved" : "rejected");
</code></pre>
<ul>
<li><strong>正式流程</strong>：领导发起请求 → 成员响应 → 领导接收确认</li>
<li><strong>状态同步</strong>：所有参与方都看到一致的状态</li>
<li><strong>可选理由</strong>：支持提供批准/拒绝的理由</li>
<li><strong>优雅关闭</strong>：成员可以完成当前工作后优雅关闭</li>
</ul>
<h2 id="计划审批协议实现">计划审批协议实现</h2>
<pre><code class="language-java">// 团队成员提交计划审批
case "plan_approval":
String planText = (String) args.get("plan");
String planReqId = BUS.generateRequestId();
synchronized (trackerLock) {
planRequests.put(planReqId, new PlanRequest(planReqId, sender, planText));
}
// 创建请求：记录计划审批请求
BUS.send(
sender, "lead", planText, "plan_approval_response",
Map.of("request_id", planReqId, "plan", planText)
);
// 提交领导：发送计划审批请求
return String.format("Plan submitted (request_id=%s). Waiting for lead approval.", planReqId);
// 确认反馈：告知成员请求已提交，等待审批
</code></pre>
<pre><code class="language-java">/**
* 处理计划审批
*/
private static String handlePlanReview(String requestId, boolean approve, String feedback) {
PlanRequest req;
synchronized (trackerLock) {
req = planRequests.get(requestId);
}
if (req == null) {
return String.format("Error: Unknown plan request_id '%s'", requestId);
}
// 验证请求：确保请求存在
synchronized (trackerLock) {
req.status = approve ? "approved" : "rejected";
}
// 状态更新：根据审批结果更新状态
BUS.send(
"lead", req.from, feedback, "plan_approval_response",
Map.of("request_id", requestId, "approve", approve, "feedback", feedback)
);
// 通知提交者：发送审批结果和反馈
// 包含反馈：提供改进建议或表扬
return String.format("Plan %s for '%s'", approve ? "approved" : "rejected", req.from);
}
</code></pre>
<ul>
<li><strong>质量保证</strong>：重要工作前需要计划审批，确保质量</li>
<li><strong>双向沟通</strong>：领导可以提出改进建议</li>
<li><strong>透明流程</strong>：明确的提交→审批→通知流程</li>
<li><strong>知识传递</strong>：通过反馈传递最佳实践</li>
</ul>
<h2 id="团队成员协议集成">团队成员协议集成</h2>
<pre><code class="language-java">// 团队成员系统提示增强
String systemPrompt = String.format(
"You are '%s', role: %s, at %s. " +
"Submit plans via plan_approval before major work. " +
"Respond to shutdown_request with shutdown_response.",
name, role, WORKDIR
);
// 协议内化：在系统提示中明确协议要求
// 行为规范：指导团队成员正确使用协议
// 责任明确：明确团队成员在协议中的责任
</code></pre>
<pre><code class="language-java">// 团队成员邮箱处理增强
for (Map<String, Object> msg : inbox) {
messages.add(Map.of("role", "user", "content", gson.toJson(msg)));
}
// 协议消息：将所有消息（包括协议消息）加入上下文
// 自动处理：团队成员自动处理收到的协议消息
</code></pre>
<pre><code class="language-java">// 团队成员退出控制
if ("shutdown_response".equals(toolName) &&
Boolean.TRUE.equals(args.get("approve"))) {
shouldExit = true;
}
// 协议驱动：当批准关机时，设置退出标志
// 优雅退出：完成当前工作后正常退出
// 状态清理：退出前清理资源
</code></pre>
<ul>
<li><strong>协议意识</strong>：团队成员知道协议要求和责任</li>
<li><strong>自动处理</strong>：自动处理收到的协议消息</li>
<li><strong>协议驱动</strong>：协议可以控制团队成员行为</li>
<li><strong>优雅管理</strong>：支持优雅的启动、运行、停止</li>
</ul>
<h2 id="领导智能体协议管理">领导智能体协议管理</h2>
<pre><code class="language-java">// 领导智能体状态监控
synchronized (trackerLock) {
long pendingShutdowns = shutdownRequests.values().stream()
.filter(r -> "pending".equals(r.status))
.count();
long pendingPlans = planRequests.values().stream()
.filter(r -> "pending".equals(r.status))
.count();
if (pendingShutdowns > 0 || pendingPlans > 0) {
System.out.printf("%n",
pendingShutdowns, pendingPlans);
}
}
// 实时监控：领导实时了解待处理的协议请求
// 优先级管理：可以基于pending请求数量调整工作优先级
// 资源分配：可以基于请求状态分配资源
</code></pre>
<pre><code class="language-java">// 领导智能体系统提示
private static final String SYSTEM_PROMPT = String.format(
"You are a team lead at %s. Manage teammates with shutdown and plan approval protocols.",
WORKDIR
);
// 领导责任：在系统提示中明确领导的管理责任
// 协议管理：强调使用协议进行团队管理
// 决策支持：为领导提供决策依据
</code></pre>
<ul>
<li><strong>领导责任</strong>：明确领导在协议中的管理责任</li>
<li><strong>全局视图</strong>：领导有团队的全局视图</li>
<li><strong>主动管理</strong>：可以主动发起和管理协议</li>
<li><strong>决策优化</strong>：基于协议状态优化决策</li>
</ul>
<h2 id="请求状态查询">请求状态查询</h2>
<pre><code class="language-java">/**
* 检查关机请求状态
*/
private static String checkShutdownStatus(String requestId) {
synchronized (trackerLock) {
ShutdownRequest req = shutdownRequests.get(requestId);
if (req == null) {
return gson.toJson(Map.of("error", "not found"));
}
return gson.toJson(Map.of(
"request_id", req.requestId,
"target", req.target,
"status", req.status,
"timestamp", req.timestamp
));
}
// 状态查询：领导可以查询任意请求的状态
// 完整信息：返回请求的所有详细信息
// JSON格式：便于程序化处理
}
</code></pre>
<ul>
<li><strong>透明管理</strong>：所有请求状态都可以查询</li>
<li><strong>程序化接口</strong>：JSON格式支持程序化处理</li>
<li><strong>调试支持</strong>：便于调试和问题定位</li>
<li><strong>历史查询</strong>：可以查询历史请求状态</li>
</ul>
<h2 id="架构演进与价值">架构演进与价值</h2>
<p><strong>从 AgentTeamsSystem 到 TeamProtocolsSystem 的升级</strong>：</p>
<table>
<thead>
<tr>
<th>维度</th>
<th>AgentTeamsSystem</th>
<th>TeamProtocolsSystem</th>
</tr>
</thead>
<tbody>
<tr>
<td>协作模式</td>
<td>松耦合消息传递</td>
<td>结构化协议协作</td>
</tr>
<tr>
<td>状态管理</td>
<td>简单状态跟踪</td>
<td>完整状态生命周期</td>
</tr>
<tr>
<td>质量控制</td>
<td>无</td>
<td>计划审批流程</td>
</tr>
<tr>
<td>团队管理</td>
<td>自由运行</td>
<td>可控优雅管理</td>
</tr>
<tr>
<td>决策支持</td>
<td>有限</td>
<td>丰富协议数据</td>
</tr>
</tbody>
</table>
</div>
<div id="MySignature" role="contentinfo">
<p>本文来自在线网站：seven的菜鸟成长之路，作者：seven，转载请注明原文链接：www.seven97.top</p><br><br>
来源：https://www.cnblogs.com/sevencoding/p/19821589
頁:
[1]