import openai

messages = [
    {"role": "system", "content": "你是一位短视频脚本专家。"},
    {"role": "user", "content": "主题是'如何用 LangChain 生成视频脚本'，风格轻松，30 秒左右。"}
]
resp = openai.chat.completions.create(model="gpt-4o", messages=messages)
script = resp.choices[0].message.content

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位短视频脚本专家，擅长把技术概念写成{style}风格的口播脚本。"),
    ("human", "主题是：{topic}，时长控制在{duration}秒，需要{scene_count}个分镜。")
])

messages = prompt.invoke({
    "style": "轻松幽默",
    "topic": "LangChain 如何自动生成视频脚本",
    "duration": 30,
    "scene_count": 3
})

response = llm.invoke(messages)
print(response.content)

{
  "scenes": [
    {"time": "0-10s", "narration": "...", "visual": "...", "bgm": "..."}
  ],
  "title": "...",
  "tags": ["..."]
}

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate

class Scene(BaseModel):
    time_range: str = Field(description="时间区间，如 0-10s")
    narration: str = Field(description="口播文案")
    visual: str = Field(description="画面描述，用于后续生成图片/视频")
    bgm_mood: str = Field(description="背景音乐情绪")

class VideoScript(BaseModel):
    title: str
    scenes: list[Scene]
    tags: list[str]

parser = PydanticOutputParser(pydantic_object=VideoScript)

prompt = PromptTemplate(
    template="根据主题生成短视频脚本。\n{format_instructions}\n主题：{topic}\n时长：{duration}秒\n",
    input_variables=["topic", "duration"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | llm | parser
result = chain.invoke({"topic": "LangChain 自动生成短视频", "duration": 30})
print(result.model_dump_json(indent=2))

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory()
conversation = ConversationChain(llm=llm, memory=memory)

conversation.predict(input="帮我写一个 30 秒的 LangChain 短视频脚本")
conversation.predict(input="第二段太正式了，改成口语化一些")
print(memory.load_memory_variables({}))

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 子链 1：生成标题
title_prompt = ChatPromptTemplate.from_template("为'{topic}'写一个 20 字以内的爆款标题")
title_chain = title_prompt | llm | StrOutputParser()

# 子链 2：根据标题和主题生成脚本
script_prompt = ChatPromptTemplate.from_template(
    "标题：{title}\n主题：{topic}\n时长：{duration}秒\n请生成 3 个分镜的脚本。"
)
script_chain = script_prompt | llm | StrOutputParser()

# 组合
full_chain = (
    {"topic": RunnablePassthrough(), "duration": RunnablePassthrough()}
    | RunnablePassthrough.assign(title=title_chain)
    | RunnablePassthrough.assign(script=script_chain)
)

result = full_chain.invoke({"topic": "LangChain 视频生成", "duration": 30})
print(result)

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains.combine_documents import create_stuff_documents_chain

# 1. 构建向量库
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["分镜案例 1：开头用悬念钩子...", "爆款标题公式：数字+痛点+解决方案..."],
    embedding=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 2. 构建 RAG Chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "参考以下素材生成脚本：\n\n{context}"),
    ("human", "主题：{topic}，时长：{duration}秒")
])

rag_chain = create_stuff_documents_chain(llm, prompt) | {
    "topic": lambda x: x["topic"],
    "context": retriever
}

# 实际调用时先检索文档，再塞入 prompt

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.tools import tool

@tool
def generate_image(description: str) -> str:
    """根据画面描述生成一张图片，返回图片 URL。"""
    # 实际接入 DALL·E / Midjourney / Stable Diffusion
    return f"https://fake-img.example.com?prompt={description}"

@tool
def generate_video_clip(description: str, duration: int) -> str:
    """根据画面描述生成一段视频片段，返回视频 URL。"""
    return f"https://fake-video.example.com?prompt={description}&duration={duration}"

tools = [generate_image, generate_video_clip]

agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位 AI 视频导演，能调用工具生成图片或视频。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_openai_tools_agent(llm, tools, agent_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

agent_executor.invoke({
    "input": "我要一个 30 秒关于 LangChain 的短视频，先出 3 张分镜图，再合成视频。"
})

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.runnables import RunnablePassthrough
from pydantic import BaseModel, Field

os.environ["OPENAI_API_KEY"] = "your-api-key"

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

class Scene(BaseModel):
    time_range: str
    narration: str
    visual: str
    tool: str = Field(description="生成该分镜所需工具：image 或 video")

class VideoProject(BaseModel):
    title: str
    target_platform: str
    scenes: list[Scene]

parser = PydanticOutputParser(pydantic_object=VideoProject)

title_chain = (
    ChatPromptTemplate.from_template("为'{topic}'写一个适合{platform}的爆款标题")
    | llm
    | StrOutputParser()
)

script_prompt = ChatPromptTemplate.from_template(
    """为平台"{target_platform}"生成短视频脚本。
{format_instructions}
标题：{title}
主题：{topic}
时长：{duration}秒
要求：每个分镜标注用 image（图片）还是 video（视频片段）生成。
"""
)

script_chain = (
    RunnablePassthrough.assign(title=title_chain)
    | script_prompt.partial(format_instructions=parser.get_format_instructions())
    | llm
    | parser
)

project = script_chain.invoke({
    "topic": "LangChain 如何帮助 AI 视频创作",
    "platform": "抖音",
    "target_platform": "抖音",
    "duration": 30
})

print(project.model_dump_json(indent=2, ensure_ascii=False))

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

builder = StateGraph(VideoGenState)
# ... 添加节点和边 ...
graph = builder.compile(checkpointer=MemorySaver())

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def invoke_with_retry(chain, inputs):
    return chain.invoke(inputs)

def process_video_task(task_id: str, inputs: dict):
    if db.get_status(task_id) in ("SUCCESS", "PROCESSING"):
        return db.get_result(task_id)

    db.set_status(task_id, "PROCESSING")
    try:
        result = video_chain.invoke(inputs)
        db.set_status(task_id, "SUCCESS", result=result)
        return result
    except Exception as e:
        db.set_status(task_id, "FAILURE", error=str(e))
        raise

config = {"configurable": {"thread_id": task_id}}
# 首次执行
graph.invoke(initial_state, config=config)

# Worker 重启后继续执行
state = graph.get_state(config)
graph.invoke(None, config=config)  # 从 checkpoint 继续

import asyncio

async def batch_generate(tasks):
    results = await script_chain.abatch(tasks)
    return results

# 并发处理 20 个脚本生成请求
tasks = [{"topic": t, "duration": 30} for t in topic_list]
results = asyncio.run(batch_generate(tasks))

async for chunk in chain.astream({"topic": "LangChain 高并发"}):
    print(chunk, end="", flush=True)

import asyncio
from asyncio import Semaphore

sem = Semaphore(10)  # 同时最多 10 个 LLM 调用

async def limited_invoke(inputs):
    async with sem:
        return await chain.ainvoke(inputs)

import hashlib

def get_cache_key(inputs: dict) -> str:
    return hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest()

key = get_cache_key(inputs)
if cached := redis.get(key):
    return json.loads(cached)

result = chain.invoke(inputs)
redis.setex(key, 3600, json.dumps(result))