test_input_messages_async() — langchain Function Reference

Architecture documentation for the test_input_messages_async() function in test_history.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_input_messages_async() — langchain Function Reference

Dependency Diagram

graph TD
  99897380_ac0a_163f_7d7e_37d8089e1b87["test_input_messages_async()"]
  0af0dfde_fe33_f3e5_b5e6_a3d18260dc67["test_history.py"]
  99897380_ac0a_163f_7d7e_37d8089e1b87 -->|defined in| 0af0dfde_fe33_f3e5_b5e6_a3d18260dc67
  e035a3b1_d1a1_c458_c801_af1fc0272f8b["_get_get_session_history()"]
  99897380_ac0a_163f_7d7e_37d8089e1b87 -->|calls| e035a3b1_d1a1_c458_c801_af1fc0272f8b
  style 99897380_ac0a_163f_7d7e_37d8089e1b87 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_history.py lines 85–116

async def test_input_messages_async() -> None:
    runnable = RunnableLambda[Any, str](
        lambda messages: (
            "you said: "
            + "\n".join(str(m.content) for m in messages if isinstance(m, HumanMessage))
        )
    )
    store: dict[str, InMemoryChatMessageHistory] = {}
    get_session_history = _get_get_session_history(store=store)
    with_history = RunnableWithMessageHistory(runnable, get_session_history)
    config = {"session_id": "1_async"}
    output = await with_history.ainvoke([HumanMessage(content="hello")], config)  # type: ignore[arg-type]
    assert output == "you said: hello"
    output = await with_history.ainvoke([HumanMessage(content="good bye")], config)  # type: ignore[arg-type]
    assert output == "you said: hello\ngood bye"
    output = [
        c
        async for c in with_history.astream([HumanMessage(content="hi again")], config)  # type: ignore[arg-type]
    ]
    assert output == ["you said: hello\ngood bye\nhi again"]
    assert store == {
        "1_async": InMemoryChatMessageHistory(
            messages=[
                HumanMessage(content="hello"),
                AIMessage(content="you said: hello"),
                HumanMessage(content="good bye"),
                AIMessage(content="you said: hello\ngood bye"),
                HumanMessage(content="hi again"),
                AIMessage(content="you said: hello\ngood bye\nhi again"),
            ]
        )
    }