openclaw/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity
Files
3183b2e28cd45d6bd77a9edd15b1e61a2dc5755f
hermes-agent/plugins/video_gen/fal/__init__.py
alt-glitch 3183b2e28c fix(video_gen): veo3.1 duration format and 4k resolution 
FAL veo3.1 API expects duration as "4s"/"6s"/"8s" (with unit suffix),
not bare "4"/"6"/"8" like other families. Add per-family duration_suffix
field and apply it in _build_payload. Also add "4k" to veo3.1 resolutions
per FAL API docs.

Note: the managed gateway currently rejects the "4s" format (expects
integer duration). Gateway-side fix needed for veo3.1 to work through
the Nous subscription path.
2026-05-29 22:26:24 +05:30

614 lines
22 KiB
Python
Raw Blame History

"""FAL.ai video generation backend.
User-facing surface: pick a **model family** (e.g. "Pixverse v6",
"Veo 3.1", "Seedance 2.0", "Kling v3 4K", "LTX 2.3", "Happy Horse").
The plugin auto-routes to the family's text-to-video endpoint when
called without ``image_url``, and to its image-to-video endpoint when
``image_url`` is provided. The agent never sees the routing — it just
calls ``video_generate(prompt=..., image_url=...)``.
Model families (each with t2v + i2v endpoints):
Cheap tier:
ltx-2.3 fal-ai/ltx-2.3-22b/text-to-video / fal-ai/ltx-2.3-22b/image-to-video
pixverse-v6 fal-ai/pixverse/v6/text-to-video / fal-ai/pixverse/v6/image-to-video
Premium tier:
veo3.1 fal-ai/veo3.1 / fal-ai/veo3.1/image-to-video
seedance-2.0 bytedance/seedance-2.0/text-to-video / bytedance/seedance-2.0/image-to-video
kling-v3-4k fal-ai/kling-video/v3/4k/text-to-video / fal-ai/kling-video/v3/4k/image-to-video
happy-horse alibaba/happy-horse/text-to-video / alibaba/happy-horse/image-to-video
Selection precedence for the active family:
1. ``model=`` arg from the tool call
2. ``FAL_VIDEO_MODEL`` env var
3. ``video_gen.fal.model`` in ``config.yaml``
4. ``video_gen.model`` in ``config.yaml`` (when it's one of our family IDs)
5. ``DEFAULT_MODEL``
Authentication via ``FAL_KEY`` or the managed Nous gateway. Output is an
HTTPS URL from FAL's CDN; the gateway downloads and delivers it.
"""
from __future__ import annotations
import logging
import os
import threading
import uuid
from typing import Any, Dict, List, Optional, Tuple
from agent.video_gen_provider import (
VideoGenProvider,
error_response,
success_response,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Family catalog
# ---------------------------------------------------------------------------
#
# Each family declares both endpoints (when available) plus a per-family
# capability sheet derived from FAL's OpenAPI schemas. Capability flags
# drive which keys get added to the request payload — keys a family doesn't
# advertise are dropped before send.
#
# Capabilities:
# aspect_ratios : tuple of supported ratios (None = endpoint decides)
# resolutions : tuple of supported resolutions (None = endpoint decides)
# durations : tuple of supported durations OR (min, max) range
# (heuristic: 2-element with gap > 1 is a range)
# audio : True if generate_audio is supported
# negative : True if negative_prompt is supported
FAL_FAMILIES: Dict[str, Dict[str, Any]] = {
# ─── Cheap / fast tier ─────────────────────────────────────────────
"ltx-2.3": {
"display": "LTX 2.3 (22B)",
"speed": "~30-60s",
"price": "cheap",
"strengths": "22B model with native audio generation. Affordable.",
"tier": "cheap",
"text_endpoint": "fal-ai/ltx-2.3-22b/text-to-video",
"image_endpoint": "fal-ai/ltx-2.3-22b/image-to-video",
# LTX docs don't expose duration/aspect/resolution enums — leave
# blank so we don't send unrecognized payload keys.
"aspect_ratios": None,
"resolutions": None,
"durations": None,
"audio": True,
"negative": True,
},
"pixverse-v6": {
"display": "Pixverse v6",
"speed": "~30-90s",
"price": "cheap",
"strengths": "Affordable. Negative prompts. 1-15s durations.",
"tier": "cheap",
"text_endpoint": "fal-ai/pixverse/v6/text-to-video",
"image_endpoint": "fal-ai/pixverse/v6/image-to-video",
"aspect_ratios": None,
"resolutions": ("360p", "540p", "720p", "1080p"),
"durations": (1, 15),
"audio": True,
"negative": True,
},
# ─── Expensive / premium tier ──────────────────────────────────────
"veo3.1": {
"display": "Veo 3.1",
"speed": "~60-120s",
"price": "premium",
"strengths": "Google DeepMind. Cinematic, native audio, strong prompt adherence.",
"tier": "premium",
"text_endpoint": "fal-ai/veo3.1",
"image_endpoint": "fal-ai/veo3.1/image-to-video",
"aspect_ratios": ("16:9", "9:16"),
"resolutions": ("720p", "1080p", "4k"),
"durations": (4, 6, 8),
"duration_suffix": "s", # FAL veo3.1 wants "4s" not "4"
"audio": True,
"negative": True,
},
"seedance-2.0": {
"display": "Seedance 2.0",
"speed": "~60-120s",
"price": "premium",
"strengths": "ByteDance. Cinematic, synchronized audio + lip-sync, 4-15s.",
"tier": "premium",
"text_endpoint": "bytedance/seedance-2.0/text-to-video",
"image_endpoint": "bytedance/seedance-2.0/image-to-video",
# Seedance accepts "auto" too — we omit it from the enum so the
# agent can't pass it; the endpoint defaults handle the rest.
"aspect_ratios": ("21:9", "16:9", "4:3", "1:1", "3:4", "9:16"),
"resolutions": ("480p", "720p", "1080p"),
"durations": (4, 15),
"audio": True,
"negative": False,
},
"kling-v3-4k": {
"display": "Kling v3 4K",
"speed": "~120-300s",
"price": "premium",
"strengths": "4K output, native audio (Chinese/English), 3-15s.",
"tier": "premium",
"text_endpoint": "fal-ai/kling-video/v3/4k/text-to-video",
"image_endpoint": "fal-ai/kling-video/v3/4k/image-to-video",
# Kling 4K image-to-video uses `start_image_url` instead of
# `image_url`. Handled in _build_payload via image_param_key.
"image_param_key": "start_image_url",
"aspect_ratios": ("16:9", "9:16", "1:1"),
"resolutions": None, # 4K is implicit
"durations": (3, 15),
"audio": True,
"negative": True,
},
"happy-horse": {
"display": "Happy Horse 1.0",
"speed": "~60-120s",
"price": "premium",
"strengths": "Alibaba. New model, sparse public docs — conservative defaults.",
"tier": "premium",
"text_endpoint": "alibaba/happy-horse/text-to-video",
"image_endpoint": "alibaba/happy-horse/image-to-video",
# Docs don't expose duration/aspect/resolution — let the endpoint
# apply its own defaults.
"aspect_ratios": None,
"resolutions": None,
"durations": None,
"audio": False,
"negative": False,
},
}
DEFAULT_MODEL = "pixverse-v6" # cheap, both modalities, sane defaults
def _is_duration_range(durations: Any) -> bool:
"""Heuristic: a 2-tuple of ints with a gap > 1 is treated as ``(min, max)``."""
if not isinstance(durations, tuple) or len(durations) != 2:
return False
if not all(isinstance(d, int) for d in durations):
return False
return durations[1] - durations[0] > 1
def _clamp_duration(family: Dict[str, Any], duration: Optional[int]) -> Optional[int]:
durations = family.get("durations")
if not durations:
return duration
if duration is None:
return durations[0]
if _is_duration_range(durations):
lo, hi = durations
return max(lo, min(hi, duration))
# enum
if duration in durations:
return duration
return min(durations, key=lambda d: abs(d - duration))
# ---------------------------------------------------------------------------
# Config / model resolution
# ---------------------------------------------------------------------------
def _load_video_gen_section() -> Dict[str, Any]:
try:
from hermes_cli.config import load_config
cfg = load_config()
section = cfg.get("video_gen") if isinstance(cfg, dict) else None
return section if isinstance(section, dict) else {}
except Exception as exc:
logger.debug("Could not load video_gen config: %s", exc)
return {}
def _resolve_family(explicit: Optional[str]) -> Tuple[str, Dict[str, Any]]:
"""Decide which FAL family to use. Returns ``(family_id, meta)``."""
candidates: List[Optional[str]] = []
candidates.append(explicit)
candidates.append(os.environ.get("FAL_VIDEO_MODEL"))
cfg = _load_video_gen_section()
fal_cfg = cfg.get("fal") if isinstance(cfg.get("fal"), dict) else {}
if isinstance(fal_cfg, dict):
candidates.append(fal_cfg.get("model"))
top = cfg.get("model")
if isinstance(top, str):
candidates.append(top)
for c in candidates:
if isinstance(c, str) and c.strip() and c.strip() in FAL_FAMILIES:
fid = c.strip()
return fid, FAL_FAMILIES[fid]
return DEFAULT_MODEL, FAL_FAMILIES[DEFAULT_MODEL]
# ---------------------------------------------------------------------------
# Payload construction
# ---------------------------------------------------------------------------
def _build_payload(
family: Dict[str, Any],
*,
prompt: str,
image_url: Optional[str],
duration: Optional[int],
aspect_ratio: str,
resolution: str,
negative_prompt: Optional[str],
audio: Optional[bool],
seed: Optional[int],
) -> Dict[str, Any]:
"""Build a family-specific payload, dropping keys the family doesn't declare."""
payload: Dict[str, Any] = {}
if prompt:
payload["prompt"] = prompt
if image_url:
# Some endpoints (e.g. Kling v3 4K image-to-video) expect
# `start_image_url` instead of `image_url`. The family entry can
# declare an override.
key = family.get("image_param_key") or "image_url"
payload[key] = image_url
if seed is not None:
payload["seed"] = seed
if family.get("aspect_ratios"):
if aspect_ratio in family["aspect_ratios"]:
payload["aspect_ratio"] = aspect_ratio
# otherwise let the endpoint auto-crop / use its default
if family.get("resolutions"):
if resolution in family["resolutions"]:
payload["resolution"] = resolution
# else: let the endpoint default
clamped = _clamp_duration(family, duration)
if clamped is not None and family.get("durations"):
# FAL exposes duration as a string in the queue API ("8" not 8).
# Some families (e.g. veo3.1) require a unit suffix ("4s" not "4").
suffix = family.get("duration_suffix", "")
payload["duration"] = f"{clamped}{suffix}"
if family.get("audio") and audio is not None:
payload["generate_audio"] = bool(audio)
if family.get("negative") and negative_prompt:
payload["negative_prompt"] = negative_prompt
return payload
# ---------------------------------------------------------------------------
# fal_client lazy import (shared with image_generation_tool via fal_common)
# ---------------------------------------------------------------------------
_fal_client: Any = None
def _load_fal_client() -> Any:
"""Lazy-load the ``fal_client`` SDK and cache it on this module.
Delegates the actual import to :func:`tools.fal_common.import_fal_client`
so the ``lazy_deps`` ensure-install handling stays in one place.
"""
global _fal_client
if _fal_client is not None:
return _fal_client
from tools.fal_common import import_fal_client
_fal_client = import_fal_client()
return _fal_client
# ---------------------------------------------------------------------------
# Managed FAL gateway (Nous Subscription)
# ---------------------------------------------------------------------------
_managed_fal_video_client: Any = None
_managed_fal_video_client_config: Any = None
_managed_fal_video_client_lock = threading.Lock()
def _resolve_managed_fal_video_gateway():
"""Return managed fal-queue gateway config when the user prefers the gateway
or direct FAL credentials are absent."""
from tools.tool_backend_helpers import fal_key_is_configured, prefers_gateway
if fal_key_is_configured() and not prefers_gateway("video_gen"):
return None
from tools.managed_tool_gateway import resolve_managed_tool_gateway
return resolve_managed_tool_gateway("fal-queue")
def _get_managed_fal_video_client(managed_gateway):
"""Reuse the managed FAL client so its internal httpx.Client is not leaked per call."""
global _managed_fal_video_client, _managed_fal_video_client_config
from tools.fal_common import _ManagedFalSyncClient
client_config = (
managed_gateway.gateway_origin.rstrip("/"),
managed_gateway.nous_user_token,
)
with _managed_fal_video_client_lock:
if _managed_fal_video_client is not None and _managed_fal_video_client_config == client_config:
return _managed_fal_video_client
_load_fal_client()
_managed_fal_video_client = _ManagedFalSyncClient(
_fal_client,
key=managed_gateway.nous_user_token,
queue_run_origin=managed_gateway.gateway_origin,
)
_managed_fal_video_client_config = client_config
return _managed_fal_video_client
def _submit_fal_video_request(endpoint: str, arguments: Dict[str, Any]):
"""Submit a FAL video request using direct credentials or the managed queue gateway.
Returns a request handle whose ``.get()`` blocks until the result is ready.
"""
_load_fal_client()
request_headers = {"x-idempotency-key": str(uuid.uuid4())}
managed_gateway = _resolve_managed_fal_video_gateway()
if managed_gateway is None:
return _fal_client.submit(endpoint, arguments=arguments, headers=request_headers)
managed_client = _get_managed_fal_video_client(managed_gateway)
try:
return managed_client.submit(
endpoint,
arguments=arguments,
headers=request_headers,
)
except Exception as exc:
from tools.fal_common import _extract_http_status
status = _extract_http_status(exc)
if status is not None and 400 <= status < 500:
raise ValueError(
f"Nous Subscription gateway rejected endpoint '{endpoint}' "
f"(HTTP {status}). This model may not yet be enabled on "
f"the Nous Portal's FAL proxy. Either:\n"
f" • Set FAL_KEY in your environment to use FAL.ai directly, or\n"
f" • Pick a different model via `hermes tools` → Video Generation."
) from exc
raise
def _check_fal_video_available() -> bool:
"""True if the FAL.ai video backend is reachable (direct key or managed gateway)."""
from tools.tool_backend_helpers import fal_key_is_configured
if fal_key_is_configured():
return True
return _resolve_managed_fal_video_gateway() is not None
# ---------------------------------------------------------------------------
# Provider
# ---------------------------------------------------------------------------
class FALVideoGenProvider(VideoGenProvider):
"""FAL.ai multi-family video generation backend.
Routes between text-to-video and image-to-video endpoints automatically
based on whether ``image_url`` was provided.
"""
@property
def name(self) -> str:
return "fal"
@property
def display_name(self) -> str:
return "FAL"
def is_available(self) -> bool:
try:
return _check_fal_video_available()
except Exception: # noqa: BLE001 — never break the picker
return False
def list_models(self) -> List[Dict[str, Any]]:
out: List[Dict[str, Any]] = []
for fid, meta in FAL_FAMILIES.items():
modalities: List[str] = []
if meta.get("text_endpoint"):
modalities.append("text")
if meta.get("image_endpoint"):
modalities.append("image")
out.append({
"id": fid,
"display": meta["display"],
"speed": meta["speed"],
"strengths": meta["strengths"],
"price": meta["price"],
"tier": meta.get("tier", "premium"),
"modalities": modalities,
})
return out
def default_model(self) -> Optional[str]:
return DEFAULT_MODEL
def get_setup_schema(self) -> Dict[str, Any]:
return {
"name": "FAL",
"badge": "paid",
"tag": "LTX, Pixverse, Veo 3.1, Seedance 2.0, Kling 4K, Happy Horse — text-to-video & image-to-video",
"env_vars": [
{
"key": "FAL_KEY",
"prompt": "FAL.ai API key",
"url": "https://fal.ai/dashboard/keys",
},
],
}
def capabilities(self) -> Dict[str, Any]:
return {
"modalities": ["text", "image"],
"aspect_ratios": ["16:9", "9:16", "1:1"],
"resolutions": ["360p", "540p", "720p", "1080p"],
"max_duration": 15,
"min_duration": 1,
"supports_audio": True,
"supports_negative_prompt": True,
"max_reference_images": 0,
}
def generate(
self,
prompt: str,
*,
model: Optional[str] = None,
image_url: Optional[str] = None,
reference_image_urls: Optional[List[str]] = None,
duration: Optional[int] = None,
aspect_ratio: str = "16:9",
resolution: str = "720p",
negative_prompt: Optional[str] = None,
audio: Optional[bool] = None,
seed: Optional[int] = None,
**kwargs: Any,
) -> Dict[str, Any]:
if not _check_fal_video_available():
return error_response(
error=(
"No FAL backend available. Either set FAL_KEY "
"(run `hermes tools` → Video Generation → FAL to configure) "
"or sign in to Nous (`hermes setup`) for managed gateway access."
),
error_type="auth_required",
provider="fal",
prompt=prompt,
)
try:
_load_fal_client()
except ImportError:
return error_response(
error="fal_client Python package not installed (pip install fal-client)",
error_type="missing_dependency",
provider="fal",
prompt=prompt,
)
prompt = (prompt or "").strip()
family_id, family = _resolve_family(model)
# Route: image_url → image-to-video endpoint; else → text-to-video.
image_url_norm = (image_url or "").strip() or None
if image_url_norm:
endpoint = family.get("image_endpoint")
modality_used = "image"
if not endpoint:
return error_response(
error=(
f"FAL family {family_id} has no image-to-video "
f"endpoint. Pick a family with image-to-video support "
f"via `hermes tools` → Video Generation."
),
error_type="modality_unsupported",
provider="fal", model=family_id, prompt=prompt,
)
else:
endpoint = family.get("text_endpoint")
modality_used = "text"
if not endpoint:
return error_response(
error=(
f"FAL family {family_id} has no text-to-video "
f"endpoint. Pass an image_url to use its "
f"image-to-video endpoint, or pick a different family."
),
error_type="modality_unsupported",
provider="fal", model=family_id, prompt=prompt,
)
if not prompt:
return error_response(
error="prompt is required.",
error_type="missing_prompt",
provider="fal", model=family_id, prompt=prompt,
)
payload = _build_payload(
family,
prompt=prompt,
image_url=image_url_norm,
duration=duration,
aspect_ratio=aspect_ratio,
resolution=resolution,
negative_prompt=negative_prompt,
audio=audio,
seed=seed,
)
try:
handle = _submit_fal_video_request(endpoint, payload)
result = handle.get()
except Exception as exc:
logger.warning(
"FAL video gen failed (family=%s, endpoint=%s): %s",
family_id, endpoint, exc, exc_info=True,
)
return error_response(
error=f"FAL video generation failed: {exc}",
error_type="api_error",
provider="fal", model=family_id, prompt=prompt,
aspect_ratio=aspect_ratio,
)
video = (result or {}).get("video") if isinstance(result, dict) else None
url: Optional[str] = None
if isinstance(video, dict):
url = video.get("url")
elif isinstance(video, str):
url = video
if not url:
return error_response(
error="FAL returned no video URL in response",
error_type="empty_response",
provider="fal", model=family_id, prompt=prompt,
)
extra: Dict[str, Any] = {"endpoint": endpoint}
if isinstance(video, dict):
if video.get("file_size"):
extra["file_size"] = video["file_size"]
if video.get("content_type"):
extra["content_type"] = video["content_type"]
return success_response(
video=url,
model=family_id,
prompt=prompt,
modality=modality_used,
aspect_ratio=aspect_ratio if "aspect_ratio" in payload else "",
duration=int(payload["duration"]) if "duration" in payload else 0,
provider="fal",
extra=extra,
)
# ---------------------------------------------------------------------------
# Plugin entry point
# ---------------------------------------------------------------------------
def register(ctx) -> None:
"""Plugin entry point — wire ``FALVideoGenProvider`` into the registry."""
ctx.register_video_gen_provider(FALVideoGenProvider())
Reference in New Issue View Git Blame Copy Permalink