lasagnastack.llm.gemini

lasagnastack.llm.gemini

GeminiClient

GeminiClient(
    api_key: str | None = None,
    model: str | None = None,
    reasoning_max_tokens: int = 4000,
    reasoning_effort: str | None = None,
    total_max_tokens: int | None = None,
)

Bases: LLMClient

LLM client backed by the Gemini Developer API.

Wraps google-genai with structured JSON output, tenacity retries, a JSON-repair loop, and MLflow span tracing for every API call.

Initialise the Gemini client.

Parameters:

Name	Type	Description	Default
api_key	str | None	Gemini Developer API key. Falls back to the LSNSTK_LLM_GEMINI_API_KEY environment variable.	None
model	str | None	Model name to use for all calls (LiteLLM naming convention). Falls back to the LSNSTK_LLM_MODEL environment variable, then "gemini/gemini-2.5-flash".	None
reasoning_max_tokens	int	Maximum thinking tokens per call, forwarded as thinking_budget to the Gemini API. Set to 0 to disable thinking entirely. reasoning_effort and total_max_tokens are accepted for interface compatibility but are not used by the Gemini API.	4000
reasoning_effort	str | None	Accepted for interface compatibility; not used by the Gemini native API.	None
total_max_tokens	int | None	Accepted for interface compatibility; not used by the Gemini native API.	None

Source code in src/lasagnastack/llm/gemini.py

def __init__(
    self,
    api_key: str | None = None,
    model: str | None = None,
    reasoning_max_tokens: int = 4000,
    reasoning_effort: str | None = None,
    total_max_tokens: int | None = None,
) -> None:
    """Initialise the Gemini client.

    Args:
        api_key: Gemini Developer API key. Falls back to the
            ``LSNSTK_LLM_GEMINI_API_KEY`` environment variable.
        model: Model name to use for all calls (LiteLLM naming convention).
            Falls back to the ``LSNSTK_LLM_MODEL`` environment
            variable, then ``"gemini/gemini-2.5-flash"``.
        reasoning_max_tokens: Maximum thinking tokens per call, forwarded
            as ``thinking_budget`` to the Gemini API. Set to ``0`` to
            disable thinking entirely. ``reasoning_effort`` and
            ``total_max_tokens`` are accepted for interface compatibility
            but are not used by the Gemini API.
        reasoning_effort: Accepted for interface compatibility; not used
            by the Gemini native API.
        total_max_tokens: Accepted for interface compatibility; not used
            by the Gemini native API.
    """
    super().__init__(
        reasoning_max_tokens=reasoning_max_tokens,
        reasoning_effort=reasoning_effort,
        total_max_tokens=total_max_tokens,
    )
    raw_model = model or os.getenv("LSNSTK_LLM_MODEL", "gemini/gemini-2.5-flash")
    self._model = raw_model.removeprefix("gemini/")
    resolved_key = api_key or os.getenv("LSNSTK_LLM_GEMINI_API_KEY")
    if not resolved_key:
        raise ValueError(
            "Gemini API key not found. Set LSNSTK_LLM_GEMINI_API_KEY in .env or pass api_key=."
        )
    self._client = genai.Client(api_key=resolved_key)

    # Per-instance accumulators — updated after every successful API call.
    self._total_input_tokens: int = 0
    self._total_output_tokens: int = 0
    self._total_thinking_tokens: int = 0
    self._total_cost_usd: float = 0.0
    self._call_count: int = 0
    self._stats_lock = threading.Lock()

session_stats property

session_stats: dict[str, float | int]

Aggregated token and cost totals across all calls on this instance.

Returns:

Type	Description
dict[str, float | int]	A dict with keys total_input_tokens, total_output_tokens,
dict[str, float | int]	total_cost_usd, and llm_call_count.

generate

generate(
    prompt: str,
    response_schema: type[BaseModel],
    *,
    temperature: float = 0.4,
) -> BaseModel

Send a text-only prompt; return a validated response_schema instance.

Parameters:

Name	Type	Description	Default
prompt	str	The plain-text prompt to send.	required
response_schema	type[BaseModel]	Pydantic model class that defines the expected JSON structure of the response.	required
temperature	float	Sampling temperature passed to the model.	0.4

Returns:

Type	Description
BaseModel	A validated instance of response_schema.

Source code in src/lasagnastack/llm/gemini.py

def generate(
    self,
    prompt: str,
    response_schema: type[BaseModel],
    *,
    temperature: float = 0.4,
) -> BaseModel:
    """Send a text-only prompt; return a validated response_schema instance.

    Args:
        prompt: The plain-text prompt to send.
        response_schema: Pydantic model class that defines the expected
            JSON structure of the response.
        temperature: Sampling temperature passed to the model.

    Returns:
        A validated instance of ``response_schema``.
    """
    return self._generate_contents(prompt, response_schema, temperature)

generate_with_video

generate_with_video(
    video_path: Path,
    prompt: str,
    response_schema: type[BaseModel],
    *,
    temperature: float = 0.4,
) -> BaseModel

Upload video_path to the Files API, generate, return validated response.

Parameters:

Name	Type	Description	Default
video_path	Path	Local path to the MP4 clip to analyse.	required
prompt	str	Text prompt sent alongside the video.	required
response_schema	type[BaseModel]	Pydantic model class for the expected JSON response.	required
temperature	float	Sampling temperature passed to the model.	0.4

Returns:

Type	Description
BaseModel	A validated instance of response_schema.

Source code in src/lasagnastack/llm/gemini.py

def generate_with_video(
    self,
    video_path: Path,
    prompt: str,
    response_schema: type[BaseModel],
    *,
    temperature: float = 0.4,
) -> BaseModel:
    """Upload video_path to the Files API, generate, return validated response.

    Args:
        video_path: Local path to the MP4 clip to analyse.
        prompt: Text prompt sent alongside the video.
        response_schema: Pydantic model class for the expected JSON response.
        temperature: Sampling temperature passed to the model.

    Returns:
        A validated instance of ``response_schema``.
    """
    file_ref = self._upload_and_wait(video_path)
    try:
        contents = [
            types.Part.from_uri(file_uri=file_ref.uri, mime_type="video/mp4"),
            prompt,
        ]
        return self._generate_contents(contents, response_schema, temperature)
    finally:
        self._delete_file(file_ref.name)