base.py — langchain Source File

Source Code

"""Chain for chatting with a vector database."""

from __future__ import annotations

import inspect
import warnings
from abc import abstractmethod
from collections.abc import Callable
from pathlib import Path
from typing import Any

from langchain_core._api import deprecated
from langchain_core.callbacks import (
    AsyncCallbackManagerForChainRun,
    CallbackManagerForChainRun,
    Callbacks,
)
from langchain_core.documents import Document
from langchain_core.language_models import BaseLanguageModel
from langchain_core.messages import BaseMessage
from langchain_core.prompts import BasePromptTemplate
from langchain_core.retrievers import BaseRetriever
from langchain_core.runnables import RunnableConfig
from langchain_core.vectorstores import VectorStore
from pydantic import BaseModel, ConfigDict, Field, model_validator
from typing_extensions import override

from langchain_classic.chains.base import Chain
from langchain_classic.chains.combine_documents.base import BaseCombineDocumentsChain
from langchain_classic.chains.combine_documents.stuff import StuffDocumentsChain
from langchain_classic.chains.conversational_retrieval.prompts import (
    CONDENSE_QUESTION_PROMPT,
)
from langchain_classic.chains.llm import LLMChain
from langchain_classic.chains.question_answering import load_qa_chain

# Depending on the memory type and configuration, the chat history format may differ.
# This needs to be consolidated.
CHAT_TURN_TYPE = tuple[str, str] | BaseMessage


_ROLE_MAP = {"human": "Human: ", "ai": "Assistant: "}


def _get_chat_history(chat_history: list[CHAT_TURN_TYPE]) -> str:
    buffer = ""
    for dialogue_turn in chat_history:
        if isinstance(dialogue_turn, BaseMessage):
            if len(dialogue_turn.content) > 0:
                role_prefix = _ROLE_MAP.get(
                    dialogue_turn.type,
                    f"{dialogue_turn.type}: ",
                )
                buffer += f"\n{role_prefix}{dialogue_turn.content}"
        elif isinstance(dialogue_turn, tuple):
            human = "Human: " + dialogue_turn[0]
            ai = "Assistant: " + dialogue_turn[1]
            buffer += f"\n{human}\n{ai}"
        else:
            msg = (  # type: ignore[unreachable]
// ... (519 more lines)