总体思路:

agent作为调用者,要按剧本->视频的调用方式进行调用这是此agent唯一的功能,所以prompt非常好写

2.script_maker_tool:
此agent为编写剧本agent,我们需要将用户的话语扩充成剧本形式,即整个视频每一分幕展现的样子,此agent只运用到了一个大模型,难度在于prompt的编写,prompt在此不提供但是如下图为此agent的输出可供参考:

3.video_maker_tool:
实现功能最重要的工具,这里运用到了一个库名为:manim,本质上是通过使用manim库输入命令行命令生成的视频,manim库通用使用格式:
 

from manim import *

    class tests(Scene):

        def construct(self):

            tri = Triangle()

            self.play(Rotate(tri, PI))

            self.wait()

在prompt中强力要求大模型只输出上述代码格式 最后使用video_maker工具即可生成视频

video_maker代码:
 

@tool('video_maker_tool',parse_docstring=True)

def video_maker_tool(script: str) -> str:

    """工具函数：根据脚本生成manim动画代码并渲染成视频。

    Args:

        script: 生成好的动画剧本

    Returns:

        返回视频的本地URL路径，如果失败则返回错误信息

    """

    try:

        # 1. 使用LLM生成Manim代码

        code_response = video_maker_llm.invoke(prompt_template.format(script=script))

        code = code_response.content if hasattr(code_response, 'content') else str(code_response)

       

        # 清理可能的代码块标记

        if code.startswith("```python"):

            code = code[9:]

        code = code.strip()

       

        # 2. 安全检查

        try:

            check_code_safety(code)

        except ValueError as e:

            return f"代码安全检查失败: {e}"

       

        # 3. 创建临时目录并渲染视频

        with tempfile.TemporaryDirectory() as tmpdir:

            script_path = Path(tmpdir) / "tests.py"

            script_path.write_text(code, encoding="utf-8")

            video_id = str(uuid.uuid4())

            host_video_path = VIDEO_DIR / f"{video_id}.mp4"

            try:

                # 设置环境变量以解决中文编码问题

                env = os.environ.copy()

                env["PYTHONIOENCODING"] = "utf-8"

               

                # 运行Manim渲染

                result = subprocess.run(

                    ["manim", "-pqm", str(script_path), "tests"],

                    cwd=tmpdir,

                    capture_output=True,

                    timeout=600,

                    env=env

                )

                # 安全解码输出

                stdout_str = result.stdout.decode('utf-8', errors='replace')

                stderr_str = result.stderr.decode('utf-8', errors='replace')

                if result.returncode != 0:

                    err = stderr_str or stdout_str

                    return f"Manim渲染失败: {err}"

                # 查找生成的视频文件

                mp4_files = list(Path(tmpdir).rglob("tests*.mp4"))

                if not mp4_files:

                    return "未找到生成的视频文件"

                # 复制视频到目标目录

                import shutil

                shutil.copy(mp4_files[0], host_video_path)

               

                # 返回本地URL

                return f"videos/{video_id}.mp4"

            except subprocess.TimeoutExpired:

                return "渲染超时"

            except Exception as e:

                return f"渲染过程中出错: {str(e)}"

       

    except Exception as e:

        return f"视频生成失败: {str(e)}"

安全验证可以让ai生成 避免篇幅过长这里省略

总结:
大部分代码可以通过ai进行生成,如安全验证,video_maker_tool此工具,以及manim代码的使用与生成,作为调用ai的人员,主要工作是构思系统以及选用工具,不只有manim库可以达到这种效果,moviepy也可以,但是trea使用moviepy会有bug所以这里不做选用,调用大模型要锻炼自身对prompt的编写,不提供prompt也是这个原因

低代码平台:
整体来说实现和上述大同小异,唯一需要注意的是:

代码节点中是调用api形式实现的,为什么要调用api,因为云端并没有manim等库,所以就算代码写的是manim代码的解码编写也是无法运行的,这里需要使用到ngrok将自己电脑上的端口暴露,再使用低平台链接即可

代码节点:

import json
import requests
import urllib3
import asyncio
from typing import Dict, Any

# 禁用SSL警告（对于ngrok等自签名证书）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

async def main(args) -> Dict[str, Any]:
    # 获取输入参数
    params = args.params
    code_input = params.get('code', '') or params.get('input', '')
    
    # 处理转义字符和格式问题
    if code_input and isinstance(code_input, str):
        # 移除多余的双引号
        if code_input.startswith('"') and code_input.endswith('"'):
            code_input = code_input[1:-1]
        
        # 处理多层转义字符
        # 将 \\n 转换为 \n，将 \\" 转换为 "
        code_input = code_input.replace('\\\\n', '\n').replace('\\\"', '"')
        code_input = code_input.replace('\\n', '\n').replace('\\"', '"')
    
    # 验证输入参数
    if not code_input:
        ret = {
            "success": False,
            "error": "Valid code is required",
            "status": 400,
            "debug_info": {
                "received_params": list(params.keys()),
                "code_preview": str(params.get('code', ''))[:100] if params.get('code') else 'No code provided'
            }
        }
        return ret
    
    # 检查代码是否包含Manim必要元素
    if 'from manim import' not in code_input and 'class' not in code_input:
        ret = {
            "success": False,
            "error": "The provided code does not appear to be valid Manim code",
            "status": 400,
            "code_preview": code_input[:200]
        }
        return ret
    
    # API端点URL
    api_url = "your_ngrok"
    
    # 准备请求数据
    request_data = {
        'code': code_input
    }
    
    try:
        # 设置请求头
        headers = {
            'Content-Type': 'application/json',
            'ngrok-skip-browser-warning': 'true'
        }
        
        # 使用线程池执行同步的requests调用，避免阻塞事件循环
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(
            None, 
            lambda: requests.post(
                api_url, 
                json=request_data, 
                headers=headers,
                verify=False,
                timeout=6000
            )
        )
        
        # 获取响应数据
        try:
            response_data = response.json()
        except ValueError:
            response_data = {'detail': response.text}
        
        # 检查响应状态
        if response.ok:
            ret = {
                "success": True,
                "video_url": response_data.get('video_url', ''),
                "message": "Animation rendered successfully",
                "status": response.status_code,
                "response": response_data
            }
        else:
            ret = {
                "success": False,
                "error": response_data.get('detail', 'Unknown error'),
                "status": response.status_code,
                "response": response_data
            }
    
    except requests.exceptions.Timeout:
        ret = {
            "success": False,
            "error": "Request timeout",
            "status": 408
        }
    
    except requests.exceptions.ConnectionError:
        ret = {
            "success": False,
            "error": "Connection error - could not reach the API",
            "status": 0
        }
    
    except Exception as e:
        ret = {
            "success": False,
            "error": f"Unexpected error: {str(e)}",
            "status": 500
        }
    
    return ret

其他低代码平台无难度,唯一需要注意仅上述内容