105 lines
4.5 KiB
Python
105 lines
4.5 KiB
Python
"""全局总监 Agent（画布级）：每段生成后判断是否跑偏，跑偏则带指导重做。
|
||
|
||
- always：每段生成完必审
|
||
- on_drift：QC 有问题/质检反复（round>1）时才审（默认，省点数）
|
||
- off：不启用
|
||
|
||
总监不写内容，只输出 JSON：{"on_track": true} 或
|
||
{"on_track": false, "reason": "为什么偏了", "guidance": "怎么拉回来"}。
|
||
guidance 会注入该段下一轮生成提示，最多重做 1 次，防止无限循环烧点数。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
from sqlmodel import Session
|
||
|
||
from ..models import Agent, Campaign, Canvas, Segment
|
||
|
||
SUPERVISOR_INSTRUCTIONS = (
|
||
"你是内容生产线的全局总监，负责保证每一段内容不跑偏。"
|
||
"给你任务目标、验收标准、本段应讲的主题和已生成的正文。"
|
||
"只判断：1) 正文是否围绕本段主题 2) 是否符合任务目标和验收标准 3) 是否夹带无关内容。"
|
||
"只输出 JSON：{\"on_track\": true} 或 {\"on_track\": false, \"reason\": \"跑偏原因（一句话）\", "
|
||
"\"guidance\": \"给写手的具体拉回指导（一两句话，怎么改）\"}"
|
||
)
|
||
|
||
|
||
def _canvas_for(campaign: Campaign, session: Session) -> Canvas | None:
|
||
"""任务运行时以画布快照为准；快照存了 supervisor 配置才能启用。"""
|
||
snap = campaign.canvas_snapshot or {}
|
||
if not isinstance(snap, dict):
|
||
return None
|
||
supervisor_id = snap.get("supervisor_agent_id")
|
||
mode = str(snap.get("supervisor_mode") or "on_drift")
|
||
if not supervisor_id or mode == "off":
|
||
return None
|
||
# 快照配置即可驱动审核；Canvas 实例仅用于传参，不落库
|
||
return Canvas(id=0, name="runtime", nodes=[], edges=[],
|
||
supervisor_agent_id=int(supervisor_id), supervisor_mode=mode,
|
||
organization_id=int(campaign.organization_id or 1))
|
||
|
||
|
||
def _segment_text(seg: Segment) -> str:
|
||
"""读取段落正文：优先产物文件，其次锚点卡标题。"""
|
||
from pathlib import Path
|
||
if seg.artifact_path and Path(seg.artifact_path).is_file():
|
||
try:
|
||
return Path(seg.artifact_path).read_text(encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
title = str((seg.anchor_card or {}).get("title") or seg.title or "")
|
||
return title
|
||
|
||
|
||
def supervisor_check(session: Session, seg: Segment, campaign: Campaign) -> dict:
|
||
"""总监审核一个刚生成完的段。返回 {"checked": bool, "on_track": bool, "guidance": str, "reason": str}。"""
|
||
canvas = _canvas_for(campaign, session)
|
||
if not canvas:
|
||
return {"checked": False}
|
||
supervisor = session.get(Agent, int(canvas.supervisor_agent_id or 0))
|
||
if not supervisor or supervisor.status != "active":
|
||
return {"checked": False}
|
||
mode = canvas.supervisor_mode
|
||
# on_drift 模式：只审可疑段（质检发现问题或已重做过一轮）
|
||
if mode == "on_drift" and seg.status == "approved" and int(seg.round or 1) <= 1:
|
||
return {"checked": False}
|
||
|
||
from ..llm import chat_json
|
||
prompt = (
|
||
f"任务目标：{campaign.objective}\n"
|
||
f"验收标准：{campaign.success_criteria}\n"
|
||
f"本段主题：{seg.title}\n"
|
||
f"已生成正文：\n{_segment_text(seg)[:4000]}"
|
||
)
|
||
try:
|
||
result = chat_json(
|
||
session,
|
||
[{"role": "system", "content": SUPERVISOR_INSTRUCTIONS},
|
||
{"role": "user", "content": prompt}],
|
||
provider_id=0, model="",
|
||
campaign_id=int(campaign.id or 0), segment_id=int(seg.id or 0),
|
||
agent_id=int(supervisor.id or 0), agent_name=f"{supervisor.name}(总监)",
|
||
user_id=int(campaign.owner_user_id or 0),
|
||
organization_id=int(campaign.organization_id or 1),
|
||
max_retries=1,
|
||
)
|
||
except Exception:
|
||
# 总监审核自身失败不阻断主流程
|
||
return {"checked": False}
|
||
if result.get("status") != "ok":
|
||
return {"checked": False}
|
||
try:
|
||
verdict = result.get("data") if isinstance(result.get("data"), dict) else json.loads(result.get("text") or "{}")
|
||
except (json.JSONDecodeError, TypeError):
|
||
verdict = {}
|
||
on_track = bool(verdict.get("on_track", True))
|
||
if on_track:
|
||
return {"checked": True, "on_track": True}
|
||
return {"checked": True, "on_track": False,
|
||
"reason": str(verdict.get("reason") or "内容偏离本段主题或任务目标")[:200],
|
||
"guidance": str(verdict.get("guidance") or "回到本段主题，紧扣任务目标与验收标准重写")[:300]}
|
||
|
||
|
||
__all__ = ["supervisor_check"]
|