Source code for langchain_nvidia_ai_endpoints.chat_models
"""Chat Model Components Derived from ChatModel/NVIDIA"""
from __future__ import annotations
import base64
import logging
import os
import urllib.parse
from typing import (
    Any,
    AsyncIterator,
    Dict,
    Iterator,
    List,
    Mapping,
    Optional,
    Sequence,
    Union,
)
import requests
from langchain_core.callbacks.manager import (
    AsyncCallbackManagerForLLMRun,
    CallbackManagerForLLMRun,
)
from langchain_core.language_models.chat_models import SimpleChatModel
from langchain_core.messages import BaseMessage, ChatMessage, ChatMessageChunk
from langchain_core.outputs import ChatGenerationChunk
from langchain_nvidia_ai_endpoints import _common as nvidia_ai_endpoints
logger = logging.getLogger(__name__)
def _is_openai_parts_format(part: dict) -> bool:
    return "type" in part
def _is_url(s: str) -> bool:
    try:
        result = urllib.parse.urlparse(s)
        return all([result.scheme, result.netloc])
    except Exception as e:
        logger.debug(f"Unable to parse URL: {e}")
        return False
def _is_b64(s: str) -> bool:
    return s.startswith("data:image")
def _url_to_b64_string(image_source: str) -> str:
    b64_template = "data:image/png;base64,{b64_string}"
    try:
        if _is_url(image_source):
            response = requests.get(image_source)
            response.raise_for_status()
            encoded = base64.b64encode(response.content).decode("utf-8")
            return b64_template.format(b64_string=encoded)
        elif _is_b64(image_source):