freedak/OpenMontage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Higgsfield provider and update Runway to v0.2.0

Browse Source 
- New: Higgsfield video provider with multi-model routing (Kling 3.0, Veo 3.1, Sora 2, WAN 2.5, Soul Cinema) and Soul ID character consistency
- Updated: Runway provider with gen4_aleph and gen3a_turbo models, proper pixel-ratio mapping, probe_output, watermark param, RUNWAYML_API_SECRET env var support
- Docs: Updated provider counts (12→13), tool counts (51→52), added Higgsfield setup/pricing sections across README, AGENT_GUIDE, ARCHITECTURE, and PROVIDERS
This commit is contained in:
calesthio
2026-04-08 12:57:43 -07:00
parent b1a078b282
commit 4f682c8b0a
6 changed files with 373 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/video/higgsfield_video.py
+243
View File
@@ -0,0 +1,243 @@
"""Higgsfield video generation via Higgsfield Cloud API.
Multi-model orchestrator with proprietary Soul model for character-consistent,
photorealistic video generation. Routes to Kling, Veo, Sora, and WAN under the hood.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class HiggsFieldVideo(BaseTool):
name = "higgsfield_video"
version = "0.1.0"
tier = ToolTier.GENERATE
capability = "video_generation"
provider = "higgsfield"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.SYNC
determinism = Determinism.STOCHASTIC
runtime = ToolRuntime.API
dependencies = []
install_instructions = (
"Set HIGGSFIELD_API_KEY and HIGGSFIELD_API_SECRET for your Higgsfield Cloud credentials.\n"
" Get them at https://cloud.higgsfield.ai/api-keys\n"
" Alternatively, set HIGGSFIELD_KEY as a combined key:secret value."
)
agent_skills = ["ai-video-gen"]
capabilities = ["text_to_video", "image_to_video"]
supports = {
"text_to_video": True,
"image_to_video": True,
"character_consistency": True,
"multi_model_routing": True,
}
best_for = [
"character-consistent video generation (Soul ID)",
"multi-model access through a single API",
"photorealistic and fashion-aware content",
]
not_good_for = ["offline generation", "fine-grained model control", "budget projects without subscription"]
fallback_tools = ["kling_video", "veo_video", "minimax_video"]
input_schema = {
"type": "object",
"required": ["prompt"],
"properties": {
"prompt": {"type": "string"},
"operation": {
"type": "string",
"enum": ["text_to_video", "image_to_video"],
"default": "text_to_video",
},
"model": {
"type": "string",
"enum": [
"kling_3.0",
"veo_3.1",
"sora_2",
"wan_2.5",
"soul_cinema",
],
"default": "kling_3.0",
"description": "Underlying model to use for generation",
},
"duration": {
"type": "string",
"enum": ["5", "10", "15"],
"default": "5",
"description": "Duration in seconds (availability varies by model)",
},
"aspect_ratio": {
"type": "string",
"enum": ["16:9", "9:16", "1:1", "21:9"],
"default": "16:9",
},
"image_url": {"type": "string", "description": "Reference image URL for image_to_video"},
"output_path": {"type": "string"},
},
}
resource_profile = ResourceProfile(
cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True
)
retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"])
idempotency_key_fields = ["prompt", "model", "operation", "duration"]
side_effects = ["writes video file to output_path", "calls Higgsfield Cloud API"]
user_visible_verification = ["Watch generated clip for motion coherence and visual quality"]
def _get_credentials(self) -> tuple[str, str] | None:
"""Return (api_key, api_secret) or None if not configured."""
combined = os.environ.get("HIGGSFIELD_KEY")
if combined and ":" in combined:
key, secret = combined.split(":", 1)
return key, secret
key = os.environ.get("HIGGSFIELD_API_KEY")
secret = os.environ.get("HIGGSFIELD_API_SECRET")
if key and secret:
return key, secret
return None
def get_status(self) -> ToolStatus:
if self._get_credentials():
return ToolStatus.AVAILABLE
return ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
model = inputs.get("model", "kling_3.0")
duration = int(inputs.get("duration", "5"))
# Approximate per-clip costs based on Higgsfield credit pricing
base_costs = {
"kling_3.0": 0.10,
"wan_2.5": 0.10,
"veo_3.1": 0.50,
"sora_2": 0.50,
"soul_cinema": 0.15,
}
base = base_costs.get(model, 0.15)
return base * (duration / 5)
def estimate_runtime(self, inputs: dict[str, Any]) -> float:
model = inputs.get("model", "kling_3.0")
if model in ("veo_3.1", "sora_2"):
return 120.0
return 60.0
def execute(self, inputs: dict[str, Any]) -> ToolResult:
creds = self._get_credentials()
if not creds:
return ToolResult(
success=False,
error="Higgsfield credentials not set. " + self.install_instructions,
)
import requests
api_key, api_secret = creds
start = time.time()
operation = inputs.get("operation", "text_to_video")
model = inputs.get("model", "kling_3.0")
payload: dict[str, Any] = {
"prompt": inputs["prompt"],
"model": model,
"task": operation.replace("_", "-"),
}
if inputs.get("duration"):
payload["duration"] = int(inputs["duration"])
if inputs.get("aspect_ratio"):
payload["aspect_ratio"] = inputs["aspect_ratio"]
if operation == "image_to_video" and inputs.get("image_url"):
payload["image_url"] = inputs["image_url"]
headers = {
"Authorization": f"Bearer {api_key}",
"X-API-Secret": api_secret,
"Content-Type": "application/json",
}
try:
# Submit generation request
submit_resp = requests.post(
"https://platform.higgsfield.ai/v1/generations",
headers=headers,
json=payload,
timeout=30,
)
submit_resp.raise_for_status()
gen_data = submit_resp.json()
generation_id = gen_data["id"]
status_url = gen_data.get("status_url", f"https://platform.higgsfield.ai/v1/generations/{generation_id}")
# Poll for completion
video_url = None
for _ in range(72): # max ~6 minutes
time.sleep(5)
poll_resp = requests.get(status_url, headers=headers, timeout=15)
poll_resp.raise_for_status()
poll_data = poll_resp.json()
status = poll_data.get("status", "Unknown")
if status in ("Completed", "COMPLETED"):
video_url = poll_data.get("output_url") or poll_data.get("url")
break
if status in ("Failed", "FAILED", "NSFW", "Cancelled", "CANCELLED"):
return ToolResult(
success=False,
error=f"Higgsfield generation {status}: {poll_data.get('error', 'unknown')}",
)
if not video_url:
return ToolResult(success=False, error="Higgsfield generation timed out.")
# Download video
video_response = requests.get(video_url, timeout=120)
video_response.raise_for_status()
output_path = Path(inputs.get("output_path", "higgsfield_output.mp4"))
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(video_response.content)
except Exception as e:
return ToolResult(success=False, error=f"Higgsfield video generation failed: {e}")
from tools.video._shared import probe_output
probed = probe_output(output_path)
return ToolResult(
success=True,
data={
"provider": "higgsfield",
"model": model,
"prompt": inputs["prompt"],
"operation": operation,
"aspect_ratio": inputs.get("aspect_ratio", "16:9"),
"output": str(output_path),
"output_path": str(output_path),
"format": "mp4",
**probed,
},
artifacts=[str(output_path)],
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - start, 2),
model=model,
)
tools/video/runway_video.py
+69 -24
View File
@@ -1,6 +1,7 @@
"""Runway Gen-4 video generation via Runway API.
Highest Elo-rated video generation model — professional quality and control.
Supports Gen-3 Alpha Turbo, Gen-4 Turbo, and Gen-4 Aleph (highest fidelity).
"""
from __future__ import annotations
@@ -23,22 +24,40 @@ from tools.base_tool import (
ToolTier,
)
_RATIO_MAP = {
"16:9": "1280:720",
"9:16": "720:1280",
"1:1": "720:720",
}
_COST_PER_SECOND = {
"gen3a_turbo": 0.05,
"gen4_turbo": 0.05,
"gen4_aleph": 0.15,
}
_RUNTIME_SECONDS = {
"gen3a_turbo": 25.0,
"gen4_turbo": 30.0,
"gen4_aleph": 60.0,
}
class RunwayVideo(BaseTool):
name = "runway_video"
version = "0.1.0"
version = "0.2.0"
tier = ToolTier.GENERATE
capability = "video_generation"
provider = "runway"
stability = ToolStability.EXPERIMENTAL
stability = ToolStability.BETA
execution_mode = ExecutionMode.SYNC
determinism = Determinism.STOCHASTIC
runtime = ToolRuntime.API
dependencies = []
install_instructions = (
"Set RUNWAY_API_KEY to your Runway API key.\n"
" Get one at https://app.runwayml.com/settings/api-keys"
"Set RUNWAY_API_KEY to your Runway API secret.\n"
" Get one at https://dev.runwayml.com/"
)
agent_skills = ["ai-video-gen"]
@@ -68,8 +87,9 @@ class RunwayVideo(BaseTool):
},
"model": {
"type": "string",
"enum": ["gen4_turbo", "gen4"],
"enum": ["gen4_turbo", "gen4_aleph", "gen3a_turbo"],
"default": "gen4_turbo",
"description": "gen4_aleph is highest fidelity, gen4_turbo is balanced, gen3a_turbo is cheapest",
},
"duration": {
"type": "integer",
@@ -82,6 +102,11 @@ class RunwayVideo(BaseTool):
"enum": ["16:9", "9:16", "1:1"],
"default": "16:9",
},
"watermark": {
"type": "boolean",
"default": False,
"description": "Include Runway watermark on output",
},
"image_url": {"type": "string", "description": "Reference image URL for image_to_video"},
"output_path": {"type": "string"},
},
@@ -90,29 +115,30 @@ class RunwayVideo(BaseTool):
resource_profile = ResourceProfile(
cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True
)
retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"])
retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout", "THROTTLED"])
idempotency_key_fields = ["prompt", "model", "operation", "duration"]
side_effects = ["writes video file to output_path", "calls Runway API"]
user_visible_verification = ["Watch generated clip for visual quality and motion coherence"]
def get_status(self) -> ToolStatus:
if os.environ.get("RUNWAY_API_KEY"):
if os.environ.get("RUNWAY_API_KEY") or os.environ.get("RUNWAYML_API_SECRET"):
return ToolStatus.AVAILABLE
return ToolStatus.UNAVAILABLE
def _get_api_key(self) -> str | None:
return os.environ.get("RUNWAY_API_KEY") or os.environ.get("RUNWAYML_API_SECRET")
def estimate_cost(self, inputs: dict[str, Any]) -> float:
model = inputs.get("model", "gen4_turbo")
duration = inputs.get("duration", 5)
# Runway charges per second of generated video
return 0.05 * duration # ~$0.25 for 5s, ~$0.50 for 10s
return _COST_PER_SECOND.get(model, 0.05) * duration
def estimate_runtime(self, inputs: dict[str, Any]) -> float:
model = inputs.get("model", "gen4_turbo")
if "turbo" in model:
return 30.0
return 60.0
return _RUNTIME_SECONDS.get(model, 30.0)
def execute(self, inputs: dict[str, Any]) -> ToolResult:
api_key = os.environ.get("RUNWAY_API_KEY")
api_key = self._get_api_key()
if not api_key:
return ToolResult(
success=False,
@@ -124,17 +150,26 @@ class RunwayVideo(BaseTool):
start = time.time()
model = inputs.get("model", "gen4_turbo")
operation = inputs.get("operation", "text_to_video")
ratio_friendly = inputs.get("ratio", "16:9")
ratio_pixels = _RATIO_MAP.get(ratio_friendly, "1280:720")
# Runway API v1 — submit task
task_payload: dict[str, Any] = {
"model": model,
"promptText": inputs["prompt"],
"duration": inputs.get("duration", 5),
"ratio": inputs.get("ratio", "16:9"),
"ratio": ratio_pixels,
"watermark": inputs.get("watermark", False),
}
if operation == "image_to_video" and inputs.get("image_url"):
task_payload["promptImage"] = inputs["image_url"]
# Choose endpoint based on operation
endpoint = (
"https://api.dev.runwayml.com/v1/image_to_video"
if operation == "image_to_video"
else "https://api.dev.runwayml.com/v1/text_to_video"
)
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
@@ -144,8 +179,7 @@ class RunwayVideo(BaseTool):
try:
# Submit generation task
submit_response = requests.post(
"https://api.dev.runwayml.com/v1/image_to_video" if operation == "image_to_video"
else "https://api.dev.runwayml.com/v1/text_to_video",
endpoint,
headers=headers,
json=task_payload,
timeout=30,
@@ -153,9 +187,9 @@ class RunwayVideo(BaseTool):
submit_response.raise_for_status()
task_id = submit_response.json()["id"]
# Poll for completion
# Poll for completion (max ~5 minutes)
video_url = None
for _ in range(60): # max 5 minutes
for _ in range(60):
time.sleep(5)
poll_response = requests.get(
f"https://api.dev.runwayml.com/v1/tasks/{task_id}",
@@ -164,20 +198,23 @@ class RunwayVideo(BaseTool):
)
poll_response.raise_for_status()
task_data = poll_response.json()
status = task_data["status"]
if task_data["status"] == "SUCCEEDED":
if status == "SUCCEEDED":
video_url = task_data["output"][0]
break
if task_data["status"] == "FAILED":
if status == "FAILED":
failure_code = task_data.get("failureCode", "unknown")
return ToolResult(
success=False,
error=f"Runway generation failed: {task_data.get('failure', 'unknown error')}",
error=f"Runway generation failed ({failure_code}): {task_data.get('failure', 'unknown error')}",
)
# PENDING, THROTTLED, RUNNING — keep polling
if not video_url:
return ToolResult(success=False, error="Runway generation timed out.")
return ToolResult(success=False, error="Runway generation timed out after 5 minutes.")
# Download video
# Download video — URLs are ephemeral (expire in 24-48h)
video_response = requests.get(video_url, timeout=120)
video_response.raise_for_status()
@@ -188,14 +225,22 @@ class RunwayVideo(BaseTool):
except Exception as e:
return ToolResult(success=False, error=f"Runway video generation failed: {e}")
from tools.video._shared import probe_output
probed = probe_output(output_path)
return ToolResult(
success=True,
data={
"provider": "runway",
"model": model,
"prompt": inputs["prompt"],
"operation": operation,
"ratio": ratio_friendly,
"output": str(output_path),
"output_path": str(output_path),
"task_id": task_id,
"format": "mp4",
**probed,
},
artifacts=[str(output_path)],
cost_usd=self.estimate_cost(inputs),