752 lines
34 KiB
Python
752 lines
34 KiB
Python
"""DSH Planner: generate an immutable, validated campaign plan version.
|
||
|
||
The workbench owns templates/resources/approvals. DSH Planner may propose a
|
||
segmented DAG, but workers can only execute a plan after it passes deterministic
|
||
validation and (for normal campaigns) human approval.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
from sqlmodel import Session, select
|
||
|
||
from . import progress
|
||
from .agents.coordinator import parse_mindmap
|
||
from .rag import backend as rag_backend
|
||
from .rag.pipeline import consumed_chunk_ids, simulate_lightweight_plan
|
||
from .models import (
|
||
Agent,
|
||
Campaign,
|
||
CampaignPlanVersion,
|
||
KnowledgeBase,
|
||
KnowledgeUsage,
|
||
Segment,
|
||
WorkflowTemplate,
|
||
now,
|
||
)
|
||
|
||
DEFAULT_TEMPLATE_KEY = "default-live-course"
|
||
TAG_ALIAS = {
|
||
"产品回指": ["产品销售", "产品体验"],
|
||
"转场": ["开场", "互动"],
|
||
"预告": ["钩子"],
|
||
"见证": ["案例"],
|
||
"金句": ["心法"],
|
||
"共鸣": ["心法"],
|
||
"安全提示": ["技法"],
|
||
"跟练": ["技法", "功法"],
|
||
"氛围": ["场景"],
|
||
"情绪": ["场景", "心法"],
|
||
}
|
||
PLAN_STATUSES = {"draft", "waiting_review", "approved", "rejected", "superseded", "failed"}
|
||
|
||
|
||
def default_template_config() -> dict[str, Any]:
|
||
return {
|
||
"version": 1,
|
||
"mode": "linear_dag",
|
||
"required_tags": [],
|
||
"optional_tags": [
|
||
"开场", "钩子", "心法", "技法", "功法", "产品销售", "产品体验",
|
||
"互动", "案例", "收尾", "预告",
|
||
],
|
||
"max_segments": 80,
|
||
"require_plan_approval": True,
|
||
"allow_auto_replan": False,
|
||
"default_kb_categories": {},
|
||
"risk_rules": {
|
||
"pending_confirm": "high",
|
||
"unassigned_agent": "high",
|
||
"knowledge_gap": "high",
|
||
"product_reference": "medium",
|
||
"duration_mismatch": "medium",
|
||
},
|
||
}
|
||
|
||
|
||
def ensure_default_workflow_template(session: Session) -> WorkflowTemplate:
|
||
"""Create the built-in template idempotently."""
|
||
tpl = session.exec(select(WorkflowTemplate).where(
|
||
WorkflowTemplate.key == DEFAULT_TEMPLATE_KEY)).first()
|
||
if tpl:
|
||
return tpl
|
||
tpl = WorkflowTemplate(
|
||
key=DEFAULT_TEMPLATE_KEY,
|
||
name="直播课标准工作流",
|
||
description="DSH Planner 在直播课模板内生成开场、主体、产品/转化与收束的线性 DAG；高风险计划必须人工批准。",
|
||
config=default_template_config(),
|
||
enabled=True,
|
||
builtin=True,
|
||
organization_id=1,
|
||
owner_user_id=0,
|
||
visibility="global",
|
||
)
|
||
session.add(tpl)
|
||
session.commit()
|
||
session.refresh(tpl)
|
||
return tpl
|
||
|
||
|
||
def get_template(session: Session, template_id: int = 0, template_key: str = "") -> WorkflowTemplate:
|
||
if template_id:
|
||
tpl = session.get(WorkflowTemplate, int(template_id))
|
||
if not tpl or not tpl.enabled:
|
||
raise ValueError("工作流模板不存在或已停用")
|
||
return tpl
|
||
if template_key:
|
||
tpl = session.exec(select(WorkflowTemplate).where(
|
||
WorkflowTemplate.key == template_key, WorkflowTemplate.enabled == True)).first() # noqa: E712
|
||
if tpl:
|
||
return tpl
|
||
return ensure_default_workflow_template(session)
|
||
|
||
|
||
def _issue(code: str, severity: str, message: str, node_id: str = "", **extra: Any) -> dict[str, Any]:
|
||
item = {"code": code, "severity": severity, "message": message, "node_id": node_id}
|
||
item.update(extra)
|
||
return item
|
||
|
||
|
||
def _candidate_agents(session: Session, tag: str) -> list[Agent]:
|
||
aliases = [tag, *TAG_ALIAS.get(tag, [])]
|
||
active = list(session.exec(select(Agent).where(
|
||
Agent.role == "production", Agent.status == "active")).all())
|
||
for alias in aliases:
|
||
hits = [a for a in active if alias in (a.content_tags or [])]
|
||
if hits:
|
||
return hits
|
||
return active
|
||
|
||
|
||
def _match_agent(session: Session, tag: str) -> tuple[int, str, list[int]]:
|
||
hits = _candidate_agents(session, tag)
|
||
if not hits:
|
||
return 0, "没有启用的生产 Agent", []
|
||
chosen = hits[0]
|
||
aliases = TAG_ALIAS.get(tag, [])
|
||
match_type = "tag" if (tag in (chosen.content_tags or []) or
|
||
any(a in (chosen.content_tags or []) for a in aliases)) else "fallback"
|
||
return int(chosen.id or 0), match_type, [int(a.id or 0) for a in hits]
|
||
|
||
|
||
def _match_plan_agent(session: Session, campaign: Campaign, tag: str,
|
||
used: dict[str, int]) -> tuple[int, str, list[int], str]:
|
||
"""Match a planner node to the selected workflow canvas first.
|
||
|
||
The canvas is a workbench-managed resource boundary. If its tagged nodes
|
||
cannot satisfy the proposal, fall back to active production agents and mark
|
||
the assignment for review instead of silently inventing permissions.
|
||
"""
|
||
aliases = [tag, *TAG_ALIAS.get(tag, [])]
|
||
snap = campaign.canvas_snapshot or {}
|
||
canvas_nodes = snap.get("nodes", []) if isinstance(snap, dict) else []
|
||
for alias in aliases:
|
||
hits = [n for n in canvas_nodes
|
||
if str(n.get("content_tag") or n.get("tag") or "") == alias
|
||
and n.get("enabled", True)]
|
||
if hits:
|
||
raw = hits[used.get(alias, 0) % len(hits)]
|
||
used[alias] = used.get(alias, 0) + 1
|
||
agent_id = int(raw.get("agent_id") or 0)
|
||
agent = session.get(Agent, agent_id) if agent_id else None
|
||
candidates = []
|
||
for n in hits:
|
||
aid = int(n.get("agent_id") or 0)
|
||
if aid and aid not in candidates:
|
||
candidates.append(aid)
|
||
source_id = str(raw.get("node_id") or raw.get("id") or "")
|
||
if agent and agent.status == "active" and agent.role == "production":
|
||
return int(agent.id or 0), "canvas_tag", candidates, source_id
|
||
return agent_id, "canvas_agent_invalid", candidates, source_id
|
||
agent_id, match_type, candidates = _match_agent(session, tag)
|
||
return agent_id, match_type, candidates, ""
|
||
|
||
|
||
def _time_range_minutes(value: str, campaign_duration_minutes: int = 0) -> int | None:
|
||
"""Parse planner ranges as mm:ss for short tasks and hh:mm for long tasks.
|
||
|
||
The LLM is asked for time_range values, but a 3 minute口播 naturally emits
|
||
``00:00-03:00`` while a 180 minute直播 emits ``00:00-180:00``. Interpreting
|
||
both as hh:mm makes short plans fail deterministic duration validation.
|
||
When the task duration is <= 60 minutes, two-component ranges are mm:ss;
|
||
otherwise they remain hh:mm. Parsed second values are rounded to whole
|
||
minutes, so a 30 second segment is represented as 1 minute in capacity checks.
|
||
"""
|
||
text = str(value or "").strip()
|
||
m = re.match(r"^(\d{1,3}):(\d{1,2})\s*(?:-|~|至|到)\s*(\d{1,3}):(\d{1,2})$", text)
|
||
if not m:
|
||
return None
|
||
first, second, third, fourth = map(int, m.groups())
|
||
if int(campaign_duration_minutes or 0) > 0 and int(campaign_duration_minutes) <= 60:
|
||
start_seconds = first * 60 + second
|
||
end_seconds = third * 60 + fourth
|
||
if end_seconds < start_seconds:
|
||
end_seconds += 60 * 60
|
||
seconds = end_seconds - start_seconds
|
||
return max(1, (seconds + 59) // 60) if seconds else 0
|
||
start, end = first * 60 + second, third * 60 + fourth
|
||
if end < start:
|
||
end += 24 * 60
|
||
return end - start
|
||
|
||
|
||
def topo_waves(node_ids: list[str], edges: list[dict[str, str]]) -> list[list[str]]:
|
||
"""Return Kahn waves for a DAG; raise ValueError when a cycle is present."""
|
||
indegree = {nid: 0 for nid in node_ids}
|
||
adjacency: dict[str, list[str]] = {nid: [] for nid in node_ids}
|
||
for edge in edges:
|
||
src = str(edge.get("from") or edge.get("from_node") or "").strip()
|
||
dst = str(edge.get("to") or edge.get("to_node") or "").strip()
|
||
if src not in indegree or dst not in indegree:
|
||
raise ValueError(f"DAG 边引用了不存在的节点：{src} -> {dst}")
|
||
adjacency[src].append(dst)
|
||
indegree[dst] += 1
|
||
wave = sorted(nid for nid, degree in indegree.items() if degree == 0)
|
||
waves: list[list[str]] = []
|
||
seen = 0
|
||
while wave:
|
||
waves.append(wave)
|
||
seen += len(wave)
|
||
next_wave: list[str] = []
|
||
for nid in wave:
|
||
for dst in adjacency[nid]:
|
||
indegree[dst] -= 1
|
||
if indegree[dst] == 0:
|
||
next_wave.append(dst)
|
||
wave = sorted(next_wave)
|
||
if seen != len(node_ids):
|
||
raise ValueError("工作流 DAG 存在环，不能执行")
|
||
return waves
|
||
|
||
|
||
def _knowledge_issues(session: Session, agent_id: int, node_id: str) -> list[dict]:
|
||
"""Validate Agent/KB bindings for every planned node.
|
||
|
||
Actual source capacity is checked once in ``_knowledge_capacity_issues``.
|
||
Checking chunk inventory per node would both duplicate errors and prevent a
|
||
plan-level reservation simulation across DAG order.
|
||
"""
|
||
issues: list[dict[str, Any]] = []
|
||
if not agent_id:
|
||
return [_issue("unassigned_agent", "error", "该节点还没有分配生产 Agent", node_id)]
|
||
agent = session.get(Agent, agent_id)
|
||
if not agent:
|
||
return [_issue("agent_missing", "error", "计划引用了不存在的 Agent", node_id)]
|
||
if agent.role != "production" or agent.status != "active":
|
||
return [_issue("agent_invalid", "error", "计划引用的 Agent 不是启用生产 Agent", node_id,
|
||
agent_id=agent_id)]
|
||
kb_ids = list(agent.kb_ids or [])
|
||
if not kb_ids:
|
||
return [_issue("knowledge_missing", "error", "生产 Agent 未绑定知识库", node_id,
|
||
agent_id=agent_id)]
|
||
for kb_id in kb_ids:
|
||
kb = session.get(KnowledgeBase, int(kb_id))
|
||
if not kb:
|
||
issues.append(_issue("knowledge_missing", "error", f"知识库 {kb_id} 不存在", node_id,
|
||
agent_id=agent_id, kb_id=kb_id))
|
||
continue
|
||
if int(kb.file_count or 0) <= 0:
|
||
issues.append(_issue("knowledge_gap", "error",
|
||
f"知识库「{kb.name}」没有素材文件，规划阶段不能确认事实来源",
|
||
node_id, agent_id=agent_id, kb_id=kb_id))
|
||
return issues
|
||
|
||
|
||
# Worker currently attempts an initial top-6 recall, then widens to top-24.
|
||
# Planner uses the widened bound for admission control; this is intentionally
|
||
# conservative because selected chunks are consumed by each production Agent.
|
||
_PLAN_SIMULATION_TOP_K = 24
|
||
_PLAN_SOURCE_CHAR_BUDGET = 12000
|
||
|
||
|
||
def _node_knowledge_query(node: dict[str, Any]) -> str:
|
||
return " ".join(str(node.get(key) or "") for key in
|
||
("title", "content_tag", "product_ref", "mindmap_node")).strip()
|
||
|
||
|
||
def _vector_knowledge_capacity(session: Session, agent: Agent,
|
||
assigned_nodes: list[dict[str, Any]]) -> tuple[list[dict], dict]:
|
||
issues: list[dict[str, Any]] = []
|
||
kb_ids = [int(kid) for kid in (agent.kb_ids or [])]
|
||
valid_kbs = [session.get(KnowledgeBase, kid) for kid in kb_ids]
|
||
valid_kbs = [kb for kb in valid_kbs if kb]
|
||
empty_kbs = [kb for kb in valid_kbs if int(kb.file_count or 0) <= 0
|
||
or int(kb.chunk_count or 0) <= 0]
|
||
locked_rows = session.exec(select(KnowledgeUsage.chunk_id, KnowledgeUsage.kb_id).where(
|
||
KnowledgeUsage.agent_id == int(agent.id or 0),
|
||
KnowledgeUsage.kb_id.in_(kb_ids),
|
||
(KnowledgeUsage.status.in_(["reserved", "consumed"]) | (KnowledgeUsage.banned == True)), # noqa: E712
|
||
)).all()
|
||
locked_count = len({(int(kb_id), str(chunk_id)) for chunk_id, kb_id in locked_rows})
|
||
total_chunks = sum(max(0, int(kb.chunk_count or 0)) for kb in valid_kbs)
|
||
available_chunks = max(0, total_chunks - locked_count)
|
||
demand = len(assigned_nodes)
|
||
report = {
|
||
"mode": "vector",
|
||
"kb_ids": kb_ids,
|
||
"total_chunks": total_chunks,
|
||
"locked_chunks": locked_count,
|
||
"available_chunks": available_chunks,
|
||
"required_nodes": demand,
|
||
"empty_kb_ids": [int(kb.id) for kb in empty_kbs],
|
||
}
|
||
first_node = assigned_nodes[0]
|
||
first_node_id = str(first_node.get("node_id") or "")
|
||
if not valid_kbs or all(int(kb.file_count or 0) <= 0 for kb in valid_kbs):
|
||
issues.append(_issue("knowledge_gap", "error",
|
||
f"Agent「{agent.name}」绑定的知识库没有素材文件，规划阶段不能确认事实来源",
|
||
first_node_id, agent_id=agent.id))
|
||
elif available_chunks <= 0:
|
||
issues.append(_issue("knowledge_exhausted", "error",
|
||
f"Agent「{agent.name}」已消费完绑定知识库的登记片段，需要补素材或重新开放",
|
||
first_node_id, agent_id=agent.id, kb_ids=kb_ids))
|
||
elif demand > available_chunks:
|
||
issues.append(_issue("knowledge_capacity", "error",
|
||
f"Agent「{agent.name}」本计划需要生成 {demand} 段，但当前只有 "
|
||
f"{available_chunks} 个未消费片段；请补素材、释放片段或减少节点",
|
||
first_node_id, agent_id=agent.id, kb_ids=kb_ids,
|
||
available_chunks=available_chunks, required_nodes=demand))
|
||
if empty_kbs and valid_kbs and available_chunks > 0:
|
||
issues.append(_issue("knowledge_unverified", "warning",
|
||
f"Agent「{agent.name}」绑定的知识库 "
|
||
f"{','.join(str(kb.id) for kb in empty_kbs)} 文件/片段计数不完整",
|
||
first_node_id, agent_id=agent.id))
|
||
return issues, report
|
||
|
||
|
||
def _agent_kb_categories(agent: Agent) -> dict[int, list[int]]:
|
||
raw = dict(getattr(agent, "kb_category_ids", None) or {})
|
||
normalized: dict[int, list[int]] = {}
|
||
for key, value in raw.items():
|
||
try:
|
||
normalized[int(key)] = [int(item) for item in (value or [])]
|
||
except (TypeError, ValueError):
|
||
continue
|
||
return normalized
|
||
|
||
|
||
def _lightweight_knowledge_capacity(session: Session, agent: Agent,
|
||
assigned_nodes: list[dict[str, Any]]) -> tuple[list[dict], dict]:
|
||
issues: list[dict[str, Any]] = []
|
||
kb_ids = [int(kid) for kid in (agent.kb_ids or [])]
|
||
queries = [_node_knowledge_query(node) for node in assigned_nodes]
|
||
report = simulate_lightweight_plan(
|
||
session, kb_ids, queries,
|
||
excluded_chunk_ids=consumed_chunk_ids(session, int(agent.id or 0), kb_ids),
|
||
category_ids_by_kb=_agent_kb_categories(agent),
|
||
initial_top_k=6,
|
||
widened_top_k=_PLAN_SIMULATION_TOP_K,
|
||
char_budget=_PLAN_SOURCE_CHAR_BUDGET,
|
||
)
|
||
report.update({"mode": "keyword-light", "required_nodes": len(assigned_nodes),
|
||
"kb_ids": kb_ids})
|
||
shortfalls = list(report.get("shortfalls", []))
|
||
affected_node_ids = [str(node.get("node_id") or "") for node in assigned_nodes]
|
||
if int(report.get("total_chunks") or 0) <= 0:
|
||
first_node = assigned_nodes[0] if assigned_nodes else {}
|
||
issues.append(_issue(
|
||
"knowledge_exhausted", "error",
|
||
f"Agent「{agent.name}」在轻量模式没有可用源文件片段：请补素材、重新开放片段，"
|
||
"或启用本地向量索引后重建片段",
|
||
str(first_node.get("node_id") or ""), agent_id=agent.id, kb_ids=kb_ids,
|
||
affected_node_ids=affected_node_ids, total_chunks=0,
|
||
locked_or_consumed_required=len(assigned_nodes)))
|
||
return issues, report
|
||
|
||
capacity_failed = False
|
||
mismatch_count = 0
|
||
for shortfall in shortfalls:
|
||
index = int(shortfall.get("order") or 1) - 1
|
||
node = assigned_nodes[max(0, min(index, len(assigned_nodes) - 1))]
|
||
node_id = str(node.get("node_id") or "")
|
||
title = str(node.get("title") or "")[:80]
|
||
if int(shortfall.get("candidate_count") or 0) <= 0:
|
||
# 主题不匹配是节点级问题；但仍限制数量，避免一个长计划刷出大量重复阻断项。
|
||
if mismatch_count >= 5:
|
||
continue
|
||
code, message = "knowledge_mismatch", (
|
||
f"节点「{title}」在 Agent「{agent.name}」绑定知识库中没有关键词匹配素材；"
|
||
"请补充同主题素材，或调整该节点/Agent 资源绑定")
|
||
mismatch_count += 1
|
||
else:
|
||
code, message = "knowledge_capacity", (
|
||
f"节点「{title}」起无法继续预留知识片段：Agent「{agent.name}」本计划还有 "
|
||
f"{len(assigned_nodes) - index} 段，但素材池已不足；请补充更多来源或减少节点")
|
||
capacity_failed = True
|
||
issues.append(_issue(code, "error", message, node_id, agent_id=agent.id,
|
||
kb_ids=kb_ids, affected_node_ids=affected_node_ids, **{
|
||
"total_chunks": report.get("total_chunks"),
|
||
"remaining_chunks": report.get("remaining_chunks"),
|
||
"candidate_count": shortfall.get("candidate_count"),
|
||
}))
|
||
if capacity_failed:
|
||
# 后续节点失败只是同一容量缺口的连锁结果，审计报告仍保留在 knowledge_capacity 中。
|
||
break
|
||
return issues, report
|
||
|
||
|
||
def _knowledge_capacity_issues(session: Session, nodes: list[dict[str, Any]]) -> tuple[list[dict], dict]:
|
||
"""Plan-level admission control for all DSH Worker knowledge reservations."""
|
||
issues: list[dict[str, Any]] = []
|
||
reports: dict[str, Any] = {}
|
||
grouped: dict[int, list[dict[str, Any]]] = {}
|
||
for node in nodes:
|
||
agent_id = int(node.get("agent_id") or 0)
|
||
if agent_id:
|
||
grouped.setdefault(agent_id, []).append(node)
|
||
for agent_id, assigned in grouped.items():
|
||
agent = session.get(Agent, agent_id)
|
||
if not agent or agent.role != "production" or agent.status != "active":
|
||
continue
|
||
if not list(agent.kb_ids or []):
|
||
continue
|
||
if rag_backend.is_disabled():
|
||
agent_issues, report = _lightweight_knowledge_capacity(session, agent, assigned)
|
||
else:
|
||
agent_issues, report = _vector_knowledge_capacity(session, agent, assigned)
|
||
issues.extend(agent_issues)
|
||
reports[str(agent_id)] = report
|
||
return issues, reports
|
||
|
||
|
||
def build_plan(session: Session, campaign: Campaign, parsed: dict, template: WorkflowTemplate) -> dict:
|
||
"""Convert coordinator segments to a planner DAG proposal."""
|
||
raw_segments = list(parsed.get("segments") or [])
|
||
config = dict(default_template_config())
|
||
config.update(template.config or {})
|
||
max_segments = int(config.get("max_segments") or 80)
|
||
nodes: list[dict[str, Any]] = []
|
||
segments: list[dict[str, Any]] = []
|
||
edges: list[dict[str, str]] = []
|
||
prev_id = ""
|
||
used: dict[str, int] = {}
|
||
for index, item in enumerate(raw_segments, start=1):
|
||
node_id = f"n{index}"
|
||
tag = str(item.get("content_tag") or "心法").strip() or "心法"
|
||
agent_id, match_type, candidates, source_canvas_node_id = _match_plan_agent(
|
||
session, campaign, tag, used)
|
||
minutes = _time_range_minutes(
|
||
str(item.get("time_range") or ""),
|
||
int(campaign.duration_minutes or 0))
|
||
node = {
|
||
"node_id": node_id,
|
||
"order": index,
|
||
"title": str(item.get("title") or f"内容段{index}")[:200],
|
||
"content_tag": tag,
|
||
"time_range": str(item.get("time_range") or "")[:60],
|
||
"duration_minutes": minutes,
|
||
"mindmap_node": str(item.get("mindmap_node") or item.get("title") or "")[:1000],
|
||
"product_ref": str(item.get("product_ref") or "")[:200],
|
||
"agent_id": agent_id,
|
||
"agent_candidate_ids": candidates,
|
||
"agent_match_type": match_type,
|
||
"source_canvas_node_id": source_canvas_node_id,
|
||
"kb_category_ids": [],
|
||
"knowledge_requirements": [],
|
||
"required": True,
|
||
"enabled": True,
|
||
"risk_level": "low",
|
||
}
|
||
segment = {
|
||
"node_id": node_id,
|
||
"title": node["title"],
|
||
"content_tag": tag,
|
||
"time_range": node["time_range"],
|
||
"mindmap_node": node["mindmap_node"],
|
||
"product_ref": node["product_ref"],
|
||
"agent_id": agent_id,
|
||
}
|
||
nodes.append(node)
|
||
segments.append(segment)
|
||
if prev_id:
|
||
edges.append({"from": prev_id, "to": node_id})
|
||
prev_id = node_id
|
||
return {
|
||
"schema_version": "campaign-plan/v1",
|
||
"template_key": template.key,
|
||
"template_id": int(template.id or 0),
|
||
"mode": config.get("mode", "linear_dag"),
|
||
"objective": campaign.objective or "",
|
||
"success_criteria": campaign.success_criteria or "",
|
||
"duration_minutes": int(campaign.duration_minutes or 0),
|
||
"nodes": nodes,
|
||
"edges": edges,
|
||
"segments": segments,
|
||
"pending_confirms": list(parsed.get("pending_confirms") or []),
|
||
"notes": list(parsed.get("notes") or []),
|
||
"source": {
|
||
"provider": parsed.get("provider", ""),
|
||
"model": parsed.get("model", ""),
|
||
"trace_id": parsed.get("trace_id", ""),
|
||
},
|
||
}
|
||
|
||
|
||
def validate_plan(session: Session, campaign: Campaign, plan: dict, template: WorkflowTemplate) -> dict[str, Any]:
|
||
"""Deterministic validation. The LLM proposal is never trusted directly."""
|
||
config = dict(default_template_config())
|
||
config.update(template.config or {})
|
||
issues: list[dict[str, Any]] = []
|
||
nodes = list(plan.get("nodes") or [])
|
||
edges = list(plan.get("edges") or [])
|
||
if not nodes:
|
||
issues.append(_issue("empty_plan", "error", "计划没有任何内容节点"))
|
||
if len(nodes) > int(config.get("max_segments") or 80):
|
||
issues.append(_issue("too_many_segments", "error",
|
||
f"计划节点数 {len(nodes)} 超过模板上限 {config.get('max_segments')}"))
|
||
ids = [str(n.get("node_id") or "") for n in nodes]
|
||
if len(set(ids)) != len(ids) or any(not x for x in ids):
|
||
issues.append(_issue("node_id_invalid", "error", "节点 ID 为空或重复"))
|
||
required_tags = set(config.get("required_tags") or [])
|
||
allowed_tags = set(config.get("optional_tags") or [])
|
||
actual_tags = {str(n.get("content_tag") or "") for n in nodes}
|
||
missing_required = sorted(required_tags - actual_tags)
|
||
if missing_required:
|
||
issues.append(_issue("required_tag_missing", "error",
|
||
"模板必备节点缺失：" + "、".join(missing_required)))
|
||
if allowed_tags and required_tags is not None:
|
||
unknown = sorted(tag for tag in actual_tags if tag and tag not in allowed_tags)
|
||
if unknown:
|
||
issues.append(_issue("tag_not_in_template", "warning",
|
||
"出现模板可选标签之外的标签：" + "、".join(unknown)))
|
||
try:
|
||
waves = topo_waves(ids, edges) if ids else []
|
||
except ValueError as exc:
|
||
waves = []
|
||
issues.append(_issue("dag_invalid", "error", str(exc)))
|
||
for node in nodes:
|
||
nid = str(node.get("node_id") or "")
|
||
title = str(node.get("title") or "").strip()
|
||
if not title:
|
||
issues.append(_issue("title_missing", "error", "节点标题为空", nid))
|
||
if not str(node.get("mindmap_node") or "").strip():
|
||
issues.append(_issue("source_missing", "warning", "节点缺少导图原文", nid))
|
||
issues.extend(_knowledge_issues(session, int(node.get("agent_id") or 0), nid))
|
||
if str(node.get("product_ref") or "").strip():
|
||
issues.append(_issue("product_reference", "medium",
|
||
"包含产品引用，批准前请核对素材证据和合规口径", nid))
|
||
capacity_issues, knowledge_capacity = _knowledge_capacity_issues(session, nodes)
|
||
issues.extend(capacity_issues)
|
||
duration = int(campaign.duration_minutes or plan.get("duration_minutes") or 0)
|
||
minute_values = [int(n.get("duration_minutes") or 0) for n in nodes if n.get("duration_minutes") is not None]
|
||
parsed_count = sum(1 for n in nodes if n.get("duration_minutes") is not None)
|
||
if duration and parsed_count == len(nodes) and minute_values and sum(minute_values) != duration:
|
||
issues.append(_issue("duration_mismatch", "medium",
|
||
f"节点时长合计 {sum(minute_values)} 分钟，任务要求 {duration} 分钟"))
|
||
elif duration and parsed_count != len(nodes):
|
||
issues.append(_issue("duration_unverified", "warning", "部分节点时间范围无法机器解析，未能核对总时长"))
|
||
for item in plan.get("pending_confirms") or []:
|
||
issues.append(_issue("pending_confirm", "high", f"待确认：{str(item)[:160]}"))
|
||
errors = [x for x in issues if x["severity"] == "error"]
|
||
warnings = [x for x in issues if x["severity"] != "error"]
|
||
return {
|
||
"ok": not errors,
|
||
"errors": errors,
|
||
"warnings": warnings,
|
||
"issues": issues,
|
||
"knowledge_capacity": knowledge_capacity,
|
||
"waves": waves,
|
||
"checked_at": now(),
|
||
}
|
||
|
||
|
||
def _next_version(session: Session, campaign_id: int) -> int:
|
||
row = session.exec(select(CampaignPlanVersion.version).where(
|
||
CampaignPlanVersion.campaign_id == campaign_id
|
||
).order_by(CampaignPlanVersion.version.desc())).first()
|
||
# SQLModel returns scalar when selecting a single column.
|
||
return int(row or 0) + 1
|
||
|
||
|
||
def save_planner_result(session: Session, campaign: Campaign, parsed: dict,
|
||
template: WorkflowTemplate, *, advice: str = "",
|
||
created_by: int = 0) -> CampaignPlanVersion:
|
||
plan = build_plan(session, campaign, parsed, template)
|
||
validation = validate_plan(session, campaign, plan, template)
|
||
risks = [x for x in validation.get("issues", []) if x.get("severity") in {"high", "medium"}]
|
||
for node in plan.get("nodes", []):
|
||
node_risks = [x for x in risks if x.get("node_id") == node["node_id"]]
|
||
node["risk_level"] = "high" if any(x["severity"] == "high" for x in node_risks) else (
|
||
"medium" if node_risks else "low")
|
||
version = CampaignPlanVersion(
|
||
campaign_id=campaign.id,
|
||
version=_next_version(session, campaign.id),
|
||
source="dsh",
|
||
template_id=int(template.id or 0),
|
||
template_key=template.key,
|
||
status="waiting_review" if validation.get("ok") else "failed",
|
||
plan=plan,
|
||
validation=validation,
|
||
summary=f"{len(plan.get('nodes') or [])} 个节点，{len(plan.get('edges') or [])} 条依赖；"
|
||
f"{len(validation.get('errors') or [])} 个错误，{len(validation.get('warnings') or [])} 个风险/提示",
|
||
risks=risks,
|
||
advice=advice[:1000],
|
||
created_by=created_by,
|
||
)
|
||
session.add(version)
|
||
session.commit()
|
||
session.refresh(version)
|
||
return version
|
||
|
||
|
||
def run_plan_job(session: Session, campaign_id: int, *, template_id: int = 0,
|
||
advice: str = "", created_by: int = 0) -> dict[str, Any]:
|
||
campaign = session.get(Campaign, campaign_id)
|
||
if not campaign:
|
||
raise ValueError(f"任务不存在：{campaign_id}")
|
||
if campaign.status == "done":
|
||
raise ValueError("任务已交付，不能重新规划")
|
||
template = get_template(session, template_id)
|
||
campaign.status = "planning"
|
||
campaign.notes = list(campaign.notes or [])
|
||
session.add(campaign)
|
||
session.commit()
|
||
progress.publish(int(campaign.owner_user_id or 0), "plan_start",
|
||
campaign_id=campaign_id, name=campaign.name, backend="dsh",
|
||
template=template.key)
|
||
parsed = parse_mindmap(session, campaign, user_advice=advice)
|
||
if parsed.get("status") != "ok" or not parsed.get("segments"):
|
||
campaign.status = "parse_failed"
|
||
campaign.notes = [f"DSH Planner 规划失败：{str(parsed.get('error') or '未产出节点')[:200]}"]
|
||
session.add(campaign)
|
||
session.commit()
|
||
progress.publish(int(campaign.owner_user_id or 0), "plan_failed",
|
||
campaign_id=campaign_id, error=str(parsed.get("error") or "未产出节点")[:200])
|
||
return {"status": "failed", "error": parsed.get("error") or "未产出节点"}
|
||
version = save_planner_result(session, campaign, parsed, template,
|
||
advice=advice, created_by=created_by)
|
||
notes = list(version.plan.get("notes") or [])
|
||
if version.status == "waiting_review":
|
||
campaign.status = "plan_review"
|
||
if version.risks:
|
||
notes.append(f"计划 v{version.version} 有 {len(version.risks)} 个中高风险项，等待人工确认")
|
||
progress_event = "plan_done"
|
||
else:
|
||
# LLM 已产出草案，但确定性校验未过。该版本只作为失败审计记录，
|
||
# 不物化为 Segment，也不能进入 Worker；工作台修正资源后必须重新规划生成新版本。
|
||
campaign.status = "plan_failed"
|
||
errors = list((version.validation or {}).get("errors") or [])
|
||
for item in errors[:5]:
|
||
notes.append(f"计划 v{version.version} 校验未过：{item.get('message', '')}")
|
||
progress_event = "plan_failed"
|
||
campaign.notes = notes[-30:]
|
||
campaign.has_failure = version.status == "failed"
|
||
session.add(campaign)
|
||
session.commit()
|
||
progress.publish(int(campaign.owner_user_id or 0), progress_event,
|
||
campaign_id=campaign_id, plan_version_id=version.id,
|
||
version=version.version, status=version.status,
|
||
risks=len(version.risks or []))
|
||
return {"status": version.status, "plan_version_id": version.id, "version": version.version}
|
||
|
||
|
||
def approve_plan(session: Session, plan_version_id: int, *, reviewer_user_id: int = 0,
|
||
review_note: str = "") -> CampaignPlanVersion:
|
||
version = session.get(CampaignPlanVersion, plan_version_id)
|
||
if not version:
|
||
raise ValueError("计划版本不存在")
|
||
campaign = session.get(Campaign, version.campaign_id)
|
||
if not campaign:
|
||
raise ValueError("内容任务不存在")
|
||
if version.status != "waiting_review":
|
||
raise ValueError(f"当前计划状态 {version.status} 不能批准；请修正后重新规划")
|
||
if campaign.status in {"done", "running"}:
|
||
raise ValueError("任务已交付或正在运行，不能更换计划")
|
||
template = get_template(session, int(version.template_id or 0), version.template_key)
|
||
validation = validate_plan(session, campaign, version.plan or {}, template)
|
||
if not validation.get("ok"):
|
||
version.status = "failed"
|
||
version.validation = validation
|
||
session.add(version)
|
||
session.commit()
|
||
raise ValueError("计划校验未通过：" + "；".join(x["message"] for x in validation.get("errors", [])[:5]))
|
||
nodes = list((version.plan or {}).get("nodes") or [])
|
||
unresolved = [n for n in nodes if not int(n.get("agent_id") or 0)]
|
||
if unresolved:
|
||
raise ValueError("仍有节点未分配 Agent，请先在计划中调整后再批准")
|
||
existing = list(session.exec(select(Segment).where(Segment.campaign_id == campaign.id)).all())
|
||
generated = [s for s in existing if (s.status not in {"pending", "skipped"} or s.artifact_path or s.generation_history)]
|
||
if generated:
|
||
raise ValueError("已有内容产出，不能直接覆盖计划；请创建新版本任务或走重规划审批")
|
||
for old in existing:
|
||
session.delete(old)
|
||
session.commit()
|
||
for index, node in enumerate(nodes, start=1):
|
||
session.add(Segment(
|
||
campaign_id=campaign.id,
|
||
node_id=str(node.get("node_id") or f"n{index}"),
|
||
agent_id=int(node.get("agent_id") or 0),
|
||
agent_assignment_mode="auto",
|
||
title=str(node.get("title") or f"内容段{index}"),
|
||
content_tag=str(node.get("content_tag") or ""),
|
||
time_range=str(node.get("time_range") or ""),
|
||
mindmap_node=str(node.get("mindmap_node") or node.get("title") or ""),
|
||
product_ref=str(node.get("product_ref") or ""),
|
||
kb_category_ids=list(node.get("kb_category_ids") or []),
|
||
status="pending",
|
||
))
|
||
campaign.canvas_snapshot = {
|
||
"nodes": [
|
||
{
|
||
"node_id": n["node_id"],
|
||
"id": n["node_id"],
|
||
"agent_id": int(n.get("agent_id") or 0),
|
||
"content_tag": n.get("content_tag") or "",
|
||
"enabled": bool(n.get("enabled", True)),
|
||
"title": n.get("title") or "",
|
||
}
|
||
for n in nodes
|
||
],
|
||
"edges": list((version.plan or {}).get("edges") or []),
|
||
"plan_version_id": version.id,
|
||
"template_key": version.template_key,
|
||
"waves": validation.get("waves") or [],
|
||
}
|
||
campaign.task_spec = list((version.plan or {}).get("segments") or [])
|
||
campaign.pending_confirms = list((version.plan or {}).get("pending_confirms") or [])
|
||
campaign.notes = list((version.plan or {}).get("notes") or [])
|
||
campaign.approved_plan_version_id = version.id
|
||
campaign.status = "parsed"
|
||
campaign.has_failure = False
|
||
for old in session.exec(select(CampaignPlanVersion).where(
|
||
CampaignPlanVersion.campaign_id == campaign.id,
|
||
CampaignPlanVersion.id != version.id,
|
||
CampaignPlanVersion.status.in_(["waiting_review", "rejected", "approved"]))).all():
|
||
old.status = "superseded"
|
||
session.add(old)
|
||
version.status = "approved"
|
||
version.reviewed_by = reviewer_user_id
|
||
version.review_note = review_note[:1000]
|
||
version.reviewed_at = now()
|
||
version.validation = validation
|
||
session.add(version)
|
||
session.add(campaign)
|
||
session.commit()
|
||
session.refresh(version)
|
||
progress.publish(int(campaign.owner_user_id or 0), "plan_approved",
|
||
campaign_id=campaign.id, plan_version_id=version.id,
|
||
version=version.version)
|
||
return version
|
||
|
||
|
||
def reject_plan(session: Session, plan_version_id: int, *, reviewer_user_id: int = 0,
|
||
review_note: str = "") -> CampaignPlanVersion:
|
||
version = session.get(CampaignPlanVersion, plan_version_id)
|
||
if not version:
|
||
raise ValueError("计划版本不存在")
|
||
if version.status not in {"waiting_review", "failed"}:
|
||
raise ValueError(f"当前计划状态 {version.status} 不能驳回")
|
||
version.status = "rejected"
|
||
version.reviewed_by = reviewer_user_id
|
||
version.review_note = review_note[:1000] or "人工驳回计划"
|
||
version.reviewed_at = now()
|
||
session.add(version)
|
||
campaign = session.get(Campaign, version.campaign_id)
|
||
if campaign and campaign.status in {"plan_review", "plan_failed"}:
|
||
campaign.status = "draft"
|
||
session.add(campaign)
|
||
session.commit()
|
||
session.refresh(version)
|
||
return version
|