Deepinfra

DeepInfraLLM

Bases: LLM

DeepInfra LLM.

例子

pip install llama-index-llms-deepinfra

from llama_index.llms.deepinfra import DeepInfraLLM

llm = DeepInfraLLM(
    model="mistralai/Mixtral-8x22B-Instruct-v0.1", # 默认模型名称
    api_key = "your-deepinfra-api-key",
    temperature=0.5,
    max_tokens=50,
    additional_kwargs={"top_p": 0.9},
)

response = llm.complete("Hello World!")
print(response)

Source code in llama_index/llms/deepinfra/base.py

class DeepInfraLLM(LLM):
    """DeepInfra LLM.

    例子:
        `pip install llama-index-llms-deepinfra`

        ```python
        from llama_index.llms.deepinfra import DeepInfraLLM

        llm = DeepInfraLLM(
            model="mistralai/Mixtral-8x22B-Instruct-v0.1", # 默认模型名称
            api_key = "your-deepinfra-api-key",
            temperature=0.5,
            max_tokens=50,
            additional_kwargs={"top_p": 0.9},
        )

        response = llm.complete("Hello World!")
        print(response)
        ```
"""

    model: str = Field(
        default=DEFAULT_MODEL_NAME, description="The DeepInfra model to use."
    )

    temperature: float = Field(
        default=DEFAULT_TEMPERATURE,
        description="The temperature to use during generation.",
        gte=0.0,
        lte=1.0,
    )
    max_tokens: Optional[int] = Field(
        description="The maximum number of tokens to generate.",
        gt=0,
    )

    timeout: Optional[float] = Field(
        default=None, description="The timeout to use in seconds.", gte=0
    )
    max_retries: int = Field(
        default=10, description="The maximum number of API retries.", gte=0
    )

    _api_key: Optional[str] = PrivateAttr()

    generate_kwargs: Dict[str, Any] = Field(
        default_factory=dict, description="Additional keyword arguments for generation."
    )

    _client: DeepInfraClient = PrivateAttr()

    def __init__(
        self,
        model: str = DEFAULT_MODEL_NAME,
        additional_kwargs: Optional[Dict[str, Any]] = None,
        temperature: float = DEFAULT_TEMPERATURE,
        max_tokens: Optional[int] = None,
        max_retries: int = 10,
        api_base: str = API_BASE,
        timeout: Optional[float] = None,
        api_key: Optional[str] = None,
        callback_manager: Optional[CallbackManager] = None,
        system_prompt: Optional[str] = None,
        messages_to_prompt: Optional[Callable[[Sequence[ChatMessage]], str]] = None,
        completion_to_prompt: Optional[Callable[[str], str]] = None,
        pydantic_program_mode: PydanticProgramMode = PydanticProgramMode.DEFAULT,
        output_parser: Optional[BaseOutputParser] = None,
    ) -> None:
        additional_kwargs = additional_kwargs or {}
        callback_manager = callback_manager or CallbackManager([])
        self._api_key = get_from_param_or_env("api_key", api_key, ENV_VARIABLE)
        self._client = DeepInfraClient(
            api_key=self._api_key,
            api_base=api_base,
            timeout=timeout,
            max_retries=max_retries,
        )
        super().__init__(
            model=model,
            api_base=api_base,
            api_key=api_key,
            temperature=temperature,
            max_tokens=max_tokens,
            timeout=timeout,
            additional_kwargs=additional_kwargs,
            max_retries=max_retries,
            callback_manager=callback_manager,
            system_prompt=system_prompt,
            messages_to_prompt=messages_to_prompt,
            completion_to_prompt=completion_to_prompt,
            pydantic_program_mode=pydantic_program_mode,
            output_parser=output_parser,
        )

    @classmethod
    def class_name(cls) -> str:
        return "DeepInfra_LLM"

    @property
    def metadata(self) -> LLMMetadata:
        return LLMMetadata(
            num_output=self.max_tokens,
            is_chat_model=self._is_chat_model,
            model=self.model,
        )

    @property
    def _is_chat_model(self) -> bool:
        return True

    # Synchronous Methods
    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs) -> CompletionResponse:
        """为给定的提示生成完成。

Args:
    prompt (str): 要生成完成的输入提示。
    **kwargs: API请求的额外关键字参数。

Returns:
    str: 生成的文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)
        result = self._client.request(INFERENCE_ENDPOINT, payload)
        return CompletionResponse(text=maybe_extract_from_json(result), raw=result)

    @llm_completion_callback()
    def stream_complete(self, prompt: str, **kwargs) -> CompletionResponseGen:
        """为给定的提示生成同步流式完成。

Args:
    prompt (str): 要生成完成的输入提示。
    **kwargs: API请求的额外关键字参数。

产出:
    CompletionResponseGen: 流式文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)

        content = ""
        for response_dict in self._client.request_stream(INFERENCE_ENDPOINT, payload):
            content_delta = maybe_extract_from_json(response_dict)
            content += content_delta
            yield CompletionResponse(
                text=content, delta=content_delta, raw=response_dict
            )

    @llm_chat_callback()
    def chat(self, messages: Sequence[ChatMessage], **kwargs) -> ChatResponse:
        """为给定的消息生成聊天响应。

Args:
    messages（Sequence[ChatMessage]）：一系列聊天消息。
    **kwargs：API请求的额外关键字参数。

Returns:
    ChatResponse：包含一系列消息的聊天响应。
"""
        messages = chat_messages_to_list(messages)
        payload = self._build_payload(messages=messages, **kwargs)
        result = self._client.request(CHAT_API_ENDPOINT, payload)

        return ChatResponse(
            message=ChatMessage(
                role=result["choices"][-1]["message"]["role"],
                content=result["choices"][-1]["message"]["content"],
            ),
            raw=result,
        )

    @llm_chat_callback()
    def stream_chat(
        self, chat_messages: Sequence[ChatMessage], **kwargs
    ) -> ChatResponseGen:
        """为给定的消息生成同步流式聊天响应。

Args:
    messages（Sequence[ChatMessage]）：聊天消息的序列。
    **kwargs：API请求的其他关键字参数。

产出：
    ChatResponseGen：包含消息序列的聊天响应。
"""
        messages = chat_messages_to_list(chat_messages)
        payload = self._build_payload(messages=messages, **kwargs)

        content = ""
        role = MessageRole.ASSISTANT
        for response_dict in self._client.request_stream(CHAT_API_ENDPOINT, payload):
            delta = response_dict["choices"][-1]["delta"]
            """
            Check if the delta contains content.
            """
            if delta.get("content", None):
                content_delta = delta["content"]
                content += delta["content"]
                message = ChatMessage(
                    role=role,
                    content=content,
                )
                yield ChatResponse(
                    message=message, raw=response_dict, delta=content_delta
                )

    # Asynchronous Methods
    @llm_completion_callback()
    async def acomplete(self, prompt: str, **kwargs) -> CompletionResponse:
        """异步生成给定提示的完成。

Args:
    prompt（str）：要生成完成的输入提示。
    **kwargs：API请求的其他关键字参数。

Returns:
    CompletionResponse：生成的文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)

        result = await self._client.arequest(INFERENCE_ENDPOINT, payload)
        return CompletionResponse(text=maybe_extract_from_json(result), raw=result)

    @llm_completion_callback()
    async def astream_complete(
        self, prompt: str, **kwargs
    ) -> CompletionResponseAsyncGen:
        """异步生成给定提示的流式完成。

Args:
    prompt (str): 要生成完成的输入提示。
    **kwargs: API请求的额外关键字参数。

Returns:
    CompletionResponseAsyncGen: 流式文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)

        async def gen():
            content = ""
            async for response_dict in self._client.arequest_stream(
                INFERENCE_ENDPOINT, payload
            ):
                content_delta = maybe_extract_from_json(response_dict)
                content += content_delta
                yield CompletionResponse(
                    text=content, delta=content_delta, raw=response_dict
                )

        return gen()

    @llm_chat_callback()
    async def achat(
        self, chat_messages: Sequence[ChatMessage], **kwargs
    ) -> ChatResponse:
        """异步生成给定消息的聊天响应。

Args:
    messages（Sequence[ChatMessage]）：一系列聊天消息。
    **kwargs：API请求的额外关键字参数。

Returns:
    ChatResponse：包含一系列消息的聊天响应。
"""
        messages = chat_messages_to_list(chat_messages)
        payload = self._build_payload(messages=messages, **kwargs)

        result = await self._client.arequest(CHAT_API_ENDPOINT, payload)
        return ChatResponse(
            message=ChatMessage(
                role=result["choices"][-1]["message"]["role"],
                content=result["choices"][-1]["message"]["content"],
            ),
            raw=result,
        )

    @llm_chat_callback()
    async def astream_chat(
        self, chat_messages: Sequence[ChatMessage], **kwargs
    ) -> ChatResponseAsyncGen:
        """
        Asynchronously generate a streaming chat response for the given messages.

        Args:
            messages (Sequence[ChatMessage]): A sequence of chat messages.
            **kwargs: Additional keyword arguments for the API request.

        Yields:
            ChatResponseAsyncGen: The chat response containing a sequence of messages.
        """
        messages = chat_messages_to_list(chat_messages)
        payload = self._build_payload(messages=messages, **kwargs)

        async def gen():
            content = ""
            role = MessageRole.ASSISTANT
            async for response_dict in self._client.arequest_stream(
                CHAT_API_ENDPOINT, payload
            ):
                delta = response_dict["choices"][-1]["delta"]
                """
                Check if the delta contains content.
                """
                if delta.get("content", None):
                    content_delta = delta["content"]
                    content += delta["content"]
                    message = ChatMessage(
                        role=role,
                        content=content,
                    )
                    yield ChatResponse(
                        message=message, raw=response_dict, delta=content_delta
                    )

        return gen()

    # Utility Methods
    def get_model_endpoint(self) -> str:
        """
        获取DeepInfra模型端点。
        """
        return f"{INFERENCE_ENDPOINT}/{self.model}"

    def _build_payload(self, **kwargs) -> Dict[str, Any]:
        """构建API请求的有效载荷。
温度和max_tokens参数明确覆盖generate_kwargs中的相应值。
任何提供的kwargs都会覆盖所有其他参数，包括温度和max_tokens。

Args:
    prompt（str）：要生成完成内容的输入提示。
    stream（bool）：是否流式传输响应。
    **kwargs：API请求的其他关键字参数。

Returns:
    Dict[str, Any]：API请求有效载荷。
"""
        return {
            **self.generate_kwargs,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
            "model": self.model,
            **kwargs,
        }

complete

complete(prompt: str, **kwargs) -> CompletionResponse

为给定的提示生成完成。

Parameters:

Name	Type	Description	Default
prompt	str	要生成完成的输入提示。	required
**kwargs		API请求的额外关键字参数。	{}

Returns:

Name	Type	Description
str	CompletionResponse	生成的文本完成。

Source code in llama_index/llms/deepinfra/base.py

    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs) -> CompletionResponse:
        """为给定的提示生成完成。

Args:
    prompt (str): 要生成完成的输入提示。
    **kwargs: API请求的额外关键字参数。

Returns:
    str: 生成的文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)
        result = self._client.request(INFERENCE_ENDPOINT, payload)
        return CompletionResponse(text=maybe_extract_from_json(result), raw=result)

stream_complete

stream_complete(
    prompt: str, **kwargs
) -> CompletionResponseGen

为给定的提示生成同步流式完成。

Parameters:

Name	Type	Description	Default
prompt	str	要生成完成的输入提示。	required
**kwargs		API请求的额外关键字参数。	{}

产出

CompletionResponseGen: 流式文本完成。

Source code in llama_index/llms/deepinfra/base.py

    @llm_completion_callback()
    def stream_complete(self, prompt: str, **kwargs) -> CompletionResponseGen:
        """为给定的提示生成同步流式完成。

Args:
    prompt (str): 要生成完成的输入提示。
    **kwargs: API请求的额外关键字参数。

产出:
    CompletionResponseGen: 流式文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)

        content = ""
        for response_dict in self._client.request_stream(INFERENCE_ENDPOINT, payload):
            content_delta = maybe_extract_from_json(response_dict)
            content += content_delta
            yield CompletionResponse(
                text=content, delta=content_delta, raw=response_dict
            )

chat

chat(
    messages: Sequence[ChatMessage], **kwargs
) -> ChatResponse

为给定的消息生成聊天响应。

Returns:

Type	Description
ChatResponse	ChatResponse：包含一系列消息的聊天响应。

Source code in llama_index/llms/deepinfra/base.py

    @llm_chat_callback()
    def chat(self, messages: Sequence[ChatMessage], **kwargs) -> ChatResponse:
        """为给定的消息生成聊天响应。

Args:
    messages（Sequence[ChatMessage]）：一系列聊天消息。
    **kwargs：API请求的额外关键字参数。

Returns:
    ChatResponse：包含一系列消息的聊天响应。
"""
        messages = chat_messages_to_list(messages)
        payload = self._build_payload(messages=messages, **kwargs)
        result = self._client.request(CHAT_API_ENDPOINT, payload)

        return ChatResponse(
            message=ChatMessage(
                role=result["choices"][-1]["message"]["role"],
                content=result["choices"][-1]["message"]["content"],
            ),
            raw=result,
        )

stream_chat

stream_chat(
    chat_messages: Sequence[ChatMessage], **kwargs
) -> ChatResponseGen

为给定的消息生成同步流式聊天响应。

产出： ChatResponseGen：包含消息序列的聊天响应。

Source code in llama_index/llms/deepinfra/base.py

    @llm_chat_callback()
    def stream_chat(
        self, chat_messages: Sequence[ChatMessage], **kwargs
    ) -> ChatResponseGen:
        """为给定的消息生成同步流式聊天响应。

Args:
    messages（Sequence[ChatMessage]）：聊天消息的序列。
    **kwargs：API请求的其他关键字参数。

产出：
    ChatResponseGen：包含消息序列的聊天响应。
"""
        messages = chat_messages_to_list(chat_messages)
        payload = self._build_payload(messages=messages, **kwargs)

        content = ""
        role = MessageRole.ASSISTANT
        for response_dict in self._client.request_stream(CHAT_API_ENDPOINT, payload):
            delta = response_dict["choices"][-1]["delta"]
            """
            Check if the delta contains content.
            """
            if delta.get("content", None):
                content_delta = delta["content"]
                content += delta["content"]
                message = ChatMessage(
                    role=role,
                    content=content,
                )
                yield ChatResponse(
                    message=message, raw=response_dict, delta=content_delta
                )

acomplete async

acomplete(prompt: str, **kwargs) -> CompletionResponse

异步生成给定提示的完成。

Returns:

Type	Description
CompletionResponse	CompletionResponse：生成的文本完成。

Source code in llama_index/llms/deepinfra/base.py

    @llm_completion_callback()
    async def acomplete(self, prompt: str, **kwargs) -> CompletionResponse:
        """异步生成给定提示的完成。

Args:
    prompt（str）：要生成完成的输入提示。
    **kwargs：API请求的其他关键字参数。

Returns:
    CompletionResponse：生成的文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)

        result = await self._client.arequest(INFERENCE_ENDPOINT, payload)
        return CompletionResponse(text=maybe_extract_from_json(result), raw=result)

astream_complete async

astream_complete(
    prompt: str, **kwargs
) -> CompletionResponseAsyncGen

异步生成给定提示的流式完成。

Parameters:

Name	Type	Description	Default
prompt	str	要生成完成的输入提示。	required
**kwargs		API请求的额外关键字参数。	{}

Returns:

Name	Type	Description
CompletionResponseAsyncGen	CompletionResponseAsyncGen	流式文本完成。

Source code in llama_index/llms/deepinfra/base.py

    @llm_completion_callback()
    async def astream_complete(
        self, prompt: str, **kwargs
    ) -> CompletionResponseAsyncGen:
        """异步生成给定提示的流式完成。

Args:
    prompt (str): 要生成完成的输入提示。
    **kwargs: API请求的额外关键字参数。

Returns:
    CompletionResponseAsyncGen: 流式文本完成。
"""
        payload = self._build_payload(prompt=prompt, **kwargs)

        async def gen():
            content = ""
            async for response_dict in self._client.arequest_stream(
                INFERENCE_ENDPOINT, payload
            ):
                content_delta = maybe_extract_from_json(response_dict)
                content += content_delta
                yield CompletionResponse(
                    text=content, delta=content_delta, raw=response_dict
                )

        return gen()

achat async

achat(
    chat_messages: Sequence[ChatMessage], **kwargs
) -> ChatResponse

异步生成给定消息的聊天响应。

Returns:

Type	Description
ChatResponse	ChatResponse：包含一系列消息的聊天响应。

Source code in llama_index/llms/deepinfra/base.py

    @llm_chat_callback()
    async def achat(
        self, chat_messages: Sequence[ChatMessage], **kwargs
    ) -> ChatResponse:
        """异步生成给定消息的聊天响应。

Args:
    messages（Sequence[ChatMessage]）：一系列聊天消息。
    **kwargs：API请求的额外关键字参数。

Returns:
    ChatResponse：包含一系列消息的聊天响应。
"""
        messages = chat_messages_to_list(chat_messages)
        payload = self._build_payload(messages=messages, **kwargs)

        result = await self._client.arequest(CHAT_API_ENDPOINT, payload)
        return ChatResponse(
            message=ChatMessage(
                role=result["choices"][-1]["message"]["role"],
                content=result["choices"][-1]["message"]["content"],
            ),
            raw=result,
        )

astream_chat async

astream_chat(
    chat_messages: Sequence[ChatMessage], **kwargs
) -> ChatResponseAsyncGen

Asynchronously generate a streaming chat response for the given messages.

Parameters:

Name	Type	Description	Default
messages	Sequence[ChatMessage]	A sequence of chat messages.	required
**kwargs		Additional keyword arguments for the API request.	{}

Yields:

Name	Type	Description
ChatResponseAsyncGen	ChatResponseAsyncGen	The chat response containing a sequence of messages.

Source code in llama_index/llms/deepinfra/base.py

@llm_chat_callback()
async def astream_chat(
    self, chat_messages: Sequence[ChatMessage], **kwargs
) -> ChatResponseAsyncGen:
    """
    Asynchronously generate a streaming chat response for the given messages.

    Args:
        messages (Sequence[ChatMessage]): A sequence of chat messages.
        **kwargs: Additional keyword arguments for the API request.

    Yields:
        ChatResponseAsyncGen: The chat response containing a sequence of messages.
    """
    messages = chat_messages_to_list(chat_messages)
    payload = self._build_payload(messages=messages, **kwargs)

    async def gen():
        content = ""
        role = MessageRole.ASSISTANT
        async for response_dict in self._client.arequest_stream(
            CHAT_API_ENDPOINT, payload
        ):
            delta = response_dict["choices"][-1]["delta"]
            """
            Check if the delta contains content.
            """
            if delta.get("content", None):
                content_delta = delta["content"]
                content += delta["content"]
                message = ChatMessage(
                    role=role,
                    content=content,
                )
                yield ChatResponse(
                    message=message, raw=response_dict, delta=content_delta
                )

    return gen()

get_model_endpoint

get_model_endpoint() -> str

获取DeepInfra模型端点。

Source code in llama_index/llms/deepinfra/base.py

def get_model_endpoint(self) -> str:
    """
    获取DeepInfra模型端点。
    """
    return f"{INFERENCE_ENDPOINT}/{self.model}"