azmat1127/yingqing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai): 实体提取解耦 — 定稿剧本后独立提取角色/场景 + 商品参考图永远带上

Browse Source 
根因:实体跟脚本生成耦合,模型不稳(尤其改稿 revise)常丢实体 → 角色提不出、
参考图全空 → 故事板/视频退化纯文生图(用户 demo 全错的真因)。

- 新 skill ecommerce-entity-extract(按 skill-creator 法写):只提角色+场景,
  角色铁律=穿整套衣服/空手/不带与商品同类物/不暴露;商品不提。4 项目实测稳定产出。
- extract_entities_for_project + /extract-entities 端点:读定稿剧本→提实体→
  落库覆盖 metadata + 回填每镜 entity_refs(提取为唯一权威来源,脚本生成完全不动)。
- 商品参考图永远带上:_storyboard_reference_images(视频继承)改为无条件带
  商品三视图→主图兜底,不再依赖 entity_refs 里有没有商品实体。
- 创建商品后端强制要主图(堵兜底洞:无图商品会让下游参考图彻底落空)。

实测:杂鱼煲(原实体0)→6实体,每镜参考图齐(角色+商品+场景),不再文生图。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
seaislee1209
2026-06-19 17:37:18 +08:00
co-authored by Claude Opus 4.8
parent 901d6920f1
commit 827fa7fa54
6 changed files with 502 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/backend/apps/ai/services.py
+216 -17
View File
@@ -221,6 +221,189 @@ def extract_cast_and_scenes(*, project, user, content: str) -> dict:
return empty
def _load_skill_system_prompt(name: str) -> str:
"""读取 skills/<name>/SKILL.md + references/*.md 拼成系统提示词(领域知识)。缺文件返回空串,不致命。"""
from pathlib import Path
from django.conf import settings
skill_dir = Path(settings.BASE_DIR).parent.parent / "skills" / name
parts: list[str] = []
main = skill_dir / "SKILL.md"
if main.exists():
parts.append(main.read_text(encoding="utf-8"))
ref_dir = skill_dir / "references"
if ref_dir.exists():
for ref in sorted(ref_dir.glob("*.md")):
parts.append(f"\n\n===== references/{ref.name} =====\n\n{ref.read_text(encoding='utf-8')}")
return "\n".join(parts)
def _normalize_extracted_entities(items: object) -> list[dict]:
"""整理提取出的 entities:只留 character/scene(绝不收 product),保留模型给的 id(供 segment refs 对应),
按 id 去重,角色/场景各最多 6 个,补 ref_index。"""
out: list[dict] = []
if not isinstance(items, list):
return out
seen_ids: set[str] = set()
n_char = n_scene = 0
for it in items:
if not isinstance(it, dict):
continue
typ = str(it.get("type") or "").strip()
if typ not in ("character", "scene"): # 丢弃 product / 未知类型
continue
name = str(it.get("name") or "").strip()
eid = str(it.get("id") or "").strip()
if not name or not eid or eid in seen_ids:
continue
if typ == "character":
n_char += 1
if n_char > 6:
continue
else:
n_scene += 1
if n_scene > 6:
continue
seen_ids.add(eid)
out.append(
{
"id": eid,
"type": typ,
"name": name,
"visual_prompt": str(it.get("visual_prompt") or "").strip(),
"ref_index": len(out) + 1,
}
)
return out
def _normalize_extracted_segment_refs(items: object, valid_ids: set[str]) -> list[dict]:
"""整理每镜 refs:只留指向存活实体(角色/场景)的 id,丢掉被剔除的(如商品/无效 id)。"""
out: list[dict] = []
if not isinstance(items, list):
return out
for it in items:
if not isinstance(it, dict):
continue
idx = it.get("index")
if not isinstance(idx, int):
continue
raw = it.get("entity_refs") or it.get("refs") or []
refs = [r for r in raw if isinstance(r, str) and r in valid_ids]
out.append({"index": idx, "entity_refs": refs})
return out
def extract_entities_for_project(*, project, user) -> dict:
"""独立实体提取步(剧本定稿后、进资产阶段时跑)。读已定稿(或最新)脚本 → 调 ecommerce-entity-extract
skill(一个文本模型)→ 产出角色 / 场景 entities + 每镜 entity_refs。**不提商品**(商品由参考图层用真实
主图 / 三视图注入)。落库:覆盖 project.metadata(cast/cast_prompts/scenes/scene_prompts/script_entities)
+ 回填每条 ScriptSegment.entity_refs。提取是实体的唯一权威来源,覆盖脚本生成期可能产出的旧实体。
走完整 AITask + 计费闭环(reserve→charge/release,任务类型 script_optimization)。失败抛 ValueError,
由端点转成用户可读错误(与脚本生成的 best-effort 不同:这是用户主动点的步骤,要让他知道成没成)。
返回 {entities, segments, cast, cast_prompts, scenes, scene_prompts}。"""
from apps.projects.models import ScriptVersion
script = (
ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first()
or ScriptVersion.objects.filter(project=project).order_by("-created_at").first()
)
if script is None:
raise ValueError("请先生成并定稿脚本,再提取角色 / 场景")
segments = list(script.segments.order_by("sort_order"))
if not segments:
raise ValueError("脚本没有分镜,无法提取")
model_config = get_default_model(ModelConfig.Capability.TEXT)
if model_config is None:
raise ValueError("没有可用的文本模型")
product = project.product
if product is not None:
sp = "、".join([p.title for p in product.selling_points.all()[:5]])
prod_line = f"名称：{product.title}\n品类：{product.category or '未填'}\n卖点：{sp or '未填'}"
else:
prod_line = "（无商品信息）"
seg_lines = [
f"镜{i} role={s.role or ''} narration={(s.narration or '').strip()} visual={(s.visual_prompt or '').strip()}"
for i, s in enumerate(segments)
]
user_msg = (
f"商品信息：\n{prod_line}\n\n分镜脚本（共 {len(segments)} 镜，index 从 0 开始）：\n" + "\n".join(seg_lines)
)
system = _load_skill_system_prompt("ecommerce-entity-extract")
messages = [{"role": "system", "content": system}, {"role": "user", "content": user_msg}]
try:
task = create_ai_task(
project=project,
user=user,
task_type=AITask.Type.SCRIPT_OPTIMIZATION,
model_config=model_config,
request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "messages": messages},
)
except Exception as exc: # 余额不足等预扣失败
raise ValueError("额度不足,无法提取(请先充值)") from exc
reservation = task.credit_reservation
try:
task.status = AITask.Status.SUBMITTED
task.submitted_at = timezone.now()
task.save(update_fields=["status", "submitted_at", "updated_at"])
provider = build_provider(model_config)
response = provider.chat_completion(model=model_config.name, endpoint=model_config.endpoint, messages=messages)
text = provider.extract_text(response)
with transaction.atomic():
task.status = AITask.Status.SUCCEEDED
task.response_payload = response
task.actual_cost = task.estimated_cost
task.completed_at = timezone.now()
task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
except Exception as exc:
with transaction.atomic():
task.status = AITask.Status.FAILED
task.error_message = str(exc)
task.completed_at = timezone.now()
task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
release_credit(reservation=reservation, reason=str(exc))
raise ValueError("提取调用失败,请重试") from exc
match = re.search(r"\{.*\}", text, re.DOTALL)
if not match:
raise ValueError("提取结果解析失败(模型未返回有效 JSON),请重试")
try:
data = json.loads(match.group(0))
except Exception as exc:
raise ValueError("提取结果解析失败,请重试") from exc
entities = _normalize_extracted_entities(data.get("entities"))
if not entities:
raise ValueError("没有从脚本里识别到角色 / 场景,可调整脚本后重试")
seg_refs = _normalize_extracted_segment_refs(data.get("segments"), {e["id"] for e in entities})
# 落库:覆盖 project.metadata + 回填 adopted 脚本每镜 entity_refs(提取为唯一权威来源)
from apps.ai.script_agent import _map_entities_to_project_metadata
_map_entities_to_project_metadata(project, entities)
refs_by_index = {r["index"]: r["entity_refs"] for r in seg_refs}
for i, seg in enumerate(segments):
new_refs = refs_by_index.get(i, [])
if (seg.entity_refs or []) != new_refs:
seg.entity_refs = new_refs
seg.save(update_fields=["entity_refs", "updated_at"])
md = project.metadata or {}
return {
"entities": entities,
"segments": seg_refs,
"cast": md.get("cast", []),
"cast_prompts": md.get("cast_prompts", {}),
"scenes": md.get("scenes", []),
"scene_prompts": md.get("scene_prompts", {}),
}
def split_script_into_segments(content: str, count: int = 4) -> list[str]:
"""把一段脚本稳健地拆成 `count` 个分镜文本,保证每镜都非空、且所有内容都被分配到某一镜。
@@ -758,9 +941,32 @@ def submit_storyboard(*, project, user, prompt: str = "") -> StoryboardVersion:
_ENTITY_TYPE_CN = {"character": "角色", "scene": "场景", "product": "商品"}
def _product_reference_image(project, groups: list | None = None) -> dict | None:
"""商品参考图:优先已采用的商品三视图(product 基础资产组)→ 否则商品真实主图。无图返回 None。
(商品是预创建的真实商品,不从脚本提取;每镜参考图都无条件带上它。)"""
if groups is None:
groups = list(
project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset")
)
pg = next(
(g for g in groups if g.kind == BaseAssetGroup.Kind.PRODUCT and g.adopted_asset_id),
None,
)
if pg is not None:
url = _asset_preview_url(pg.adopted_asset)
if url:
return {"url": url, "label": "商品", "type": "product"}
cover = _product_cover_url(project.product)
if cover:
return {"url": cover, "label": "商品", "type": "product"}
return None
def _storyboard_reference_images(project, segment) -> list[dict]:
"""按本镜 entity_refs 取参考图(角色/场景/商品的已采用基础资产),供 gpt-image-2 多图合成 @图N。
返回 [{url,label,type}],最多 4 张;无匹配时兜底商品组。依赖脚本 agent 落进 metadata 的 script_entities。"""
"""按本镜 entity_refs 取参考图(角色 / 场景 已采用基础资产)+ **无条件带上商品参考图**,
供 gpt-image-2 多图合成 @图N。返回 [{url,label,type}],最多 4 张(角色/场景 ≤3 + 商品 1)。
商品不靠 entity_refs(预创建真实商品、不从脚本提),统一用商品三视图 / 主图带上,从根上保证
商品参考永不缺失。依赖提取步落进 metadata 的 script_entities。"""
entities = {
e.get("id"): e
for e in (project.metadata or {}).get("script_entities", [])
@@ -769,18 +975,17 @@ def _storyboard_reference_images(project, segment) -> list[dict]:
kind_by_type = {
"character": BaseAssetGroup.Kind.PERSON,
"scene": BaseAssetGroup.Kind.SCENE,
"product": BaseAssetGroup.Kind.PRODUCT,
}
groups = list(project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset"))
out: list[dict] = []
used: set = set()
# 商品天生有真实主图,不必先生成"商品基础资产":匹配不到商品组时直接拿主图当参考(原先这里会丢掉商品参考图)
product_cover = _product_cover_url(project.product)
for rid in (segment.entity_refs or []):
ent = entities.get(rid)
if not ent:
continue
kind = kind_by_type.get(ent.get("type"))
if kind is None: # 商品 / 未知类型不在此处理,商品统一在末尾无条件带上
continue
name = (ent.get("name") or "").strip()
match = next(
(g for g in groups if g.kind == kind and (g.metadata or {}).get("label", "").strip() == name and g.id not in used),
@@ -791,22 +996,16 @@ def _storyboard_reference_images(project, segment) -> list[dict]:
url = _asset_preview_url(match.adopted_asset)
if url:
out.append({"url": url, "label": name or _ENTITY_TYPE_CN.get(ent.get("type"), "参考"), "type": ent.get("type")})
elif ent.get("type") == "product" and product_cover and product_cover not in {r["url"] for r in out}:
out.append({"url": product_cover, "label": name or "商品", "type": "product"})
if len(out) >= 4:
if len(out) >= 3: # 给商品留一个位置(总计最多 4 张)
break
if not out:
pg = next((g for g in groups if g.kind == BaseAssetGroup.Kind.PRODUCT), None)
if pg:
url = _asset_preview_url(pg.adopted_asset)
if url:
out.append({"url": url, "label": "商品", "type": "product"})
elif product_cover:
out.append({"url": product_cover, "label": "商品", "type": "product"})
# 商品参考图永远带上:优先已采用商品三视图组 → 否则真实主图
product_ref = _product_reference_image(project, groups)
if product_ref is not None and product_ref["url"] not in {r["url"] for r in out}:
out.append(product_ref)
# 规范 @图N 顺序:角色 → 商品 → 场景(与下游 image_edit 传图顺序一致,标注不错位)
_ord = {"character": 0, "product": 1, "scene": 2}
out.sort(key=lambda r: _ord.get(r.get("type"), 9))
return out
return out[:4]
def build_storyboard_frame_prompt_refs(project, version, segment, refs: list[dict]) -> str: