Source code for langchain_community.chat_models.vertexai
"""Wrapper around Google VertexAI chat-based models."""
from __future__ import annotations
import base64
import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Union, cast
from langchain_core.callbacks import (
    AsyncCallbackManagerForLLMRun,
    CallbackManagerForLLMRun,
)
from langchain_core.language_models.chat_models import (
    BaseChatModel,
    generate_from_stream,
)
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    HumanMessage,
    SystemMessage,
)
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
from langchain_core.pydantic_v1 import root_validator
from langchain_community.llms.vertexai import (
    _VertexAICommon,
    is_codey_model,
    is_gemini_model,
)
from langchain_community.utilities.vertexai import (
    load_image_from_gcs,
    raise_vertex_import_error,
)
if TYPE_CHECKING:
    from vertexai.language_models import (
        ChatMessage,
        ChatSession,
        CodeChatSession,
        InputOutputTextPair,
    )
    from vertexai.preview.generative_models import Content
logger = logging.getLogger(__name__)
@dataclass
class _ChatHistory:
    """Represents a context and a history of messages."""
    history: List["ChatMessage"] = field(default_factory=list)
    context: Optional[str] = None
def _parse_chat_history(history: List[BaseMessage]) -> _ChatHistory:
    """Parse a sequence of messages into history.
    Args: