218 lines
13 KiB
Python
218 lines
13 KiB
Python
"""ORM models for the external-task multi-agent scheduled analysis (任务分析定时).
|
||
|
||
Feature summary (design doc in the frontend ``official-task-analysis`` module):
|
||
an external system pushes tasks via our create API; each task binds 1:1 to a
|
||
business-chain orchestration plus a cron schedule; the scheduler runs the chain
|
||
stage-by-stage (seats within a stage in parallel) and persists per-seat process
|
||
messages plus a content "timeline" version per run. Manual runs wrap the
|
||
roundtable Step-2 coordinator Q&A and are stored with the same message table
|
||
(``seat_agent_id`` distinguishes seats; the coordinator uses ``__coordinator__``).
|
||
|
||
Tables:
|
||
- ``task_analysis_tasks`` — external tasks (source_task_id unique)
|
||
- ``task_analysis_bindings`` — 1:1 binding (chain snapshot + cron + notes)
|
||
- ``task_analysis_runs`` — executions (trigger: schedule | manual)
|
||
- ``task_analysis_run_messages`` — per-seat process messages for a run
|
||
- ``task_analysis_content_versions`` — one timeline version per run
|
||
- ``task_data_block_defs`` — the 5 fixed data blocks per task (title/sort editable)
|
||
- ``task_data_periods`` — data-delivery timeline ticks (period_index per task)
|
||
- ``task_data_blocks`` — merged payload per (task, period, block)
|
||
- ``task_data_submissions`` — append-only submission log feeding the merge
|
||
|
||
JSON payloads are stored as LONGTEXT-backed JSON strings (PortableLongText),
|
||
mirroring the sibling roundtable persistence modules for headroom.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime
|
||
|
||
from sqlalchemy import Boolean, Integer, String, Text, UniqueConstraint
|
||
from sqlalchemy.orm import Mapped, mapped_column
|
||
|
||
from deerflow.persistence.base import Base
|
||
from deerflow.persistence.types import BeijingDateTime, PortableLongText
|
||
|
||
|
||
def _utcnow() -> datetime:
|
||
return datetime.now(UTC)
|
||
|
||
|
||
class TaskAnalysisTaskRow(Base):
|
||
__tablename__ = "task_analysis_tasks"
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
# 对方系统任务号（创建接口传入；唯一）。
|
||
source_task_id: Mapped[str] = mapped_column(String(128), nullable=False, unique=True, index=True)
|
||
name: Mapped[str] = mapped_column(String(512), nullable=False, default="")
|
||
# 任务简介：创建时落库的存量即执行时用的版本（不回拉对方接口）。
|
||
description: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow, onupdate=_utcnow)
|
||
|
||
|
||
class TaskAnalysisBindingRow(Base):
|
||
__tablename__ = "task_analysis_bindings"
|
||
__table_args__ = (UniqueConstraint("external_task_id", name="uq_task_analysis_binding_task"),)
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
external_task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
# 引用 roundtable_chains.id；执行前校验，链缺失/停用 → 跳过并标记 chain_missing。
|
||
chain_id: Mapped[str] = mapped_column(String(64), nullable=False)
|
||
# 执行编排快照 JSON：{seats, stages, stage_goals, coordinator_prompt}。
|
||
# 每次执行前刷新（跟随链最新版），历史 run 用自己的 stages_snapshot 回放。
|
||
chain_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
cron_expr: Mapped[str] = mapped_column(String(64), nullable=False, default="")
|
||
timezone: Mapped[str] = mapped_column(String(64), nullable=False, default="Asia/Shanghai")
|
||
# 执行方式：cron = 按频率定时；once = 只执行一次（创建即执行，next_run_at 恒空）。
|
||
schedule_type: Mapped[str] = mapped_column(String(16), nullable=False, default="cron", server_default="cron")
|
||
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="1")
|
||
next_run_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True, index=True)
|
||
# 绑定级总控提示（与链内 coordinator_prompt 拼接使用）。
|
||
coordinator_prompt: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
# 问答沉淀的「执行方向备注」JSON 数组（逐条追加，注入后续每次定时执行）。
|
||
direction_notes: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
last_run_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
|
||
last_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
|
||
# 执行模型（席位 run 的 model_name；NULL = 系统/席位默认）。
|
||
model: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||
created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow, onupdate=_utcnow)
|
||
|
||
|
||
class TaskAnalysisRunRow(Base):
|
||
__tablename__ = "task_analysis_runs"
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
binding_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
external_task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
# schedule = cron 定时执行；manual = 总控问答（对方详情页发起）。
|
||
trigger: Mapped[str] = mapped_column(String(16), nullable=False, default="schedule")
|
||
# 幂等键：同 binding + scheduled_for 只建一条（对齐 scheduled_tasks.create_run_once）。
|
||
scheduled_for: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False)
|
||
started_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
finished_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
|
||
# pending | running | succeeded | failed | partial
|
||
status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
|
||
# 本次编排快照 JSON：[[seat_id,...],...]（回放不受链后续修改影响）。
|
||
stages_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
stage_goals_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
seat_names_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
# 各席位结果 JSON：{agent_id: {state, duration_ms, thread_id, error}}。
|
||
seat_results: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
error: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||
retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow, onupdate=_utcnow)
|
||
|
||
|
||
class TaskAnalysisRunMessageRow(Base):
|
||
__tablename__ = "task_analysis_run_messages"
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
run_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||
# 消息归属席位（agent_id）；总控消息用固定标识 __coordinator__。
|
||
seat_agent_id: Mapped[str] = mapped_column(String(128), nullable=False, default="")
|
||
stage_index: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||
# input | llm.ai | tool.result | artifact（对齐现有 run messages 事件分类）。
|
||
event_type: Mapped[str] = mapped_column(String(32), nullable=False, default="llm.ai")
|
||
content: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
|
||
|
||
class TaskAnalysisContentVersionRow(Base):
|
||
__tablename__ = "task_analysis_content_versions"
|
||
__table_args__ = (UniqueConstraint("run_id", name="uq_task_analysis_version_run"),)
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
external_task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
# 一次执行 = 一期时间线版本（1:1）。
|
||
run_id: Mapped[str] = mapped_column(String(64), nullable=False)
|
||
# JSON：{key: {name, seat_agent_id, content, format}} —— 对方详情的各块内容。
|
||
blocks: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
|
||
|
||
class TaskDataBlockDefRow(Base):
|
||
"""数据块定义：块 key 定死 block_1~block_6，标题默认「数据块一~六」可改。"""
|
||
|
||
__tablename__ = "task_data_block_defs"
|
||
__table_args__ = (UniqueConstraint("task_id", "block_key", name="uq_task_data_block_def_key"),)
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
# 关联 task_analysis_tasks.id（内部任务 id）。
|
||
task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
block_key: Mapped[str] = mapped_column(String(32), nullable=False)
|
||
title: Mapped[str] = mapped_column(String(128), nullable=False, default="")
|
||
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow, onupdate=_utcnow)
|
||
|
||
|
||
class TaskDataPeriodRow(Base):
|
||
"""数据交付时间轴刻度：每个任务 period_index 从 1 递增；run_id 作幂等锚点。"""
|
||
|
||
__tablename__ = "task_data_periods"
|
||
__table_args__ = (
|
||
UniqueConstraint("task_id", "period_index", name="uq_task_data_period_index"),
|
||
UniqueConstraint("run_id", name="uq_task_data_period_run"),
|
||
)
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
period_index: Mapped[int] = mapped_column(Integer, nullable=False)
|
||
opened_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
# cron = 定时执行开期；manual = 手动触发/总控问答开期；external = 首次提交兜底开期。
|
||
trigger: Mapped[str] = mapped_column(String(16), nullable=False, default="cron")
|
||
# 触发本期的执行记录（问答/定时执行共用 run 表）；提交兜底开的期没有 run。
|
||
run_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
|
||
|
||
class TaskDataBlockRow(Base):
|
||
"""每期每块的合并结果（由 task_data_submissions 流水按序合并得出）。"""
|
||
|
||
__tablename__ = "task_data_blocks"
|
||
__table_args__ = (
|
||
UniqueConstraint("task_id", "period_index", "block_key", name="uq_task_data_block_tpb"),
|
||
)
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
period_index: Mapped[int] = mapped_column(Integer, nullable=False)
|
||
block_key: Mapped[str] = mapped_column(String(32), nullable=False)
|
||
# JSON 对象（顶层 key 合并、同 key 后写覆盖）。
|
||
payload: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow, onupdate=_utcnow)
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
|
||
|
||
class TaskDataSubmissionRow(Base):
|
||
"""入库流水（追加只读）：块 payload 的一致性来源。"""
|
||
|
||
__tablename__ = "task_data_submissions"
|
||
|
||
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||
period_index: Mapped[int] = mapped_column(Integer, nullable=False)
|
||
block_key: Mapped[str] = mapped_column(String(32), nullable=False)
|
||
# 本次提交的 JSON 对象片段（合并语义：顶层 key 合并、同 key 后写覆盖）。
|
||
payload: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
||
# 提交方（席位 agent id / 用户 id，仅审计）。
|
||
submitted_by: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||
# 同 (task, period, block) 内的提交序号，从 0 递增 —— 合并顺序依据。
|
||
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
|
||
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow)
|
||
|
||
|
||
class TaskAnalysisSettingRow(Base):
|
||
"""模块级 KV 设置（当前仅默认业务链）。"""
|
||
|
||
__tablename__ = "task_analysis_settings"
|
||
|
||
key: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||
value: Mapped[str] = mapped_column(PortableLongText, nullable=False, default="")
|
||
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=_utcnow, onupdate=_utcnow)
|