使用 cron 作业

在很多情况下，按计划运行助手会很有用。

例如，假设你正在构建一个每天运行并发送当日新闻邮件摘要的助手。 你可以使用 cron 作业让该助手每天 20:00 运行。

LangSmith Deployment 支持 cron 作业，可按用户定义的计划运行。用户指定计划、助手和一些输入。之后，服务器会按指定计划执行以下操作：

• 使用指定助手创建新线程
• 将指定输入发送到该线程

请注意，这每次都会向线程发送相同的输入。

LangSmith Deployment API 提供了多个用于创建和管理 cron 作业的端点。详情请参阅 API 参考。

有时你不想基于用户交互运行图，而是希望按计划运行图——例如，你希望图每周为团队撰写并发送待办事项邮件。LangSmith Deployment 允许你使用 Crons 客户端实现这一点，而无需编写自己的脚本。要计划图作业，你需要传入一个 cron 表达式，告知客户端何时运行图。Cron 作业在后台运行，不会干扰图的正常调用。

NOTE

所有 cron 计划都按 UTC 解释。指定计划时，请确保将所需执行时间转换为 UTC。

设置

首先，让我们设置 SDK 客户端、助手和线程：

from langgraph_sdk import get_client

client = get_client(url=<DEPLOYMENT_URL>)
# Using the graph deployed with the name "agent"
assistant_id = "agent"
# create thread
thread = await client.threads.create()
print(thread)

输出：

{
'thread_id': '9dde5490-2b67-47c8-aa14-4bfec88af217',
'created_at': '2024-08-30T23:07:38.242730+00:00',
'updated_at': '2024-08-30T23:07:38.242730+00:00',
'metadata': {},
'status': 'idle',
'config': {},
'values': None
}

线程上的 cron 作业

要创建与特定线程关联的 cron 作业，可以这样写：

# This schedules a job to run at 15:27 (3:27PM) UTC every day
cron_job = await client.crons.create_for_thread(
    thread["thread_id"],
    assistant_id,
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "What time is it?"}]},
)

请注意，删除不再有用的 Cron 作业非常重要。否则你可能会产生不必要的 LLM API 费用！你可以使用以下代码删除 Cron 作业：

await client.crons.delete(cron_job["cron_id"])

无状态 cron 作业

你也可以使用以下代码创建无状态 cron 作业。无状态 cron 作业会为每次执行创建一个新线程：

# This schedules a job to run at 15:27 (3:27PM) UTC every day
cron_job_stateless = await client.crons.create(
    assistant_id,
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "What time is it?"}]},
)

再次提醒，完成后请记得删除作业！

await client.crons.delete(cron_job_stateless["cron_id"])

无状态 cron 的线程清理

NOTE

此功能需要 LangGraph API 版本 0.5.18 或更高，以及 Python SDK 0.3.2 或更高，或 JavaScript SDK 1.4.0 或更高。

每次触发无状态 cron 时，都会创建一个新线程。使用 on_run_completed 参数控制运行完成后该线程的处理方式：

• "delete"（默认）：运行完成后自动删除线程。
• "keep"：保留线程以便稍后检索。你需要负责清理这些线程。有关推荐方法，请参阅如何为应用添加 TTL。

示例：保留线程以便稍后检索

# Create a stateless cron that keeps threads after execution.
# Configure checkpointer.ttl in langgraph.json to auto-delete old threads.
# See: https://docs.langchain.com/langsmith/configure-ttl
cron_job = await client.crons.create(
    assistant_id,
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "Daily report"}]},
    on_run_completed="keep"
)

# You can later retrieve the runs and their results
runs = await client.runs.search(
    metadata={"cron_id": cron_job["cron_id"]}
)

---