Skip to content

gemini.calls

A module for calling Google's Gemini Chat API.

GeminiCall

Bases: BaseCall[GeminiCallResponse, GeminiCallResponseChunk, GeminiTool]

A class for prompting Google's Gemini Chat API.

This prompt supports the message types: USER, MODEL, TOOL

Example:

from google.generativeai import configure  # type: ignore
from mirascope.gemini import GeminiCall

configure(api_key="YOUR_API_KEY")


class BookRecommender(GeminiCall):
    prompt_template = """
    USER: You're the world's greatest librarian.
    MODEL: Ok, I understand I'm the world's greatest librarian. How can I help?
    USER: Please recommend some {genre} books.

    genre: str


response = BookRecommender(genre="fantasy").call()
print(response.content)
#> As the world's greatest librarian, I am delighted to recommend...
Source code in mirascope/gemini/calls.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class GeminiCall(BaseCall[GeminiCallResponse, GeminiCallResponseChunk, GeminiTool]):
    '''A class for prompting Google's Gemini Chat API.

    This prompt supports the message types: USER, MODEL, TOOL

    Example:

    ```python
    from google.generativeai import configure  # type: ignore
    from mirascope.gemini import GeminiCall

    configure(api_key="YOUR_API_KEY")


    class BookRecommender(GeminiCall):
        prompt_template = """
        USER: You're the world's greatest librarian.
        MODEL: Ok, I understand I'm the world's greatest librarian. How can I help?
        USER: Please recommend some {genre} books.

        genre: str


    response = BookRecommender(genre="fantasy").call()
    print(response.content)
    #> As the world's greatest librarian, I am delighted to recommend...
    ```
    '''

    call_params: ClassVar[GeminiCallParams] = GeminiCallParams()

    def messages(self) -> ContentsType:
        """Returns the `ContentsType` messages for Gemini `generate_content`.

        Raises:
            ValueError: if the docstring contains an unknown role.
        """
        return [
            {"role": message["role"], "parts": [message["content"]]}
            for message in self._parse_messages(
                [MessageRole.MODEL, MessageRole.USER, MessageRole.TOOL]
            )
        ]

    @retry
    def call(
        self, retries: Union[int, Retrying] = 1, **kwargs: Any
    ) -> GeminiCallResponse:
        """Makes an call to the model using this `GeminiCall` instance.

        Args:
            **kwargs: Additional keyword arguments that will be used for generating the
                response. These will override any existing argument settings in call
                params.

        Returns:
            A `GeminiCallResponse` instance.
        """
        kwargs, tool_types = self._setup(kwargs, GeminiTool)
        model_name = kwargs.pop("model")
        gemini_pro_model = GenerativeModel(model_name=model_name)
        generate_content = gemini_pro_model.generate_content
        if self.call_params.weave is not None:
            generate_content = self.call_params.weave(
                generate_content
            )  # pragma: no cover
        if self.call_params.logfire:
            generate_content = self.call_params.logfire(
                generate_content,
                "gemini",
                response_type=GeminiCallResponse,
                tool_types=tool_types,
            )  # pragma: no cover
            kwargs["model"] = model_name  # pragma: no cover
        start_time = datetime.datetime.now().timestamp() * 1000
        response = generate_content(
            self.messages(),
            stream=False,
            tools=kwargs.pop("tools") if "tools" in kwargs else None,
            **kwargs,
        )
        return GeminiCallResponse(
            response=response,
            tool_types=tool_types,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            cost=None,
        )

    @retry
    async def call_async(
        self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
    ) -> GeminiCallResponse:
        """Makes an asynchronous call to the model using this `GeminiCall` instance.

        Args:
            **kwargs: Additional keyword arguments that will be used for generating the
                response. These will override any existing argument settings in call
                params.

        Returns:
            A `GeminiCallResponse` instance.
        """
        kwargs, tool_types = self._setup(kwargs, GeminiTool)
        model_name = kwargs.pop("model")
        gemini_pro_model = GenerativeModel(model_name=model_name)
        generate_content_async = gemini_pro_model.generate_content_async
        if self.call_params.weave is not None:
            generate_content_async = self.call_params.weave(
                generate_content_async
            )  # pragma: no cover
        if self.call_params.logfire:
            generate_content_async = self.call_params.logfire(
                generate_content_async,
                "gemini",
                response_type=GeminiCallResponse,
                tool_types=tool_types,
            )  # pragma: no cover
            kwargs["model"] = model_name  # pragma: no cover
        start_time = datetime.datetime.now().timestamp() * 1000
        response = await generate_content_async(
            self.messages(),
            stream=False,
            tools=kwargs.pop("tools") if "tools" in kwargs else None,
            **kwargs,
        )
        return GeminiCallResponse(
            response=response,
            tool_types=tool_types,
            start_time=start_time,
            end_time=datetime.datetime.now().timestamp() * 1000,
            cost=None,
        )

    @retry
    def stream(
        self, retries: Union[int, Retrying] = 1, **kwargs: Any
    ) -> Generator[GeminiCallResponseChunk, None, None]:
        """Streams the response for a call using this `GeminiCall`.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Yields:
            A `GeminiCallResponseChunk` for each chunk of the response.
        """
        kwargs, tool_types = self._setup(kwargs, GeminiTool)
        model_name = kwargs.pop("model")
        gemini_pro_model = GenerativeModel(model_name=model_name)
        generate_content = gemini_pro_model.generate_content
        if self.call_params.logfire:
            generate_content = self.call_params.logfire(
                generate_content, "gemini", response_chunk_type=GeminiCallResponseChunk
            )  # pragma: no cover
            kwargs["model"] = model_name  # pragma: no cover
        stream = generate_content(
            self.messages(),
            stream=True,
            tools=kwargs.pop("tools") if "tools" in kwargs else None,
            **kwargs,
        )
        for chunk in stream:
            yield GeminiCallResponseChunk(chunk=chunk, tool_types=tool_types)

    @retry
    async def stream_async(
        self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
    ) -> AsyncGenerator[GeminiCallResponseChunk, None]:
        """Streams the response asynchronously for a call using this `GeminiCall`.

        Args:
            **kwargs: Additional keyword arguments parameters to pass to the call. These
                will override any existing arguments in `call_params`.

        Yields:
            A `GeminiCallResponseChunk` for each chunk of the response.
        """
        kwargs, tool_types = self._setup(kwargs, GeminiTool)
        model_name = kwargs.pop("model")
        gemini_pro_model = GenerativeModel(model_name=model_name)
        generate_content_async = gemini_pro_model.generate_content_async
        if self.call_params.logfire:
            generate_content_async = self.call_params.logfire(
                generate_content_async,
                "gemini",
                response_chunk_type=GeminiCallResponseChunk,
            )  # pragma: no cover
            kwargs["model"] = model_name  # pragma: no cover
        stream = await generate_content_async(
            self.messages(),
            stream=True,
            tools=kwargs.pop("tools") if "tools" in kwargs else None,
            **kwargs,
        )
        async for chunk in stream:
            yield GeminiCallResponseChunk(chunk=chunk, tool_types=tool_types)

call(retries=1, **kwargs)

Makes an call to the model using this GeminiCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments that will be used for generating the response. These will override any existing argument settings in call params.

 {}

Returns:

Type Description
GeminiCallResponse

A GeminiCallResponse instance.
Source code in mirascope/gemini/calls.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@retry
def call(
    self, retries: Union[int, Retrying] = 1, **kwargs: Any
) -> GeminiCallResponse:
    """Makes an call to the model using this `GeminiCall` instance.

    Args:
        **kwargs: Additional keyword arguments that will be used for generating the
            response. These will override any existing argument settings in call
            params.

    Returns:
        A `GeminiCallResponse` instance.
    """
    kwargs, tool_types = self._setup(kwargs, GeminiTool)
    model_name = kwargs.pop("model")
    gemini_pro_model = GenerativeModel(model_name=model_name)
    generate_content = gemini_pro_model.generate_content
    if self.call_params.weave is not None:
        generate_content = self.call_params.weave(
            generate_content
        )  # pragma: no cover
    if self.call_params.logfire:
        generate_content = self.call_params.logfire(
            generate_content,
            "gemini",
            response_type=GeminiCallResponse,
            tool_types=tool_types,
        )  # pragma: no cover
        kwargs["model"] = model_name  # pragma: no cover
    start_time = datetime.datetime.now().timestamp() * 1000
    response = generate_content(
        self.messages(),
        stream=False,
        tools=kwargs.pop("tools") if "tools" in kwargs else None,
        **kwargs,
    )
    return GeminiCallResponse(
        response=response,
        tool_types=tool_types,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        cost=None,
    )

call_async(retries=1, **kwargs) async

Makes an asynchronous call to the model using this GeminiCall instance.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments that will be used for generating the response. These will override any existing argument settings in call params.

 {}

Returns:

Type Description
GeminiCallResponse

A GeminiCallResponse instance.
Source code in mirascope/gemini/calls.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@retry
async def call_async(
    self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
) -> GeminiCallResponse:
    """Makes an asynchronous call to the model using this `GeminiCall` instance.

    Args:
        **kwargs: Additional keyword arguments that will be used for generating the
            response. These will override any existing argument settings in call
            params.

    Returns:
        A `GeminiCallResponse` instance.
    """
    kwargs, tool_types = self._setup(kwargs, GeminiTool)
    model_name = kwargs.pop("model")
    gemini_pro_model = GenerativeModel(model_name=model_name)
    generate_content_async = gemini_pro_model.generate_content_async
    if self.call_params.weave is not None:
        generate_content_async = self.call_params.weave(
            generate_content_async
        )  # pragma: no cover
    if self.call_params.logfire:
        generate_content_async = self.call_params.logfire(
            generate_content_async,
            "gemini",
            response_type=GeminiCallResponse,
            tool_types=tool_types,
        )  # pragma: no cover
        kwargs["model"] = model_name  # pragma: no cover
    start_time = datetime.datetime.now().timestamp() * 1000
    response = await generate_content_async(
        self.messages(),
        stream=False,
        tools=kwargs.pop("tools") if "tools" in kwargs else None,
        **kwargs,
    )
    return GeminiCallResponse(
        response=response,
        tool_types=tool_types,
        start_time=start_time,
        end_time=datetime.datetime.now().timestamp() * 1000,
        cost=None,
    )

messages()

Returns the ContentsType messages for Gemini generate_content.

Raises:

Type Description
ValueError

if the docstring contains an unknown role.
Source code in mirascope/gemini/calls.py 
59
60
61
62
63
64
65
66
67
68
69
70
def messages(self) -> ContentsType:
    """Returns the `ContentsType` messages for Gemini `generate_content`.

    Raises:
        ValueError: if the docstring contains an unknown role.
    """
    return [
        {"role": message["role"], "parts": [message["content"]]}
        for message in self._parse_messages(
            [MessageRole.MODEL, MessageRole.USER, MessageRole.TOOL]
        )
    ]

stream(retries=1, **kwargs)

Streams the response for a call using this GeminiCall.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Yields:

Type Description
GeminiCallResponseChunk

A GeminiCallResponseChunk for each chunk of the response.
Source code in mirascope/gemini/calls.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@retry
def stream(
    self, retries: Union[int, Retrying] = 1, **kwargs: Any
) -> Generator[GeminiCallResponseChunk, None, None]:
    """Streams the response for a call using this `GeminiCall`.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Yields:
        A `GeminiCallResponseChunk` for each chunk of the response.
    """
    kwargs, tool_types = self._setup(kwargs, GeminiTool)
    model_name = kwargs.pop("model")
    gemini_pro_model = GenerativeModel(model_name=model_name)
    generate_content = gemini_pro_model.generate_content
    if self.call_params.logfire:
        generate_content = self.call_params.logfire(
            generate_content, "gemini", response_chunk_type=GeminiCallResponseChunk
        )  # pragma: no cover
        kwargs["model"] = model_name  # pragma: no cover
    stream = generate_content(
        self.messages(),
        stream=True,
        tools=kwargs.pop("tools") if "tools" in kwargs else None,
        **kwargs,
    )
    for chunk in stream:
        yield GeminiCallResponseChunk(chunk=chunk, tool_types=tool_types)

stream_async(retries=1, **kwargs) async

Streams the response asynchronously for a call using this GeminiCall.

Parameters:

Name Type Description Default
**kwargs Any

Additional keyword arguments parameters to pass to the call. These will override any existing arguments in call_params.

 {}

Yields:

Type Description
AsyncGenerator[GeminiCallResponseChunk, None]

A GeminiCallResponseChunk for each chunk of the response.
Source code in mirascope/gemini/calls.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@retry
async def stream_async(
    self, retries: Union[int, AsyncRetrying] = 1, **kwargs: Any
) -> AsyncGenerator[GeminiCallResponseChunk, None]:
    """Streams the response asynchronously for a call using this `GeminiCall`.

    Args:
        **kwargs: Additional keyword arguments parameters to pass to the call. These
            will override any existing arguments in `call_params`.

    Yields:
        A `GeminiCallResponseChunk` for each chunk of the response.
    """
    kwargs, tool_types = self._setup(kwargs, GeminiTool)
    model_name = kwargs.pop("model")
    gemini_pro_model = GenerativeModel(model_name=model_name)
    generate_content_async = gemini_pro_model.generate_content_async
    if self.call_params.logfire:
        generate_content_async = self.call_params.logfire(
            generate_content_async,
            "gemini",
            response_chunk_type=GeminiCallResponseChunk,
        )  # pragma: no cover
        kwargs["model"] = model_name  # pragma: no cover
    stream = await generate_content_async(
        self.messages(),
        stream=True,
        tools=kwargs.pop("tools") if "tools" in kwargs else None,
        **kwargs,
    )
    async for chunk in stream:
        yield GeminiCallResponseChunk(chunk=chunk, tool_types=tool_types)