Agent 团队协作从 “异步通信” 到 “结构化协议”作为编程学习的视角我们先把这两季的核心逻辑串起来第九季解决了 “团队能沟通” 的问题基础通信层第十季解决了 “团队沟通有规则” 的问题协议层就像先给团队装了邮箱再制定了 “请假审批” “方案评审” 的公司制度。一、核心背景为什么要做这两季的升级在第九季之前第八季及更早AI代理存在两个致命问题一次性死亡子 Agent 用完就销毁没有 “身份” 和 “记忆” 没法持续协作无规则沟通就算能通信也只是零散消息没有 “请求-响应” 的规范比如关线程时容易数据丢失、改代码时没有审批流程。第九季先搭建 “能持续沟通的团队”第十季再给沟通加“规则和流程”一步步接近真实的人类团队协作。二、第九季搭建 “有身份、能异步通信的持久 Agent 团队”核心目标让AI Agent从 “一次性工具” 变成 “持久存在的队友” 支持跨 Agent 的异步消息传递比如A发消息给BB忙完再看。关键组件拆解带代码解释1. 团队管理核心TeammateManager队友生命周期管理作用创建、跟踪队友Agent给每个队友Agent分配独立线程运行保证同时工作。classTeammateManager:def__init__(self,team_dir:Path):self.dirteam_dir# 团队数据目录存配置、邮箱self.dir.mkdir(exist_okTrue)# 目录不存在则创建self.config_pathself.dir/config.json# 团队配置文件存队友名单、状态self.configself._load_config()# 加载配置比如{members: [{name:alice,role:coder,status:working}]}self.threads{}# 存储每个队友Agent的运行线程# 查找对应 Agent 的信息def_find_member(self,name:str)-dict:forminself.config[members]:ifm[name]name:returnmreturnNonedef_save_config(self,configNone):保存团队配置到文件configconfigorself.config self.config_path.write_text(json.dumps(config,indent2))defspawn(self,name:str,role:str,prompt:str)-str:# 1. 记录队友信息存储到配置文件持久化不会丢member{name:name,role:role,status:working}self.config[members].append(member)self._save_config()# 写入config.json就算程序重启也能找到队友# 2. 启动独立线程运行队友持久存在不是用完就扔threadthreading.Thread(targetself._teammate_loop,# 队友的核心工作循环args(name,role,prompt),daemonTrue# 守护线程主程序退出时自动结束)thread.start()self.threads[name]thread# 跟踪线程状态returnfSpawned teammate {name} (role:{role})问题一从 Agent 的创建上是如何实现异步通信的这和 Agent 的生命周期相关。具体可分为1Agent 的持久化、2线程创建跟管理 Agent 工作逻辑相关、3线程的管理第一部分 - 持久化通过对每个Agent创建json文件的方式持久化Agent的信息包含Agent的名称、角色、状态第二部分 - 创建线程给每个Agent开独立线程让他们一直活着为什么要开独立线程呢是为了让每个Agent的工作循环互不干扰比如 A 队友工作时B 队友可以摸鱼实现异步通信功能第三部分 - 管理线程通过字典的形式存储所有的Agent线程方便管理比如1查询线程是否存活2线程状态是否空闲3防止重复创建线程问题二Agent 的工作逻辑是如何实现异步通信的呢如以下代码所示def_teammate_loop(self,name:str,role:str,prompt:str): 单个队友的核心工作循环独立线程中运行 核心功能 1. 初始化队友的系统提示和对话上下文 2. 循环读取消息队列inbox接收其他队友/管理器的消息 3. 调用大模型LLM生成响应支持工具调用 4. 执行工具调用并返回结果给大模型 5. 循环最多50次或直到停止非工具调用 6. 循环结束后更新队友状态为idle并保存配置 参数 name: str - 队友名称唯一标识 role: str - 队友角色如后端开发、测试工程师 prompt: str - 初始任务提示队友的核心任务 # 1. 构建系统提示词定义队友的身份、工作目录、行为规则# WORKDIR是全局变量代表队友的工作目录send_message是预设的通信方式sys_prompt(fYou are {name}, role:{role}, at{WORKDIR}. # 身份与环境fUse send_message to communicate. Complete your task.# 行为规则用send_message通信完成任务)# 2. 初始化对话消息列表初始只有用户管理器下发的任务提示# messages是大模型的对话上下文格式遵循OpenAI API规范messages[{role:user,content:prompt}]# 3. 获取该队友可使用的工具列表如文件操作、命令执行、消息发送等# _teammate_tools()是管理器的私有方法返回该队友支持的工具定义符合OpenAI工具调用规范toolsself._teammate_tools()# 4. 核心工作循环最多执行50轮避免无限循环for_inrange(50):# 4.1 读取当前队友的收件箱消息从全局消息总线BUS中读取# BUS是全局消息总线read_inbox(name)获取发给该队友的所有未读消息inboxBUS.read_inbox(name)# 4.2 将收件箱的新消息添加到对话上下文转JSON保证格式统一formsgininbox:messages.append({role:user,content:json.dumps(msg)})try:# 4.3 调用大模型API生成响应# modelMODEL全局变量指定使用的大模型如gpt-4o# systemsys_prompt设置大模型的系统角色# toolstools传入该队友可调用的工具列表# max_tokens8000限制单次响应的最大token数responseclient.messages.create(modelMODEL,systemsys_prompt,messagesmessages,toolstools,max_tokens8000,)exceptException:# 捕获API调用异常如网络、额度、模型服务问题直接终止循环break# 4.4 将大模型的响应添加到对话上下文维护多轮对话状态messages.append({role:assistant,content:response.content})# 4.5 判断是否停止循环如果响应不是因为工具调用停止则结束工作# stop_reasontool_use表示大模型要求调用工具需要继续循环否则任务完成ifresponse.stop_reason!tool_use:break# 4.6 处理工具调用遍历响应中的所有工具调用块results[]forblockinresponse.content:# 筛选出类型为tool_use的响应块大模型要求执行的工具ifblock.typetool_use:# 执行工具调用_exec是管理器的私有方法封装工具执行逻辑# 参数调用者名称、工具名称、工具入参outputself._exec(name,block.name,block.input)# 打印工具执行结果截断到120字符避免输出过长print(f [{name}]{block.name}:{str(output)[:120]})# 整理工具执行结果符合OpenAI工具结果格式用于返回给大模型results.append({type:tool_result,tool_use_id:block.id,# 关联对应的工具调用IDcontent:str(output),# 工具执行的结果内容})# 4.7 将工具执行结果作为用户消息添加到上下文供下一轮大模型处理messages.append({role:user,content:results})# 5. 循环结束后更新队友状态并保存配置# 5.1 查找该队友的信息从管理器的成员列表中memberself._find_member(name)# 5.2 如果队友存在且未被关闭shutdown将状态改为idle闲置ifmemberandmember[status]!shutdown:member[status]idle# 5.3 保存配置到文件如config.json持久化状态变更self._save_config()异步功能实现的逻辑如下读取消息队列中的内容读取类似异步邮箱中的内容并将其封装到messages列表中第一次回答大模型回答解决邮箱中的问题两种情况:1无需调用工具说明一次QA就解决了2需要调用工具则和原来的Agent Loop一样的调用所需工具调用工具得到 tool response然后封装到messages列表中Agent Loop循环直至无需使用工具说明任务解决更新线程状态更新线程状态2. 通信核心MessageBus基于JSONL的异步邮箱作用实现队友间的异步消息传递就像公司的邮箱系统——A发消息给BB忙完再看消息存在文件里不会丢。classMessageBus:defsend(self,sender,to,content,msg_typemessage,extraNone):# 构造标准化消息谁发的、发给谁、内容、时间、类型msg{type:msg_type,# 消息类型普通消息/关机请求/计划审批from:sender,content:content,timestamp:time.time()# 时间戳保证顺序}ifextra:msg.update(extra)# 附加字段比如request_id# 追加写入收件人的JSONL文件append-only只加不删保证不丢消息withopen(self.dir/f{to}.jsonl,a)asf:f.write(json.dumps(msg)\n)defread_inbox(self,name):pathself.dir/f{name}.jsonlifnotpath.exists():return[]# 读取所有消息并解析msgs[json.loads(l)forlinpath.read_text().strip().splitlines()ifl]path.write_text()# 读完清空drain避免重复处理returnjson.dumps(msgs,indent2)MessageBus 的目的持久化 Agent 的异步邮箱每个Agent有自己的专属邮箱文件比如 alice.jsonl、bob.jsonl异步操作实现每个Agent发消息、读消息的功能消息存储到json文件中3. 队友的工作循环_teammate_loop作用让队友持续运行定期查邮箱和LLM交互实现“持久工作”。def_teammate_loop(self,name,role,prompt):# 初始化对话上下文messages[{role:user,content:prompt}]# 循环运行最多50轮避免无限循环for_inrange(50):# 第一步查邮箱有消息就注入上下文inboxBUS.read_inbox(name)ifinbox![]:messages.append({role:user,content:finbox{inbox}/inbox})messages.append({role:assistant,content:Noted inbox messages.})# 第二步调用LLM获取响应responseclient.messages.create(...)# 第三步如果LLM不调用工具了就停止本轮工作ifresponse.stop_reason!tool_use:break# 第四步执行工具比如发消息、写代码把结果追加到上下文# ...工具执行逻辑# 第五步标记队友状态为“空闲”self._find_member(name)[status]idle关键理解队友不是“一次性的”而是在循环里持续运行直到完成任务或达到循环上限每次和LLM对话前先查邮箱保证队友能及时收到其他成员的消息状态从“working”变为“idle”实现生命周期管理。第九季核心亮点特性解决的问题实现方式持久化队友Agent 用完就丢无身份线程config.json持久化队友信息异步通信Agent 间无法传递消息JSONL邮箱send/read_inbox生命周期管理Agent 只有“运行/死亡”两种状态working/idle状态切换三、第十季给 Agent 团队沟通加“结构化协议”核心目标第九季的队友能通信但沟通是“无规则的”——比如领导让队友关机队友可能直接退出导致文件没写完队友改代码没有审批流程直接动手。第十季通过“结构化协议”解决这些问题。 s10_team_protocols.py - 团队协议 关机协议和计划审批协议均使用相同的request_id关联模式。基于s09的团队消息功能扩展。 关机状态机FSM待处理pending→ 批准approved | 拒绝rejected 领导 队友 --------------------- --------------------- | 关机请求 | | | | { | ------- | 接收请求 | | request_id: abc | | 决策是否批准 | | } | | | --------------------- --------------------- | --------------------- -------v------------- | 关机响应 | ------- | 关机响应 | | { | | { | | request_id: abc | | request_id: abc | | approve: true | | approve: true | | } | | } | --------------------- --------------------- | v 状态 - 关机完成线程停止 计划审批状态机FSM待处理pending→ 批准approved | 拒绝rejected 队友 领导 --------------------- --------------------- | 计划审批 | | | | 提交{plan:...}| ------- | 审核计划文本 | --------------------- | 批准/拒绝 | --------------------- | --------------------- -------v------------- | 计划审批响应 | ------- | 计划审批 | | {approve: true} | | 审核{req_id, | --------------------- | approve: true} | --------------------- 跟踪器{request_id: {目标|发起方: 名称, 状态: 待处理|已批准|已拒绝}} 核心思路相同的request_id关联模式适配两个业务场景。 核心概念请求-响应协议 FSM有限状态机1. 核心思路所有重要沟通都遵循“请求带唯一ID→ 响应引用同一ID→ 状态更新”的流程用FSM有限状态机管理请求的状态pending→approved/rejected。2. 协议1关机协议Shutdown Protocol解决“强制关机导致数据丢失”的问题要求“领导发关机请求→队友确认批准/拒绝→再关机”。步骤1领导发送关机请求生成唯一request_idshutdown_requests{}# 跟踪所有关机请求的状态defhandle_shutdown_request(teammate:str)-str:# 生成8位唯一请求ID避免冲突req_idstr(uuid.uuid4())[:8]# 记录请求状态待处理shutdown_requests[req_id]{target:teammate,status:pending}# 发消息给目标队友消息类型是shutdown_request带request_idBUS.send(lead,# 发送者teammate,# 接收者Please shut down gracefully.,# 内容shutdown_request,# 消息类型{request_id:req_id}# 附加请求ID)returnfShutdown request{req_id}sent (status: pending)步骤2队友处理关机请求并响应# 队友收到shutdown_request后调用shutdown_response工具iftool_nameshutdown_response:req_idargs[request_id]# 引用领导的请求IDapproveargs[approve]# 批准/拒绝关机# 更新请求状态shutdown_requests[req_id][status]approvedifapproveelserejected# 回复领导同样带request_id保证关联BUS.send(sender,# 队友自己lead,# 接收者是领导args.get(reason,),# 拒绝理由可选shutdown_response,# 消息类型{request_id:req_id,approve:approve})3. 协议2计划审批协议Plan Approval Protocol解决“高风险任务无审批”的问题要求“队友提交计划带ID→领导审批引用ID→队友执行/修改”。plan_requests{}# 跟踪所有计划请求的状态# 队友提交计划defsubmit_plan(sender,plan_content):req_idstr(uuid.uuid4())[:8]plan_requests[req_id]{from:sender,plan:plan_content,status:pending}# 发消息给领导请求审批BUS.send(sender,lead,plan_content,plan_request,{request_id:req_id})returnfPlan submitted (ID:{req_id}), waiting for approval.# 领导审批计划defhandle_plan_review(request_id,approve,feedback):reqplan_requests[request_id]# 通过ID找到对应的计划req[status]approvedifapproveelserejected# 回复队友审批结果BUS.send(lead,req[from],feedback,# 审批意见比如“代码需要加注释”plan_approval_response,{request_id:request_id,approve:approve})4. 统一的FSM有限状态机两个协议共享同一个状态机逻辑简化代码[pending待处理] ↙ ↘ [approved批准] [rejected拒绝]不管是关机请求还是计划请求都只需要维护这三种状态逻辑统一容易扩展。第十季核心亮点特性解决的问题实现方式结构化协议沟通无规则易出错请求-响应模式唯一request_id状态跟踪不知道请求的处理结果shutdown_requests/plan_requests字典优雅关机强制关机导致数据丢失关机请求-响应握手计划审批高风险任务无审核直接执行计划提交-审批流程四、两季核心对比学生视角维度第九季基础版第十季进阶版核心能力队友能持久运行、异步发消息在第九季基础上沟通有规则、可追溯消息特点无结构化普通文本结构化带类型、request_id状态管理仅队友的工作状态working/idle新增请求状态pending/approved/rejected工具数量9个基础通信、执行12个关机请求/响应、计划审批协作规范无明确规范有请求-响应的协议规范五、代码运行与验证实操步骤作为学生你可以按以下步骤跑通代码理解实际效果1. 第九季运行# 进入代码目录cdlearn-claude-code# 运行第九季代码python agents/s09_agent_teams.py# 交互操作示例# 1. 创建alice编码员和bob测试员# 2. 让alice给bob发消息# 3. 广播“status update: phase 1 complete”给所有队友# 4. 查看领导的收件箱# 5. 查看团队名单/team和收件箱/inbox2. 第十季运行cdlearn-claude-code python agents/s10_team_protocols.py# 交互操作示例# 1. 创建alice编码员发送关机请求查看她的状态# 2. 创建bob高风险重构任务提交计划领导拒绝# 3. 创建charlie提交计划领导批准# 4. 查看团队状态/team总结核心知识点回顾第九季核心用TeammateManager实现队友持久化线程配置文件用MessageBusJSONL文件实现异步通信解决“代理能组队、能发消息”的基础问题第十季核心在通信基础上增加“请求-响应协议”通过request_id关联请求和响应用状态机管理请求状态解决“沟通有规则、可追溯、不混乱”的进阶问题关键设计思想从“无结构”到“结构化”是编程中常见的升级思路比如先做功能再做规范这和真实团队协作的逻辑一致。这两季的代码核心是“模拟人类团队的协作模式”把编程中的“线程、文件IO、状态管理、协议设计”等知识点结合起来是非常好的实战案例。
面试09-Agent 的团队协作
发布时间：2026/7/8 19:20:58
Agent 团队协作从 “异步通信” 到 “结构化协议”作为编程学习的视角我们先把这两季的核心逻辑串起来第九季解决了 “团队能沟通” 的问题基础通信层第十季解决了 “团队沟通有规则” 的问题协议层就像先给团队装了邮箱再制定了 “请假审批” “方案评审” 的公司制度。一、核心背景为什么要做这两季的升级在第九季之前第八季及更早AI代理存在两个致命问题一次性死亡子 Agent 用完就销毁没有 “身份” 和 “记忆” 没法持续协作无规则沟通就算能通信也只是零散消息没有 “请求-响应” 的规范比如关线程时容易数据丢失、改代码时没有审批流程。第九季先搭建 “能持续沟通的团队”第十季再给沟通加“规则和流程”一步步接近真实的人类团队协作。二、第九季搭建 “有身份、能异步通信的持久 Agent 团队”核心目标让AI Agent从 “一次性工具” 变成 “持久存在的队友” 支持跨 Agent 的异步消息传递比如A发消息给BB忙完再看。关键组件拆解带代码解释1. 团队管理核心TeammateManager队友生命周期管理作用创建、跟踪队友Agent给每个队友Agent分配独立线程运行保证同时工作。classTeammateManager:def__init__(self,team_dir:Path):self.dirteam_dir# 团队数据目录存配置、邮箱self.dir.mkdir(exist_okTrue)# 目录不存在则创建self.config_pathself.dir/config.json# 团队配置文件存队友名单、状态self.configself._load_config()# 加载配置比如{members: [{name:alice,role:coder,status:working}]}self.threads{}# 存储每个队友Agent的运行线程# 查找对应 Agent 的信息def_find_member(self,name:str)-dict:forminself.config[members]:ifm[name]name:returnmreturnNonedef_save_config(self,configNone):保存团队配置到文件configconfigorself.config self.config_path.write_text(json.dumps(config,indent2))defspawn(self,name:str,role:str,prompt:str)-str:# 1. 记录队友信息存储到配置文件持久化不会丢member{name:name,role:role,status:working}self.config[members].append(member)self._save_config()# 写入config.json就算程序重启也能找到队友# 2. 启动独立线程运行队友持久存在不是用完就扔threadthreading.Thread(targetself._teammate_loop,# 队友的核心工作循环args(name,role,prompt),daemonTrue# 守护线程主程序退出时自动结束)thread.start()self.threads[name]thread# 跟踪线程状态returnfSpawned teammate {name} (role:{role})问题一从 Agent 的创建上是如何实现异步通信的这和 Agent 的生命周期相关。具体可分为1Agent 的持久化、2线程创建跟管理 Agent 工作逻辑相关、3线程的管理第一部分 - 持久化通过对每个Agent创建json文件的方式持久化Agent的信息包含Agent的名称、角色、状态第二部分 - 创建线程给每个Agent开独立线程让他们一直活着为什么要开独立线程呢是为了让每个Agent的工作循环互不干扰比如 A 队友工作时B 队友可以摸鱼实现异步通信功能第三部分 - 管理线程通过字典的形式存储所有的Agent线程方便管理比如1查询线程是否存活2线程状态是否空闲3防止重复创建线程问题二Agent 的工作逻辑是如何实现异步通信的呢如以下代码所示def_teammate_loop(self,name:str,role:str,prompt:str): 单个队友的核心工作循环独立线程中运行 核心功能 1. 初始化队友的系统提示和对话上下文 2. 循环读取消息队列inbox接收其他队友/管理器的消息 3. 调用大模型LLM生成响应支持工具调用 4. 执行工具调用并返回结果给大模型 5. 循环最多50次或直到停止非工具调用 6. 循环结束后更新队友状态为idle并保存配置 参数 name: str - 队友名称唯一标识 role: str - 队友角色如后端开发、测试工程师 prompt: str - 初始任务提示队友的核心任务 # 1. 构建系统提示词定义队友的身份、工作目录、行为规则# WORKDIR是全局变量代表队友的工作目录send_message是预设的通信方式sys_prompt(fYou are {name}, role:{role}, at{WORKDIR}. # 身份与环境fUse send_message to communicate. Complete your task.# 行为规则用send_message通信完成任务)# 2. 初始化对话消息列表初始只有用户管理器下发的任务提示# messages是大模型的对话上下文格式遵循OpenAI API规范messages[{role:user,content:prompt}]# 3. 获取该队友可使用的工具列表如文件操作、命令执行、消息发送等# _teammate_tools()是管理器的私有方法返回该队友支持的工具定义符合OpenAI工具调用规范toolsself._teammate_tools()# 4. 核心工作循环最多执行50轮避免无限循环for_inrange(50):# 4.1 读取当前队友的收件箱消息从全局消息总线BUS中读取# BUS是全局消息总线read_inbox(name)获取发给该队友的所有未读消息inboxBUS.read_inbox(name)# 4.2 将收件箱的新消息添加到对话上下文转JSON保证格式统一formsgininbox:messages.append({role:user,content:json.dumps(msg)})try:# 4.3 调用大模型API生成响应# modelMODEL全局变量指定使用的大模型如gpt-4o# systemsys_prompt设置大模型的系统角色# toolstools传入该队友可调用的工具列表# max_tokens8000限制单次响应的最大token数responseclient.messages.create(modelMODEL,systemsys_prompt,messagesmessages,toolstools,max_tokens8000,)exceptException:# 捕获API调用异常如网络、额度、模型服务问题直接终止循环break# 4.4 将大模型的响应添加到对话上下文维护多轮对话状态messages.append({role:assistant,content:response.content})# 4.5 判断是否停止循环如果响应不是因为工具调用停止则结束工作# stop_reasontool_use表示大模型要求调用工具需要继续循环否则任务完成ifresponse.stop_reason!tool_use:break# 4.6 处理工具调用遍历响应中的所有工具调用块results[]forblockinresponse.content:# 筛选出类型为tool_use的响应块大模型要求执行的工具ifblock.typetool_use:# 执行工具调用_exec是管理器的私有方法封装工具执行逻辑# 参数调用者名称、工具名称、工具入参outputself._exec(name,block.name,block.input)# 打印工具执行结果截断到120字符避免输出过长print(f [{name}]{block.name}:{str(output)[:120]})# 整理工具执行结果符合OpenAI工具结果格式用于返回给大模型results.append({type:tool_result,tool_use_id:block.id,# 关联对应的工具调用IDcontent:str(output),# 工具执行的结果内容})# 4.7 将工具执行结果作为用户消息添加到上下文供下一轮大模型处理messages.append({role:user,content:results})# 5. 循环结束后更新队友状态并保存配置# 5.1 查找该队友的信息从管理器的成员列表中memberself._find_member(name)# 5.2 如果队友存在且未被关闭shutdown将状态改为idle闲置ifmemberandmember[status]!shutdown:member[status]idle# 5.3 保存配置到文件如config.json持久化状态变更self._save_config()异步功能实现的逻辑如下读取消息队列中的内容读取类似异步邮箱中的内容并将其封装到messages列表中第一次回答大模型回答解决邮箱中的问题两种情况:1无需调用工具说明一次QA就解决了2需要调用工具则和原来的Agent Loop一样的调用所需工具调用工具得到 tool response然后封装到messages列表中Agent Loop循环直至无需使用工具说明任务解决更新线程状态更新线程状态2. 通信核心MessageBus基于JSONL的异步邮箱作用实现队友间的异步消息传递就像公司的邮箱系统——A发消息给BB忙完再看消息存在文件里不会丢。classMessageBus:defsend(self,sender,to,content,msg_typemessage,extraNone):# 构造标准化消息谁发的、发给谁、内容、时间、类型msg{type:msg_type,# 消息类型普通消息/关机请求/计划审批from:sender,content:content,timestamp:time.time()# 时间戳保证顺序}ifextra:msg.update(extra)# 附加字段比如request_id# 追加写入收件人的JSONL文件append-only只加不删保证不丢消息withopen(self.dir/f{to}.jsonl,a)asf:f.write(json.dumps(msg)\n)defread_inbox(self,name):pathself.dir/f{name}.jsonlifnotpath.exists():return[]# 读取所有消息并解析msgs[json.loads(l)forlinpath.read_text().strip().splitlines()ifl]path.write_text()# 读完清空drain避免重复处理returnjson.dumps(msgs,indent2)MessageBus 的目的持久化 Agent 的异步邮箱每个Agent有自己的专属邮箱文件比如 alice.jsonl、bob.jsonl异步操作实现每个Agent发消息、读消息的功能消息存储到json文件中3. 队友的工作循环_teammate_loop作用让队友持续运行定期查邮箱和LLM交互实现“持久工作”。def_teammate_loop(self,name,role,prompt):# 初始化对话上下文messages[{role:user,content:prompt}]# 循环运行最多50轮避免无限循环for_inrange(50):# 第一步查邮箱有消息就注入上下文inboxBUS.read_inbox(name)ifinbox![]:messages.append({role:user,content:finbox{inbox}/inbox})messages.append({role:assistant,content:Noted inbox messages.})# 第二步调用LLM获取响应responseclient.messages.create(...)# 第三步如果LLM不调用工具了就停止本轮工作ifresponse.stop_reason!tool_use:break# 第四步执行工具比如发消息、写代码把结果追加到上下文# ...工具执行逻辑# 第五步标记队友状态为“空闲”self._find_member(name)[status]idle关键理解队友不是“一次性的”而是在循环里持续运行直到完成任务或达到循环上限每次和LLM对话前先查邮箱保证队友能及时收到其他成员的消息状态从“working”变为“idle”实现生命周期管理。第九季核心亮点特性解决的问题实现方式持久化队友Agent 用完就丢无身份线程config.json持久化队友信息异步通信Agent 间无法传递消息JSONL邮箱send/read_inbox生命周期管理Agent 只有“运行/死亡”两种状态working/idle状态切换三、第十季给 Agent 团队沟通加“结构化协议”核心目标第九季的队友能通信但沟通是“无规则的”——比如领导让队友关机队友可能直接退出导致文件没写完队友改代码没有审批流程直接动手。第十季通过“结构化协议”解决这些问题。 s10_team_protocols.py - 团队协议 关机协议和计划审批协议均使用相同的request_id关联模式。基于s09的团队消息功能扩展。 关机状态机FSM待处理pending→ 批准approved | 拒绝rejected 领导 队友 --------------------- --------------------- | 关机请求 | | | | { | ------- | 接收请求 | | request_id: abc | | 决策是否批准 | | } | | | --------------------- --------------------- | --------------------- -------v------------- | 关机响应 | ------- | 关机响应 | | { | | { | | request_id: abc | | request_id: abc | | approve: true | | approve: true | | } | | } | --------------------- --------------------- | v 状态 - 关机完成线程停止 计划审批状态机FSM待处理pending→ 批准approved | 拒绝rejected 队友 领导 --------------------- --------------------- | 计划审批 | | | | 提交{plan:...}| ------- | 审核计划文本 | --------------------- | 批准/拒绝 | --------------------- | --------------------- -------v------------- | 计划审批响应 | ------- | 计划审批 | | {approve: true} | | 审核{req_id, | --------------------- | approve: true} | --------------------- 跟踪器{request_id: {目标|发起方: 名称, 状态: 待处理|已批准|已拒绝}} 核心思路相同的request_id关联模式适配两个业务场景。 核心概念请求-响应协议 FSM有限状态机1. 核心思路所有重要沟通都遵循“请求带唯一ID→ 响应引用同一ID→ 状态更新”的流程用FSM有限状态机管理请求的状态pending→approved/rejected。2. 协议1关机协议Shutdown Protocol解决“强制关机导致数据丢失”的问题要求“领导发关机请求→队友确认批准/拒绝→再关机”。步骤1领导发送关机请求生成唯一request_idshutdown_requests{}# 跟踪所有关机请求的状态defhandle_shutdown_request(teammate:str)-str:# 生成8位唯一请求ID避免冲突req_idstr(uuid.uuid4())[:8]# 记录请求状态待处理shutdown_requests[req_id]{target:teammate,status:pending}# 发消息给目标队友消息类型是shutdown_request带request_idBUS.send(lead,# 发送者teammate,# 接收者Please shut down gracefully.,# 内容shutdown_request,# 消息类型{request_id:req_id}# 附加请求ID)returnfShutdown request{req_id}sent (status: pending)步骤2队友处理关机请求并响应# 队友收到shutdown_request后调用shutdown_response工具iftool_nameshutdown_response:req_idargs[request_id]# 引用领导的请求IDapproveargs[approve]# 批准/拒绝关机# 更新请求状态shutdown_requests[req_id][status]approvedifapproveelserejected# 回复领导同样带request_id保证关联BUS.send(sender,# 队友自己lead,# 接收者是领导args.get(reason,),# 拒绝理由可选shutdown_response,# 消息类型{request_id:req_id,approve:approve})3. 协议2计划审批协议Plan Approval Protocol解决“高风险任务无审批”的问题要求“队友提交计划带ID→领导审批引用ID→队友执行/修改”。plan_requests{}# 跟踪所有计划请求的状态# 队友提交计划defsubmit_plan(sender,plan_content):req_idstr(uuid.uuid4())[:8]plan_requests[req_id]{from:sender,plan:plan_content,status:pending}# 发消息给领导请求审批BUS.send(sender,lead,plan_content,plan_request,{request_id:req_id})returnfPlan submitted (ID:{req_id}), waiting for approval.# 领导审批计划defhandle_plan_review(request_id,approve,feedback):reqplan_requests[request_id]# 通过ID找到对应的计划req[status]approvedifapproveelserejected# 回复队友审批结果BUS.send(lead,req[from],feedback,# 审批意见比如“代码需要加注释”plan_approval_response,{request_id:request_id,approve:approve})4. 统一的FSM有限状态机两个协议共享同一个状态机逻辑简化代码[pending待处理] ↙ ↘ [approved批准] [rejected拒绝]不管是关机请求还是计划请求都只需要维护这三种状态逻辑统一容易扩展。第十季核心亮点特性解决的问题实现方式结构化协议沟通无规则易出错请求-响应模式唯一request_id状态跟踪不知道请求的处理结果shutdown_requests/plan_requests字典优雅关机强制关机导致数据丢失关机请求-响应握手计划审批高风险任务无审核直接执行计划提交-审批流程四、两季核心对比学生视角维度第九季基础版第十季进阶版核心能力队友能持久运行、异步发消息在第九季基础上沟通有规则、可追溯消息特点无结构化普通文本结构化带类型、request_id状态管理仅队友的工作状态working/idle新增请求状态pending/approved/rejected工具数量9个基础通信、执行12个关机请求/响应、计划审批协作规范无明确规范有请求-响应的协议规范五、代码运行与验证实操步骤作为学生你可以按以下步骤跑通代码理解实际效果1. 第九季运行# 进入代码目录cdlearn-claude-code# 运行第九季代码python agents/s09_agent_teams.py# 交互操作示例# 1. 创建alice编码员和bob测试员# 2. 让alice给bob发消息# 3. 广播“status update: phase 1 complete”给所有队友# 4. 查看领导的收件箱# 5. 查看团队名单/team和收件箱/inbox2. 第十季运行cdlearn-claude-code python agents/s10_team_protocols.py# 交互操作示例# 1. 创建alice编码员发送关机请求查看她的状态# 2. 创建bob高风险重构任务提交计划领导拒绝# 3. 创建charlie提交计划领导批准# 4. 查看团队状态/team总结核心知识点回顾第九季核心用TeammateManager实现队友持久化线程配置文件用MessageBusJSONL文件实现异步通信解决“代理能组队、能发消息”的基础问题第十季核心在通信基础上增加“请求-响应协议”通过request_id关联请求和响应用状态机管理请求状态解决“沟通有规则、可追溯、不混乱”的进阶问题关键设计思想从“无结构”到“结构化”是编程中常见的升级思路比如先做功能再做规范这和真实团队协作的逻辑一致。这两季的代码核心是“模拟人类团队的协作模式”把编程中的“线程、文件IO、状态管理、协议设计”等知识点结合起来是非常好的实战案例。
)
