Own the Retrieval and the Agent

1Use my brief and selected files for chapter 4 only. Propose one small2retrieval/agent increment; wait for approval to apply it. Cite current official3API contracts, not invented constructors or fields. Preserve dimension/modality4checks, configuration-keyed resume, strict SDK-shaped mocks, and source/timecode5citations; similarity alone is not grounded evidence.6Run local tests; inspect actual responses and footage only after an approved live run.7Explain changes and debug with redacted errors. Do not read secrets, commit/push,8provision, or make paid calls without my explicit approval. Do not jump chapters.9I review the Git diff and commit.

1Review chapter 4's diff/tests against current official API contracts.2Check strict constructor/method mocks, nested parsers, ready/processing/failed3states, interruption, dimension/config mismatches, and grounded source citations.4Do not invent model results; separate offline tests from actual response evidence.5Flag unsupported claims and use redacted errors. Do not read secrets or jump chapters.6Do not commit/push, provision, or make paid calls without my explicit approval.7Show the Git changes for my review; I make the commit.

1"""Chapter 4: Marengo 3.5 direct embeddings — async video, sync text.2 3Documentation-checked against the create-embeddings quickstart and the v24SDK references; live validation pending. Contract notes:5- Async video: client.embed.v_2.tasks.create(model_name="marengo3.5",6  input_type="video", video=AsyncVideoInputRequest(media_source=MediaSource(asset_id=...),7  embedding_option=[...]), embedding_dimension=DIMENSION)8- Completed data is parsed from done.data — NOT done.video_embedding.segments.9- Sync text: input_type="multi_input", multi_input=MultiInputRequest(input_text=text),10  vector at result.data[0].embedding.11- State is keyed by checksum + model + dimension + options so a 512-dim task is12  never reused for a 128-dim exercise.13"""14import math15import json16import os17import time18from pathlib import Path19 20from dotenv import load_dotenv21from twelvelabs import TwelveLabs22from twelvelabs.types import AsyncVideoInputRequest, MediaSource, MultiInputRequest23from rank import normalize24 25load_dotenv()26client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])27 28ROOT = Path(__file__).parent29VECTORS = ROOT / "vectors"30VECTORS.mkdir(exist_ok=True)31STATE_PATH = ROOT / "state.json"32state = json.loads(STATE_PATH.read_text())33 34MODEL = "marengo3.5"35DIMENSION = 51236EMBEDDING_OPTIONS = ["visual", "audio"]37 38 39def save_state() -> None:40    STATE_PATH.write_text(json.dumps(state, indent=2))41 42 43def embed_key(clip: dict) -> str:44    """Reuse is only safe when the file, model, dimension and options all match."""45    return f"{clip['id']}:{clip['sha256'][:12]}:{MODEL}:{DIMENSION}:{'+'.join(EMBEDDING_OPTIONS)}"46 47 48def start_video_task(clip: dict) -> str:49    state.setdefault("embed_tasks", {})50    key = embed_key(clip)51    if key in state["embed_tasks"]:52        print(f"resuming {clip['id']}: task {state['embed_tasks'][key]}")53        return state["embed_tasks"][key]54    task = client.embed.v_2.tasks.create(55        model_name=MODEL,56        input_type="video",57        video=AsyncVideoInputRequest(58            media_source=MediaSource(asset_id=state["assets"][clip["id"]]["asset_id"]),59            embedding_option=EMBEDDING_OPTIONS,60        ),61        embedding_dimension=DIMENSION,62    )63    state["embed_tasks"][key] = task.id  # persist BEFORE polling64    save_state()65    return task.id66 67 68def wait_task(task_id: str, timeout_sec: int = 1800):69    deadline = time.time() + timeout_sec70    delay = 571    while time.time() < deadline:72        done = client.embed.v_2.tasks.retrieve(task_id)73        status = getattr(done, "status", "unknown")74        if status == "ready":75            return done76        if status in ("failed", "canceled"):77            raise RuntimeError(f"embedding task {task_id} ended with status={status}")78        time.sleep(delay)79        delay = min(delay * 2, 30)80    raise TimeoutError(f"local timeout — task {task_id} may still run; re-run to resume")81 82 83def parse_video_segments(done) -> list[dict]:84    """Parse done.data into normalized segment vectors with dimension checks."""85    segments = []86    for i, seg in enumerate(done.data or []):87        getter = seg.get if isinstance(seg, dict) else lambda k: getattr(seg, k, None)88        emb = getter("embedding")89        validate_vector(emb)90        start, end = getter("start_sec"), getter("end_sec")91        if not all(isinstance(x, (int, float)) and not isinstance(x, bool) and math.isfinite(x) for x in (start, end)) or start < 0 or end <= start:92            raise RuntimeError(f"segment {i}: invalid start_sec/end_sec")93        segments.append({94            "start_sec": float(start), "end_sec": float(end),95            "embedding_option": getter("embedding_option"),96            "embedding_scope": getter("embedding_scope"),97            "embedding": emb,98        })99    return segments100 101 102def validate_vector(vector) -> list[float]:103    if not isinstance(vector, list) or len(vector) != DIMENSION:104        raise RuntimeError(f"expected {DIMENSION}-dim vector")105    normalize(vector)  # reject non-finite and zero-norm vectors, preserve original values106    return vector107 108 109def embed_text(text: str) -> list[float]:110    """Synchronous text embedding via multi_input; vector at data[0].embedding."""111    result = client.embed.v_2.create(112        model_name=MODEL,113        input_type="multi_input",114        multi_input=MultiInputRequest(input_text=text),115        embedding_dimension=DIMENSION,116    )117    vector = result.data[0].embedding118    return validate_vector(vector)119 120 121if __name__ == "__main__":122    manifest = json.loads((ROOT / "media-manifest.json").read_text())123    for clip in manifest["clips"]:124        task_id = start_video_task(clip)125        done = wait_task(task_id)126        segments = parse_video_segments(done)127        (VECTORS / f"{clip['id']}.json").write_text(json.dumps({128            "clip_id": clip["id"], "model": MODEL, "dimension": DIMENSION,129            "options": EMBEDDING_OPTIONS, "segments": segments,130        }, indent=2))131        print(f"{clip['id']}: {len(segments)} segment vectors")132    query = embed_text("a person opens a blue notebook")133    (VECTORS / "query-notebook.json").write_text(json.dumps({134        "text": "a person opens a blue notebook", "model": MODEL,135        "dimension": DIMENSION, "embedding": query,136    }))137    print("query vector saved")

1"""Chapter 4: local cosine retrieval over saved segment vectors. No API calls."""2import json3import math4from pathlib import Path5 6ROOT = Path(__file__).parent7 8 9def normalize(v: list[float]) -> list[float]:10    if not all(isinstance(x, (int, float)) and not isinstance(x, bool) and math.isfinite(x) for x in v):11        raise ValueError("non-finite or nonnumeric value in vector")12    norm = math.sqrt(sum(x * x for x in v))13    if not math.isfinite(norm):14        raise ValueError("non-finite vector norm")15    if norm == 0:16        raise ValueError("zero-norm vector cannot be ranked")17    return [x / norm for x in v]18 19 20def rank(segments: list[dict], query: list[float], top_k: int = 5) -> list[dict]:21    q = normalize(query)22    scored = []23    for seg in segments:24        v = normalize(seg["embedding"])25        if len(v) != len(q):26            raise ValueError(f"dimension mismatch: {len(v)} vs query {len(q)}")27        scored.append({**seg, "score": sum(a * b for a, b in zip(v, q))})28    scored.sort(key=lambda s: s["score"], reverse=True)29    return scored[:top_k]30 31 32def load_segments(clip_ids: list[str]) -> list[dict]:33    segments = []34    for clip_id in clip_ids:35        data = json.loads((ROOT / "vectors" / f"{clip_id}.json").read_text())36        for seg in data["segments"]:37            segments.append({"clip_id": clip_id, **seg})38    return segments39 40 41if __name__ == "__main__":42    state = json.loads((ROOT / "state.json").read_text())43    query = json.loads((ROOT / "vectors" / "query-notebook.json").read_text())44    hits = rank(load_segments(list(state["assets"])), query["embedding"])45    for h in hits:46        print(f"{h['score']:.3f}  {h['clip_id']}  {h['start_sec']:.1f}-{h['end_sec']:.1f}s")

1"""Chapter 4 exercise: composed image + text query.2 3Embed a reference frame (e.g. "the red mug") and a text constraint4("after the notebook opens") in one native multi-input request,5and rank locally. Implemented, not a prose placeholder.6"""7import json8import os9from pathlib import Path10 11from dotenv import load_dotenv12from twelvelabs import TwelveLabs13from twelvelabs.types import MultiInputRequest, MultiInputMediaSource14from embeddings import validate_vector15from still_image import extract_frame, ensure_frame_asset16from rank import load_segments, rank17 18load_dotenv()19ROOT = Path(__file__).parent20MODEL = "marengo3.5"21DIMENSION = 51222client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])23 24 25def embed_composed(asset_id: str, constraint: str) -> list[float]:26    """A single native image + text multi-input request, not invented image_file."""27    result = client.embed.v_2.create(28        model_name=MODEL,29        input_type="multi_input",30        multi_input=MultiInputRequest(31            input_text=constraint,32            media_sources=[MultiInputMediaSource(media_type="image", asset_id=asset_id)],33        ),34        embedding_dimension=DIMENSION,35    )36    return validate_vector(result.data[0].embedding)37 38 39if __name__ == "__main__":40    clip = json.loads((ROOT / "media-manifest.json").read_text())["clips"][0]41    frame = extract_frame(clip, clip["duration_sec"] / 2)42    asset_id = ensure_frame_asset(frame)  # same persisted, ready image used in chapter three43    query = embed_composed(asset_id, "after the notebook opens")44    state = json.loads((ROOT / "state.json").read_text())45    for h in rank(load_segments(list(state["assets"])), query):46        print(f"{h['score']:.3f}  {h['clip_id']}  {h['start_sec']:.1f}-{h['end_sec']:.1f}s")

1"""Jockey knowledge store via the API — per-item readiness."""2import json3import os4import time5from pathlib import Path6 7from dotenv import load_dotenv8from twelvelabs import TwelveLabs9 10load_dotenv()11client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])12 13ROOT = Path(__file__).parent14STATE_PATH = ROOT / "state.json"15state = json.loads(STATE_PATH.read_text())16 17 18def save_state() -> None:19    STATE_PATH.write_text(json.dumps(state, indent=2))20 21 22def main() -> None:23    if "knowledge_store_id" not in state:24        store = client.knowledge_stores.create(name="video-lab-store")25        state["knowledge_store_id"] = store.id26        save_state()27        print(f"created store {store.id}")28    store_id = state["knowledge_store_id"]29    state.setdefault("store_items", {})30    for clip_id, info in state["assets"].items():31        if clip_id in state["store_items"]:32            print(f"skip {clip_id}: already added")33            continue34        item = client.knowledge_store_items.create(35            store_id, asset_id=info["asset_id"]36        )37        state["store_items"][clip_id] = {"item_id": item.id}38        save_state()39        print(f"added {clip_id} -> {item.id}")40 41 42def wait_items_ready(timeout_sec: int = 900) -> None:43    """Poll each persisted item; the store has no aggregate status."""44    store_id = state["knowledge_store_id"]45    pending = dict(state["store_items"])46    deadline = time.time() + timeout_sec47    delay = 548    while pending and time.time() < deadline:49        for clip_id, item in list(pending.items()):50            current = client.knowledge_store_items.retrieve(51                store_id, item["item_id"]52            )53            status = getattr(current, "status", "unknown")54            if status == "ready":55                print(f"ready: {clip_id}")56                del pending[clip_id]57            elif status == "failed":58                raise RuntimeError(f"{clip_id}: item {item['item_id']} failed")59        if pending:60            time.sleep(delay)61            delay = min(delay * 2, 30)62    if pending:63        raise TimeoutError(64            f"items still processing: {sorted(pending)} — re-run to resume"65        )66 67 68if __name__ == "__main__":69    main()70    wait_items_ready()71    print("all store items ready")

1"""Jockey Responses with sessions and reviewed output parsing."""2import json3import os4from pathlib import Path5 6from dotenv import load_dotenv7from twelvelabs import TwelveLabs8 9load_dotenv()10client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])11 12ROOT = Path(__file__).parent13RESULTS = ROOT / "local-results"14RESULTS.mkdir(exist_ok=True)15state = json.loads((ROOT / "state.json").read_text())16STORE_ID = state["knowledge_store_id"]17 18 19def ask(prompt: str, session_id: str | None = None) -> dict:20    kwargs = {21        "input": [{"type": "message", "role": "user", "content": prompt}],22        "knowledge_store_id": STORE_ID,23    }24    if session_id:25        kwargs["session_id"] = session_id26    response = client.responses.create(**kwargs)27    status = getattr(response, "status", None)28    if status and status != "completed":29        raise RuntimeError(30            f"response not completed: status={status} "31            f"incomplete_details={getattr(response, 'incomplete_details', None)}"32        )33    texts, annotations = [], []34    for item in getattr(response, "output", []) or []:35        for content in getattr(item, "content", []) or []:36            text = getattr(content, "text", None)37            if text:38                texts.append(text)39            annotations.extend(getattr(content, "annotations", []) or [])40    return {41        "text": "\n".join(texts),42        "annotations": [getattr(a, "model_dump", lambda: a)() for a in annotations],43        "session_id": response.session_id,44    }45 46 47if __name__ == "__main__":48    first = ask(49        "Across all clips in this store, what happens to the red mug? "50        "Cite the clip and approximate time for each event."51    )52    print("ANSWER 1:", first["text"])53    second = ask(54        "Which of those moments would make the best 5-second highlight, "55        "and why? Propose exactly three intervals with start and end times.",56        session_id=first["session_id"],57    )58    print("ANSWER 2:", second["text"])59    out = RESULTS / "jockey-highlights.json"60    out.write_text(json.dumps({"first": first, "second": second},61                              indent=2, default=str))62    print(f"saved {out} — review each interval against your footage")

1git status && git diff --cached2git add embeddings.py rank.py composed_query.py jockey_store.py jockey_responses.py3git commit -m "Chapter 4: Marengo 3.5 embeddings, local retrieval, Jockey from code"4git push