Openai llm

OpenAILLM

Bases: LLM

The openai llm class.

Attributes:

Name	Type	Description
openai_api_key	Optional[str]	The API key for the OpenAI API.
openai_organization	Optional[str]	The OpenAI organization.
openai_api_base	Optional[str]	The OpenAI base url.
openai_client_args	Optional[dict]	Additional arguments to pass to the OpenAI client.

Source code in agentuniverse/llm/openai_llm.py

Python

class OpenAILLM(LLM):
    """The openai llm class.

    Attributes:
        openai_api_key (Optional[str], optional): The API key for the OpenAI API.
        This automatically infers the `openai_api_key` from the environment variable `OPENAI_API_KEY` if not provided.

        openai_organization (Optional[str], optional): The OpenAI organization.
        This automatically infers the `openai_organization` from the environment variable `OPENAI_ORGANIZATION` if not provided.

        openai_api_base (Optional[str], optional): The OpenAI base url.
        This automatically infers the `openai_api_base` from the environment variable `OPENAI_API_BASE` if not provided.

        openai_client_args (Optional[dict], optional): Additional arguments to pass to the OpenAI client.
   """

    openai_api_key: Optional[str] = Field(default_factory=lambda: get_from_env("OPENAI_API_KEY"))
    openai_organization: Optional[str] = Field(default_factory=lambda: get_from_env("OPENAI_ORGANIZATION"))
    openai_api_base: Optional[str] = Field(default_factory=lambda: get_from_env("OPENAI_API_BASE"))
    openai_proxy: Optional[str] = Field(default_factory=lambda: get_from_env("OPENAI_PROXY"))
    openai_client_args: Optional[dict] = None

    def _new_client(self):
        """Initialize the openai client."""
        return OpenAI(
            api_key=self.openai_api_key,
            organization=self.openai_organization,
            base_url=self.openai_api_base,
            timeout=self.request_timeout,
            max_retries=self.max_retries,
            http_client=httpx.Client(proxy=self.openai_proxy) if self.openai_proxy else None,
            **(self.openai_client_args or {}),
        )

    def _new_async_client(self):
        """Initialize the openai async client."""
        return AsyncOpenAI(
            api_key=self.openai_api_key,
            organization=self.openai_organization,
            base_url=self.openai_api_base,
            timeout=self.request_timeout,
            max_retries=self.max_retries,
            http_client=httpx.AsyncClient(proxy=self.openai_proxy) if self.openai_proxy else None,
            **(self.openai_client_args or {}),
        )

    def call(self, messages: list, **kwargs: Any) -> Union[LLMOutput, Iterator[LLMOutput]]:
        """Run the OpenAI LLM.

        Args:
            messages (list): The messages to send to the LLM.
            **kwargs: Arbitrary keyword arguments.
        """
        streaming = kwargs.pop("streaming") if "streaming" in kwargs else self.streaming
        if self.client is None:
            self.client = self._new_client()
        chat_completion = self.client.chat.completions.create(
            messages=messages,
            model=kwargs.pop('model', self.model_name),
            temperature=kwargs.pop('temperature', self.temperature),
            stream=kwargs.pop('stream', streaming),
            max_tokens=kwargs.pop('max_tokens', self.max_tokens),
            **kwargs,
        )
        if not streaming:
            text = chat_completion.choices[0].message.content
            return LLMOutput(text=text, raw=chat_completion.model_dump())
        return self.generate_stream_result(chat_completion)

    async def acall(self, messages: list, **kwargs: Any) -> Union[LLMOutput, AsyncIterator[LLMOutput]]:
        """Asynchronously run the OpenAI LLM.

        Args:
            messages (list): The messages to send to the LLM.
            **kwargs: Arbitrary keyword arguments.
        """
        streaming = kwargs.pop("streaming") if "streaming" in kwargs else self.streaming
        if self.async_client is None:
            self.async_client = self._new_async_client()
        chat_completion = await self.async_client.chat.completions.create(
            messages=messages,
            model=kwargs.pop('model', self.model_name),
            temperature=kwargs.pop('temperature', self.temperature),
            stream=kwargs.pop('stream', streaming),
            max_tokens=kwargs.pop('max_tokens', self.max_tokens),
            **kwargs,
        )
        if not streaming:
            text = chat_completion.choices[0].message.content
            return LLMOutput(text=text, raw=chat_completion.model_dump())
        return self.agenerate_stream_result(chat_completion)

    def as_langchain(self) -> BaseLanguageModel:
        """Convert the agentUniverse(aU) openai llm class to the langchain openai llm class."""
        return LangchainOpenAI(self)

    def set_by_agent_model(self, **kwargs) -> None:
        """ Assign values of parameters to the OpenAILLM model in the agent configuration."""
        super().set_by_agent_model(**kwargs)
        if 'openai_api_key' in kwargs and kwargs['openai_api_key']:
            self.openai_api_key = kwargs['openai_api_key']
        if 'openai_api_base' in kwargs and kwargs['openai_api_base']:
            self.openai_api_base = kwargs['openai_api_base']
        if 'openai_proxy' in kwargs and kwargs['openai_proxy']:
            self.openai_proxy = kwargs['openai_proxy']
        if 'openai_client_args' in kwargs and kwargs['openai_client_args']:
            self.openai_client_args = kwargs['openai_client_args']

    def max_context_length(self) -> int:
        """Max context length.

          The total length of input tokens and generated tokens is limited by the openai model's context length.
          """
        return OPENAI_MAX_CONTEXT_LENGTH.get(self.model_name, 4096)

    def get_num_tokens(self, text: str) -> int:
        """Get the number of tokens present in the text.

        Useful for checking if an input will fit in an openai model's context window.

        Args:
            text: The string input to tokenize.

        Returns:
            The integer number of tokens in the text.
        """
        try:
            encoding = tiktoken.encoding_for_model(self.model_name)
        except KeyError:
            encoding = tiktoken.get_encoding("cl100k_base")
        return len(encoding.encode(text))

    @staticmethod
    def parse_result(chunk):
        """Generate the result of the stream."""
        chat_completion = chunk
        if not isinstance(chunk, dict):
            chunk = chunk.dict()
        if len(chunk["choices"]) == 0:
            return
        choice = chunk["choices"][0]
        message = choice.get("delta")
        text = message.get("content")
        if not text:
            return
        return LLMOutput(text=text, raw=chat_completion.model_dump())

    @classmethod
    def generate_stream_result(cls, stream: Iterator) -> Iterator[LLMOutput]:
        """Generate the result of the stream."""
        for chunk in stream:
            llm_output = cls.parse_result(chunk)
            if llm_output:
                yield llm_output

    @classmethod
    async def agenerate_stream_result(cls, stream: AsyncIterator) -> AsyncIterator[LLMOutput]:
        """Generate the result of the stream."""
        async for chunk in stream:
            llm_output = cls.parse_result(chunk)
            if llm_output:
                yield llm_output

acall(messages, **kwargs) async

Asynchronously run the OpenAI LLM.

Parameters:

Name	Type	Description	Default
messages	list	The messages to send to the LLM.	required
**kwargs	Any	Arbitrary keyword arguments.	{}

Source code in agentuniverse/llm/openai_llm.py

Python

async def acall(self, messages: list, **kwargs: Any) -> Union[LLMOutput, AsyncIterator[LLMOutput]]:
    """Asynchronously run the OpenAI LLM.

    Args:
        messages (list): The messages to send to the LLM.
        **kwargs: Arbitrary keyword arguments.
    """
    streaming = kwargs.pop("streaming") if "streaming" in kwargs else self.streaming
    if self.async_client is None:
        self.async_client = self._new_async_client()
    chat_completion = await self.async_client.chat.completions.create(
        messages=messages,
        model=kwargs.pop('model', self.model_name),
        temperature=kwargs.pop('temperature', self.temperature),
        stream=kwargs.pop('stream', streaming),
        max_tokens=kwargs.pop('max_tokens', self.max_tokens),
        **kwargs,
    )
    if not streaming:
        text = chat_completion.choices[0].message.content
        return LLMOutput(text=text, raw=chat_completion.model_dump())
    return self.agenerate_stream_result(chat_completion)

agenerate_stream_result(stream) async classmethod

Generate the result of the stream.

Source code in agentuniverse/llm/openai_llm.py

Python

@classmethod
async def agenerate_stream_result(cls, stream: AsyncIterator) -> AsyncIterator[LLMOutput]:
    """Generate the result of the stream."""
    async for chunk in stream:
        llm_output = cls.parse_result(chunk)
        if llm_output:
            yield llm_output

as_langchain()

Convert the agentUniverse(aU) openai llm class to the langchain openai llm class.

Source code in agentuniverse/llm/openai_llm.py

Python

def as_langchain(self) -> BaseLanguageModel:
    """Convert the agentUniverse(aU) openai llm class to the langchain openai llm class."""
    return LangchainOpenAI(self)

call(messages, **kwargs)

Run the OpenAI LLM.

Parameters:

Name	Type	Description	Default
messages	list	The messages to send to the LLM.	required
**kwargs	Any	Arbitrary keyword arguments.	{}

Source code in agentuniverse/llm/openai_llm.py

Python

def call(self, messages: list, **kwargs: Any) -> Union[LLMOutput, Iterator[LLMOutput]]:
    """Run the OpenAI LLM.

    Args:
        messages (list): The messages to send to the LLM.
        **kwargs: Arbitrary keyword arguments.
    """
    streaming = kwargs.pop("streaming") if "streaming" in kwargs else self.streaming
    if self.client is None:
        self.client = self._new_client()
    chat_completion = self.client.chat.completions.create(
        messages=messages,
        model=kwargs.pop('model', self.model_name),
        temperature=kwargs.pop('temperature', self.temperature),
        stream=kwargs.pop('stream', streaming),
        max_tokens=kwargs.pop('max_tokens', self.max_tokens),
        **kwargs,
    )
    if not streaming:
        text = chat_completion.choices[0].message.content
        return LLMOutput(text=text, raw=chat_completion.model_dump())
    return self.generate_stream_result(chat_completion)

generate_stream_result(stream) classmethod

Generate the result of the stream.

Source code in agentuniverse/llm/openai_llm.py

Python

@classmethod
def generate_stream_result(cls, stream: Iterator) -> Iterator[LLMOutput]:
    """Generate the result of the stream."""
    for chunk in stream:
        llm_output = cls.parse_result(chunk)
        if llm_output:
            yield llm_output

get_num_tokens(text)

Get the number of tokens present in the text.

Useful for checking if an input will fit in an openai model's context window.

Parameters:

Name	Type	Description	Default
text	str	The string input to tokenize.	required

Returns:

Type	Description
int	The integer number of tokens in the text.

Source code in agentuniverse/llm/openai_llm.py

Python

def get_num_tokens(self, text: str) -> int:
    """Get the number of tokens present in the text.

    Useful for checking if an input will fit in an openai model's context window.

    Args:
        text: The string input to tokenize.

    Returns:
        The integer number of tokens in the text.
    """
    try:
        encoding = tiktoken.encoding_for_model(self.model_name)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

max_context_length()

Max context length.

The total length of input tokens and generated tokens is limited by the openai model's context length.

Source code in agentuniverse/llm/openai_llm.py

Python

def max_context_length(self) -> int:
    """Max context length.

      The total length of input tokens and generated tokens is limited by the openai model's context length.
      """
    return OPENAI_MAX_CONTEXT_LENGTH.get(self.model_name, 4096)

parse_result(chunk) staticmethod

Generate the result of the stream.

Source code in agentuniverse/llm/openai_llm.py

Python

@staticmethod
def parse_result(chunk):
    """Generate the result of the stream."""
    chat_completion = chunk
    if not isinstance(chunk, dict):
        chunk = chunk.dict()
    if len(chunk["choices"]) == 0:
        return
    choice = chunk["choices"][0]
    message = choice.get("delta")
    text = message.get("content")
    if not text:
        return
    return LLMOutput(text=text, raw=chat_completion.model_dump())

set_by_agent_model(**kwargs)

Assign values of parameters to the OpenAILLM model in the agent configuration.

Source code in agentuniverse/llm/openai_llm.py

Python

def set_by_agent_model(self, **kwargs) -> None:
    """ Assign values of parameters to the OpenAILLM model in the agent configuration."""
    super().set_by_agent_model(**kwargs)
    if 'openai_api_key' in kwargs and kwargs['openai_api_key']:
        self.openai_api_key = kwargs['openai_api_key']
    if 'openai_api_base' in kwargs and kwargs['openai_api_base']:
        self.openai_api_base = kwargs['openai_api_base']
    if 'openai_proxy' in kwargs and kwargs['openai_proxy']:
        self.openai_proxy = kwargs['openai_proxy']
    if 'openai_client_args' in kwargs and kwargs['openai_client_args']:
        self.openai_client_args = kwargs['openai_client_args']