lasagnastack.stages.reel_pipeline.analyse

lasagnastack.stages.reel_pipeline.analyse

AnalyseStage

AnalyseStage(
    client: LLMClient | None = None, max_workers: int = 4
)

Bases: Stage

Initialise AnalyseStage.

Parameters:

Name	Type	Description	Default
client	LLMClient | None	LLM client to use. Defaults to GeminiClient.	None
max_workers	int	Maximum number of concurrent LLM calls.	4

Source code in src/lasagnastack/stages/reel_pipeline/analyse.py

def __init__(self, client: LLMClient | None = None, max_workers: int = 4) -> None:
    """Initialise AnalyseStage.

    Args:
        client: LLM client to use. Defaults to GeminiClient.
        max_workers: Maximum number of concurrent LLM calls.
    """
    self._client = client
    self._max_workers = max_workers

run

run(
    clips: list[NormalisedClip],
    output_dir: Path,
    client: LLMClient | None = None,
    max_workers: int = 4,
) -> list[ClipInventory]

Analyse each normalised clip with Gemini; cache results by content hash.

Parameters:

Name	Type	Description	Default
clips	list[NormalisedClip]	Normalised clips from Stage 1.	required
output_dir	Path	Pipeline root; cache lives at output_dir/.cache, inventories written to output_dir/inventories/.	required
client	LLMClient | None	LLM client to use. Defaults to GeminiClient.	None
max_workers	int	Maximum number of concurrent LLM calls. Uses asyncio.gather with a semaphore; all calls run as threads via asyncio.to_thread since the LLM client is synchronous.	4

Returns:

Type	Description
list[ClipInventory]	One ClipInventory per clip, in the same order as clips.

Source code in src/lasagnastack/stages/reel_pipeline/analyse.py

def run(
    clips: list[NormalisedClip],
    output_dir: Path,
    client: LLMClient | None = None,
    max_workers: int = 4,
) -> list[ClipInventory]:
    """Analyse each normalised clip with Gemini; cache results by content hash.

    Args:
        clips: Normalised clips from Stage 1.
        output_dir: Pipeline root; cache lives at output_dir/.cache,
            inventories written to output_dir/inventories/.
        client: LLM client to use. Defaults to GeminiClient.
        max_workers: Maximum number of concurrent LLM calls. Uses
            ``asyncio.gather`` with a semaphore; all calls run as threads
            via ``asyncio.to_thread`` since the LLM client is synchronous.

    Returns:
        One ClipInventory per clip, in the same order as clips.
    """
    if client is None:
        client = GeminiClient()

    cache = DiskCache(io.cache_dir(output_dir))
    inventories = asyncio.run(_run_async(clips, cache, client, max_workers))

    for clip, inventory in zip(clips, inventories):
        io.write_json(inventory, io.inventory_path(output_dir, clip.source_path.name))

    return inventories