test_history.py — langchain Source File

Source Code

import re
from collections.abc import Callable, Sequence
from typing import Any

import pytest
from pydantic import BaseModel, RootModel
from typing_extensions import override

from langchain_core.callbacks import (
    CallbackManagerForLLMRun,
)
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.runnables import Runnable
from langchain_core.runnables.base import RunnableBinding, RunnableLambda
from langchain_core.runnables.config import RunnableConfig
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables.utils import ConfigurableFieldSpec, Input, Output
from langchain_core.tracers import Run
from langchain_core.tracers.root_listeners import (
    AsyncListener,
    AsyncRootListenersTracer,
    RootListenersTracer,
)
from tests.unit_tests.pydantic_utils import _schema


def test_interfaces() -> None:
    history = InMemoryChatMessageHistory()
    history.add_message(SystemMessage(content="system"))
    history.add_message(HumanMessage(content="human 1"))
    history.add_message(AIMessage(content="ai"))
    assert str(history) == "System: system\nHuman: human 1\nAI: ai"


def _get_get_session_history(
    *,
    store: dict[str, InMemoryChatMessageHistory] | None = None,
) -> Callable[..., InMemoryChatMessageHistory]:
    chat_history_store = store if store is not None else {}

    def get_session_history(
        session_id: str, **_kwargs: Any
    ) -> InMemoryChatMessageHistory:
        if session_id not in chat_history_store:
            chat_history_store[session_id] = InMemoryChatMessageHistory()
        return chat_history_store[session_id]

    return get_session_history


def test_input_messages() -> None:
    runnable = RunnableLambda[Any, str](
        lambda messages: (
            "you said: "
            + "\n".join(str(m.content) for m in messages if isinstance(m, HumanMessage))
        )
    )
// ... (847 more lines)