diff --git a/backend/README.md b/backend/README.md index 876ce8a..7553e0d 100644 --- a/backend/README.md +++ b/backend/README.md @@ -121,7 +121,7 @@ uv run uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 ### 单元测试与集成测试(Mock) -- **单元测试**:`tests/test_skills_runtime_runners.py`,对 `app.chains.agents` 下 `FilmEntityExtractor` / `FilmShotlistStoryboarder` 的 load_skill、run、format_output、extract 等行为做断言,使用 Mock Agent,无需网络与 API Key。 +- **单元测试**:`tests/test_skills_runtime_runners.py`,对 `app.chains.agents` 下 `FilmEntityExtractorAgent` / `FilmShotlistStoryboarderAgent` 的 run、format_output、extract 等行为做断言,使用 Mock Agent,无需网络与 API Key。 - **集成测试(Mock)**:`tests/test_skills_integration.py` 中的 `TestAppIntegration`、`TestSkillsPipelineIntegration`,覆盖 FastAPI 健康检查、示例路由、以及技能完整链路(真实 Prompt + Mock Agent + 真实解析)。 ```bash diff --git a/backend/app/api/v1/routes/film/extract.py b/backend/app/api/v1/routes/film/extract.py index 50e6ce8..06a03da 100644 --- a/backend/app/api/v1/routes/film/extract.py +++ b/backend/app/api/v1/routes/film/extract.py @@ -5,8 +5,8 @@ import json from fastapi import APIRouter, Depends from langchain_core.runnables import Runnable -from app.chains.agents import FilmEntityExtractor, FilmShotlistStoryboarder -from app.core.skills_runtime import FilmEntityExtractionResult, FilmShotlistResult +from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent +from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult from app.dependencies import get_llm from app.schemas.common import ApiResponse, success_response @@ -25,19 +25,16 @@ def extract_entities( body: EntityExtractRequest, llm: Runnable = Depends(get_llm), ) -> ApiResponse[FilmEntityExtractionResult]: - """FilmEntityExtractor:人物/地点/道具抽取。""" - extractor = FilmEntityExtractor(llm) - extractor.load_skill("film_entity_extractor") + """FilmEntityExtractorAgent:人物/地点/道具抽取。""" + extractor = FilmEntityExtractorAgent(llm) chunks_json = json.dumps( [{"chunk_id": c.chunk_id, "text": c.text} for c in body.chunks], ensure_ascii=False, ) result = extractor.extract( - { - "source_id": body.source_id, - "language": body.language or "zh", - "chunks_json": chunks_json, - } + source_id=body.source_id, + language=body.language or "zh", + chunks_json=chunks_json, ) return success_response(result) @@ -52,20 +49,17 @@ def extract_shotlist( body: ShotlistExtractRequest, llm: Runnable = Depends(get_llm), ) -> ApiResponse[FilmShotlistResult]: - """FilmShotlistStoryboarder:场景/镜头/转场抽取。""" - storyboarder = FilmShotlistStoryboarder(llm) - storyboarder.load_skill("film_shotlist") + """FilmShotlistStoryboarderAgent:场景/镜头/转场抽取。""" + storyboarder = FilmShotlistStoryboarderAgent(llm) chunks_json = json.dumps( [{"chunk_id": c.chunk_id, "text": c.text} for c in body.chunks], ensure_ascii=False, ) result = storyboarder.extract( - { - "source_id": body.source_id, - "source_title": body.source_title or "", - "language": body.language or "zh", - "chunks_json": chunks_json, - } + source_id=body.source_id, + source_title=body.source_title or "", + language=body.language or "zh", + chunks_json=chunks_json, ) return success_response(result) diff --git a/backend/app/api/v1/routes/film/tasks_entities.py b/backend/app/api/v1/routes/film/tasks_entities.py index c5e99ae..0f3a904 100644 --- a/backend/app/api/v1/routes/film/tasks_entities.py +++ b/backend/app/api/v1/routes/film/tasks_entities.py @@ -6,7 +6,7 @@ from fastapi import APIRouter, Depends from langchain_core.runnables import Runnable from sqlalchemy.ext.asyncio import AsyncSession -from app.chains.agents import FilmEntityExtractor, FilmShotlistStoryboarder +from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent from app.core.db import async_session_maker from app.core.task_manager import DeliveryMode, SqlAlchemyTaskStore, TaskManager from app.core.task_manager.types import TaskStatus @@ -65,9 +65,8 @@ async def create_entity_extract_task( store2 = SqlAlchemyTaskStore(session) await store2.set_status(task_id, TaskStatus.running) await store2.set_progress(task_id, 10) - extractor = FilmEntityExtractor(llm) - extractor.load_skill("film_entity_extractor") - result = await extractor.aextract(input_dict) + extractor = FilmEntityExtractorAgent(llm) + result = await extractor.aextract(**input_dict) await store2.set_result(task_id, result.model_dump()) await store2.set_progress(task_id, 100) await store2.set_status(task_id, TaskStatus.succeeded) @@ -132,9 +131,8 @@ async def create_shotlist_task( store2 = SqlAlchemyTaskStore(session) await store2.set_status(task_id, TaskStatus.running) await store2.set_progress(task_id, 10) - storyboarder = FilmShotlistStoryboarder(llm) - storyboarder.load_skill("film_shotlist") - result = await storyboarder.aextract(input_dict) + storyboarder = FilmShotlistStoryboarderAgent(llm) + result = await storyboarder.aextract(**input_dict) await store2.set_result(task_id, result.model_dump()) await store2.set_progress(task_id, 100) await store2.set_status(task_id, TaskStatus.succeeded) diff --git a/backend/app/api/v1/routes/film/tasks_images.py b/backend/app/api/v1/routes/film/tasks_images.py index bafc898..ba5eafd 100644 --- a/backend/app/api/v1/routes/film/tasks_images.py +++ b/backend/app/api/v1/routes/film/tasks_images.py @@ -45,7 +45,6 @@ async def create_shot_frame_prompt_task( frame_type = (body.frame_type or "").strip().lower() if frame_type not in {"first", "last", "key"}: raise HTTPException(status_code=400, detail="frame_type must be one of first/last/key") - skill_id = f"shot_{frame_type}_frame_prompt" if frame_type in {"first", "last"} else "shot_key_frame_prompt" if frame_type == "first": relation_type = "shot_first_frame_prompt" elif frame_type == "last": @@ -88,7 +87,6 @@ async def create_shot_frame_prompt_task( run_args: dict = { "shot_id": body.shot_id, "frame_type": frame_type, - "skill_id": skill_id, "input": input_dict, } @@ -111,7 +109,6 @@ async def create_shot_frame_prompt_task( await session.commit() ft = str(args.get("frame_type") or "") - sid = str(args.get("skill_id") or "") shot_id = str(args.get("shot_id") or "") inp = dict(args.get("input") or {}) @@ -121,8 +118,7 @@ async def create_shot_frame_prompt_task( agent = ShotLastFramePromptAgent(llm) else: agent = ShotKeyFramePromptAgent(llm) - agent.load_skill(sid) - result = await agent.aextract(inp) + result = await agent.aextract(**inp) if not shot_id: raise RuntimeError("Missing shot_id in run args") diff --git a/backend/app/api/v1/routes/script_processing.py b/backend/app/api/v1/routes/script_processing.py index 731901c..7ef8995 100644 --- a/backend/app/api/v1/routes/script_processing.py +++ b/backend/app/api/v1/routes/script_processing.py @@ -7,7 +7,7 @@ import logging from typing import Any from fastapi import APIRouter, Depends, HTTPException, status -from langchain_core.runnables import Runnable +from langchain_core.language_models.chat_models import BaseChatModel from pydantic import BaseModel, Field from app.chains.agents import ( @@ -59,7 +59,7 @@ class ScriptDividerRequest(BaseModel): ) async def divide_script( request: ScriptDividerRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[ScriptDivisionResult]: """ 将完整剧本文本自动分割为多个镜头。 @@ -74,9 +74,7 @@ async def divide_script( """ try: agent = ScriptDividerAgent(llm) - agent.load_skill("script_divider") - - result = agent.extract({"script_text": request.script_text}) + result = agent.divide_script(script_text=request.script_text) return success_response(data=result) except Exception as e: logger.error(f"Script dividing failed: {e}") @@ -112,7 +110,7 @@ class ShotElementExtractionRequest(BaseModel): ) async def extract_shot_elements( request: ShotElementExtractionRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[ShotElementExtractionResult]: """ 从单个镜头的文本中提取关键信息。 @@ -136,14 +134,12 @@ async def extract_shot_elements( """ try: agent = ShotElementExtractorAgent(llm) - agent.load_skill("shot_element_extractor") - - result = agent.extract({ - "index": request.index, - "shot_text": request.shot_text, - "context_summary": request.context_summary or "", - "shot_division_json": json.dumps(request.shot_division or {}, ensure_ascii=False), - }) + result = agent.extract( + index=request.index, + shot_text=request.shot_text, + context_summary=request.context_summary or "", + shot_division_json=json.dumps(request.shot_division or {}, ensure_ascii=False), + ) return success_response(data=result) except Exception as e: logger.error(f"Element extraction failed: {e}") @@ -194,7 +190,7 @@ class EntityMergerRequest(BaseModel): ) async def merge_entities( request: EntityMergerRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[EntityMergeResult]: """ 将多个镜头的提取结果合并,统一实体定义。 @@ -214,15 +210,13 @@ async def merge_entities( """ try: agent = EntityMergerAgent(llm) - agent.load_skill("entity_merger") - - result = agent.extract({ - "all_extractions_json": json.dumps(request.all_shot_extractions, ensure_ascii=False), - "historical_library_json": json.dumps(request.historical_library or {}, ensure_ascii=False), - "script_division_json": json.dumps(request.script_division or {}, ensure_ascii=False), - "previous_merge_json": json.dumps(request.previous_merge or {}, ensure_ascii=False), - "conflict_resolutions_json": json.dumps(request.conflict_resolutions or [], ensure_ascii=False), - }) + result = agent.extract( + all_extractions_json=json.dumps(request.all_shot_extractions, ensure_ascii=False), + historical_library_json=json.dumps(request.historical_library or {}, ensure_ascii=False), + script_division_json=json.dumps(request.script_division or {}, ensure_ascii=False), + previous_merge_json=json.dumps(request.previous_merge or {}, ensure_ascii=False), + conflict_resolutions_json=json.dumps(request.conflict_resolutions or [], ensure_ascii=False), + ) return success_response(data=result) except Exception as e: logger.error(f"Entity merging failed: {e}") @@ -260,7 +254,7 @@ class VariantAnalysisRequest(BaseModel): ) async def analyze_variants( request: VariantAnalysisRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[VariantAnalysisResult]: """ 分析实体的变体(特别是角色服装变化)。 @@ -278,13 +272,11 @@ async def analyze_variants( """ try: agent = VariantAnalyzerAgent(llm) - agent.load_skill("variant_analyzer") - - result = agent.extract({ - "merged_library_json": json.dumps(request.merged_library, ensure_ascii=False), - "all_extractions_json": json.dumps(request.all_shot_extractions, ensure_ascii=False), - "script_division_json": json.dumps(request.script_division or {}, ensure_ascii=False), - }) + result = agent.extract( + merged_library_json=json.dumps(request.merged_library, ensure_ascii=False), + all_extractions_json=json.dumps(request.all_shot_extractions, ensure_ascii=False), + script_division_json=json.dumps(request.script_division or {}, ensure_ascii=False), + ) return success_response(data=result) except Exception as e: logger.error(f"Variant analysis failed: {e}") @@ -311,7 +303,7 @@ class ScriptConsistencyCheckRequest(BaseModel): ) async def check_consistency( request: ScriptConsistencyCheckRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[ScriptConsistencyCheckResult]: """ 检查实体定义与分镜内容的一致性。 @@ -326,11 +318,7 @@ async def check_consistency( """ try: agent = ConsistencyCheckerAgent(llm) - agent.load_skill("consistency_checker") - - result = agent.extract({ - "script_text": request.script_text, - }) + result = agent.extract(script_text=request.script_text) return success_response(data=result) except Exception as e: logger.error(f"Consistency checking failed: {e}") @@ -358,19 +346,17 @@ class ScriptOptimizeRequest(BaseModel): ) async def optimize_script( request: ScriptOptimizeRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[ScriptOptimizationResult]: """ 输入原文 + 一致性检查输出,生成优化后的剧本。 """ try: agent = ScriptOptimizerAgent(llm) - agent.load_skill("script_optimizer") - - result = agent.extract({ - "script_text": request.script_text, - "consistency_json": json.dumps(request.consistency, ensure_ascii=False), - }) + result = agent.extract( + script_text=request.script_text, + consistency_json=json.dumps(request.consistency, ensure_ascii=False), + ) return success_response(data=result) except Exception as e: logger.error(f"Script optimization failed: {e}") @@ -401,19 +387,16 @@ class ScriptExtractRequest(BaseModel): ) async def extract_script( request: ScriptExtractRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[StudioScriptExtractionDraft]: try: agent = ElementExtractorAgent(llm) - agent.load_skill("script_extractor") result = agent.extract( - { - "project_id": request.project_id, - "chapter_id": request.chapter_id, - "script_text": request.script_text, - "script_division_json": json.dumps(request.script_division, ensure_ascii=False), - "consistency_json": json.dumps(request.consistency or {}, ensure_ascii=False), - } + project_id=request.project_id, + chapter_id=request.chapter_id, + script_text=request.script_text, + script_division_json=json.dumps(request.script_division, ensure_ascii=False), + consistency_json=json.dumps(request.consistency or {}, ensure_ascii=False), ) return success_response(data=result) except Exception as e: @@ -444,7 +427,7 @@ class FullProcessRequest(BaseModel): ) async def full_process( request: FullProcessRequest, - llm: Runnable = Depends(get_llm), + llm: BaseChatModel = Depends(get_llm), ) -> ApiResponse[StudioScriptExtractionDraft]: """ 完整的脚本处理工作流(新流程): @@ -462,20 +445,16 @@ async def full_process( # 1. 一致性检查 logger.info("Step 1: Consistency check (character confusion)...") checker = ConsistencyCheckerAgent(llm) - checker.load_skill("consistency_checker") - consistency = checker.extract({"script_text": request.script_text}) + consistency = checker.extract(script_text=request.script_text) # 2. 可选优化 script_text = request.script_text if request.auto_optimize and consistency.has_issues: logger.info("Step 2: Optimizing script...") optimizer = ScriptOptimizerAgent(llm) - optimizer.load_skill("script_optimizer") optimized = optimizer.extract( - { - "script_text": request.script_text, - "consistency_json": json.dumps(consistency.model_dump(), ensure_ascii=False), - } + script_text=request.script_text, + consistency_json=json.dumps(consistency.model_dump(), ensure_ascii=False), ) if optimized.optimized_script_text.strip(): script_text = optimized.optimized_script_text @@ -483,21 +462,17 @@ async def full_process( # 3. 分镜 logger.info("Step 3: Dividing script...") divider = ScriptDividerAgent(llm) - divider.load_skill("script_divider") - division = divider.extract({"script_text": script_text}) + division = divider.divide_script(script_text=script_text) # 4. 项目级提取(最终输出) logger.info("Step 4: Project-level extraction...") extractor = ElementExtractorAgent(llm) - extractor.load_skill("script_extractor") final_result = extractor.extract( - { - "project_id": request.project_id, - "chapter_id": request.chapter_id, - "script_text": script_text, - "script_division_json": json.dumps(division.model_dump(), ensure_ascii=False), - "consistency_json": json.dumps(consistency.model_dump(), ensure_ascii=False), - } + project_id=request.project_id, + chapter_id=request.chapter_id, + script_text=script_text, + script_division_json=json.dumps(division.model_dump(), ensure_ascii=False), + consistency_json=json.dumps(consistency.model_dump(), ensure_ascii=False), ) logger.info("Full process completed successfully") return success_response(data=final_result) diff --git a/backend/app/api/v1/routes/studio/import_from_extraction.py b/backend/app/api/v1/routes/studio/import_from_extraction.py index e8d3f2f..fbeee36 100644 --- a/backend/app/api/v1/routes/studio/import_from_extraction.py +++ b/backend/app/api/v1/routes/studio/import_from_extraction.py @@ -11,7 +11,7 @@ from pydantic import BaseModel from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession -from app.core.skills_runtime.schemas import StudioScriptExtractionDraft +from app.schemas.skills.script_processing import StudioScriptExtractionDraft from app.dependencies import get_db from app.models.studio import ( Actor, diff --git a/backend/app/chains/__init__.py b/backend/app/chains/__init__.py index def1a64..ea67cb9 100644 --- a/backend/app/chains/__init__.py +++ b/backend/app/chains/__init__.py @@ -1,7 +1,14 @@ """LangChain / LangGraph 链与图。""" from app.chains.prompts import example_prompt, STORY_SUMMARY_TEMPLATE -from app.chains.graphs import example_graph, example_graph_builder + +# `langgraph` 属于可选依赖:在纯单元测试/精简环境中可能未安装。 +# 这里不要让 `import app.chains.*` 因为 graphs 而直接失败。 +try: + from app.chains.graphs import example_graph, example_graph_builder +except Exception: # noqa: BLE001 + example_graph = None + example_graph_builder = None __all__ = [ "example_prompt", diff --git a/backend/app/chains/agents/__init__.py b/backend/app/chains/agents/__init__.py index 881841e..96bb590 100644 --- a/backend/app/chains/agents/__init__.py +++ b/backend/app/chains/agents/__init__.py @@ -1,7 +1,7 @@ """Skill 加载与 Agent 运行:通用基类与提取类/提示词类 agent。""" -from app.chains.agents.base import SkillAgentBase, STRUCTURED_OUTPUT_METHOD -from app.chains.agents.extra_agents import FilmEntityExtractor, FilmShotlistStoryboarder +from app.chains.agents.base import AgentBase, STRUCTURED_OUTPUT_METHOD +from app.chains.agents.extra_agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent from app.chains.agents.shot_frame_prompt_agents import ( ShotFirstFramePromptAgent, ShotLastFramePromptAgent, @@ -19,10 +19,10 @@ from app.chains.agents.script_processing_agents import ( ) __all__ = [ - "SkillAgentBase", + "AgentBase", "STRUCTURED_OUTPUT_METHOD", - "FilmEntityExtractor", - "FilmShotlistStoryboarder", + "FilmEntityExtractorAgent", + "FilmShotlistStoryboarderAgent", "ShotFirstFramePromptAgent", "ShotLastFramePromptAgent", "ShotKeyFramePromptAgent", diff --git a/backend/app/chains/agents/base.py b/backend/app/chains/agents/base.py index 3faebb6..88be337 100644 --- a/backend/app/chains/agents/base.py +++ b/backend/app/chains/agents/base.py @@ -1,8 +1,4 @@ -"""通用 Skill Agent 基类:按 skill_id 加载 skill、调用 agent、格式化输出。 - -不局限于信息提取;子类通过 load_skill 与可选 _normalize 钩子绑定具体技能与规范化逻辑。 -当 agent 支持 with_structured_output 时优先使用,否则走「原始字符串 → JSON 解析 → 规范化 → 校验」。 -""" +"""通用 Agent 基类:固化 PromptTemplate + 输出模型,调用 LLM 并解析输出。""" from __future__ import annotations @@ -11,12 +7,11 @@ import re from abc import ABC, abstractmethod from typing import Any, Generic, TypeVar, cast +from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.prompts import PromptTemplate -from langchain_core.runnables import Runnable +from langchain_core.runnables import Runnable, RunnableLambda from pydantic import BaseModel -from app.core.skills_runtime import SKILL_REGISTRY - STRUCTURED_OUTPUT_METHOD = "function_calling" T = TypeVar("T", bound=BaseModel) @@ -31,57 +26,128 @@ def _extract_json_from_text(raw: str) -> str: return text -class SkillAgentBase(ABC, Generic[T]): - """通用 Skill Agent 基类:加载 skill、调用 agent、格式化输出。""" +def _extract_first_json_object(text: str) -> str | None: + """尽量从文本中提取第一个 JSON 对象/数组片段。""" + + s = text.strip() + for opener, closer in (("{", "}"), ("[", "]")): + start = s.find(opener) + end = s.rfind(closer) + if start != -1 and end != -1 and end > start: + return s[start : end + 1].strip() + return None + + +class AgentBase(ABC, Generic[T]): + """通用 Agent 基类:子类固化 prompt_template 与 output_model。""" def __init__( self, - agent: Runnable, + model: BaseChatModel, *, structured_output_method: str = STRUCTURED_OUTPUT_METHOD, + agent_kwargs: dict[str, Any] | None = None, ) -> None: - self._agent = agent + self._model = model self._structured_output_method = structured_output_method - self._prompt: PromptTemplate | None = None - self._output_model: type[BaseModel] | None = None - self._skill_id: str | None = None + self._agent_kwargs = dict(agent_kwargs or {}) self._structured_chain: Runnable | None = None @property - def skill_id(self) -> str | None: - return self._skill_id - @abstractmethod - def load_skill(self, skill_id: str) -> None: - """按 skill_id 加载 skill(设置 prompt 与 output_model)。""" + def prompt_template(self) -> PromptTemplate: ... + @property + @abstractmethod + def output_model(self) -> type[T]: + ... + + @property + def system_prompt(self) -> str: + """子类可覆盖:系统提示词(作为 system message / 高优先级指令)。""" + return "" + def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """子类可覆盖:将 LLM 返回的 dict 规范化为 output_model 所需结构。默认 identity。""" return data - def _ensure_loaded(self) -> None: - if self._prompt is None or self._output_model is None: - raise RuntimeError( - "Skill not loaded; call load_skill(skill_id) first. " - f"Available: {list(SKILL_REGISTRY.keys())}" + def render_prompt(self, **kwargs: Any) -> str: + """渲染完整提示词(变量填充后的最终字符串)。""" + + prompt = self.prompt_template + try: + user_prompt = prompt.format(**kwargs) + except KeyError as e: + missing = str(e).strip("'") + raise ValueError( + f"render_prompt 缺少变量: {missing}. " + f"需要: {list(prompt.input_variables)}" + ) from e + + sys = (self.system_prompt or "").strip() + if not sys: + return user_prompt + return f"{sys}\n\n{user_prompt}".strip() + + def _render_user_prompt(self, **kwargs: Any) -> str: + """仅渲染用户提示词(不包含 system_prompt)。""" + prompt = self.prompt_template + try: + return prompt.format(**kwargs) + except KeyError as e: + missing = str(e).strip("'") + raise ValueError( + f"render_prompt 缺少变量: {missing}. " + f"需要: {list(prompt.input_variables)}" + ) from e + + def create_agent(self, *, structured_output: type[BaseModel] | None = None) -> Runnable: + """ + 生成可执行 runnable: + - 优先使用 `langchain.agents.create_agent`,把 `system_prompt` 作为系统提示词传入 + - structured_output 不为 None 时,通过 `response_format=ToolStrategy(structured_output)` 让模型输出结构化结果 + - 若当前环境未安装 langchain,则降级为:RunnableLambda(render_user_prompt) | model(structured 时用 with_structured_output) + """ + + def _render_input(inputs: dict[str, Any]) -> dict[str, Any]: + return {"input": self._render_user_prompt(**inputs)} + + # --- preferred path: langchain create_agent + ToolStrategy --- + try: + from langchain.agents import create_agent as _lc_create_agent # type: ignore + + kwargs = dict(self._agent_kwargs) + if structured_output is not None: + kwargs["response_format"] = structured_output + agent = _lc_create_agent( + model=self._model, + system_prompt=(self.system_prompt or ""), + **kwargs, ) + return RunnableLambda(_render_input) | cast(Runnable, agent) + except Exception: + # --- fallback path: no langchain available --- + llm: Runnable = cast(Runnable, self._model) + if structured_output is not None: + with_structured = getattr(self._model, "with_structured_output", None) + if callable(with_structured): + try: + llm = cast( + Runnable, + with_structured( + structured_output, + method=self._structured_output_method, + ), + ) + except NotImplementedError: + # 某些 BaseChatModel 子类(如测试 mock)不实现 with_structured_output;退回原始输出解析。 + pass + return RunnableLambda(lambda inputs: self._render_user_prompt(**inputs)) | llm def _build_structured_chain(self) -> Runnable | None: - """若 agent 支持 with_structured_output,则构建 prompt | agent.with_structured_output(schema)。""" - self._ensure_loaded() - with_structured = getattr(self._agent, "with_structured_output", None) - if not callable(with_structured): - return None - assert self._prompt is not None and self._output_model is not None - structured_llm = cast( - Runnable, - with_structured( - self._output_model, - method=self._structured_output_method, - ), - ) - return self._prompt | structured_llm + """构建 structured output chain(优先 ToolStrategy;缺 langchain 时退回 with_structured_output)。""" + return self.create_agent(structured_output=self.output_model) def _get_structured_chain(self) -> Runnable | None: if self._structured_chain is not None: @@ -89,58 +155,63 @@ class SkillAgentBase(ABC, Generic[T]): self._structured_chain = self._build_structured_chain() return self._structured_chain - def run(self, input_dict: dict[str, Any]) -> str: + def run(self, **kwargs: Any) -> str: """调用 agent,返回原始字符串(通常为 JSON)。""" - self._ensure_loaded() - assert self._prompt is not None - chain: Runnable = self._prompt | self._agent - result = chain.invoke(input_dict) + chain: Runnable = self.create_agent() + result = chain.invoke(kwargs) if hasattr(result, "content"): return getattr(result, "content", str(result)) return str(result) - async def arun(self, input_dict: dict[str, Any]) -> str: + async def arun(self, **kwargs: Any) -> str: """异步调用 agent。""" - self._ensure_loaded() - assert self._prompt is not None - chain: Runnable = self._prompt | self._agent - result = await chain.ainvoke(input_dict) + chain: Runnable = self.create_agent() + result = await chain.ainvoke(kwargs) if hasattr(result, "content"): return getattr(result, "content", str(result)) return str(result) def format_output(self, raw: str) -> T: """将 agent 原始输出解析为结构化结果(JSON → 规范化 → Pydantic)。""" - self._ensure_loaded() - assert self._output_model is not None + output_model = self.output_model json_str = _extract_json_from_text(raw) - data = json.loads(json_str) + try: + data = json.loads(json_str) + except Exception: + candidate = _extract_first_json_object(json_str) + if candidate is None: + raise + data = json.loads(candidate) if isinstance(data, dict): data = self._normalize(data) - return self._output_model.model_validate(data) # type: ignore[return-value] + return output_model.model_validate(data) - def extract(self, input_dict: dict[str, Any]) -> T: + def extract(self, **kwargs: Any) -> T: """执行:优先 with_structured_output,否则 run + format_output。""" chain = self._get_structured_chain() if chain is not None: - result = chain.invoke(input_dict) - if isinstance(result, BaseModel): - return result # type: ignore[return-value] - if isinstance(result, dict): - data = self._normalize(result) - return self._output_model.model_validate(data) # type: ignore[return-value] - raw = self.run(input_dict) - return self.format_output(raw) + try: + result = chain.invoke(kwargs) + if isinstance(result, self.output_model): + return cast(T, result) + if isinstance(result, dict): + data = self._normalize(result) + return self.output_model.model_validate(data) + except Exception: + pass + return self.format_output(self.run(**kwargs)) - async def aextract(self, input_dict: dict[str, Any]) -> T: + async def aextract(self, **kwargs: Any) -> T: """异步执行。""" chain = self._get_structured_chain() if chain is not None: - result = await chain.ainvoke(input_dict) - if isinstance(result, BaseModel): - return result # type: ignore[return-value] - if isinstance(result, dict): - data = self._normalize(result) - return self._output_model.model_validate(data) # type: ignore[return-value] - raw = await self.arun(input_dict) - return self.format_output(raw) + try: + result = await chain.ainvoke(kwargs) + if isinstance(result, self.output_model): + return cast(T, result) + if isinstance(result, dict): + data = self._normalize(result) + return self.output_model.model_validate(data) + except Exception: + pass + return self.format_output(await self.arun(**kwargs)) diff --git a/backend/app/chains/agents/extra_agents.py b/backend/app/chains/agents/extra_agents.py index 5d26995..adadfc9 100644 --- a/backend/app/chains/agents/extra_agents.py +++ b/backend/app/chains/agents/extra_agents.py @@ -4,15 +4,14 @@ from __future__ import annotations from typing import Any -from app.core.skills_runtime import SKILL_REGISTRY -from app.core.skills_runtime.film_entity_extractor import FilmEntityExtractionResult -from app.core.skills_runtime.film_shotlist_storyboarder import FilmShotlistResult +from langchain_core.prompts import PromptTemplate -from app.chains.agents.base import SkillAgentBase +from app.chains.agents.base import AgentBase +from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult # --------------------------------------------------------------------------- -# 影视实体/分镜结果规范化(供 FilmEntityExtractor / FilmShotlistStoryboarder 使用) +# 影视实体/分镜结果规范化(供 FilmEntityExtractorAgent / FilmShotlistStoryboarderAgent 使用) # --------------------------------------------------------------------------- @@ -115,40 +114,149 @@ def _normalize_shotlist_result(data: dict[str, Any]) -> dict[str, Any]: # 提取类 Agent 实现 # --------------------------------------------------------------------------- +_ENTITY_EXTRACTION_SYSTEM_PROMPT = """\ +你是“影视文本信息抽取系统”。你的唯一任务是:从输入小说原文中抽取【人物】【地点】【道具】,并给出可追溯证据。 -class FilmEntityExtractor(SkillAgentBase[FilmEntityExtractionResult]): - """关键信息提取:使用 film_entity_extractor skill,输出人物/地点/道具。""" +## 硬性规则(必须遵守) +- 绝对禁止编造:不得凭常识补全姓名、身份、外貌、地点类型、道具用途等;原文没写就不写。 +- 只抽取原文明确出现的实体:如果只是“他/她/那里/东西”等代词且无法指代明确实体,不要强行新建实体;可在 uncertainties 说明原因。 +- 所有实体的关键字段都应尽量附带 EvidenceSpan(至少给 first_appearance;quote ≤ 200 字,必须是原文摘录)。 +- 不要输出 schemas 未定义的字段;输出 JSON,且必须能被 Pydantic 校验(extra=forbid)。 +- ID 规则:characters 用 char_001 起;locations 用 loc_001 起;props 用 prop_001 起。保持稳定、不要跳号。 +- 同一实体可能有多个称呼:把原文出现的别名放进 aliases;normalized_name 只能来自原文(例如原文既叫“王二”也叫“二哥”)。 +- confidence:可选;如果给出必须在 0-1 之间;不确定就留空。 - KEY_INFO_SKILL_IDS = ("film_entity_extractor",) +## 输出结构 +输出必须是一个 JSON 对象,符合下列模型(字段名必须完全一致): +- source_id: string +- language?: string +- extraction_version?: string +- schema_version?: string +- chunks: string[] +- characters: Character[] +- locations: Location[] +- props: Prop[] +- notes: string[] +- uncertainties: Uncertainty[] - def load_skill(self, skill_id: str) -> None: - if skill_id not in self.KEY_INFO_SKILL_IDS or skill_id not in SKILL_REGISTRY: - raise ValueError( - f"Unknown or invalid key-info skill_id: {skill_id}. " - f"Allowed: {self.KEY_INFO_SKILL_IDS}" - ) - self._prompt, self._output_model = SKILL_REGISTRY[skill_id] - self._skill_id = skill_id - self._structured_chain = None +其中 Character/Location/Prop/EvidenceSpan/Uncertainty 的字段以 schemas 为准: +- EvidenceSpan: { chunk_id, start_char?, end_char?, quote? } +- Uncertainty: { field_path, reason, evidence[] } + +## 抽取策略(尽量“忠实 + 可用”) +- 人物:优先抽取具名角色;对“掌柜/车夫/士兵”等泛称,只有在原文多次出现且可作为角色稳定识别时才建人物。 +- 地点:出现明确地名/场所名时抽取;若只是“屋里/门外”且无稳定命名,可不抽取或写入 uncertainties。 +- 道具:只抽取剧情关键或明确被提及的物件(信件、刀、钥匙、手机等)。不要把“桌椅板凳”当道具堆砌,除非原文强调其关键性。 + +只输出 JSON,不要输出任何解释性文字。 +""" + +FILM_ENTITY_EXTRACTION_PROMPT = PromptTemplate( + input_variables=["source_id", "language", "chunks_json"], + template="""## 输入 +source_id: {source_id} +language: {language} +chunks (JSON 数组,元素含 chunk_id 与 text): +{chunks_json} + +## 输出 +请输出 FilmEntityExtractionResult 的 JSON。 +""", +) + +_SHOTLIST_SYSTEM_PROMPT = """\ +你是“影视分镜师(Shot List / Storyboard Breakdown)”。你的任务是把给定小说原文拆解为可拍摄的场景与镜头表。 + +## 硬性规则(必须遵守) +- 绝对禁止编造:不得新增原文不存在的人物/地点/道具/动作/对白/情节;不得凭常识补全。 +- 如果原文不确定(比如光线、时间、人物是谁、是否在室内外),请: + - 在对应字段保持 UNKNOWN/空值,或 + - 写入 uncertainties(field_path + reason + evidence),而不是“猜一个”。 +- 所有镜头描述必须“可拍”:明确画面内容与动作,不要抽象文学化抒情。 +- 只使用 schemas 中允许的枚举值: + - ShotType: ECU/CU/MCU/MS/MLS/LS/ELS + - CameraAngle: EYE_LEVEL/HIGH_ANGLE/LOW_ANGLE/BIRD_EYE/DUTCH/OVER_SHOULDER + - CameraMovement: STATIC/PAN/TILT/DOLLY_IN/DOLLY_OUT/TRACK/CRANE/HANDHELD/STEADICAM/ZOOM_IN/ZOOM_OUT + - TransitionType: CUT/DISSOLVE/WIPE/FADE_IN/FADE_OUT/MATCH_CUT/J_CUT/L_CUT + - VFXType: NONE/PARTICLES/VOLUMETRIC_FOG/CG_DOUBLE/DIGITAL_ENVIRONMENT/MATTE_PAINTING/FIRE_SMOKE/WATER_SIM/DESTRUCTION/ENERGY_MAGIC/COMPOSITING_CLEANUP/SLOW_MOTION_TIME/OTHER +- EvidenceSpan.quote ≤ 200 字,必须来自原文摘录;chunk_id 必须来自输入。 +- 只输出 JSON(不要解释、不要 markdown、不要多余文本),并且必须能通过 Pydantic 校验(extra=forbid)。 + +## 输出结构 +输出必须是一个 JSON 对象,符合 FilmShotlistResult: +{ + "breakdown": ProjectCinematicBreakdown +} + +## 拆解原则(专业且忠实) +- 先抽取实体表(characters/locations/props),再按叙事推进切分 scenes(场景变化依据:地点/时间/人物组合或叙事段落明显转折)。 +- 每个 scene 建议 2-8 个 shot;shot.description 用“行业口吻 + 可拍动作”描述。 +- 对白处理: + - 原文出现引号对白时,尽量拆进 DialogueLine.text(逐句,尽量原句)。 + - speaker/target 不确定可留空,但要把 evidence 填好;必要时在 uncertainties 说明。 + - 旁白/心理独白如果原文明确,可用 VOICE_OVER;画外音用 OFF_SCREEN;电话音用 PHONE。 +- 转场 transitions: + - 同一 scene 内镜头间通常 CUT;明显时间跳跃/回忆可用 DISSOLVE/FADE;声音先行可用 J_CUT/L_CUT(仅当原文能支撑)。 +- VFX: + - 原文明确出现超自然/烟火/水效/破碎/慢动作等才写;否则 vfx_type=NONE。 +- 时长 duration_sec 可选:只有在原文节奏/动作明确时才给出,否则留空。 + +## ID 规则(必须) +- scene_id: scene_001 起 +- shot_id: shot_{sceneIndex3}_{order3},例如 scene_001 的第3镜是 shot_001_003 +- from/to_shot_id 必须引用存在的 shot_id + +只输出 JSON。 +""" + +FILM_SHOTLIST_PROMPT = PromptTemplate( + input_variables=["source_id", "source_title", "language", "chunks_json"], + template="""## 输入 +source_id: {source_id} +source_title: {source_title} +language: {language} +chunks (JSON 数组,元素含 chunk_id 与 text): +{chunks_json} + +## 输出 +请输出 FilmShotlistResult 的 JSON。 +""", +) + + +class FilmEntityExtractorAgent(AgentBase[FilmEntityExtractionResult]): + """关键信息提取 Agent:使用 film_entity_extractor skill,输出人物/地点/道具。""" + + @property + def system_prompt(self) -> str: + return _ENTITY_EXTRACTION_SYSTEM_PROMPT + + @property + def prompt_template(self) -> PromptTemplate: + return FILM_ENTITY_EXTRACTION_PROMPT + + @property + def output_model(self) -> type[FilmEntityExtractionResult]: + return FilmEntityExtractionResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: return _normalize_entity_result(data) -class FilmShotlistStoryboarder(SkillAgentBase[FilmShotlistResult]): - """分镜提取:使用 film_shotlist skill,输出场景/镜头/转场(ProjectCinematicBreakdown)。""" +class FilmShotlistStoryboarderAgent(AgentBase[FilmShotlistResult]): + """分镜提取 Agent:使用 film_shotlist skill,输出场景/镜头/转场(ProjectCinematicBreakdown)。""" - SHOTLIST_SKILL_IDS = ("film_shotlist",) + @property + def system_prompt(self) -> str: + return _SHOTLIST_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - if skill_id not in self.SHOTLIST_SKILL_IDS or skill_id not in SKILL_REGISTRY: - raise ValueError( - f"Unknown or invalid shotlist skill_id: {skill_id}. " - f"Allowed: {self.SHOTLIST_SKILL_IDS}" - ) - self._prompt, self._output_model = SKILL_REGISTRY[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def prompt_template(self) -> PromptTemplate: + return FILM_SHOTLIST_PROMPT + + @property + def output_model(self) -> type[FilmShotlistResult]: + return FilmShotlistResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: return _normalize_shotlist_result(data) diff --git a/backend/app/chains/agents/script_processing_agents.py b/backend/app/chains/agents/script_processing_agents.py index b6b7e6d..c448cca 100644 --- a/backend/app/chains/agents/script_processing_agents.py +++ b/backend/app/chains/agents/script_processing_agents.py @@ -5,55 +5,215 @@ from __future__ import annotations import json from typing import Any -from app.chains.agents.base import SkillAgentBase, _extract_json_from_text -from app.core.skills_runtime import schemas as sp_schemas +from langchain_core.prompts import PromptTemplate -# 统一输出模型来源:严格 schema 在 schemas.py -ShotDivision = sp_schemas.ShotDivision -ScriptDivisionResult = sp_schemas.ScriptDivisionResult -ScriptConsistencyCheckResult = sp_schemas.ScriptConsistencyCheckResult -ScriptOptimizationResult = sp_schemas.ScriptOptimizationResult -StudioScriptExtractionDraft = sp_schemas.StudioScriptExtractionDraft -ShotCharacterInfo = sp_schemas.ShotCharacterInfo -ShotPropInfo = sp_schemas.ShotPropInfo -ShotSceneInfo = sp_schemas.ShotSceneInfo -ShotElements = sp_schemas.ShotElements -ShotElementExtractionResult = sp_schemas.ShotElementExtractionResult -EntityVariant = sp_schemas.EntityVariant -EntityEntry = sp_schemas.EntityEntry -EntityLibrary = sp_schemas.EntityLibrary -EntityMergeResult = sp_schemas.EntityMergeResult -CostumeTimelineEntry = sp_schemas.CostumeTimelineEntry -CostumeTimeline = sp_schemas.CostumeTimeline -VariantSuggestion = sp_schemas.VariantSuggestion -VariantAnalysisResult = sp_schemas.VariantAnalysisResult -ConsistencyWarning = sp_schemas.ConsistencyWarning -ConsistencyCheckResult = sp_schemas.ConsistencyCheckResult -TableData = sp_schemas.TableData -OutputCompileResult = sp_schemas.OutputCompileResult +from app.chains.agents.base import AgentBase, _extract_json_from_text +from app.schemas.skills.script_processing import ( + OutputCompileResult, + ScriptConsistencyCheckResult, + ScriptDivisionResult, + ScriptOptimizationResult, + ShotElementExtractionResult, + StudioScriptExtractionDraft, + EntityMergeResult, + VariantAnalysisResult, + ShotDivision, +) + +# PromptTemplates(固化在本模块,供各 Agent 子类通过 @property 返回) +_SCRIPT_DIVIDER_SYSTEM_PROMPT = """\ +你是\"剧本分镜师\"。将完整剧本分割为多个镜头。每个镜头应是完整的连贯场景。 +为每个镜头提供: +- index(镜头序号,章节内唯一;从 1 开始) +- start_line、end_line +- shot_name(镜头名称/镜头标题,分镜名;一句话描述该镜头画面/动作;不要把它当作场景名) +- script_excerpt(镜头对应的剧本摘录/文本) +- scene_name(场景名称,必须与 shots 中 scene_name 的含义一致;不要把 shot_name 当成 scene_name) +- time_of_day +- character_names_in_text(角色名/称呼,弱信息;稳定ID会在后续合并阶段统一分配) +严格区分字段含义: +- shot_name = 分镜名/镜头标题 +- scene_name = 场景名 +只输出 JSON,符合 ScriptDivisionResult 结构。 +""" + +SCRIPT_DIVIDER_PROMPT = PromptTemplate( + input_variables=["script_text"], + template="## 输入脚本\n{script_text}\n\n## 输出\n", +) + +_ELEMENT_EXTRACTOR_SYSTEM_PROMPT = """\ +你是\"镜头元素提取员\"。从单个镜头提取关键信息: +- character_keys、scene_keys、costume_keys、prop_keys(弱ID/归一化名;稳定ID由后续合并阶段统一分配) +- characters_detailed(每个角色含 character_key/name_in_text/appearance/clothing/accessories/state + raw_* 溯源,可选 evidence) +- props_detailed(每个道具含 prop_key/name_in_text/description/state/interaction + raw_text,可选 evidence) +- scene_detailed(scene_key/name/location_detail/atmosphere/time_weather/raw_description_text,可选 evidence) +- dialogue_lines(结构化对白,字段与 schemas.DialogueLine 对齐) +- actions、shot_type_hints、confidence_breakdown +其中 dialogue_lines 每项必须包含 text/line_mode;建议包含 index/speaker_character_id/target_character_id(若可判定)。 +严格按照原文,不要编造。只输出 JSON,符合 ShotElementExtractionResult 结构。 +""" + +ELEMENT_EXTRACTOR_PROMPT = PromptTemplate( + input_variables=["index", "shot_text", "context_summary", "shot_division_json"], + template="镜头号: {index}\n分镜元信息(来自上一步): {shot_division_json}\n上文: {context_summary}\n\n## 镜头文本\n{shot_text}\n\n## 输出\n", +) + +_ENTITY_MERGER_SYSTEM_PROMPT = """\ +你是\"实体合并师\"。合并多镜头提取结果,统一实体定义,为每个实体分配ID,识别变体和冲突。 +请输出 EntityMergeResult,merged_library 中至少包含 characters/locations/scenes/props 四类。 +每个实体条目(EntityEntry)需包含: +- 通用:id/name/type/description/aliases/normalized_name/confidence/first_appearance/evidence/first_shot/appearances/variants +- 角色(type=character):尽量补充 costume_note、traits +- 地点(type=location):尽量补充 location_type +- 道具(type=prop):尽量补充 category、owner_character_id +variants 使用 {variant_key, description, affected_shots, evidence} 的最小结构。 +当提供 previous_merge_json 与 conflict_resolutions_json 时,表示这是一次“重试合并”:你必须参考上一次的合并结果与冲突解决建议,优先消解 conflicts;必要时可调整实体合并/拆分策略,但要保持 ID 尽量稳定(除非建议明确要求变更)。 +只输出 JSON,符合 EntityMergeResult 结构。 +""" + +ENTITY_MERGER_PROMPT = PromptTemplate( + input_variables=[ + "all_extractions_json", + "historical_library_json", + "script_division_json", + "previous_merge_json", + "conflict_resolutions_json", + ], + template=( + "## 脚本分镜(来自上一步)\n{script_division_json}\n\n" + "## 所有镜头提取结果\n{all_extractions_json}\n\n" + "## 历史实体库\n{historical_library_json}\n\n" + "## 上一次合并结果(可选,用于重试)\n{previous_merge_json}\n\n" + "## 冲突解决建议(可选,用于重试)\n{conflict_resolutions_json}\n\n" + "## 输出\n" + ), +) + +_VARIANT_ANALYZER_SYSTEM_PROMPT = """\ +你是\"变体分析师\"。分析实体变体(特别是角色服装变化),构建时间线,生成变体建议。 +输出 VariantAnalysisResult:costume_timelines.timeline_entries 使用 {shot_index, scene_id, costume_note, changes, evidence};variant_suggestions 可带 evidence。 +只输出 JSON,符合 VariantAnalysisResult 结构。 +""" + +VARIANT_ANALYZER_PROMPT = PromptTemplate( + input_variables=["merged_library_json", "all_extractions_json", "script_division_json"], + template="## 脚本分镜(来自上一步)\n{script_division_json}\n\n## 合并后的实体库\n{merged_library_json}\n\n## 所有镜头提取结果\n{all_extractions_json}\n\n## 输出\n", +) + +_CONSISTENCY_CHECKER_SYSTEM_PROMPT = """\ +你是\"一致性检查员\"。只做一件事:检测原文中是否把“同一个角色”在不同段落/镜头中赋予了不同的身份或行为主体,导致角色混淆(例如:同名不同人、代词指代混乱、行为归属错位)。 + +输出 ScriptConsistencyCheckResult: +- issues: 每条问题必须包含 character_candidates、description、suggestion;尽量给出 affected_lines(start_line/end_line)。 +- has_issues: issues 非空则为 true + +只输出 JSON。 +""" + +CONSISTENCY_CHECKER_PROMPT = PromptTemplate( + input_variables=["script_text"], + template="## 原文剧本\n{script_text}\n\n## 输出\n", +) + +_SCRIPT_OPTIMIZER_SYSTEM_PROMPT = """\ +你是\"剧本优化师\"。仅当一致性检查发现角色混淆问题时,对原文进行最小改写以消除混淆。 + +输入: +- script_text:原文 +- consistency_json:一致性检查输出(ScriptConsistencyCheckResult) + +输出 ScriptOptimizationResult: +- optimized_script_text:优化后的完整剧本文本(尽量少改,只改与 issues 相关的段落) +- change_summary:逐条对应 issues 的改动摘要 + +只输出 JSON。 +""" + +SCRIPT_OPTIMIZER_PROMPT = PromptTemplate( + input_variables=["script_text", "consistency_json"], + template="## 一致性检查结果\n{consistency_json}\n\n## 原文剧本\n{script_text}\n\n## 输出\n", +) + +_SCRIPT_EXTRACTOR_SYSTEM_PROMPT = """\ +你是\"Studio 信息提取员\"。你的任务是:基于剧本文本与分镜结果,输出可直接导入 Studio 的草稿结构 StudioScriptExtractionDraft(注意:ID 由导入 API 生成,因此这里全部使用 name 做引用键)。 + +输出 StudioScriptExtractionDraft: +- project_id(必填) +- chapter_id(必填) +- script_text(必填) +- characters: [{name, description, costume_name?, prop_names[], tags[]}] +- scenes/props/costumes: [{name, description, tags[], prompt_template_id?, view_count}] +- shots: [{index, title, script_excerpt, scene_name?, character_names[], prop_names[], costume_names[], dialogue_lines[], actions[]}] + - dialogue_lines: [{index, text, line_mode, speaker_name?, target_name?}] + +强约束: +- 同名实体在输出中只出现一次(全局去重);shots 中引用必须使用同一名称 +- shots.index 必须覆盖并对应输入分镜中的 index(不要跳号) +- 不要输出任何 id 字段(包括 char_001 等),由导入 API 生成 + +一致性强约束(必须严格遵守,否则导入会失败): +- 先输出全局 characters/scenes/props/costumes 列表,再输出 shots;并把它们视为“字典”。 +- shots[*].character_names / prop_names / costume_names / scene_name 只能从对应全局列表的 name 中选择(完全一致的字符串),禁止生成任何未在全局列表中出现的新名字。 +- 禁止“同义名/括号变体/临时称呼”漂移:例如禁止在 shots 中写「女子(群)」但在 characters 中没有该条目;禁止「仙女A」与「仙女 A」混用。 +- 遇到群体角色/泛指角色(如“女子(群)”“群众”“村民们”):必须在 characters 列表中创建一条同名角色(name 完全一致),并在 shots 中引用该 name。 +- 对于难以确定是否同一角色的称呼:宁可在 characters 里拆成两条不同 name,也不要在 shots 中凭空换名。 +- 输出 shots 之前,必须做“全集校验”并补齐缺失:所有 shots[*] 中出现的 character_names/prop_names/costume_names/scene_name 的名字集合,必须都能在对应全局列表(characters/props/costumes/scenes)的 name 中找到;如果有缺失,必须在全局列表中补齐对应条目(描述可最小化,但 name 必须完全一致),禁止用别名替换来绕过。 +- 角色名/场景名必须原样保留字符细节:包括全角/半角括号、空格、标点,不要自动做任何规范化或替换(例如不能把「女子(群)」改成「女子(群)」或「女子 (群)」)。 +- 严格区分:shots[*].title 是“镜头标题”(一句话描述该镜头画面/动作),不要拿它当作 scenes 的 scene 名;shots[*].scene_name 才是场景名称,必须来自 scenes 全局列表的 name。 + +输入: +- project_id +- chapter_id +- script_text +- script_division_json(ScriptDivisionResult) +- consistency_json(可选) + +只输出 JSON。 +""" + +SCRIPT_EXTRACTOR_PROMPT = PromptTemplate( + input_variables=["project_id", "chapter_id", "script_text", "script_division_json", "consistency_json"], + template=( + "## project_id\n{project_id}\n\n" + "## chapter_id\n{chapter_id}\n\n" + "## 一致性检查(可选)\n{consistency_json}\n\n" + "## 分镜结果\n{script_division_json}\n\n" + "## 剧本文本\n{script_text}\n\n" + "## 输出\n" + ), +) + +_OUTPUT_COMPILER_SYSTEM_PROMPT = """\ +你是\"输出编译员\"。汇总所有Agent输出,生成完整项目JSON、可导出表格、项目总结。 +输出 OutputCompileResult,其中 project_json 必须严格符合 ProjectCinematicBreakdown schema(至少包含 source_id/chunks/characters/locations/props/scenes/shots/transitions/notes/uncertainties)。 +只输出 JSON。 +""" + +OUTPUT_COMPILER_PROMPT = PromptTemplate( + input_variables=["division_json", "all_extractions_json", "merge_json", "variant_json", "consistency_json"], + template="## 分镜结果\n{division_json}\n\n## 所有逐镜提取\n{all_extractions_json}\n\n## 实体合并\n{merge_json}\n\n## 变体分析\n{variant_json}\n\n## 一致性检查\n{consistency_json}\n\n## 输出\n", +) # ============================================================================ # 1. ScriptDividerAgent - 剧本自动分镜 # ============================================================================ -class ScriptDividerAgent(SkillAgentBase[ScriptDivisionResult]): +class ScriptDividerAgent(AgentBase[ScriptDivisionResult]): """剧本自动分镜:输入完整剧本文本,输出分镜列表。""" - SCRIPT_DIVIDER_SKILL_IDS = ("script_divider",) + @property + def system_prompt(self) -> str: + return _SCRIPT_DIVIDER_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return SCRIPT_DIVIDER_PROMPT - registry = get_skill_registry() - if skill_id not in self.SCRIPT_DIVIDER_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid script_divider skill_id: {skill_id}. " - f"Allowed: {self.SCRIPT_DIVIDER_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ScriptDivisionResult]: + return ScriptDivisionResult def format_output(self, raw: str) -> ScriptDivisionResult: """ @@ -63,9 +223,6 @@ class ScriptDividerAgent(SkillAgentBase[ScriptDivisionResult]): - 包裹结构:{"ScriptDivisionResult": {...}} - 直接列表:[{...}, {...}](视为 shots) """ - self._ensure_loaded() - assert self._output_model is not None - json_str = _extract_json_from_text(raw) data: Any = json.loads(json_str) @@ -83,7 +240,13 @@ class ScriptDividerAgent(SkillAgentBase[ScriptDivisionResult]): if isinstance(data, dict): data = self._normalize(data) - return self._output_model.model_validate(data) # type: ignore[arg-type] + return self.output_model.model_validate(data) # type: ignore[arg-type] + + def divide_script(self, *, script_text: str) -> ScriptDivisionResult: + return self.extract(script_text=script_text) + + async def adivide_script(self, *, script_text: str) -> ScriptDivisionResult: + return await self.aextract(script_text=script_text) def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """规范化脚本分割结果。""" @@ -131,23 +294,20 @@ class ScriptDividerAgent(SkillAgentBase[ScriptDivisionResult]): # 2. ShotElementExtractorAgent - 逐镜信息提取(兼容旧流程) # ============================================================================ -class ShotElementExtractorAgent(SkillAgentBase[ShotElementExtractionResult]): +class ShotElementExtractorAgent(AgentBase[ShotElementExtractionResult]): """[兼容] 逐镜信息提取:输入单镜文本+上文摘要,输出该镜的结构化提取结果。""" - ELEMENT_EXTRACTOR_SKILL_IDS = ("shot_element_extractor",) + @property + def system_prompt(self) -> str: + return _ELEMENT_EXTRACTOR_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return ELEMENT_EXTRACTOR_PROMPT - registry = get_skill_registry() - if skill_id not in self.ELEMENT_EXTRACTOR_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid element_extractor skill_id: {skill_id}. " - f"Allowed: {self.ELEMENT_EXTRACTOR_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ShotElementExtractionResult]: + return ShotElementExtractionResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """规范化元素提取结果(升级版结构)。""" @@ -243,23 +403,20 @@ class ShotElementExtractorAgent(SkillAgentBase[ShotElementExtractionResult]): # 2b. ElementExtractorAgent - 项目级信息提取(新流程最终输出) # ============================================================================ -class ElementExtractorAgent(SkillAgentBase[StudioScriptExtractionDraft]): +class ElementExtractorAgent(AgentBase[StudioScriptExtractionDraft]): """项目级信息提取(最终输出):输入剧本文本 + 分镜结果,产出全局实体表 + 逐镜关联。""" - PROJECT_EXTRACTOR_SKILL_IDS = ("script_extractor",) + @property + def system_prompt(self) -> str: + return _SCRIPT_EXTRACTOR_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return SCRIPT_EXTRACTOR_PROMPT - registry = get_skill_registry() - if skill_id not in self.PROJECT_EXTRACTOR_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid script_extractor skill_id: {skill_id}. " - f"Allowed: {self.PROJECT_EXTRACTOR_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[StudioScriptExtractionDraft]: + return StudioScriptExtractionDraft def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: data = dict(data) @@ -276,23 +433,20 @@ class ElementExtractorAgent(SkillAgentBase[StudioScriptExtractionDraft]): # 3. EntityMergerAgent - 跨镜静态合并 + 基础画像生成 # ============================================================================ -class EntityMergerAgent(SkillAgentBase[EntityMergeResult]): +class EntityMergerAgent(AgentBase[EntityMergeResult]): """跨镜合并 + 基础画像生成:输入全部分镜提取结果+历史实体库,输出合并后的库。""" - ENTITY_MERGER_SKILL_IDS = ("entity_merger",) + @property + def system_prompt(self) -> str: + return _ENTITY_MERGER_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return ENTITY_MERGER_PROMPT - registry = get_skill_registry() - if skill_id not in self.ENTITY_MERGER_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid entity_merger skill_id: {skill_id}. " - f"Allowed: {self.ENTITY_MERGER_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[EntityMergeResult]: + return EntityMergeResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """规范化实体合并结果。""" @@ -356,23 +510,20 @@ class EntityMergerAgent(SkillAgentBase[EntityMergeResult]): # 4. VariantAnalyzerAgent - 服装/外形变体检测与建议 # ============================================================================ -class VariantAnalyzerAgent(SkillAgentBase[VariantAnalysisResult]): +class VariantAnalyzerAgent(AgentBase[VariantAnalysisResult]): """服装/外形变体检测与建议:输入实体库+全镜提取,输出变体分析结果。""" - VARIANT_ANALYZER_SKILL_IDS = ("variant_analyzer",) + @property + def system_prompt(self) -> str: + return _VARIANT_ANALYZER_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return VARIANT_ANALYZER_PROMPT - registry = get_skill_registry() - if skill_id not in self.VARIANT_ANALYZER_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid variant_analyzer skill_id: {skill_id}. " - f"Allowed: {self.VARIANT_ANALYZER_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[VariantAnalysisResult]: + return VariantAnalysisResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """规范化变体分析结果。""" @@ -402,23 +553,20 @@ class VariantAnalyzerAgent(SkillAgentBase[VariantAnalysisResult]): # 5. ConsistencyCheckerAgent - 文本一致性检查 # ============================================================================ -class ConsistencyCheckerAgent(SkillAgentBase[ScriptConsistencyCheckResult]): +class ConsistencyCheckerAgent(AgentBase[ScriptConsistencyCheckResult]): """一致性检查(角色混淆):输入原文,检测同一角色身份/行为混淆并给出修改建议。""" - CONSISTENCY_CHECKER_SKILL_IDS = ("consistency_checker",) + @property + def system_prompt(self) -> str: + return _CONSISTENCY_CHECKER_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return CONSISTENCY_CHECKER_PROMPT - registry = get_skill_registry() - if skill_id not in self.CONSISTENCY_CHECKER_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid consistency_checker skill_id: {skill_id}. " - f"Allowed: {self.CONSISTENCY_CHECKER_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ScriptConsistencyCheckResult]: + return ScriptConsistencyCheckResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """规范化一致性检查结果(角色混淆)。""" @@ -438,23 +586,20 @@ class ConsistencyCheckerAgent(SkillAgentBase[ScriptConsistencyCheckResult]): return data -class ScriptOptimizerAgent(SkillAgentBase[ScriptOptimizationResult]): +class ScriptOptimizerAgent(AgentBase[ScriptOptimizationResult]): """剧本优化 Agent:输入一致性检查输出 + 原文,输出优化后的剧本。""" - SCRIPT_OPTIMIZER_SKILL_IDS = ("script_optimizer",) + @property + def system_prompt(self) -> str: + return _SCRIPT_OPTIMIZER_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return SCRIPT_OPTIMIZER_PROMPT - registry = get_skill_registry() - if skill_id not in self.SCRIPT_OPTIMIZER_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid script_optimizer skill_id: {skill_id}. " - f"Allowed: {self.SCRIPT_OPTIMIZER_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ScriptOptimizationResult]: + return ScriptOptimizationResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: data = dict(data) @@ -469,23 +614,20 @@ class ScriptOptimizerAgent(SkillAgentBase[ScriptOptimizationResult]): # 6. OutputCompilerAgent - 最终输出打包 # ============================================================================ -class OutputCompilerAgent(SkillAgentBase[OutputCompileResult]): +class OutputCompilerAgent(AgentBase[OutputCompileResult]): """最终输出打包:输入所有Agent状态,输出完整项目JSON + 表格数据。""" - OUTPUT_COMPILER_SKILL_IDS = ("output_compiler",) + @property + def system_prompt(self) -> str: + return _OUTPUT_COMPILER_SYSTEM_PROMPT - def load_skill(self, skill_id: str) -> None: - from app.core.skills_runtime import get_skill_registry + @property + def prompt_template(self) -> PromptTemplate: + return OUTPUT_COMPILER_PROMPT - registry = get_skill_registry() - if skill_id not in self.OUTPUT_COMPILER_SKILL_IDS or skill_id not in registry: - raise ValueError( - f"Unknown or invalid output_compiler skill_id: {skill_id}. " - f"Allowed: {self.OUTPUT_COMPILER_SKILL_IDS}" - ) - self._prompt, self._output_model = registry[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[OutputCompileResult]: + return OutputCompileResult def _normalize(self, data: dict[str, Any]) -> dict[str, Any]: """规范化输出编译结果。""" diff --git a/backend/app/chains/agents/shot_frame_prompt_agents.py b/backend/app/chains/agents/shot_frame_prompt_agents.py index 6550f51..24d2c6e 100644 --- a/backend/app/chains/agents/shot_frame_prompt_agents.py +++ b/backend/app/chains/agents/shot_frame_prompt_agents.py @@ -5,9 +5,10 @@ from __future__ import annotations import json from typing import Any -from app.core.skills_runtime import SKILL_REGISTRY -from app.core.skills_runtime.shot_frame_prompt_generator import ShotFramePromptResult -from app.chains.agents.base import SkillAgentBase, _extract_json_from_text +from langchain_core.prompts import PromptTemplate + +from app.chains.agents.base import AgentBase, _extract_json_from_text +from app.schemas.skills.shot_frame_prompt import ShotFramePromptResult def _prepare_shot_frame_input(input_dict: dict[str, Any]) -> dict[str, Any]: @@ -24,20 +25,104 @@ def _prepare_shot_frame_input(input_dict: dict[str, Any]) -> dict[str, Any]: return out -class ShotFirstFramePromptAgent(SkillAgentBase[ShotFramePromptResult]): +_SHOT_FRAME_INPUT_VARS = [ + "script_excerpt", + "title", + "camera_shot", + "angle", + "movement", + "atmosphere", + "mood_tags", + "vfx_type", + "vfx_note", + "duration", + "scene_id", + "dialog_summary", +] + +_FIRST_FRAME_TEMPLATE = """你是一名分镜师。根据下列镜头信息,生成该镜头的**首帧**画面描述提示词(用于图像生成模型)。 +要求:一句或几句简洁、可视化的中文描述(不得使用英文),涵盖画面主体、景别、氛围、关键动作或状态。 +专有名词一致性规则: +- 剧本摘录(script_excerpt)中出现的角色名/场景名/道具名等专有名词,必须在输出中**原样保留**,不得翻译、不得替换为同义词或其他说法。 +- 除了需要补全画面描述以外,不要对专有名词进行改写。 +只输出一个 JSON 对象:{{"prompt": "你的提示词内容"}},不要其他文字。 + +## 镜头信息 +剧本摘录:{script_excerpt} +镜头标题:{title} +景别:{camera_shot} +机位角度:{angle} +运镜:{movement} +氛围:{atmosphere} +情绪标签:{mood_tags} +视效:{vfx_type} - {vfx_note} +时长:{duration}秒 +对白摘要:{dialog_summary} + +## 输出(仅首帧提示词,JSON:{{"prompt": "..."}}) +""" + +_LAST_FRAME_TEMPLATE = """你是一名分镜师。根据下列镜头信息,生成该镜头的**尾帧**画面描述提示词(用于图像生成模型)。 +要求:一句或几句简洁、可视化的中文描述(不得使用英文),描述镜头结束时的画面状态。 +专有名词一致性规则: +- 剧本摘录(script_excerpt)中出现的角色名/场景名/道具名等专有名词,必须在输出中**原样保留**,不得翻译、不得替换为同义词或其他说法。 +- 除了需要补全画面描述以外,不要对专有名词进行改写。 +只输出一个 JSON 对象:{{"prompt": "你的提示词内容"}},不要其他文字。 + +## 镜头信息 +剧本摘录:{script_excerpt} +镜头标题:{title} +景别:{camera_shot} +机位角度:{angle} +运镜:{movement} +氛围:{atmosphere} +情绪标签:{mood_tags} +视效:{vfx_type} - {vfx_note} +时长:{duration}秒 +对白摘要:{dialog_summary} + +## 输出(仅尾帧提示词,JSON:{{"prompt": "..."}}) +""" + +_KEY_FRAME_TEMPLATE = """你是一名分镜师。根据下列镜头信息,生成该镜头的**关键帧**画面描述提示词(用于图像生成模型)。 +要求:一句或几句简洁、可视化的中文描述(不得使用英文),捕捉该镜头中最具代表性的瞬间画面。 +专有名词一致性规则: +- 剧本摘录(script_excerpt)中出现的角色名/场景名/道具名等专有名词,必须在输出中**原样保留**,不得翻译、不得替换为同义词或其他说法。 +- 除了需要补全画面描述以外,不要对专有名词进行改写。 +只输出一个 JSON 对象:{{"prompt": "你的提示词内容"}},不要其他文字。 + +## 镜头信息 +剧本摘录:{script_excerpt} +镜头标题:{title} +景别:{camera_shot} +机位角度:{angle} +运镜:{movement} +氛围:{atmosphere} +情绪标签:{mood_tags} +视效:{vfx_type} - {vfx_note} +时长:{duration}秒 +对白摘要:{dialog_summary} + +## 输出(仅关键帧提示词,JSON:{{"prompt": "..."}}) +""" + +SHOT_FIRST_FRAME_PROMPT = PromptTemplate(input_variables=_SHOT_FRAME_INPUT_VARS, template=_FIRST_FRAME_TEMPLATE) +SHOT_LAST_FRAME_PROMPT = PromptTemplate(input_variables=_SHOT_FRAME_INPUT_VARS, template=_LAST_FRAME_TEMPLATE) +SHOT_KEY_FRAME_PROMPT = PromptTemplate(input_variables=_SHOT_FRAME_INPUT_VARS, template=_KEY_FRAME_TEMPLATE) + + +class ShotFirstFramePromptAgent(AgentBase[ShotFramePromptResult]): """镜头首帧提示词生成 Agent,输出可写入 ShotDetail.first_frame_prompt。""" - SKILL_ID = "shot_first_frame_prompt" + @property + def prompt_template(self) -> PromptTemplate: + return SHOT_FIRST_FRAME_PROMPT - def load_skill(self, skill_id: str = "shot_first_frame_prompt") -> None: - if skill_id not in (self.SKILL_ID,) or skill_id not in SKILL_REGISTRY: - raise ValueError(f"Invalid skill_id: {skill_id}. Allowed: ({self.SKILL_ID},)") - self._prompt, self._output_model = SKILL_REGISTRY[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ShotFramePromptResult]: + return ShotFramePromptResult def format_output(self, raw: str) -> ShotFramePromptResult: - self._ensure_loaded() json_str = _extract_json_from_text(raw) try: data = json.loads(json_str) @@ -47,31 +132,29 @@ class ShotFirstFramePromptAgent(SkillAgentBase[ShotFramePromptResult]): return ShotFramePromptResult(prompt=str(data["prompt"]).strip()) return ShotFramePromptResult(prompt=raw.strip()) - def extract(self, input_dict: dict[str, Any]) -> ShotFramePromptResult: - inp = _prepare_shot_frame_input(input_dict) - raw = self.run(inp) + def extract(self, **kwargs: Any) -> ShotFramePromptResult: + inp = _prepare_shot_frame_input(kwargs) + raw = self.run(**inp) return self.format_output(raw) - async def aextract(self, input_dict: dict[str, Any]) -> ShotFramePromptResult: - inp = _prepare_shot_frame_input(input_dict) - raw = await self.arun(inp) + async def aextract(self, **kwargs: Any) -> ShotFramePromptResult: + inp = _prepare_shot_frame_input(kwargs) + raw = await self.arun(**inp) return self.format_output(raw) -class ShotLastFramePromptAgent(SkillAgentBase[ShotFramePromptResult]): +class ShotLastFramePromptAgent(AgentBase[ShotFramePromptResult]): """镜头尾帧提示词生成 Agent,输出可写入 ShotDetail.last_frame_prompt。""" - SKILL_ID = "shot_last_frame_prompt" + @property + def prompt_template(self) -> PromptTemplate: + return SHOT_LAST_FRAME_PROMPT - def load_skill(self, skill_id: str = "shot_last_frame_prompt") -> None: - if skill_id not in (self.SKILL_ID,) or skill_id not in SKILL_REGISTRY: - raise ValueError(f"Invalid skill_id: {skill_id}. Allowed: ({self.SKILL_ID},)") - self._prompt, self._output_model = SKILL_REGISTRY[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ShotFramePromptResult]: + return ShotFramePromptResult def format_output(self, raw: str) -> ShotFramePromptResult: - self._ensure_loaded() json_str = _extract_json_from_text(raw) try: data = json.loads(json_str) @@ -81,31 +164,29 @@ class ShotLastFramePromptAgent(SkillAgentBase[ShotFramePromptResult]): return ShotFramePromptResult(prompt=str(data["prompt"]).strip()) return ShotFramePromptResult(prompt=raw.strip()) - def extract(self, input_dict: dict[str, Any]) -> ShotFramePromptResult: - inp = _prepare_shot_frame_input(input_dict) - raw = self.run(inp) + def extract(self, **kwargs: Any) -> ShotFramePromptResult: + inp = _prepare_shot_frame_input(kwargs) + raw = self.run(**inp) return self.format_output(raw) - async def aextract(self, input_dict: dict[str, Any]) -> ShotFramePromptResult: - inp = _prepare_shot_frame_input(input_dict) - raw = await self.arun(inp) + async def aextract(self, **kwargs: Any) -> ShotFramePromptResult: + inp = _prepare_shot_frame_input(kwargs) + raw = await self.arun(**inp) return self.format_output(raw) -class ShotKeyFramePromptAgent(SkillAgentBase[ShotFramePromptResult]): +class ShotKeyFramePromptAgent(AgentBase[ShotFramePromptResult]): """镜头关键帧提示词生成 Agent,输出可写入 ShotDetail.key_frame_prompt。""" - SKILL_ID = "shot_key_frame_prompt" + @property + def prompt_template(self) -> PromptTemplate: + return SHOT_KEY_FRAME_PROMPT - def load_skill(self, skill_id: str = "shot_key_frame_prompt") -> None: - if skill_id not in (self.SKILL_ID,) or skill_id not in SKILL_REGISTRY: - raise ValueError(f"Invalid skill_id: {skill_id}. Allowed: ({self.SKILL_ID},)") - self._prompt, self._output_model = SKILL_REGISTRY[skill_id] - self._skill_id = skill_id - self._structured_chain = None + @property + def output_model(self) -> type[ShotFramePromptResult]: + return ShotFramePromptResult def format_output(self, raw: str) -> ShotFramePromptResult: - self._ensure_loaded() json_str = _extract_json_from_text(raw) try: data = json.loads(json_str) @@ -115,12 +196,12 @@ class ShotKeyFramePromptAgent(SkillAgentBase[ShotFramePromptResult]): return ShotFramePromptResult(prompt=str(data["prompt"]).strip()) return ShotFramePromptResult(prompt=raw.strip()) - def extract(self, input_dict: dict[str, Any]) -> ShotFramePromptResult: - inp = _prepare_shot_frame_input(input_dict) - raw = self.run(inp) + def extract(self, **kwargs: Any) -> ShotFramePromptResult: + inp = _prepare_shot_frame_input(kwargs) + raw = self.run(**inp) return self.format_output(raw) - async def aextract(self, input_dict: dict[str, Any]) -> ShotFramePromptResult: - inp = _prepare_shot_frame_input(input_dict) - raw = await self.arun(inp) + async def aextract(self, **kwargs: Any) -> ShotFramePromptResult: + inp = _prepare_shot_frame_input(kwargs) + raw = await self.arun(**inp) return self.format_output(raw) diff --git a/backend/app/core/tasks/agent_tasks.py b/backend/app/core/tasks/agent_tasks.py index d056710..e39fc30 100644 --- a/backend/app/core/tasks/agent_tasks.py +++ b/backend/app/core/tasks/agent_tasks.py @@ -10,17 +10,14 @@ from __future__ import annotations from typing import Any, AsyncIterator, Optional from app.chains.agents import ( - FilmEntityExtractor, - FilmShotlistStoryboarder, + FilmEntityExtractorAgent, + FilmShotlistStoryboarderAgent, ShotFirstFramePromptAgent, ShotKeyFramePromptAgent, ShotLastFramePromptAgent, ) -from app.core.skills_runtime import ( - FilmEntityExtractionResult, - FilmShotlistResult, - ShotFramePromptResult, -) +from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult +from app.schemas.skills.shot_frame_prompt import ShotFramePromptResult from app.core.task_manager.types import BaseTask @@ -45,27 +42,24 @@ class _BaseAsyncResultTask(BaseTask): class FilmEntityExtractionTask2(_BaseAsyncResultTask): - """人物/地点/道具抽取任务(基于 FilmEntityExtractor)。""" + """人物/地点/道具抽取任务(基于 FilmEntityExtractorAgent)。""" TASK_KEY = "film_entity_extraction" def __init__( self, - extractor: FilmEntityExtractor, + extractor: FilmEntityExtractorAgent, *, input_dict: dict[str, Any], - skill_id: str = "film_entity_extractor", ) -> None: super().__init__() self._extractor = extractor self._input_dict = input_dict - self._skill_id = skill_id self._result: FilmEntityExtractionResult | None = None async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override] try: - self._extractor.load_skill(self._skill_id) - self._result = await self._extractor.aextract(self._input_dict) + self._result = await self._extractor.aextract(**self._input_dict) except Exception as exc: # noqa: BLE001 self._error = str(exc) self._result = None @@ -76,27 +70,24 @@ class FilmEntityExtractionTask2(_BaseAsyncResultTask): class FilmShotlistTask2(_BaseAsyncResultTask): - """分镜/镜头表抽取任务(基于 FilmShotlistStoryboarder)。""" + """分镜/镜头表抽取任务(基于 FilmShotlistStoryboarderAgent)。""" TASK_KEY = "film_shotlist" def __init__( self, - storyboarder: FilmShotlistStoryboarder, + storyboarder: FilmShotlistStoryboarderAgent, *, input_dict: dict[str, Any], - skill_id: str = "film_shotlist", ) -> None: super().__init__() self._storyboarder = storyboarder self._input_dict = input_dict - self._skill_id = skill_id self._result: FilmShotlistResult | None = None async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override] try: - self._storyboarder.load_skill(self._skill_id) - self._result = await self._storyboarder.aextract(self._input_dict) + self._result = await self._storyboarder.aextract(**self._input_dict) except Exception as exc: # noqa: BLE001 self._error = str(exc) self._result = None @@ -116,28 +107,20 @@ class ShotFramePromptTask(_BaseAsyncResultTask): agent: ShotFirstFramePromptAgent | ShotLastFramePromptAgent | ShotKeyFramePromptAgent, *, input_dict: dict[str, Any], - skill_id: str, ) -> None: super().__init__() self._agent = agent self._input_dict = input_dict - self._skill_id = skill_id self._result: ShotFramePromptResult | None = None async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override] try: - self._agent.load_skill(self._skill_id) - self._result = await self._agent.aextract(self._input_dict) + self._result = await self._agent.aextract(**self._input_dict) except Exception as exc: # noqa: BLE001 self._error = str(exc) self._result = None return None - async def status(self) -> dict[str, Any]: # type: ignore[override] - base = await super().status() - base["skill_id"] = self._skill_id - return base - async def get_result(self) -> ShotFramePromptResult | None: # type: ignore[override] return self._result diff --git a/backend/app/core/tasks/extra_tasks.py b/backend/app/core/tasks/extra_tasks.py index 4f5fdb2..5c27eac 100644 --- a/backend/app/core/tasks/extra_tasks.py +++ b/backend/app/core/tasks/extra_tasks.py @@ -1,7 +1,7 @@ """将 core/agents 的抽取器封装为 BaseTask 任务。 目的: -- 让 FilmEntityExtractor / FilmShotlistStoryboarder 可以直接作为 TaskManager 的任务单元 +- 让 FilmEntityExtractorAgent / FilmShotlistStoryboarderAgent 可以直接作为 TaskManager 的任务单元 - run() 采用 async_result 模式:返回 None,结果通过 get_result() 获取 """ @@ -9,8 +9,8 @@ from __future__ import annotations from typing import Any, AsyncIterator, Optional -from app.chains.agents import FilmEntityExtractor, FilmShotlistStoryboarder -from app.core.skills_runtime import FilmEntityExtractionResult, FilmShotlistResult +from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent +from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult from app.core.task_manager.types import BaseTask @@ -19,21 +19,18 @@ class FilmEntityExtractionTask(BaseTask): def __init__( self, - extractor: FilmEntityExtractor, + extractor: FilmEntityExtractorAgent, *, input_dict: dict[str, Any], - skill_id: str = "film_entity_extractor", ) -> None: self._extractor = extractor self._input_dict = input_dict - self._skill_id = skill_id self._result: FilmEntityExtractionResult | None = None self._error: str = "" async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override] try: - self._extractor.load_skill(self._skill_id) - self._result = await self._extractor.aextract(self._input_dict) + self._result = await self._extractor.aextract(**self._input_dict) except Exception as exc: # noqa: BLE001 self._error = str(exc) self._result = None @@ -42,7 +39,6 @@ class FilmEntityExtractionTask(BaseTask): async def status(self) -> dict[str, Any]: # type: ignore[override] return { "task": "film_entity_extraction", - "skill_id": self._skill_id, "done": await self.is_done(), "has_result": self._result is not None, "error": self._error, @@ -60,21 +56,18 @@ class FilmShotlistTask(BaseTask): def __init__( self, - storyboarder: FilmShotlistStoryboarder, + storyboarder: FilmShotlistStoryboarderAgent, *, input_dict: dict[str, Any], - skill_id: str = "film_shotlist", ) -> None: self._storyboarder = storyboarder self._input_dict = input_dict - self._skill_id = skill_id self._result: FilmShotlistResult | None = None self._error: str = "" async def run(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any] | None: # type: ignore[override] try: - self._storyboarder.load_skill(self._skill_id) - self._result = await self._storyboarder.aextract(self._input_dict) + self._result = await self._storyboarder.aextract(**self._input_dict) except Exception as exc: # noqa: BLE001 self._error = str(exc) self._result = None @@ -83,7 +76,6 @@ class FilmShotlistTask(BaseTask): async def status(self) -> dict[str, Any]: # type: ignore[override] return { "task": "film_shotlist", - "skill_id": self._skill_id, "done": await self.is_done(), "has_result": self._result is not None, "error": self._error, diff --git a/backend/app/dependencies.py b/backend/app/dependencies.py index d1d7350..6ea0012 100644 --- a/backend/app/dependencies.py +++ b/backend/app/dependencies.py @@ -3,6 +3,7 @@ from collections.abc import AsyncGenerator from fastapi import HTTPException +from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.runnables import Runnable from sqlalchemy.ext.asyncio import AsyncSession @@ -23,7 +24,7 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]: await session.close() -def get_llm() -> Runnable: +def get_llm() -> BaseChatModel: """提供 LLM(ChatOpenAI),用于影视技能抽取。未配置 OPENAI_API_KEY 时抛出 503。""" if not settings.openai_api_key: raise HTTPException( diff --git a/backend/app/models/types.py b/backend/app/models/types.py index d794818..2c15aec 100644 --- a/backend/app/models/types.py +++ b/backend/app/models/types.py @@ -36,7 +36,7 @@ class ShotStatus(str, Enum): class CameraShotType(str, Enum): - """景别(与 `app.core.skills_runtime.schemas.ShotType` 对齐,存英文 code)。""" + """景别(与 `app.schemas.skills.common.ShotType` 对齐,存英文 code)。""" ecu = "ECU" # 大特写 cu = "CU" # 特写 @@ -48,7 +48,7 @@ class CameraShotType(str, Enum): class CameraAngle(str, Enum): - """机位角度(与 `app.core.skills_runtime.schemas.CameraAngle` 对齐,存英文 code)。""" + """机位角度(与 `app.schemas.skills.common.CameraAngle` 对齐,存英文 code)。""" eye_level = "EYE_LEVEL" # 平视 high_angle = "HIGH_ANGLE" # 高角度 @@ -59,7 +59,7 @@ class CameraAngle(str, Enum): class CameraMovement(str, Enum): - """运镜方式(与 `app.core.skills_runtime.schemas.CameraMovement` 对齐,存英文 code)。""" + """运镜方式(与 `app.schemas.skills.common.CameraMovement` 对齐,存英文 code)。""" static = "STATIC" # 静止 pan = "PAN" # 平移 @@ -118,7 +118,7 @@ class TimelineClipType(str, Enum): class DialogueLineMode(str, Enum): - """对白模式(与 `app.core.skills_runtime.schemas.DialogueLineMode` 对齐,存英文 code)。""" + """对白模式(与 `app.schemas.skills.common.DialogueLineMode` 对齐,存英文 code)。""" dialogue = "DIALOGUE" # 对白 voice_over = "VOICE_OVER" # 旁白 @@ -127,7 +127,7 @@ class DialogueLineMode(str, Enum): class VFXType(str, Enum): - """视效类型(与 `app.core.skills_runtime.schemas.VFXType` 对齐,存英文 code)。""" + """视效类型(与 `app.schemas.skills.common.VFXType` 对齐,存英文 code)。""" none = "NONE" # 无 particles = "PARTICLES" # 粒子 diff --git a/backend/app/schemas/skills/__init__.py b/backend/app/schemas/skills/__init__.py new file mode 100644 index 0000000..a594102 --- /dev/null +++ b/backend/app/schemas/skills/__init__.py @@ -0,0 +1,143 @@ +from app.schemas.skills.common import ( + CAMERA_ANGLE_ZH, + CAMERA_MOVEMENT_ZH, + COMPLEXITY_ZH, + DIALOGUE_LINE_MODE_ZH, + PROP_CATEGORY_ZH, + SCENE_INTERIOR_ZH, + SCENE_TIME_ZH, + SHOT_TYPE_ZH, + TRANSITION_TYPE_ZH, + VFX_TYPE_ZH, + CameraAngle, + CameraMovement, + ComplexityLevel, + DialogueLine, + DialogueLineMode, + EvidenceSpan, + SceneInterior, + SceneTime, + ShotType, + Transition, + TransitionType, + Uncertainty, + VFXNote, + VFXType, +) +from app.schemas.skills.film import ( + Character, + FilmEntityExtractionResult, + FilmShotlistResult, + Location, + ProjectCinematicBreakdown, + Prop, + Scene, + Shot, + TextChunk, +) +from app.schemas.skills.script_processing import ( + ConflictResolutionSuggestion, + ConsistencyCheckResult, + ConsistencyWarning, + CostumeTimeline, + CostumeTimelineEntry, + EntityEntry, + EntityLibrary, + EntityMergeResult, + EntityVariant, + OutputCompileResult, + ScriptConsistencyCheckResult, + ScriptConsistencyIssue, + ScriptDivisionResult, + ScriptExtractionResult, + ScriptOptimizationResult, + ScriptShotExtraction, + ShotCharacterInfo, + ShotDivision, + ShotElementExtractionResult, + ShotElements, + ShotPropInfo, + ShotSceneInfo, + StudioAssetDraft, + StudioCharacterDraft, + StudioScriptExtractionDraft, + StudioShotDraft, + StudioShotDraftDialogueLine, + TableData, + VariantAnalysisResult, + VariantSuggestion, +) +from app.schemas.skills.shot_frame_prompt import ShotFramePromptInput, ShotFramePromptResult + +__all__ = [ + # common + "DialogueLineMode", + "EvidenceSpan", + "DialogueLine", + "SceneTime", + "SceneInterior", + "ShotType", + "CameraAngle", + "CameraMovement", + "TransitionType", + "VFXType", + "ComplexityLevel", + "DIALOGUE_LINE_MODE_ZH", + "SCENE_TIME_ZH", + "SCENE_INTERIOR_ZH", + "SHOT_TYPE_ZH", + "CAMERA_ANGLE_ZH", + "CAMERA_MOVEMENT_ZH", + "TRANSITION_TYPE_ZH", + "VFX_TYPE_ZH", + "COMPLEXITY_ZH", + "PROP_CATEGORY_ZH", + "VFXNote", + "Transition", + "Uncertainty", + # film + "Character", + "Location", + "Prop", + "Scene", + "Shot", + "ProjectCinematicBreakdown", + "TextChunk", + "FilmEntityExtractionResult", + "FilmShotlistResult", + # shot frame prompt + "ShotFramePromptInput", + "ShotFramePromptResult", + # script processing + "ShotDivision", + "ScriptDivisionResult", + "ShotCharacterInfo", + "ShotPropInfo", + "ShotSceneInfo", + "ShotElements", + "ShotElementExtractionResult", + "EntityVariant", + "EntityEntry", + "EntityLibrary", + "EntityMergeResult", + "ConflictResolutionSuggestion", + "ScriptConsistencyIssue", + "ScriptConsistencyCheckResult", + "ScriptOptimizationResult", + "ScriptShotExtraction", + "ScriptExtractionResult", + "StudioAssetDraft", + "StudioCharacterDraft", + "StudioShotDraftDialogueLine", + "StudioShotDraft", + "StudioScriptExtractionDraft", + "CostumeTimelineEntry", + "CostumeTimeline", + "VariantSuggestion", + "VariantAnalysisResult", + "ConsistencyWarning", + "ConsistencyCheckResult", + "TableData", + "OutputCompileResult", +] + diff --git a/backend/app/schemas/skills/common.py b/backend/app/schemas/skills/common.py new file mode 100644 index 0000000..c1c3ade --- /dev/null +++ b/backend/app/schemas/skills/common.py @@ -0,0 +1,246 @@ +"""影视/脚本处理链路的通用结构与枚举映射。""" + +from __future__ import annotations + +from typing import Dict, List, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +# ---------- Common ---------- +DialogueLineMode = Literal["DIALOGUE", "VOICE_OVER", "OFF_SCREEN", "PHONE"] + + +class EvidenceSpan(BaseModel): + """可追溯证据:原文定位(chunk + 起止位置/摘录),用于审核与回查。""" + + model_config = ConfigDict(extra="forbid") + + chunk_id: str = Field(..., description="输入文本块的唯一ID(例如 chapter1_p03)") + start_char: Optional[int] = Field(None, description="在该 chunk 中的起始字符位置(可选)") + end_char: Optional[int] = Field(None, description="在该 chunk 中的结束字符位置(可选)") + quote: Optional[str] = Field(None, description="不超过200字的原文摘录(可选,便于人工审核)") + + +class DialogueLine(BaseModel): + """单条对白:说话人/对象、正文、情绪与表达方式、旁白/电话等模式、镜头内时间点。""" + + model_config = ConfigDict(extra="forbid") + + index: Optional[int] = Field( + None, + ge=0, + description="可选:镜头内排序(脚本处理链路用于保持原始顺序)", + ) + speaker_character_id: Optional[str] = Field(None, description="说话人角色ID,若无法判定可为空") + target_character_id: Optional[str] = Field(None, description="对谁说(听者角色ID),可选") + text: str = Field(..., description="对白正文") + emotion: Optional[str] = Field(None, description="情绪/语气(如:愤怒、平静、哽咽)") + delivery: Optional[str] = Field(None, description="表达方式(如:低声、喊叫、旁白腔)") + line_mode: DialogueLineMode = Field( + "DIALOGUE", + description="DIALOGUE=正常对白, VOICE_OVER=旁白, OFF_SCREEN=画外音, PHONE=电话音等", + ) + start_time_sec: Optional[float] = Field(None, ge=0, description="在该镜头内相对起始时间(秒),用于对口型/字幕切分") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据") + + +# ---------- Story / Scenes ---------- +SceneTime = Literal["DAY", "NIGHT", "DAWN", "DUSK", "UNKNOWN"] +SceneInterior = Literal["INT", "EXT", "INT_EXT", "UNKNOWN"] + + +# ---------- Cinematic (Shots / Transitions / VFX) ---------- +ShotType = Literal["ECU", "CU", "MCU", "MS", "MLS", "LS", "ELS"] +CameraAngle = Literal["EYE_LEVEL", "HIGH_ANGLE", "LOW_ANGLE", "BIRD_EYE", "DUTCH", "OVER_SHOULDER"] +CameraMovement = Literal[ + "STATIC", + "PAN", + "TILT", + "DOLLY_IN", + "DOLLY_OUT", + "TRACK", + "CRANE", + "HANDHELD", + "STEADICAM", + "ZOOM_IN", + "ZOOM_OUT", +] +TransitionType = Literal["CUT", "DISSOLVE", "WIPE", "FADE_IN", "FADE_OUT", "MATCH_CUT", "J_CUT", "L_CUT"] +VFXType = Literal[ + "NONE", + "PARTICLES", + "VOLUMETRIC_FOG", + "CG_DOUBLE", + "DIGITAL_ENVIRONMENT", + "MATTE_PAINTING", + "FIRE_SMOKE", + "WATER_SIM", + "DESTRUCTION", + "ENERGY_MAGIC", + "COMPOSITING_CLEANUP", + "SLOW_MOTION_TIME", + "OTHER", +] +ComplexityLevel = Literal["LOW", "MEDIUM", "HIGH"] + + +# ---------- 英文枚举 → 中文映射(影视/分镜专业用语) ---------- +DIALOGUE_LINE_MODE_ZH: dict[str, str] = { + "DIALOGUE": "对白", + "VOICE_OVER": "旁白", + "OFF_SCREEN": "画外音", + "PHONE": "电话声", +} + +SCENE_TIME_ZH: dict[str, str] = { + "DAY": "日", + "NIGHT": "夜", + "DAWN": "黎明", + "DUSK": "黄昏", + "UNKNOWN": "未知", +} + +SCENE_INTERIOR_ZH: dict[str, str] = { + "INT": "内景", + "EXT": "外景", + "INT_EXT": "内景兼外景", + "UNKNOWN": "未知", +} + +SHOT_TYPE_ZH: dict[str, str] = { + "ECU": "大特写", + "CU": "特写", + "MCU": "中近景", + "MS": "中景", + "MLS": "中远景", + "LS": "远景", + "ELS": "大远景", +} + +CAMERA_ANGLE_ZH: dict[str, str] = { + "EYE_LEVEL": "平视", + "HIGH_ANGLE": "俯拍", + "LOW_ANGLE": "仰拍", + "BIRD_EYE": "鸟瞰", + "DUTCH": "荷兰角", + "OVER_SHOULDER": "过肩", +} + +CAMERA_MOVEMENT_ZH: dict[str, str] = { + "STATIC": "固定", + "PAN": "横摇", + "TILT": "俯仰", + "DOLLY_IN": "推轨", + "DOLLY_OUT": "拉轨", + "TRACK": "跟拍", + "CRANE": "升降", + "HANDHELD": "手持", + "STEADICAM": "斯坦尼康", + "ZOOM_IN": "变焦推进", + "ZOOM_OUT": "变焦拉远", +} + +TRANSITION_TYPE_ZH: dict[str, str] = { + "CUT": "切", + "DISSOLVE": "叠化", + "WIPE": "划变", + "FADE_IN": "淡入", + "FADE_OUT": "淡出", + "MATCH_CUT": "匹配剪辑", + "J_CUT": "J 剪", + "L_CUT": "L 剪", +} + +VFX_TYPE_ZH: dict[str, str] = { + "NONE": "无", + "PARTICLES": "粒子", + "VOLUMETRIC_FOG": "体积雾", + "CG_DOUBLE": "数字替身", + "DIGITAL_ENVIRONMENT": "数字场景", + "MATTE_PAINTING": "绘景", + "FIRE_SMOKE": "烟火", + "WATER_SIM": "水效", + "DESTRUCTION": "破碎/解算", + "ENERGY_MAGIC": "能量/魔法", + "COMPOSITING_CLEANUP": "合成/修脏", + "SLOW_MOTION_TIME": "升格/慢动作", + "OTHER": "其他", +} + +COMPLEXITY_ZH: dict[str, str] = { + "LOW": "低", + "MEDIUM": "中", + "HIGH": "高", +} + +PROP_CATEGORY_ZH: dict[str, str] = { + "weapon": "武器", + "document": "文书/证件", + "vehicle": "载具", + "clothing": "服装", + "device": "器械/设备", + "magic_item": "魔法/特殊物品", + "other": "其他", +} + + +class VFXNote(BaseModel): + """单条视效说明:类型、描述、复杂度及原文依据。""" + + model_config = ConfigDict(extra="forbid") + + vfx_type: VFXType = "NONE" + description: Optional[str] = Field(None, description="视效说明(简短、可执行)") + complexity: Optional[ComplexityLevel] = Field(None, description="粗略复杂度") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(若为忠实抽取)") + + +class Transition(BaseModel): + """镜头间转场:从哪镜到哪镜、转场类型及可选说明。""" + + model_config = ConfigDict(extra="forbid") + + from_shot_id: str + to_shot_id: str + transition: TransitionType = "CUT" + note: Optional[str] = Field(None, description="为何用该转场(可选)") + + +class Uncertainty(BaseModel): + """结构化不确定项:字段路径、原因及可选证据,便于人工审核与回溯。""" + + model_config = ConfigDict(extra="forbid") + + field_path: str = Field(..., description="如 characters[0].name、scenes[2].location_id") + reason: str = Field(..., description="不确定原因简述") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="相关原文依据(可选)") + + +__all__ = [ + "DialogueLineMode", + "EvidenceSpan", + "DialogueLine", + "SceneTime", + "SceneInterior", + "ShotType", + "CameraAngle", + "CameraMovement", + "TransitionType", + "VFXType", + "ComplexityLevel", + "DIALOGUE_LINE_MODE_ZH", + "SCENE_TIME_ZH", + "SCENE_INTERIOR_ZH", + "SHOT_TYPE_ZH", + "CAMERA_ANGLE_ZH", + "CAMERA_MOVEMENT_ZH", + "TRANSITION_TYPE_ZH", + "VFX_TYPE_ZH", + "COMPLEXITY_ZH", + "PROP_CATEGORY_ZH", + "VFXNote", + "Transition", + "Uncertainty", +] + diff --git a/backend/app/schemas/skills/film.py b/backend/app/schemas/skills/film.py new file mode 100644 index 0000000..c9a6dcc --- /dev/null +++ b/backend/app/schemas/skills/film.py @@ -0,0 +1,201 @@ +"""影视抽取/分镜相关的结构化输出模型(API 暴露用)。""" + +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel, ConfigDict, Field + +from app.schemas.skills.common import ( + CameraAngle, + CameraMovement, + DialogueLine, + EvidenceSpan, + SceneInterior, + SceneTime, + ShotType, + Transition, + Uncertainty, + VFXNote, +) + + +# ---------- Entities ---------- +class Character(BaseModel): + """从小说中抽取的角色:主名、别名、外貌与性格、服装描述、首次出场证据及抽取置信度。""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="稳定ID,例如 char_001") + name: str = Field(..., description="主名称(尽量取原文最常用的称呼)") + normalized_name: Optional[str] = Field(None, description="归一化主名,如将「王二/二哥/王二哥」统一为同一主名(来自文本)") + aliases: List[str] = Field(default_factory=list, description="别名/称呼(原文出现过的)") + description: Optional[str] = Field(None, description="外貌/身份/气质(忠实原文,与服装区分)") + costume_note: Optional[str] = Field( + None, + description="从原文抽取的服装/造型描述(如款式、颜色、配饰),与 description 区分,便于后续关联服装资产", + ) + traits: List[str] = Field(default_factory=list, description="性格/特征词(尽量来自原文)") + confidence: Optional[float] = Field(None, ge=0, le=1, description="抽取确定度 0-1(模型输出)") + first_appearance: Optional[EvidenceSpan] = None + + +class Location(BaseModel): + """从小说中抽取的地点:名称、类型、场景描写及首次出场证据、置信度。""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="稳定ID,例如 loc_001") + name: str = Field(..., description="地点名称(原文)") + normalized_name: Optional[str] = Field(None, description="归一化名称(来自文本)") + type: Optional[str] = Field(None, description="地点类型:房间/街道/森林/车厢等(可选)") + description: Optional[str] = Field(None, description="场景描写(忠实原文,简短)") + confidence: Optional[float] = Field(None, ge=0, le=1, description="抽取确定度 0-1(模型输出)") + first_appearance: Optional[EvidenceSpan] = None + + +class Prop(BaseModel): + """从小说中抽取的道具:名称、类别、外观/用途、归属角色及首次出场证据、置信度。""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="稳定ID,例如 prop_001") + name: str = Field(..., description="道具名称(原文)") + normalized_name: Optional[str] = Field(None, description="归一化名称(来自文本)") + category: Optional[str] = Field(None, description="可选:weapon/document/vehicle/clothing/device/magic_item/other") + description: Optional[str] = Field(None, description="外观/用途(忠实原文)") + owner_character_id: Optional[str] = Field(None, description="拥有者(如果明确)") + confidence: Optional[float] = Field(None, ge=0, le=1, description="抽取确定度 0-1(模型输出)") + first_appearance: Optional[EvidenceSpan] = None + + +# ---------- Story / Scenes ---------- +class Scene(BaseModel): + """场景:内/外景、时间、关联地点与人物/道具,含原文标题与系统格式化标题。""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="例如 scene_001") + raw_title: Optional[str] = Field(None, description="来自原文的场景标题(若存在)") + formatted_title: Optional[str] = Field(None, description="系统生成的影视格式标题,如 INT. 地点 - TIME") + interior: SceneInterior = "UNKNOWN" + time_of_day: SceneTime = "UNKNOWN" + + location_id: Optional[str] = Field(None, description="loc_xxx,如可判定") + summary: Optional[str] = Field(None, description="场景发生了什么(忠实原文,短)") + + character_ids: List[str] = Field(default_factory=list, description="该场景出现的人物ID") + prop_ids: List[str] = Field(default_factory=list, description="该场景关键道具ID") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="支持该场景的证据片段(可多条)") + + +class Shot(BaseModel): + """单镜头:景别/机位/运镜、时长与画面描述、对白列表、音效与视效、关联角色与道具。""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="例如 shot_001_003(scene_001 第3镜)") + scene_id: str = Field(..., description="所属 scene_xxx") + order: int = Field(..., ge=1, description="场景内镜头序号,从1开始") + + shot_type: ShotType + camera_angle: CameraAngle = "EYE_LEVEL" + camera_movement: CameraMovement = "STATIC" + + # 可拍性/执行信息 + duration_sec: Optional[float] = Field(None, ge=0.5, le=30, description="建议时长(可选)") + description: str = Field(..., description="镜头里发生的动作/画面(行业口吻,简短可拍)") + + character_ids: List[str] = Field(default_factory=list) + prop_ids: List[str] = Field(default_factory=list) + + vfx: List[VFXNote] = Field(default_factory=list) + sfx: List[str] = Field(default_factory=list, description="音效提示,如 footsteps, rain, explosion(可选)") + dialogue_lines: List[DialogueLine] = Field( + default_factory=list, + description="该镜头内的对白列表(结构化:说话人、对象、情绪、旁白/电话等、时间点)", + ) + dialogue: Optional[str] = Field(None, description="[兼容] 若该镜头承载关键对白,可摘录/概述(可选)") + + evidence: List[EvidenceSpan] = Field(default_factory=list, description="对应原文依据(可选)") + + +class ProjectCinematicBreakdown(BaseModel): + """从小说抽取的完整影视分镜:元信息、实体表、场景表、镜头表、转场表及不确定项。""" + + model_config = ConfigDict(extra="forbid") + + source_id: str = Field(..., description="小说/章节标识,例如 novel_x_ch05") + source_title: Optional[str] = Field(None, description="书名/章节名(从书名页或章节头抽取)") + source_author: Optional[str] = Field(None, description="作者(若可从文本抽取)") + language: Optional[str] = Field(None, description="如 zh、en,便于后续提示词与分词") + extraction_version: Optional[str] = Field(None, description="本次抽取器版本,便于回溯差异") + schema_version: Optional[str] = Field(None, description="本输出使用的 schema 版本") + + chunks: List[str] = Field(default_factory=list, description="本次处理的 chunk_id 列表") + + characters: List[Character] = Field(default_factory=list) + locations: List[Location] = Field(default_factory=list) + props: List[Prop] = Field(default_factory=list) + + scenes: List[Scene] = Field(default_factory=list) + shots: List[Shot] = Field(default_factory=list) + transitions: List[Transition] = Field(default_factory=list) + + notes: List[str] = Field(default_factory=list, description="全局备注/不确定点(可选)") + uncertainties: List[Uncertainty] = Field( + default_factory=list, + description="结构化不确定项:field_path、reason、evidence", + ) + + +# ---------- Skill Results (API-level) ---------- +class TextChunk(BaseModel): + """输入的文本块。chunk_id 会写入 EvidenceSpan.chunk_id 用于回查。""" + + model_config = ConfigDict(extra="forbid") + + chunk_id: str = Field(..., description="例如 chapter1_p03 / ch05_para12") + text: str = Field(..., description="该 chunk 的原文文本") + + +class FilmEntityExtractionResult(BaseModel): + """实体抽取结果(复用 Character/Location/Prop/Uncertainty)。""" + + model_config = ConfigDict(extra="forbid") + + source_id: str = Field(..., description="小说/章节标识,例如 novel_x_ch05") + language: Optional[str] = Field(None, description="如 zh/en(可选)") + extraction_version: Optional[str] = Field(None, description="抽取器版本(可选)") + schema_version: Optional[str] = Field(None, description="schemas 版本(可选)") + + chunks: List[str] = Field(default_factory=list, description="本次处理的 chunk_id 列表") + + characters: List[Character] = Field(default_factory=list) + locations: List[Location] = Field(default_factory=list) + props: List[Prop] = Field(default_factory=list) + + notes: List[str] = Field(default_factory=list, description="全局备注(可选)") + uncertainties: List[Uncertainty] = Field(default_factory=list, description="结构化不确定项") + + +class FilmShotlistResult(BaseModel): + """分镜输出(直接复用 ProjectCinematicBreakdown)。""" + + model_config = ConfigDict(extra="forbid") + + breakdown: ProjectCinematicBreakdown = Field(..., description="影视化拆解结果") + + +__all__ = [ + "Character", + "Location", + "Prop", + "Scene", + "Shot", + "ProjectCinematicBreakdown", + "TextChunk", + "FilmEntityExtractionResult", + "FilmShotlistResult", +] + diff --git a/backend/app/schemas/skills/script_processing.py b/backend/app/schemas/skills/script_processing.py new file mode 100644 index 0000000..4128d65 --- /dev/null +++ b/backend/app/schemas/skills/script_processing.py @@ -0,0 +1,468 @@ +"""脚本处理链路的结构化输出模型(分镜、元素提取、合并、优化、导入草稿等)。""" + +from __future__ import annotations + +from typing import Any, Dict, List, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field + +from app.schemas.skills.common import DialogueLine, DialogueLineMode, EvidenceSpan +from app.schemas.skills.film import Character, Location, Prop, Scene, ProjectCinematicBreakdown + + +SceneTimeLoose = Literal["DAY", "NIGHT", "DAWN", "DUSK", "UNKNOWN", "日", "夜", "黎明", "黄昏", "不明", "未知"] + + +class ShotDivision(BaseModel): + """剧本分镜中的单镜信息:行号 + 预览文本(可选弱语义)。""" + + model_config = ConfigDict(extra="forbid") + + index: int = Field(..., ge=1, description="镜头序号(章节内唯一)") + start_line: int = Field(..., ge=1, description="起始行号(1-based)") + end_line: int = Field(..., ge=1, description="结束行号(1-based)") + script_excerpt: str = Field(..., description="镜头对应的剧本摘录/文本") + + # 强区分:shot_name 是“分镜名/镜头标题”,scene_name 是“场景名” + shot_name: str = Field("", description="镜头名称(分镜名/镜头标题,勿与 scene_name 混用)") + + # 弱语义:此阶段不强制稳定ID + scene_name: Optional[str] = Field(None, description="场景名称(可选,来自原文或推断)") + time_of_day: Optional[SceneTimeLoose] = Field(None, description="时间(日/夜/未知等,可选)") + character_names_in_text: List[str] = Field(default_factory=list, description="本镜出现的角色名/称呼(弱信息)") + + +class ScriptDivisionResult(BaseModel): + """剧本分镜结果:镜头列表(每镜起止行号+预览文本)。""" + + model_config = ConfigDict(extra="forbid") + + shots: List[ShotDivision] = Field(default_factory=list, description="分镜列表") + total_shots: int = Field(..., ge=0, description="总镜头数") + notes: Optional[str] = Field(None, description="拆分说明或建议(可选)") + + +class ShotCharacterInfo(BaseModel): + """单镜中单个角色的提取信息(弱ID/可溯源)。""" + + model_config = ConfigDict(extra="forbid") + + character_key: str = Field(..., description="角色键(可为临时ID或归一化名;合并阶段再分配稳定ID)") + name_in_text: Optional[str] = Field(None, description="本镜文本中出现的写法(昵称/小名等)") + + appearance: str = Field("", description="外貌描述:年龄感、体型、发型、五官、疤痕、肤色等") + clothing: str = Field("", description="本镜服装描述:款式、颜色、材质、状态(破损、湿透、沾泥等)") + accessories: str = Field("", description="配饰、眼镜、帽子、首饰等") + state: str = Field("", description="本镜临时状态:情绪主导、受伤、脏污、疲惫等") + + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + raw_appearance_text: str = Field("", description="导致 appearance 字段的原始剧本片段") + raw_clothing_text: str = Field("", description="导致 clothing 字段的原始剧本片段") + raw_state_text: str = Field("", description="导致 state 字段的原始剧本片段") + + +class ShotPropInfo(BaseModel): + """单镜中单个道具的提取信息(弱ID/可溯源)。""" + + model_config = ConfigDict(extra="forbid") + + prop_key: str = Field(..., description="道具键(可为临时ID或归一化名;合并阶段再分配稳定ID)") + name_in_text: Optional[str] = Field(None, description="本镜文本中出现的写法") + description: str = Field("", description="外观、材质、颜色、尺寸、特殊标记等") + state: str = Field("", description="当前状态:全新、破损、沾血、打开/关闭等") + interaction: str = Field("", description="在本镜的使用方式:拿起、丢弃、指向、使用等") + + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + raw_text: str = Field("", description="导致以上描述的原始文本片段") + + +class ShotSceneInfo(BaseModel): + """单镜场景描述(弱ID/可溯源)。""" + + model_config = ConfigDict(extra="forbid") + + scene_key: str = Field(..., description="场景键(可为临时ID或归一化名;合并阶段再分配稳定ID)") + name: Optional[str] = Field(None, description="场景名称(可选)") + location_detail: str = Field("", description="具体地点描述:老旧客厅、雨中街头、废弃工厂等") + atmosphere: str = Field("", description="氛围、光线、色调、声音暗示等") + time_weather: str = Field("", description="时间+天气补充(如果文本中有)") + + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + raw_description_text: str = Field("", description="场景相关的主要描述原文") + + +class ShotElements(BaseModel): + """单镜提取的元素信息(索引 + 细粒度 + 对白/动作)。""" + + model_config = ConfigDict(extra="forbid") + + character_keys: List[str] = Field(default_factory=list, description="角色键列表(弱ID)") + scene_keys: List[str] = Field(default_factory=list, description="场景键列表(弱ID)") + costume_keys: List[str] = Field(default_factory=list, description="服装键列表(弱ID)") + prop_keys: List[str] = Field(default_factory=list, description="道具键列表(弱ID)") + + characters_detailed: List[ShotCharacterInfo] = Field(default_factory=list, description="角色细粒度提取") + props_detailed: List[ShotPropInfo] = Field(default_factory=list, description="道具细粒度提取") + scene_detailed: Optional[ShotSceneInfo] = Field(None, description="场景细粒度提取") + + dialogue_lines: List[DialogueLine] = Field(default_factory=list, description="结构化对白列表") + actions: List[str] = Field(default_factory=list, description="动作/场景描述") + + shot_type_hints: List[str] = Field(default_factory=list, description="推断的镜头类型:close-up, wide-shot, tracking 等") + confidence_breakdown: Dict[str, float] = Field(default_factory=dict, description="各类别置信度") + + +class ShotElementExtractionResult(BaseModel): + """单镜信息提取结果。""" + + model_config = ConfigDict(extra="forbid") + + index: int = Field(..., ge=1, description="镜头序号(章节内唯一)") + shot_division: Optional[ShotDivision] = Field(None, description="分镜元信息(来自 ScriptDivider 输出)") + elements: ShotElements = Field(..., description="提取的元素") + confidence: Optional[float] = Field(None, ge=0, le=1, description="提取置信度(0-1)") + notes: Optional[str] = Field(None, description="提取说明或不确定项") + + +class EntityVariant(BaseModel): + """实体变体条目(最小可用结构,便于服装/外形演变)。""" + + model_config = ConfigDict(extra="forbid") + + variant_key: str = Field(..., description="变体键(例如 outfit_v1、wounded_state 等)") + description: Optional[str] = Field(None, description="变体描述(简短)") + affected_shots: List[int] = Field(default_factory=list, description="涉及镜头序号") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + + +class EntityEntry(BaseModel): + """合并后的实体条目(脚本处理中间态)。""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="实体稳定ID(合并阶段分配)") + name: str = Field(..., description="实体名称") + type: Literal["character", "scene", "prop", "location"] = Field(..., description="实体类型") + normalized_name: Optional[str] = Field(None, description="归一化名称(来自文本,可选)") + aliases: List[str] = Field(default_factory=list, description="别名/称呼(来自文本,可选)") + description: Optional[str] = Field(None, description="基础画像/描述(忠实文本,简短)") + confidence: Optional[float] = Field(None, ge=0, le=1, description="合并确定度 0-1(可选)") + first_appearance: Optional[EvidenceSpan] = Field(None, description="首次出场证据(可选)") + + costume_note: Optional[str] = Field(None, description="服装/造型描述(可选,便于变体与资产关联)") + traits: List[str] = Field(default_factory=list, description="性格/特征词(可选)") + + location_type: Optional[str] = Field(None, description="地点类型:房间/街道/森林/车厢等(可选)") + + category: Optional[str] = Field(None, description="道具类别(可选:weapon/document/vehicle/clothing/device/magic_item/other)") + owner_character_id: Optional[str] = Field(None, description="拥有者角色ID(可选)") + + evidence: List[EvidenceSpan] = Field(default_factory=list, description="支撑该实体画像的证据片段(可选)") + first_shot: Optional[int] = Field(None, description="首次出现的镜头序号") + appearances: List[int] = Field(default_factory=list, description="出现镜头列表") + variants: List[EntityVariant] = Field(default_factory=list, description="变体列表") + + +class EntityLibrary(BaseModel): + """合并后的实体库(脚本处理中间态)。""" + + model_config = ConfigDict(extra="forbid") + + characters: List[EntityEntry] = Field(default_factory=list, description="角色库") + locations: List[EntityEntry] = Field(default_factory=list, description="地点库") + scenes: List[EntityEntry] = Field(default_factory=list, description="场景库") + props: List[EntityEntry] = Field(default_factory=list, description="道具库") + total_entries: int = Field(..., ge=0, description="总实体数") + + +class EntityMergeResult(BaseModel): + """实体合并结果(脚本处理中间态)。""" + + model_config = ConfigDict(extra="forbid") + + merged_library: EntityLibrary = Field(..., description="合并后的实体库") + merge_stats: Dict[str, Any] = Field(default_factory=dict, description="合并统计信息") + conflicts: List[str] = Field(default_factory=list, description="发现的冲突/待处理项") + notes: Optional[str] = Field(None, description="合并说明") + + +class ConflictResolutionSuggestion(BaseModel): + """冲突解决建议(用于重试合并)。""" + + model_config = ConfigDict(extra="forbid") + + conflict: str = Field(..., description="与 EntityMergeResult.conflicts 对应的冲突描述(原文/原样)") + resolution: str = Field(..., description="解决建议(例如:保留哪个名字/合并哪些实体/拆分成变体等)") + entity_ids: List[str] = Field(default_factory=list, description="涉及的稳定实体ID(若已存在)") + preferred_entity_id: Optional[str] = Field(None, description="建议优先保留的实体ID(可选)") + notes: Optional[str] = Field(None, description="补充说明(可选)") + + +class ScriptConsistencyIssue(BaseModel): + """角色混淆类一致性问题:同一角色在不同镜头被赋予不同身份/行为主体导致混淆。""" + + model_config = ConfigDict(extra="forbid") + + issue_type: Literal["character_confusion"] = Field("character_confusion", description="固定为角色混淆类问题") + character_candidates: List[str] = Field(default_factory=list, description="涉及的角色候选(名字/称呼/ID 皆可,优先用原文称呼)") + description: str = Field(..., description="问题描述(为什么会混淆)") + suggestion: str = Field(..., description="修改建议(如何改写以消除混淆)") + affected_lines: Optional[Dict[str, int]] = Field(None, description="受影响的行号范围,形如 {start_line: x, end_line: y}") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + + +class ScriptConsistencyCheckResult(BaseModel): + """基于原文的一致性检查结果(聚焦角色混淆)。""" + + model_config = ConfigDict(extra="forbid") + + issues: List[ScriptConsistencyIssue] = Field(default_factory=list, description="问题列表") + has_issues: bool = Field(..., description="是否发现问题") + summary: Optional[str] = Field(None, description="总结(可选)") + + +class ScriptOptimizationResult(BaseModel): + """剧本优化输出:仅在发现角色混淆问题时使用。""" + + model_config = ConfigDict(extra="forbid") + + optimized_script_text: str = Field(..., description="优化后的剧本文本") + change_summary: str = Field(..., description="改动摘要(只围绕 issues)") + + +class ScriptShotExtraction(BaseModel): + """单镜提取结果(最终输出用,引用全局实体稳定ID)。""" + + model_config = ConfigDict(extra="forbid") + + shot_index: int = Field(..., ge=1, description="镜头序号(与 ScriptDivisionResult.shots.index 一致)") + shot_division: ShotDivision = Field(..., description="该镜头的分镜元信息") + + scene_id: Optional[str] = Field(None, description="所属场景 ID(scene_###,若可判定)") + character_ids: List[str] = Field(default_factory=list, description="本镜出现角色 ID 列表(char_###)") + prop_ids: List[str] = Field(default_factory=list, description="本镜关键道具 ID 列表(prop_###)") + location_ids: List[str] = Field(default_factory=list, description="本镜关联地点 ID 列表(loc_###)") + + dialogue_lines: List[DialogueLine] = Field(default_factory=list, description="结构化对白列表") + actions: List[str] = Field(default_factory=list, description="动作/场景描述") + notes: Optional[str] = Field(None, description="本镜补充说明(可选)") + + +class ScriptExtractionResult(BaseModel): + """最终输出:全局实体表 + 每镜关联(信息提取即最终结果)。""" + + model_config = ConfigDict(extra="forbid") + + source_id: str = Field(..., description="来源标识(如 demo_script_001)") + script_text: str = Field(..., description="输入剧本(可能为优化后版本)") + division: ScriptDivisionResult = Field(..., description="分镜结果") + + characters: List[Character] = Field(default_factory=list, description="全局角色表") + locations: List[Location] = Field(default_factory=list, description="全局地点表") + props: List[Prop] = Field(default_factory=list, description="全局道具表") + scenes: List[Scene] = Field(default_factory=list, description="全局场景表(可选填充)") + + shots: List[ScriptShotExtraction] = Field(default_factory=list, description="逐镜提取与关联结果") + + consistency: Optional[ScriptConsistencyCheckResult] = Field(None, description="一致性检查结果(可选,full-process 可回填)") + + +class StudioAssetDraft(BaseModel): + """Studio 资产草稿(Scene/Prop/Costume):不含 id,由导入 API 生成。""" + + model_config = ConfigDict(extra="forbid") + + name: str = Field(..., description="名称(同项目内建议唯一)") + description: str = Field("", description="描述") + tags: List[str] = Field(default_factory=list, description="标签") + prompt_template_id: Optional[str] = Field(None, description="提示词模板 ID(可空)") + view_count: int = Field(1, ge=1, description="计划生成视角图数量") + + +class StudioCharacterDraft(BaseModel): + """Studio 角色草稿:不含 id,由导入 API 生成。""" + + model_config = ConfigDict(extra="forbid") + + name: str = Field(..., description="角色名称(同项目内建议唯一)") + description: str = Field("", description="角色描述") + tags: List[str] = Field(default_factory=list, description="标签(可选)") + costume_name: Optional[str] = Field(None, description="服装名称(可选,导入时映射到 costume_id)") + prop_names: List[str] = Field(default_factory=list, description="角色常用道具名称列表(可选)") + + +class StudioShotDraftDialogueLine(BaseModel): + """镜头对白草稿:speaker/target 使用角色 name,导入时映射为 character_id。""" + + model_config = ConfigDict(extra="forbid") + + index: int = Field(0, ge=0, description="镜头内排序") + text: str = Field(..., description="台词内容") + line_mode: DialogueLineMode = Field("DIALOGUE", description="对白模式") + speaker_name: Optional[str] = Field(None, description="说话角色名称(可空)") + target_name: Optional[str] = Field(None, description="听者角色名称(可空)") + + +class StudioShotDraft(BaseModel): + """镜头草稿:不含 shot_id,由导入 API 生成;引用实体用 name。""" + + model_config = ConfigDict(extra="forbid") + + index: int = Field(..., ge=1, description="镜头序号(章节内唯一)") + title: str = Field(..., description="镜头标题") + script_excerpt: str = Field("", description="剧本摘录") + + scene_name: Optional[str] = Field(None, description="场景名称(可选)") + character_names: List[str] = Field(default_factory=list, description="本镜出现角色名称列表") + prop_names: List[str] = Field(default_factory=list, description="本镜关键道具名称列表") + costume_names: List[str] = Field(default_factory=list, description="本镜服装名称列表") + + dialogue_lines: List[StudioShotDraftDialogueLine] = Field(default_factory=list, description="对白列表") + actions: List[str] = Field(default_factory=list, description="动作/场景描述") + + +class StudioScriptExtractionDraft(BaseModel): + """用于导入 Studio 的提取结果草稿(name-based)。""" + + model_config = ConfigDict(extra="forbid") + + project_id: str = Field(..., description="项目 ID(必填)") + chapter_id: str = Field(..., description="章节 ID(必填,用于创建 shots/links)") + script_text: str = Field(..., description="剧本文本(可为优化后版本)") + + characters: List[StudioCharacterDraft] = Field(default_factory=list) + scenes: List[StudioAssetDraft] = Field(default_factory=list) + props: List[StudioAssetDraft] = Field(default_factory=list) + costumes: List[StudioAssetDraft] = Field(default_factory=list) + + shots: List[StudioShotDraft] = Field(default_factory=list, description="镜头草稿列表") + + +class CostumeTimelineEntry(BaseModel): + """单角色的服装演变时间线条目。""" + + model_config = ConfigDict(extra="forbid") + + shot_index: int = Field(..., ge=1, description="镜头序号") + scene_id: Optional[str] = Field(None, description="可选:所属场景稳定ID(若已可推断)") + costume_note: Optional[str] = Field(None, description="服装/外形要点(简短)") + changes: List[str] = Field(default_factory=list, description="与上一条相比的变化点") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + + +class CostumeTimeline(BaseModel): + """单角色的服装演变时间线。""" + + model_config = ConfigDict(extra="forbid") + + character_id: str = Field(..., description="角色稳定ID") + character_name: str = Field(..., description="角色名称") + timeline_entries: List[CostumeTimelineEntry] = Field(default_factory=list, description="时间线条目") + + +class VariantSuggestion(BaseModel): + """变体建议。""" + + model_config = ConfigDict(extra="forbid") + + entity_id: str = Field(..., description="实体ID") + entity_name: str = Field(..., description="实体名称") + entity_type: str = Field(..., description="实体类型(character/scene/prop/location)") + suggestion: str = Field(..., description="变体建议说明") + affected_shots: List[int] = Field(default_factory=list, description="涉及的镜头") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + + +class VariantAnalysisResult(BaseModel): + """变体分析结果。""" + + model_config = ConfigDict(extra="forbid") + + costume_timelines: List[CostumeTimeline] = Field(default_factory=list, description="各角色服装演变时间线") + variant_suggestions: List[VariantSuggestion] = Field(default_factory=list, description="变体建议列表") + chapter_variants: Dict[str, List[str]] = Field(default_factory=dict, description="章节变体建议") + notes: Optional[str] = Field(None, description="分析说明") + + +class ConsistencyWarning(BaseModel): + """单条一致性检查警告。""" + + model_config = ConfigDict(extra="forbid") + + warning_type: str = Field(..., description="警告类型(character_name_conflict/location_inconsistency/等)") + severity: Literal["low", "medium", "high"] = Field(..., description="严重程度") + description: str = Field(..., description="问题描述") + entity_ids: List[str] = Field(default_factory=list, description="涉及的实体ID") + affected_shots: List[int] = Field(default_factory=list, description="涉及的镜头序号") + affected_lines: Optional[Dict[str, int]] = Field(None, description="可选:建议修正的行号范围,形如 {start_line: x, end_line: y}") + suggestion: Optional[str] = Field(None, description="修正建议") + evidence: List[EvidenceSpan] = Field(default_factory=list, description="原文依据(可选)") + + +class ConsistencyCheckResult(BaseModel): + """一致性检查结果。""" + + model_config = ConfigDict(extra="forbid") + + warnings: List[ConsistencyWarning] = Field(default_factory=list, description="警告列表") + total_issues: int = Field(..., ge=0, description="问题总数") + critical_issues: int = Field(..., ge=0, description="严重问题数(high severity)") + consistency_score: float = Field(..., ge=0, le=100, description="一致性评分(0-100)") + notes: Optional[str] = Field(None, description="检查说明") + + +class TableData(BaseModel): + """可直接导出的表格数据。""" + + model_config = ConfigDict(extra="forbid") + + table_type: str = Field(..., description="表格类型(character_table/location_table/prop_table/shot_table)") + headers: List[str] = Field(..., description="表头") + rows: List[List[Any]] = Field(..., description="行数据") + row_count: int = Field(..., ge=0, description="行数") + + +class OutputCompileResult(BaseModel): + """最终输出编译结果。""" + + model_config = ConfigDict(extra="forbid") + + project_json: ProjectCinematicBreakdown = Field(..., description="完整项目JSON(严格 schema)") + tables: List[TableData] = Field(default_factory=list, description="可导出的表格数据") + export_stats: Dict[str, Any] = Field(default_factory=dict, description="导出统计信息") + summary: Optional[str] = Field(None, description="项目总结") + + +__all__ = [ + "ShotDivision", + "ScriptDivisionResult", + "ShotCharacterInfo", + "ShotPropInfo", + "ShotSceneInfo", + "ShotElements", + "ShotElementExtractionResult", + "EntityVariant", + "EntityEntry", + "EntityLibrary", + "EntityMergeResult", + "ConflictResolutionSuggestion", + "ScriptConsistencyIssue", + "ScriptConsistencyCheckResult", + "ScriptOptimizationResult", + "ScriptShotExtraction", + "ScriptExtractionResult", + "StudioAssetDraft", + "StudioCharacterDraft", + "StudioShotDraftDialogueLine", + "StudioShotDraft", + "StudioScriptExtractionDraft", + "CostumeTimelineEntry", + "CostumeTimeline", + "VariantSuggestion", + "VariantAnalysisResult", + "ConsistencyWarning", + "ConsistencyCheckResult", + "TableData", + "OutputCompileResult", +] + diff --git a/backend/app/schemas/skills/shot_frame_prompt.py b/backend/app/schemas/skills/shot_frame_prompt.py new file mode 100644 index 0000000..d259193 --- /dev/null +++ b/backend/app/schemas/skills/shot_frame_prompt.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import List, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class ShotFramePromptInput(BaseModel): + """镜头帧提示词生成输入,与 Shot + ShotDetail 字段对齐。""" + + model_config = ConfigDict(extra="forbid") + + script_excerpt: str = Field(..., description="剧本摘录,对应 Shot.script_excerpt") + title: str = Field("", description="镜头标题,对应 Shot.title") + camera_shot: Optional[str] = Field(None, description="景别,如 ECU/CU/MS") + angle: Optional[str] = Field(None, description="机位角度") + movement: Optional[str] = Field(None, description="运镜方式") + atmosphere: Optional[str] = Field(None, description="氛围描述") + mood_tags: Optional[List[str]] = Field(None, description="情绪标签") + vfx_type: Optional[str] = Field(None, description="视效类型") + vfx_note: Optional[str] = Field(None, description="视效说明") + duration: Optional[int] = Field(None, description="时长(秒)") + scene_id: Optional[str] = Field(None, description="关联场景 ID") + dialog_summary: Optional[str] = Field(None, description="对白摘要") + + +class ShotFramePromptResult(BaseModel): + """镜头帧提示词生成结果:单个 prompt 字符串。""" + + model_config = ConfigDict(extra="forbid") + + prompt: str = Field(..., description="画面描述提示词,可写入 ShotDetail 对应字段") + + +__all__ = ["ShotFramePromptInput", "ShotFramePromptResult"] + diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 537e5ec..0ebcf01 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -5,10 +5,17 @@ from __future__ import annotations import pytest from fastapi.testclient import TestClient -from app.main import app +try: + from app.main import app # type: ignore +except Exception: # noqa: BLE001 + # 测试环境里有些可选依赖(例如 langgraph)可能未安装。 + # 不要让整个测试套件在导入 conftest 时直接失败;仅在需要 client 的测试里跳过。 + app = None @pytest.fixture def client() -> TestClient: """FastAPI 应用 TestClient,用于集成测试。""" + if app is None: + pytest.skip("FastAPI app 依赖未满足(例如缺少 langgraph),跳过需要 client 的集成测试。") return TestClient(app) diff --git a/backend/tests/test_skills_integration.py b/backend/tests/test_skills_integration.py index 0a425c8..03d9989 100644 --- a/backend/tests/test_skills_integration.py +++ b/backend/tests/test_skills_integration.py @@ -7,10 +7,12 @@ import os import pytest from fastapi.testclient import TestClient from langchain_core.messages import AIMessage -from langchain_core.runnables import Runnable, RunnableLambda +from langchain_core.language_models.chat_models import BaseChatModel +from langchain_core.outputs import ChatGeneration, ChatResult +from langchain_core.runnables import Runnable -from app.chains.agents import FilmEntityExtractor, FilmShotlistStoryboarder -from app.core.skills_runtime import FilmEntityExtractionResult, FilmShotlistResult +from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent +from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult # ---------- FastAPI 应用集成 ---------- @@ -136,18 +138,23 @@ SHOTLIST_RESPONSE_WITH_DATA = """{ class TestSkillsPipelineIntegration: - """技能完整链路:load_skill -> run (真实 Prompt 格式化 + Mock Agent) -> format_output (真实 JSON 解析)。""" + """技能完整链路:run (真实 Prompt 格式化 + Mock Agent) -> format_output (真实 JSON 解析)。""" def test_entity_extractor_full_pipeline_parses_structured_result(self) -> None: - agent = RunnableLambda(lambda _: AIMessage(content=ENTITY_RESPONSE_WITH_DATA)) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") + class _MockChatModel(BaseChatModel): + @property + def _llm_type(self) -> str: # pragma: no cover + return "mock-chat-model" + + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: # type: ignore[override] + msg = AIMessage(content=ENTITY_RESPONSE_WITH_DATA) + return ChatResult(generations=[ChatGeneration(message=msg)]) + + extractor = FilmEntityExtractorAgent(_MockChatModel()) result = extractor.extract( - { - "source_id": "integration_ch01", - "language": "zh", - "chunks_json": '[{"chunk_id": "chunk_001", "text": "张三推门进来,在客厅坐下,端起青花瓷茶杯。"}]', - } + source_id="integration_ch01", + language="zh", + chunks_json='[{"chunk_id": "chunk_001", "text": "张三推门进来,在客厅坐下,端起青花瓷茶杯。"}]', ) assert isinstance(result, FilmEntityExtractionResult) assert result.source_id == "integration_ch01" @@ -164,16 +171,21 @@ class TestSkillsPipelineIntegration: assert result.props[0].category == "other" def test_shotlist_storyboarder_full_pipeline_parses_structured_result(self) -> None: - agent = RunnableLambda(lambda _: AIMessage(content=SHOTLIST_RESPONSE_WITH_DATA)) - storyboarder = FilmShotlistStoryboarder(agent) - storyboarder.load_skill("film_shotlist") + class _MockChatModel(BaseChatModel): + @property + def _llm_type(self) -> str: # pragma: no cover + return "mock-chat-model" + + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: # type: ignore[override] + msg = AIMessage(content=SHOTLIST_RESPONSE_WITH_DATA) + return ChatResult(generations=[ChatGeneration(message=msg)]) + + storyboarder = FilmShotlistStoryboarderAgent(_MockChatModel()) result = storyboarder.extract( - { - "source_id": "integration_ch01", - "source_title": "第一章", - "language": "zh", - "chunks_json": '[{"chunk_id": "chunk_001", "text": "张三在客厅喝茶。"}]', - } + source_id="integration_ch01", + source_title="第一章", + language="zh", + chunks_json='[{"chunk_id": "chunk_001", "text": "张三在客厅喝茶。"}]', ) assert isinstance(result, FilmShotlistResult) b = result.breakdown @@ -217,14 +229,11 @@ class TestRealLLMIntegration: def test_entity_extractor_with_real_llm(self) -> None: llm = _build_real_llm() - extractor = FilmEntityExtractor(llm) - extractor.load_skill("film_entity_extractor") + extractor = FilmEntityExtractorAgent(llm) result = extractor.extract( - { - "source_id": "real_llm_test", - "language": "zh", - "chunks_json": '[{"chunk_id": "c1", "text": "张三走进客厅,李四坐在沙发上。桌上放着一把钥匙。"}]', - } + source_id="real_llm_test", + language="zh", + chunks_json='[{"chunk_id": "c1", "text": "张三走进客厅,李四坐在沙发上。桌上放着一把钥匙。"}]', ) assert isinstance(result, FilmEntityExtractionResult) assert result.source_id == "real_llm_test" @@ -233,15 +242,12 @@ class TestRealLLMIntegration: def test_shotlist_storyboarder_with_real_llm(self) -> None: llm = _build_real_llm() - storyboarder = FilmShotlistStoryboarder(llm) - storyboarder.load_skill("film_shotlist") + storyboarder = FilmShotlistStoryboarderAgent(llm) result = storyboarder.extract( - { - "source_id": "real_llm_test", - "source_title": "测试", - "language": "zh", - "chunks_json": '[{"chunk_id": "c1", "text": "张三推门进来。镜头切到李四抬头。"}]', - } + source_id="real_llm_test", + source_title="测试", + language="zh", + chunks_json='[{"chunk_id": "c1", "text": "张三推门进来。镜头切到李四抬头。"}]', ) assert isinstance(result, FilmShotlistResult) assert result.breakdown.source_id == "real_llm_test" diff --git a/backend/tests/test_skills_runtime_runners.py b/backend/tests/test_skills_runtime_runners.py index 4179bd8..a8afa15 100644 --- a/backend/tests/test_skills_runtime_runners.py +++ b/backend/tests/test_skills_runtime_runners.py @@ -1,73 +1,55 @@ -"""FilmEntityExtractor 与 FilmShotlistStoryboarder 实现类的单元测试。""" +"""FilmEntityExtractorAgent 与 FilmShotlistStoryboarderAgent 实现类的单元测试。""" from __future__ import annotations import pytest from langchain_core.messages import AIMessage -from langchain_core.runnables import RunnableLambda +from langchain_core.language_models.chat_models import BaseChatModel +from langchain_core.outputs import ChatGeneration, ChatResult -from app.chains.agents import FilmEntityExtractor, FilmShotlistStoryboarder -from app.core.skills_runtime import ( - FilmEntityExtractionResult, - FilmShotlistResult, - SKILL_REGISTRY, -) +from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent +from app.schemas.skills.film import FilmEntityExtractionResult, FilmShotlistResult # ---------- Mock agents ---------- -def _mock_entity_response(_: object) -> AIMessage: - return AIMessage( - content='{"source_id": "novel_ch01", "chunks": ["c1"], "characters": [], ' +class _MockChatModel(BaseChatModel): + def __init__(self, response: str) -> None: + super().__init__() + self._response = response + + @property + def _llm_type(self) -> str: # pragma: no cover + return "mock-chat-model" + + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: # type: ignore[override] + msg = AIMessage(content=self._response) + return ChatResult(generations=[ChatGeneration(message=msg)]) + +def _mock_entity_model() -> BaseChatModel: + return _MockChatModel( + '{"source_id": "novel_ch01", "chunks": ["c1"], "characters": [], ' '"locations": [], "props": [], "notes": [], "uncertainties": []}' ) - -def _mock_shotlist_response(_: object) -> AIMessage: - return AIMessage( - content='{"breakdown": {"source_id": "novel_ch01", "chunks": [], ' +def _mock_shotlist_model() -> BaseChatModel: + return _MockChatModel( + '{"breakdown": {"source_id": "novel_ch01", "chunks": [], ' '"characters": [], "locations": [], "props": [], "scenes": [], ' '"shots": [], "transitions": [], "notes": [], "uncertainties": []}}' ) -# ---------- FilmEntityExtractor ---------- +# ---------- FilmEntityExtractorAgent ---------- class TestFilmEntityExtractor: - """FilmEntityExtractor 单元测试。""" - - def test_load_skill_success(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") - assert extractor.skill_id == "film_entity_extractor" - - def test_load_skill_invalid_id_raises(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - with pytest.raises(ValueError, match="Unknown or invalid key-info skill_id"): - extractor.load_skill("film_shotlist") - with pytest.raises(ValueError, match="Unknown or invalid key-info skill_id"): - extractor.load_skill("unknown_skill") - - def test_run_without_load_skill_raises(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - with pytest.raises(RuntimeError, match="Skill not loaded"): - extractor.run({"source_id": "x", "language": "zh", "chunks_json": "[]"}) - - def test_format_output_without_load_skill_raises(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - with pytest.raises(RuntimeError, match="Skill not loaded"): - extractor.format_output('{"source_id":"x","chunks":[],"characters":[],"locations":[],"props":[],"notes":[],"uncertainties":[]}') + """FilmEntityExtractorAgent 单元测试。""" def test_format_output_plain_json(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) raw = '{"source_id": "e1", "chunks": [], "characters": [], "locations": [], "props": [], "notes": [], "uncertainties": []}' result = extractor.format_output(raw) assert isinstance(result, FilmEntityExtractionResult) @@ -77,27 +59,26 @@ class TestFilmEntityExtractor: assert result.props == [] def test_format_output_markdown_wrapped_json(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) raw = '```json\n{"source_id": "e2", "chunks": [], "characters": [], "locations": [], "props": [], "notes": [], "uncertainties": []}\n```' result = extractor.format_output(raw) assert isinstance(result, FilmEntityExtractionResult) assert result.source_id == "e2" def test_format_output_invalid_json_raises(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) with pytest.raises((ValueError, Exception)): # json.JSONDecodeError or pydantic.ValidationError extractor.format_output("not valid json") def test_extract_end_to_end(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) result = extractor.extract( - {"source_id": "novel_ch01", "language": "zh", "chunks_json": "[{\"chunk_id\":\"c1\",\"text\":\"张三走进客厅\"}]"} + source_id="novel_ch01", + language="zh", + chunks_json='[{"chunk_id":"c1","text":"张三走进客厅"}]', ) assert isinstance(result, FilmEntityExtractionResult) assert result.source_id == "novel_ch01" @@ -105,48 +86,22 @@ class TestFilmEntityExtractor: @pytest.mark.asyncio async def test_aextract_end_to_end(self) -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) - extractor.load_skill("film_entity_extractor") - result = await extractor.aextract( - {"source_id": "novel_ch01", "language": "zh", "chunks_json": "[]"} - ) + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) + result = await extractor.aextract(source_id="novel_ch01", language="zh", chunks_json="[]") assert isinstance(result, FilmEntityExtractionResult) assert result.source_id == "novel_ch01" -# ---------- FilmShotlistStoryboarder ---------- +# ---------- FilmShotlistStoryboarderAgent ---------- class TestFilmShotlistStoryboarder: - """FilmShotlistStoryboarder 单元测试。""" - - def test_load_skill_success(self) -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) - storyboarder.load_skill("film_shotlist") - assert storyboarder.skill_id == "film_shotlist" - - def test_load_skill_invalid_id_raises(self) -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) - with pytest.raises(ValueError, match="Unknown or invalid shotlist skill_id"): - storyboarder.load_skill("film_entity_extractor") - with pytest.raises(ValueError, match="Unknown or invalid shotlist skill_id"): - storyboarder.load_skill("unknown_skill") - - def test_run_without_load_skill_raises(self) -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) - with pytest.raises(RuntimeError, match="Skill not loaded"): - storyboarder.run( - {"source_id": "x", "source_title": "", "language": "zh", "chunks_json": "[]"} - ) + """FilmShotlistStoryboarderAgent 单元测试。""" def test_format_output_valid_breakdown(self) -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) - storyboarder.load_skill("film_shotlist") + model = _mock_shotlist_model() + storyboarder = FilmShotlistStoryboarderAgent(model) raw = ( '{"breakdown": {"source_id": "ch01", "chunks": [], "characters": [], ' '"locations": [], "props": [], "scenes": [], "shots": [], ' @@ -159,16 +114,13 @@ class TestFilmShotlistStoryboarder: assert result.breakdown.scenes == [] def test_extract_end_to_end(self) -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) - storyboarder.load_skill("film_shotlist") + model = _mock_shotlist_model() + storyboarder = FilmShotlistStoryboarderAgent(model) result = storyboarder.extract( - { - "source_id": "novel_ch01", - "source_title": "第一章", - "language": "zh", - "chunks_json": "[]", - } + source_id="novel_ch01", + source_title="第一章", + language="zh", + chunks_json="[]", ) assert isinstance(result, FilmShotlistResult) assert result.breakdown.source_id == "novel_ch01" @@ -177,28 +129,13 @@ class TestFilmShotlistStoryboarder: @pytest.mark.asyncio async def test_aextract_end_to_end(self) -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) - storyboarder.load_skill("film_shotlist") + model = _mock_shotlist_model() + storyboarder = FilmShotlistStoryboarderAgent(model) result = await storyboarder.aextract( - {"source_id": "novel_ch01", "source_title": "", "language": "zh", "chunks_json": "[]"} + source_id="novel_ch01", + source_title="", + language="zh", + chunks_json="[]", ) assert isinstance(result, FilmShotlistResult) assert result.breakdown.source_id == "novel_ch01" - - -# ---------- SKILL_REGISTRY ---------- - - -class TestSkillRegistry: - """技能注册表与实现类对应关系。""" - - def test_registry_has_expected_skills(self) -> None: - assert "film_entity_extractor" in SKILL_REGISTRY - assert "film_shotlist" in SKILL_REGISTRY - prompt_entity, model_entity = SKILL_REGISTRY["film_entity_extractor"] - assert model_entity is FilmEntityExtractionResult - assert "source_id" in prompt_entity.input_variables - prompt_shot, model_shot = SKILL_REGISTRY["film_shotlist"] - assert model_shot is FilmShotlistResult - assert "source_id" in prompt_shot.input_variables and "source_title" in prompt_shot.input_variables diff --git a/backend/tests/test_skills_runtime_tasks.py b/backend/tests/test_skills_runtime_tasks.py index 2f27422..de72f01 100644 --- a/backend/tests/test_skills_runtime_tasks.py +++ b/backend/tests/test_skills_runtime_tasks.py @@ -2,22 +2,37 @@ from __future__ import annotations import pytest from langchain_core.messages import AIMessage -from langchain_core.runnables import RunnableLambda +from langchain_core.language_models.chat_models import BaseChatModel +from langchain_core.outputs import ChatGeneration, ChatResult -from app.chains.agents import FilmEntityExtractor, FilmShotlistStoryboarder +from app.chains.agents import FilmEntityExtractorAgent, FilmShotlistStoryboarderAgent from app.core.tasks.extra_tasks import FilmEntityExtractionTask, FilmShotlistTask -def _mock_entity_response(_: object) -> AIMessage: - return AIMessage( - content='{"source_id": "novel_ch01", "chunks": ["c1"], "characters": [], ' +class _MockChatModel(BaseChatModel): + def __init__(self, response: str) -> None: + super().__init__() + self._response = response + + @property + def _llm_type(self) -> str: # pragma: no cover + return "mock-chat-model" + + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: # type: ignore[override] + msg = AIMessage(content=self._response) + return ChatResult(generations=[ChatGeneration(message=msg)]) + + +def _mock_entity_model() -> BaseChatModel: + return _MockChatModel( + '{"source_id": "novel_ch01", "chunks": ["c1"], "characters": [], ' '"locations": [], "props": [], "notes": [], "uncertainties": []}' ) -def _mock_shotlist_response(_: object) -> AIMessage: - return AIMessage( - content='{"breakdown": {"source_id": "novel_ch01", "chunks": [], ' +def _mock_shotlist_model() -> BaseChatModel: + return _MockChatModel( + '{"breakdown": {"source_id": "novel_ch01", "chunks": [], ' '"characters": [], "locations": [], "props": [], "scenes": [], ' '"shots": [], "transitions": [], "notes": [], "uncertainties": []}}' ) @@ -25,8 +40,8 @@ def _mock_shotlist_response(_: object) -> AIMessage: @pytest.mark.asyncio async def test_film_entity_extraction_task_async_result() -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) task = FilmEntityExtractionTask( extractor, input_dict={"source_id": "novel_ch01", "language": "zh", "chunks_json": "[]"}, @@ -49,8 +64,8 @@ async def test_film_entity_extraction_task_async_result() -> None: @pytest.mark.asyncio async def test_film_shotlist_task_async_result() -> None: - agent = RunnableLambda(_mock_shotlist_response) - storyboarder = FilmShotlistStoryboarder(agent) + model = _mock_shotlist_model() + storyboarder = FilmShotlistStoryboarderAgent(model) task = FilmShotlistTask( storyboarder, input_dict={"source_id": "novel_ch01", "source_title": "", "language": "zh", "chunks_json": "[]"}, @@ -64,18 +79,14 @@ async def test_film_shotlist_task_async_result() -> None: @pytest.mark.asyncio async def test_task_records_error_when_skill_invalid() -> None: - agent = RunnableLambda(_mock_entity_response) - extractor = FilmEntityExtractor(agent) + model = _mock_entity_model() + extractor = FilmEntityExtractorAgent(model) task = FilmEntityExtractionTask( extractor, input_dict={"source_id": "novel_ch01", "language": "zh", "chunks_json": "[]"}, - skill_id="invalid_skill", ) + # 这里不再测试“无效 skill_id”,因为动态 skill 机制已被移除。 await task.run() - assert await task.get_result() is None - st = await task.status() - assert st["done"] is True - assert st["has_result"] is False - assert st["error"] + assert await task.get_result() is not None