# 代码材料（前30页）
软件名称：StudioAgent AI视频制片平台软件
版本号：V0.1.0
## 第 1 页
```text
// File: frontend/src/app/layout.tsx
import type { Metadata } from "next";
import { Providers } from "./providers";
import "./globals.css";
export const metadata: Metadata = {
title: "StudioAgent — AI 制片平台",
description: "多 Agent 协作的 AI 制片平台",
};
export default function RootLayout({
children,
}: {
children: React.ReactNode;
}) {
return (
{children}
);
}
// File: frontend/src/app/page.tsx
export default function Home() {
return (
StudioAgent
多 Agent 协作的 AI 制片平台
);
}
// File: backend/app/agents/compaction.py
"""Context compaction service (inspired by pi-mono session.compact())."""
async def compact_conversation(conversation_id: str) -> None:
"""Compress conversation history.
Strategy:
1. Keep last 10 messages intact
2. Summarize older messages into a single system message via LLM
```
## 第 2 页
```text
3. Preserve all interrupt/confirm checkpoint decisions
4. Preserve tool_call result summaries (not full params)
"""
# TODO: implement
pass
// File: backend/app/agents/graph.py
"""LangGraph Swarm graph definition — 6-agent production crew."""
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph.state import CompiledStateGraph
from langgraph_swarm import create_handoff_tool, create_swarm
from langchain.agents import create_agent
def _load_prompt(agent_name: str) -> str:
"""Load system prompt from markdown file."""
from pathlib import Path
prompt_path = Path(__file__).parent.parent / "llm" / "prompts" / f"{agent_name}.md"
if prompt_path.exists():
return prompt_path.read_text(encoding="utf-8")
return f"You are the {agent_name} agent of the StudioAgent production crew."
def _build_swarm():
"""Build the swarm graph (deferred to avoid import-time LLM initialization)."""
from app.agents.tools.producer_tools import (
plan_production,
estimate_cost,
query_project_status,
)
from app.agents.tools.screenwriter_tools import (
analyze_text,
extract_characters,
extract_locations,
generate_script,
edit_script,
)
from app.agents.tools.director_tools import (
generate_storyboard,
plan_shots,
review_continuity,
approve_visual,
)
from app.agents.tools.camera_tools import (
generate_character_image,
generate_location_image,
generate_panel_image,
modify_image,
)
from app.agents.tools.editor_tools import (
generate_video,
extend_video,
compose_timeline,
)
from app.agents.tools.sound_tools import (
analyze_dialogue,
generate_voice,
design_sfx,
```
## 第 3 页
```text
match_bgm,
)
from app.llm import get_llm
from app.config import settings
# ═══════════ Handoff Tools ═══════════
handoff_to_producer = create_handoff_tool(
agent_name="producer",
description="将控制权交还给制片人，用于汇报工作结果或请求下一步指示",
)
handoff_to_screenwriter = create_handoff_tool(
agent_name="screenwriter",
description="将任务交给编剧，用于文本分析、角色提取、剧本生成",
)
handoff_to_director = create_handoff_tool(
agent_name="director",
description="将任务交给导演，用于分镜生成、镜头规划、视觉一致性审核",
)
handoff_to_camera = create_handoff_tool(
agent_name="camera",
description="将任务交给摄影，用于角色形象/场景图/分镜画面生成",
)
handoff_to_editor = create_handoff_tool(
agent_name="editor",
description="将任务交给剪辑，用于视频生成和时间轴编排",
)
handoff_to_sound = create_handoff_tool(
agent_name="sound",
description="将任务交给音效，用于配音生成和音效设计",
)
# ═══════════ Agent Definitions ═══════════
producer = create_agent(
model=get_llm(settings.producer_model),
tools=[
plan_production,
estimate_cost,
query_project_status,
handoff_to_screenwriter,
handoff_to_director,
handoff_to_camera,
handoff_to_editor,
handoff_to_sound,
],
system_prompt=_load_prompt("producer"),
name="producer",
)
screenwriter = create_agent(
model=get_llm(settings.screenwriter_model),
tools=[
analyze_text,
extract_characters,
extract_locations,
generate_script,
edit_script,
handoff_to_producer,
],
```
## 第 4 页
```text
system_prompt=_load_prompt("screenwriter"),
name="screenwriter",
)
director = create_agent(
model=get_llm(settings.director_model),
tools=[
generate_storyboard,
plan_shots,
review_continuity,
approve_visual,
handoff_to_producer,
],
system_prompt=_load_prompt("director"),
name="director",
)
camera = create_agent(
model=get_llm(settings.camera_model),
tools=[
generate_character_image,
generate_location_image,
generate_panel_image,
modify_image,
handoff_to_producer,
],
system_prompt=_load_prompt("camera"),
name="camera",
)
editor = create_agent(
model=get_llm(settings.editor_model),
tools=[
generate_video,
extend_video,
compose_timeline,
handoff_to_producer,
],
system_prompt=_load_prompt("editor"),
name="editor",
)
sound = create_agent(
model=get_llm(settings.sound_model),
tools=[
analyze_dialogue,
generate_voice,
design_sfx,
match_bgm,
handoff_to_producer,
],
system_prompt=_load_prompt("sound"),
name="sound",
)
# ═══════════ Build Swarm ═══════════
return create_swarm(
agents=[producer, screenwriter, director, camera, editor, sound],
default_active_agent="producer",
```
## 第 5 页
```text
)
async def build_graph(db_url: str):
"""Compile the swarm graph with PostgreSQL checkpointing.
Returns (compiled_graph, checkpointer_context) — caller must keep
the context alive for the lifetime of the app.
"""
swarm_graph = _build_swarm()
checkpointer_ctx = AsyncPostgresSaver.from_conn_string(db_url)
checkpointer = await checkpointer_ctx.__aenter__()
await checkpointer.setup()
return swarm_graph.compile(checkpointer=checkpointer), checkpointer_ctx
// File: backend/app/agents/session.py
"""Conversation session management (inspired by pi-mono AgentSession)."""
from app.agents.state import ProductionState
class ConversationSession:
"""Manages a conversation session's lifecycle.
Provides equivalents to pi-mono's AgentSession interface:
- send (prompt) → POST /conversations/{id}/messages
- confirm (resume) → POST /conversations/{id}/resume
- steer → POST /conversations/{id}/steer
- fork → POST /conversations/{id}/fork
- compact → POST /conversations/{id}/compact
- abort → POST /conversations/{id}/abort
"""
def __init__(self, conversation_id: str, project_id: str):
self.conversation_id = conversation_id
self.project_id = project_id
async def get_state(self) -> ProductionState | None:
"""Get current production state from LangGraph checkpoint."""
# TODO: implement
return None
// File: backend/app/agents/sse_transformer.py
"""LangGraph → SSE bridge — transforms graph stream events to SSE protocol."""
import json
import uuid
from collections.abc import AsyncGenerator
from langchain_core.messages import AIMessageChunk, ToolMessage
from langgraph.graph.state import CompiledStateGraph
def sse_event(event_type: str, data: dict) -> str:
"""Format a single SSE event."""
return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
async def stream_graph_to_sse(
graph: CompiledStateGraph,
```
## 第 6 页
```text
input_state: dict,
config: dict,
) -> AsyncGenerator[str, None]:
"""Stream LangGraph execution as SSE events.
Maps LangGraph stream events to the SSE protocol:
- AIMessageChunk.content → agent.text_start / text_delta / text_end
- AIMessageChunk.tool_call_chunks → agent.tool_call
- ToolMessage → agent.tool_result
- Exception → error
- Normal end → done
"""
message_id = str(uuid.uuid4())
current_agent = "producer"
text_started = False
full_response = ""
try:
async for event in graph.astream(input_state, config, stream_mode="messages"):
# stream_mode="messages" yields (message_chunk, metadata) tuples
if not isinstance(event, tuple) or len(event) != 2:
continue
chunk, metadata = event
# Track current agent from metadata
agent_name = metadata.get("langgraph_node", current_agent)
if agent_name != current_agent:
# Agent handoff
yield sse_event("agent.handoff", {
"from": current_agent,
"to": agent_name,
})
current_agent = agent_name
# Reset text state for new agent
if text_started:
yield sse_event("agent.text_end", {
"agent": current_agent,
"message_id": message_id,
})
text_started = False
message_id = str(uuid.uuid4())
if isinstance(chunk, AIMessageChunk):
# Text content
if chunk.content:
content = chunk.content if isinstance(chunk.content, str) else str(chunk.content)
if content:
if not text_started:
yield sse_event("agent.text_start", {
"agent": current_agent,
"message_id": message_id,
})
text_started = True
yield sse_event("agent.text_delta", {
"agent": current_agent,
"content": content,
})
full_response += content
```
## 第 7 页
```text
# Tool calls
if chunk.tool_call_chunks:
for tc in chunk.tool_call_chunks:
if tc.get("name"):
yield sse_event("agent.tool_call", {
"agent": current_agent,
"tool": tc.get("name"),
"args": tc.get("args", ""),
"id": tc.get("id", ""),
})
elif isinstance(chunk, ToolMessage):
yield sse_event("agent.tool_result", {
"agent": current_agent,
"tool_call_id": chunk.tool_call_id,
"content": chunk.content if isinstance(chunk.content, str) else json.dumps(chunk.content, ensure_ascii=False),
})
# End text stream if still open
if text_started:
yield sse_event("agent.text_end", {
"agent": current_agent,
"message_id": message_id,
})
yield sse_event("done", {"reason": "complete", "full_response": full_response})
except Exception as e:
if text_started:
yield sse_event("agent.text_end", {
"agent": current_agent,
"message_id": message_id,
})
yield sse_event("error", {"message": str(e)})
// File: backend/app/agents/state.py
"""Production State — LangGraph state schema for the Swarm."""
from __future__ import annotations
from typing import Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from pydantic import BaseModel
from typing_extensions import TypedDict
class ProductionPlan(BaseModel):
"""A production plan created by Producer."""
title: str
description: str
phases: list[str]
estimated_cost: float
estimated_duration: str
class CharacterRef(BaseModel):
"""Character reference for production state."""
```
## 第 8 页
```text
id: str
name: str
description: str | None = None
appearance_url: str | None = None
class LocationRef(BaseModel):
"""Location reference for production state."""
id: str
name: str
description: str | None = None
image_url: str | None = None
class ScriptData(BaseModel):
"""Structured script data."""
episode_index: int = 0
title: str | None = None
scenes: list[dict] = []
raw_text: str | None = None
class StoryboardData(BaseModel):
"""Structured storyboard data."""
panels: list[dict] = []
class ProductionState(TypedDict):
"""Global state for the LangGraph Swarm.
This state is shared across all agents and persisted via checkpointing.
Producer uses `completed_steps` and `available_assets` for dynamic state
awareness instead of a fixed `current_phase` enum.
"""
messages: Annotated[list[BaseMessage], add_messages]
project_id: str
plan: ProductionPlan | None
plan_approved: bool
characters: list[CharacterRef]
locations: list[LocationRef]
script: ScriptData | None
storyboard: StoryboardData | None
# ── Dynamic state (replaces fixed current_phase) ──
completed_steps: list[str] # e.g. ["characters_extracted", "script_generated"]
available_assets: dict # e.g. {"characters": 5, "locations": 3, "panels": 0}
# ── User interaction preference ──
interaction_mode: str # "collaborative" | "autonomous" | "supervised"
# ── Agent runtime ──
awaiting_confirmation: str | None
active_agent: str # tracked by LangGraph Swarm
// File: backend/app/agents/tools/base.py
```
## 第 9 页
```text
"""Tool definition base classes."""
from pydantic import BaseModel
from typing import Any, Callable, Awaitable
class ToolDefinition(BaseModel):
"""Tool definition (inspired by pi-mono Tool interface).
Each tool uses a Pydantic Model for parameters, auto-generating
JSON Schema for the LLM.
"""
name: str
label: str
description: str
parameters: type[BaseModel]
execute: Callable[..., Awaitable[Any]]
model_config = {"arbitrary_types_allowed": True}
class ToolContext(BaseModel):
"""Tool execution context."""
project_id: str
user_id: str
conversation_id: str
stream_writer: Any = None # LangGraph get_stream_writer() reference
// File: backend/app/agents/tools/camera_tools.py
"""Camera Agent tools."""
from langchain_core.tools import tool
@tool
def generate_character_image(
character_id: str,
prompt: str,
style: str = "anime",
model: str = "seedream-3.0",
count: int = 4,
width: int = 1024,
height: int = 1024,
) -> dict:
"""生成角色形象候选图。为指定角色生成多张外貌参考图供选择。
Args:
character_id: 角色ID
prompt: 角色外貌描述 prompt
style: 画风
model: 图片生成模型
count: 生成数量
width: 图片宽度
height: 图片高度
"""
return {"candidates": [], "task_ids": []}
```
## 第 10 页
```text
@tool
def generate_location_image(
location_id: str,
prompt: str,
style: str = "anime",
model: str = "seedream-3.0",
count: int = 4,
width: int = 1280,
height: int = 720,
) -> dict:
"""生成场景图候选。为指定场景生成多张场景图供选择。
Args:
location_id: 场景ID
prompt: 场景描述 prompt
style: 画风
model: 图片生成模型
count: 生成数量
width: 图片宽度
height: 图片高度
"""
return {"candidates": [], "task_ids": []}
@tool
def generate_panel_image(
panel_id: str,
prompt: str,
character_refs: list[str] | None = None,
location_ref: str | None = None,
style: str = "anime",
model: str = "seedream-3.0",
width: int = 1280,
height: int = 720,
) -> dict:
"""生成分镜画面候选图。根据角色和场景参考生成分镜画面。
Args:
panel_id: 分镜ID
prompt: 画面描述 prompt
character_refs: 参考角色形象 URL 列表
location_ref: 参考场景图 URL
style: 画风
model: 图片生成模型
width: 图片宽度
height: 图片高度
"""
return {"candidates": [], "task_ids": []}
@tool
def modify_image(prompt: str, image_url: str, mask_url: str | None = None) -> dict:
"""修改已有图片。支持局部编辑和整体调整。
Args:
prompt: 修改描述
image_url: 原图URL
mask_url: 蒙版URL（局部编辑时使用）
"""
return {"image_url": ""}
```
## 第 11 页
```text
// File: backend/app/agents/tools/candidate.py
"""Candidate card-draw tool — interrupt + confirm for AI-generated candidates."""
from pydantic import BaseModel
class CandidateState(BaseModel):
"""Universal candidate state (character appearance / location image / panel)."""
entity_type: str # "character_appearance" | "location_image" | "panel"
entity_id: str
original_url: str | None = None
candidates: list[str] = [] # URLs, may contain "PENDING:{task_id}" placeholders
selected_index: int = -1 # -1 = original, 0~N = candidate
previous_url: str | None = None
// File: backend/app/agents/tools/confirm.py
"""Agent confirmation tool — Human-in-the-Loop via LangGraph interrupt()."""
from langgraph.types import interrupt
from pydantic import BaseModel
class ConfirmationResult(BaseModel):
action: str # "confirm" | "modify" | "redo" | "skip"
feedback: str | None = None
def request_user_confirmation(
confirmation_type: str,
summary: str,
data: dict,
urgency: str = "normal",
) -> dict:
"""Agent-driven confirmation tool.
Unlike traditional workflow checkpoints, this is called by Producer
when it autonomously decides user input is needed — guided by
system prompt strategy, not hardcoded if-else.
LangGraph mechanism:
- interrupt() pauses current node
- Frontend receives awaiting_confirmation event
- User chooses confirm/modify/redo/skip
- Frontend calls POST /conversations/{id}/resume with Command(resume=...)
- Graph resumes from pause point
"""
result = interrupt({
"type": confirmation_type,
"summary": summary,
"data": data,
"urgency": urgency,
"options": ["confirm", "modify", "redo", "skip"],
})
return result
// File: backend/app/agents/tools/director_tools.py
"""Director Agent tools."""
```
## 第 12 页
```text
from langchain_core.tools import tool
@tool
def generate_storyboard(script_id: str, panel_count: int = 10) -> dict:
"""根据剧本生成分镜脚本，规划每个镜头的画面描述。
Args:
script_id: 剧本ID
panel_count: 分镜数量
"""
return {"storyboard_id": "placeholder", "panels": []}
@tool
def plan_shots(panel_id: str, description: str) -> dict:
"""为单个分镜规划镜头构图、运镜方式。
Args:
panel_id: 分镜ID
description: 画面描述
"""
return {"shot_plan": {}}
@tool
def review_continuity(
episode_id: str,
check_characters: bool = True,
check_locations: bool = True,
) -> dict:
"""审核视觉连续性，检查角色形象和场景的一致性。
Args:
episode_id: 集ID
check_characters: 是否检查角色一致性
check_locations: 是否检查场景一致性
"""
return {"issues": [], "approved": True}
@tool
def approve_visual(panel_id: str, image_url: str) -> dict:
"""审批视觉输出，确认画面质量。
Args:
panel_id: 分镜ID
image_url: 待审批的图片URL
"""
return {"approved": True}
// File: backend/app/agents/tools/editor_tools.py
"""Editor Agent tools."""
from langchain_core.tools import tool
@tool
def generate_video(
panel_id: str,
```
## 第 13 页
```text
image_url: str,
prompt: str,
camera_move: str | None = None,
duration: float = 5.0,
model: str = "seedance-1.5",
) -> dict:
"""从图片生成视频。使用首帧图片和运动描述生成视频片段。
Args:
panel_id: 分镜ID
image_url: 首帧图片 URL
prompt: 运动描述 prompt
camera_move: 运镜指令
duration: 时长（秒）
model: 视频生成模型
"""
return {"video_url": "", "task_id": ""}
@tool
def extend_video(video_url: str, duration: float = 5.0) -> dict:
"""延长视频时长。
Args:
video_url: 原视频URL
duration: 延长时长（秒）
"""
return {"video_url": ""}
@tool
def compose_timeline(
episode_id: str,
panel_ids: list[str],
transitions: list[str] | None = None,
) -> dict:
"""编排视频时间轴。将多个分镜视频按顺序组合，添加转场效果。
Args:
episode_id: 集ID
panel_ids: 按顺序排列的分镜 ID
transitions: 转场效果列表
"""
return {"timeline_url": ""}
// File: backend/app/agents/tools/global_asset_picker.py
"""Global asset picker tool."""
async def pick_global_asset(asset_type: str, user_id: str) -> dict | None:
"""Pick an asset from the global Asset Hub.
Triggered when user says "use my character from asset library".
Frontend renders a GlobalAssetPicker modal for user selection.
"""
# TODO: integrate with interrupt() for frontend modal
return None
// File: backend/app/agents/tools/producer_tools.py
"""Producer Agent tools."""
```
## 第 14 页
```text
from langchain_core.tools import tool
@tool
def plan_production(
title: str,
description: str,
style: str = "anime",
episode_count: int = 1,
panel_count_per_episode: int = 10,
) -> dict:
"""制定制作方案。根据用户需求分析项目规模，规划制作阶段和资源分配。
Args:
title: 项目标题
description: 用户原始需求描述
style: 画风偏好：写实/动漫/3D/水墨
episode_count: 集数
panel_count_per_episode: 每集分镜数
"""
total_panels = episode_count * panel_count_per_episode
phases = []
if total_panels > 0:
phases.append("文本分析与角色提取")
phases.append("场景设定与角色形象设计")
phases.append("剧本生成与分镜规划")
phases.append("画面生成")
if total_panels > 5:
phases.append("视频合成与音效制作")
return {
"title": title,
"description": description,
"style": style,
"episode_count": episode_count,
"panel_count_per_episode": panel_count_per_episode,
"total_panels": total_panels,
"phases": phases,
"estimated_cost": round(total_panels * 0.5, 2),
"estimated_duration": f"{max(1, total_panels // 5)} 分钟",
}
@tool
def estimate_cost(
image_count: int,
video_count: int,
voice_count: int,
image_model: str = "seedream-3.0",
video_model: str = "seedance-1.5",
voice_model: str = "qwen-tts",
) -> dict:
"""估算制作成本。根据各类素材数量和选用模型计算预估费用。
Args:
image_count: 图片数量
video_count: 视频数量
voice_count: 配音数量
image_model: 图片模型
```
## 第 15 页
```text
video_model: 视频模型
voice_model: 配音模型
"""
pricing = {
"seedream-3.0": 0.2,
"imagen-3": 0.4,
"flux-1.1": 0.3,
"seedance-1.5": 2.0,
"veo-2": 3.0,
"kling-1.5": 2.5,
"qwen-tts": 0.05,
"elevenlabs": 0.15,
}
image_cost = image_count * pricing.get(image_model, 0.3)
video_cost = video_count * pricing.get(video_model, 2.0)
voice_cost = voice_count * pricing.get(voice_model, 0.1)
return {
"total_cost": round(image_cost + video_cost + voice_cost, 2),
"breakdown": {
"image": {"count": image_count, "model": image_model, "cost": round(image_cost, 2)},
"video": {"count": video_count, "model": video_model, "cost": round(video_cost, 2)},
"voice": {"count": voice_count, "model": voice_model, "cost": round(voice_cost, 2)},
},
}
@tool
def query_project_status(project_id: str, include_details: bool = False) -> dict:
"""查询项目制作进度。返回当前已完成的步骤和可用资产统计。
Args:
project_id: 项目ID
include_details: 是否包含详细信息
"""
# TODO: implement with database query
return {"project_id": project_id, "status": "active", "completed_steps": [], "available_assets": {}}
// File: backend/app/agents/tools/reference_collector.py
"""Reference image auto-collector for panel generation."""
from pydantic import BaseModel
class CharacterRef(BaseModel):
character_id: str
image_url: str
description: str | None = None
class PanelReferences(BaseModel):
"""References collected for panel image generation."""
sketch: str | None = None
character_refs: list[CharacterRef] = []
location_ref: str | None = None
async def collect_panel_references(panel_id: str, project_id: str) -> PanelReferences:
"""Collect reference images for panel generation.
```
## 第 16 页
```text
Priority:
1. Sketch reference (user-uploaded hand-drawn sketch)
2. Character appearance reference (confirmed selectedIndex image)
3. Location reference (confirmed isSelected location image)
"""
# TODO: implement with database queries
return PanelReferences()
// File: backend/app/agents/tools/screenwriter_tools.py
"""Screenwriter Agent tools."""
from langchain_core.tools import tool
@tool
def analyze_text(text: str, analysis_type: str = "full") -> dict:
"""分析文本内容（小说/故事大纲/用户描述），提取关键信息。
Args:
text: 待分析的文本内容
analysis_type: 分析类型：full/characters/locations/plot
"""
return {"analysis": {}, "analysis_type": analysis_type, "text_length": len(text)}
@tool
def extract_characters(text: str, max_characters: int = 10) -> dict:
"""从文本中提取角色信息，包括名称、外貌、性格等。
Args:
text: 待分析文本
max_characters: 最大角色数
"""
return {"characters": []}
@tool
def extract_locations(text: str, max_locations: int = 10) -> dict:
"""从文本中提取场景/地点信息。
Args:
text: 待分析文本
max_locations: 最大场景数
"""
return {"locations": []}
@tool
def generate_script(
plot_summary: str,
characters: list[dict] | None = None,
locations: list[dict] | None = None,
episode_index: int = 0,
style: str = "drama",
panel_count: int = 10,
) -> dict:
"""生成剧本。根据角色、场景和剧情生成结构化剧本。
Args:
plot_summary: 剧情概要
```
## 第 17 页
```text
characters: 角色列表
locations: 场景列表
episode_index: 集数索引
style: 风格
panel_count: 分镜数量
"""
return {"script_id": "placeholder", "script": {}}
@tool
def edit_script(script_id: str, modifications: str) -> dict:
"""修改剧本。根据用户反馈调整剧本内容。
Args:
script_id: 剧本ID
modifications: 用户要求的修改内容
"""
return {"script_id": script_id, "updated": True}
// File: backend/app/agents/tools/sound_tools.py
"""Sound Agent tools."""
from langchain_core.tools import tool
@tool
def analyze_dialogue(script_id: str, episode_index: int = 0) -> dict:
"""分析剧本中的对白，提取需要配音的文本段落。
Args:
script_id: 剧本ID
episode_index: 集索引
"""
return {"dialogues": []}
@tool
def generate_voice(
panel_id: str,
text: str,
character_id: str,
voice_id: str | None = None,
emotion: str = "neutral",
model: str = "qwen-tts",
) -> dict:
"""生成角色配音。根据角色音色和情感生成对白语音。
Args:
panel_id: 分镜ID
text: 对白文本
character_id: 角色 ID，用于匹配音色
voice_id: 指定音色 ID
emotion: 情感
model: 配音模型
"""
return {"voice_url": "", "task_id": ""}
@tool
def design_sfx(panel_id: str, scene_description: str, mood: str = "neutral") -> dict:
```
## 第 18 页
```text
"""设计音效。根据场景描述和氛围生成环境音效。
Args:
panel_id: 分镜ID
scene_description: 场景描述
mood: 氛围
"""
return {"sfx_url": ""}
@tool
def match_bgm(episode_id: str, mood: str = "neutral", duration: float = 60.0) -> dict:
"""匹配背景音乐。根据集的整体氛围推荐/生成背景音乐。
Args:
episode_id: 集ID
mood: 氛围
duration: 时长（秒）
"""
return {"bgm_url": ""}
// File: backend/app/api/asset_hub.py
"""Global Asset Hub API routes."""
from fastapi import APIRouter
router = APIRouter()
# ── Folders ──
@router.get("/folders")
async def list_folders():
"""List asset folders."""
...
@router.post("/folders")
async def create_folder():
"""Create a folder."""
...
@router.patch("/folders/{folder_id}")
async def rename_folder(folder_id: str):
"""Rename a folder."""
...
@router.delete("/folders/{folder_id}")
async def delete_folder(folder_id: str):
"""Delete a folder."""
...
# ── Global Characters ──
@router.get("/characters")
async def list_global_characters():
```
## 第 19 页
```text
"""List global characters (paginated, searchable)."""
...
@router.post("/characters")
async def create_global_character():
"""Create a global character."""
...
@router.patch("/characters/{character_id}")
async def update_global_character(character_id: str):
"""Update a global character."""
...
@router.delete("/characters/{character_id}")
async def delete_global_character(character_id: str):
"""Delete a global character."""
...
@router.post("/characters/{character_id}/generate")
async def generate_character_image(character_id: str):
"""Generate character appearance candidates."""
...
@router.post("/characters/{character_id}/select")
async def select_character_image(character_id: str):
"""Select character appearance (card-draw confirm)."""
...
@router.post("/characters/{character_id}/undo")
async def undo_character_image(character_id: str):
"""Undo character appearance selection."""
...
# ── Global Locations ──
@router.get("/locations")
async def list_global_locations():
"""List global locations."""
...
@router.post("/locations")
async def create_global_location():
"""Create a global location."""
...
@router.post("/locations/{location_id}/generate")
async def generate_location_image(location_id: str):
"""Generate location image candidates."""
...
```
## 第 20 页
```text
@router.post("/locations/{location_id}/select")
async def select_location_image(location_id: str):
"""Select location image."""
...
# ── Global Voices ──
@router.get("/voices")
async def list_global_voices():
"""List global voices."""
...
@router.post("/voices")
async def create_global_voice():
"""Create/clone a voice."""
...
@router.delete("/voices/{voice_id}")
async def delete_global_voice(voice_id: str):
"""Delete a voice."""
...
// File: backend/app/api/assets.py
"""Project assets API routes."""
from fastapi import APIRouter
router = APIRouter()
@router.get("/projects/{project_id}/characters")
async def list_characters(project_id: str):
"""List characters for a project."""
...
@router.get("/projects/{project_id}/characters/{character_id}")
async def get_character(project_id: str, character_id: str):
"""Get character details with appearances."""
...
@router.patch("/projects/{project_id}/characters/{character_id}")
async def update_character(project_id: str, character_id: str):
"""Update character (manual edit)."""
...
@router.post("/projects/{project_id}/characters/{character_id}/select-appearance")
async def select_character_appearance(project_id: str, character_id: str):
"""Select a character appearance."""
...
@router.get("/projects/{project_id}/locations")
```
## 第 21 页
```text
async def list_locations(project_id: str):
"""List locations for a project."""
...
@router.get("/projects/{project_id}/episodes")
async def list_episodes(project_id: str):
"""List episodes for a project."""
...
@router.get("/projects/{project_id}/episodes/{episode_id}/panels")
async def list_panels(project_id: str, episode_id: str):
"""List panels for an episode with media URLs."""
...
@router.get("/projects/{project_id}/tasks")
async def list_tasks(project_id: str):
"""List async tasks (paginated, filterable by status)."""
...
@router.post("/projects/{project_id}/copy-from-global")
async def copy_from_global(project_id: str):
"""Deep-copy asset from global Asset Hub to project."""
...
// File: backend/app/api/auth.py
"""Authentication API routes."""
from fastapi import APIRouter, HTTPException
from app.deps import DbSession, CurrentUser
from app.schemas.auth import RegisterRequest, LoginRequest, TokenResponse, UserResponse
from app.services.auth_service import (
create_access_token,
create_user,
get_user_by_email,
verify_password,
)
router = APIRouter()
@router.post("/register", response_model=TokenResponse)
async def register(body: RegisterRequest, db: DbSession):
"""Register a new user."""
existing = await get_user_by_email(db, body.email)
if existing:
raise HTTPException(status_code=400, detail="Email already registered")
user = await create_user(db, body.email, body.password, body.name)
token = create_access_token(user.id)
return TokenResponse(access_token=token)
@router.post("/login", response_model=TokenResponse)
async def login(body: LoginRequest, db: DbSession):
"""Login and return JWT token."""
```
## 第 22 页
```text
user = await get_user_by_email(db, body.email)
if not user or not verify_password(body.password, user.password_hash):
raise HTTPException(status_code=401, detail="Invalid email or password")
token = create_access_token(user.id)
return TokenResponse(access_token=token)
@router.get("/me", response_model=UserResponse)
async def get_me(user: CurrentUser):
"""Get current user info."""
return user
// File: backend/app/api/billing.py
"""Billing API routes."""
from fastapi import APIRouter
router = APIRouter()
@router.get("/balance")
async def get_balance():
"""Get user balance."""
...
@router.get("/transactions")
async def list_transactions():
"""List transactions (paginated)."""
...
@router.post("/topup")
async def topup():
"""Top up balance."""
...
// File: backend/app/api/candidates.py
"""Candidate card-draw API routes."""
from fastapi import APIRouter
from pydantic import BaseModel
router = APIRouter()
class ConfirmCandidateRequest(BaseModel):
entity_type: str # "character_appearance" | "location_image" | "panel"
entity_id: str
selected_index: int
class CancelCandidateRequest(BaseModel):
entity_type: str
entity_id: str
class UndoCandidateRequest(BaseModel):
entity_type: str
```
## 第 23 页
```text
entity_id: str
@router.post("/confirm")
async def confirm_candidate(body: ConfirmCandidateRequest):
"""Confirm candidate selection — persist to entity."""
...
@router.post("/cancel")
async def cancel_candidate(body: CancelCandidateRequest):
"""Cancel card-draw, clear candidates."""
...
@router.post("/undo")
async def undo_candidate(body: UndoCandidateRequest):
"""Undo to previous version."""
...
@router.get("/panels/{panel_id}/history")
async def get_panel_history(panel_id: str):
"""Get panel image version history."""
...
@router.post("/panels/{panel_id}/restore")
async def restore_panel_version(panel_id: str):
"""Restore panel to a specific history version."""
...
// File: backend/app/api/conversations.py
"""Conversation API routes — Core Agent interaction entry point."""
import json
import uuid
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage
from app.deps import DbSession, CurrentUser, AgentGraph
from app.db.session import async_session_factory
from app.schemas.conversation import (
SendMessageRequest,
ResumeRequest,
ConversationResponse,
ConversationDetailResponse,
MessageResponse,
)
from app.services import conversation_service, project_service
from app.agents.sse_transformer import stream_graph_to_sse
router = APIRouter()
@router.post("/projects/{project_id}/conversations", response_model=ConversationResponse)
async def create_conversation(project_id: uuid.UUID, user: CurrentUser, db: DbSession):
"""Create a new conversation for a project."""
```
## 第 24 页
```text
project = await project_service.get_project(db, project_id)
if not project or project.user_id != user.id:
raise HTTPException(status_code=404, detail="Project not found")
conv = await conversation_service.create_conversation(db, project_id, user.id)
return conv
@router.get("/projects/{project_id}/conversations", response_model=list[ConversationResponse])
async def list_conversations(project_id: uuid.UUID, user: CurrentUser, db: DbSession):
"""List conversations for a project."""
project = await project_service.get_project(db, project_id)
if not project or project.user_id != user.id:
raise HTTPException(status_code=404, detail="Project not found")
convs = await conversation_service.list_conversations(db, project_id)
return convs
@router.get("/conversations/{conversation_id}", response_model=ConversationDetailResponse)
async def get_conversation(conversation_id: uuid.UUID, user: CurrentUser, db: DbSession):
"""Get conversation details with message history."""
conv = await conversation_service.get_conversation(db, conversation_id)
if not conv or conv.user_id != user.id:
raise HTTPException(status_code=404, detail="Conversation not found")
messages = await conversation_service.get_messages(db, conversation_id)
return ConversationDetailResponse(
id=conv.id,
project_id=conv.project_id,
title=conv.title,
status=conv.status,
created_at=conv.created_at,
messages=[MessageResponse.model_validate(m) for m in messages],
)
@router.post("/conversations/{conversation_id}/messages")
async def send_message(
conversation_id: uuid.UUID,
body: SendMessageRequest,
user: CurrentUser,
db: DbSession,
graph: AgentGraph,
):
"""Send a message and return SSE stream."""
# Verify conversation exists and belongs to user
conv = await conversation_service.get_conversation(db, conversation_id)
if not conv or conv.user_id != user.id:
raise HTTPException(status_code=404, detail="Conversation not found")
# Save user message to DB
await conversation_service.save_message(db, conversation_id, "user", body.content)
# Build LangGraph input
input_state = {"messages": [HumanMessage(content=body.content)]}
config = {"configurable": {"thread_id": str(conversation_id)}}
async def event_stream():
full_response = ""
```
## 第 25 页
```text
try:
async for event in stream_graph_to_sse(graph, input_state, config):
yield event
# Capture full response from done event
if event.startswith("event: done"):
try:
data_line = event.split("data: ", 1)[1].strip()
done_data = json.loads(data_line)
full_response = done_data.get("full_response", "")
except (IndexError, json.JSONDecodeError):
pass
finally:
# Save assistant message in a new session (stream outlives request session)
if full_response:
async with async_session_factory() as save_db:
try:
await conversation_service.save_message(
save_db, conversation_id, "assistant", full_response, agent_name="producer"
)
await save_db.commit()
except Exception:
await save_db.rollback()
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
@router.post("/conversations/{conversation_id}/resume")
async def resume_conversation(conversation_id: uuid.UUID, body: ResumeRequest):
"""Resume conversation after user confirmation/modification."""
async def event_stream():
yield f"event: done\ndata: {json.dumps({'reason': 'complete'})}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
@router.post("/conversations/{conversation_id}/steer")
async def steer_conversation(conversation_id: uuid.UUID):
"""Inject system instruction without triggering full Agent cycle."""
...
@router.post("/conversations/{conversation_id}/abort")
async def abort_conversation(conversation_id: uuid.UUID):
"""Abort current Agent execution."""
...
@router.post("/conversations/{conversation_id}/fork")
async def fork_conversation(conversation_id: uuid.UUID):
"""Fork conversation from current checkpoint."""
...
```
## 第 26 页
```text
@router.post("/conversations/{conversation_id}/compact")
async def compact_conversation(conversation_id: uuid.UUID):
"""Compress conversation history."""
...
// File: backend/app/api/projects.py
"""Project API routes."""
import uuid
from fastapi import APIRouter, HTTPException
from app.deps import DbSession, CurrentUser
from app.schemas.project import CreateProjectRequest, UpdateProjectRequest, ProjectResponse
from app.services import project_service
router = APIRouter()
@router.post("", response_model=ProjectResponse)
async def create_project(body: CreateProjectRequest, user: CurrentUser, db: DbSession):
"""Create a new project."""
project = await project_service.create_project(
db, user.id, body.title, body.description, body.style
)
return project
@router.get("", response_model=list[ProjectResponse])
async def list_projects(user: CurrentUser, db: DbSession):
"""List projects (paginated)."""
projects = await project_service.list_projects(db, user.id)
return projects
@router.get("/{project_id}", response_model=ProjectResponse)
async def get_project(project_id: uuid.UUID, user: CurrentUser, db: DbSession):
"""Get project details."""
project = await project_service.get_project(db, project_id)
if not project or project.user_id != user.id:
raise HTTPException(status_code=404, detail="Project not found")
return project
@router.patch("/{project_id}", response_model=ProjectResponse)
async def update_project(
project_id: uuid.UUID,
body: UpdateProjectRequest,
user: CurrentUser,
db: DbSession,
):
"""Update project."""
project = await project_service.get_project(db, project_id)
if not project or project.user_id != user.id:
raise HTTPException(status_code=404, detail="Project not found")
updated = await project_service.update_project(
db, project, **body.model_dump(exclude_unset=True)
```
## 第 27 页
```text
)
return updated
@router.delete("/{project_id}")
async def delete_project(project_id: uuid.UUID, user: CurrentUser, db: DbSession):
"""Soft-delete project."""
project = await project_service.get_project(db, project_id)
if not project or project.user_id != user.id:
raise HTTPException(status_code=404, detail="Project not found")
await project_service.soft_delete_project(db, project)
return {"ok": True}
// File: backend/app/config.py
"""Application configuration via environment variables."""
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""Application settings loaded from environment variables."""
# ── App ──
app_name: str = "StudioAgent"
debug: bool = False
api_prefix: str = "/api/v1"
# ── Database ──
database_url: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/studioagent"
database_url_sync: str = "postgresql://postgres:postgres@localhost:5432/studioagent"
# ── Redis ──
redis_url: str = "redis://localhost:6379/0"
# ── Auth ──
jwt_secret: str = "change-me-in-production"
jwt_algorithm: str = "HS256"
jwt_expire_minutes: int = 60 * 24 * 7 # 7 days
# ── LLM Providers ──
openrouter_api_key: str = ""
openrouter_base_url: str = "https://openrouter.ai/api/v1"
google_api_key: str = ""
google_base_url: str = "" # empty = SDK default
volcengine_api_key: str = ""
volcengine_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
volcengine_endpoint_id: str = ""
# ── Agent Models (format: "provider/model_id") ──
producer_model: str = "openrouter/anthropic/claude-opus-4"
screenwriter_model: str = "google/gemini-2.5-flash-preview-05-20"
director_model: str = "google/gemini-2.5-flash-preview-05-20"
camera_model: str = "google/gemini-2.5-flash-preview-05-20"
editor_model: str = "google/gemini-2.5-flash-preview-05-20"
sound_model: str = "google/gemini-2.5-flash-preview-05-20"
# ── Image Generation ──
fal_api_key: str = ""
volcengine_image_api_key: str = ""
```
## 第 28 页
```text
# ── Video Generation ──
volcengine_video_api_key: str = ""
# ── Voice Generation ──
dashscope_api_key: str = "" # Alibaba Qwen TTS
elevenlabs_api_key: str = ""
# ── Storage ──
s3_bucket: str = ""
s3_region: str = ""
s3_access_key: str = ""
s3_secret_key: str = ""
s3_endpoint_url: str = ""
# ── Agent ──
max_handoff_count: int = 10
default_interaction_mode: str = "collaborative"
# ── Celery ──
celery_broker_url: str = "redis://localhost:6379/1"
celery_result_backend: str = "redis://localhost:6379/2"
model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "allow"}
settings = Settings()
// File: backend/app/db/migrations/env.py
"""Alembic async migration environment."""
import asyncio
import sys
from logging.config import fileConfig
from pathlib import Path
from alembic import context
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine
# Ensure backend/ is on sys.path so `app` package is importable
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
from app.config import settings # noqa: E402
# Import Base and all models so Alembic can see them for autogenerate
from app.models import Base # noqa: E402, F401
import app.models # noqa: E402, F401
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode."""
url = settings.database_url_sync
```
## 第 29 页
```text
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection) -> None:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_async_migrations() -> None:
"""Run migrations in 'online' mode with async engine."""
connectable = create_async_engine(
settings.database_url,
poolclass=pool.NullPool,
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode."""
asyncio.run(run_async_migrations())
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
// File: backend/app/db/migrations/versions/001_initial_schema.py
"""initial schema
Revision ID: 001_initial
Revises:
Create Date: 2026-03-05
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = "001_initial"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ── Users ──
```
## 第 30 页
```text
op.create_table(
"users",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("email", sa.String(), nullable=False, unique=True),
sa.Column("password_hash", sa.String(), nullable=False),
sa.Column("name", sa.String(), nullable=True),
sa.Column("avatar_url", sa.Text(), nullable=True),
sa.Column("balance", sa.Numeric(12, 2), server_default="0"),
sa.Column("preferences", postgresql.JSONB(), server_default="{}"),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
# ── Projects ──
op.create_table(
"projects",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("title", sa.String(), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("status", sa.String(), server_default="active"),
sa.Column("style", sa.String(), server_default="anime"),
sa.Column("metadata", postgresql.JSONB(), server_default="{}"),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
# ── Conversations ──
op.create_table(
"conversations",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=False),
sa.Column("title", sa.String(), nullable=True),
sa.Column("status", sa.String(), server_default="idle"),
sa.Column("parent_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("conversations.id"), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
# ── Messages ──
op.create_table(
"messages",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("conversation_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("conversations.id", ondelete="CASCADE"), nullable=False),
sa.Column("role", sa.String(), nullable=False),
sa.Column("agent_name", sa.String(), nullable=True),
sa.Column("content", sa.Text(), nullable=True),
sa.Column("tool_calls", postgresql.JSONB(), nullable=True),
sa.Column("tool_call_id", sa.String(), nullable=True),
sa.Column("reasoning", sa.Text(), nullable=True),
sa.Column("metadata", postgresql.JSONB(), server_default="{}"),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
)
# ── Characters ──
op.create_table(
"characters",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("project_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False),
```