ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

smolagents Human-in-the-Loop 实战:用 step callback 交互式定制 Agent 计划并保留记忆

smolagents Human-in-the-Loop 实战:用 step callback 交互式定制 Agent 计划并保留记忆 smolagents Human-in-the-Loop 实战用 step callback 交互式定制 Agent 计划并保留记忆【免费下载链接】smolagents smolagents: a barebones library for agents that think in code.项目地址: https://gitcode.com/gh_mirrors/smo/smolagents导读本篇文章围绕 smolagents 官方示例 examples/plan_customization/plan_customization.py 展开讲解如何在CodeAgent生成计划PlanningStep后立即暂停执行把计划交给人类用户审查、批准、修改或取消再携带完整记忆恢复执行。读完本文你将掌握 smolagents 的 step callback 注册机制、planning_interval的调度原理、resetFalse的内存保留语义以及如何用这些能力构建人在回路Human-in-the-Loop, HITL的可控 Agent 工作流。本文同时结合仓库源码src/smolagents/agents.py、src/smolagents/memory.py说明每个配置项背后的实现细节。为什么需要人在回路的计划定制smolagents 的核心设计是让 Agent 用代码思考think in codeCodeAgent在每一步执行 Python 代码、调用工具、观察结果直到给出最终答案。但完全自动化的执行存在两个痛点失控风险复杂任务一旦方向跑偏Agent 会沿着错误路径消耗大量步数token 与时间缺乏干预窗口默认情况下计划生成后 Agent 会直接继续执行人类没有机会在关键时刻介入。HITLHuman-in-the-Loop策略把计划生成这个关键节点变成人机交互点Agent 先产出计划人类审查并决定批准Approve、修改Modify或取消Cancel之后 Agent 才基于最终确定的计划继续执行。示例文档给出了这一策略的四个目标见 docs/source/en/examples/plan_customization.md使用 step callback 在计划生成后中断 Agent 执行允许用户在执行前审查并修改计划在保留 Agent 记忆的前提下恢复执行基于用户反馈动态更新计划让人类始终掌握控制权。核心概念一PlanningStep 与 step callback 的中断机制PlanningStep 是什么在 smolagents 中一次运行产生的所有过程记录都被抽象为MemoryStep及其子类其中计划步骤对应PlanningStep。从 src/smolagents/memory.py 可以看到其定义dataclass class PlanningStep(MemoryStep): model_input_messages: list[ChatMessage] model_output_message: ChatMessage plan: str # 计划文本后续可直接修改 timing: Timing token_usage: TokenUsage | None None关键字段是plan字符串类型的计划内容——示例正是通过改写memory_step.plan实现计划修改。此外PlanningStep.to_messages()会把计划以ASSISTANT消息 Now proceed and carry out this plan. 的用户消息形式回灌给模型因此修改plan字段会直接影响后续模型收到的上下文。step_callbacks按步骤类型注册回调CodeAgent接受step_callbacks参数支持两种形态见 src/smolagents/agents.py 的参数文档与 src/smolagents/agents.py 的_setup_step_callbacks实现# 形态一list —— 为兼容旧版本统一注册到 ActionStep step_callbacks[my_callback] # 形态二dict —— 按具体 MemoryStep 子类注册 step_callbacks{PlanningStep: interrupt_after_plan}源码中_setup_step_callbacks的逻辑是传入list时为兼容旧行为回调被注册到ActionStep每个动作步骤都会触发传入dict时按{步骤类: 回调或回调列表}逐个注册无论哪种方式都会额外把监控器monitor.update_metrics注册到ActionStep。回调最终存于CallbackRegistrysrc/smolagents/memory.py它是一个按步骤类维护回调列表的注册表。每次步骤完成时agents.py 的_finalize_step会调用def _finalize_step(self, memory_step: ActionStep | PlanningStep | FinalAnswerStep): if not isinstance(memory_step, FinalAnswerStep): memory_step.timing.end_time time.time() self.step_callbacks.callback(memory_step, agentself)CallbackRegistry.callback在派发时沿memory_step.__class__.__mro__查找所有匹配步骤类的回调并根据回调函数签名决定调用方式单参数回调只接收memory_step多参数回调接收(memory_step, agentself)。这解释了示例中interrupt_after_plan(memory_step, agent)为何能拿到agent实例——正因它声明了两个参数。回调触发时机与中断检查计划步骤并非每一步都生成而是由planning_interval控制。在_run_streamsrc/smolagents/agents.py中调度条件为if self.planning_interval is not None and ( self.step_number 1 or (self.step_number - 1) % self.planning_interval 0 ):即第 1 步必然规划之后每隔planning_interval步规划一次。计划生成后追加到self.memory.steps并调用_finalize_step从而触发我们注册的PlanningStep回调。真正的中断检查发生在每轮循环开头src/smolagents/agents.pywhile not returned_final_answer and self.step_number max_steps: if self.interrupt_switch: raise AgentError(Agent interrupted., self.logger)而Agent.interrupt()src/smolagents/agents.py只是把self.interrupt_switch置为True。也就是说回调里调用agent.interrupt()并不会立刻终止而是设置开关循环在下一轮开始时抛出AgentError(Agent interrupted.)示例代码正是通过捕获这个异常来感知用户取消了执行。核心概念二完整的回调实现与三选一交互示例 plan_customization.py 定义了三个辅助函数加一个核心回调1. 展示计划display_plan(plan_content)def display_plan(plan_content): Display the plan in a formatted way print(\n * 60) print( AGENT PLAN CREATED) print( * 60) print(plan_content) print( * 60)2. 获取用户选择get_user_choice()def get_user_choice(): Get users choice for plan approval while True: choice input(\nChoose an option:\n1. Approve plan\n2. Modify plan\n3. Cancel\nYour choice (1-3): ).strip() if choice in [1, 2, 3]: return int(choice) print(Invalid choice. Please enter 1, 2, or 3.)3. 修改计划get_modified_plan(original_plan)def get_modified_plan(original_plan): Allow user to modify the plan print(\n - * 40) print(MODIFY PLAN) print(- * 40) print(Current plan:) print(original_plan) print(- * 40) print(Enter your modified plan (press Enter twice to finish):) lines [] empty_line_count 0 while empty_line_count 2: line input() if line.strip() : empty_line_count 1 else: empty_line_count 0 lines.append(line) # Remove the last two empty lines modified_plan \n.join(lines[:-2]) return modified_plan if modified_plan.strip() else original_plan这里用连续输入两个空行作为多行输入结束标志并且如果用户清空了内容则回退到原计划避免产生空计划。4. 核心回调interrupt_after_plan(memory_step, agent)def interrupt_after_plan(memory_step, agent): Step callback that interrupts the agent after a planning step is created. This allows for user interaction to review and potentially modify the plan. if isinstance(memory_step, PlanningStep): print(\n Agent interrupted after plan creation...) # Display the created plan display_plan(memory_step.plan) # Get user choice choice get_user_choice() if choice 1: # Approve plan print(✅ Plan approved! Continuing execution...) # Dont interrupt - let the agent continue return elif choice 2: # Modify plan # Get modified plan from user modified_plan get_modified_plan(memory_step.plan) # Update the plan in the memory step memory_step.plan modified_plan print(\nPlan updated!) display_plan(modified_plan) print(✅ Continuing with modified plan...) # Dont interrupt - let the agent continue with modified plan return elif choice 3: # Cancel print(❌ Execution cancelled by user.) agent.interrupt() return这段代码展示了三种分支的完整语义选择行为实现方式1 批准打印提示后直接returnAgent 按原计划继续不调用interrupt()2 修改读取用户新计划直接改写memory_step.plan然后继续执行修改内存中的计划对象3 取消打印提示后调用agent.interrupt()置位中断开关下一轮抛出AgentError注意修改分支的巧妙之处由于PlanningStep已追加进agent.memory.steps直接修改其plan字段就等价于修改了 Agent 的记忆上下文——后续模型看到的是新计划从而以新计划为导向继续行动。运行时交互示意按文档描述实际运行时的终端交互大致如下 AGENT PLAN CREATED 1. Search for recent AI developments 2. Analyze the top results 3. Summarize the 3 most significant breakthroughs 4. Include sources for each breakthrough Choose an option: 1. Approve plan 2. Modify plan 3. Cancel Your choice (1-3):核心概念三Agent 配置与各参数的作用示例中的 Agent 构造如下agent CodeAgent( modelInferenceClientModel(), tools[DuckDuckGoSearchTool()], # Add a search tool for more interesting plans planning_interval5, # Plan every 5 steps for demonstration step_callbacks{PlanningStep: interrupt_after_plan}, max_steps10, verbosity_level1, # Show agent thoughts )各参数的作用与源码依据modelInferenceClientModel()使用 Hugging Face Inference Providers 作为模型后端。InferenceClientModel 默认模型为Qwen/Qwen3-Next-80B-A3B-Thinkingprovider默认为auto按用户账户中启用的供应商顺序自动选择。需要配置 Hugging Face API Token通过HF_TOKEN环境变量或token/api_key参数传入且该 token 需被授权 Make calls to the serverless Inference Providers。若模型为 gated 模型还需对 gated 仓库的读取权限。tools[DuckDuckGoSearchTool()]注册搜索工具让计划更有实际内容。DuckDuckGoSearchTool 是内置工具name web_search支持max_results默认 10与rate_limit默认每秒 1 次请求参数。planning_interval5计划调度间隔。由 agents.py 的step_number 1 or (step_number - 1) % planning_interval 0决定即首次必规划、之后每 5 步规划一次。step_callbacks{PlanningStep: interrupt_after_plan}按步骤类型注册回调是本例 HITL 的核心开关详见上文。max_steps10最大步数上限。_run_stream的while条件self.step_number max_steps会限制总步数超限时走_handle_max_steps_reached路径返回兜底答案src/smolagents/agents.py、src/smolagents/agents.py。verbosity_level1控制日志详细程度设为 1 可看到 Agent 的思考过程便于观察计划-执行的完整轨迹。核心概念四内存保留与resetFalse恢复执行reset 参数的语义Agent.run()的签名src/smolagents/agents.py中reset: bool True默认每次运行前清空记忆。源码在运行开始时self.memory.system_prompt SystemPromptStep(system_promptself.system_prompt) if reset: self.memory.reset() self.monitor.reset() self.memory.steps.append(TaskStep(taskself.task, task_imagesimages))也就是说resetTrue清空历史步骤与监控指标从零开始resetFalse保留memory.steps中全部历史TaskStep、PlanningStep、ActionStep 等只追加新的TaskStepAgent 带着完整上下文继续执行。示例中的两种运行模式# First run - may be interrupted首次运行可能被中断 result agent.run(task) # Resume with preserved memory携带保留的记忆恢复 agent.run(task, resetFalse)文档 docs/source/ko/examples/plan_customization.md 明确建议首次运行使用resetTrue默认即如此中断或修改计划后使用resetFalse恢复执行。示例的main()中还演示了如何在捕获到 interrupted 异常后询问用户是否演示恢复流程except Exception as e: if interrupted in str(e).lower(): print(\n Agent execution was cancelled by user.) print(\nTo resume execution later, you could call:) print(agent.run(task, resetFalse) # This preserves the agents memory) ... resume_choice input(\nWould you like to see resume demonstration? (y/n): ).strip().lower() if resume_choice y: print(\n Resuming execution...) try: # Resume without resetting - preserves memory agent.run(task, resetFalse)这里匹配的正是_run_stream中raise AgentError(Agent interrupted., self.logger)抛出的异常——取消执行并不会销毁已产生的步骤记忆仍然完好。核心概念五检查 Agent 记忆无论运行被中断还是成功结束都可以遍历agent.memory.steps查看到目前为止产生的全部步骤类型print(fCurrent memory contains {len(agent.memory.steps)} steps:) for i, step in enumerate(agent.memory.steps): step_type type(step).__name__ print(f {i1}. {step_type})在 smolagents 中memory.steps可能包含的步骤类型主要有TaskStep任务描述、SystemPromptStep系统提示、PlanningStep计划、ActionStep工具调用/代码执行动作与FinalAnswerStep最终答案。通过打印类型列表可以直观确认计划步骤确实已生成、用户取消后执行停在哪一步、恢复运行后新增了哪些步骤——这正是文档强调的透明性与控制权。完整 HITL 工作流综合文档与示例代码一个完整的 Human-in-the-Loop 工作流如下Agent 接收复杂任务示例任务为搜索 2024 年 AI 领域最新进展总结最重要的 3 项突破并附来源计划步骤自动生成首次运行即触发规划PlanningStep被追加到记忆并触发回调执行暂停等待人工审查回调展示计划等待用户选择人在回路用户批准继续、修改改写memory_step.plan后继续或取消agent.interrupt()下一轮抛出中断异常携带记忆恢复执行若中断可随时用agent.run(task, resetFalse)在保留全部历史步骤的基础上续跑全部步骤保留每一步都留在agent.memory.steps中供事后检查、审计与后续复用。错误处理文档与示例覆盖了三类错误场景用户取消通过捕获包含interrupted的AgentError识别并给出恢复执行指引计划修改错误get_modified_plan对空输入回退到原计划避免空计划进入执行get_user_choice对非法输入循环重试保证选择值恒为 1/2/3恢复执行失败agent.run(task, resetFalse)包在try/except中恢复失败时打印❌ Error during resume: {resume_error}不会让程序直接崩溃。运行环境与前置要求smolagents 库确保已安装当前仓库版本pip install -e .或按 pyproject.toml 安装依赖DuckDuckGoSearchTool随 smolagents 内置无需额外安装InferenceClientModel需要 Hugging Face API Token。可通过环境变量HF_TOKEN设置也可在构造时传token示例运行会发起真实网络请求请确保网络可访问 Hugging Face Inference Providers。运行方式见 examples/plan_customization/README.mdpython plan_customization.py小结能学到什么这个示例虽然代码量不大却浓缩了构建可控 Agent所需的全部关键模式step callback 实现自定义 Agent 行为按MemoryStep子类精准挂载回调理解CallbackRegistry的 MRO 派发与签名自适应调用多步 Agent 的记忆管理reset参数的真实语义、memory.steps的结构与检查方法交互式 Agent 的用户交互模式阻塞式input()三选一 多行文本编辑的完整落地动态控制计划的技术直接改写PlanningStep.plan即可让后续执行沿新计划推进交互系统的错误处理用异常传播表达用户取消并优雅地提供恢复路径。如果你想在此基础上扩展可以尝试把计划导出为 JSON 供程序化审查、为回调增加超时自动批准逻辑、或将同样的PlanningStep回调模式应用到ToolCallingAgent该参数在 CodeAgent 与 ToolCallingAgent 的构造中均已透传。完整可运行代码见 examples/plan_customization/plan_customization.py英文版讲解见 docs/source/en/examples/plan_customization.md。【免费下载链接】smolagents smolagents: a barebones library for agents that think in code.项目地址: https://gitcode.com/gh_mirrors/smo/smolagents创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表