test_async_interface() — langchain Function Reference

Architecture documentation for the test_async_interface() function in test_chat_history.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 2

Entity Profile

LangChainCore→ LanguageModelBase→ test_async_interface() — langchain Function Reference

Dependency Diagram

graph TD
  71473c26_d138_2063_09ba_2464c8206318["test_async_interface()"]
  bb44d82a_7f9d_8d19_baa4_96f8844d68a3["test_chat_history.py"]
  71473c26_d138_2063_09ba_2464c8206318 -->|defined in| bb44d82a_7f9d_8d19_baa4_96f8844d68a3
  7a3e1ff0_9d2e_5680_8a98_060fc7fcba81["add_messages()"]
  71473c26_d138_2063_09ba_2464c8206318 -->|calls| 7a3e1ff0_9d2e_5680_8a98_060fc7fcba81
  b857d03c_63dd_64d5_f020_089b6ab4ef34["clear()"]
  71473c26_d138_2063_09ba_2464c8206318 -->|calls| b857d03c_63dd_64d5_f020_089b6ab4ef34
  style 71473c26_d138_2063_09ba_2464c8206318 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/chat_history/test_chat_history.py lines 77–110

async def test_async_interface() -> None:
    """Test async interface for BaseChatMessageHistory."""

    class BulkAddHistory(BaseChatMessageHistory):
        def __init__(self) -> None:
            self.messages = []

        def add_messages(self, message: Sequence[BaseMessage]) -> None:
            """Add a message to the store."""
            self.messages.extend(message)

        def clear(self) -> None:
            """Clear the store."""
            self.messages.clear()

    chat_history = BulkAddHistory()
    await chat_history.aadd_messages(
        [
            HumanMessage(content="Hello"),
            HumanMessage(content="World"),
        ]
    )
    assert await chat_history.aget_messages() == [
        HumanMessage(content="Hello"),
        HumanMessage(content="World"),
    ]
    await chat_history.aadd_messages([HumanMessage(content="!")])
    assert await chat_history.aget_messages() == [
        HumanMessage(content="Hello"),
        HumanMessage(content="World"),
        HumanMessage(content="!"),
    ]
    await chat_history.aclear()
    assert await chat_history.aget_messages() == []