Source code for langchain_community.adapters.openai
from __future__ import annotations
import importlib
from typing import (
    Any,
    AsyncIterator,
    Dict,
    Iterable,
    List,
    Mapping,
    Sequence,
    Union,
    overload,
)
from langchain_core.chat_sessions import ChatSession
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    BaseMessageChunk,
    ChatMessage,
    FunctionMessage,
    HumanMessage,
    SystemMessage,
    ToolMessage,
)
from langchain_core.pydantic_v1 import BaseModel
from typing_extensions import Literal
[docs]async def aenumerate(
    iterable: AsyncIterator[Any], start: int = 0
) -> AsyncIterator[tuple[int, Any]]:
    """Async version of enumerate function."""
    i = start
    async for x in iterable:
        yield i, x
        i += 1
[docs]class IndexableBaseModel(BaseModel):
    """Allows a BaseModel to return its fields by string variable indexing"""
    def __getitem__(self, item: str) -> Any:
        return getattr(self, item)
[docs]class Choice(IndexableBaseModel):
    message: dict
[docs]class ChatCompletions(IndexableBaseModel):
    choices: List[Choice]
[docs]class ChoiceChunk(IndexableBaseModel):
    delta: dict
[docs]class ChatCompletionChunk(IndexableBaseModel):
    choices: List[ChoiceChunk]
[docs]def convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
    """Convert a dictionary to a LangChain message.
    Args:
        _dict: The dictionary.
    Returns: