第 4 章 · 02 planning_interval 规划与序列化

第 4 章 · 02 planning_interval 规划与序列化

本节摘要:本节讲两件事:一是 planning_interval 参数——让 agent 每 N 步插入一个 PlanningStep,由 LLM 重新梳理「已知事实/待查事实/下一步行动」,这是 human-in-the-loop(人工审批与修改计划)的基础;二是序列化与分享——write_memory_to_messages 的 summary_mode 视图、serialization.py 的 SafeSerializer(JSON 优先、pickle 可选回退并伴随安全警告)、agent 的 save/load 与 push_to_hub(把工具与提示词完整打包上传 HF Hub)。

内容来源:src/smolagents/agents.py、src/smolagents/serialization.py、src/smolagents/prompts/code_agent.yaml、examples/plan_customization/plan_customization.py

⚠️ 注意:serialization 模块的文档字符串开宗明义——pickle 反序列化可以执行任意代码,该模块默认只用安全 JSON;只有在你完全信任执行环境时才可开启 pickle 回退。不要加载来历不明的 agent 文件。

学习目标

1. 理解 planning_interval 的触发时机与规划提示词结构(facts survey + plan)。
2. 读懂 plan_customization 示例中「打断—人工改计划—续跑」的实现。
3. 掌握 SafeSerializer 的类型标记方案与 pickle 安全警告。
4. 了解 agent.save 的产物目录结构与 push_to_hub 的打包内容。

一、planning_interval:每 N 步停下来重新规划

构造 agent 时传入 planning_interval=N,运行循环就会在特定时机插入规划步。agents.py 中的触发条件:

if self.planning_interval is not None and (
    self.step_number == 1 or (self.step_number - 1) % self.planning_interval == 0
):
    planning_start_time = time.time()
    for element in self._generate_planning_step(
        task, is_first_step=len(self.memory.steps) == 1, step=self.step_number
    ):
        yield element
    ...
    self.memory.steps.append(planning_step)
​

解读:第 1 步必规划(开局先出计划),之后每 N 步再规划一次(planning_interval=5 意味着第 1、6、11…步先规划再行动)。生成的 PlanningStep 追加进记忆——上一节看到它在 summary_mode 下不进消息、正常模式下以「计划 + Now proceed and carry out this plan.」两条消息呈现。

1.1 规划提示词:事实调查表 + 行动计划

prompts/code_agent.yaml 中的 initial_plan 模板规定了 LLM 的规划格式,核心是两部分:

planning:
  initial_plan : |-
    You are a world expert at analyzing a situation to derive facts, and plan accordingly towards solving a task.
    You will need to 1. build a survey of facts known or needed to solve the task, then 2. make a plan of action.
    ## 1. Facts survey
    ### 1.1. Facts given in the task
    ### 1.2. Facts to look up
    List here any facts that we may need to look up.  # 还要注明去哪查:网站、文件……
    ### 1.3. Facts to derive
    List here anything that we want to derive by logical reasoning, for instance computation or simulation.
    ## 2. Plan
    Then develop a step-by-step high-level plan ...  # 强制只写高层计划,不细化到具体工具调用
    After writing the final step of the plan, write the '<end_plan>' tag and stop there.
​

「Facts survey」分三类——任务给定的 facts、需要查的 facts(注明去哪查)、需要推导的 facts,这是把「先审题再动笔」工程化;「Plan」部分强制只写高层计划、不细化到具体工具调用,以 <end_plan> 标签收尾供解析。更新计划的 update_plan 消息还会注入 remaining_steps(还剩几步可用的预算):

plan_update_post = ChatMessage(
    role=MessageRole.USER,
    content=[{"type": "text",
        "text": populate_template(
            self.prompt_templates["planning"]["update_plan_post_messages"],
            variables={"task": task, "tools": self.tools,
                       "managed_agents": self.managed_agents,
                       "remaining_steps": (self.max_steps - step)},
        )}],
)
​

注意生成更新计划时的输入构造——第 4.1 节埋的伏笔在这里兑现:

# Summary mode removes the system prompt and previous planning messages output by the model.
# Removing previous planning messages avoids influencing too much the new plan.
memory_messages = self.write_memory_to_messages(summary_mode=True)
input_messages = [plan_update_pre] + memory_messages + [plan_update_post]
​

write_memory_to_messages(summary_mode=True) 只保留任务与动作史,剔除系统提示与旧计划——旧计划已过时,留在上下文里只会让新计划「抄旧作业」。

1.2 plan_customization:人工审批计划(human-in-the-loop)

examples/plan_customization/plan_customization.py 演示了三步玩法:注册 PlanningStep 回调 → 在回调里打断 agent 展示计划 → 用户批准/修改后用 reset=False 续跑。核心回调:

def interrupt_after_plan(memory_step, agent):
    """Step callback that interrupts the agent after a planning step is created."""
    if isinstance(memory_step, PlanningStep):
        display_plan(memory_step.plan)        # 展示 LLM 生成的计划
        choice = get_user_choice()            # 1 批准 / 2 修改 / 3 取消
        if choice == 1:
            return                            # 直接放行
        elif choice == 2:
            memory_step.plan = get_modified_plan(memory_step.plan)  # 人为改写计划文本
        else:
            agent.interrupt()                 # 拉起中断开关
​

注册与续跑:

agent = CodeAgent(tools=[DuckDuckGoSearchTool()], model=model, planning_interval=1)
agent.step_callbacks.append(interrupt_after_plan)
agent.run(task)                    # 第一次运行在计划后被打断
...
agent.run(task, reset=False)       # 记忆保留,从中断处继续
​

这就是上一节 CallbackRegistry 的实战:回调收到 memory_step(还带 agent 引用),直接改写 memory_step.plan——因为下一步的输入消息是从记忆实时渲染的,改了计划文本就等于改了 agent 接下来「心里装着的计划」。不需要任何私有 API。

二、serialization.py:JSON 优先、pickle 兜底的安全序列化

远程执行器要把变量跨进程/跨机器传给沙箱,序列化是刚需。serialization.py 的模块文档开宗明义:「Pickle deserialization can execute arbitrary code. This module defaults to safe JSON-only serialization. Only enable pickle fallback ... if you fully trust the execution environment.」——能用 JSON 绝不用 pickle,pickle 只作可选回退。

SafeSerializer 用「类型标记」把 Python 对象映射成 JSON 可表达的形状,to_json_safe 的分层处理(节选):

@staticmethod
def to_json_safe(obj: Any) -> Any:
    obj_type = type(obj)
    if obj_type is str or obj_type is int or obj_type is float or obj_type is bool or obj is None:
        return obj
    if obj_type is list:
        return [SafeSerializer.to_json_safe(item) for item in obj]
    if obj_type is tuple:
        return {"__type__": "tuple", "data": [SafeSerializer.to_json_safe(item) for item in obj]}
    if obj_type is bytes:
        return {"__type__": "bytes", "data": base64.b64encode(obj).decode()}
    if obj_type is complex:
        return {"__type__": "complex", "real": obj.real, "imag": obj.imag}
    # datetime/Decimal/Path/PIL.Image/numpy.ndarray/dataclass ... 同样打标记
    ...
    raise SerializationError(f"Cannot safely serialize object of type {type_name}")
​

支持范围:基础类型(str/int/float/bool/None/list/dict)+ 扩展类型(tuple/set/frozenset/bytes/complex/datetime 系)+ 可选类型(numpy 数组、PIL 图片、dataclass、Decimal、Path——用缓存惰性导入,没装就跳过)。PIL 图片转 PNG 字节再 base64,numpy 数组转 tolist() 加 dtype 字符串——都是纯数据,反序列化时不执行任何代码。遇到不认识的类型直接抛 SerializationError,绝不静默。

dumps/loads 用前缀区分格式并落实安全策略:

@staticmethod
def dumps(obj: Any, allow_pickle: bool = False) -> str:
    if not allow_pickle:
        json_safe = SafeSerializer.to_json_safe(obj)  # Raises SerializationError if fails
        return SafeSerializer.SAFE_PREFIX + json.dumps(json_safe)
    else:
        try:
            json_safe = SafeSerializer.to_json_safe(obj)
            return SafeSerializer.SAFE_PREFIX + json.dumps(json_safe)
        except SerializationError:
            import warnings
            warnings.warn(
                "Falling back to insecure pickle serialization. "
                "This is a security risk and will be removed in a future version. ...",
                FutureWarning, stacklevel=2,
            )
            return "pickle:" + base64.b64encode(pickle.dumps(obj)).decode()
​

"safe:" 前缀走 JSON,"pickle:" 前缀走 pickle。关键是 loads 侧:默认 allow_pickle=False,收到 pickle 数据直接拒绝——「反序列化即执行」是 pickle 的根本问题,防线必须设在读入端。这个模块同时服务于第 6 章的远程沙箱:工具变量传给沙箱、最终答案传回主进程,都走这一套。

三、agent 的 save/load 与 push_to_hub

agent.save(output_dir) 落盘的不只是配置,而是完整可复活的工程目录:

# - a `tools` folder containing the logic for each of the tools under `tools/{tool_name}.py`.
# - a `managed_agents` folder containing the logic for each of the managed agents.
# - an `agent.json` file containing a dictionary representing your agent.
# - a `prompt.yaml` file containing the prompt templates used by your agent.
# - an `app.py` file providing a UI for your agent when it is exported to a Space with `agent.push_to_hub()`
# - a `requirements.txt` containing the names of the modules used by your tool
​

实现要点:每个工具经 validate_tool_attributes 静态校验(保证代码自包含、可重建)后用 instance_to_source 还原成源码;提示词模板用 yaml 块字面量原样导出;agent.json 记录类名、模型配置、工具清单;app.py 由 Jinja2 模板渲染出一个 Gradio 界面。load 反向执行:读 agent.json、从 tools 目录重建工具类、恢复提示词模板。注意 save 的日志提示——step_callbacks 与 final_answer_checks 这类函数无法序列化,会被忽略。

push_to_hub 把上面整个目录上传到 HF Hub 并创建为一个 Space:

def push_to_hub(self, repo_id: str, ...) -> str:
    repo_url = create_repo(repo_id=repo_id, token=token, private=private,
                           exist_ok=True, repo_type="space", space_sdk="gradio")
    metadata_update(repo_id, {"tags": ["smolagents", "agent"]}, ...)
    with tempfile.TemporaryDirectory() as work_dir:
        self.save(work_dir)
        return upload_folder(repo_id=repo_id, folder_path=work_dir, ...)
​

打包内容 = 工具源码 + 提示词模板 + 模型配置 + 依赖清单 + Gradio 界面——别人从 Hub 拉下来 agent = CodeAgent.from_hub(repo_id, trust_remote_code=True) 即可复现你的 agent(trust_remote_code 的强调正是第 6.2 节安全话题的前奏:拉取即执行远端代码)。记忆本身(steps)不在打包范围——分享的是「能力」,不是「经历」。

💡 阶梯要点:本阶拿到「规划」与「分享」两件装备。规划的本质是给 LLM 一个强制复盘节奏:planning_interval 定期触发、summary_mode 剔除旧计划防干扰、回调机制打开人工介入的口子。序列化则处处体现「安全默认」:JSON 优先、pickle 拒之门外、分享需显式信任。

本节要点回顾

• planning_interval:第 1 步必规划,之后每 N 步触发;规划提示词分「事实调查(给定/待查/待推导)+ 高层计划」两段,以 <end_plan> 结束。
• 重新规划用 write_memory_to_messages(summary_mode=True)——剔除系统提示与旧计划,只看动作史。
• plan_customization:PlanningStep 回调里改 memory_step.plan,run(task, reset=False) 保留记忆续跑,实现 human-in-the-loop。
• SafeSerializer:类型标记法转 JSON(tuple/bytes/PIL/numpy 等都能表达),safe:/pickle: 前缀区分格式,loads 默认拒绝 pickle——反序列化即任意代码执行。
• agent.save 落盘 tools/、managed_agents/、agent.json、prompts.yaml、app.py、requirements.txt;push_to_hub 上传整个目录为 HF Space;记忆不随包分享。

下一节:进入第 5 章「工具体系」——01 Tool 基类与 @tool 装饰器,看 smolagents 如何用最少的抽象把普通函数变成 agent 的手和脚。