如何用 Managed Deep Agents 的 Schedules SDK 构建主动常驻智能体

听说过巴德-迈因霍夫现象吗？就是你第一次学到一个词，然后突然发现到处都能听到它。我认为同样的原理也适用于软件工程：你了解到一个简单但有效的设计模式，然后突然发现它无处不在。

我们在主动式智能体上看到了这种情况，我们注意到几乎所有你喜爱的新兴主动式智能体系统都由一个非常简单的机制驱动：智能体作为 cron 任务运行，利用一个已有五十多年历史的操作系统原理，服务于一个非常现代的用例。

未来属于主动式、始终在线、环境感知的智能体。Muse、Instinct、Dots、Claude Duet、Hone 等产品的发布都反映了这一点。

Cron 驱动了几种不同的能力，它们结合起来使这些智能体具备主动性、个性化以及环境感知——也就是说，它们可以自行执行任务，无需人工指示。

在本博客中，我们将介绍 cron 解锁的各项能力、构建此类智能体所需的两个核心底层部分，然后展示如何在Managed Deep Agents 中实现这些。

我们已经看到了 cron 能带来的影响。主动性已成为智能体行为的新默认，因为它解锁了大量新功能。LangSmith 上超过 70% 的 Managed Deepagent 运行来自定时运行，而非人工输入消息。

Cron 解锁了什么？

Cron 虽然概念简单，却解锁了几种不同的行为，塑造了这些主动式、始终在线的智能体。

记忆：所有这些智能体的一个显著特点是，它们能在更长的时间跨度内记住关于你的事实。实现这一点部分涉及运行一个后台任务，将之前的对话浓缩为更密集的记忆形式。这俗称“做梦”。在底层，这只是一个按某种节奏（例如每天一次）运行的 cron，通过反思对话中的收获来更新长期记忆。

主动性：这些智能体之所以有效，部分在于它们会主动就某个它们认为你可能感兴趣的想法发起沟通。在底层，其运作方式是存在某种关于该智能体应该完成什么的“想法”。

对于通用智能体，这可能简单到只是做一个好助手：采取措施预判你接下来需要什么，比如标记即将到来的截止日期、跟进一个未完结的话题，或推送与你工作相关的新闻。对于专用智能体，比如 Hone，智能体在朝着一个明确的目标努力。然后 cron 按某种节奏（每小时一次）运行，提示智能体探索新的行动以推进其目标。这些行动随后会以主动消息的形式呈现给用户。

智能分诊：世界上的噪音太多了。始终在线智能体的一个重要用例是监控信息源（Slack、邮件等），并向用户推送相关消息。在底层，这可以实现为一个按某种节奏（每十分钟一次）运行的 cron，提示智能体查看事件流，并将任何值得注意的内容通知用户。

跟进提醒：Cron 不仅仅用于后台。用户常常可能要求在某时提醒他们某件事。Cron 可以用来设定一个时间表，到点唤醒并提醒用户。

编排：Agent 集群可以由单个编排 agent 管理，该 agent 具备在未来自我唤醒的能力。编排 agent 可以通过 cron 监控整个集群。它可以在等待异步子任务完成时休眠，并按固定间隔自我唤醒以检查进度。

构建主动型 agent 的两个关键部分

为了构建这类 agent，我们发现通常有两个关键部分。

第一是一个健壮的 cron 服务器。你需要一个能让开发者轻松添加 cron 的服务器。开发者应当能够按所需的任意调度添加 cron，并附带一个要作为该调度一部分运行的 prompt。如果 agent 在运行时需要特定技能或权限，这些也应一并包含。如果在某个特定时间点有大量运行，服务器应稍作“抖动”以分散负载。

第二，你需要一个对 agent 友好的接口。对于跟进之类的操作，agent 需要能够与这些提醒进行交互。它需要一种简单但受管控的方式来设置和修改提醒。

用 MDA 构建

Managed Deep Agents——通过 LangSmith Deployments 托管——提供内置的 cron 服务器，可动态触发 agent 运行。定时运行看似简单：它只是传入的 prompt，按调度触发，其线程和记忆基于创建该调度的主体。上述工具包为仅接受 prompt 和 cron 调度的工具提供了基础。

使用 Managed Deep Agents，调度可以由 agent 静态或动态定义。

例如，将固定调度作为核心设计的一部分来构建 agent——比如每日销售播报——只需在 schedules/ 中放一个文件：

# schedules/daily_digest.py
from managed_deepagents import define_schedule

schedule = define_schedule(
    cron="0 8 * * 1-5",
    timezone="America/Los_Angeles",
    prompt="Inspect the sales report using your tools and send a KPI readout.",
)

Managed Deep Agents 还提供动态调度，最终用户可以提示你的 agent 安排你在开发时未曾预料到的周期性任务。为支持这一点，将 Schedules SDK 封装在工具中：

# tools/scheduling.py
from langchain.tools import tool
from managed_deepagents import ManagedDeepAgentRuntime


@tool
async def set_schedule(prompt: str, cron: str, runtime: ManagedDeepAgentRuntime) -> str:
    """Run a prompt for the current user on a cron schedule."""
    item = await runtime.schedules.create(owner={"type": "user"}, cron=cron, prompt=prompt)
    return f"Created schedule {item['id']}."


@tool
async def list_schedules(runtime: ManagedDeepAgentRuntime) -> str:
    """List the current user's schedules."""
    page = await runtime.schedules.list(owner={"type": "user"})
    return str(page["items"])


@tool
async def delete_schedule(schedule_id: str, runtime: ManagedDeepAgentRuntime) -> str:
    """Delete one of the current user's schedules."""
    await runtime.schedules.delete(schedule_id, owner={"type": "user"})
    return f"Deleted schedule {schedule_id}."

‍开发者需要灵活性，Managed Deep Agents 的工具系统正是为此而设计。不同的 agent 类型可能需要暴露不同的调度工具包。在某些情况下，一个包装一次性运行的基础 remind_me 或一个 check_in_later 可能就足够了：

from datetime import datetime, timedelta, timezone

from langchain.tools import tool
from managed_deepagents import ManagedDeepAgentRuntime


@tool
async def check_in_later(prompt: str, minutes: int, runtime: ManagedDeepAgentRuntime) -> str:
    """Run a prompt once for the current user, a number of minutes from now."""
    at = datetime.now(timezone.utc) + timedelta(minutes=minutes)
    item = await runtime.schedules.create(owner={"type": "user"}, at=at, prompt=prompt)
    return f"Scheduled a check-in {item['id']}."

从我们的文档开始为你的 Managed Deep Agents 添加调度：langch.in/mda。你距离在组织内分发一个主动型 agent 只差一个 mda init 和一个 mda deploy。

‍