面试11-Agent如何自动接任务

一、整体场景与核心目标解释首先我们先明确这段代码要解决的核心问题在第9-10季的代码中队友代理teammate agent只能在负责人lead明确分配任务时才工作10个未认领任务需要负责人手动分配无法规模化。而第11季的核心目标是实现自治代理让队友代理自己扫描任务板、认领无人处理的任务、完成任务无需负责人逐个分配同时通过工作-闲置循环和身份重注入保证代理的自主性和身份不丢失。二、核心流程拆解与代码逐段解释整个自治代理的核心逻辑是生成代理 → 工作阶段 → 闲置阶段轮询任务/消息→ 要么回到工作阶段要么超时关机。下面按流程拆解每段代码。流程1代理核心循环_loop函数—— 工作闲置的主逻辑def_loop(self,name,role,prompt):whileTrue:# -- WORK PHASE --messages[{role:user,content:prompt}]for_inrange(50):responseclient.messages.create(...)# 调用大模型LLMifresponse.stop_reason!tool_use:break# 大模型停止调用工具退出工作阶段# execute tools... # 执行大模型调用的工具比如处理任务ifidle_requested:# 如果收到闲置请求break# 主动退出工作阶段# -- IDLE PHASE --self._set_status(name,idle)# 将代理状态设为闲置resumeself._idle_poll(name,messages)# 进入闲置轮询ifnotresume:# 轮询后无需恢复工作超时self._set_status(name,shutdown)# 设为关机return# 退出循环代理停止self._set_status(name,working)# 轮询到任务/消息恢复工作状态代码解释小白友好版def _loop(...)定义代理的核心循环函数name是代理名字比如alicerole是代理角色比如程序员prompt是初始任务提示。while True无限循环直到代理超时关机才会退出。工作阶段WORK PHASEmessages [...]初始化大模型的输入消息把初始任务提示传给大模型。for _ in range(50)最多让大模型执行50轮工具调用防止无限工作。client.messages.create(...)调用大模型比如Claude/OpenAI让代理处理任务。stop_reason ! tool_use专业词汇解释 →stop_reason是大模型的停止原因tool_use表示大模型调用了工具比如执行修复认证错误任务如果停止原因不是调用工具说明任务处理完了退出工作阶段。idle_requested如果代理收到闲置指令比如任务暂时做不了主动退出工作阶段。闲置阶段IDLE PHASE_set_status(name, idle)更新代理状态为闲置方便监控比如在/team目录能看到。_idle_poll(...)调用闲置轮询函数后面会讲检查有没有新任务/消息。如果resume是False轮询超时就把代理设为关机并退出循环如果是True有新任务/消息就把状态切回工作重新进入工作阶段。本段总结_loop函数是代理的心脏核心逻辑是无限循环的工作-闲置交替先让代理调用大模型处理任务工作阶段任务停了就进入闲置阶段闲置时检查有没有新任务/消息有就继续工作没有就超时关机。流程2闲置轮询_idle_poll函数—— 代理主动找活干的核心# 定义常量小白解释固定不变的数值方便修改IDLE_TIMEOUT60# 闲置超时时间60秒POLL_INTERVAL5# 轮询间隔5秒def_idle_poll(self,name,messages):# 循环次数 总超时时间 / 轮询间隔 → 60/512次for_inrange(IDLE_TIMEOUT//POLL_INTERVAL):time.sleep(POLL_INTERVAL)# 等待5秒避免频繁检查节省资源# 第一步检查收件箱有没有新消息inboxBUS.read_inbox(name)# 读取当前代理的收件箱ifinbox:# 如果有新消息# 把消息加到大模型的输入里告诉代理有新消息messages.append({role:user,content:finbox{inbox}/inbox})returnTrue# 有消息需要恢复工作# 第二步扫描任务板找未认领任务unclaimedscan_unclaimed_tasks()# 调用扫描函数后面讲ifunclaimed:# 如果有未认领任务claim_task(unclaimed[0][id],name)# 认领第一个未认领任务# 把认领的任务加到大模型输入里告诉代理要处理这个任务messages.append({role:user,content:fauto-claimedTask #{unclaimed[0][id]}: f{unclaimed[0][subject]}/auto-claimed})returnTrue# 有任务需要恢复工作returnFalse# 12次轮询60秒都没消息/任务返回False超时代码解释小白友好版IDLE_TIMEOUT/POLL_INTERVAL定义闲置的核心参数60秒内每5秒检查一次总共检查12次。time.sleep(POLL_INTERVAL)专业词汇解释 →sleep是让程序暂停执行指定秒数这里暂停5秒避免代理每秒都检查任务板浪费电脑资源。检查收件箱BUS.read_inbox(name)BUS是代理之间的消息总线可以理解为公司的内部邮箱系统读取当前代理比如alice的收件箱。如果有消息比如负责人发了新指令就把消息加到大模型的输入里返回True告诉核心循环要继续工作。扫描并认领任务scan_unclaimed_tasks()调用扫描函数找无人认领的任务。claim_task(...)专业词汇解释 →claim是认领这个函数把任务的所有者设为当前代理比如把修复认证错误的所有者设为alice。把认领的任务信息加到大模型输入里返回True继续工作。循环结束后如果没找到消息/任务返回False核心循环会让代理关机。本段总结_idle_poll是代理主动找活干的逻辑闲置时每5秒检查一次先看有没有人发消息再看任务板有没有没人做的任务只要找到一个就认领并恢复工作60秒都没找到就关机。流程3扫描未认领任务scan_unclaimed_tasks函数—— 找没人做的任务TASKS_DIRPath(./tasks)# 任务板目录代码里的任务板就是这个文件夹defscan_unclaimed_tasks()-list:unclaimed[]# 初始化空列表存未认领任务# 遍历tasks目录下所有task_*.json文件每个文件是一个任务forfinsorted(TASKS_DIR.glob(task_*.json)):taskjson.loads(f.read_text())# 读取并解析任务文件JSON格式# 检查任务是否满足未认领条件# 1. status是pending待处理2. 没有owner所有者3. 没有blockedBy被其他任务阻塞if(task.get(status)pendingandnottask.get(owner)andnottask.get(blockedBy)):unclaimed.append(task)# 满足条件加入未认领列表returnunclaimed# 返回所有未认领任务代码解释小白友好版TASKS_DIR Path(./tasks)专业词汇解释 →Path是Python处理文件路径的工具这里指定任务文件都存在./tasks文件夹里就是你说的扫描板/任务板。def scan_unclaimed_tasks() - list函数返回值类型是列表list里面存的是未认领任务。TASKS_DIR.glob(task_*.json)专业词汇解释 →glob是通配符匹配这里匹配所有以task_开头、.json结尾的文件比如task_1.json、task_2.json每个文件对应一个任务。json.loads(f.read_text())专业词汇解释 →JSON是一种通用的任务数据格式比如{“task”:“修复认证错误”,“status”:“pending”,“owner”:“”}loads是把文件里的文字转换成Python能识别的字典格式。任务未认领的三个条件status pending任务状态是待处理不是完成或阻塞not task.get(owner)没有所有者比如没人填alicenot task.get(blockedBy)没有被其他任务阻塞比如写测试不会因为修复认证错误没做而被阻塞。满足条件的任务会被加入unclaimed列表最后返回这个列表。本段总结scan_unclaimed_tasks函数是代理的任务扫描器核心是遍历./tasks文件夹里的任务文件筛选出待处理、无所有者、无阻塞的任务给代理提供可认领的任务列表。流程4身份重注入 —— 防止代理忘了自己是谁team_namedev_team# 代理所属团队名iflen(messages)3:# 插入身份块到消息列表的第一个位置messages.insert(0,{role:user,content:fidentityYou are {name}, role:{role}, fteam:{team_name}. Continue your work./identity})# 插入代理的确认回复强化身份messages.insert(1,{role:assistant,content:fI am{name}. Continuing.})代码解释小白友好版背景第6季的上下文压缩会删减历史消息比如把100条消息压缩成3条可能导致代理忘记自己的名字、角色比如我是alice程序员。len(messages) 3检查消息列表的长度如果≤3说明压缩过了就重新注入身份。messages.insert(0, ...)专业词汇解释 →insert(0, ...)是在列表的第一个位置插入内容保证大模型优先看到身份信息。identity.../identity身份块明确告诉代理你是谁、你的角色、所属团队防止失忆。第二个insert是让代理自己确认身份比如我是alice继续工作进一步强化身份记忆。本段总结身份重注入是解决代理失忆的关键当上下文消息太少压缩过就主动把代理的身份信息加到消息最前面保证代理始终知道自己的角色和任务。流程5运行代码的实操指令cdlearn-claude-code# 进入代码目录python agents/s11_autonomous_agents.py# 运行第11季的自治代理代码代码解释小白友好版cd learn-claude-code专业词汇解释 →cd是终端命令意思是切换目录进入存放代码的文件夹。python agents/s11_autonomous_agents.py运行指定路径下的Python文件启动自治代理程序。后续操作说明创建3个任务到任务板往./tasks文件夹里加task_*.json文件启动alice和bob两个代理它们会自动扫描并认领任务启动程序员代理它会自己找任务板里的活干创建有依赖的任务比如写测试依赖修复认证错误代理会遵守依赖顺序先做前者再做后者。查看任务板直接打开./tasks文件夹看任务文件里的owner字段知道谁认领了任务监控代理状态查看./team文件夹知道谁在工作、谁在闲置。本段总结这部分是实操指令核心是通过终端命令启动代码然后创建任务、观察代理的自治行为同时通过指定文件夹监控任务和代理状态。三、核心变化对比第10季 vs 第11季组成部分第10季第11季工具数量12个14个新增闲置和认领任务工具自治权负责人分配任务代理自组织自己找任务空闲阶段无轮询收件箱任务板任务认领仅手动分配自动认领未认领任务身份仅初始系统提示压缩后重注入身份暂停/关机无60秒闲置后自动关机四、示例好我直接给你一套完整、可跑通、从头到尾的真实示例用最简单的 Python 代码 任务文件 运行流程完全模拟第11季自治代理。你是小白也能一步看懂代理怎么启动 → 怎么闲置 → 怎么扫任务板 → 怎么自动认领 → 怎么工作。完整自治代理示例可直接理解1. 先搭建环境模拟任务板我们在当前目录新建一个文件夹tasks/里面放 3 个任务文件task_1.json{id:1,subject:修复认证错误,status:pending,owner:null,blockedBy:null}task_2.json{id:2,subject:添加速率限制器,status:pending,owner:null,blockedBy:null}task_3.json{id:3,subject:写测试用例,status:pending,owner:null,blockedBy:1}注意task_3 被 task_1 阻塞代理不会抢这个。2. 完整可理解代码简化版 s11 自治代理我把原逻辑精简、可运行、每一步都打印日志你能清晰看到流程。importjsonimporttimefrompathlibimportPath# 配置 TASKS_DIRPath(./tasks)IDLE_TIMEOUT60# 最大闲置60秒POLL_INTERVAL5# 每5秒扫一次任务板team_nameAI-Team# 工具1扫描未认领任务 defscan_unclaimed_tasks():unclaimed[]forfinsorted(TASKS_DIR.glob(task_*.json)):taskjson.loads(f.read_text(utf-8))# 条件待处理 没人认领 不被阻塞if(task.get(status)pendingandnottask.get(owner)andnottask.get(blockedBy)):unclaimed.append(task)returnunclaimed# 工具2认领任务 defclaim_task(task_id,agent_name):fTASKS_DIR/ftask_{task_id}.jsontaskjson.loads(f.read_text(utf-8))task[owner]agent_name f.write_text(json.dumps(task,indent2),utf-8)print(f【系统】{agent_name}已认领任务{task_id}:{task[subject]})# 闲置轮询核心主动找活干 defidle_poll(agent_name,messages):print(f\n【{agent_name}】进入闲置每5秒扫描任务板……)# 60秒 / 5秒 12轮foriinrange(IDLE_TIMEOUT//POLL_INTERVAL):print(f 第{i1}轮扫描...)time.sleep(POLL_INTERVAL)# 1. 先看有没有消息这里简化模拟无消息# 2. 扫描任务板unclaimedscan_unclaimed_tasks()ifunclaimed:# 认领第一个可做任务taskunclaimed[0]claim_task(task[id],agent_name)# 把任务加入上下文messages.append({role:user,content:fauto-claimed任务 #{task[id]}:{task[subject]}})returnTrue# 回去工作# 60秒超时print(f【{agent_name}】60秒无任务自动关机)returnFalse# 身份重注入防止失忆 definject_identity(name,role,messages):iflen(messages)3:print(f【系统】上下文太短重新注入{name}的身份)messages.insert(0,{role:user,content:fidentity你是{name}角色{role}团队{team_name}继续工作})messages.insert(1,{role:assistant,content:f我是{name}继续执行任务})# 代理主循环 defagent_loop(name,role,prompt):print(f 启动自治代理{name}{role})messages[{role:user,content:prompt}]whileTrue:# 工作阶段 print(f\n【{name}】进入工作阶段)inject_identity(name,role,messages)# 防止失忆# 模拟 LLM 工作最多5轮工具调用work_step0for_inrange(5):work_step1print(f 工作步骤{work_step}处理任务中...)# 假设工作完成停止调用工具break# 进入闲置阶段 print(f【{name}】工作暂停切换到闲置)resumeidle_poll(name,messages)ifnotresume:print(f\n{name}已关闭 )return# 启动两个代理Alice Bob if__name____main__:# 启动代理Alice程序员agent_loop(nameAlice,role后端开发,prompt你是自治AI代理会自动从任务板认领并完成任务)# 启动代理Bob测试# agent_loop(# nameBob,# role测试工程师,# prompt你是自治AI代理自动认领任务# )3. 运行一遍你会看到完整真实流程直接运行代码你会看到如下输出我给你翻译成人话流程第一步启动代理 启动自治代理Alice后端开发第二步进入工作阶段【Alice】进入工作阶段 【系统】上下文太短重新注入 Alice 的身份 工作步骤 1处理任务中...因为消息只有1条触发身份重注入防止代理失忆。第三步工作结束 → 进入闲置【Alice】工作暂停切换到闲置 【Alice】进入闲置每5秒扫描任务板…… 第1轮扫描...第四步5秒后扫描任务板 → 发现无人认领任务【系统】Alice 已认领任务 1: 修复认证错误第五步返回 True → 回到工作阶段【Alice】回到工作阶段 工作步骤 1处理任务中...第六步再次闲置 → 再次扫描 → 认领 task_2【Alice】进入闲置每5秒扫描任务板…… 第1轮扫描... 【系统】Alice 已认领任务 2: 添加速率限制器第七步task_3 被阻塞 → 不会认领扫描到 task_3但 blockedBy1 → 跳过第八步所有任务做完 → 60秒后超时关机【Alice】60秒无任务自动关机 Alice 已关闭 4. 任务板最终状态你打开文件能看到task_1.jsonowner:Alicetask_2.jsonowner:Alicetask_3.jsonowner:null,blockedBy:15. 这个例子完整对应第11季所有核心逻辑代理自主循环工作 ↔ 闲置主动扫描任务板不需要负责人分配自动认领任务谁先扫到谁领遵守依赖阻塞不领被 blocked 的任务身份重注入防止上下文压缩后失忆超时自动关机60秒没活就停6. 如果你想看“两个代理抢任务”只需要把代码里的 Bob 也启动if__name____main__:agent_loop(Alice,后端开发,...)agent_loop(Bob,测试工程师,...)运行后你会看到Alice 领 task_1Bob 领 task_2完全自治不需要你插手如果你需要我可以再给你做一个**带 FSM 状态机怠速→投票→主张→工作**的可视化流程图版本或者把代码扩展成 14 个工具完整版。五、总结核心逻辑回顾核心循环代理通过工作-闲置无限循环实现自治工作阶段处理任务闲置阶段主动找任务/消息60秒没活干就关机。任务认领代理通过扫描./tasks文件夹的JSON任务文件自动筛选并认领待处理、无所有者、无阻塞的任务无需人工分配。身份保障当上下文消息被压缩≤3条会主动注入身份信息防止代理忘记自己的角色和所属团队。整个代码的核心是让代理从被动等分配变成主动找活干实现真正的自治同时通过超时、身份重注入保证代理的稳定性和正确性。