Skip to content

cohere.types

Types for interacting with Cohere chat models using Mirascope.

BaseAsyncStream

Bases: Generic[BaseCallResponseChunkT, UserMessageParamT, AssistantMessageParamT, BaseToolT], ABC

A base class for async streaming responses from LLMs.

Source code in mirascope/base/types.py
class BaseAsyncStream(
    Generic[
        BaseCallResponseChunkT,
        UserMessageParamT,
        AssistantMessageParamT,
        BaseToolT,
    ],
    ABC,
):
    """A base class for async streaming responses from LLMs."""

    stream: AsyncGenerator[BaseCallResponseChunkT, None]
    message_param_type: type[AssistantMessageParamT]

    cost: Optional[float] = None
    user_message_param: Optional[UserMessageParamT] = None
    message_param: AssistantMessageParamT

    def __init__(
        self,
        stream: AsyncGenerator[BaseCallResponseChunkT, None],
        message_param_type: type[AssistantMessageParamT],
    ):
        """Initializes an instance of `BaseAsyncStream`."""
        self.stream = stream
        self.message_param_type = message_param_type

    def __aiter__(
        self,
    ) -> AsyncGenerator[tuple[BaseCallResponseChunkT, Optional[BaseToolT]], None]:
        """Iterates over the stream and stores useful information."""

        async def generator():
            content = ""
            async for chunk in self.stream:
                content += chunk.content
                if chunk.cost is not None:
                    self.cost = chunk.cost
                yield chunk, None
                self.user_message_param = chunk.user_message_param
            kwargs = {"role": "assistant"}
            if "message" in self.message_param_type.__annotations__:
                kwargs["message"] = content
            else:
                kwargs["content"] = content
            self.message_param = self.message_param_type(**kwargs)

        return generator()

BaseCallParams

Bases: BaseModel, Generic[BaseToolT]

The parameters with which to make a call.

Source code in mirascope/base/types.py
class BaseCallParams(BaseModel, Generic[BaseToolT]):
    """The parameters with which to make a call."""

    model: str
    tools: Optional[list[Union[Callable, Type[BaseToolT]]]] = None

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    def kwargs(
        self,
        tool_type: Optional[Type[BaseToolT]] = None,
        exclude: Optional[set[str]] = None,
    ) -> dict[str, Any]:
        """Returns all parameters for the call as a keyword arguments dictionary."""
        extra_exclude = {"tools"}
        exclude = extra_exclude if exclude is None else exclude.union(extra_exclude)
        kwargs = {
            key: value
            for key, value in self.model_dump(exclude=exclude).items()
            if value is not None
        }
        if not self.tools or tool_type is None:
            return kwargs
        kwargs["tools"] = [
            tool if isclass(tool) else convert_function_to_tool(tool, tool_type)
            for tool in self.tools
        ]
        return kwargs

kwargs(tool_type=None, exclude=None)

Returns all parameters for the call as a keyword arguments dictionary.

Source code in mirascope/base/types.py
def kwargs(
    self,
    tool_type: Optional[Type[BaseToolT]] = None,
    exclude: Optional[set[str]] = None,
) -> dict[str, Any]:
    """Returns all parameters for the call as a keyword arguments dictionary."""
    extra_exclude = {"tools"}
    exclude = extra_exclude if exclude is None else exclude.union(extra_exclude)
    kwargs = {
        key: value
        for key, value in self.model_dump(exclude=exclude).items()
        if value is not None
    }
    if not self.tools or tool_type is None:
        return kwargs
    kwargs["tools"] = [
        tool if isclass(tool) else convert_function_to_tool(tool, tool_type)
        for tool in self.tools
    ]
    return kwargs

BaseCallResponse

Bases: BaseModel, Generic[ResponseT, BaseToolT], ABC

A base abstract interface for LLM call responses.

Attributes:

Name Type Description
response ResponseT

The original response from whichever model response this wraps.

Source code in mirascope/base/types.py
class BaseCallResponse(BaseModel, Generic[ResponseT, BaseToolT], ABC):
    """A base abstract interface for LLM call responses.

    Attributes:
        response: The original response from whichever model response this wraps.
    """

    response: ResponseT
    user_message_param: Optional[Any] = None
    tool_types: Optional[list[Type[BaseToolT]]] = None
    start_time: float  # The start time of the completion in ms
    end_time: float  # The end time of the completion in ms
    cost: Optional[float] = None  # The cost of the completion in dollars

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    @property
    @abstractmethod
    def message_param(self) -> Any:
        """Returns the assistant's response as a message parameter."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def tools(self) -> Optional[list[BaseToolT]]:
        """Returns the tools for the 0th choice message."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def tool(self) -> Optional[BaseToolT]:
        """Returns the 0th tool for the 0th choice message."""
        ...  # pragma: no cover

    @classmethod
    @abstractmethod
    def tool_message_params(
        cls, tools_and_outputs: list[tuple[BaseToolT, Any]]
    ) -> list[Any]:
        """Returns the tool message parameters for tool call results."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def content(self) -> str:
        """Should return the string content of the response.

        If there are multiple choices in a response, this method should select the 0th
        choice and return it's string content.

        If there is no string content (e.g. when using tools), this method must return
        the empty string.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def finish_reasons(self) -> Union[None, list[str]]:
        """Should return the finish reasons of the response.

        If there is no finish reason, this method must return None.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def model(self) -> Optional[str]:
        """Should return the name of the response model."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def id(self) -> Optional[str]:
        """Should return the id of the response."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def usage(self) -> Any:
        """Should return the usage of the response.

        If there is no usage, this method must return None.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def input_tokens(self) -> Optional[Union[int, float]]:
        """Should return the number of input tokens.

        If there is no input_tokens, this method must return None.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def output_tokens(self) -> Optional[Union[int, float]]:
        """Should return the number of output tokens.

        If there is no output_tokens, this method must return None.
        """
        ...  # pragma: no cover

content: str abstractmethod property

Should return the string content of the response.

If there are multiple choices in a response, this method should select the 0th choice and return it's string content.

If there is no string content (e.g. when using tools), this method must return the empty string.

finish_reasons: Union[None, list[str]] abstractmethod property

Should return the finish reasons of the response.

If there is no finish reason, this method must return None.

id: Optional[str] abstractmethod property

Should return the id of the response.

input_tokens: Optional[Union[int, float]] abstractmethod property

Should return the number of input tokens.

If there is no input_tokens, this method must return None.

message_param: Any abstractmethod property

Returns the assistant's response as a message parameter.

model: Optional[str] abstractmethod property

Should return the name of the response model.

output_tokens: Optional[Union[int, float]] abstractmethod property

Should return the number of output tokens.

If there is no output_tokens, this method must return None.

tool: Optional[BaseToolT] abstractmethod property

Returns the 0th tool for the 0th choice message.

tools: Optional[list[BaseToolT]] abstractmethod property

Returns the tools for the 0th choice message.

usage: Any abstractmethod property

Should return the usage of the response.

If there is no usage, this method must return None.

tool_message_params(tools_and_outputs) abstractmethod classmethod

Returns the tool message parameters for tool call results.

Source code in mirascope/base/types.py
@classmethod
@abstractmethod
def tool_message_params(
    cls, tools_and_outputs: list[tuple[BaseToolT, Any]]
) -> list[Any]:
    """Returns the tool message parameters for tool call results."""
    ...  # pragma: no cover

BaseCallResponseChunk

Bases: BaseModel, Generic[ChunkT, BaseToolT], ABC

A base abstract interface for LLM streaming response chunks.

Attributes:

Name Type Description
response

The original response chunk from whichever model response this wraps.

Source code in mirascope/base/types.py
class BaseCallResponseChunk(BaseModel, Generic[ChunkT, BaseToolT], ABC):
    """A base abstract interface for LLM streaming response chunks.

    Attributes:
        response: The original response chunk from whichever model response this wraps.
    """

    chunk: ChunkT
    user_message_param: Optional[Any] = None
    tool_types: Optional[list[Type[BaseToolT]]] = None
    cost: Optional[float] = None  # The cost of the completion in dollars
    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    @property
    @abstractmethod
    def content(self) -> str:
        """Should return the string content of the response chunk.

        If there are multiple choices in a chunk, this method should select the 0th
        choice and return it's string content.

        If there is no string content (e.g. when using tools), this method must return
        the empty string.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def model(self) -> Optional[str]:
        """Should return the name of the response model."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def id(self) -> Optional[str]:
        """Should return the id of the response."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def finish_reasons(self) -> Union[None, list[str]]:
        """Should return the finish reasons of the response.

        If there is no finish reason, this method must return None.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def usage(self) -> Any:
        """Should return the usage of the response.

        If there is no usage, this method must return None.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def input_tokens(self) -> Optional[Union[int, float]]:
        """Should return the number of input tokens.

        If there is no input_tokens, this method must return None.
        """
        ...  # pragma: no cover

    @property
    @abstractmethod
    def output_tokens(self) -> Optional[Union[int, float]]:
        """Should return the number of output tokens.

        If there is no output_tokens, this method must return None.
        """
        ...  # pragma: no cover

content: str abstractmethod property

Should return the string content of the response chunk.

If there are multiple choices in a chunk, this method should select the 0th choice and return it's string content.

If there is no string content (e.g. when using tools), this method must return the empty string.

finish_reasons: Union[None, list[str]] abstractmethod property

Should return the finish reasons of the response.

If there is no finish reason, this method must return None.

id: Optional[str] abstractmethod property

Should return the id of the response.

input_tokens: Optional[Union[int, float]] abstractmethod property

Should return the number of input tokens.

If there is no input_tokens, this method must return None.

model: Optional[str] abstractmethod property

Should return the name of the response model.

output_tokens: Optional[Union[int, float]] abstractmethod property

Should return the number of output tokens.

If there is no output_tokens, this method must return None.

usage: Any abstractmethod property

Should return the usage of the response.

If there is no usage, this method must return None.

BaseEmbeddingParams

Bases: BaseModel

The parameters with which to make an embedding.

Source code in mirascope/rag/types.py
class BaseEmbeddingParams(BaseModel):
    """The parameters with which to make an embedding."""

    model: str

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    def kwargs(self) -> dict[str, Any]:
        """Returns all parameters for the embedder as a keyword arguments dictionary."""
        kwargs = {
            key: value for key, value in self.model_dump().items() if value is not None
        }
        return kwargs

kwargs()

Returns all parameters for the embedder as a keyword arguments dictionary.

Source code in mirascope/rag/types.py
def kwargs(self) -> dict[str, Any]:
    """Returns all parameters for the embedder as a keyword arguments dictionary."""
    kwargs = {
        key: value for key, value in self.model_dump().items() if value is not None
    }
    return kwargs

BaseEmbeddingResponse

Bases: BaseModel, Generic[ResponseT], ABC

A base abstract interface for LLM embedding responses.

Attributes:

Name Type Description
response ResponseT

The original response from whichever model response this wraps.

Source code in mirascope/rag/types.py
class BaseEmbeddingResponse(BaseModel, Generic[ResponseT], ABC):
    """A base abstract interface for LLM embedding responses.

    Attributes:
        response: The original response from whichever model response this wraps.
    """

    response: ResponseT
    start_time: float  # The start time of the embedding in ms
    end_time: float  # The end time of the embedding in ms

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    @property
    @abstractmethod
    def embeddings(self) -> Optional[Union[list[list[float]], list[list[int]]]]:
        """Should return the embedding of the response.

        If there are multiple choices in a response, this method should select the 0th
        choice and return it's embedding.
        """
        ...  # pragma: no cover

embeddings: Optional[Union[list[list[float]], list[list[int]]]] abstractmethod property

Should return the embedding of the response.

If there are multiple choices in a response, this method should select the 0th choice and return it's embedding.

BaseStream

Bases: Generic[BaseCallResponseChunkT, UserMessageParamT, AssistantMessageParamT, BaseToolT], ABC

A base class for streaming responses from LLMs.

Source code in mirascope/base/types.py
class BaseStream(
    Generic[
        BaseCallResponseChunkT,
        UserMessageParamT,
        AssistantMessageParamT,
        BaseToolT,
    ],
    ABC,
):
    """A base class for streaming responses from LLMs."""

    stream: Generator[BaseCallResponseChunkT, None, None]
    message_param_type: type[AssistantMessageParamT]

    cost: Optional[float] = None
    user_message_param: Optional[UserMessageParamT] = None
    message_param: AssistantMessageParamT

    def __init__(
        self,
        stream: Generator[BaseCallResponseChunkT, None, None],
        message_param_type: type[AssistantMessageParamT],
    ):
        """Initializes an instance of `BaseStream`."""
        self.stream = stream
        self.message_param_type = message_param_type

    def __iter__(
        self,
    ) -> Generator[tuple[BaseCallResponseChunkT, Optional[BaseToolT]], None, None]:
        """Iterator over the stream and stores useful information."""
        content = ""
        for chunk in self.stream:
            content += chunk.content
            if chunk.cost is not None:
                self.cost = chunk.cost
            yield chunk, None
            self.user_message_param = chunk.user_message_param
        kwargs = {"role": "assistant"}
        if "message" in self.message_param_type.__annotations__:
            kwargs["message"] = content
        else:
            kwargs["content"] = content
        self.message_param = self.message_param_type(**kwargs)

CohereAsyncStream

Bases: BaseAsyncStream[CohereCallResponseChunk, ChatMessage, ChatMessage, CohereTool]

A class for streaming responses from Cohere's API.

Source code in mirascope/cohere/types.py
class CohereAsyncStream(
    BaseAsyncStream[
        CohereCallResponseChunk,
        ChatMessage,
        ChatMessage,
        CohereTool,
    ]
):
    """A class for streaming responses from Cohere's API."""

    def __init__(self, stream: AsyncGenerator[CohereCallResponseChunk, None]):
        """Initializes an instance of `CohereAsyncStream`."""
        super().__init__(stream, ChatMessage)

CohereCallParams

Bases: BaseCallParams[CohereTool]

The parameters to use when calling the Cohere chat API.

Source code in mirascope/cohere/types.py
class CohereCallParams(BaseCallParams[CohereTool]):
    """The parameters to use when calling the Cohere chat API."""

    model: str = "command-r-plus"
    preamble: Optional[str] = None
    chat_history: Optional[Sequence[ChatMessage]] = None
    conversation_id: Optional[str] = None
    prompt_truncation: Optional[ChatRequestPromptTruncation] = None
    connectors: Optional[Sequence[ChatConnector]] = None
    search_queries_only: Optional[bool] = None
    documents: Optional[Sequence[ChatDocument]] = None
    temperature: Optional[float] = None
    max_tokens: Optional[int] = None
    max_input_tokens: Optional[int] = None
    k: Optional[int] = None
    p: Optional[float] = None
    seed: Optional[float] = None
    stop_sequences: Optional[Sequence[str]] = None
    frequency_penalty: Optional[float] = None
    presence_penalty: Optional[float] = None
    raw_prompting: Optional[bool] = None
    tool_results: Optional[Sequence[ToolResult]] = None
    request_options: Optional[RequestOptions] = None

    model_config = ConfigDict(arbitrary_types_allowed=True)

    def kwargs(
        self,
        tool_type: Optional[Type[CohereTool]] = CohereTool,
        exclude: Optional[set[str]] = None,
    ) -> dict[str, Any]:
        """Returns the keyword argument call parameters."""
        extra_exclude = {"wrapper", "wrapper_async"}
        exclude = extra_exclude if exclude is None else exclude.union(extra_exclude)
        return super().kwargs(tool_type, exclude)

kwargs(tool_type=CohereTool, exclude=None)

Returns the keyword argument call parameters.

Source code in mirascope/cohere/types.py
def kwargs(
    self,
    tool_type: Optional[Type[CohereTool]] = CohereTool,
    exclude: Optional[set[str]] = None,
) -> dict[str, Any]:
    """Returns the keyword argument call parameters."""
    extra_exclude = {"wrapper", "wrapper_async"}
    exclude = extra_exclude if exclude is None else exclude.union(extra_exclude)
    return super().kwargs(tool_type, exclude)

CohereCallResponse

Bases: BaseCallResponse[NonStreamedChatResponse, CohereTool]

A convenience wrapper around the Cohere NonStreamedChatResponse response.

When using Mirascope's convenience wrappers to interact with Cohere chat models via CohereCall, responses using CohereCall.call() will return a CohereCallResponse whereby the implemented properties allow for simpler syntax and a convenient developer experience.

Example:

from mirascope.cohere import CohereCall


class BookRecommender(CohereCall):
    prompt_template = "Please recommend a {genre} book"

    genre: str


response = Bookrecommender(genre="fantasy").call()
print(response.content)
#> The Name of the Wind

print(response.message)
#> ...

print(response.choices)
#> ...
Source code in mirascope/cohere/types.py
class CohereCallResponse(BaseCallResponse[NonStreamedChatResponse, CohereTool]):
    """A convenience wrapper around the Cohere `NonStreamedChatResponse` response.

    When using Mirascope's convenience wrappers to interact with Cohere chat models via
    `CohereCall`, responses using `CohereCall.call()` will return a `CohereCallResponse`
    whereby the implemented properties allow for simpler syntax and a convenient
    developer experience.

    Example:

    ```python
    from mirascope.cohere import CohereCall


    class BookRecommender(CohereCall):
        prompt_template = "Please recommend a {genre} book"

        genre: str


    response = Bookrecommender(genre="fantasy").call()
    print(response.content)
    #> The Name of the Wind

    print(response.message)
    #> ...

    print(response.choices)
    #> ...
    ```
    """

    # We need to skip validation since it's a pydantic_v1 model and breaks validation.
    response: SkipValidation[NonStreamedChatResponse]
    user_message_param: SkipValidation[Optional[ChatMessage]] = None

    @property
    def message_param(self) -> ChatMessage:
        """Returns the assistant's response as a message parameter."""
        return ChatMessage(
            message=self.response.text,
            tool_calls=self.response.tool_calls,
            role="assistant",  # type: ignore
        )

    @property
    def content(self) -> str:
        """Returns the content of the chat completion for the 0th choice."""
        return self.response.text

    @property
    def model(self) -> Optional[str]:
        """Returns the name of the response model.

        Cohere does not return model, so we return None
        """
        return None

    @property
    def id(self) -> Optional[str]:
        """Returns the id of the response."""
        return self.response.generation_id

    @property
    def finish_reasons(self) -> Optional[list[str]]:
        """Returns the finish reasons of the response."""
        return [str(self.response.finish_reason)]

    @property
    def search_queries(self) -> Optional[list[ChatSearchQuery]]:
        """Returns the search queries for the 0th choice message."""
        return self.response.search_queries

    @property
    def search_results(self) -> Optional[list[ChatSearchResult]]:
        """Returns the search results for the 0th choice message."""
        return self.response.search_results

    @property
    def documents(self) -> Optional[list[ChatDocument]]:
        """Returns the documents for the 0th choice message."""
        return self.response.documents

    @property
    def citations(self) -> Optional[list[ChatCitation]]:
        """Returns the citations for the 0th choice message."""
        return self.response.citations

    @property
    def tool_calls(self) -> Optional[list[ToolCall]]:
        """Returns the tool calls for the 0th choice message."""
        return self.response.tool_calls

    @property
    def tools(self) -> Optional[list[CohereTool]]:
        """Returns the tools for the 0th choice message.

        Raises:
            ValidationError: if a tool call doesn't match the tool's schema.
        """
        if not self.tool_types or not self.tool_calls:
            return None

        if self.response.finish_reason == "MAX_TOKENS":
            raise RuntimeError(
                "Generation stopped with MAX_TOKENS finish reason. This means that the "
                "response hit the token limit before completion."
            )

        extracted_tools = []
        for tool_call in self.tool_calls:
            for tool_type in self.tool_types:
                if tool_call.name == tool_type.name():
                    extracted_tools.append(tool_type.from_tool_call(tool_call))
                    break

        return extracted_tools

    @property
    def tool(self) -> Optional[CohereTool]:
        """Returns the 0th tool for the 0th choice message.

        Raises:
            ValidationError: if the tool call doesn't match the tool's schema.
        """
        tools = self.tools
        if tools:
            return tools[0]
        return None

    @classmethod
    def tool_message_params(
        self, tools_and_outputs: list[tuple[CohereTool, list[dict[str, Any]]]]
    ) -> list[ToolResult]:
        """Returns the tool message parameters for tool call results."""
        return [
            {"call": tool.tool_call, "outputs": output}  # type: ignore
            for tool, output in tools_and_outputs
        ]

    @property
    def usage(self) -> Optional[ApiMetaBilledUnits]:
        """Returns the usage of the response."""
        if self.response.meta:
            return self.response.meta.billed_units
        return None

    @property
    def input_tokens(self) -> Optional[float]:
        """Returns the number of input tokens."""
        if self.usage:
            return self.usage.input_tokens
        return None

    @property
    def output_tokens(self) -> Optional[float]:
        """Returns the number of output tokens."""
        if self.usage:
            return self.usage.output_tokens
        return None

    def dump(self) -> dict[str, Any]:
        """Dumps the response to a dictionary."""
        return {
            "start_time": self.start_time,
            "end_time": self.end_time,
            "output": self.response.dict(),
            "cost": self.cost,
        }

citations: Optional[list[ChatCitation]] property

Returns the citations for the 0th choice message.

content: str property

Returns the content of the chat completion for the 0th choice.

documents: Optional[list[ChatDocument]] property

Returns the documents for the 0th choice message.

finish_reasons: Optional[list[str]] property

Returns the finish reasons of the response.

id: Optional[str] property

Returns the id of the response.

input_tokens: Optional[float] property

Returns the number of input tokens.

message_param: ChatMessage property

Returns the assistant's response as a message parameter.

model: Optional[str] property

Returns the name of the response model.

Cohere does not return model, so we return None

output_tokens: Optional[float] property

Returns the number of output tokens.

search_queries: Optional[list[ChatSearchQuery]] property

Returns the search queries for the 0th choice message.

search_results: Optional[list[ChatSearchResult]] property

Returns the search results for the 0th choice message.

tool: Optional[CohereTool] property

Returns the 0th tool for the 0th choice message.

Raises:

Type Description
ValidationError

if the tool call doesn't match the tool's schema.

tool_calls: Optional[list[ToolCall]] property

Returns the tool calls for the 0th choice message.

tools: Optional[list[CohereTool]] property

Returns the tools for the 0th choice message.

Raises:

Type Description
ValidationError

if a tool call doesn't match the tool's schema.

usage: Optional[ApiMetaBilledUnits] property

Returns the usage of the response.

dump()

Dumps the response to a dictionary.

Source code in mirascope/cohere/types.py
def dump(self) -> dict[str, Any]:
    """Dumps the response to a dictionary."""
    return {
        "start_time": self.start_time,
        "end_time": self.end_time,
        "output": self.response.dict(),
        "cost": self.cost,
    }

tool_message_params(tools_and_outputs) classmethod

Returns the tool message parameters for tool call results.

Source code in mirascope/cohere/types.py
@classmethod
def tool_message_params(
    self, tools_and_outputs: list[tuple[CohereTool, list[dict[str, Any]]]]
) -> list[ToolResult]:
    """Returns the tool message parameters for tool call results."""
    return [
        {"call": tool.tool_call, "outputs": output}  # type: ignore
        for tool, output in tools_and_outputs
    ]

CohereCallResponseChunk

Bases: BaseCallResponseChunk[StreamedChatResponse, CohereTool]

Convenience wrapper around chat completion streaming chunks.

When using Mirascope's convenience wrappers to interact with Cohere models via CohereCall.stream, responses will return an CohereCallResponseChunk, whereby the implemented properties allow for simpler syntax and a convenient developer experience.

Example:

from mirascope.cohere import CohereCall


class Math(CohereCall):
    prompt_template = "What is 1 + 2?"


content = ""
for chunk in Math().stream():
    content += chunk.content
    print(content)
#> 1
#  1 +
#  1 + 2
#  1 + 2 equals
#  1 + 2 equals
#  1 + 2 equals 3
#  1 + 2 equals 3.
Source code in mirascope/cohere/types.py
class CohereCallResponseChunk(BaseCallResponseChunk[StreamedChatResponse, CohereTool]):
    """Convenience wrapper around chat completion streaming chunks.

    When using Mirascope's convenience wrappers to interact with Cohere models via
    `CohereCall.stream`, responses will return an `CohereCallResponseChunk`, whereby
    the implemented properties allow for simpler syntax and a convenient developer
    experience.

    Example:

    ```python
    from mirascope.cohere import CohereCall


    class Math(CohereCall):
        prompt_template = "What is 1 + 2?"


    content = ""
    for chunk in Math().stream():
        content += chunk.content
        print(content)
    #> 1
    #  1 +
    #  1 + 2
    #  1 + 2 equals
    #  1 + 2 equals
    #  1 + 2 equals 3
    #  1 + 2 equals 3.
    ```
    """

    chunk: SkipValidation[StreamedChatResponse]
    user_message_param: SkipValidation[Optional[ChatMessage]] = None

    @property
    def event_type(
        self,
    ) -> Literal[
        "stream-start",
        "search-queries-generation",
        "search-results",
        "text-generation",
        "citation-generation",
        "tool-calls-generation",
        "stream-end",
        "tool-calls-chunk",
    ]:
        """Returns the type of the chunk."""
        return self.chunk.event_type

    @property
    def content(self) -> str:
        """Returns the content for the 0th choice delta."""
        if isinstance(self.chunk, StreamedChatResponse_TextGeneration):
            return self.chunk.text
        return ""

    @property
    def search_queries(self) -> Optional[list[ChatSearchQuery]]:
        """Returns the search queries for search-query event type else None."""
        if isinstance(self.chunk, StreamedChatResponse_SearchQueriesGeneration):
            return self.chunk.search_queries  # type: ignore
        return None

    @property
    def search_results(self) -> Optional[list[ChatSearchResult]]:
        """Returns the search results for search-results event type else None."""
        if isinstance(self.chunk, StreamedChatResponse_SearchResults):
            return self.chunk.search_results
        return None

    @property
    def documents(self) -> Optional[list[ChatDocument]]:
        """Returns the documents for search-results event type else None."""
        if isinstance(self.chunk, StreamedChatResponse_SearchResults):
            return self.chunk.documents
        return None

    @property
    def citations(self) -> Optional[list[ChatCitation]]:
        """Returns the citations for citation-generation event type else None."""
        if isinstance(self.chunk, StreamedChatResponse_CitationGeneration):
            return self.chunk.citations
        return None

    @property
    def model(self) -> Optional[str]:
        """Returns the name of the response model.

        Cohere does not return model, so we return None
        """
        return None

    @property
    def id(self) -> Optional[str]:
        """Returns the id of the response."""
        if isinstance(self.chunk, StreamedChatResponse_StreamStart):
            return self.chunk.generation_id
        return None

    @property
    def finish_reasons(self) -> Optional[list[str]]:
        """Returns the finish reasons of the response."""
        if isinstance(self.chunk, StreamedChatResponse_StreamEnd):
            return [str(self.chunk.finish_reason)]
        return None

    @property
    def response(self) -> Optional[NonStreamedChatResponse]:
        """Returns the full response for the stream-end event type else None."""
        if isinstance(self.chunk, StreamedChatResponse_StreamEnd):
            return self.chunk.response
        return None

    @property
    def tool_calls(self) -> Optional[list[ToolCall]]:
        """Returns the partial tool calls for the 0th choice message."""
        if isinstance(self.chunk, StreamedChatResponse_ToolCallsGeneration):
            return self.chunk.tool_calls
        return None

    @property
    def usage(self) -> Optional[ApiMetaBilledUnits]:
        """Returns the usage of the response."""
        if isinstance(self.chunk, StreamedChatResponse_StreamEnd):
            if self.chunk.response.meta:
                return self.chunk.response.meta.billed_units
        return None

    @property
    def input_tokens(self) -> Optional[float]:
        """Returns the number of input tokens."""
        if self.usage:
            return self.usage.input_tokens
        return None

    @property
    def output_tokens(self) -> Optional[float]:
        """Returns the number of output tokens."""
        if self.usage:
            return self.usage.output_tokens
        return None

citations: Optional[list[ChatCitation]] property

Returns the citations for citation-generation event type else None.

content: str property

Returns the content for the 0th choice delta.

documents: Optional[list[ChatDocument]] property

Returns the documents for search-results event type else None.

event_type: Literal['stream-start', 'search-queries-generation', 'search-results', 'text-generation', 'citation-generation', 'tool-calls-generation', 'stream-end', 'tool-calls-chunk'] property

Returns the type of the chunk.

finish_reasons: Optional[list[str]] property

Returns the finish reasons of the response.

id: Optional[str] property

Returns the id of the response.

input_tokens: Optional[float] property

Returns the number of input tokens.

model: Optional[str] property

Returns the name of the response model.

Cohere does not return model, so we return None

output_tokens: Optional[float] property

Returns the number of output tokens.

response: Optional[NonStreamedChatResponse] property

Returns the full response for the stream-end event type else None.

search_queries: Optional[list[ChatSearchQuery]] property

Returns the search queries for search-query event type else None.

search_results: Optional[list[ChatSearchResult]] property

Returns the search results for search-results event type else None.

tool_calls: Optional[list[ToolCall]] property

Returns the partial tool calls for the 0th choice message.

usage: Optional[ApiMetaBilledUnits] property

Returns the usage of the response.

CohereEmbeddingResponse

Bases: BaseEmbeddingResponse[SkipValidation[EmbedResponse]]

A convenience wrapper around the Cohere EmbedResponse response.

Source code in mirascope/cohere/types.py
class CohereEmbeddingResponse(BaseEmbeddingResponse[SkipValidation[EmbedResponse]]):
    """A convenience wrapper around the Cohere `EmbedResponse` response."""

    embedding_type: Optional[
        Literal["float", "int8", "uint8", "binary", "ubinary"]
    ] = None

    @property
    def embeddings(
        self,
    ) -> Optional[Union[list[list[float]], list[list[int]]]]:
        """Returns the embeddings"""
        if self.response.response_type == "embeddings_floats":
            return self.response.embeddings
        else:
            embedding_type = self.embedding_type
            if embedding_type == "float":
                embedding_type == "float_"

            # TODO: Update to model_dump when Cohere updates to Pydantic v2
            embeddings_by_type: EmbedByTypeResponseEmbeddings = self.response.embeddings
            embedding_dict = embeddings_by_type.dict()
            return embedding_dict.get(str(embedding_type), None)

embeddings: Optional[Union[list[list[float]], list[list[int]]]] property

Returns the embeddings

CohereStream

Bases: BaseStream[CohereCallResponseChunk, ChatMessage, ChatMessage, CohereTool]

A class for streaming responses from Cohere's API.

Source code in mirascope/cohere/types.py
class CohereStream(
    BaseStream[
        CohereCallResponseChunk,
        ChatMessage,
        ChatMessage,
        CohereTool,
    ]
):
    """A class for streaming responses from Cohere's API."""

    def __init__(self, stream: Generator[CohereCallResponseChunk, None, None]):
        """Initializes an instance of `CohereStream`."""
        super().__init__(stream, ChatMessage)

CohereTool

Bases: BaseTool[ToolCall]

A base class for easy use of tools with the Cohere chat client.

CohereTool internally handles the logic that allows you to use tools with simple calls such as CohereCallResponse.tool or CohereTool.fn, as seen in the examples below.

Example:

from mirascope.cohere import CohereCall, CohereCallParams


def animal_matcher(fav_food: str, fav_color: str) -> str:
    """Tells you your most likely favorite animal from personality traits.

    Args:
        fav_food: your favorite food.
        fav_color: your favorite color.

    Returns:
        The animal most likely to be your favorite based on traits.
    """
    return "Your favorite animal is the best one, a frog."


class AnimalMatcher(CohereCall):
    prompt_template = """
    Tell me my favorite animal if my favorite food is {food} and my
    favorite color is {color}.
    """

    food: str
    color: str

    call_params = CohereCallParams(tools=[animal_matcher])


response = AnimalMatcher(food="pizza", color="red").call
tool = response.tool
print(tool.fn(**tool.args))
#> Your favorite animal is the best one, a frog.
Source code in mirascope/cohere/tools.py
class CohereTool(BaseTool[ToolCall]):
    '''A base class for easy use of tools with the Cohere chat client.

    `CohereTool` internally handles the logic that allows you to use tools with simple
    calls such as `CohereCallResponse.tool` or `CohereTool.fn`, as seen in the
    examples below.

    Example:

    ```python
    from mirascope.cohere import CohereCall, CohereCallParams


    def animal_matcher(fav_food: str, fav_color: str) -> str:
        """Tells you your most likely favorite animal from personality traits.

        Args:
            fav_food: your favorite food.
            fav_color: your favorite color.

        Returns:
            The animal most likely to be your favorite based on traits.
        """
        return "Your favorite animal is the best one, a frog."


    class AnimalMatcher(CohereCall):
        prompt_template = """
        Tell me my favorite animal if my favorite food is {food} and my
        favorite color is {color}.
        """

        food: str
        color: str

        call_params = CohereCallParams(tools=[animal_matcher])


    response = AnimalMatcher(food="pizza", color="red").call
    tool = response.tool
    print(tool.fn(**tool.args))
    #> Your favorite animal is the best one, a frog.
    ```
    '''

    tool_call: SkipJsonSchema[SkipValidation[ToolCall]]

    @classmethod
    def tool_schema(cls) -> Tool:
        """Constructs a tool schema for use with the Cohere chat client.

        A Mirascope `CohereTool` is deconstructed into a JSON schema, and relevant keys
        are renamed to match the Cohere tool schema used to make function/tool calls in
        the Cohere chat API.

        Returns:
            The constructed tool schema.
        """
        tool_schema = super().tool_schema()
        parameter_definitions = None
        if "parameters" in tool_schema:
            if "$defs" in tool_schema["parameters"]:
                raise ValueError(
                    "Unfortunately Cohere's chat API cannot handle nested structures "
                    "with $defs."
                )
            parameter_definitions = {
                prop: ToolParameterDefinitionsValue(
                    description=prop_schema["description"]
                    if "description" in prop_schema
                    else None,
                    type=prop_schema["type"],
                    required="required" in tool_schema["parameters"]
                    and prop in tool_schema["parameters"]["required"],
                )
                for prop, prop_schema in tool_schema["parameters"]["properties"].items()
            }
        return Tool(
            name=tool_schema["name"],
            description=tool_schema["description"],
            parameter_definitions=parameter_definitions,
        )

    @classmethod
    def from_tool_call(cls, tool_call: ToolCall) -> CohereTool:
        """Extracts an instance of the tool constructed from a tool call response.

        Given `ToolCall` from an Cohere chat completion response, takes its function
        arguments and creates an `CohereTool` instance from it.

        Args:
            tool_call: The `...` to extract the tool from.

        Returns:
            An instance of the tool constructed from the tool call.

        Raises:
            ValidationError: if the tool call doesn't match the tool schema.
        """
        model_json = tool_call.parameters
        model_json["tool_call"] = tool_call  # type: ignore
        return cls.model_validate(model_json)

    @classmethod
    def from_model(cls, model: Type[BaseModel]) -> Type[CohereTool]:
        """Constructs a `CohereTool` type from a `BaseModel` type."""
        return convert_base_model_to_tool(model, CohereTool)

    @classmethod
    def from_fn(cls, fn: Callable) -> Type[CohereTool]:
        """Constructs a `CohereTool` type from a function."""
        return convert_function_to_tool(fn, CohereTool)

    @classmethod
    def from_base_type(cls, base_type: Type[BaseType]) -> Type[CohereTool]:
        """Constructs a `CohereTool` type from a `BaseType` type."""
        return convert_base_type_to_tool(base_type, CohereTool)

from_base_type(base_type) classmethod

Constructs a CohereTool type from a BaseType type.

Source code in mirascope/cohere/tools.py
@classmethod
def from_base_type(cls, base_type: Type[BaseType]) -> Type[CohereTool]:
    """Constructs a `CohereTool` type from a `BaseType` type."""
    return convert_base_type_to_tool(base_type, CohereTool)

from_fn(fn) classmethod

Constructs a CohereTool type from a function.

Source code in mirascope/cohere/tools.py
@classmethod
def from_fn(cls, fn: Callable) -> Type[CohereTool]:
    """Constructs a `CohereTool` type from a function."""
    return convert_function_to_tool(fn, CohereTool)

from_model(model) classmethod

Constructs a CohereTool type from a BaseModel type.

Source code in mirascope/cohere/tools.py
@classmethod
def from_model(cls, model: Type[BaseModel]) -> Type[CohereTool]:
    """Constructs a `CohereTool` type from a `BaseModel` type."""
    return convert_base_model_to_tool(model, CohereTool)

from_tool_call(tool_call) classmethod

Extracts an instance of the tool constructed from a tool call response.

Given ToolCall from an Cohere chat completion response, takes its function arguments and creates an CohereTool instance from it.

Parameters:

Name Type Description Default
tool_call ToolCall

The ... to extract the tool from.

required

Returns:

Type Description
CohereTool

An instance of the tool constructed from the tool call.

Raises:

Type Description
ValidationError

if the tool call doesn't match the tool schema.

Source code in mirascope/cohere/tools.py
@classmethod
def from_tool_call(cls, tool_call: ToolCall) -> CohereTool:
    """Extracts an instance of the tool constructed from a tool call response.

    Given `ToolCall` from an Cohere chat completion response, takes its function
    arguments and creates an `CohereTool` instance from it.

    Args:
        tool_call: The `...` to extract the tool from.

    Returns:
        An instance of the tool constructed from the tool call.

    Raises:
        ValidationError: if the tool call doesn't match the tool schema.
    """
    model_json = tool_call.parameters
    model_json["tool_call"] = tool_call  # type: ignore
    return cls.model_validate(model_json)

tool_schema() classmethod

Constructs a tool schema for use with the Cohere chat client.

A Mirascope CohereTool is deconstructed into a JSON schema, and relevant keys are renamed to match the Cohere tool schema used to make function/tool calls in the Cohere chat API.

Returns:

Type Description
Tool

The constructed tool schema.

Source code in mirascope/cohere/tools.py
@classmethod
def tool_schema(cls) -> Tool:
    """Constructs a tool schema for use with the Cohere chat client.

    A Mirascope `CohereTool` is deconstructed into a JSON schema, and relevant keys
    are renamed to match the Cohere tool schema used to make function/tool calls in
    the Cohere chat API.

    Returns:
        The constructed tool schema.
    """
    tool_schema = super().tool_schema()
    parameter_definitions = None
    if "parameters" in tool_schema:
        if "$defs" in tool_schema["parameters"]:
            raise ValueError(
                "Unfortunately Cohere's chat API cannot handle nested structures "
                "with $defs."
            )
        parameter_definitions = {
            prop: ToolParameterDefinitionsValue(
                description=prop_schema["description"]
                if "description" in prop_schema
                else None,
                type=prop_schema["type"],
                required="required" in tool_schema["parameters"]
                and prop in tool_schema["parameters"]["required"],
            )
            for prop, prop_schema in tool_schema["parameters"]["properties"].items()
        }
    return Tool(
        name=tool_schema["name"],
        description=tool_schema["description"],
        parameter_definitions=parameter_definitions,
    )

RequestOptions

Bases: TypedDict

Redefining their class to use typing_extensions.TypedDict for Pydantic.

Source code in mirascope/cohere/types.py
class RequestOptions(TypedDict):
    """Redefining their class to use `typing_extensions.TypedDict` for Pydantic."""

    timeout_in_seconds: NotRequired[int]
    max_retries: NotRequired[int]
    additional_headers: NotRequired[dict[str, Any]]
    additional_query_parameters: NotRequired[dict[str, Any]]
    additional_body_parameters: NotRequired[dict[str, Any]]