Skip to content

AI auto-editing

LLM-authored editing: build a scene catalog from one or more sources and let a local vision-language model plan a VideoEdit from it. End-to-end usage is in Let a local LLM edit for you; the agent-driven variant is the MCP server.

from videopython.ai import AutoEditor, OllamaVisionLLM

editor = AutoEditor(planner=OllamaVisionLLM(model="qwen3.6:27b"))
edit = editor.edit(["a.mp4", "b.mp4"], brief="A 15s teaser, most dynamic shot first.")
edit.run_to_file("teaser.mp4")

The catalog and the by-id plan

build_catalog projects VideoAnalysis results into an EditCatalog of candidate CatalogScenes — each with a stable id, exact bounds, caption and transcript — plus one keyframe per scene. The planner authors an EditPlan whose segments reference scenes by scene_id, and resolve_plan maps those ids back to a runnable VideoEdit. The model never authors timestamps (why).

from videopython.ai import VideoAnalyzer, build_catalog
from videopython.ai.auto_edit import EditPlan, resolve_plan

analyses = [VideoAnalyzer().analyze_path("a.mp4")]
bundle = build_catalog(analyses)              # bundle.catalog + bundle.keyframes

plan = EditPlan.model_validate({"segments": [{"scene_id": bundle.catalog.scenes[0].id}]})
edit = resolve_plan(plan, bundle.catalog)     # -> VideoEdit

Speech candidates

build_catalog(..., mode="speech", speech=SpeechCandidateConfig(...)) selects timed speech passages. mode="visual" is the default and requires speech=None. SpeechCandidateConfig requires positive min_duration and max_duration in seconds, with max_duration >= min_duration. pause_duration is positive and defaults to 0.8 seconds.

Speech mode uses these deterministic rules:

  • Each nonempty transcript segment must contain timed words. Otherwise that source contributes no candidates. Empty or absent transcriptions also contribute none.
  • Words are sorted by start, then end. Times must be finite with 0 <= start <= end; invalid times raise ValueError. Zero-duration words retain their supplied timestamp and text ownership. Their duration is not inferred.
  • A span ends after ., ?, !, …, 。, ？, or ！ (with closing quotes or brackets allowed), or before a gap of at least pause_duration. A boundary cannot pass through another overlapping word. An unfinished trailing span is discarded.
  • Consecutive spans are combined until the minimum duration is met. If adding a span would exceed the maximum or cross a gap of at least pause_duration, the shorter pending group is discarded. A single span longer than the maximum is also discarded. Durations include shorter internal pauses.
  • Candidates are chronological within each source and do not overlap. Source order follows the supplied analyses. Start/end are the first word start and the latest word end, without added padding. Source duration is required. Passages extending beyond it are omitted, not clamped through a timed word; supplied word records stay unchanged. An empty catalog.scenes is a valid outcome.

Speech mode does not assign a shot caption, shot type, or face flag. It can cross visual cuts. Sentence punctuation and pauses are selection cues, not a topic or meaning model: abbreviations, recognition errors, and an excerpt that starts mid-sentence can produce poor editorial boundaries. Review the retrieved text and rendered cuts. Supplied timing precision limits audio alignment; frame timing still follows the renderer's source-frame contract.

CatalogBundle.transcripts maps every candidate ID to its full normalized transcript text, in either mode. CatalogScene.transcript remains a short excerpt (280 characters by default). Whitespace is normalized; source words and times are not changed.

IDs and migration

Visual mode keeps the existing IDs. Speech IDs include a digest of the resolved source path, transcript, and speech settings, followed by the candidate index. Repeated builds with the same inputs produce the same IDs. Sources with identical file stems in different directories have distinct speech IDs.

Rebuild the catalog and regenerate saved by-ID plans when changing mode, transcript, source path, or speech settings. Old speech IDs will fail resolution rather than select new ranges. The plan field remains scene_id. This is catalog identity, not a media-content integrity check.

SpeechCandidateConfig

Bases: BaseModel

Duration limits and minimum pause for deterministic speech candidates.

Source code in src/videopython/ai/auto_edit/models.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class SpeechCandidateConfig(BaseModel):
    """Duration limits and minimum pause for deterministic speech candidates."""

    model_config = ConfigDict(extra="forbid", frozen=True, allow_inf_nan=False)

    min_duration: float = Field(gt=0)
    max_duration: float = Field(gt=0)
    pause_duration: float = Field(0.8, gt=0)

    @model_validator(mode="after")
    def _ordered_limits(self) -> SpeechCandidateConfig:
        if self.max_duration < self.min_duration:
            raise ValueError("max_duration must be at least min_duration")
        return self

Planner

AutoEditor

Plan a VideoEdit from source videos using a structured-vision planner.

Source code in src/videopython/ai/auto_edit/editor.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class AutoEditor:
    """Plan a VideoEdit from source videos using a structured-vision planner."""

    def __init__(
        self,
        planner: StructuredVisionLLM,
        *,
        analyzer: VideoAnalyzer | None = None,
        max_rounds: int = 3,
        normalize_target: _NormalizeTarget = "largest",
    ) -> None:
        self.planner = planner
        self._analyzer = analyzer
        self.max_rounds = max_rounds
        self.normalize_target = normalize_target

    def edit(self, sources: Sequence[str | Path], brief: str, *, context: dict[str, Any] | None = None) -> VideoEdit:
        """Analyze ``sources`` and plan an edit for ``brief`` (runs the analyzer)."""
        analyses = [self._get_analyzer().analyze_path(source) for source in sources]
        return self.edit_from_analyses(analyses, brief, context=context)

    def edit_from_analyses(
        self, analyses: Sequence[VideoAnalysis], brief: str, *, context: dict[str, Any] | None = None
    ) -> VideoEdit:
        """Plan an edit from precomputed VideoAnalysis results (no model download)."""
        bundle = build_catalog(analyses)
        metadata = _metadata_by_source(analyses)
        run_context = _merge_context(analyses, context)
        schema = EditPlan.json_schema(strict=True)
        base_text, images = _build_prompt(brief, bundle)

        feedback: str | None = None
        for _ in range(self.max_rounds):
            text = base_text if feedback is None else f"{base_text}\n\n{feedback}"
            try:
                raw = self.planner.generate_json(system=_SYSTEM_PROMPT, text=text, images=images or None, schema=schema)
                edit = resolve_plan(EditPlan.model_validate(raw), bundle.catalog)
            except (PlannerError, ValidationError, UnknownSceneIdsError) as exc:
                feedback = _shape_feedback(exc)
                continue
            edit, _ = edit.repair(metadata, context=run_context, clamp_segment_end=True)
            edit, _ = edit.normalize_dimensions(metadata, self.normalize_target, context=run_context)
            errors = edit.check(metadata, context=run_context)
            if not errors:
                return edit
            feedback = "The previous plan had these problems:\n" + "\n".join(e.to_prompt_line() for e in errors)

        raise AutoEditError(f"No valid edit after {self.max_rounds} round(s). Last feedback:\n{feedback}")

    def _get_analyzer(self) -> VideoAnalyzer:
        if self._analyzer is None:
            from videopython.ai.video_analysis import VideoAnalyzer

            self._analyzer = VideoAnalyzer()
        return self._analyzer

edit 

edit(
    sources: Sequence[str | Path],
    brief: str,
    *,
    context: dict[str, Any] | None = None,
) -> VideoEdit

Analyze sources and plan an edit for brief (runs the analyzer).

Source code in src/videopython/ai/auto_edit/editor.py 
55
56
57
58
def edit(self, sources: Sequence[str | Path], brief: str, *, context: dict[str, Any] | None = None) -> VideoEdit:
    """Analyze ``sources`` and plan an edit for ``brief`` (runs the analyzer)."""
    analyses = [self._get_analyzer().analyze_path(source) for source in sources]
    return self.edit_from_analyses(analyses, brief, context=context)

edit_from_analyses 

edit_from_analyses(
    analyses: Sequence[VideoAnalysis],
    brief: str,
    *,
    context: dict[str, Any] | None = None,
) -> VideoEdit

Plan an edit from precomputed VideoAnalysis results (no model download).

Source code in src/videopython/ai/auto_edit/editor.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def edit_from_analyses(
    self, analyses: Sequence[VideoAnalysis], brief: str, *, context: dict[str, Any] | None = None
) -> VideoEdit:
    """Plan an edit from precomputed VideoAnalysis results (no model download)."""
    bundle = build_catalog(analyses)
    metadata = _metadata_by_source(analyses)
    run_context = _merge_context(analyses, context)
    schema = EditPlan.json_schema(strict=True)
    base_text, images = _build_prompt(brief, bundle)

    feedback: str | None = None
    for _ in range(self.max_rounds):
        text = base_text if feedback is None else f"{base_text}\n\n{feedback}"
        try:
            raw = self.planner.generate_json(system=_SYSTEM_PROMPT, text=text, images=images or None, schema=schema)
            edit = resolve_plan(EditPlan.model_validate(raw), bundle.catalog)
        except (PlannerError, ValidationError, UnknownSceneIdsError) as exc:
            feedback = _shape_feedback(exc)
            continue
        edit, _ = edit.repair(metadata, context=run_context, clamp_segment_end=True)
        edit, _ = edit.normalize_dimensions(metadata, self.normalize_target, context=run_context)
        errors = edit.check(metadata, context=run_context)
        if not errors:
            return edit
        feedback = "The previous plan had these problems:\n" + "\n".join(e.to_prompt_line() for e in errors)

    raise AutoEditError(f"No valid edit after {self.max_rounds} round(s). Last feedback:\n{feedback}")

OllamaVisionLLM

A StructuredVisionLLM backed by a local Ollama server.

The model must be vision-capable (it is sent keyframes) AND support Ollama's structured-output format (the EditPlan schema constrains the decode). The default qwen3.6:27b is an Apache-2.0 vision model; not every model supports schema conditioning (some builds, e.g. certain MLX ones, fail it), so confirm format works for a custom model locally. ollama pull <model> first; options are extra generation options merged over temperature=0.

Thin wrapper over the shared :class:OllamaStructuredClient: its only job is to translate :class:OllamaError into the :class:PlannerError the editor retries on.

Source code in src/videopython/ai/auto_edit/local.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class OllamaVisionLLM:
    """A StructuredVisionLLM backed by a local Ollama server.

    The model must be vision-capable (it is sent keyframes) AND support Ollama's
    structured-output ``format`` (the EditPlan schema constrains the decode). The
    default ``qwen3.6:27b`` is an Apache-2.0 vision model; not every model supports
    schema conditioning (some builds, e.g. certain MLX ones, fail it), so confirm
    ``format`` works for a custom model locally. ``ollama pull <model>`` first;
    ``options`` are extra generation options merged over ``temperature=0``.

    Thin wrapper over the shared :class:`OllamaStructuredClient`: its only job is
    to translate :class:`OllamaError` into the :class:`PlannerError` the editor
    retries on.
    """

    def __init__(
        self,
        model: str = DEFAULT_OLLAMA_MODEL,
        *,
        host: str | None = None,
        options: dict[str, Any] | None = None,
    ) -> None:
        self._client = OllamaStructuredClient(model=model, host=host, options=options)

    def generate_json(
        self, *, system: str, text: str, images: list[np.ndarray] | None, schema: dict[str, Any]
    ) -> dict[str, Any]:
        try:
            return self._client.generate_json(system=system, text=text, schema=schema, images=images or None)
        except OllamaError as exc:
            raise PlannerError(str(exc)) from exc

StructuredVisionLLM

Bases: Protocol

Returns schema-shaped JSON from a system prompt + text + optional keyframes.

The signature mirrors :meth:videopython.ai._ollama.OllamaStructuredClient.generate_json, so any structured-generation client satisfies it structurally. Implementations raise :class:PlannerError on unusable output (the editor retries those); infra errors should propagate so they are not silently retried.

Source code in src/videopython/ai/auto_edit/backend.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@runtime_checkable
class StructuredVisionLLM(Protocol):
    """Returns schema-shaped JSON from a system prompt + text + optional keyframes.

    The signature mirrors
    :meth:`videopython.ai._ollama.OllamaStructuredClient.generate_json`, so any
    structured-generation client satisfies it structurally. Implementations
    raise :class:`PlannerError` on unusable output (the editor retries those);
    infra errors should propagate so they are not silently retried.
    """

    def generate_json(
        self, *, system: str, text: str, images: list[np.ndarray] | None, schema: dict[str, Any]
    ) -> dict[str, Any]: ...

Catalog and plan

build_catalog 

build_catalog(
    analyses: Sequence[VideoAnalysis],
    *,
    keyframes: bool = True,
    max_transcript_chars: int = DEFAULT_TRANSCRIPT_CHARS,
    mode: Literal["visual", "speech"] = "visual",
    speech: SpeechCandidateConfig | None = None,
) -> CatalogBundle

Build visual scenes or timed speech passages; speech mode requires speech settings.

Source code in src/videopython/ai/auto_edit/catalog.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def build_catalog(
    analyses: Sequence[VideoAnalysis],
    *,
    keyframes: bool = True,
    max_transcript_chars: int = DEFAULT_TRANSCRIPT_CHARS,
    mode: Literal["visual", "speech"] = "visual",
    speech: SpeechCandidateConfig | None = None,
) -> CatalogBundle:
    """Build visual scenes or timed speech passages; speech mode requires speech settings."""
    if mode not in ("visual", "speech") or (mode == "speech") != (speech is not None):
        raise ValueError("Supply speech settings exactly when mode='speech'")
    scenes: list[CatalogScene] = []
    transcripts: dict[str, str] = {}
    used_ids: set[str] = set()

    for analysis in analyses:
        source_path = analysis.source.path
        samples = analysis.scenes.samples if analysis.scenes else []
        transcription = analysis.audio.transcription if analysis.audio else None
        stem = Path(source_path).stem if source_path else "clip"

        if speech is not None:
            if analysis.source.duration is None:
                raise ValueError("Speech candidates require the source duration")
            identity = json.dumps(
                [
                    str(Path(source_path).resolve()) if source_path else None,
                    transcription.model_dump() if transcription else None,
                    speech.model_dump(),
                ],
                sort_keys=True,
            )
            digest = hashlib.sha256(identity.encode()).hexdigest()[:24]
            for index, passage in enumerate(speech_passages(transcription, speech)):
                start, end = passage[0].start, max(word.end for word in passage)
                if end > analysis.source.duration or start >= analysis.source.duration:
                    continue
                scene_id = _unique_id(f"{stem}#speech-{digest}", index, used_ids)
                text = " ".join(" ".join(word.word.split()) for word in passage).strip()
                scenes.append(
                    CatalogScene(
                        id=scene_id,
                        source=Path(source_path) if source_path else Path(stem),
                        start=start,
                        end=end,
                        duration=end - start,
                        transcript=_shorten(text, max_transcript_chars),
                        has_speech=True,
                    )
                )
                transcripts[scene_id] = text
                if keyframes and source_path is None:
                    raise ValueError(f"Scene {scene_id!r} has no source path to extract a keyframe from.")
            continue

        for sample in samples:
            scene_id = _unique_id(stem, sample.scene_index, used_ids)
            caption, shot_type = _description(sample)
            text = _transcript_text(transcription, sample.start_second, sample.end_second)
            transcripts[scene_id] = text
            transcript = _shorten(text, max_transcript_chars)
            scenes.append(
                CatalogScene(
                    id=scene_id,
                    source=Path(source_path) if source_path else Path(stem),
                    start=sample.start_second,
                    end=sample.end_second,
                    duration=max(0.0, sample.end_second - sample.start_second),
                    shot_type=shot_type,
                    caption=caption,
                    transcript=transcript,
                    has_speech=bool(transcript),
                    has_faces=bool(sample.faces),
                )
            )
            if keyframes and source_path is None:
                raise ValueError(f"Scene {scene_id!r} has no source path to extract a keyframe from.")

    frames = extract_catalog_keyframes(scenes) if keyframes else {}
    return CatalogBundle(catalog=EditCatalog(scenes=scenes), keyframes=frames, transcripts=transcripts)

resolve_plan 

resolve_plan(
    plan: EditPlan, catalog: EditCatalog
) -> VideoEdit

Map each plan segment's scene_id to its exact source/start/end.

Source code in src/videopython/ai/auto_edit/resolve.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def resolve_plan(plan: EditPlan, catalog: EditCatalog) -> VideoEdit:
    """Map each plan segment's scene_id to its exact source/start/end."""
    by_id = catalog.by_id()
    unknown = [seg.scene_id for seg in plan.segments if seg.scene_id not in by_id]
    if unknown:
        raise UnknownSceneIdsError(unknown)
    segments = [
        SegmentConfig(
            source=by_id[seg.scene_id].source,
            start=by_id[seg.scene_id].start,
            end=by_id[seg.scene_id].end,
            operations=seg.operations,
            transition_in=seg.transition_in,
        )
        for seg in plan.segments
    ]
    return VideoEdit(segments=segments, post_operations=plan.post_operations)

EditPlan

Bases: BaseModel

The planner's output: an ordered selection of catalog scenes, referenced by id.

Source code in src/videopython/ai/auto_edit/models.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class EditPlan(BaseModel):
    """The planner's output: an ordered selection of catalog scenes, referenced by id."""

    model_config = ConfigDict(extra="forbid")

    segments: list[PlanSegment] = Field(min_length=1, description="Ordered plan segments.")
    post_operations: list[OperationInput] = Field(
        default_factory=list, description="Operations applied once to the whole assembled program."
    )

    @classmethod
    def json_schema(cls, *, strict: bool = False) -> dict[str, Any]:
        """The by-id mirror of VideoEdit.json_schema, reusing the op union and strict rewrite."""
        op_schema = Operation.json_schema()

        segment_schema: dict[str, Any] = {
            "type": "object",
            "description": PlanSegment.__doc__,
            "properties": {
                "scene_id": field_schema(PlanSegment, "scene_id"),
                "operations": array_field_schema(PlanSegment, "operations", op_schema),
                "transition_in": optional_model_field_schema(TransitionSpec, PlanSegment, "transition_in"),
            },
            "required": ["scene_id"],
            "additionalProperties": False,
        }
        segments = field_schema(cls, "segments")
        segments["items"] = segment_schema
        schema: dict[str, Any] = {
            "$schema": "http://json-schema.org/draft-07/schema#",
            "type": "object",
            "description": cls.__doc__,
            "properties": {
                "segments": segments,
                "post_operations": array_field_schema(cls, "post_operations", op_schema),
            },
            "required": ["segments"],
            "additionalProperties": False,
        }
        if not strict:
            return schema
        op_defs = op_schema.pop("$defs", None)
        if op_defs:
            schema["$defs"] = op_defs
        return _to_strict_schema(schema)

json_schema classmethod 

json_schema(*, strict: bool = False) -> dict[str, Any]

The by-id mirror of VideoEdit.json_schema, reusing the op union and strict rewrite.

Source code in src/videopython/ai/auto_edit/models.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def json_schema(cls, *, strict: bool = False) -> dict[str, Any]:
    """The by-id mirror of VideoEdit.json_schema, reusing the op union and strict rewrite."""
    op_schema = Operation.json_schema()

    segment_schema: dict[str, Any] = {
        "type": "object",
        "description": PlanSegment.__doc__,
        "properties": {
            "scene_id": field_schema(PlanSegment, "scene_id"),
            "operations": array_field_schema(PlanSegment, "operations", op_schema),
            "transition_in": optional_model_field_schema(TransitionSpec, PlanSegment, "transition_in"),
        },
        "required": ["scene_id"],
        "additionalProperties": False,
    }
    segments = field_schema(cls, "segments")
    segments["items"] = segment_schema
    schema: dict[str, Any] = {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",
        "description": cls.__doc__,
        "properties": {
            "segments": segments,
            "post_operations": array_field_schema(cls, "post_operations", op_schema),
        },
        "required": ["segments"],
        "additionalProperties": False,
    }
    if not strict:
        return schema
    op_defs = op_schema.pop("$defs", None)
    if op_defs:
        schema["$defs"] = op_defs
    return _to_strict_schema(schema)

EditCatalog

Bases: BaseModel

The candidate scenes across all source videos.

Source code in src/videopython/ai/auto_edit/models.py 
48
49
50
51
52
53
54
class EditCatalog(BaseModel):
    """The candidate scenes across all source videos."""

    scenes: list[CatalogScene]

    def by_id(self) -> dict[str, CatalogScene]:
        return {scene.id: scene for scene in self.scenes}

CatalogScene

Bases: BaseModel

A candidate scene the planner picks by id; carries the exact source bounds.

Source code in src/videopython/ai/auto_edit/models.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
class CatalogScene(BaseModel):
    """A candidate scene the planner picks by id; carries the exact source bounds."""

    id: str
    source: Path
    start: float
    end: float
    duration: float
    shot_type: str | None = None
    caption: str = ""
    transcript: str = ""
    has_speech: bool = False
    has_faces: bool = False

Errors

AutoEditError

Bases: AiError, RuntimeError

The planner could not produce a valid edit within the retry budget.

Source code in src/videopython/ai/auto_edit/editor.py 
35
36
class AutoEditError(AiError, RuntimeError):
    """The planner could not produce a valid edit within the retry budget."""

PlannerError

Bases: AiError, RuntimeError

A backend produced unusable output; the editor retries (infra errors should propagate instead).

Source code in src/videopython/ai/auto_edit/backend.py 
13
14
class PlannerError(AiError, RuntimeError):
    """A backend produced unusable output; the editor retries (infra errors should propagate instead)."""

UnknownSceneIdsError

Bases: AiError, ValueError

An EditPlan referenced scene ids absent from the catalog.

Source code in src/videopython/ai/auto_edit/resolve.py 
11
12
13
14
15
16
class UnknownSceneIdsError(AiError, ValueError):
    """An EditPlan referenced scene ids absent from the catalog."""

    def __init__(self, ids: list[str]) -> None:
        self.ids = ids
        super().__init__(f"Plan references unknown scene ids: {sorted(set(ids))}")

Keyframe memory

Python catalogs with keyframes=True retain every selected full-resolution RGB frame. Selected-frame extraction reads directly into one output array, preserving request order and duplicate timestamps. An unreachable frame raises VideoLoadError rather than returning a shorter array. The array needs approximately N × width × height × 3 bytes for N frames, plus decoder and process overhead. At 300 frames of 1920×1080, that array alone is about 1.87 GB. Use keyframes=False for text-only selection, then request a shortlist. This analysis allocation is separate from the renderer's bounded frame buffers. MCP has its own request and cache limits.