{
  "tool": "RetrieveMaterials",
  "params": {"query": "海边日落"}
}

[{"element_type": "TitleCard", "content": "AI 剪辑实战"}]

{
  "result": [{"element_type": "TitleCard", "content": "AI 剪辑实战"}]
}

{
  "duration": "30秒"
}

visual_plan_schema = {
    "type": "array",
    "items": {
        "type": "object",
        "properties": {
            "sentence_index": {"type": "integer"},
            "element_type": {
                "type": "string",
                "enum": ["TitleCard", "BRoll", "QuoteCard", "CodeBlock", "ListCard", "TalkingHead"]
            },
            "reason": {"type": "string"},
            "suggested_content": {"type": "string"}
        },
        "required": ["sentence_index", "element_type", "reason"]
    }
}

import openai

client = openai.OpenAI(
    api_key="your-api-key",
    base_url="https://api.deepseek.com/v1"
)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个专业的视频剪辑规划助手。"},
        {"role": "user", "content": prompt}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "visual_plan",
            "schema": visual_plan_schema
        }
    },
    temperature=0.2
)

result = json.loads(response.choices[0].message.content)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": prompt}],
    tools=[
        {
            "type": "function",
            "function": {
                "name": "plan_visual_elements",
                "description": "为口播视频规划视觉元素",
                "parameters": visual_plan_schema
            }
        }
    ],
    tool_choice={"type": "function", "function": {"name": "plan_visual_elements"}}
)

tool_call = response.choices[0].message.tool_calls[0]
result = json.loads(tool_call.function.arguments)

from pydantic import BaseModel, Field, field_validator
from typing import Literal

class VisualElement(BaseModel):
    sentence_index: int = Field(ge=0)
    element_type: Literal["TitleCard", "BRoll", "QuoteCard", "CodeBlock", "ListCard", "TalkingHead"]
    reason: str = Field(min_length=5, max_length=200)
    suggested_content: str = ""

    @field_validator('suggested_content')
    @classmethod
    def validate_content(cls, v, info):
        element_type = info.data.get('element_type')
        if element_type == "TitleCard" and len(v) < 2:
            raise ValueError("TitleCard 必须有标题内容")
        return v

class VisualPlan(BaseModel):
    elements: list[VisualElement]

def parse_visual_plan(raw_output: str) -> VisualPlan:
    try:
        data = json.loads(raw_output)
    except json.JSONDecodeError as e:
        raise ValueError(f"JSON 解析失败: {e}")

    # 兼容 LLM 外层多套一层 result 的情况
    if isinstance(data, dict) and "result" in data and isinstance(data["result"], list):
        data = data["result"]

    return VisualPlan(elements=data)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=5),
    retry=retry_if_exception_type(ValueError)
)
def generate_with_retry(prompt: str, schema: dict) -> dict:
    raw = llm.generate(prompt, response_format=schema)

    try:
        return parse_visual_plan(raw)
    except ValueError as e:
        # 把错误信息加入 Prompt，重新生成
        prompt += f"\n\n注意：你之前的输出校验失败，错误是：{e}\n请严格按照 JSON Schema 输出。"
        raise

temperature_schedule = [0.2, 0.1, 0.0]

def safe_generate_visual_plan(script: str) -> VisualPlan:
    try:
        return generate_with_retry(script)
    except Exception:
        # 降级：所有句子保持 TalkingHead
        return VisualPlan(elements=[
            VisualElement(
                sentence_index=i,
                element_type="TalkingHead",
                reason="LLM 规划失败，降级为默认方案"
            )
            for i in range(len(script.split('。')))
        ])

import json
from pydantic import BaseModel, Field
from typing import Literal

class VisualElement(BaseModel):
    sentence_index: int = Field(ge=0)
    element_type: Literal["TitleCard", "BRoll", "QuoteCard", "CodeBlock", "ListCard", "TalkingHead"]
    reason: str = Field(min_length=5, max_length=200)
    suggested_content: str = ""

def clean_llm_output(raw: str) -> str:
    """清理 LLM 可能产生的多余格式"""
    raw = raw.strip()
    if raw.startswith("```"):
        raw = raw.strip("`")
        if raw.startswith("json"):
            raw = raw[4:].strip()
    return raw

def robust_parse(raw: str) -> list[VisualElement]:
    raw = clean_llm_output(raw)

    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        # 尝试找到第一个 [ 和最后一个 ]
        start = raw.find("[")
        end = raw.rfind("]")
        if start != -1 and end != -1:
            data = json.loads(raw[start:end+1])
        else:
            raise

    # 兼容外层嵌套
    if isinstance(data, dict):
        for key in ["result", "data", "elements"]:
            if key in data and isinstance(data[key], list):
                data = data[key]
                break

    if not isinstance(data, list):
        raise ValueError(f"期望数组，实际得到 {type(data)}")

    return [VisualElement(**item) for item in data]