feat: 实现 Agent 定时任务系统 — 按 cron 表达式周期执行 Agent
- 新增 AgentSchedule 模型、CRUD API、调度服务 - 集成 Celery Beat 每分钟检查到期任务并自动触发执行 - 支持手动触发、cron 表达式解析、执行状态跟踪 - 依赖: croniter (cron 表达式解析) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
renjianbo
194
backend/app/api/agent_schedules.py
Normal file
194
backend/app/api/agent_schedules.py
Normal file
@@ -0,0 +1,194 @@
|
||||
"""Agent 定时任务 CRUD API"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.auth import get_current_user
|
||||
from app.core.database import get_db
|
||||
from app.models.agent import Agent
|
||||
from app.models.agent_schedule import AgentSchedule
|
||||
from app.models.user import User
|
||||
from app.services.agent_schedule_service import (
|
||||
compute_next_run,
|
||||
create_execution_for_schedule,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/v1/agent-schedules", tags=["agent-schedules"])
|
||||
|
||||
|
||||
# ─── Pydantic Schemas ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class ScheduleCreate(BaseModel):
|
||||
agent_id: str
|
||||
name: str = Field(..., max_length=100)
|
||||
cron_expression: str = Field(..., description="标准 5 位 cron，如 0 9 * * *")
|
||||
input_message: str = Field(..., description="每次触发时发给 Agent 的消息")
|
||||
timezone: str = "Asia/Shanghai"
|
||||
|
||||
|
||||
class ScheduleUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
cron_expression: Optional[str] = None
|
||||
input_message: Optional[str] = None
|
||||
timezone: Optional[str] = None
|
||||
enabled: Optional[bool] = None
|
||||
|
||||
|
||||
class ScheduleResponse(BaseModel):
|
||||
id: str
|
||||
agent_id: str
|
||||
name: str
|
||||
cron_expression: str
|
||||
input_message: str
|
||||
timezone: str
|
||||
enabled: bool
|
||||
last_run_at: Optional[datetime] = None
|
||||
last_run_status: Optional[str] = None
|
||||
next_run_at: datetime
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ─── API Endpoints ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("", response_model=List[ScheduleResponse])
|
||||
async def list_schedules(
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""获取当前用户的所有定时任务。"""
|
||||
schedules = (
|
||||
db.query(AgentSchedule)
|
||||
.filter(AgentSchedule.user_id == current_user.id)
|
||||
.order_by(AgentSchedule.created_at.desc())
|
||||
.all()
|
||||
)
|
||||
return schedules
|
||||
|
||||
|
||||
@router.post("", response_model=ScheduleResponse, status_code=201)
|
||||
async def create_schedule(
|
||||
data: ScheduleCreate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""创建定时任务。"""
|
||||
# 验证 Agent 存在
|
||||
agent = db.query(Agent).filter(Agent.id == data.agent_id).first()
|
||||
if not agent:
|
||||
raise HTTPException(status_code=404, detail="Agent 不存在")
|
||||
if agent.user_id and agent.user_id != current_user.id and current_user.role != "admin":
|
||||
raise HTTPException(status_code=403, detail="无权使用该 Agent")
|
||||
|
||||
# 验证 cron 表达式
|
||||
try:
|
||||
next_run = compute_next_run(data.cron_expression)
|
||||
except (ValueError, KeyError) as e:
|
||||
raise HTTPException(status_code=400, detail=f"cron 表达式无效: {e}")
|
||||
|
||||
schedule = AgentSchedule(
|
||||
agent_id=data.agent_id,
|
||||
name=data.name,
|
||||
cron_expression=data.cron_expression,
|
||||
input_message=data.input_message,
|
||||
timezone=data.timezone or "Asia/Shanghai",
|
||||
enabled=True,
|
||||
next_run_at=next_run,
|
||||
user_id=current_user.id,
|
||||
)
|
||||
db.add(schedule)
|
||||
db.commit()
|
||||
db.refresh(schedule)
|
||||
logger.info(
|
||||
"定时任务创建: user=%s agent=%s name=%s cron=%s next_run=%s",
|
||||
current_user.id, data.agent_id, data.name, data.cron_expression, next_run,
|
||||
)
|
||||
return schedule
|
||||
|
||||
|
||||
@router.put("/{schedule_id}", response_model=ScheduleResponse)
|
||||
async def update_schedule(
|
||||
schedule_id: str,
|
||||
data: ScheduleUpdate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""更新定时任务配置。"""
|
||||
schedule = db.query(AgentSchedule).filter(AgentSchedule.id == schedule_id).first()
|
||||
if not schedule:
|
||||
raise HTTPException(status_code=404, detail="定时任务不存在")
|
||||
if schedule.user_id != current_user.id and current_user.role != "admin":
|
||||
raise HTTPException(status_code=403, detail="无权修改该定时任务")
|
||||
|
||||
if data.name is not None:
|
||||
schedule.name = data.name
|
||||
if data.cron_expression is not None:
|
||||
try:
|
||||
schedule.next_run_at = compute_next_run(data.cron_expression, after=datetime.utcnow())
|
||||
schedule.cron_expression = data.cron_expression
|
||||
except (ValueError, KeyError) as e:
|
||||
raise HTTPException(status_code=400, detail=f"cron 表达式无效: {e}")
|
||||
if data.input_message is not None:
|
||||
schedule.input_message = data.input_message
|
||||
if data.timezone is not None:
|
||||
schedule.timezone = data.timezone
|
||||
if data.enabled is not None:
|
||||
schedule.enabled = data.enabled
|
||||
|
||||
schedule.updated_at = datetime.utcnow()
|
||||
db.commit()
|
||||
db.refresh(schedule)
|
||||
return schedule
|
||||
|
||||
|
||||
@router.delete("/{schedule_id}")
|
||||
async def delete_schedule(
|
||||
schedule_id: str,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""删除定时任务。"""
|
||||
schedule = db.query(AgentSchedule).filter(AgentSchedule.id == schedule_id).first()
|
||||
if not schedule:
|
||||
raise HTTPException(status_code=404, detail="定时任务不存在")
|
||||
if schedule.user_id != current_user.id and current_user.role != "admin":
|
||||
raise HTTPException(status_code=403, detail="无权删除该定时任务")
|
||||
|
||||
db.delete(schedule)
|
||||
db.commit()
|
||||
return {"message": "定时任务已删除"}
|
||||
|
||||
|
||||
@router.post("/{schedule_id}/trigger")
|
||||
async def trigger_schedule(
|
||||
schedule_id: str,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""手动触发一次定时任务。"""
|
||||
schedule = db.query(AgentSchedule).filter(AgentSchedule.id == schedule_id).first()
|
||||
if not schedule:
|
||||
raise HTTPException(status_code=404, detail="定时任务不存在")
|
||||
if schedule.user_id != current_user.id and current_user.role != "admin":
|
||||
raise HTTPException(status_code=403, detail="无权触发该定时任务")
|
||||
|
||||
execution_id = create_execution_for_schedule(db, schedule)
|
||||
if not execution_id:
|
||||
raise HTTPException(status_code=500, detail="触发执行失败")
|
||||
|
||||
return {
|
||||
"message": "定时任务已触发",
|
||||
"execution_id": execution_id,
|
||||
}
|
||||
@@ -2,13 +2,18 @@
|
||||
Celery 应用配置
|
||||
"""
|
||||
from celery import Celery
|
||||
from celery.schedules import crontab
|
||||
from app.core.config import settings
|
||||
|
||||
celery_app = Celery(
|
||||
"aiagent",
|
||||
broker=settings.REDIS_URL,
|
||||
backend=settings.REDIS_URL,
|
||||
include=["app.tasks.workflow_tasks", "app.tasks.agent_tasks"]
|
||||
include=[
|
||||
"app.tasks.workflow_tasks",
|
||||
"app.tasks.agent_tasks",
|
||||
"app.tasks.scheduler_tasks",
|
||||
]
|
||||
)
|
||||
|
||||
celery_app.conf.update(
|
||||
@@ -21,3 +26,11 @@ celery_app.conf.update(
|
||||
task_time_limit=30 * 60, # 30分钟
|
||||
task_soft_time_limit=25 * 60, # 25分钟
|
||||
)
|
||||
|
||||
# Celery Beat 定时调度配置
|
||||
celery_app.conf.beat_schedule = {
|
||||
"check-agent-schedules-every-minute": {
|
||||
"task": "app.tasks.scheduler_tasks.check_agent_schedules_task",
|
||||
"schedule": crontab(minute="*"), # 每分钟检查
|
||||
},
|
||||
}
|
||||
|
||||
@@ -49,5 +49,6 @@ def init_db():
|
||||
import app.models.agent_llm_log
|
||||
import app.models.agent_vector_memory
|
||||
import app.models.agent_learning_pattern
|
||||
import app.models.agent_schedule
|
||||
import app.models.knowledge_base
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
@@ -213,7 +213,7 @@ async def startup_event():
|
||||
logger.error(f"自定义工具加载失败: {e}")
|
||||
|
||||
# 注册路由
|
||||
from app.api import auth, uploads, workflows, executions, websocket, execution_logs, data_sources, agents, platform_templates, model_configs, webhooks, template_market, batch_operations, collaboration, permissions, monitoring, alert_rules, node_test, node_templates, tools, agent_chat, agent_monitoring, knowledge_base
|
||||
from app.api import auth, uploads, workflows, executions, websocket, execution_logs, data_sources, agents, platform_templates, model_configs, webhooks, template_market, batch_operations, collaboration, permissions, monitoring, alert_rules, node_test, node_templates, tools, agent_chat, agent_monitoring, knowledge_base, agent_schedules
|
||||
|
||||
app.include_router(auth.router)
|
||||
app.include_router(uploads.router)
|
||||
@@ -238,6 +238,7 @@ app.include_router(tools.router)
|
||||
app.include_router(agent_chat.router)
|
||||
app.include_router(agent_monitoring.router)
|
||||
app.include_router(knowledge_base.router)
|
||||
app.include_router(agent_schedules.router)
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
@@ -15,6 +15,7 @@ from app.models.persistent_user_memory import PersistentUserMemory
|
||||
from app.models.agent_llm_log import AgentLLMLog
|
||||
from app.models.agent_vector_memory import AgentVectorMemory
|
||||
from app.models.agent_learning_pattern import AgentLearningPattern
|
||||
from app.models.agent_schedule import AgentSchedule
|
||||
from app.models.knowledge_base import KnowledgeBase, Document, DocumentChunk
|
||||
|
||||
__all__ = ["User", "Workflow", "WorkflowVersion", "Agent", "Execution", "ExecutionLog", "ModelConfig", "DataSource", "WorkflowTemplate", "TemplateRating", "TemplateFavorite", "NodeTemplate", "Role", "Permission", "WorkflowPermission", "AgentPermission", "AlertRule", "AlertLog", "PersistentUserMemory", "AgentLLMLog", "AgentVectorMemory", "AgentLearningPattern", "KnowledgeBase", "Document", "DocumentChunk"]
|
||||
__all__ = ["User", "Workflow", "WorkflowVersion", "Agent", "Execution", "ExecutionLog", "ModelConfig", "DataSource", "WorkflowTemplate", "TemplateRating", "TemplateFavorite", "NodeTemplate", "Role", "Permission", "WorkflowPermission", "AgentPermission", "AlertRule", "AlertLog", "PersistentUserMemory", "AgentLLMLog", "AgentVectorMemory", "AgentLearningPattern", "AgentSchedule", "KnowledgeBase", "Document", "DocumentChunk"]
|
||||
28
backend/app/models/agent_schedule.py
Normal file
28
backend/app/models/agent_schedule.py
Normal file
@@ -0,0 +1,28 @@
|
||||
"""Agent 定时任务表：按 cron 表达式周期执行 Agent"""
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from sqlalchemy import Column, String, Text, Integer, DateTime, ForeignKey, Boolean
|
||||
from sqlalchemy.dialects.mysql import CHAR
|
||||
from app.core.database import Base
|
||||
|
||||
|
||||
class AgentSchedule(Base):
|
||||
"""Agent 定时任务 — 按 cron 表达式周期执行指定 Agent"""
|
||||
__tablename__ = "agent_schedules"
|
||||
|
||||
id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()))
|
||||
agent_id = Column(CHAR(36), ForeignKey("agents.id"), nullable=False, index=True, comment="关联 Agent ID")
|
||||
name = Column(String(100), nullable=False, comment="任务名称")
|
||||
cron_expression = Column(String(100), nullable=False, comment="cron 表达式，如 0 9 * * *")
|
||||
input_message = Column(Text, nullable=False, comment="定时执行时发送的消息内容")
|
||||
timezone = Column(String(64), default="Asia/Shanghai", comment="时区")
|
||||
enabled = Column(Boolean, default=True, comment="是否启用")
|
||||
last_run_at = Column(DateTime, nullable=True, comment="上次执行时间")
|
||||
last_run_status = Column(String(32), nullable=True, comment="上次执行状态: success/failed")
|
||||
next_run_at = Column(DateTime, nullable=False, comment="下次执行时间")
|
||||
user_id = Column(CHAR(36), ForeignKey("users.id"), nullable=False, index=True, comment="创建者 ID")
|
||||
created_at = Column(DateTime, default=datetime.utcnow, comment="创建时间")
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, comment="更新时间")
|
||||
|
||||
def __repr__(self):
|
||||
return f"<AgentSchedule(id={self.id}, name={self.name}, cron={self.cron_expression})>"
|
||||
140
backend/app/services/agent_schedule_service.py
Normal file
140
backend/app/services/agent_schedule_service.py
Normal file
@@ -0,0 +1,140 @@
|
||||
"""Agent 定时任务服务：cron 解析、执行触发、下次执行时间计算"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional
|
||||
from croniter import croniter
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.database import SessionLocal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def compute_next_run(cron_expression: str, after: Optional[datetime] = None) -> datetime:
|
||||
"""根据 cron 表达式计算下一次执行时间。
|
||||
|
||||
Args:
|
||||
cron_expression: 标准 5 位 cron 表达式，如 "0 9 * * *"
|
||||
after: 从哪个时间点开始计算，默认为当前 UTC 时间
|
||||
|
||||
Returns:
|
||||
下次执行的 UTC datetime
|
||||
"""
|
||||
base = after or datetime.now(timezone.utc)
|
||||
# croniter 需要以 datetime 对象作为基准
|
||||
base_naive = base.replace(tzinfo=None) if base.tzinfo else base
|
||||
cron = croniter(cron_expression, base_naive)
|
||||
next_dt = cron.get_next(datetime)
|
||||
return next_dt
|
||||
|
||||
|
||||
def create_execution_for_schedule(db: Session, schedule) -> Optional[str]:
|
||||
"""为定时任务创建 Execution 记录并投递 Celery 任务。
|
||||
|
||||
Args:
|
||||
db: 数据库会话
|
||||
schedule: AgentSchedule ORM 对象
|
||||
|
||||
Returns:
|
||||
创建的 execution_id，失败返回 None
|
||||
"""
|
||||
from app.models.execution import Execution
|
||||
from app.models.agent import Agent
|
||||
|
||||
agent = db.query(Agent).filter(Agent.id == schedule.agent_id).first()
|
||||
if not agent:
|
||||
logger.warning("定时任务 %s 关联的 Agent %s 不存在", schedule.id, schedule.agent_id)
|
||||
return None
|
||||
if not agent.workflow_config:
|
||||
logger.warning("Agent %s 缺少 workflow_config，无法执行定时任务", schedule.agent_id)
|
||||
return None
|
||||
|
||||
# 创建执行记录
|
||||
execution = Execution(
|
||||
agent_id=schedule.agent_id,
|
||||
input_data={"message": schedule.input_message},
|
||||
status="pending",
|
||||
)
|
||||
db.add(execution)
|
||||
db.flush() # 获取 id
|
||||
|
||||
# 投递到 Celery
|
||||
from app.tasks.workflow_tasks import execute_workflow_task
|
||||
|
||||
try:
|
||||
task = execute_workflow_task.delay(
|
||||
str(execution.id),
|
||||
f"agent_{schedule.agent_id}",
|
||||
agent.workflow_config,
|
||||
{"message": schedule.input_message},
|
||||
)
|
||||
execution.task_id = task.id
|
||||
execution.status = "running"
|
||||
db.commit()
|
||||
logger.info(
|
||||
"定时任务 %s 已投递执行: execution=%s task=%s",
|
||||
schedule.id, execution.id, task.id,
|
||||
)
|
||||
return str(execution.id)
|
||||
except Exception as e:
|
||||
execution.status = "failed"
|
||||
execution.error_message = f"定时任务投递失败: {e!s}"
|
||||
db.commit()
|
||||
logger.warning("定时任务 %s 投递失败: %s", schedule.id, e)
|
||||
return str(execution.id)
|
||||
|
||||
|
||||
def check_and_run_due_schedules() -> int:
|
||||
"""检查所有启用的定时任务，执行到期的任务。
|
||||
|
||||
被 Celery Beat 每分钟调用一次。
|
||||
|
||||
Returns:
|
||||
本次触发的任务数
|
||||
"""
|
||||
db: Optional[Session] = None
|
||||
try:
|
||||
db = SessionLocal()
|
||||
now = datetime.utcnow()
|
||||
|
||||
due_schedules = (
|
||||
db.query(AgentSchedule)
|
||||
.filter(
|
||||
AgentSchedule.enabled == True, # noqa: E712
|
||||
AgentSchedule.next_run_at <= now,
|
||||
)
|
||||
.all()
|
||||
)
|
||||
|
||||
triggered = 0
|
||||
for sched in due_schedules:
|
||||
try:
|
||||
# 创建执行记录
|
||||
create_execution_for_schedule(db, sched)
|
||||
|
||||
# 更新定时任务状态
|
||||
sched.last_run_at = now
|
||||
sched.last_run_status = "success"
|
||||
# 计算下次执行时间
|
||||
sched.next_run_at = compute_next_run(sched.cron_expression, after=now)
|
||||
db.commit()
|
||||
triggered += 1
|
||||
logger.info(
|
||||
"定时任务触发完成: name=%s next_run=%s",
|
||||
sched.name, sched.next_run_at,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("定时任务 %s 执行失败: %s", sched.id, e)
|
||||
sched.last_run_at = now
|
||||
sched.last_run_status = "failed"
|
||||
db.commit()
|
||||
|
||||
return triggered
|
||||
except Exception as e:
|
||||
logger.error("检查定时任务失败: %s", e)
|
||||
return 0
|
||||
finally:
|
||||
if db:
|
||||
db.close()
|
||||
27
backend/app/tasks/scheduler_tasks.py
Normal file
27
backend/app/tasks/scheduler_tasks.py
Normal file
@@ -0,0 +1,27 @@
|
||||
"""
|
||||
Celery Beat 定时调度任务 — 每分钟检查并触发到期的 Agent 定时任务。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from app.core.tools_bootstrap import ensure_builtin_tools_registered
|
||||
|
||||
ensure_builtin_tools_registered()
|
||||
|
||||
from app.core.celery_app import celery_app
|
||||
from app.services.agent_schedule_service import check_and_run_due_schedules
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@celery_app.task(bind=True, max_retries=0)
|
||||
def check_agent_schedules_task(self):
|
||||
"""检查所有到期的 Agent 定时任务并触发执行。
|
||||
|
||||
由 Celery Beat 每分钟调度一次。若 Redis/Worker 短暂不可用，
|
||||
下次心跳仍会捕获到期的任务（next_run_at 不会因错过而丢失）。
|
||||
"""
|
||||
triggered = check_and_run_due_schedules()
|
||||
if triggered:
|
||||
logger.info("定时任务检查完成: 本次触发 %d 个任务", triggered)
|
||||
Reference in New Issue
Block a user