Skip to content

base.ops_utils

BaseCall

Bases: BasePrompt, Generic[BaseCallResponseT, BaseCallResponseChunkT, BaseToolT, MessageParamT], ABC

The base class abstract interface for calling LLMs.

Source code in mirascope/base/calls.py 
 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
class BaseCall(
    BasePrompt,
    Generic[BaseCallResponseT, BaseCallResponseChunkT, BaseToolT, MessageParamT],
    ABC,
):
    """The base class abstract interface for calling LLMs."""

    api_key: ClassVar[Optional[str]] = None
    base_url: ClassVar[Optional[str]] = None
    call_params: ClassVar[BaseCallParams] = BaseCallParams[BaseToolT](
        model="gpt-3.5-turbo-0125"
    )
    configuration: ClassVar[BaseConfig] = BaseConfig(llm_ops=[], client_wrappers=[])
    _provider: ClassVar[str] = "base"

    @abstractmethod
    def call(
        self, retries: Union[int, Retrying] = 0, **kwargs: Any
    ) -> BaseCallResponseT:
        """A call to an LLM.

        An implementation of this function must return a response that extends
        `BaseCallResponse`. This ensures a consistent API and convenience across e.g.
        different model providers.
        """
        ...  # pragma: no cover

    @abstractmethod
    async def call_async(
        self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
    ) -> BaseCallResponseT:
        """An asynchronous call to an LLM.

        An implementation of this function must return a response that extends
        `BaseCallResponse`. This ensures a consistent API and convenience across e.g.
        different model providers.
        """
        ...  # pragma: no cover

    @abstractmethod
    def stream(
        self, retries: Union[int, Retrying] = 0, **kwargs: Any
    ) -> Generator[BaseCallResponseChunkT, None, None]:
        """A call to an LLM that streams the response in chunks.

        An implementation of this function must yield response chunks that extend
        `BaseCallResponseChunk`. This ensures a consistent API and convenience across
        e.g. different model providers.
        """
        ...  # pragma: no cover

    @abstractmethod
    async def stream_async(
        self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
    ) -> AsyncGenerator[BaseCallResponseChunkT, None]:
        """A asynchronous call to an LLM that streams the response in chunks.

        An implementation of this function must yield response chunks that extend
        `BaseCallResponseChunk`. This ensures a consistent API and convenience across
        e.g. different model providers."""
        yield ...  # type: ignore # pragma: no cover

    @classmethod
    def from_prompt(
        cls, prompt_type: type[BasePromptT], call_params: BaseCallParams
    ) -> type[BasePromptT]:
        """Returns a call_type generated dynamically from this base call.

        Args:
            prompt_type: The prompt class to use for the call. Properties and class
                variables of this class will be used to create the new call class. Must
                be a class that can be instantiated.
            call_params: The call params to use for the call.

        Returns:
            A new call class with new call_type.
        """

        fields: dict[str, Any] = {
            name: (field.annotation, field.default)
            for name, field in prompt_type.model_fields.items()
        }

        class_vars = {
            name: value
            for name, value in prompt_type.__dict__.items()
            if name not in prompt_type.model_fields
        }
        new_call = create_model(prompt_type.__name__, __base__=cls, **fields)

        for var_name, var_value in class_vars.items():
            setattr(new_call, var_name, var_value)
        setattr(new_call, "call_params", call_params)

        return cast(type[BasePromptT], new_call)

    ############################## PRIVATE METHODS ###################################

    def _setup(
        self,
        kwargs: dict[str, Any],
        base_tool_type: Optional[Type[BaseToolT]] = None,
    ) -> tuple[dict[str, Any], Optional[list[Type[BaseToolT]]]]:
        """Returns the call params kwargs and tool types.

        The tools in the call params first get converted into BaseToolT types. We then
        need both the converted tools for the response (so it can construct actual tool
        instances if present in the response) as well as the actual schemas injected
        through kwargs. This function handles that setup.
        """
        call_params = self.call_params.model_copy(update=kwargs)
        kwargs = call_params.kwargs(tool_type=base_tool_type)
        tool_types = None
        if "tools" in kwargs and base_tool_type is not None:
            tool_types = kwargs.pop("tools")
            kwargs["tools"] = [tool_type.tool_schema() for tool_type in tool_types]
        return kwargs, tool_types

    def _get_possible_user_message(
        self, messages: list[Any]
    ) -> Optional[MessageParamT]:
        """Returns the most recent message if it's a user message, otherwise `None`."""
        return messages[-1] if messages[-1]["role"] == "user" else None

call(retries=0, **kwargs) abstractmethod

A call to an LLM.

An implementation of this function must return a response that extends BaseCallResponse. This ensures a consistent API and convenience across e.g. different model providers.

Source code in mirascope/base/calls.py 
48
49
50
51
52
53
54
55
56
57
58
@abstractmethod
def call(
    self, retries: Union[int, Retrying] = 0, **kwargs: Any
) -> BaseCallResponseT:
    """A call to an LLM.

    An implementation of this function must return a response that extends
    `BaseCallResponse`. This ensures a consistent API and convenience across e.g.
    different model providers.
    """
    ...  # pragma: no cover

call_async(retries=0, **kwargs) abstractmethod async

An asynchronous call to an LLM.

An implementation of this function must return a response that extends BaseCallResponse. This ensures a consistent API and convenience across e.g. different model providers.

Source code in mirascope/base/calls.py 
60
61
62
63
64
65
66
67
68
69
70
@abstractmethod
async def call_async(
    self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
) -> BaseCallResponseT:
    """An asynchronous call to an LLM.

    An implementation of this function must return a response that extends
    `BaseCallResponse`. This ensures a consistent API and convenience across e.g.
    different model providers.
    """
    ...  # pragma: no cover

from_prompt(prompt_type, call_params) classmethod

Returns a call_type generated dynamically from this base call.

Parameters:

Name Type Description Default
prompt_type type[BasePromptT]

The prompt class to use for the call. Properties and class variables of this class will be used to create the new call class. Must be a class that can be instantiated.

 required
call_params BaseCallParams

The call params to use for the call.

 required

Returns:

Type Description
type[BasePromptT]

A new call class with new call_type.
Source code in mirascope/base/calls.py 
 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
@classmethod
def from_prompt(
    cls, prompt_type: type[BasePromptT], call_params: BaseCallParams
) -> type[BasePromptT]:
    """Returns a call_type generated dynamically from this base call.

    Args:
        prompt_type: The prompt class to use for the call. Properties and class
            variables of this class will be used to create the new call class. Must
            be a class that can be instantiated.
        call_params: The call params to use for the call.

    Returns:
        A new call class with new call_type.
    """

    fields: dict[str, Any] = {
        name: (field.annotation, field.default)
        for name, field in prompt_type.model_fields.items()
    }

    class_vars = {
        name: value
        for name, value in prompt_type.__dict__.items()
        if name not in prompt_type.model_fields
    }
    new_call = create_model(prompt_type.__name__, __base__=cls, **fields)

    for var_name, var_value in class_vars.items():
        setattr(new_call, var_name, var_value)
    setattr(new_call, "call_params", call_params)

    return cast(type[BasePromptT], new_call)

stream(retries=0, **kwargs) abstractmethod

A call to an LLM that streams the response in chunks.

An implementation of this function must yield response chunks that extend BaseCallResponseChunk. This ensures a consistent API and convenience across e.g. different model providers.

Source code in mirascope/base/calls.py 
72
73
74
75
76
77
78
79
80
81
82
@abstractmethod
def stream(
    self, retries: Union[int, Retrying] = 0, **kwargs: Any
) -> Generator[BaseCallResponseChunkT, None, None]:
    """A call to an LLM that streams the response in chunks.

    An implementation of this function must yield response chunks that extend
    `BaseCallResponseChunk`. This ensures a consistent API and convenience across
    e.g. different model providers.
    """
    ...  # pragma: no cover

stream_async(retries=0, **kwargs) abstractmethod async

A asynchronous call to an LLM that streams the response in chunks.

An implementation of this function must yield response chunks that extend BaseCallResponseChunk. This ensures a consistent API and convenience across e.g. different model providers.

Source code in mirascope/base/calls.py 
84
85
86
87
88
89
90
91
92
93
@abstractmethod
async def stream_async(
    self, retries: Union[int, AsyncRetrying] = 0, **kwargs: Any
) -> AsyncGenerator[BaseCallResponseChunkT, None]:
    """A asynchronous call to an LLM that streams the response in chunks.

    An implementation of this function must yield response chunks that extend
    `BaseCallResponseChunk`. This ensures a consistent API and convenience across
    e.g. different model providers."""
    yield ...  # type: ignore # pragma: no cover

BaseEmbedder

Bases: BaseModel, Generic[BaseEmbeddingT], ABC

The base class abstract interface for interacting with LLM embeddings.

Source code in mirascope/rag/embedders.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class BaseEmbedder(BaseModel, Generic[BaseEmbeddingT], ABC):
    """The base class abstract interface for interacting with LLM embeddings."""

    api_key: ClassVar[Optional[str]] = None
    base_url: ClassVar[Optional[str]] = None
    embedding_params: ClassVar[BaseEmbeddingParams] = BaseEmbeddingParams(
        model="text-embedding-ada-002"
    )
    dimensions: Optional[int] = None
    configuration: ClassVar[BaseConfig] = BaseConfig(llm_ops=[], client_wrappers=[])
    _provider: ClassVar[str] = "base"

    @abstractmethod
    def embed(self, input: list[str]) -> BaseEmbeddingT:
        """A call to the embedder with a single input"""
        ...  # pragma: no cover

    @abstractmethod
    async def embed_async(self, input: list[str]) -> BaseEmbeddingT:
        """Asynchronously call the embedder with a single input"""
        ...  # pragma: no cover

embed(input) abstractmethod

A call to the embedder with a single input

Source code in mirascope/rag/embedders.py 
26
27
28
29
@abstractmethod
def embed(self, input: list[str]) -> BaseEmbeddingT:
    """A call to the embedder with a single input"""
    ...  # pragma: no cover

embed_async(input) abstractmethod async

Asynchronously call the embedder with a single input

Source code in mirascope/rag/embedders.py 
31
32
33
34
@abstractmethod
async def embed_async(self, input: list[str]) -> BaseEmbeddingT:
    """Asynchronously call the embedder with a single input"""
    ...  # pragma: no cover

get_class_functions(cls)

Get the class functions of a BaseModel.

Source code in mirascope/base/ops_utils.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def get_class_functions(cls: type[BaseModel]) -> Generator[str, None, None]:
    """Get the class functions of a `BaseModel`."""
    ignore_functions = [
        "copy",
        "dict",
        "dump",
        "json",
        "messages",
        "model_copy",
        "model_dump",
        "model_dump_json",
        "model_post_init",
    ]
    for name, _ in inspect.getmembers(cls, predicate=inspect.isfunction):
        if not name.startswith("_") and name not in ignore_functions:
            yield name

get_class_vars(self)

Get the class variables of a BaseModel removing any dangerous variables.

Source code in mirascope/base/ops_utils.py 
13
14
15
16
17
18
19
def get_class_vars(self: BaseModel) -> dict[str, Any]:
    """Get the class variables of a `BaseModel` removing any dangerous variables."""
    class_vars = {}
    for classvars in self.__class_vars__:
        if not classvars == "api_key":
            class_vars[classvars] = getattr(self.__class__, classvars)
    return class_vars

get_wrapped_async_client(client, self)

Get a wrapped async client.

Source code in mirascope/base/ops_utils.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def get_wrapped_async_client(client: T, self: Union[BaseCall, BaseEmbedder]) -> T:
    """Get a wrapped async client."""
    if self.configuration.client_wrappers:
        for op in self.configuration.client_wrappers:
            if op == "langfuse":  # pragma: no cover
                from langfuse.openai import AsyncOpenAI as LangfuseAsyncOpenAI

                client = LangfuseAsyncOpenAI(
                    api_key=self.api_key, base_url=self.base_url
                )
            elif op == "logfire":  # pragma: no cover
                import logfire

                if self._provider == "openai":
                    logfire.instrument_openai(client)  # type: ignore
                elif self._provider == "anthropic":
                    logfire.instrument_anthropic(client)  # type: ignore
            elif callable(op):
                client = op(client)
    return client

get_wrapped_call(call, self, **kwargs)

Wrap a call to add the llm_ops parameter if it exists.

Source code in mirascope/base/ops_utils.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def get_wrapped_call(call: C, self: Union[BaseCall, BaseEmbedder], **kwargs) -> C:
    """Wrap a call to add the `llm_ops` parameter if it exists."""
    if self.configuration.llm_ops:
        wrapped_call = call
        for op in self.configuration.llm_ops:
            if op == "weave":  # pragma: no cover
                import weave

                wrapped_call = weave.op()(wrapped_call)
            elif callable(op):
                wrapped_call = op(
                    wrapped_call,
                    self._provider,
                    **kwargs,
                )
        return wrapped_call
    return call

get_wrapped_client(client, self)

Get a wrapped client.

Source code in mirascope/base/ops_utils.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def get_wrapped_client(client: T, self: Union[BaseCall, BaseEmbedder]) -> T:
    """Get a wrapped client."""
    if self.configuration.client_wrappers:
        for op in self.configuration.client_wrappers:  # pragma: no cover
            if op == "langfuse":
                from langfuse.openai import OpenAI as LangfuseOpenAI

                client = LangfuseOpenAI(api_key=self.api_key, base_url=self.base_url)
            elif op == "logfire":  # pragma: no cover
                import logfire

                if self._provider == "openai":
                    logfire.instrument_openai(client)  # type: ignore
                elif self._provider == "anthropic":
                    logfire.instrument_anthropic(client)  # type: ignore
            elif callable(op):
                client = op(client)
    return client

mirascope_span(fn, *, handle_before_call=None, handle_before_call_async=None, handle_after_call=None, handle_after_call_async=None, decorator=None, **custom_kwargs)

Wraps a pydantic class method.

Source code in mirascope/base/ops_utils.py 
 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
def mirascope_span(
    fn: Callable,
    *,
    handle_before_call: Optional[Callable[..., Any]] = None,
    handle_before_call_async: Optional[Callable[..., Awaitable[Any]]] = None,
    handle_after_call: Optional[Callable[..., Any]] = None,
    handle_after_call_async: Optional[Callable[..., Awaitable[Any]]] = None,
    decorator: Optional[DecoratorType] = None,
    **custom_kwargs: Any,
):
    """Wraps a pydantic class method."""

    async def run_after_call_handler(self, result, result_before_call, joined_kwargs):
        if handle_after_call_async is not None:
            await handle_after_call_async(
                self, fn, result, result_before_call, **joined_kwargs
            )
        elif handle_after_call is not None:
            handle_after_call(self, fn, result, result_before_call, **joined_kwargs)

    @wraps(fn)
    def wrapper(self: BaseModel, *args, **kwargs):
        """Wraps a pydantic class method that returns a value."""
        joined_kwargs = {**kwargs, **custom_kwargs}
        before_call = (
            handle_before_call(self, fn, **joined_kwargs)
            if handle_before_call is not None
            else None
        )
        if isinstance(before_call, AbstractContextManager):
            with before_call as result_before_call:
                result = fn(self, *args, **kwargs)
                if handle_after_call is not None:
                    handle_after_call(
                        self, fn, result, result_before_call, **joined_kwargs
                    )
                return result
        else:
            result = fn(self, *args, **kwargs)
            if handle_after_call is not None:
                handle_after_call(self, fn, result, before_call, **joined_kwargs)
            return result

    @wraps(fn)
    async def wrapper_async(self: BaseModel, *args, **kwargs):
        """Wraps a pydantic async class method that returns a value."""
        joined_kwargs = {**kwargs, **custom_kwargs}
        before_call = None
        if handle_before_call_async is not None:
            before_call = await handle_before_call_async(self, fn, **joined_kwargs)
        elif handle_before_call is not None:
            before_call = handle_before_call(self, fn, **joined_kwargs)

        if isinstance(before_call, AbstractContextManager):
            with before_call as result_before_call:
                result = await fn(self, *args, **kwargs)
                await run_after_call_handler(
                    self, result, result_before_call, joined_kwargs
                )
                return result
        else:
            result = await fn(self, *args, **kwargs)
            await run_after_call_handler(self, result, before_call, joined_kwargs)
            return result

    @wraps(fn)
    def wrapper_generator(self: BaseModel, *args, **kwargs):
        """Wraps a pydantic class method that returns a generator."""
        joined_kwargs = {**kwargs, **custom_kwargs}
        before_call = (
            handle_before_call(self, fn, **joined_kwargs)
            if handle_before_call is not None
            else None
        )
        if isinstance(before_call, AbstractContextManager):
            with before_call as result_before_call:
                result = fn(self, *args, **kwargs)
                output = []
                for value in result:
                    output.append(value)
                    yield value
                if handle_after_call is not None:
                    handle_after_call(
                        self, fn, output, result_before_call, **joined_kwargs
                    )
        else:
            result = fn(self, *args, **kwargs)
            output = []
            for value in result:
                output.append(value)
                yield value
            if handle_after_call is not None:
                handle_after_call(self, fn, output, before_call, **joined_kwargs)

    @wraps(fn)
    async def wrapper_generator_async(self: BaseModel, *args, **kwargs):
        """Wraps a pydantic async class method that returns a generator."""
        joined_kwargs = {**kwargs, **custom_kwargs}
        before_call = None
        if handle_before_call_async is not None:
            before_call = await handle_before_call_async(self, fn, **joined_kwargs)
        elif handle_before_call is not None:
            before_call = handle_before_call(self, fn, **joined_kwargs)
        if isinstance(before_call, AbstractContextManager):
            with before_call as result_before_call:
                result = fn(self, *args, **kwargs)
                output = []
                async for value in result:
                    output.append(value)
                    yield value
                await run_after_call_handler(
                    self, output, result_before_call, joined_kwargs
                )
        else:
            result = fn(self, *args, **kwargs)
            output = []
            async for value in result:
                output.append(value)
                yield value
            await run_after_call_handler(self, output, before_call, joined_kwargs)

    wrapper_function = wrapper
    if inspect.isasyncgenfunction(fn):
        wrapper_function = wrapper_generator_async
    elif inspect.iscoroutinefunction(fn):
        wrapper_function = wrapper_async
    elif inspect.isgeneratorfunction(fn):
        wrapper_function = wrapper_generator
    if decorator is not None:
        wrapper_function = decorator(wrapper_function)
    return wrapper_function

wrap_mirascope_class_functions(cls, *, handle_before_call=None, handle_before_call_async=None, handle_after_call=None, handle_after_call_async=None, decorator=None, **custom_kwargs)

Wraps Mirascope class functions with a decorator.

Parameters:

Name Type Description Default
cls type[BaseModel]

The Mirascope class to wrap.

 required
handle_before_call Optional[Callable[..., Any]]

A function to call before the call to the wrapped function.

 None
handle_after_call Optional[Callable[..., Any]]

A function to call after the call to the wrapped function.

 None
custom_kwargs Any

Additional keyword arguments to pass to the decorator.

 {}
Source code in mirascope/base/ops_utils.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def wrap_mirascope_class_functions(
    cls: type[BaseModel],
    *,
    handle_before_call: Optional[Callable[..., Any]] = None,
    handle_before_call_async: Optional[Callable[..., Awaitable[Any]]] = None,
    handle_after_call: Optional[Callable[..., Any]] = None,
    handle_after_call_async: Optional[Callable[..., Awaitable[Any]]] = None,
    decorator: Optional[DecoratorType] = None,
    **custom_kwargs: Any,
):
    """Wraps Mirascope class functions with a decorator.

    Args:
        cls: The Mirascope class to wrap.
        handle_before_call: A function to call before the call to the wrapped function.
        handle_after_call: A function to call after the call to the wrapped function.
        custom_kwargs: Additional keyword arguments to pass to the decorator.
    """

    for name in get_class_functions(cls):
        setattr(
            cls,
            name,
            mirascope_span(
                getattr(cls, name),
                handle_before_call=handle_before_call,
                handle_before_call_async=handle_before_call_async,
                handle_after_call=handle_after_call,
                handle_after_call_async=handle_after_call_async,
                decorator=decorator,
                **custom_kwargs,
            ),
        )
    return cls