test_input_dict_with_history_key_async() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/runnables/test_history.py lines 197–222

async def test_input_dict_with_history_key_async() -> None:
    runnable = RunnableLambda[Any, str](
        lambda params: (
            "you said: "
            + "\n".join(
                [
                    str(m.content)
                    for m in params["history"]
                    if isinstance(m, HumanMessage)
                ]
                + [params["input"]]
            )
        )
    )
    get_session_history = _get_get_session_history()
    with_history = RunnableWithMessageHistory(
        runnable,
        get_session_history,
        input_messages_key="input",
        history_messages_key="history",
    )
    config: RunnableConfig = {"configurable": {"session_id": "3_async"}}
    output = await with_history.ainvoke({"input": "hello"}, config)
    assert output == "you said: hello"
    output = await with_history.ainvoke({"input": "good bye"}, config)
    assert output == "you said: hello\ngood bye"