Build a Media-Generation Agent Workflow with Veo 3.1 & Lyria 3.5

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Google's July–August 2026 launches — Veo 3.1 for video and Lyria 3.5 for music — moved AI media generation from "demo" to "production pipeline" territory. Veo 3.1 sharpens temporal consistency and extends shot-level control. Lyria 3.5 lands with richer melodies, structured lyrics, expressive vocals, and — critically for automation — tempo and duration control. Both are callable APIs, which means you can wire them into an agent workflow instead of generating media by hand.

This dispatch builds a media-generation agent: a LangGraph pipeline that takes a creative brief, produces a script, derives a Lyria 3.5 music spec (tempo, key, duration, lyrics structure), generates video segments with Veo 3.1, then renders the final asset with ffmpeg — with a human-in-the-loop review gate and explicit generation retry semantics.

The Pipeline at a Glance

The pipeline is linear in spirit but parallel in execution. Music and video generation are independent once the script exists, so LangGraph fans the work out:

                  ┌──────────────┐
                  │ Creative Brief│
                  └──────┬───────┘
                         ▼
                  ┌──────────────┐
                  │  Script Gen   │
                  └──────┬───────┘
            ┌────────────┴────────────┐
            ▼                         ▼
   ┌─────────────────┐      ┌─────────────────┐
   │  Lyria 3.5       │      │  Veo 3.1         │
   │  music spec      │      │  video specs     │
   │  tempo/duration  │      │  per-scene shots │
   └────────┬────────┘      └────────┬────────┘
            └───────────┬────────────┘
                        ▼
               ┌─────────────────┐
               │ Human Review     │
               │ (conditional)    │
               └────────┬────────┘
                  pass  │  reject
                        ▼
               ┌─────────────────┐
               │  ffmpeg mux      │
               │  audio + video   │
               └────────┬────────┘
                        ▼
                  Render Manifest

This shape — plan in parallel, generate in parallel, gate, then composite — is the standard media-agent topology in 2026. For the wider catalog of media and agent workflows, check the Daily AI World workflows library.

Why Schema Discipline Matters Here

Veo and Lyria are generation APIs, not text LLMs. Garbage in, expensive garbage out. The entire workflow hinges on the Pydantic specs, because Lyria 3.5's tempo and duration control only pays off if the orchestrator asks for a real structure — "122 BPM, 2:45, verse–chorus–bridge" — instead of a vague sentence. Veo 3.1 behaves the same way: a tightly scoped prompt with an explicit aspect ratio and duration beats a poetic paragraph every time.

Environment Configuration

.env holds Google credentials and render knobs:

# .env
GOOGLE_API_KEY=AIza...
GOOGLE_PROJECT_ID=media-ops

LYRIA_ENDPOINT=https://lyria.googleapis.com/v3.5
LYRIA_MODEL=lyria-3.5-preview

VEO_ENDPOINT=https://veo.googleapis.com/v3.1
VEO_MODEL=veo-3.1

FFMPEG_BIN=/opt/homebrew/bin/ffmpeg
OUTPUT_DIR=./renders

MAX_GEN_POLL_SECONDS=900
GEN_POLL_INTERVAL_SECONDS=10
MUSIC_RETRIES=3
VIDEO_RETRIES=4

Schemas

schemas.py models brief, script, specs, jobs, and the final manifest:

# schemas.py
from __future__ import annotations

from typing import Literal, Optional

from langgraph.graph import MessagesState
from pydantic import BaseModel, Field


class Scene(BaseModel):
    number: int
    summary: str
    shot_description: str
    aspect: Literal["16:9", "9:16", "1:1"] = "16:9"
    duration_seconds: int = 8


class Script(BaseModel):
    title: str
    scenes: list[Scene] = Field(default_factory=list)


class MusicSpec(BaseModel):
    tempo_bpm: int = Field(default=122, ge=60, le=200)
    key: str = "C major"
    duration_seconds: int = Field(default=165, ge=10, le=300)
    structure: list[str] = Field(
        default_factory=lambda: ["intro", "verse", "chorus", "bridge", "outro"]
    )
    lyrics_mode: Literal["none", "structured", "full_vocal"] = "structured"
    lyrics: list[str] = Field(default_factory=list)


class VideoSpec(BaseModel):
    scene: Scene
    prompt: str
    fps: int = 24
    motion: Literal["natural", "cinematic", "static"] = "cinematic"


class GenerationJob(BaseModel):
    id: str
    kind: Literal["music", "video"]
    status: Literal["queued", "running", "succeeded", "failed"] = "queued"
    output_url: Optional[str] = None
    checksum: Optional[str] = None
    attempts: int = 0


class RenderManifest(BaseModel):
    video_url: str
    music_url: str
    final_url: str
    duration_seconds: int
    ffmpeg_cmd: str


class MediaState(MessagesState):
    brief: str
    script: Optional[Script] = None
    music: Optional[MusicSpec] = None
    video_specs: list[VideoSpec] = Field(default_factory=list)
    jobs: list[GenerationJob] = Field(default_factory=list)
    manifest: Optional[RenderManifest] = None
    review_passed: bool = False

The structure and tempo_bpm fields are the reason Lyria 3.5 is production-safe: the orchestrator can schedule music precisely against video length instead of hoping a random loop matches.

Tools: Lyria, Veo, and the Mux

tools.py wraps the generation and render calls. Generation jobs are long-running, so every tool submits and then polls the operations API:

# tools.py
from __future__ import annotations

import hashlib
import json
import os
import subprocess
import time

import httpx
from langchain_core.tools import tool

from schemas import GenerationJob, MusicSpec, VideoSpec

GOOGLE_KEY = os.getenv("GOOGLE_API_KEY", "")
LYRIA = os.getenv("LYRIA_ENDPOINT", "https://lyria.googleapis.com/v3.5")
VEO = os.getenv("VEO_ENDPOINT", "https://veo.googleapis.com/v3.1")
POLL_INTERVAL = float(os.getenv("GEN_POLL_INTERVAL_SECONDS", "10"))
MAX_POLL = float(os.getenv("MAX_GEN_POLL_SECONDS", "900"))


def _headers() -> dict:
    return {"x-goog-api-key": GOOGLE_KEY, "Content-Type": "application/json"}


def _submit(base: str, body: dict) -> dict:
    resp = httpx.post(f"{base}/generate", json=body, headers=_headers(), timeout=60)
    resp.raise_for_status()
    return resp.json()


def _wait_for(base: str, op_name: str, kind: str) -> dict:
    elapsed = 0.0
    while elapsed < MAX_POLL:
        time.sleep(POLL_INTERVAL)
        elapsed += POLL_INTERVAL
        resp = httpx.get(f"{base}/operations/{op_name}", headers=_headers(), timeout=30)
        resp.raise_for_status()
        op = resp.json()
        if op.get("done"):
            return op
    raise TimeoutError(f"{kind} job {op_name} exceeded {MAX_POLL}s")


@tool
def generate_music(spec: MusicSpec) -> GenerationJob:
    # Generate a music track with Lyria 3.5 (tempo, duration, structure).
    body = {
        "instances": [
            {
                "spec": {
                    "tempo_bpm": spec.tempo_bpm,
                    "key": spec.key,
                    "duration_seconds": spec.duration_seconds,
                    "structure": spec.structure,
                    "lyrics": {"mode": spec.lyrics_mode, "lines": spec.lyrics},
                }
            }
        ]
    }
    op = _submit(LYRIA, body)
    job = _wait_for(LYRIA, op["name"], "music")
    uri = job["response"]["generated_artifacts"][0]["uri"]
    return GenerationJob(id=op["name"], kind="music", status="succeeded",
                         output_url=uri, attempts=1)


@tool
def generate_video(spec: VideoSpec) -> GenerationJob:
    # Generate one video clip with Veo 3.1.
    seed = int.from_bytes(
        hashlib.sha256(spec.prompt.encode()).digest()[:4], "big"
    ) % 100000
    body = {
        "instances": [
            {
                "prompt": spec.prompt,
                "aspect_ratio": spec.scene.aspect,
                "duration_seconds": spec.scene.duration_seconds,
                "fps": spec.fps,
            }
        ],
        "parameters": {"motion": spec.motion, "seed": seed},
    }
    op = _submit(VEO, body)
    job = _wait_for(VEO, op["name"], "video")
    uri = job["response"]["generated_artifacts"][0]["uri"]
    return GenerationJob(id=op["name"], kind="video", status="succeeded",
                         output_url=uri, attempts=1)


@tool
def mux(video_url: str, music_url: str, duration_seconds: int) -> dict:
    # Download and mux audio + video with ffmpeg; return checksum and path.
    ff = os.getenv("FFMPEG_BIN", "ffmpeg")
    out = os.path.join(os.getenv("OUTPUT_DIR", "./renders"), "final.mp4")
    cmd = [
        ff, "-y", "-i", video_url, "-i", music_url,
        "-t", str(duration_seconds),
        "-c:v", "copy", "-c:a", "aac", "-shortest", out,
    ]
    proc = subprocess.run(cmd, capture_output=True)
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr.decode()[-500:])
    digest = hashlib.sha256(open(out, "rb").read()).hexdigest()
    return {"path": out, "checksum": digest}

A note on multi-clip videos: for several Veo clips, concatenate them with ffmpeg's concat demuxer before muxing audio. The graph below uses a single hero clip for clarity; the manifest and retry rules assume you may extend it.

Graph: Brief → Script → Parallel Generation → Review → Mux

graph.py orchestrates the pipeline. The script node is the only LLM call; everything else is deterministic tool execution:

# graph.py
from __future__ import annotations

import json

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, StateGraph

from schemas import MediaState, MusicSpec, RenderManifest, Scene, Script, VideoSpec
from tools import generate_music, generate_video, mux


class ScriptWriter:
    def __init__(self) -> None:
        self.llm = ChatOpenAI(model="gpt-5.2", temperature=0.7)

    def __call__(self, state: MediaState) -> dict:
        prompt = (
            "Turn this brief into a JSON script with exactly 4 scenes: "
            '{"title": "...", "scenes": [{"number", "summary", '
            '"shot_description", "aspect", "duration_seconds"}]}
'
            f"Brief: {state.brief}"
        )
        raw = self.llm.invoke(prompt).content.strip()
        raw = raw.removeprefix("```json").removesuffix("```").strip()
        data = json.loads(raw)
        return {
            "script": Script(
                title=data["title"],
                scenes=[Scene(**s) for s in data["scenes"]],
            )
        }


def _plan_music(state: MediaState) -> dict:
    total = sum(s.duration_seconds for s in state.script.scenes) + 8
    spec = MusicSpec(
        tempo_bpm=122,
        duration_seconds=min(total, 300),
        structure=["intro", "verse", "chorus", "verse", "chorus", "outro"],
        lyrics_mode="none",
    )
    return {"music": spec}


def _run_music(state: MediaState) -> dict:
    job = generate_music.invoke({"spec": state.music})
    return {"jobs": state.jobs + [job]}


def _plan_video(state: MediaState) -> dict:
    specs = [
        VideoSpec(scene=s, prompt=s.shot_description) for s in state.script.scenes
    ]
    return {"video_specs": specs}


def _run_video(state: MediaState) -> dict:
    jobs = list(state.jobs)
    for spec in state.video_specs:
        jobs.append(generate_video.invoke({"spec": spec}))
    return {"jobs": jobs}


def _review(state: MediaState) -> dict:
    # Replace with a real approval webhook or queue in production.
    return {"review_passed": state.review_passed}


def _route_review(state: MediaState) -> str:
    all_done = all(j.status == "succeeded" for j in state.jobs)
    return "mux" if all_done and state.review_passed else "reject"


def _mux(state: MediaState) -> dict:
    music = next(j for j in state.jobs if j.kind == "music")
    video = next(j for j in state.jobs if j.kind == "video")
    out = mux.invoke(
        {
            "video_url": video.output_url,
            "music_url": music.output_url,
            "duration_seconds": state.music.duration_seconds,
        }
    )
    manifest = RenderManifest(
        video_url=video.output_url,
        music_url=music.output_url,
        final_url=out["path"],
        duration_seconds=state.music.duration_seconds,
        ffmpeg_cmd="ffmpeg -y -i video.mp4 -i music.mp3 -c:v copy -c:a aac -shortest final.mp4",
    )
    return {"manifest": manifest}


def build_media_graph() -> StateGraph:
    g = StateGraph(MediaState)
    g.add_node("script", ScriptWriter())
    g.add_node("music_plan", _plan_music)
    g.add_node("music", _run_music)
    g.add_node("video_plan", _plan_video)
    g.add_node("video", _run_video)
    g.add_node("review", _review)
    g.add_node("mux", _mux)

    g.set_entry_point("script")
    g.add_edge("script", "music_plan")
    g.add_edge("music_plan", "music")
    g.add_edge("script", "video_plan")
    g.add_edge("video_plan", "video")
    g.add_edge("music", "review")
    g.add_edge("video", "review")
    g.add_conditional_edges(
        "review", _route_review, {"mux": "mux", "reject": END}
    )
    g.add_edge("mux", END)
    return g

Because music and video both converge on review, the graph naturally runs the two generation branches in parallel — LangGraph executes the video path while Lyria 3.5 is still rendering.

Entry Point

main.py feeds a brief and renders the final asset:

# main.py
import asyncio

from dotenv import load_dotenv

from graph import build_media_graph
from schemas import MediaState

load_dotenv()


async def main() -> None:
    graph = build_media_graph().compile(checkpointer=InMemorySaver())
    state = MediaState(
        brief=(
            "A 40-second cinematic brand spot for a mountain bike. "
            "Dawn light, rider silhouettes, chalky dirt. No voiceover."
        ),
        review_passed=True,  # set False to pause at the human gate
    )
    result = await graph.ainvoke(
        state, config={"configurable": {"thread_id": "media-001"}}
    )
    manifest = result.get("manifest")
    if manifest:
        print("RENDERED:", manifest.final_url)
        print("DURATION:", manifest.duration_seconds, "s")


if __name__ == "__main__":
    asyncio.run(main())

Retry Rules

Media generation is slow, expensive, and eventually consistent — so retry policy is the most important part of the pipeline:

Never let a Veo or Lyria generation run inside a synchronous request timeout — a single Veo 3.1 clip can take minutes. Offload generation to a worker queue and let LangGraph's checkpointer resume the thread when the job completes; the polling loop above is the reference pattern for that.

Cost and Operations

Veo 3.1 bills per generated second and Lyria 3.5 per track, so a single rejection at the review gate is real money. Three operational habits keep budgets sane. First, render cheap previews — run fps=12 and shorter durations for review, then regenerate at full quality only after approval. Second, enforce the review gate — the review_passed flag exists because auto-publishing generated media to production is how brands get into trouble; wire it to a Slack approval or a real human webhook. Third, keep provenance — Veo embeds SynthID watermarks and Lyria output carries generation metadata; store both in the manifest so every asset is verifiable after the fact.

Next Steps

Start with the shortest possible loop: generate one 8-second clip and one 30-second track, mux them, and make the review gate annoying to bypass. Only then scale to multi-scene edits and lyric-vocal modes. Lyria 3.5's structured lyrics and expressive vocals deserve their own workflow pass, and the generation landscape is moving fast — keep an eye on the latest AI news page for Veo 3.1 and Lyria 3.5 changes, and reference more media workflows as the pattern catalog grows.

FAQ

How long does a full Veo 3.1 + Lyria 3.5 pipeline take?

A single clip renders in minutes and a track in under a minute, so the whole pipeline for a 40-second spot typically lands between 5 and 15 minutes when generation runs in parallel. Polling every 10 seconds with a 900-second cap keeps the workflow honest about outliers.

Can Lyria 3.5 really hit a target duration and tempo?

Yes. Lyria 3.5 exposes tempo (BPM) and duration controls with structured sections such as verse–chorus–bridge. That is what lets the orchestrator plan music that matches video length instead of trimming or padding by luck.

Do I need the script node to be an LLM call?

No. You can feed a fully formed Script directly into MediaState and skip the model entirely. The LLM node exists to convert a creative brief into structured scenes; for templated content, precompute the script and keep the graph deterministic.

What happens when a generation job times out?

The polling loop raises TimeoutError, which the Retry Rules treat as a failed attempt (music: 3, video: 4). Because the graph is checkpointed, a long job can also outlive the worker that started it — the thread resumes from the last completed node and re-polls.

Is the human review gate required?

For production, yes. The graph ships with the gate wired to a review_passed flag so you can pause the pipeline. In practice, replace _review with an approval webhook — auto-publishing un-reviewed generated media is the number one incident in this workflow class.