很多人每天三件事:打开头条看新闻,打开邮箱看日报,打开数据库看业务数据。三件事做完,整理成一份简报,发到群里。日复一日,像个机器人。

这里我用 LangGraph 写了一个"定时任务 Agent",每天早上 8 点自动完成这三件事,把报告推到我微信上。从那以后,每天可以轻松很多。

LangGraph 定时任务的两种模式

LangGraph 的定时任务(Cron Jobs)有两种模式,选错了后面全是坑。

模式一:有状态(Stateful)——带记忆的定时任务

每次执行都在同一个线程里,Agent 能记住上一次执行的结果。

# 创建线程(持久化上下文)
thread = await client.threads.create()
# 在指定线程上创建定时任务
cron_job = await client.crons.create_for_thread(
thread["thread_id"],
"daily_reporter",
schedule="0 8 * * *",       # 每天早上 8 点
input={"messages": [{"role": "user", "content": "生成今日简报"}]}
)

适合场景:需要上下文的持续任务,比如"每天早上生成日报,且能记住昨天的建议"。

模式二:无状态(Stateless)——每次都是全新的

每次执行新建一个临时线程,执行完自动删除。

# 不传 thread_id,就是无状态
cron_job = await client.crons.create(
"daily_checker",
schedule="0 9 * * *",
input={"messages": [{"role": "user", "content": "检查系统状态"}]},
on_run_completed="delete"  # 执行完自动删除临时线程
)

适合场景:不需要上下文的独立任务,如定时健康检查、定时抓取数据。

生产级实战:每日简报 Agent

这是我的"每日简报 Agent"完整架构,直接抄作业:

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph_sdk import get_client
class BriefState(TypedDict):
news: str
email_summary: str
data_report: str
final_brief: str
# 三个并行节点
def fetch_news(state: BriefState) -> BriefState:
"""抓取今日热点新闻"""
return {"news": search_and_summarize()}
def fetch_emails(state: BriefState) -> BriefState:
"""汇总未读邮件"""
return {"email_summary": summarize_emails()}
def fetch_data(state: BriefState) -> BriefState:
"""拉取业务数据看板"""
return {"data_report": query_dashboard()}
def compose_brief(state: BriefState) -> BriefState:
"""LLM 整合三路信息,生成结构化简报"""
prompt = f"""
请整合以下信息,生成一份结构化简报:
## 今日新闻
{state['news']}
## 邮件摘要
{state['email_summary']}
## 数据报告
{state['data_report']}
"""
result = llm.invoke(prompt)
return {"final_brief": result.content}
# 构建图
graph = StateGraph(BriefState)
graph.add_node("news", fetch_news)
graph.add_node("emails", fetch_emails)
graph.add_node("data", fetch_data)
graph.add_node("compose", compose_brief)
# 前三路并行出发
graph.add_edge(START, "news")
graph.add_edge(START, "emails")
graph.add_edge(START, "data")
# 都跑完才汇总
graph.add_edge("news", "compose")
graph.add_edge("emails", "compose")
graph.add_edge("data", "compose")
graph.add_edge("compose", END)
# 配置持久化
app = graph.compile(checkpointer=PostgresSaver())

然后用 Cron 定时触发:

# 创建有状态的定时任务(记住每一次的简报内容)
client = get_client(url="http://localhost:8123")
thread = await client.threads.create()
cron = await client.crons.create_for_thread(
thread["thread_id"],
"daily_brief",
schedule="0 8 * * *",          # 每天早上 8 点
timezone="Asia/Shanghai",
input={
"messages": [{"role": "user", "content": "生成今日简报"}]
}
)
print(f"定时任务已创建: {cron['cron_id']}")

高级玩法:Ambient Agent(全天候 Agent)

LangGraph 的定时任务还有一个更高级的玩法——Ambient Agent,翻译过来就是"全天候 Agent"。

它不是一个"到点执行一次"的任务,而是一个持续在线、定期自检、主动推送的 Agent。

比如这个"每日热点追踪 Agent":

# 每 2 小时检查一次热点,有重大新闻才推送
cron_hot = await client.crons.create(
"hot_tracker",
schedule="0 */2 * * *",  # 每 2 小时
input={"messages": [{"role": "user", "content": "检查是否有重大新闻"}]}
)

你还可以用 langgraph-lite-cron 这个开源库,自己搭建定时任务服务,不需要用 LangGraph 的付费平台:

pip install langgraph-lite-cron

然后加到你的 FastAPI 应用里:

from langgraph_lite_cron import crons
from langgraph_lite_cron.scheduler import create_scheduler
app = FastAPI(lifespan=lifespan)
app.include_router(crons.router)

5 个值常用定时任务场景

场景 方案 Cron 表达式
每日早报推送 有状态,记住日报历史 0 8 * * *
每 2 小时热点追踪 无状态,每次独立 0 */2 * * *
每周一项目周报 有状态,跟踪进度 0 9 * * 1
每月数据清洗 无状态,执行完即删 0 0 1 * *
每分钟监控告警 无状态,轻量检查 */1 * * * *

避坑指南

1. 时区问题

Cron 表达式默认是 UTC,如果你是北京时间(UTC+8),一定要设置 timezone 参数:

cron = await client.crons.create(
"daily_check",
schedule="0 8 * * *",
timezone="Asia/Shanghai"  # 这个必填,否则默认 UTC
)

2. 任务超时

生产环境建议配置超时 + 重试:

cron = await client.crons.create(
"heavy_task",
schedule="0 9 * * *",
timeout=600,           # 10 分钟超时
retry_config={
"maxRetries": 3,
"retryDelay": 60,
"backoffMultiplier": 2
}
)

3. 并发控制

有些定时任务可能会跑超时,跟下一次触发重叠,用 maxConcurrency 防止:

cron = await client.crons.create(
"long_task",
schedule="*/30 * * * *",
maxConcurrency=1  # 上一次没跑完,下一次不触发
)

最后

Agent 的价值,不应该只停留在"你问它答"。

被动响应的 Agent,叫工具;主动服务的 Agent,才叫助手。

定时任务,就是把 Agent 从"被动工具"变成"主动助手"的关键一步。

别让你的 Agent 闲着,让它自己"上班"去。

LangGraph 定时任务是 Agent 从"被动响应"到"主动服务"的分水岭,让 Agent 自己"上班",才是真正的自动化。

学AI大模型的正确顺序,千万不要搞错了

🤔2026年AI风口已来!各行各业的AI渗透肉眼可见,超多公司要么转型做AI相关产品,要么高薪挖AI技术人才,机遇直接摆在眼前!

有往AI方向发展,或者本身有后端编程基础的朋友,直接冲AI大模型应用开发转岗超合适!

就算暂时不打算转岗,了解大模型、RAG、Prompt、Agent这些热门概念,能上手做简单项目,也绝对是求职加分王🔋

在这里插入图片描述

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料,手把手帮你快速入门!👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型(GPT、文心一言等)特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架(LangChain等)实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块,看似清晰好上手,实则每个部分都有扎实的核心内容需要吃透!

我把大模型的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~

这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费

在这里插入图片描述

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐