1 介绍loop是Agent实现最重要的一个 part其中分为lead agent的 loop 和sub_agent的 loop2 代码-接收控制台输入的命令definteractive_mode(): 启动交互式命令行模式# 打印Bannerprint_banner()# 打印使用提示print_info(输入任务描述或输入 [bold]quit[/] 退出)print_info(使用 [bold]--plan[/] 前缀启用任务规划模式)console.print()# 循环接收任务whileTrue:try:user_inputget_user_input()# 退出命令ifuser_input.lower()in[quit,exit,q]:console.print(\n[bold cyan] 感谢使用 EasyAgent[/]\n);breakifnotuser_input.strip():continue# 检测是否启用任务规划use_planuser_input.startswith(--plan )taskuser_input.replace(--plan ,).strip()ifnottask:print_error(请输入有效的任务描述);continueconsole.print()# 执行智能体run_agent_plus(task,use_planuse_plan)console.print()exceptKeyboardInterrupt:console.print(\n[yellow]⚠️ 任务中断[/]);continueexceptExceptionase:print_error(f执行异常:{e});continue第一步边界条件的判断如果输入包含quit等退出指令则打印提示并break循环 loop第二步判断用户是否使用任务分解--plan如果使用的话作为智能体执行run_agent_plus()的参数3 代码-Agent的入口defrun_agent_plus(task,use_planFalse,keep_chattingFalse,use_teamFalse):增强版智能体入口# 加载历史记忆memoryload_memory()system_prompt您是一位能够与系统交互的得力助手。请言简意赅。# 拼接记忆到系统提示ifmemory:system_promptf\n\n之前的上下文:\n{memory}messages[{role:system,content:system_prompt}]# 使用 Agent 团队处理如果启用ifuse_teamandagent_team:print_section( 使用 Agent 团队处理,styleTHEME[primary])task_idagent_team.submit_task({content:task,description:task})print_info(f任务已提交到队列:{task_id})returnfTask{task_id}submitted to team# 生成任务步骤stepscreate_plan(task)ifuse_planelse[task]print_section( 开始执行任务,styleTHEME[success])all_results,step_index[],0# 遍历执行所有步骤whilestep_indexlen(steps):step_index1current_stepsteps[step_index-1]# 多步骤时显示当前步骤面板iflen(steps)1:console.print(Panel(f[bold]{current_step}[/],titlef[{THEME[primary]}] 步骤{step_index}/{len(steps)}[/],border_styleTHEME[border],padding(1,2)))# 执行单步任务result,actions,messagesrun_agent_step(current_step,messages)all_results.append(result)print_result(result,titlef✨ 步骤{step_index}执行完成)# 获取用户反馈feedback,should_continueget_step_advice(step_index,len(steps),current_step,result)# 处理反馈指令iffeedback__skip_all__:console.print([yellow]⏭️ 跳过剩余步骤任务提前结束[/]);break# 处理控制指令iffeedback__retry__:print_info(f 重新执行步骤{step_index}...)step_index-1# 回退索引重试当前步骤continueeliffeedbackconfirmed:# 用户确认无需额外操作继续下一步passeliffeedback:# 非空 用户建议直接透传给 AIconsole.print(Panel([dim] 正在处理您的建议...[/],border_styleTHEME[primary],padding(0,2)))messages.append({role:user,content:f[步骤反馈]{current_step}\n用户建议:{feedback}})responsewith_spinner( 思考中...,client.chat.completions.create,modelMODEL_NAME,messagesmessages,toolsTOOLS)ai_responseresponse.choices[0].message.content# 展示 AI 响应ifai_response:print_result(ai_response,titlef 响应)messages.append({role:assistant,content:ai_response})ifnotshould_continue:break# 合并所有结果并保存记忆final_result\n.join(all_results)save_memory(task,final_result)console.print(Panel([bold green] 任务执行完成[/],border_styleTHEME[success],padding(1,2)))# 开启连续对话ifkeep_chatting:messagescontinue_chatting(messages,task)returnfinal_result第一部分加载历史记忆这里我们用的是.md存储用户以往的对话对话内容包含Query和Resultsummary之后的内容所以 Token 消耗并不大具体操作就是简单的拼接然后以System的方式封装到messages列表中memoryload_memory()system_prompt您是一位能够与系统交互的得力助手。请言简意赅。# 拼接记忆到系统提示ifmemory:system_promptf\n\n之前的上下文:\n{memory}messages[{role:system,content:system_prompt}]第二部分是否使用 Agent Team如果使用Agent Team那么任务就提交到任务队列中字段包含task_id、description等顶层任务无依赖方# 使用 Agent 团队处理如果启用ifuse_teamandagent_team:print_section( 使用 Agent 团队处理,styleTHEME[primary])task_idagent_team.submit_task({content:task,description:task})print_info(f任务已提交到队列:{task_id})returnfTask{task_id}submitted to team第三部分任务分解本质上就是通过提示词的方式指定response_format让大模型生成json结构的分解步骤# 生成任务步骤stepscreate_plan(task)ifuse_planelse[task]print_section( 开始执行任务,styleTHEME[success])如果设置任务分解这里会是第一次问答第四部分回答每个 step 任务返回最终的结果其中result是模型对一个Step最后一次回答的结果完成一个Step至少需要回答两次而actions的内容是所有执行的工具的name和argsmessages列表中包含了tools_result和model_responseall_results,step_index[],0# 遍历执行所有步骤whilestep_indexlen(steps):step_index1current_stepsteps[step_index-1]# 多步骤时显示当前步骤面板iflen(steps)1:console.print(Panel(f[bold]{current_step}[/],titlef[{THEME[primary]}] 步骤{step_index}/{len(steps)}[/],border_styleTHEME[border],padding(1,2)))# 执行单步任务result,actions,messagesrun_agent_step(current_step,messages)all_results.append(result)print_result(result,titlef✨ 步骤{step_index}执行完成)单步骤执行的具体逻辑run_agent_step()先对任务进行请求提取tool_calling字段看需要调用的工具内容和参数注入工具再进行问答封装到messages列表中工具可以通过懒加载先把元数据加载到system_prompt中然后在具体使用具体某个 tool 的时候再加载详细的描述# 单步执行 defrun_agent_step(task,messages,max_iterations5):执行单个步骤# 将当前任务加入对话上下文messages.append({role:user,content:task})# 存储执行的动作actions[]# 最大迭代次数防止死循环foriterationinrange(max_iterations):ifiteration0:print_info(f第{iteration1}次迭代...)# 调用AI带加载动画responsewith_spinner(AI 思考中...,client.chat.completions.create,modelMODEL_NAME,messagesmessages,toolsTOOLS)messageresponse.choices[0].message# 将AI回复加入上下文messages.append({role:message.role,content:message.content,tool_calls:message.tool_callsifhasattr(message,tool_calls)elseNone})# 无工具调用直接返回结果ifnotmessage.tool_calls:resultmessage.contentifmessage.contentelse[dim]任务已完成无额外输出[/]returnresult,actions,messages# 遍历所有工具调用fortool_callinmessage.tool_calls:function_payloadgetattr(tool_call,function,None)iffunction_payloadisNone:continue# 获取函数名和参数function_namestr(getattr(function_payload,name,))raw_argumentsstr(getattr(function_payload,arguments,))function_argsparse_tool_arguments(raw_arguments)# 打印工具调用信息print_tool_call(function_name,function_args)# 查找并执行对应函数function_implAVAILABLE_FUNCTIONS.get(function_name)iffunction_implisNone:function_responsefError: Unknown tool {function_name}elif_argument_errorinfunction_args:function_responsefError:{function_args[_argument_error]}else:function_responsefunction_impl(**function_args)actions.append({tool:function_name,args:function_args})# 将工具执行结果加入上下文messages.append({role:tool,tool_call_id:tool_call.id,content:function_response})# 达到最大迭代次数终止执行print_error(达到最大迭代次数终止执行)returnMax iterations reached,actions,messages第五部分对每个 step 的结果提出建议get_step_advice从控制台读取user_input作为messages列表的输入第四次问答模型将根据用户的建议对step重新回答一次# 获取用户反馈feedback,should_continueget_step_advice(step_index,len(steps),current_step,result)# 处理反馈指令iffeedback__skip_all__:console.print([yellow]⏭️ 跳过剩余步骤任务提前结束[/]);break# 处理控制指令iffeedback__retry__:print_info(f 重新执行步骤{step_index}...)step_index-1# 回退索引重试当前步骤continueeliffeedbackconfirmed:# 用户确认无需额外操作继续下一步passeliffeedback:# 非空 用户建议直接透传给 AIconsole.print(Panel([dim] 正在处理您的建议...[/],border_styleTHEME[primary],padding(0,2)))messages.append({role:user,content:f[步骤反馈]{current_step}\n用户建议:{feedback}})responsewith_spinner( 思考中...,client.chat.completions.create,modelMODEL_NAME,messagesmessages,toolsTOOLS)ai_responseresponse.choices[0].message.content# 展示 AI 响应ifai_response:print_result(ai_response,titlef 响应)messages.append({role:assistant,content:ai_response})ifnotshould_continue:break第六部分存储记忆将第三次问答的response存储到.md中# 合并所有结果并保存记忆final_result\n.join(all_results)save_memory(task,final_result)console.print(Panel([bold green] 任务执行完成[/],border_styleTHEME[success],padding(1,2)))# 开启连续对话ifkeep_chatting:messagescontinue_chatting(messages,task)returnfinal_result总结第 1 次问答任务分解假设分为 n 个子任务1次问答第 2 ~ n1 次问答查看每个子任务需要调用的工具n次问答第 n2 ~ 2n2 次问答每次调用完工具至少还需要一次问答生成答案总结n次问答 第 2n3 ~ 3n3 次问答每个子任务可以提出建议n次问答一共3n1次问答如果你的任务很复杂那么单 Agent 会堵死去。这也是为什么要用懒加载和任务规划的原因
项目01-手搓Agent之loop
1 介绍loop是Agent实现最重要的一个 part其中分为lead agent的 loop 和sub_agent的 loop2 代码-接收控制台输入的命令definteractive_mode(): 启动交互式命令行模式# 打印Bannerprint_banner()# 打印使用提示print_info(输入任务描述或输入 [bold]quit[/] 退出)print_info(使用 [bold]--plan[/] 前缀启用任务规划模式)console.print()# 循环接收任务whileTrue:try:user_inputget_user_input()# 退出命令ifuser_input.lower()in[quit,exit,q]:console.print(\n[bold cyan] 感谢使用 EasyAgent[/]\n);breakifnotuser_input.strip():continue# 检测是否启用任务规划use_planuser_input.startswith(--plan )taskuser_input.replace(--plan ,).strip()ifnottask:print_error(请输入有效的任务描述);continueconsole.print()# 执行智能体run_agent_plus(task,use_planuse_plan)console.print()exceptKeyboardInterrupt:console.print(\n[yellow]⚠️ 任务中断[/]);continueexceptExceptionase:print_error(f执行异常:{e});continue第一步边界条件的判断如果输入包含quit等退出指令则打印提示并break循环 loop第二步判断用户是否使用任务分解--plan如果使用的话作为智能体执行run_agent_plus()的参数3 代码-Agent的入口defrun_agent_plus(task,use_planFalse,keep_chattingFalse,use_teamFalse):增强版智能体入口# 加载历史记忆memoryload_memory()system_prompt您是一位能够与系统交互的得力助手。请言简意赅。# 拼接记忆到系统提示ifmemory:system_promptf\n\n之前的上下文:\n{memory}messages[{role:system,content:system_prompt}]# 使用 Agent 团队处理如果启用ifuse_teamandagent_team:print_section( 使用 Agent 团队处理,styleTHEME[primary])task_idagent_team.submit_task({content:task,description:task})print_info(f任务已提交到队列:{task_id})returnfTask{task_id}submitted to team# 生成任务步骤stepscreate_plan(task)ifuse_planelse[task]print_section( 开始执行任务,styleTHEME[success])all_results,step_index[],0# 遍历执行所有步骤whilestep_indexlen(steps):step_index1current_stepsteps[step_index-1]# 多步骤时显示当前步骤面板iflen(steps)1:console.print(Panel(f[bold]{current_step}[/],titlef[{THEME[primary]}] 步骤{step_index}/{len(steps)}[/],border_styleTHEME[border],padding(1,2)))# 执行单步任务result,actions,messagesrun_agent_step(current_step,messages)all_results.append(result)print_result(result,titlef✨ 步骤{step_index}执行完成)# 获取用户反馈feedback,should_continueget_step_advice(step_index,len(steps),current_step,result)# 处理反馈指令iffeedback__skip_all__:console.print([yellow]⏭️ 跳过剩余步骤任务提前结束[/]);break# 处理控制指令iffeedback__retry__:print_info(f 重新执行步骤{step_index}...)step_index-1# 回退索引重试当前步骤continueeliffeedbackconfirmed:# 用户确认无需额外操作继续下一步passeliffeedback:# 非空 用户建议直接透传给 AIconsole.print(Panel([dim] 正在处理您的建议...[/],border_styleTHEME[primary],padding(0,2)))messages.append({role:user,content:f[步骤反馈]{current_step}\n用户建议:{feedback}})responsewith_spinner( 思考中...,client.chat.completions.create,modelMODEL_NAME,messagesmessages,toolsTOOLS)ai_responseresponse.choices[0].message.content# 展示 AI 响应ifai_response:print_result(ai_response,titlef 响应)messages.append({role:assistant,content:ai_response})ifnotshould_continue:break# 合并所有结果并保存记忆final_result\n.join(all_results)save_memory(task,final_result)console.print(Panel([bold green] 任务执行完成[/],border_styleTHEME[success],padding(1,2)))# 开启连续对话ifkeep_chatting:messagescontinue_chatting(messages,task)returnfinal_result第一部分加载历史记忆这里我们用的是.md存储用户以往的对话对话内容包含Query和Resultsummary之后的内容所以 Token 消耗并不大具体操作就是简单的拼接然后以System的方式封装到messages列表中memoryload_memory()system_prompt您是一位能够与系统交互的得力助手。请言简意赅。# 拼接记忆到系统提示ifmemory:system_promptf\n\n之前的上下文:\n{memory}messages[{role:system,content:system_prompt}]第二部分是否使用 Agent Team如果使用Agent Team那么任务就提交到任务队列中字段包含task_id、description等顶层任务无依赖方# 使用 Agent 团队处理如果启用ifuse_teamandagent_team:print_section( 使用 Agent 团队处理,styleTHEME[primary])task_idagent_team.submit_task({content:task,description:task})print_info(f任务已提交到队列:{task_id})returnfTask{task_id}submitted to team第三部分任务分解本质上就是通过提示词的方式指定response_format让大模型生成json结构的分解步骤# 生成任务步骤stepscreate_plan(task)ifuse_planelse[task]print_section( 开始执行任务,styleTHEME[success])如果设置任务分解这里会是第一次问答第四部分回答每个 step 任务返回最终的结果其中result是模型对一个Step最后一次回答的结果完成一个Step至少需要回答两次而actions的内容是所有执行的工具的name和argsmessages列表中包含了tools_result和model_responseall_results,step_index[],0# 遍历执行所有步骤whilestep_indexlen(steps):step_index1current_stepsteps[step_index-1]# 多步骤时显示当前步骤面板iflen(steps)1:console.print(Panel(f[bold]{current_step}[/],titlef[{THEME[primary]}] 步骤{step_index}/{len(steps)}[/],border_styleTHEME[border],padding(1,2)))# 执行单步任务result,actions,messagesrun_agent_step(current_step,messages)all_results.append(result)print_result(result,titlef✨ 步骤{step_index}执行完成)单步骤执行的具体逻辑run_agent_step()先对任务进行请求提取tool_calling字段看需要调用的工具内容和参数注入工具再进行问答封装到messages列表中工具可以通过懒加载先把元数据加载到system_prompt中然后在具体使用具体某个 tool 的时候再加载详细的描述# 单步执行 defrun_agent_step(task,messages,max_iterations5):执行单个步骤# 将当前任务加入对话上下文messages.append({role:user,content:task})# 存储执行的动作actions[]# 最大迭代次数防止死循环foriterationinrange(max_iterations):ifiteration0:print_info(f第{iteration1}次迭代...)# 调用AI带加载动画responsewith_spinner(AI 思考中...,client.chat.completions.create,modelMODEL_NAME,messagesmessages,toolsTOOLS)messageresponse.choices[0].message# 将AI回复加入上下文messages.append({role:message.role,content:message.content,tool_calls:message.tool_callsifhasattr(message,tool_calls)elseNone})# 无工具调用直接返回结果ifnotmessage.tool_calls:resultmessage.contentifmessage.contentelse[dim]任务已完成无额外输出[/]returnresult,actions,messages# 遍历所有工具调用fortool_callinmessage.tool_calls:function_payloadgetattr(tool_call,function,None)iffunction_payloadisNone:continue# 获取函数名和参数function_namestr(getattr(function_payload,name,))raw_argumentsstr(getattr(function_payload,arguments,))function_argsparse_tool_arguments(raw_arguments)# 打印工具调用信息print_tool_call(function_name,function_args)# 查找并执行对应函数function_implAVAILABLE_FUNCTIONS.get(function_name)iffunction_implisNone:function_responsefError: Unknown tool {function_name}elif_argument_errorinfunction_args:function_responsefError:{function_args[_argument_error]}else:function_responsefunction_impl(**function_args)actions.append({tool:function_name,args:function_args})# 将工具执行结果加入上下文messages.append({role:tool,tool_call_id:tool_call.id,content:function_response})# 达到最大迭代次数终止执行print_error(达到最大迭代次数终止执行)returnMax iterations reached,actions,messages第五部分对每个 step 的结果提出建议get_step_advice从控制台读取user_input作为messages列表的输入第四次问答模型将根据用户的建议对step重新回答一次# 获取用户反馈feedback,should_continueget_step_advice(step_index,len(steps),current_step,result)# 处理反馈指令iffeedback__skip_all__:console.print([yellow]⏭️ 跳过剩余步骤任务提前结束[/]);break# 处理控制指令iffeedback__retry__:print_info(f 重新执行步骤{step_index}...)step_index-1# 回退索引重试当前步骤continueeliffeedbackconfirmed:# 用户确认无需额外操作继续下一步passeliffeedback:# 非空 用户建议直接透传给 AIconsole.print(Panel([dim] 正在处理您的建议...[/],border_styleTHEME[primary],padding(0,2)))messages.append({role:user,content:f[步骤反馈]{current_step}\n用户建议:{feedback}})responsewith_spinner( 思考中...,client.chat.completions.create,modelMODEL_NAME,messagesmessages,toolsTOOLS)ai_responseresponse.choices[0].message.content# 展示 AI 响应ifai_response:print_result(ai_response,titlef 响应)messages.append({role:assistant,content:ai_response})ifnotshould_continue:break第六部分存储记忆将第三次问答的response存储到.md中# 合并所有结果并保存记忆final_result\n.join(all_results)save_memory(task,final_result)console.print(Panel([bold green] 任务执行完成[/],border_styleTHEME[success],padding(1,2)))# 开启连续对话ifkeep_chatting:messagescontinue_chatting(messages,task)returnfinal_result总结第 1 次问答任务分解假设分为 n 个子任务1次问答第 2 ~ n1 次问答查看每个子任务需要调用的工具n次问答第 n2 ~ 2n2 次问答每次调用完工具至少还需要一次问答生成答案总结n次问答 第 2n3 ~ 3n3 次问答每个子任务可以提出建议n次问答一共3n1次问答如果你的任务很复杂那么单 Agent 会堵死去。这也是为什么要用懒加载和任务规划的原因
