code: """Wrapper around Cohere APIs."""
  from __future__ import annotations
  
  import logging
  from typing import Any, Callable, Dict, List, Optional
  
  from pydantic import Extra, root_validator
  from tenacity import (
      before_sleep_log,
      retry,
      retry_if_exception_type,
      stop_after_attempt,
      wait_exponential,
  )
  
  from langchain.callbacks.manager import CallbackManagerForLLMRun
  from langchain.llms.base import LLM
  from langchain.llms.utils import enforce_stop_tokens
  from langchain.utils import get_from_dict_or_env
  
  logger = logging.getLogger(__name__)
  
  
  def _create_retry_decorator(llm: Cohere) -> Callable[[Any], Any]:
      import cohere
  
      min_seconds = 4
      max_seconds = 10
      # Wait 2^x * 1 second between each retry starting with
      # 4 seconds, then up to 10 seconds, then 10 seconds afterwards
      return retry(
          reraise=True,
          stop=stop_after_attempt(llm.max_retries),
          wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds),
          retry=(retry_if_exception_type(cohere.error.CohereError)),
          before_sleep=before_sleep_log(logger, logging.WARNING),
      )
  
  
  def completion_with_retry(llm: Cohere, **kwargs: Any) -> Any:
      """Use tenacity to retry the completion call."""
      retry_decorator = _create_retry_decorator(llm)
  
      @retry_decorator
      def _completion_with_retry(**kwargs: Any) -> Any:
          return llm.client.generate(**kwargs)
  
      return _completion_with_retry(**kwargs)
  
  
  class Cohere(LLM):
      """Wrapper around Cohere large language models.
  
      To use, you should have the ``cohere`` python package installed, and the
      environment variable ``COHERE_API_KEY`` set with your API key, or pass
      it as a named parameter to the constructor.
  
      Example:
          .. code-block:: python
  
              from langchain.llms import Cohere
              cohere = Cohere(model="gptd-instruct-tft", cohere_api_key="my-api-key")
      """
  
      client: Any  #: :meta private:
      model: Optional[str] = None
      """Model name to use."""
  
      max_tokens: int = 256
      """Denotes the number of tokens to predict per generation."""
  
      temperature: float = 0.75
      """A non-negative float that tunes the degree of randomness in generation."""
  
      k: int = 0
      """Number of most likely tokens to consider at each step."""
  
      p: int = 1
      """Total probability mass of tokens to consider at each step."""
  
      frequency_penalty: float = 0.0
      """Penalizes repeated tokens according to frequency. Between 0 and 1."""
  
      presence_penalty: float = 0.0
      """Penalizes repeated tokens. Between 0 and 1."""
  
      truncate: Optional[str] = None
      """Specify how the client handles inputs longer than the maximum token
      length: Truncate from START, END or NONE"""
  
      max_retries: int = 10
      """Maximum number of retries to make when generating."""
  
      cohere_api_key: Optional[str] = None
  
      stop: Optional[List[str]] = None
  
      class Config:
          """Configuration for this pydantic object."""
  
          extra = Extra.forbid
  
      @root_validator()
      def validate_environment(cls, values: Dict) -> Dict:
          """Validate that api key and python package exists in environment."""
          cohere_api_key = get_from_dict_or_env(
              values, "cohere_api_key", "COHERE_API_KEY"
          )
          try:
              import cohere
  
              values["client"] = cohere.Client(cohere_api_key)
          except ImportError:
              raise ImportError(
                  "Could not import cohere python package. "
                  "Please install it with `pip install cohere`."
              )
          return values
  
      @property
      def _default_params(self) -> Dict[str, Any]:
          """Get the default parameters for calling Cohere API."""
          return {
              "max_tokens": self.max_tokens,
              "temperature": self.temperature,
              "k": self.k,
              "p": self.p,
              "frequency_penalty": self.frequency_penalty,
              "presence_penalty": self.presence_penalty,
              "truncate": self.truncate,
          }
  
      @property
      def _identifying_params(self) -> Dict[str, Any]:
          """Get the identifying parameters."""
          return {**{"model": self.model}, **self._default_params}
  
      @property
      def _llm_type(self) -> str:
          """Return type of llm."""
          return "cohere"
  
      def _call(
          self,
          prompt: str,
          stop: Optional[List[str]] = None,
          run_manager: Optional[CallbackManagerForLLMRun] = None,
      ) -> str:
          """Call out to Cohere's generate endpoint.
  
          Args:
              prompt: The prompt to pass into the model.
              stop: Optional list of stop words to use when generating.
  
          Returns:
              The string generated by the model.
  
          Example:
              .. code-block:: python
  
                  response = cohere("Tell me a joke.")
          """
          params = self._default_params
          if self.stop is not None and stop is not None:
              raise ValueError("`stop` found in both the input and default params.")
          elif self.stop is not None:
              params["stop_sequences"] = self.stop
          else:
              params["stop_sequences"] = stop
  
          response = completion_with_retry(
              self, model=self.model, prompt=prompt, **params
          )
          text = response.generations[0].text
          # If stop tokens are provided, Cohere's endpoint returns them.
          # In order to make this consistent with other endpoints, we strip them.
          if stop is not None or self.stop is not None:
              text = enforce_stop_tokens(text, params["stop_sequences"])
          return text
  
apis: langchain.llms.utils.enforce_stop_tokens langchain.utils.get_from_dict_or_env
extract_api: [((531, 558), 'logging.getLogger', 'logging.getLogger', (['__name__'], {}), '(__name__)\n', (548, 558), False, 'import logging\n'), ((3018, 3034), 'pydantic.root_validator', 'root_validator', ([], {}), '()\n', (3032, 3034), False, 'from pydantic import Extra, root_validator\n'), ((3195, 3259), 'langchain.utils.get_from_dict_or_env', 'get_from_dict_or_env', (['values', '"""cohere_api_key"""', '"""COHERE_API_KEY"""'], {}), "(values, 'cohere_api_key', 'COHERE_API_KEY')\n", (3215, 3259), False, 'from langchain.utils import get_from_dict_or_env\n'), ((866, 901), 'tenacity.stop_after_attempt', 'stop_after_attempt', (['llm.max_retries'], {}), '(llm.max_retries)\n', (884, 901), False, 'from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_exponential\n'), ((916, 980), 'tenacity.wait_exponential', 'wait_exponential', ([], {'multiplier': '(1)', 'min': 'min_seconds', 'max': 'max_seconds'}), '(multiplier=1, min=min_seconds, max=max_seconds)\n', (932, 980), False, 'from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_exponential\n'), ((997, 1046), 'tenacity.retry_if_exception_type', 'retry_if_exception_type', (['cohere.error.CohereError'], {}), '(cohere.error.CohereError)\n', (1020, 1046), False, 'from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_exponential\n'), ((1070, 1111), 'tenacity.before_sleep_log', 'before_sleep_log', (['logger', 'logging.WARNING'], {}), '(logger, logging.WARNING)\n', (1086, 1111), False, 'from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_exponential\n'), ((3353, 3382), 'cohere.Client', 'cohere.Client', (['cohere_api_key'], {}), '(cohere_api_key)\n', (3366, 3382), False, 'import cohere\n'), ((5575, 5626), 'langchain.llms.utils.enforce_stop_tokens', 'enforce_stop_tokens', (['text', "params['stop_sequences']"], {}), "(text, params['stop_sequences'])\n", (5594, 5626), False, 'from langchain.llms.utils import enforce_stop_tokens\n')]