|
|
@@ -1,215 +1,288 @@
|
|
|
-"""
|
|
|
-定时任务:对 multi_demand_pool_di 需求池中的需求循环打分级。
|
|
|
-
|
|
|
-循环控制在本文件(调度任务侧),而不是让 agent 在一次运行内自行分页遍历——
|
|
|
-每一轮并行启动多个 worker(默认 5 个线程),每个 worker 处理一批互不重叠的
|
|
|
-需求词(默认每批 20 个),调用一次 demand_grade_agent 并等待其运行结束
|
|
|
-(agent 内部会调用 batch_save_demand_grades 落库),再重新查询"已分级"集合、
|
|
|
-分配下一批新词,如此循环直到没有更多待分级需求或达到本次运行的批次上限,
|
|
|
-从而避免单次 agent 会话上下文无限增长,同时提升吞吐。
|
|
|
-"""
|
|
|
+"""统筹落库后执行分级任务。"""
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import logging
|
|
|
-import math
|
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
from datetime import datetime
|
|
|
+from typing import Any
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
from agents.demand_grade_agent.run import main as grade_demand_words
|
|
|
+from agents.demand_grade_orchestrator_agent.run import orchestrate_daily_grade_plan
|
|
|
from supply_infra.config import get_infra_settings
|
|
|
+from supply_infra.db.repositories.demand_grade_plan_repo import DemandGradePlanRepository
|
|
|
from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
|
|
|
from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
|
|
|
from supply_infra.db.session import get_session
|
|
|
+from supply_infra.scheduler.plan_group_batch import MAX_DEMANDS_PER_BATCH, split_even_batches
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
-_DEFAULT_BATCH_SIZE = 20
|
|
|
_DEFAULT_WORKERS = 5
|
|
|
|
|
|
|
|
|
-def _resolve_biz_dt(biz_dt: str | None) -> str | None:
|
|
|
+def _resolve_biz_dt(biz_dt: str | None) -> str:
|
|
|
if biz_dt:
|
|
|
return biz_dt
|
|
|
- # 定时任务默认按“当天业务日”执行;使用 scheduler_timezone 以免机器时区不同导致 biz_dt 偏移。
|
|
|
- settings = get_infra_settings()
|
|
|
- tz = ZoneInfo(settings.scheduler_timezone)
|
|
|
- return datetime.now(tz).strftime("%Y%m%d")
|
|
|
+ return datetime.now(ZoneInfo(get_infra_settings().scheduler_timezone)).strftime("%Y%m%d")
|
|
|
|
|
|
|
|
|
-def _fetch_next_batch(biz_dt: str, batch_size: int, exclude_names: set[str]) -> list[str]:
|
|
|
- """按 exclude_names 过滤已分配/已分级词,取下一批(不用 offset,避免和落库进度错位)。"""
|
|
|
+def _materialize_pending_group_items(biz_dt: str) -> int:
|
|
|
+ """执行前为 pending 计划组物化待分级需求明细。"""
|
|
|
with get_session() as session:
|
|
|
- summaries = MultiDemandPoolDiRepository(session).list_distinct_demand_name_summaries(
|
|
|
+ graded_names = DemandGradeRepository(session).get_existing_demand_names(biz_dt)
|
|
|
+ return DemandGradePlanRepository(session).materialize_pending_groups(
|
|
|
biz_dt,
|
|
|
- limit=batch_size,
|
|
|
- offset=0,
|
|
|
- exclude_names=list(exclude_names) if exclude_names else None,
|
|
|
+ graded_names=graded_names,
|
|
|
)
|
|
|
- return [item["demand_name"] for item in summaries]
|
|
|
|
|
|
|
|
|
-def _allocate_parallel_batches(
|
|
|
- biz_dt: str,
|
|
|
- batch_size: int,
|
|
|
- exclude_names: set[str],
|
|
|
- *,
|
|
|
- num_workers: int,
|
|
|
-) -> list[list[str]]:
|
|
|
- """
|
|
|
- 为同一轮并行 worker 分配互不重叠的批次。
|
|
|
-
|
|
|
- 每取出一批后立即加入 reserved,下一批查询时排除,保证单轮内不会重复分配。
|
|
|
- """
|
|
|
- batches: list[list[str]] = []
|
|
|
- reserved = set(exclude_names)
|
|
|
- for _ in range(num_workers):
|
|
|
- batch = _fetch_next_batch(biz_dt, batch_size, reserved)
|
|
|
- if not batch:
|
|
|
- break
|
|
|
- batches.append(batch)
|
|
|
- reserved.update(batch)
|
|
|
- return batches
|
|
|
+def _group_success(item_summary: dict[str, int], processed_batches: int) -> bool:
|
|
|
+ if item_summary.get("pending", 0) > 0:
|
|
|
+ return False
|
|
|
+ total = sum(item_summary.values())
|
|
|
+ if total == 0:
|
|
|
+ return True
|
|
|
+ if processed_batches > 0:
|
|
|
+ return True
|
|
|
+ return item_summary.get("failed", 0) == 0 and (
|
|
|
+ item_summary.get("finished", 0) > 0 or item_summary.get("skipped", 0) > 0
|
|
|
+ )
|
|
|
|
|
|
|
|
|
-def _fetch_graded_names(biz_dt: str) -> set[str]:
|
|
|
+def _run_group(biz_dt: str, group_id: int, *, max_demands_per_batch: int) -> int:
|
|
|
+ """领取并执行一个固定任务,从组内需求明细表按批读取。"""
|
|
|
with get_session() as session:
|
|
|
- return DemandGradeRepository(session).get_existing_demand_names(biz_dt)
|
|
|
-
|
|
|
+ group = DemandGradePlanRepository(session).claim_group(biz_dt, group_id)
|
|
|
+ if group is None:
|
|
|
+ logger.warning("分级任务已被其他 worker 领取或状态已变化: biz_dt=%s group_id=%s", biz_dt, group_id)
|
|
|
+ return 0
|
|
|
+
|
|
|
+ gid = int(group["id"])
|
|
|
+ processed_batches = 0
|
|
|
+ batch_errors: list[str] = []
|
|
|
+ try:
|
|
|
+ with get_session() as session:
|
|
|
+ items = DemandGradePlanRepository(session).list_pending_group_items(gid)
|
|
|
+
|
|
|
+ for batch_items in split_even_batches(items, max_per_batch=max_demands_per_batch):
|
|
|
+ demands = [
|
|
|
+ {"pool_id": item["pool_id"], "demand_name": item["demand_name"]}
|
|
|
+ for item in batch_items
|
|
|
+ ]
|
|
|
+ item_ids = [int(item["item_id"]) for item in batch_items]
|
|
|
+ try:
|
|
|
+ grade_demand_words(demands, biz_dt=biz_dt)
|
|
|
+ with get_session() as session:
|
|
|
+ DemandGradePlanRepository(session).mark_group_items_status(
|
|
|
+ item_ids,
|
|
|
+ status="finished",
|
|
|
+ )
|
|
|
+ processed_batches += 1
|
|
|
+ except Exception as exc:
|
|
|
+ logger.exception(
|
|
|
+ "分级子批次失败,跳过并继续本组其余需求: biz_dt=%s group=%s count=%s",
|
|
|
+ biz_dt,
|
|
|
+ group["group_key"],
|
|
|
+ len(demands),
|
|
|
+ )
|
|
|
+ batch_errors.append(str(exc))
|
|
|
+ with get_session() as session:
|
|
|
+ DemandGradePlanRepository(session).mark_group_items_status(
|
|
|
+ item_ids,
|
|
|
+ status="failed",
|
|
|
+ error_message=str(exc),
|
|
|
+ )
|
|
|
|
|
|
-def _run_batch_in_thread(batch: list[str], biz_dt: str, batch_label: str) -> None:
|
|
|
- logger.info("Grade demand pool %s (size=%d): %s", batch_label, len(batch), batch)
|
|
|
- grade_demand_words(batch, biz_dt=biz_dt)
|
|
|
+ with get_session() as session:
|
|
|
+ repo = DemandGradePlanRepository(session)
|
|
|
+ item_summary = repo.summarize_group_items(gid)
|
|
|
+ repo.finish_group(
|
|
|
+ gid,
|
|
|
+ success=_group_success(item_summary, processed_batches),
|
|
|
+ error_message="; ".join(batch_errors) if batch_errors else None,
|
|
|
+ )
|
|
|
+ except Exception as exc:
|
|
|
+ logger.exception(
|
|
|
+ "分级计划任务失败: biz_dt=%s group=%s",
|
|
|
+ biz_dt,
|
|
|
+ group["group_key"],
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ with get_session() as session:
|
|
|
+ DemandGradePlanRepository(session).finish_group(
|
|
|
+ gid,
|
|
|
+ success=False,
|
|
|
+ error_message=str(exc),
|
|
|
+ )
|
|
|
+ except Exception:
|
|
|
+ logger.exception(
|
|
|
+ "记录分级计划任务失败状态时发生错误: biz_dt=%s group=%s",
|
|
|
+ biz_dt,
|
|
|
+ group["group_key"],
|
|
|
+ )
|
|
|
+ return 0
|
|
|
+ return processed_batches
|
|
|
|
|
|
|
|
|
-def grade_demand_pool(
|
|
|
- biz_dt: str | None = None,
|
|
|
+def _execute_plan_tasks(
|
|
|
+ biz_dt: str,
|
|
|
*,
|
|
|
- batch_size: int = _DEFAULT_BATCH_SIZE,
|
|
|
- max_batches: int | None = None,
|
|
|
- workers: int = _DEFAULT_WORKERS,
|
|
|
-) -> dict:
|
|
|
- """
|
|
|
- 循环对需求池中尚未分级的需求打分级,每轮并行调用多个 agent 并等待其完成。
|
|
|
-
|
|
|
- Args:
|
|
|
- biz_dt: 业务日期 YYYYMMDD;不传则取需求池最新业务日。
|
|
|
- batch_size: 每批交给 agent 的需求词数量。
|
|
|
- max_batches: 本次运行最多执行多少批(安全阀,避免单次运行时间过长/无限循环)。
|
|
|
- 不传/为 None 时,将根据当天数据库待分级需求词数量动态计算上限:
|
|
|
- `ceil(total / batch_size) + 10`。
|
|
|
- workers: 并行线程数;每轮最多同时跑 workers 个互不重叠的批次。
|
|
|
-
|
|
|
- Returns:
|
|
|
- 运行统计:{"biz_dt", "total", "graded_before", "graded_after", "batches_run", "stopped_reason"}
|
|
|
- """
|
|
|
- resolved_biz_dt = _resolve_biz_dt(biz_dt)
|
|
|
+ workers: int,
|
|
|
+ max_demands_per_batch: int,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """并发执行当天全部 pending 计划任务,仅执行一轮。"""
|
|
|
with get_session() as session:
|
|
|
- total = MultiDemandPoolDiRepository(session).count_distinct_demand_names(resolved_biz_dt)
|
|
|
-
|
|
|
- graded_names = _fetch_graded_names(resolved_biz_dt)
|
|
|
- graded_before = len(graded_names)
|
|
|
-
|
|
|
- # max_batches 默认为动态上限:ceil(total / batch_size) + 10
|
|
|
- # 其中 total 使用“当天需求词去重数”,与本 job 的分页维度一致。
|
|
|
- if max_batches is None:
|
|
|
- batches_needed = math.ceil(total / batch_size) if batch_size > 0 else 0
|
|
|
- max_batches = batches_needed + 10
|
|
|
+ group_ids = DemandGradePlanRepository(session).list_pending_group_ids(biz_dt)
|
|
|
+ if not group_ids:
|
|
|
+ with get_session() as session:
|
|
|
+ final_snapshot = DemandGradePlanRepository(session).get_execution_snapshot(biz_dt)
|
|
|
+ return {
|
|
|
+ "attempted_groups": 0,
|
|
|
+ "groups_run": 0,
|
|
|
+ "workers": 0,
|
|
|
+ "final_snapshot": final_snapshot,
|
|
|
+ "execution_complete": final_snapshot["execution_complete"],
|
|
|
+ }
|
|
|
+
|
|
|
+ worker_count = max(1, min(int(workers), len(group_ids)))
|
|
|
+ groups_run = 0
|
|
|
+ with ThreadPoolExecutor(max_workers=worker_count) as executor:
|
|
|
+ futures = [
|
|
|
+ executor.submit(_run_group, biz_dt, group_id, max_demands_per_batch=max_demands_per_batch)
|
|
|
+ for group_id in group_ids
|
|
|
+ ]
|
|
|
+ for future in as_completed(futures):
|
|
|
+ try:
|
|
|
+ groups_run += future.result()
|
|
|
+ except Exception:
|
|
|
+ logger.exception("分级任务 worker 出现未捕获错误: biz_dt=%s", biz_dt)
|
|
|
|
|
|
- worker_count = max(1, workers)
|
|
|
-
|
|
|
- logger.info(
|
|
|
- "Grade demand pool start: biz_dt=%s total=%d graded=%d batch_size=%d "
|
|
|
- "workers=%d max_batches=%s",
|
|
|
- resolved_biz_dt,
|
|
|
- total,
|
|
|
- graded_before,
|
|
|
- batch_size,
|
|
|
- worker_count,
|
|
|
- max_batches,
|
|
|
- )
|
|
|
+ with get_session() as session:
|
|
|
+ final_snapshot = DemandGradePlanRepository(session).get_execution_snapshot(biz_dt)
|
|
|
+ return {
|
|
|
+ "attempted_groups": len(group_ids),
|
|
|
+ "groups_run": groups_run,
|
|
|
+ "workers": worker_count,
|
|
|
+ "final_snapshot": final_snapshot,
|
|
|
+ "execution_complete": final_snapshot["execution_complete"],
|
|
|
+ }
|
|
|
|
|
|
- batches_run = 0
|
|
|
- stopped_reason = "no_more_pending"
|
|
|
|
|
|
+def execute_plan_tasks_until_complete(
|
|
|
+ biz_dt: str,
|
|
|
+ *,
|
|
|
+ workers: int,
|
|
|
+ max_demands_per_batch: int = MAX_DEMANDS_PER_BATCH,
|
|
|
+ max_rounds: int = 0,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """循环执行 pending 计划任务,直到全部完成或达到 max_rounds。"""
|
|
|
+ rounds: list[dict[str, Any]] = []
|
|
|
+ groups_run = 0
|
|
|
+ round_no = 0
|
|
|
while True:
|
|
|
- if max_batches is not None and batches_run >= max_batches:
|
|
|
- stopped_reason = "max_batches_reached"
|
|
|
- break
|
|
|
-
|
|
|
- if max_batches is not None:
|
|
|
- slots = max_batches - batches_run
|
|
|
- num_workers = min(worker_count, slots)
|
|
|
- else:
|
|
|
- num_workers = worker_count
|
|
|
-
|
|
|
- if num_workers <= 0:
|
|
|
- stopped_reason = "max_batches_reached"
|
|
|
+ if max_rounds > 0 and round_no >= max_rounds:
|
|
|
break
|
|
|
-
|
|
|
- batches = _allocate_parallel_batches(
|
|
|
- resolved_biz_dt,
|
|
|
- batch_size,
|
|
|
- graded_names,
|
|
|
- num_workers=num_workers,
|
|
|
+ round_no += 1
|
|
|
+ result = _execute_plan_tasks(
|
|
|
+ biz_dt,
|
|
|
+ workers=workers,
|
|
|
+ max_demands_per_batch=max_demands_per_batch,
|
|
|
)
|
|
|
- if not batches:
|
|
|
- stopped_reason = "no_more_pending"
|
|
|
+ rounds.append(result)
|
|
|
+ groups_run += int(result.get("groups_run") or 0)
|
|
|
+ if int(result.get("attempted_groups") or 0) == 0:
|
|
|
break
|
|
|
|
|
|
- round_start_batch = batches_run + 1
|
|
|
- round_failed = False
|
|
|
-
|
|
|
- with ThreadPoolExecutor(max_workers=len(batches)) as executor:
|
|
|
- futures = {
|
|
|
- executor.submit(
|
|
|
- _run_batch_in_thread,
|
|
|
- batch,
|
|
|
- resolved_biz_dt,
|
|
|
- f"batch {round_start_batch + idx}",
|
|
|
- ): batch
|
|
|
- for idx, batch in enumerate(batches)
|
|
|
- }
|
|
|
- for future in as_completed(futures):
|
|
|
- batch = futures[future]
|
|
|
- try:
|
|
|
- future.result()
|
|
|
- except Exception as e:
|
|
|
- logger.error(
|
|
|
- "Grade demand pool batch failed: %s (batch=%s)",
|
|
|
- e,
|
|
|
- batch,
|
|
|
- exc_info=True,
|
|
|
- )
|
|
|
- round_failed = True
|
|
|
+ final_snapshot = rounds[-1]["final_snapshot"] if rounds else None
|
|
|
+ if final_snapshot is None:
|
|
|
+ with get_session() as session:
|
|
|
+ final_snapshot = DemandGradePlanRepository(session).get_execution_snapshot(biz_dt)
|
|
|
+
|
|
|
+ return {
|
|
|
+ "rounds": round_no,
|
|
|
+ "groups_run": groups_run,
|
|
|
+ "workers": max(1, int(workers)),
|
|
|
+ "round_details": rounds,
|
|
|
+ "final_snapshot": final_snapshot,
|
|
|
+ "execution_complete": bool(final_snapshot["execution_complete"]),
|
|
|
+ }
|
|
|
|
|
|
- batches_run += len(batches)
|
|
|
|
|
|
- if round_failed:
|
|
|
- stopped_reason = "batch_failed"
|
|
|
- break
|
|
|
+def _grade_demand_pool_impl(
|
|
|
+ resolved_biz_dt: str,
|
|
|
+ *,
|
|
|
+ workers: int,
|
|
|
+ max_demands_per_batch: int = MAX_DEMANDS_PER_BATCH,
|
|
|
+ with_orchestrate: bool = True,
|
|
|
+ max_rounds: int = 0,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ with get_session() as session:
|
|
|
+ total = MultiDemandPoolDiRepository(session).count_distinct_demand_names(resolved_biz_dt)
|
|
|
+ graded_before = DemandGradeRepository(session).count_by_biz_dt(resolved_biz_dt)
|
|
|
|
|
|
- new_graded_names = _fetch_graded_names(resolved_biz_dt)
|
|
|
- if len(new_graded_names) <= len(graded_names):
|
|
|
- # agent 没有对本轮产生任何新的落库结果,避免死循环重复拿到同一批
|
|
|
- logger.warning(
|
|
|
- "Grade demand pool round made no progress (graded still %d), stop",
|
|
|
- len(graded_names),
|
|
|
- )
|
|
|
- stopped_reason = "stalled"
|
|
|
- graded_names = new_graded_names
|
|
|
- break
|
|
|
- graded_names = new_graded_names
|
|
|
+ if with_orchestrate:
|
|
|
+ try:
|
|
|
+ orchestrate_daily_grade_plan(biz_dt=resolved_biz_dt)
|
|
|
+ except Exception:
|
|
|
+ logger.exception("统筹 Agent 执行失败,继续处理数据库中已有任务: biz_dt=%s", resolved_biz_dt)
|
|
|
+
|
|
|
+ materialized = _materialize_pending_group_items(resolved_biz_dt)
|
|
|
+ logger.info("物化计划组需求明细: biz_dt=%s items=%s", resolved_biz_dt, materialized)
|
|
|
|
|
|
+ plan_execution = execute_plan_tasks_until_complete(
|
|
|
+ resolved_biz_dt,
|
|
|
+ workers=max(1, int(workers)),
|
|
|
+ max_demands_per_batch=max(1, min(int(max_demands_per_batch), MAX_DEMANDS_PER_BATCH)),
|
|
|
+ max_rounds=max_rounds,
|
|
|
+ )
|
|
|
+
|
|
|
+ with get_session() as session:
|
|
|
+ graded_after = DemandGradeRepository(session).count_by_biz_dt(resolved_biz_dt)
|
|
|
+ final_snapshot = plan_execution["final_snapshot"]
|
|
|
result = {
|
|
|
+ "success": bool(plan_execution["execution_complete"]),
|
|
|
"biz_dt": resolved_biz_dt,
|
|
|
"total": total,
|
|
|
"graded_before": graded_before,
|
|
|
- "graded_after": len(graded_names),
|
|
|
- "batches_run": batches_run,
|
|
|
- "workers": worker_count,
|
|
|
- "stopped_reason": stopped_reason,
|
|
|
+ "graded_after": graded_after,
|
|
|
+ "materialized_items": materialized,
|
|
|
+ "planned_category_count": len(final_snapshot["assigned_category_ids"]),
|
|
|
+ "planned_groups": final_snapshot["planned_groups"],
|
|
|
+ "group_status": final_snapshot["group_status"],
|
|
|
+ "plan_execution": plan_execution,
|
|
|
+ "workers": plan_execution["workers"],
|
|
|
+ "groups_run": plan_execution["groups_run"],
|
|
|
"run_at": datetime.now().isoformat(),
|
|
|
}
|
|
|
logger.info("Grade demand pool completed: %s", result)
|
|
|
return result
|
|
|
+
|
|
|
+
|
|
|
+def grade_demand_pool(
|
|
|
+ biz_dt: str | None = None,
|
|
|
+ *,
|
|
|
+ workers: int = _DEFAULT_WORKERS,
|
|
|
+ max_demands_per_batch: int = MAX_DEMANDS_PER_BATCH,
|
|
|
+ with_orchestrate: bool = True,
|
|
|
+ max_rounds: int = 0,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ """执行完整分级闭环;任何错误只记录日志并返回,不向上中断定时任务。"""
|
|
|
+ resolved_biz_dt = str(biz_dt or "")
|
|
|
+ try:
|
|
|
+ resolved_biz_dt = _resolve_biz_dt(biz_dt)
|
|
|
+ return _grade_demand_pool_impl(
|
|
|
+ resolved_biz_dt,
|
|
|
+ workers=workers,
|
|
|
+ max_demands_per_batch=max_demands_per_batch,
|
|
|
+ with_orchestrate=with_orchestrate,
|
|
|
+ max_rounds=max_rounds,
|
|
|
+ )
|
|
|
+ except Exception as exc:
|
|
|
+ logger.exception("需求分级任务发生未捕获错误,已阻止异常中断定时任务: biz_dt=%s", resolved_biz_dt)
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "biz_dt": resolved_biz_dt,
|
|
|
+ "error": str(exc),
|
|
|
+ "run_at": datetime.now().isoformat(),
|
|
|
+ }
|