Gemini
===============================================================================

Agent
-------------------------------------------------------------------------------

.. autoclass:: core_genai.agents.gemini.agent.GeminiAgent
   :members:
   :undoc-members:
   :show-inheritance:



Examples
-------------------------------------------------------------------------------

Inference
~~~~~~~~~

.. code-block:: python

    import asyncio

    from core_genai.agents.gemini.agent import GeminiAgent
    from core_genai.interfaces import IAgent
    from core_mixins.decorators.async_ import SyncWrapper
    from environs import Env

    env = Env()
    env.read_env(".env")

    agent: GeminiAgent = IAgent.create_agent("GeminiAgent", api_key=env("GEMINI_API_KEY"))

    MODEL = "gemini-2.5-flash"
    PROMPT = "Explain how AI works in a few words"


    async def run_async() -> None:
        output = await agent.analyze(model=MODEL, prompt=PROMPT)
        print("[async]", agent.get_text(output))


    def run_sync() -> None:
        with SyncWrapper(agent) as sync_agent:
            output = sync_agent.analyze(model=MODEL, prompt=PROMPT)
            print("[sync]", agent.get_text(output))


    if __name__ == "__main__":
        asyncio.run(run_async())
        run_sync()


Batch inference
~~~~~~~~~~~~~~~

Each request is an :class:`~google.genai.types.InlinedRequestDict` with the
model and prompt. ``schedule_job`` submits the batch; poll
``check_job_status`` until the state is terminal, then call
``extract_job_results``.

.. code-block:: python

    import asyncio

    from core_genai.agents.gemini.agent import GeminiAgent
    from core_genai.interfaces import IAgent
    from environs import Env
    from google.genai.types import InlinedRequestDict
    from google.genai.types import JobState

    env = Env()
    env.read_env(".env")

    MODEL = "gemini-2.5-flash-lite"
    POLL_INTERVAL = 30

    PROMPTS = [
        "Explain what machine learning is in one sentence.",
        "Explain what a neural network is in one sentence.",
        "Explain what reinforcement learning is in one sentence.",
    ]

    agent: GeminiAgent = IAgent.create_agent("GeminiAgent", api_key=env("GEMINI_API_KEY"))


    def build_requests() -> list[InlinedRequestDict]:
        return [
            InlinedRequestDict(model=MODEL, contents=prompt)
            for prompt in PROMPTS
        ]


    def print_results(result: dict) -> None:
        if result["error"]:
            print(f"Error: {result['error']}")
            return

        responses = result["results"] or []
        print(f"{len(responses)} response(s):\n")
        for i, resp in enumerate(responses, 1):
            if resp.error:
                text = f"[error {resp.error.code}: {resp.error.message}]"
            elif resp.response:
                text = agent.get_text(resp.response)
            else:
                text = "[no response]"
            print(f"[{i}] {PROMPTS[i - 1]!r}")
            print(f"    {text}\n")


    async def poll_until_done(job_id: str) -> JobState:
        while True:
            state = await agent.check_job_status(job_id)
            print(f"  status: {state.name}")

            if state in GeminiAgent.TERMINAL_STATES:
                return state

            await asyncio.sleep(POLL_INTERVAL)


    async def main() -> None:
        requests = build_requests()
        print(f"Scheduling batch job with {len(requests)} request(s)...")

        job = await agent.schedule_job(requests=requests, model=MODEL)
        job_id = job["job_id"]
        print(f"Job scheduled: {job_id}  (created: {job.get('created_at')})")

        print("Polling for completion...")
        final_state = await poll_until_done(job_id)
        print(f"Job finished with state: {final_state.name}")

        print_results(await agent.extract_job_results(job_id))


    async def extract_results(job_id: str) -> None:
        print_results(await agent.extract_job_results(job_id))


    async def check_batches(*job_ids: str) -> None:
        for job_id in job_ids:
            state = await agent.check_job_status(job_id)
            print(f"{job_id}: {state.name}")


    if __name__ == "__main__":
        asyncio.run(main())
        asyncio.run(check_batches("batches/..."))
        asyncio.run(extract_results("batches/..."))