core

The source of truth

This library follows the fastai style guide, and is crafted with nbdev.

This is a general purpose library that allows you to use a VLM (Vision Language Model) to thoroughly describe the contents of a video, with subtitles included for additional context.

In essence, this library provides a second set of eyes.

The output is a database object containing the video description.

If you want to directly run this notebook, you want to have an OPENROUTER_API_KEY set.

Preface

This is a library that allows you to thoroughly describe what occurs in a video.

Concisely:

  • A VLM describes the video frame by frame. Each time, it is provided with an empty history together with the frame, the subtitle, and instructions about what exactly to do to describe the frame. If one is processing a 5 minute video with a sample rate of 1 frame per second, the output is 300 individual, isolated frame descriptions.
  • An LLM then takes all the isolated frame descriptions and pieces them together to form a summary/description of the video. The user could, for instance, piece together a video description whos summary windows consist of 60 frames (in which case, when the LLM produces the summary of the next window, it will keep the previous windows in its chat history). Or, the user could piece toether an overall video description consisting of a single window comprising of 300 frames.

The biggest beneficiary of this approach is LLM context. Traditional VLM description systems keep the image in the chat history. Images are token heavy. Storing the desription of the image, rather than the image itself, saves the necessary information whilst allowing higher definition: you can describe videos at 1 frame per second, or even lower if you desire so.

More concretely, this library works as follows:

  1. Set up a database to store the data.
  2. Load the videos and their frames into the database.
  3. Allow a VLM to describe the frames.
  4. Allow a LLM to piece together a description.

And at an even lower level, as follows:

  1. Set up a database consisting of 4 tables.
    • A video table to store metadata about your videos
    • A frame table to store metadata about the frames in each of your videos
    • A run table to store metadata about each description process
    • A runframe table to store metadata about each described frame
  2. Populate the video and frame tables
  3. Define the VLM settings
  4. Process the frames through the VLM, storing the frame descriptions in runframe
  5. Process the resulting descriptions through the LLM, storing the resulting summary in video

What follows is an exposition of the source code. You’ll typically see source code written first, and then some exposition afterward.

Database

Exported source 
from fastcore.all import *

A database is needed to store information about the:

  • videos
  • frames each video
  • each deployed run
  • and the descriptions made for each frame in a run.

These classes hold the definition of the tables.

Exported source 
from fastlite import *

I first need a database object. For that, I’ll be using fastlite which is a wrapper over sqlite-utils.

!rm db.db
db = database('db.db'); db
<Database <apsw.Connection "/app/data/vlm-monitor/nbs/db.db">>
videos = db.create(Video, transform=True); print(videos.schema)
CREATE TABLE [video] (
   [id] INTEGER PRIMARY KEY,
   [title] TEXT,
   [overview] TEXT,
   [description] TEXT,
   [transcript] TEXT,
   [length] INTEGER,
   [sample_rate] INTEGER,
   [path] TEXT
)

I’m now creating the tables from the classes I’ve defined.

While I can view the schema, I’ll add some markdown highlighting to make things easier to visually distinguish. sqlite-utils tables inherit from Queryable. So I’ll patch its .schema method with fastcore’s hl_md function.

??Queryable.schema
@property
def schema(self) -> str:
    "SQL schema for this table or view."
    return self.db.execute(
        "select sql from sqlite_master where name = ?", (self.name,)
    ).fetchone()[0]

File: /usr/local/lib/python3.12/site-packages/apswutils/db.py; line: 1359

?hl_md
def hl_md(
    s, lang:str='html', show:bool=True
):
    "Syntax highlight `s` using `lang`."

File: ~/.local/lib/python3.12/site-packages/fastcore/xtras.py; line: 1098

Type: function

Queryable.schema

def schema()->str:

SQL schema for this table or view.

Exported source 
@patch(as_prop=True)
def schema(self:Queryable) -> str:
    "SQL schema for this table or view."
    return hl_md(self.db.execute(
        "select sql from sqlite_master where name = ?", (self.name,)
    ).fetchone()[0], lang='sql')

Things should now look much visually nicer.

videos.schema
CREATE TABLE [video] (
   [id] INTEGER PRIMARY KEY,
   [title] TEXT,
   [overview] TEXT,
   [description] TEXT,
   [transcript] TEXT,
   [length] INTEGER,
   [sample_rate] INTEGER,
   [path] TEXT
)

I’ll create the remaining tables.

frames = db.create(Frame, transform=True, foreign_keys=[('video_id', 'video', 'id')]); frames.schema
CREATE TABLE [frame] (
   [id] INTEGER PRIMARY KEY,
   [video_id] INTEGER REFERENCES [video]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [frame_number] INTEGER,
   [subtitle] TEXT
)
runs = db.create(Run, transform=True); runs.schema
CREATE TABLE [run] (
   [id] INTEGER PRIMARY KEY,
   [deploy_time] TEXT,
   [finish_time] TEXT,
   [start_time] TEXT,
   [total_duration] TEXT,
   [video_id] INTEGER,
   [model] TEXT,
   [usage] TEXT,
   [num_frames] INTEGER,
   [start_sec] INTEGER,
   [end_sec] INTEGER,
   [step] INTEGER,
   [description] TEXT
)
runframes = db.create(RunFrame, pk=['run_id', 'frame_id', 'type'], foreign_keys=[('run_id', 'run', 'id'), ('frame_id', 'frame', 'id')], transform=True); runframes.schema
CREATE TABLE [run_frame] (
   [run_id] INTEGER REFERENCES [run]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [frame_id] INTEGER REFERENCES [frame]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [type] TEXT,
   [system_prompt] TEXT,
   [prompt] TEXT,
   [description] TEXT,
   [usage] TEXT,
   [skipped] INTEGER,
   PRIMARY KEY ([run_id], [frame_id], [type])
)

init_db

def init_db(
    path:str | pathlib.Path='db.db', # Path to database
)->Database:

Initialize a database and return it.

Exported source 
from apswutils.db import Database
Exported source 
def init_db(
    path:str|Path='db.db' # Path to database
) -> Database:
    "Initialize a database and return it."
    db = database(path)
    db.create(Video, transform=True)
    db.create(Frame, transform=True, foreign_keys=[('video_id', 'video', 'id')])
    db.create(Run, transform=True)
    db.create(RunFrame, pk=['run_id', 'frame_id', 'type'], foreign_keys=[('run_id', 'run', 'id'), ('frame_id', 'frame', 'id')], transform=True)
    for t in db.t: t.dataclass()
    return db

I’ve now wrapped everything together that initializes the database in a single function.

!rm db.db
db = init_db(); db
<Database <apsw.Connection "/app/data/vlm-monitor/nbs/db.db">>
db.t.video.schema
CREATE TABLE [video] (
   [id] INTEGER PRIMARY KEY,
   [title] TEXT,
   [overview] TEXT,
   [description] TEXT,
   [transcript] TEXT,
   [length] INTEGER,
   [sample_rate] INTEGER,
   [path] TEXT
)

view_col

def view_col(
    table:Table, # Database table
    col:str, # Column to view
    where:str | None=None, # SQL lookup statement
    where_args:str | None=None, # SQL lookup statement arguments
)->list: # List of rows

Return all rows of a given column in table, optionally setting where.

Exported source 
from operator import attrgetter, itemgetter
Exported source 
def view_col(
    table:Table, # Database table
    col:str, # Column to view
    where:str|None=None, # SQL lookup statement
    where_args:str|None=None # SQL lookup statement arguments
)->list: # List of rows
    "Return all rows of a given column in table, optionally setting `where`."
    if where is None: return L(table()).map(attrgetter(col))
    else: return L(table(where=where, where_args=where_args)).map(attrgetter(col))

I’ve also defined a little helper function here that makes it easier to view all rows of a given column in a table.

Populate

In this section, I write the logic that will populate the video and frame tables in one function.

For this to work, the following assumptions are made: - the provided directory is flat, - all folders contain the video frames, and - the same folders contain the respective video’s transcript.

!tree ../../data/timss -L 1
../../data/timss
├── M-AU1
├── M-AU2
├── M-AU3
├── M-AU4
├── M-CZ1
├── M-CZ2
├── M-CZ3
├── M-CZ4
├── M-HK1
├── M-HK2
├── M-HK3
├── M-HK4
├── M-JP1
├── M-JP2
├── M-JP3
├── M-JP4
├── M-NL1
├── M-NL2
├── M-NL3
├── M-NL4
├── M-SW1
├── M-SW2
├── M-SW3
├── M-SW4
├── M-US1
├── M-US2
├── M-US3
├── M-US4
├── S-AU1
├── S-AU2
├── S-AU3
├── S-AU4
├── S-AU5
├── S-CZ1
├── S-CZ2
├── S-CZ3
├── S-CZ4
├── S-CZ5
├── S-JP1
├── S-JP2
├── S-JP3
├── S-JP4
├── S-JP5
├── S-NL1
├── S-NL2
├── S-NL3
├── S-NL4
├── S-NL5
├── S-US1
├── S-US3
├── S-US4
└── S-US5

53 directories, 0 files
dpath = Path('../../data/timss/'); dpath.ls()
[Path('../../data/timss/M-CZ3'), Path('../../data/timss/M-AU2'), Path('../../data/timss/M-CZ4'), Path('../../data/timss/M-JP2'), Path('../../data/timss/S-AU4'), Path('../../data/timss/S-US3'), Path('../../data/timss/S-CZ1'), Path('../../data/timss/M-AU1'), Path('../../data/timss/S-AU5'), Path('../../data/timss/S-US4'), Path('../../data/timss/S-NL2'), Path('../../data/timss/S-CZ5'), Path('../../data/timss/S-AU2'), Path('../../data/timss/S-JP5'), Path('../../data/timss/M-NL2'), Path('../../data/timss/M-AU3'), Path('../../data/timss/S-US5'), Path('../../data/timss/M-SW2'), Path('../../data/timss/M-HK4'), Path('../../data/timss/S-AU3'), Path('../../data/timss/M-HK2'), Path('../../data/timss/S-NL5'), Path('../../data/timss/S-CZ4'), Path('../../data/timss/.DS_Store'), Path('../../data/timss/M-NL4'), Path('../../data/timss/M-CZ1'), Path('../../data/timss/S-JP2'), Path('../../data/timss/M-NL1'), Path('../../data/timss/M-US3'), Path('../../data/timss/M-SW3'), Path('../../data/timss/M-US1'), Path('../../data/timss/M-JP4'), Path('../../data/timss/M-US2'), Path('../../data/timss/M-JP3'), Path('../../data/timss/M-CZ2'), Path('../../data/timss/M-HK3'), Path('../../data/timss/S-JP1'), Path('../../data/timss/S-CZ3'), Path('../../data/timss/S-US1'), Path('../../data/timss/M-SW4'), Path('../../data/timss/M-HK1'), Path('../../data/timss/M-JP1'), Path('../../data/timss/S-NL3'), Path('../../data/timss/M-SW1'), Path('../../data/timss/M-AU4'), Path('../../data/timss/S-NL4'), Path('../../data/timss/S-NL1'), Path('../../data/timss/S-AU1'), Path('../../data/timss/S-JP4'), Path('../../data/timss/S-JP3'), Path('../../data/timss/M-NL3'), Path('../../data/timss/M-US4'), Path('../../data/timss/S-CZ2')]

I can see that my data has some unneccessary files, such as .DS_Store.

filter_paths

def filter_paths(
    paths:list, # List of paths to filter
    chs:str='.', # Characters to check for in the component
    comp:str='stem', # Path attribute to inspect (e.g. 'stem', 'name')
    negate:bool=True, # If True, exclude paths whose `comp` contains `chs`; if False, keep only those
)->list: # Filtered list of paths

Filter paths by whether a path component contains specified characters.

Exported source 
def filter_paths(
    paths:list[Path], # List of paths to filter
    chs:str='.', # Characters to check for in the component
    comp:str='stem', # Path attribute to inspect (e.g. 'stem', 'name')
    negate:bool=True, # If True, exclude paths whose `comp` contains `chs`; if False, keep only those
)->list[Path]: # Filtered list of paths
    "Filter paths by whether a path component contains specified characters."
    return paths.filter(~getattr(Self, comp).count(chs), negate=negate)

I’ve written a function that both filters out those unnecessary files, at the same time also filters in the necessary ones!

dpaths = filter_paths(dpath.ls()).sorted(lambda o: (o.stem[:-1], o.stem[-1])); dpaths
[Path('../../data/timss/M-AU1'), Path('../../data/timss/M-AU2'), Path('../../data/timss/M-AU3'), Path('../../data/timss/M-AU4'), Path('../../data/timss/M-CZ1'), Path('../../data/timss/M-CZ2'), Path('../../data/timss/M-CZ3'), Path('../../data/timss/M-CZ4'), Path('../../data/timss/M-HK1'), Path('../../data/timss/M-HK2'), Path('../../data/timss/M-HK3'), Path('../../data/timss/M-HK4'), Path('../../data/timss/M-JP1'), Path('../../data/timss/M-JP2'), Path('../../data/timss/M-JP3'), Path('../../data/timss/M-JP4'), Path('../../data/timss/M-NL1'), Path('../../data/timss/M-NL2'), Path('../../data/timss/M-NL3'), Path('../../data/timss/M-NL4'), Path('../../data/timss/M-SW1'), Path('../../data/timss/M-SW2'), Path('../../data/timss/M-SW3'), Path('../../data/timss/M-SW4'), Path('../../data/timss/M-US1'), Path('../../data/timss/M-US2'), Path('../../data/timss/M-US3'), Path('../../data/timss/M-US4'), Path('../../data/timss/S-AU1'), Path('../../data/timss/S-AU2'), Path('../../data/timss/S-AU3'), Path('../../data/timss/S-AU4'), Path('../../data/timss/S-AU5'), Path('../../data/timss/S-CZ1'), Path('../../data/timss/S-CZ2'), Path('../../data/timss/S-CZ3'), Path('../../data/timss/S-CZ4'), Path('../../data/timss/S-CZ5'), Path('../../data/timss/S-JP1'), Path('../../data/timss/S-JP2'), Path('../../data/timss/S-JP3'), Path('../../data/timss/S-JP4'), Path('../../data/timss/S-JP5'), Path('../../data/timss/S-NL1'), Path('../../data/timss/S-NL2'), Path('../../data/timss/S-NL3'), Path('../../data/timss/S-NL4'), Path('../../data/timss/S-NL5'), Path('../../data/timss/S-US1'), Path('../../data/timss/S-US3'), Path('../../data/timss/S-US4'), Path('../../data/timss/S-US5')]

The way I’ve defined filter_paths means I can filter for any file. I’ll filter for the transcripts and take a look inside one of them.

dp = dpaths[0]; dp
Path('../../data/timss/M-AU1')
tr = filter_paths(dp.ls(), chs='txt', comp='suffix', negate=False)[0]
print(tr.read_text()[:500])
1
00:00:20,000 --> 00:00:42,980
[T] I'm wired.

2
00:00:43,000 --> 00:00:53,980
[T] It's running.

3
00:00:54,000 --> 00:00:56,980
[SN] (inaudible) please turn off the air.

4
00:00:57,000 --> 00:01:06,980
[T] I'll make it a bit warmer.

5
00:01:07,000 --> 00:01:16,980
[T] Well not really, but if you really have to I suppose. Okay.

6
00:01:17,000 --> 00:01:20,980
[SN] (inaudible) go out to my locker and get my maths book?

7
00:01:21,000 --> 00:01:22,980
[T] Oh, you won't need it today.

8
00:0

The transcripts are in SRT format. I’ll be converting them to TSV, making them token efficient. To do that, I’ll need a way to convert all timestamps to seconds.

f'45 seconds is {int(45)*60**1} seconds'
'45 seconds is 2700 seconds'
f'45 minutes is {int(45)*60**2} seconds'
'45 minutes is 162000 seconds'
f'45 hours is {int(45)*60**3} seconds'
'45 hours is 9720000 seconds'
'01:50'.split(':')
['01', '50']
L(enumerate(reversed('01:50'.split(':'))))
[(0, '50'), (1, '01')]
sum(int(x)*60**i for i,x in enumerate(reversed('01:50'.split(':'))))
110

time2sec

def time2sec(
    time:str, # Time string in HH:MM:SS or MM:SS format
)->int: # Total seconds

Convert a time string to total seconds.

Exported source 
import re
Exported source 
def time2sec(
    time:str, # Time string in HH:MM:SS or MM:SS format
)->int: # Total seconds
    "Convert a time string to total seconds."
    return sum(int(x)*60**i for i,x in enumerate(reversed(time.split(':'))))
time2sec('1:50'), time2sec('00:01:30')
(110, 90)

With that sorted, I can now write the function that will convert from SRT to TSV.

srt2tsv

def srt2tsv(
    srt:str, # SRT subtitle text to parse
)->str: # TSV string with columns: timerange, speaker, text

Convert SRT format to TSV string.

Exported source 
def srt2tsv(
    srt:str, # SRT subtitle text to parse
)->str: # TSV string with columns: timerange, speaker, text
    "Convert SRT format to TSV string."
    blocks = re.split(r'\n\n+', srt.strip())
    res = ''
    for b in blocks:
        lines = b.strip().splitlines()
        if len(lines)<3: continue
        parts = lines[1].split(' --> ')
        start = time2sec(parts[0].replace(',', '')[:-3])
        end = time2sec(parts[1].replace(',', '')[:-3])
        m = re.match(r'\[(.*?)\]\s*(.*)', lines[2]); spk,text = (m[1], m[2]) if m else ('', lines[2])
        res += f"{start}→{end}\t{spk}\t{text}\n"
    return res
print(srt2tsv(tr.read_text())[:500])
20→42   T   I'm wired.
43→53   T   It's running.
54→56   SN  (inaudible) please turn off the air.
57→66   T   I'll make it a bit warmer.
67→76   T   Well not really, but if you really have to I suppose. Okay.
77→80   SN  (inaudible) go out to my locker and get my maths book?
81→82   T   Oh, you won't need it today.
83→97   SN  Oh, okay.
98→98   T   Are you going to be sitting down?
99→100  SN  I haven't (inaudible)
101→102 T   Oh, then you've got to go outside then.
103→105 S   (inaudible) books (inaudible)
106→135 T   There'll be some

I can now try create an entry in the video table.

dt = db.t; dt
frame, run, run_frame, video
for t in dt: t.dataclass()
v = dt.video.insert(title=dp.stem, transcript=srt2tsv(tr.read_text()),
                   length=len(dp.ls()), path=str(dp), sample_rate=1)
type(v), v.transcript[:50]
(fastlite.core.Video,
 "20→42\tT\tI'm wired.\n43→53\tT\tIt's running.\n54→56\tSN\t")

Now I want to attempt creating some entries in the frame table. For that, I’ll have to fetch the corresponding subtitle for each frame.

db.t.frame.schema
CREATE TABLE [frame] (
   [id] INTEGER PRIMARY KEY,
   [video_id] INTEGER REFERENCES [video]([id]) ON UPDATE CASCADE ON DELETE CASCADE,
   [frame_number] INTEGER,
   [subtitle] TEXT
)
dp
Path('../../data/timss/M-AU1')
dp.ls()[:5]
[Path('../../data/timss/M-AU1/frame_001886.jpg'), Path('../../data/timss/M-AU1/frame_001303.jpg'), Path('../../data/timss/M-AU1/frame_001378.jpg'), Path('../../data/timss/M-AU1/frame_002173.jpg'), Path('../../data/timss/M-AU1/frame_002259.jpg')]
filter_paths(dp.ls(), chs='.', comp='stem')[:3]
[Path('../../data/timss/M-AU1/frame_001886.jpg'), Path('../../data/timss/M-AU1/frame_001303.jpg'), Path('../../data/timss/M-AU1/frame_001378.jpg')]
filter_paths(dp.ls(), chs='txt', comp='suffix')[:3]
[Path('../../data/timss/M-AU1/frame_001886.jpg'), Path('../../data/timss/M-AU1/frame_001303.jpg'), Path('../../data/timss/M-AU1/frame_001378.jpg')]

To connect each frame to their corresponding transcript, I’ll be converting the TSV transcripts to Python dictionaries. If the video had a subtitle between 3 seconds and 7 seconds. Then all frames sampled in that range should have the corresponding same subtitle.

tsv2dict

def tsv2dict(
    tsv:str, # TSV string with columns: timerange, speaker, text
    max_len:int, # Max frame number to cover
    default:str='', # Value for timestamps without subtitles
)->dict: # {second: "speaker: text"} lookup dict

Convert TSV string to a {second: subtitle} lookup dict.

Exported source 
def tsv2dict(
    tsv:str, # TSV string with columns: timerange, speaker, text
    max_len:int, # Max frame number to cover
    default:str='', # Value for timestamps without subtitles
)->dict: # {second: "speaker: text"} lookup dict
    "Convert TSV string to a {second: subtitle} lookup dict."
    lookup = {}
    for line in tsv.splitlines():
        if not line.strip(): continue
        tr, spk, text = line.split('\t')
        start, end = tr.split('→')
        start, end = int(start), int(end)
        entry = f"{spk}: {text}" if spk else text
        for s in range(start, end+1): lookup[s] = entry
    end = max(max(lookup) if lookup else 0, max_len)
    for s in range(end+1): lookup.setdefault(s, default)
    return lookup
r = srt2tsv(tr.read_text())

Double checking whether I got what I wanted…

from itertools import islice
list(islice(tsv2dict(r, 2000).items(), 10))
[(20, "T: I'm wired."),
 (21, "T: I'm wired."),
 (22, "T: I'm wired."),
 (23, "T: I'm wired."),
 (24, "T: I'm wired."),
 (25, "T: I'm wired."),
 (26, "T: I'm wired."),
 (27, "T: I'm wired."),
 (28, "T: I'm wired."),
 (29, "T: I'm wired.")]
lookup = tsv2dict(srt2tsv(tr.read_text()), len(dp.ls()))
lookup[20], lookup[100]
("T: I'm wired.", "SN: I haven't (inaudible)")

I can go ahead an produce an entry.

fpaths = filter_paths(Path(v.path).ls(), chs='txt', comp='suffix')
fpaths = filter_paths(fpaths, chs='.', comp='stem').sorted(key=~Self.stem.split('_'))
fpaths[:5]
[Path('../../data/timss/M-AU1/frame_000001.jpg'), Path('../../data/timss/M-AU1/frame_000002.jpg'), Path('../../data/timss/M-AU1/frame_000003.jpg'), Path('../../data/timss/M-AU1/frame_000004.jpg'), Path('../../data/timss/M-AU1/frame_000005.jpg')]
fp = fpaths[0]; fp
Path('../../data/timss/M-AU1/frame_000001.jpg')
fnum = int(fp.stem.split('_')[1]); fnum
1
db.t.frame.insert(video_id=v.id, frame_number=fnum, subtitle=tsv2dict(v.transcript, v.length)[fnum])
Frame(id=1, video_id=1, frame_number=1, subtitle='')

I can wrap this all up into a function that will perform this on all videos and all frames.

populate_db

def populate_db(
    db:Database, # Database to populate
    paths:list, # List of video directories
    sample_rate:int=1, # Sampling rate for frames
    trans_suffix:str='txt', # Transcript file suffix
)->None:

Populate video and frame tables from a list of video directories.

Exported source 
from fastprogress.fastprogress import master_bar, progress_bar
Exported source 
def populate_db(
    db:Database, # Database to populate
    paths:list[Path], # List of video directories
    sample_rate:int=1, # Sampling rate for frames
    trans_suffix:str='txt', # Transcript file suffix
)->None:
    "Populate video and frame tables from a list of video directories."
    t = db.t
    filter_trans = partial(filter_paths, chs=trans_suffix, comp='suffix', negate=False)
    filter_dots = partial(filter_paths, chs='.', comp='stem')
    for p in (mb:=master_bar(paths)):
        mb.main_bar.comment = f'video {p.stem}'
        tr = filter_trans(p.ls())[0]
        v = t.video.insert(title=p.stem, transcript=srt2tsv(tr.read_text()),
                           length=len(p.ls()), path=str(p), sample_rate=sample_rate)
        lookup = tsv2dict(v.transcript, v.length)
        fpaths = filter_dots(filter_paths(Path(v.path).ls(), chs=trans_suffix, comp='suffix')).sorted(key=~Self.stem.split('_'))
        for fp in (pb:=progress_bar(fpaths, parent=mb)):
            fnum = int(fp.stem.split('_')[1])
            t.frame.insert(video_id=v.id, frame_number=fnum, subtitle=lookup[fnum])
!rm db.db
db = init_db()
populate_db(db, dpaths)
len(db.t.video()), len(db.t.frame())
(52, 149880)

VLM

In this section I create helper functions for working with the LLM, using the fastllm library. fastllm is still in active development, and the helpers here work for version 0.0.36.

Exported source 
from aidialog.msg_parts import Msg, Part, PartType
?Msg
def Msg(
    role:str, content:List
)->None:
    "A normalized message."

File: /usr/local/lib/python3.12/site-packages/aidialog/msg_parts.py; line: 55

Type: type

?Part
def Part(
    type:str, text:str=None, data:dict=None
)->None:
    "A normalized content part."

File: /usr/local/lib/python3.12/site-packages/aidialog/msg_parts.py; line: 21

Type: type

?PartType
def PartType(
    *args, **kwds
):

File: /usr/local/lib/python3.12/site-packages/aidialog/msg_parts.py

Type: EnumType

user

def user(
    txt:str, # User message text
    img:str | None=None, # Base64 data URL of image to include
)->Msg: # User message with optional image

Build a user message with optional image.

Exported source 
def user(
    txt:str, # User message text
    img:str|None=None, # Base64 data URL of image to include
)->Msg: # User message with optional image
    "Build a user message with optional image."
    if img is None: return Msg(role='user', content=[Part(PartType.text, text=txt)])
    else: return Msg(role='user', content=[Part(PartType.input_image, text=img), Part(PartType.text, text=txt)])
user('你好')

Msg

  • role: user

Part (text)

你好

  • data: None

assistant

def assistant(
    txt:str, # Assistant message text
    citations:list=None, # Optional citation list
)->Msg: # Assistant message

Build an assistant message.

Exported source 
def assistant(
    txt:str, # Assistant message text
    citations:list=None, # Optional citation list
)->Msg: # Assistant message
    "Build an assistant message."
    return Msg(role='assistant', content=[Part(PartType.text, text=txt, data={'citations': citations} if citations else None)])
assistant('嗨')

Msg

  • role: assistant

Part (text)

嗨

  • data: None

Exported source 
from fastllm.acomplete import acomplete
?acomplete
async def acomplete(
    msgs, model, api_name:NoneType=None, vendor_name:NoneType=None, api_key:NoneType=None, base_url:NoneType=None,
    xtra_body:NoneType=None, xtra_hdrs:NoneType=None, stream:bool=False, stop_callables:NoneType=None, retries:int=2,
    retry_delay:float=0.5, system:NoneType=None, max_tokens:NoneType=None, temperature:NoneType=None,
    tools:NoneType=None, tool_choice:NoneType=None, reasoning_effort:NoneType=None, web_search_options:NoneType=None
):
    "Unified completion across different APIs."

File: /usr/local/lib/python3.12/site-packages/fastllm/acomplete.py; line: 170

Type: function

Exported source 
from cachy import enable_cachy, disable_cachy

Cachy caches http requests and stores the responses locally. This reduces spend and avoids the need to also repeatedly wait for the remote server to process my request.

I’ll now send a test query.

enable_cachy()
await acomplete([user('hi')], 'deepseek-v4-flash', vendor_name='deepseek')
Thinking

好的，用户只发了一个“hi”，这是非常简单的打招呼。用户可能刚进入对话，想测试我是否在线或者开始一个友好的交流。深层需求应该是希望得到热情、友好的回应，开启一次对话。我不需要复杂分析，直接礼貌问候并表达乐于助人的态度，用开放式的邀请让用户提出具体问题。想到了用“你好！”开头，加上表情符号显得亲切，然后自我介绍并说明能力范围，最后用提问引导对话继续。

你好！很高兴见到你！😊

有什么我可以帮你的吗？无论是回答问题、帮你整理信息、提供创作灵感，还是聊聊天，我都很乐意陪你一起。你只需告诉我需求，剩下的交给我！

  • model: deepseek-v4-flash
  • finish_reason: stop
  • usage: Usage(prompt_tokens=5, completion_tokens=143, total_tokens=148, cached_tokens=0, cache_creation_tokens=0, reasoning_tokens=97, raw={'prompt_tokens': 5, 'completion_tokens': 143, 'total_tokens': 148, 'prompt_tokens_details': {'cached_tokens': 0}, 'completion_tokens_details': {'reasoning_tokens': 97}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 5})
Exported source 
from fastllm.types import Completion

I’m now creating a function that’ll stream in the response with the Completion object. We’ll see the response here as it generates.

stream

async def stream(
    msgs:list | None=None, # Messages to send
    model:str='', # Model name (e.g. 'deepseek-v4-flash')
    max_think:float=inf, # Max thinking tokens to display
    usage:bool=True, # Show usage info in output
    display:bool=True, # Print text/thinking as it arrives
    **kwargs
)->Completion: # Return the final completion

Stream a response, printing text/thinking as it arrives. Returns the final completion.

Exported source 
async def stream(
    msgs:list|None=None,  # Messages to send
    model:str='',  # Model name (e.g. 'deepseek-v4-flash')
    max_think:float=float('inf'),  # Max thinking tokens to display
    usage:bool=True,  # Show usage info in output
    display:bool=True,  # Print text/thinking as it arrives
    **kwargs,  # Passed to acomplete
) -> Completion:  # Return the final completion
    "Stream a response, printing text/thinking as it arrives. Returns the final completion."
    assert msgs is not None, 'no messages provided'
    assert model!='', 'no model name provided'
    think_cnt, seen_txt = 0, False
    async for o in await acomplete(msgs, model, stream=True, **kwargs):
        if not isinstance(o, Completion) and display:
            if isinstance(o, Part) and o.type==PartType.thinking and think_cnt<max_think: print('🤔', end='', flush=True)
            if isinstance(o, Part) and o.type==PartType.text and (txt:=o.text): print(f"{'\n\n' if not seen_txt else ''}{txt}", end='', flush=True) or not seen_txt and (seen_txt:=True)
            think_cnt+=1
    if display: print()
    return o
r = await stream([user('hi')], 'deepseek-v4-flash', vendor_name='deepseek')

img2b64

def img2b64(
    path:Path, # Path to image file
)->str: # Base64 data URL

Encode an image file as a base64 data URL.

Exported source 
from base64 import b64encode
Exported source 
def img2b64(
    path:Path, # Path to image file
)->str: # Base64 data URL
    "Encode an image file as a base64 data URL."
    return 'data:image/png;base64,'+b64encode(Path.read_bytes(path)).decode()

And a helper function to more easily pass images to the VLM.

r = await stream([user('what do ye elf eyes see', img2b64(Path('./test.jpg')))], 'bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high')
r.usage.raw
{'prompt_tokens': 1343,
 'completion_tokens': 702,
 'total_tokens': 2045,
 'cost': 0.00173975,
 'is_byok': False,
 'prompt_tokens_details': {'cached_tokens': 0,
  'cache_write_tokens': 0,
  'audio_tokens': 0,
  'video_tokens': 0},
 'cost_details': {'upstream_inference_cost': 0.00173975,
  'upstream_inference_prompt_cost': 0.00033575,
  'upstream_inference_completions_cost': 0.001404},
 'completion_tokens_details': {'reasoning_tokens': 559,
  'image_tokens': 0,
  'audio_tokens': 0}}

session

def session(
    msgs:list | None=None, model:str='', max_think:float=inf, usage:bool=True, display:bool=True, **kwargs
):

Create a stream partial with preset model/kwargs.

Exported source 
@delegates(stream, keep=True)
def session(
    **kwargs,
):
    "Create a stream partial with preset model/kwargs."
    return partial(stream, **kwargs)

And another helper function to create sessions more easily without having to write the same params each time.

Run

Now it’s time to run the VLM across all frames of a video. I’ll begin small by working on a single frame: fetch its subtitle from the database, attach the image, and see what comes back.

frame = db.t.frame.selectone('video_id=? AND frame_number=?', (v.id, int(fp.stem.split('_')[1]))); frame
Frame(id=1, video_id=1, frame_number=1, subtitle='')
sub = frame.subtitle or '[No speech]'
prompt = f"Describe what you see\n\nSubtitle: {sub}"
r = await stream([user(prompt, img=img2b64(fp))], 'bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high'); r.message.text[:200]
'This is a low-resolution (blurry, likely older recorded) scene of a school classroom, with no spoken audio:\nThe space is a standard classroom, with rows of student desks holding seated school-aged chi'

Running a single frame works nicely, but when scaling to hundreds of frames across a video, problems emerge:

  • Rate limits (429): the API throttles concurrent requests, so some frames will fail
  • Connection timeouts: the provider may be temporarily unreachable
  • Insufficient credits (402): if the budget runs out mid-run, we need to stop gracefully without losing completed work

To handle these problems, I’ll wrap the single-frame call that catches APIErrors, stores successful results to the run_frame table via upsert, and flags any failed frame as skipped for later retry.

Exported source 
from fastspec.errors import APIError
Exported source 
async def _process_frame(
    p:Path, # Path to frame image
    db, # Database object
    session, # Stream session partial
    run_id:int, # Run ID to store results under
    video_id:int, # Video ID the frame belongs to
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt (e.g. 'description')
    include_subs:bool=True, # Whether to append subtitle to prompt
):
    "Run VLM on a single frame and store the result."
    fnum = int(p.stem.split('_')[1])
    frame = db.t.frame.selectone('video_id=? AND frame_number=?', (video_id, fnum))
    if include_subs:
        sub = frame.subtitle or '[No speech]'
        prompt = f"{prompt}\n\nSubtitle: {sub}"
    try:
        r = await session([user(prompt, img=img2b64(p))])
        u = r.usage.raw if r.usage else {}
        db.t.run_frame.upsert(run_id=run_id, frame_id=frame.id, type=prompt_type, prompt=prompt, description=r.message.text, usage={**u, 'cost': u.get('cost', 0.0)}, skipped=False)
    except Exception as e:
        if not isinstance(e, APIError): raise
        sc = getattr(e,'status_code',None)
        print(f'!! APIError sc={sc} frame {fnum}: {e}')
        if sc==402:
            print(f'!! Insufficient credits — stopped at frame {fnum}')
            raise
        db.t.run_frame.upsert(run_id=run_id, frame_id=frame.id, type=prompt_type, prompt=prompt, description='', skipped=True)
        print(f'!! Skipped: frame {fnum}')

Throughput is a bottleneck. Running frames one at a time across hundreds of frames would take too long. I’ll query multiple requests in parallel with fastcore’s parallel_async_gen.

?parallel_async_gen
def parallel_async_gen(
    f, items, *args, n_workers:int=16, pause:int=0, timeout:NoneType=None, return_exceptions:bool=False,
    cancel_on_exit:bool=True, **kwargs
):
    "Yield `(index,result)` pairs as `f` applied to each of `items` completes, in completion order"

File: ~/.local/lib/python3.12/site-packages/fastcore/parallel.py; line: 154

Type: function

s = session(model='bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high'); s
functools.partial(<function stream>, model='bytedance-seed/seed-2.0-lite', vendor_name='openrouter', reasoning_effort='high')
run = db.t.run.insert(deploy_time='', video_id=v.id, model='bytedance-seed/seed-2.0-lite'); run
Run(id=1, deploy_time='', finish_time=None, start_time=None, total_duration=None, video_id=1, model='bytedance-seed/seed-2.0-lite', usage=None, num_frames=None, start_sec=None, end_sec=None, step=None, description=None)
run_id = run.id
video_id, prompt_type = v.id, 'description'
async for i,r in parallel_async_gen(_process_frame, fpaths[:4], db, s, run_id, video_id, prompt, prompt_type, True, n_workers=4, pause=0.1): print(f'frame {i} done')
view_col(db.t.run_frame, 'description', where='run_id=?', where_args=(run_id,))

frame 0 done

frame 1 done


frame 2 done
frame 3 done
['This is a low-resolution, blurry clip filmed from the back of a school classroom, looking toward the front of the room.\nNearly all students, viewed from behind as they face the front of class, are seated at student desks, oriented toward the front of the room. At the front, a male teacher wearing a collared shirt and tie leans over a student at a front desk, interacting with or assisting that child.\nStandard classroom features fill the space: large windows on the left wall let in natural light, a retractable overhead projector screen is mounted to the center wall above a whiteboard, with colorful educational banners hung above the whiteboard. Fluorescent lights are set into the ceiling, and a brick wall section is visible on the right side of the room. There is no speech or audio in this clip.', "This is a quiet scene inside a school classroom, captured from a static in-room camera, with no speech/audio.\nThe space is a standard classroom: student desks and chairs including small green student seating are arranged in rows facing the front of the room. Natural light filters through a large window on the left wall. The front wall has a whiteboard, with small colorful educational posters mounted above it, plus a large rolled-down projection screen to the whiteboard's right; a brick accent wall on the right has a security camera mounted to it, which matches the footage's camera perspective.\n\nAt the front of the room, an adult man who appears to be the teacher, with light hair, a grey long-sleeve shirt, and a dark strap across his chest leans over a front desk, interacting with a seated male student in a white shirt at that desk. A small number of other students are seated at their desks across the room, facing the front of the class: including a blonde girl in a dark top, and another girl with dark hair in a ponytail wearing a white shirt. The space feels calm, like an ongoing quiet classroom session.", 'This is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.', "This is an elementary school classroom, hosting a small tech demonstration for students:\n1.  At the front of the room, a tall light-haired adult man (wearing a light grey long-sleeve shirt and dark trousers) stands at the head of the class, behind a small robotic arm set up on a student desk. A young boy (seen from behind, in a white school shirt) is at this desk interacting with the robotic arm.\n2.  Other students are seated at grouped classroom desks (with standard green plastic school chairs) all facing the front to watch the demo: a girl with a blonde ponytail in a dark top sits to the boy's left, and another student's head is visible in the bottom right corner, also observing.\n3.  The background has typical classroom features: a back wall with a whiteboard, a pulled-down projector screen to the whiteboard's right, colorful educational banners mounted above the whiteboard, a closed white door on the far left, and a ledge under the whiteboards holding a small supply box."]

Time to wrap that up.

Exported source 
async def _run_batch(
    frames:L, # List of frame paths to process
    db, # Database object
    session, # Stream session partial
    run_id:int, # Run ID to store results under
    video_id:int, # Video ID the frames belong to
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    include_subs:bool, # Whether to append subtitle to prompt
    n_workers:int, # Number of parallel workers
    pause:float, # Seconds to pause between dispatches
):
    "Run _process_frame across a batch of frame paths in parallel."
    done = 0
    async for i,r in parallel_async_gen(_process_frame, frames, db, session, run_id, video_id,
                                         prompt, prompt_type, include_subs,
                                         n_workers=n_workers, pause=pause):
        done += 1; print(f'\r{done}/{len(frames)}', end='', flush=True)
    print()

Skipped frames are flagged with skipped=1. I’ll query those rows once again, repeating until they succeed.

I’ll manually mark a couple of frames as skipped.

for rf in L(db.t.run_frame('run_id=?', (run_id,)))[:2]:
    db.t.run_frame.update(run_id=rf.run_id, frame_id=rf.frame_id, type=rf.type, skipped=True)
L(db.t.run_frame('run_id=? AND skipped=1', (run_id,))).map(attrgetter('frame_id'))
[1, 2]
rl_rows = L(db.t.run_frame('run_id=? AND skipped=1', (run_id,)))
rl_fids = rl_rows.map(attrgetter('frame_id')); rl_fids
[1, 2]
ph = ','.join('?'*len(rl_fids)); ph
'?,?'
rl_frames = L(db.t.frame(f'id IN ({ph})', tuple(rl_fids))); rl_frames
[Frame(id=1, video_id=1, frame_number=1, subtitle=''), Frame(id=2, video_id=1, frame_number=2, subtitle='')]
rl_paths = rl_frames.map(lambda f: Path(v.path)/f'frame_{f.frame_number:06d}.jpg'); rl_paths
[Path('../../data/timss/M-AU1/frame_000001.jpg'), Path('../../data/timss/M-AU1/frame_000002.jpg')]
Exported source 
async def _retry_skipped(
    db, # Database object
    video_path:Path, # Path to the video's frame directory
    run_id:int, # Run ID to retry frames for
    video_id:int, # Video ID the frames belong to
    session, # Stream session partial
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    include_subs:bool, # Whether to append subtitle to prompt
    n_workers:int, # Number of parallel workers
    pause:float, # Seconds to pause between dispatches
    max_retries:int=2, # Max retry attempts for skipped frames
):
    "Retry skipped frames up to max_retries times."
    for attempt in range(max_retries):
        rl_rows = L(db.t.run_frame('run_id=? AND skipped=1', (run_id,)))
        if not rl_rows: break
        rl_fids = rl_rows.map(attrgetter('frame_id'))
        ph = ','.join('?'*len(rl_fids))
        rl_frames = L(db.t.frame(f'id IN ({ph})', tuple(rl_fids)))
        rl_paths = rl_frames.map(lambda f: video_path/f'frame_{f.frame_number:06d}.jpg')
        print(f'!! Retrying {len(rl_paths)} skipped frames (attempt {attempt+1}/{max_retries})')
        try: await _run_batch(rl_paths, db, session, run_id, video_id, prompt, prompt_type, include_subs, n_workers, pause)
        except Exception as e:
            if not isinstance(e, APIError): raise
            print(f'!! APIError in retry: {e}')
            break

With the run complete, I want to see how much it cost. Each run_frame row stores a usage JSON string from the API response. I’ll look at one.

rows = db.t.run_frame('run_id=? AND skipped=0', (run_id,))
if rows: print(loads(rows[0].usage))
else: print('No successful frames yet')
{'prompt_tokens': 1369, 'completion_tokens': 954, 'total_tokens': 2323, 'cost': 0.00225025, 'is_byok': False, 'prompt_tokens_details': {'cached_tokens': 0, 'cache_write_tokens': 0, 'audio_tokens': 0, 'video_tokens': 0}, 'cost_details': {'upstream_inference_cost': 0.00225025, 'upstream_inference_prompt_cost': 0.00034225, 'upstream_inference_completions_cost': 0.001908}, 'completion_tokens_details': {'reasoning_tokens': 735, 'image_tokens': 0, 'audio_tokens': 0}}
usgs = L(db.t.run_frame('run_id=? AND skipped=0', (run_id,))).map(lambda r: loads(r.usage))
usgs.map(lambda u: tuple(u.keys())).unique()
[('prompt_tokens', 'completion_tokens', 'total_tokens', 'cost', 'is_byok', 'prompt_tokens_details', 'cost_details', 'completion_tokens_details')]
sum(u['cost'] for u in usgs)
0.0044785
sum(u['prompt_tokens'] for u in usgs)
2738
u = usgs[0]; u['prompt_tokens_details']
{'cached_tokens': 0,
 'cache_write_tokens': 0,
 'audio_tokens': 0,
 'video_tokens': 0}
sum(u['prompt_tokens_details']['cached_tokens'] for u in usgs)
0

Scalar fields sum directly; nested dict fields need to be unpacked first. Now I’ll aggregate usage across all frames in a run, and stamp the run with its finish time and total duration.

Exported source 
from datetime import datetime
from zoneinfo import ZoneInfo
Exported source 
tz = ZoneInfo('Asia/Hong_Kong')

I’ll also add a header box at the start of each run, so the notebook output frames the run visually.

compute_usage

def compute_usage(
    db, # Database object
    run_id:int, # Run ID to aggregate usage for
)->str: # JSON string of aggregated usage stats

Aggregate usage stats across all frames in a run. Returns JSON string.

Exported source 
def _run_header(
    run, # Run record
    model:str, # Model name
    start:int, # Start second
    stop:int, # Stop second
    step:int, # Frame step interval
    cache:bool, # Whether cache is enabled
):
    "Print run header box."
    print(f'╭─ Run #{run.id} ═══════════════════════════════╮\n│ Model    {model}\n│ Start    {start}\n│ Stop     {stop}\n│ Step     {step}\n│ Frames   {(stop-start)//step}\n│ Cache    {cache}\n│ Time     {datetime.now(tz).strftime("%H:%M:%S")}\n╰──────────────────────────────────────────────╯')
Exported source 
def compute_usage(
    db, # Database object
    run_id:int, # Run ID to aggregate usage for
)->str: # JSON string of aggregated usage stats
    "Aggregate usage stats across all frames in a run. Returns JSON string."
    usgs = L(db.t.run_frame('run_id=? AND skipped=0', (run_id,))).map(lambda r: loads(r.usage))
    if not usgs: return '{}'
    tot = {k: ({k2:0 for k2 in v} if isinstance(v,dict) else 0) for k,v in usgs[0].items()}
    for u in usgs:
        for k,v in u.items():
            if isinstance(v,dict):
                for k2,v2 in v.items(): tot[k][k2] += v2
            else: tot[k] += v
    return dumps(tot)

def _finish_run(
    db, # Database object
    run_id:int, # Run ID to finalize
)->Run: # Updated run record
    "Update run with finish time and usage, print summary box."
    finish = datetime.now(tz)
    run = db.t.run[run_id]
    start = datetime.fromisoformat(run.deploy_time)
    elapsed = finish - start
    db.t.run.update(id=run_id, finish_time=finish, start_time=start.isoformat(), total_duration=str(elapsed).split('.')[0], usage=compute_usage(db, run_id))
    run = db.t.run[run_id]
    tot = dict2obj(loads(run.usage))
    cost = getattr(tot, 'cost', 0)
    print(f'╭─ Run #{run.id} Complete ═════════════════════╮\n│ Finish   {datetime.fromisoformat(run.finish_time).strftime("%H:%M:%S")}\n│ Elapsed  {str(elapsed).split(".")[0]}\n│ Cost     ${cost:.4f} (HKD {cost*7.84:.2f})\n╰──────────────────────────────────────────────╯')
    return run
Exported source 
from typing import Callable
from fastprogress.fastprogress import NBMasterBar as master_bar

Finally, I’ll tie everything together, batch processing, retry, usage tracking, and visual framing, into a single deploy_run function. It takes a video ID, sets up the run record, dispatches frames in parallel, retries rate-limited ones, and stamps the run with its finish time and total cost.

deploy_run

async def deploy_run(
    video_id:int, # Video ID to process
    db:Database, # Database object
    session:Callable, # Stream session partial
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    start:int=0, # Start frame index
    stop:int | None=None, # Stop frame index (defaults to last)
    step:int=1, # Frame step interval
    cache:bool=False, # Whether to use HTTP cache
    include_subs:bool=True, # Whether to append subtitle to prompt
    n_workers:int=8, # Number of parallel workers
    pause:float=3, # Seconds to pause between dispatches
    max_retries:int=2, # Max retry attempts for skipped frames
)->Run: # Completed run record with usage stats

Run a single prompt across a range of frames, storing results in the database.

Exported source 
from fastcore.parallel import parallel_async_gen
async def deploy_run(
    video_id:int, # Video ID to process
    db:Database, # Database object
    session:Callable, # Stream session partial
    prompt:str, # Prompt template for the VLM
    prompt_type:str, # Type label for this prompt
    start:int=0, # Start frame index
    stop:int|None=None, # Stop frame index (defaults to last)
    step:int=1, # Frame step interval
    cache:bool=False, # Whether to use HTTP cache
    include_subs:bool=True, # Whether to append subtitle to prompt
    n_workers:int=8, # Number of parallel workers
    pause:float=3, # Seconds to pause between dispatches
    max_retries:int=2, # Max retry attempts for skipped frames
)->Run: # Completed run record with usage stats
    "Run a single prompt across a range of frames, storing results in the database."
    video = db.t.video[video_id]
    fpath = L(Path(video.path).glob('frame_*.jpg')).sorted(key=~Self.stem.split('_'))
    if stop is None: stop = len(fpath)
    if not cache: disable_cachy(); print('!! Cache disabled')
    else: print('!! Using cache')

    run = db.t.run.insert(deploy_time=datetime.now(tz), video_id=video_id, model=session.keywords['model'], num_frames=(stop-start)//step, start_sec=start, end_sec=stop, step=step)
    _run_header(run, session.keywords['model'], start, stop, step, cache)

    frames = fpath[start:stop:step]
    try: await _run_batch(frames, db, session, run.id, video_id, prompt, prompt_type, include_subs, n_workers, pause)
    except Exception as e:
        if not isinstance(e, APIError): raise
        print(f'!! APIError in deploy: {e}')
    await _retry_skipped(db, Path(video.path), run.id, video_id, session, prompt, prompt_type, include_subs, n_workers, pause, max_retries)

    if not cache: enable_cachy(); print('!! Cache enabled')
    return _finish_run(db, run.id)

Summary

Each frame now has a VLM description stored in the database. But a video with hundreds of frames means hundreds of separate descriptions. I want to now take a window of frames and compress it into a single coherent passage. Then for each of those passages, I stich them together into a full narrative.

To do that, I first need to pull the frame descriptions back out of the database, joined with their frame numbers so I can order them chronologically. A run might cover a subset of frames, and I might want to merge multiple runs for the same video, so the query should accept either a single run ID or a list.

I’ll also need to count tokens, since I’ll be feeding windows of frame descriptions into the LLM and want to track how much gets compressed. For that, tiktoken gives me an encoder that approximates what the model sees.

Exported source 
from tiktoken import encoding_for_model

Let me start by querying the run frames for the run I just did, and joining with the frame table to get frame numbers. First I’ll see what the raw rows look like:

rows = L(db.t.run_frame('run_id=? AND skipped=0', (run_id,)))
rows[:3]
[Run_Frame(run_id=1, frame_id=3, type='description', system_prompt=None, prompt='Describe what you see\n\nSubtitle: [No speech]\n\nSubtitle: [No speech]', description='This is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.', usage='{"prompt_tokens": 1369, "completion_tokens": 954, "total_tokens": 2323, "cost": 0.00225025, "is_byok": false, "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0, "audio_tokens": 0, "video_tokens": 0}, "cost_details": {"upstream_inference_cost": 0.00225025, "upstream_inference_prompt_cost": 0.00034225, "upstream_inference_completions_cost": 0.001908}, "completion_tokens_details": {"reasoning_tokens": 735, "image_tokens": 0, "audio_tokens": 0}}', skipped=0), Run_Frame(run_id=1, frame_id=4, type='description', system_prompt=None, prompt='Describe what you see\n\nSubtitle: [No speech]\n\nSubtitle: [No speech]', description="This is an elementary school classroom, hosting a small tech demonstration for students:\n1.  At the front of the room, a tall light-haired adult man (wearing a light grey long-sleeve shirt and dark trousers) stands at the head of the class, behind a small robotic arm set up on a student desk. A young boy (seen from behind, in a white school shirt) is at this desk interacting with the robotic arm.\n2.  Other students are seated at grouped classroom desks (with standard green plastic school chairs) all facing the front to watch the demo: a girl with a blonde ponytail in a dark top sits to the boy's left, and another student's head is visible in the bottom right corner, also observing.\n3.  The background has typical classroom features: a back wall with a whiteboard, a pulled-down projector screen to the whiteboard's right, colorful educational banners mounted above the whiteboard, a closed white door on the far left, and a ledge under the whiteboards holding a small supply box.", usage='{"prompt_tokens": 1369, "completion_tokens": 943, "total_tokens": 2312, "cost": 0.00222825, "is_byok": false, "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0, "audio_tokens": 0, "video_tokens": 0}, "cost_details": {"upstream_inference_cost": 0.00222825, "upstream_inference_prompt_cost": 0.00034225, "upstream_inference_completions_cost": 0.001886}, "completion_tokens_details": {"reasoning_tokens": 726, "image_tokens": 0, "audio_tokens": 0}}', skipped=0)]

Now let me join with the frame table to get frame numbers, and group by frame number.

fids = sorted(set(r.frame_id for r in rows))
fn_map = {f.id: f.frame_number for f in db.t.frame(f'id IN ({",".join("?"*len(fids))})', tuple(fids))}
fn_map
{3: 3, 4: 4}
grouped = rows.groupby(lambda r: fn_map[r.frame_id])
sorted(grouped.keys())
[3, 4]
Exported source 
def _get_runframes(
    db:Database, # Database object
    run_ids:int|list[int], # Single run ID or list of run IDs
    start:int|None=None, # Filter frames at or after this second
    stop:int|None=None, # Filter frames at or before this second
)->L: # Grouped (frame_number, L[RunFrame]) pairs sorted by frame_number
    "Query runframes for given run(s), join with frame table for frame_number, group by frame_number."
    if isinstance(run_ids, int): run_ids = [run_ids]
    ph = ','.join('?' * len(run_ids))
    rows = L(db.t.run_frame(f'run_id IN ({ph}) AND skipped=0', tuple(run_ids)))
    fids = sorted(set(r.frame_id for r in rows))
    fn_map = {f.id: f.frame_number for f in db.t.frame(f'id IN ({",".join("?"*len(fids))})', tuple(fids))}
    grouped = rows.groupby(lambda r: fn_map[r.frame_id])
    fnums = sorted(grouped.keys())
    if start is not None: fnums = [f for f in fnums if f >= start]
    if stop is not None: fnums = [f for f in fnums if f <= stop]
    return L((fn, L(grouped[fn])) for fn in fnums)
?encoding_for_model
def encoding_for_model(
    model_name:str
)->Encoding:
    "Returns the encoding used by a model.

    Raises a KeyError if the model name is not recognised.
    "

File: /usr/local/lib/python3.12/site-packages/tiktoken/model.py; line: 113

Type: function

Now I need to format those grouped runframes into a single text block the LLM can read. Each frame’s description should be prefixed with its timestamp, and multiple descriptions for the same frame (e.g. from different prompt types) separated by --. This is a “window”. A unit I’ll feed into the LLM for summarization.

Let me take the first grouped runframe and see what I’m working with.

rfs = _get_runframes(db, run_id, 0, 10)
i, rf = rfs[0]; i, rf
(3,
 [Run_Frame(run_id=1, frame_id=3, type='description', system_prompt=None, prompt='Describe what you see\n\nSubtitle: [No speech]\n\nSubtitle: [No speech]', description='This is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.', usage='{"prompt_tokens": 1369, "completion_tokens": 954, "total_tokens": 2323, "cost": 0.00225025, "is_byok": false, "prompt_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0, "audio_tokens": 0, "video_tokens": 0}, "cost_details": {"upstream_inference_cost": 0.00225025, "upstream_inference_prompt_cost": 0.00034225, "upstream_inference_completions_cost": 0.001908}, "completion_tokens_details": {"reasoning_tokens": 735, "image_tokens": 0, "audio_tokens": 0}}', skipped=0)])

Each frame gets a timestamp header. Let me build that prefix.

prefix = f'TIMESTAMP {i}s\n'; prefix + len(prefix.strip())*'=' + '\n'
'TIMESTAMP 3s\n============\n'

Then each description entry under that header, separated by --.

r = rf[0]; '\n--\n' + f'{r.type.upper()}\n{r.description}' + '\n\n'
'\n--\nDESCRIPTION\nThis is a classroom scene, set in a school:\n1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.\n2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and several boys in white school uniform shirts, all oriented toward the man at the front.\n3.  Additional classroom details: above the front whiteboards, three colorful decorative educational banners are mounted. A large window on the left side of the frame lets in natural light, next to a white interior door. The right side of the back wall has an exposed brick column, with a security camera mounted to it, and a framed poster hung above the projection screen. This is a small class, with only a handful of students visible in this view.\n\n'

Now I can assemble a few frames into a window and see what the full text looks like.

window = ''
for i,rf in rfs[:3]:
    prefix = f'TIMESTAMP {i}s\n'
    window += prefix + len(prefix.strip())*'=' + '\n'
    for r in rf:
        window += '\n--\n' + f'{r.type.upper()}\n{r.description}' + '\n\n'
print(window[:500])
TIMESTAMP 3s
============

--
DESCRIPTION
This is a classroom scene, set in a school:
1.  At the front of the room, an adult man (likely a teacher, with an ID lanyard, wearing a light grey long-sleeve shirt) stands before a wall of whiteboards plus a pulled-down blank projection screen, positioned to address the class.
2.  Seated at student desks, all facing front in standard green plastic school chairs, are a small number of students: visible are a blonde girl with a ponytail in a dark top, and

That looks right. Let me extract this into a function, and add a step parameter so I can subsample frames when windows get large.

Exported source 
def _build_window(
    runframes, # Grouped runframes from `_get_runframes`
    step:int=1, # Subsample every Nth frame
)->str: # Formatted window text for LLM input
    "Build window text from grouped runframes."
    window = ''
    for i,rf in runframes[::step]:
        prefix = f'TIMESTAMP {i}s\n'
        window += prefix+len(prefix.strip())*'='+'\n'
        for r in rf:
            window += '\n--\n'+f'{r.type.upper()}\n{r.description}'+'\n\n'
    return window

With the window built, I want to see how many tokens it is before feeding it to the LLM. That tells me how much I’m asking the model to compress. Let me try the encoder on a sample window.

enc = encoding_for_model('gpt-4o')
rfs = _get_runframes(db, run_id, 0, 10)
win = _build_window(rfs)
len(enc.encode(win))
451

I’ll add visual framing like the run header — a box at the start showing the window size and token count, and a box at the end showing the summary size, compression ratio, and cost.

Exported source 
def _summary_header(
    run_id:int, # Run ID (or label for multi-run)
    start:int, # Start frame second
    stop:int, # Stop frame second
    step:int, # Frame step interval
    model:str, # Model name
    window:str, # Full window text
    win_tokens:int, # Token count of window
    cache:bool, # Whether cache is enabled
    t0:datetime, # Start timestamp
):
    "Print summary run header box."
    print(f'╭─ Summary Run #{run_id} ═══════════════════════╮\n│ Frames   {start}–{stop} (step {step})\n│ Model    {model}\n│ Window   {len(window)} chars / {win_tokens} tokens\n│ Cache    {cache}\n│ Start    {t0.strftime("%H:%M:%S")}\n╰──────────────────────────────────────────────╯')
Exported source 
def _summary_footer(
    summary:str, # Summary text from LLM
    win_tokens:int, # Token count of original window
    t0:datetime, # Start timestamp
    t1:datetime, # Finish timestamp
    cost:float, # Dollar cost of this summary call
):
    "Print summary completion box."
    enc = encoding_for_model('gpt-4o')
    sum_tokens = len(enc.encode(summary))
    reduction = (1 - sum_tokens/win_tokens)*100 if win_tokens else 0
    elapsed = t1 - t0
    print(f'╭─ Summary Complete ═══════════════════════════╮\n│ Finish   {t1.strftime("%H:%M:%S")}\n│ Elapsed  {str(elapsed).split(".")[0]}\n│ Summary  {len(summary)} chars / {sum_tokens} tokens\n│ Reduced  {reduction:.1f}%\n│ Cost     ${cost:.4f} (HKD {cost*7.84:.2f})\n╰──────────────────────────────────────────────╯')

Let me try running a single window through the LLM to see what kind of summary comes back.

sys_prompt = "You are a careful observer. Summarize the following frame descriptions into a coherent narrative passage."
r = await s([user(win)], system=sys_prompt)
print(r.message.text[:500])

Across two consecutive moments in a small elementary school classroom, the scene shifts from the teacher preparing to address his group to the start of a hands-on tech demonstration. At the 3-second mark, a male teacher in a light grey long-sleeve shirt and ID lanyard stands at the front of the sunlit room, positioned before a wall of whiteboards and a blank pulled-down projection screen to speak to his handful of students. All the children sit in standard green plastic school chairs at forward-

With those in place, I can wrap the whole call into a single summarize_window function that builds the window, sends it to the LLM with a system prompt, and prints the boxes.

summarize_window

async def summarize_window(
    db:Database, # Database object
    run_ids:int | list[int], # Single run ID or list of run IDs
    start:int, # Start frame second
    stop:int, # Stop frame second
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    step:int=1, # Subsample every Nth frame
    cache:bool=False, # Whether to use HTTP cache
    context:str='', # Prior summary text to prepend as context
)->str: # Summary text for this window

Summarize a window of frames from runframes. Returns summary text.

Exported source 
async def summarize_window(
    db:Database, # Database object
    run_ids:int|list[int], # Single run ID or list of run IDs
    start:int, # Start frame second
    stop:int, # Stop frame second
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    step:int=1, # Subsample every Nth frame
    cache:bool=False, # Whether to use HTTP cache
    context:str='', # Prior summary text to prepend as context
)->str: # Summary text for this window
    "Summarize a window of frames from runframes. Returns summary text."
    if not cache: disable_cachy()
    rfs = _get_runframes(db, run_ids, start, stop)
    window = _build_window(rfs, step)
    if context: window = f'PRIOR CONTEXT\n{"="*13}\n{context}\n\n{window}'
    win_tokens = len(encoding_for_model('gpt-4o').encode(window))
    t0 = datetime.now(tz)
    label = run_ids if isinstance(run_ids, int) else f'{run_ids[0]}+{len(run_ids)-1}more'
    _summary_header(label, start, stop, step, session.keywords['model'], window, win_tokens, cache, t0)
    r = await session([user(window)], system=sys_prompt)
    t1 = datetime.now(tz)
    if not cache: enable_cachy()
    summary = r.message.text
    _summary_footer(summary, win_tokens, t0, t1, r.usage.raw.get('cost', 0))
    return summary

Let me try this manually with two windows to see if the context chaining produces a flowing narrative.

fnums = rfs.itemgot(0)
chunks = list(chunked(fnums, 300)); chunks[:3]
[[3, 4]]
s1 = await summarize_window(db, run_id, chunks[0][0], chunks[0][-1], s, sys_prompt)
if len(chunks) > 1: s2 = await summarize_window(db, run_id, chunks[1][0], chunks[1][-1], s, sys_prompt, context=s1)
╭─ Summary Run #1 ═══════════════════════╮
│ Frames   3–4 (step 1)
│ Model    bytedance-seed/seed-2.0-lite
│ Window   2072 chars / 451 tokens
│ Cache    False
│ Start    15:38:08
╰──────────────────────────────────────────────╯

╭─ Summary Complete ═══════════════════════════╮
│ Finish   15:38:33
│ Elapsed  0:00:25
│ Summary  1578 chars / 323 tokens
│ Reduced  28.4%
│ Cost     $0.0039 (HKD 0.03)
╰──────────────────────────────────────────────╯

A single window gives me one passage. But a full video has many windows. I want to feed each window to the LLM with the previous window’s summary as context, so the narrative flows continuously. That means chunking frames into window_sec-sized groups (default 300 seconds / 5 minutes), summarizing each in turn, and concatenating the results. The final summary gets written back to the database, onto the run record if I started from a run ID, or onto the video record if I started from a video ID (merging all its runs).

summarize_run

async def summarize_run(
    db:Database, # Database object
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    perspective:str, # Key for storing this summary in description JSON
    run_id:int | None=None, # Single run ID to summarize
    video_id:int | None=None, # Summarize all runs for this video
    window_sec:int=300, # Rolling window size in seconds
    step:int=1, # Subsample every Nth frame within windows
    cache:bool=False, # Whether to use HTTP cache
)->str: # Full concatenated summary across all windows

Summarize a single run or all frames for a video in rolling windows.

Exported source 
from IPython.display import clear_output
Exported source 
async def summarize_run(
    db:Database, # Database object
    session:Callable, # Stream session partial
    sys_prompt:str, # System prompt for the summarizing LLM
    perspective:str, # Key for storing this summary in description JSON
    run_id:int|None=None, # Single run ID to summarize
    video_id:int|None=None, # Summarize all runs for this video
    window_sec:int=300, # Rolling window size in seconds
    step:int=1, # Subsample every Nth frame within windows
    cache:bool=False, # Whether to use HTTP cache
)->str: # Full concatenated summary across all windows
    "Summarize a single run or all frames for a video in rolling windows."
    if run_id is not None: rids = [run_id]
    elif video_id is not None: rids = L(db.t.run('video_id=?', (video_id,))).map(lambda r: r.id)
    else: raise ValueError('Either run_id or video_id required')
    rfs = _get_runframes(db, rids)
    if not rfs: return ''
    fnums = rfs.itemgot(0)
    full_summary = ''
    chunks = list(chunked(fnums, window_sec))
    for chunk in (mb:=master_bar(chunks)):
        mb.main_bar.comment = f'window {chunk[0]}–{chunk[-1]}s'
        summary = await summarize_window(db, rids, chunk[0], chunk[-1], session, sys_prompt, step=step, cache=cache, context=full_summary)
        heading = f'[{chunk[0]}–{chunk[-1]}s]'
        summary = summary.strip()
        full_summary = full_summary + f'\n\n{heading} {summary}' if full_summary else f'{heading} {summary}' if summary else ''
        clear_output(wait=True)
        print(full_summary)
    target = db.t.run[run_id] if run_id is not None else db.t.video[video_id]
    desc = loads(target.description) if target.description else {}
    desc[perspective or 'default'] = full_summary
    if run_id is not None: db.t.run.update(id=run_id, description=dumps(desc))
    else: db.t.video.update(id=video_id, description=dumps(desc))
    return full_summary

vlm_monitor supports all providers fastllm supports. vlm_monitor uses an older version of fastllm and thus requires the vendor_name parameter amongst other differences.