参数说明
基本参数
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| prompt | 字符串 | "" | 详细描述要生成视频内容的文本 |
| aspect_ratio | 选择项 | ”16:9” | 输出视频的宽高比，可选”16:9”或”9:16” |
| negative_prompt | 字符串 | "" | 指导系统避免在视频中出现的内容 |
| duration_seconds | 整数 | 5 | 输出视频的持续时间，5-8秒 |
| enhance_prompt | 布尔值 | True | 是否使用AI辅助增强提示词 |
| person_generation | 选择项 | ”ALLOW” | 是否允许生成人物，可选”ALLOW”或”BLOCK” |
| seed | 整数 | 0 | 随机种子，0表示随机生成 |
可选参数
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| image | 图像 | None | 可选的参考图像，用于引导视频生成 |
输出
| 输出 | 类型 | 说明 |
|---|---|---|
| VIDEO | 视频 | 生成的视频结果 |
源码参考
[节点源码 (更新于2025-05-03)]
class VeoVideoGenerationNode(ComfyNodeABC):
"""
Generates videos from text prompts using Google's Veo API.
This node can create videos from text descriptions and optional image inputs,
with control over parameters like aspect ratio, duration, and more.
"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Text description of the video",
},
),
"aspect_ratio": (
IO.COMBO,
{
"options": ["16:9", "9:16"],
"default": "16:9",
"tooltip": "Aspect ratio of the output video",
},
),
},
"optional": {
"negative_prompt": (
IO.STRING,
{
"multiline": True,
"default": "",
"tooltip": "Negative text prompt to guide what to avoid in the video",
},
),
"duration_seconds": (
IO.INT,
{
"default": 5,
"min": 5,
"max": 8,
"step": 1,
"display": "number",
"tooltip": "Duration of the output video in seconds",
},
),
"enhance_prompt": (
IO.BOOLEAN,
{
"default": True,
"tooltip": "Whether to enhance the prompt with AI assistance",
}
),
"person_generation": (
IO.COMBO,
{
"options": ["ALLOW", "BLOCK"],
"default": "ALLOW",
"tooltip": "Whether to allow generating people in the video",
},
),
"seed": (
IO.INT,
{
"default": 0,
"min": 0,
"max": 0xFFFFFFFF,
"step": 1,
"display": "number",
"control_after_generate": True,
"tooltip": "Seed for video generation (0 for random)",
},
),
"image": (IO.IMAGE, {
"default": None,
"tooltip": "Optional reference image to guide video generation",
}),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
},
}
RETURN_TYPES = (IO.VIDEO,)
FUNCTION = "generate_video"
CATEGORY = "api node/video/Veo"
DESCRIPTION = "Generates videos from text prompts using Google's Veo API"
API_NODE = True
def generate_video(
self,
prompt,
aspect_ratio="16:9",
negative_prompt="",
duration_seconds=5,
enhance_prompt=True,
person_generation="ALLOW",
seed=0,
image=None,
auth_token=None,
):
# 准备请求的 instances
instances = []
instance = {
"prompt": prompt
}
# 如果提供了图像就添加
if image is not None:
image_base64 = convert_image_to_base64(image)
if image_base64:
instance["image"] = {
"bytesBase64Encoded": image_base64,
"mimeType": "image/png"
}
instances.append(instance)
# 创建参数字典
parameters = {
"aspectRatio": aspect_ratio,
"personGeneration": person_generation,
"durationSeconds": duration_seconds,
"enhancePrompt": enhance_prompt,
}
# 如果提供了可选参数就添加
if negative_prompt:
parameters["negativePrompt"] = negative_prompt
if seed > 0:
parameters["seed"] = seed
# 发起启动视频生成的初始请求
initial_operation = SynchronousOperation(
endpoint=ApiEndpoint(
path="/proxy/veo/generate",
method=HttpMethod.POST,
request_model=Veo2GenVidRequest,
response_model=Veo2GenVidResponse
),
request=Veo2GenVidRequest(
instances=instances,
parameters=parameters
),
auth_token=auth_token
)
initial_response = initial_operation.execute()
operation_name = initial_response.name
logging.info(f"Veo generation started with operation name: {operation_name}")
# 定义状态提取函数
def status_extractor(response):
# 只要操作已结束就返回 "completed"，不区分成功或失败
# 轮询结束后我们再检查错误
return "completed" if response.done else "pending"
# 定义进度提取函数
def progress_extractor(response):
# 如果 API 提供了进度信息，这里可以进一步增强
return None
# 定义轮询操作
poll_operation = PollingOperation(
poll_endpoint=ApiEndpoint(
path="/proxy/veo/poll",
method=HttpMethod.POST,
request_model=Veo2GenVidPollRequest,
response_model=Veo2GenVidPollResponse
),
completed_statuses=["completed"],
failed_statuses=[], # 没有失败状态，我们在轮询结束后处理错误
status_extractor=status_extractor,
progress_extractor=progress_extractor,
request=Veo2GenVidPollRequest(
operationName=operation_name
),
auth_token=auth_token,
poll_interval=5.0
)
# 执行轮询操作
poll_response = poll_operation.execute()
# 现在检查最终响应中的错误
# 检查轮询响应中的错误
if hasattr(poll_response, 'error') and poll_response.error:
error_message = f"Veo API error: {poll_response.error.message} (code: {poll_response.error.code})"
logging.error(error_message)
raise Exception(error_message)
# 检查被 RAI 过滤的内容
if (hasattr(poll_response.response, 'raiMediaFilteredCount') and
poll_response.response.raiMediaFilteredCount > 0):
# 如果有原因信息就提取
if (hasattr(poll_response.response, 'raiMediaFilteredReasons') and
poll_response.response.raiMediaFilteredReasons):
reason = poll_response.response.raiMediaFilteredReasons[0]
error_message = f"Content filtered by Google's Responsible AI practices: {reason} ({poll_response.response.raiMediaFilteredCount} videos filtered.)"
else:
error_message = f"Content filtered by Google's Responsible AI practices ({poll_response.response.raiMediaFilteredCount} videos filtered.)"
logging.error(error_message)
raise Exception(error_message)
# 提取视频数据
video_data = None
if poll_response.response and hasattr(poll_response.response, 'videos') and poll_response.response.videos and len(poll_response.response.videos) > 0:
video = poll_response.response.videos[0]
# 检查视频是以 base64 还是 URL 形式提供
if hasattr(video, 'bytesBase64Encoded') and video.bytesBase64Encoded:
# 将 base64 字符串解码为字节
video_data = base64.b64decode(video.bytesBase64Encoded)
elif hasattr(video, 'gcsUri') and video.gcsUri:
# 从 URL 下载
video_url = video.gcsUri
video_response = requests.get(video_url)
video_data = video_response.content
else:
raise Exception("Video returned but no data or URL was provided")
else:
raise Exception("Video generation completed but no video was returned")
if not video_data:
raise Exception("No video data was returned")
logging.info("Video generation completed successfully")
# 将视频数据转换为 BytesIO 对象
video_io = io.BytesIO(video_data)
# 返回 VideoFromFile 对象
return (VideoFromFile(video_io),)