211 lines
9.6 KiB
Python
211 lines
9.6 KiB
Python
"""
|
||
app/services/fission_pipeline.py — 第11环裂变 编排层（从 fission_service 拆出）
|
||
|
||
Celery 内调用：一次 LLM 出 N 套完整笔记包 → 解析/兜底 → 评分@80 → 落 FissionNote
|
||
→ 逐套生图存 images_json。拆分原因：fission_service.py 超 200 行红线上限。
|
||
🔴 生图引擎零改动：复用单篇 generate_storyboard_images（codeproxy edits 带产品参考图）。
|
||
🔴 评分沿用 llm_score_copy，合格线 QUALITY_PASS_SCORE=80（禁改）。
|
||
🔴 FISSION_SYSTEM 作 messages[0].role=system 传（chat_complete 是 OpenAI 兼容）。
|
||
"""
|
||
import logging
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
MAX_FANOUT = 5 # 每用户并发上限5(红线)；裂变 N 套直出受同等约束
|
||
|
||
|
||
def _parse_fission_response(
|
||
raw: str, source_note: dict, product: dict, note_count: int, image_count: int,
|
||
) -> list[dict]:
|
||
"""解析 LLM 返回的 N 套笔记；不可解析则品类兜底（不卡用户）。
|
||
|
||
解析链：parse_json_array(容错markdown) → notes_array_from_parsed(拎数组) →
|
||
空则 build_fallback_notes 品类草稿。每套补 imagePlan/tags 归一化。
|
||
"""
|
||
import json as _json
|
||
from app.services.ai_engine._text_prompt import parse_json_array
|
||
from app.services.ai_engine.fission_prompt import (
|
||
normalize_tags, notes_array_from_parsed,
|
||
)
|
||
from app.services.ai_engine.fission_fallback import (
|
||
build_fallback_image_plan, build_fallback_notes,
|
||
)
|
||
|
||
def _fallback(reason: str) -> list[dict]:
|
||
notes = build_fallback_notes(source_note, product, note_count, image_count)
|
||
for note in notes:
|
||
note["is_fallback"] = True
|
||
note["fallback_reason"] = reason
|
||
return notes
|
||
|
||
notes = notes_array_from_parsed(parse_json_array(raw))
|
||
if not notes:
|
||
# parse_json_array 只认数组；再试整段当对象解析（容错单对象返回）
|
||
try:
|
||
notes = notes_array_from_parsed(_json.loads(raw))
|
||
except (ValueError, TypeError):
|
||
notes = []
|
||
if not notes:
|
||
logger.warning("裂变 LLM 返回不可解析，启用品类兜底。raw[:120]=%s", str(raw)[:120])
|
||
return _fallback("llm_parse_failed")
|
||
|
||
out: list[dict] = []
|
||
for n in notes:
|
||
if not isinstance(n, dict):
|
||
continue
|
||
n["tags"] = normalize_tags(n.get("tags", []), n.get("keywords", []))
|
||
plan = n.get("imagePlan")
|
||
if not isinstance(plan, list) or len(plan) != image_count:
|
||
n["imagePlan"] = build_fallback_image_plan(n, image_count)
|
||
out.append(n)
|
||
return out or _fallback("llm_empty_notes")
|
||
|
||
|
||
def _score_notes(
|
||
clients, notes: list[dict], source_note: dict, banned: list[str],
|
||
product: dict | None = None,
|
||
) -> None:
|
||
"""对每套笔记 LLM 评分（限2并发），结果就地写回 note['_score']/_passed/_block。
|
||
|
||
D1修复：product 传给 llm_score_copy → build_score_prompt 组装【产品语境】，
|
||
评委不再盲评，买点转化/痛点人群精准两维能基于产品信息给分。
|
||
Celery 同步环境：用 asyncio.run 跑一个内部 gather（信号量限2并发）。
|
||
"""
|
||
import asyncio
|
||
from app.services.ai_engine.llm_scorer import llm_score_copy, ScoringUnavailableError
|
||
from app.services.ai_engine.constants import QUALITY_PASS_SCORE
|
||
|
||
async def _run() -> None:
|
||
sem = asyncio.Semaphore(2)
|
||
|
||
async def _one(note: dict) -> None:
|
||
copy = {
|
||
"title": note.get("title", ""),
|
||
"content": note.get("content", ""),
|
||
"tags": note.get("tags", []),
|
||
}
|
||
async with sem:
|
||
try:
|
||
r = await llm_score_copy(
|
||
clients, copy, product or source_note, banned,
|
||
pass_score=QUALITY_PASS_SCORE,
|
||
)
|
||
except ScoringUnavailableError as exc:
|
||
# B1红线对齐：评分两通道都挂≠质量差，明确标记不可用，绝不静默记0分糊弄
|
||
logger.error("裂变评分通道不可用(非质量问题): %s", exc)
|
||
note["_scoring_unavailable"] = True
|
||
note["_score"] = 0
|
||
note["_block"] = False
|
||
note["_passed"] = False
|
||
return
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("裂变评分异常，记0分不达标: %s", exc)
|
||
r = {"score": 0, "passed": False, "banned_words_found": []}
|
||
note["_score"] = int(r.get("score", 0))
|
||
note["_block"] = bool(r.get("banned_words_found"))
|
||
# 过线 = score≥80 且无硬拦违禁词
|
||
note["_passed"] = bool(r.get("passed")) and not note["_block"]
|
||
|
||
await asyncio.gather(*(_one(n) for n in notes))
|
||
|
||
asyncio.run(_run())
|
||
# 整批评分全因通道不可用而失败 → 抛错让上层(execute_fission_pipeline)感知,
|
||
# 标记任务"评分服务不可用"而非误判成完成/质量差
|
||
if notes and all(n.get("_scoring_unavailable") for n in notes):
|
||
raise ScoringUnavailableError(
|
||
f"裂变 {len(notes)} 套笔记评分全部失败：评分通道(apiports+codeproxy)均不可用")
|
||
|
||
|
||
def execute_fission_pipeline(db: Session, clients, fission_id: int, source_task_id: int) -> dict:
|
||
"""裂变主编排（Celery 内调用，clients 已构建好）。
|
||
|
||
流程：查 FissionTask+源产品 → 一次 chat_complete 出 N 套 → 解析/兜底
|
||
→ 每套评分@80 → 按分排序取 N 套（不够用兜底补，不达标标 needs_optimization）
|
||
→ 落 FissionNote → 逐套生图存 images_json → 回写 FissionTask.status。
|
||
"""
|
||
import json as _json
|
||
|
||
from app.models.fission import FissionTask, FissionNote
|
||
from app.models.task import GenerationTask
|
||
from app.models.product import Product
|
||
from app.workers.pipeline_steps import build_product_dict, load_benchmark_features
|
||
from app.services.ai_engine.fission_prompt import build_fission_prompt
|
||
|
||
ft = db.query(FissionTask).filter(FissionTask.id == fission_id).first()
|
||
if not ft:
|
||
return {"fission_id": fission_id, "status": "not_found"}
|
||
|
||
n = max(1, min(ft.fanout_count or 3, MAX_FANOUT))
|
||
try:
|
||
source_note = _json.loads(ft.source_note) if ft.source_note else {}
|
||
except (ValueError, TypeError):
|
||
source_note = {}
|
||
|
||
src = db.query(GenerationTask).filter(GenerationTask.id == source_task_id).first()
|
||
product_row = db.query(Product).filter(Product.id == src.product_id).first() if src else None
|
||
if not src or not product_row:
|
||
ft.status = "failed"; db.commit()
|
||
return {"fission_id": fission_id, "status": "failed", "reason": "源任务或产品缺失"}
|
||
|
||
product = build_product_dict(product_row)
|
||
product["benchmark_refs"] = load_benchmark_features(db, src, ft.workspace_id)
|
||
image_count = max(1, src.image_count or 3)
|
||
banned = [] # 违禁词分级表后续接入；评分器内置 BANNED_WORDS_DEFAULT 已兜底
|
||
|
||
# 一次 LLM 出 N 套（FISSION_SYSTEM 作 messages[0].role=system）
|
||
from app.services.ai_engine.fission_prompt import FISSION_SYSTEM
|
||
user_prompt = build_fission_prompt(
|
||
source_note, product, ft.reference_level, n, image_count,
|
||
)
|
||
# max_tokens 按套数缩放：每套完整笔记包约 1200 token，留足余量
|
||
max_tokens = min(8192, 1800 + n * 1400)
|
||
raw = ""
|
||
try:
|
||
import asyncio
|
||
raw = asyncio.run(clients.chat_complete(
|
||
messages=[{"role": "system", "content": FISSION_SYSTEM},
|
||
{"role": "user", "content": user_prompt}],
|
||
model=clients._model, max_tokens=max_tokens, temperature=0.8,
|
||
))
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("裂变 LLM 调用失败，启用品类兜底: %s", exc)
|
||
|
||
notes = _parse_fission_response(raw, source_note, product, n, image_count)
|
||
# D1修复：传 product 给评委，不再盲评（买点转化/痛点人群精准需要产品语境）
|
||
_score_notes(clients, notes, source_note, banned, product=product)
|
||
|
||
# 排序：过线优先，再按分降序；取前 N 套（不足用兜底草稿补齐）
|
||
notes.sort(key=lambda x: (x.get("_passed", False), x.get("_score", 0)), reverse=True)
|
||
chosen = notes[:n]
|
||
|
||
saved_ids: list[int] = []
|
||
for seq, note in enumerate(chosen):
|
||
passed = bool(note.get("_passed"))
|
||
fn = FissionNote(
|
||
fission_id=fission_id, workspace_id=ft.workspace_id, seq=seq,
|
||
note_json=_json.dumps(note, ensure_ascii=False),
|
||
score=int(note.get("_score", 0)),
|
||
passed=1 if passed else 0,
|
||
needs_optimization=0 if passed else 1, # 不达标不丢弃，标降级草稿
|
||
dimension=str(note.get("dimension", ""))[:64],
|
||
status="scored",
|
||
)
|
||
db.add(fn); db.flush()
|
||
saved_ids.append(fn.id)
|
||
db.commit()
|
||
|
||
logger.info("裂变出 %s 套已落库 fission=%s ids=%s", len(saved_ids), fission_id, saved_ids)
|
||
|
||
# 逐套生图（复用单篇引擎），存 images_json
|
||
from app.services.fission_images import generate_fission_images
|
||
generate_fission_images(db, clients, ft, product, image_count, saved_ids)
|
||
|
||
has_fallback = any(bool(note.get("is_fallback")) for note in chosen)
|
||
# status 字段是 varchar(20)，"completed_with_fallback"(23字符)会撑爆触发 DataError 1406，
|
||
# 故用短码 "completed_fb"(倩倩姐2026-06-30修)；语义=完成但含降级草稿。
|
||
ft.status = "completed_fb" if has_fallback else "completed"
|
||
db.commit()
|
||
return {"fission_id": fission_id, "status": ft.status, "note_ids": saved_ids}
|