test_ainvoke() — langchain Function Reference

Architecture documentation for the test_ainvoke() function in test_time_weighted_retriever.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_ainvoke() — langchain Function Reference

Dependency Diagram

graph TD
  fa195f96_394b_f23f_4114_5fe9a34f0ab7["test_ainvoke()"]
  3ccece1f_f844_33e7_bdea_1996eb777e35["test_time_weighted_retriever.py"]
  fa195f96_394b_f23f_4114_5fe9a34f0ab7 -->|defined in| 3ccece1f_f844_33e7_bdea_1996eb777e35
  6fe805a4_6ac0_60a5_d86a_5523f5336e7e["_get_example_memories()"]
  fa195f96_394b_f23f_4114_5fe9a34f0ab7 -->|calls| 6fe805a4_6ac0_60a5_d86a_5523f5336e7e
  style fa195f96_394b_f23f_4114_5fe9a34f0ab7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/tests/unit_tests/retrievers/test_time_weighted_retriever.py lines 162–177

async def test_ainvoke(
    time_weighted_retriever: TimeWeightedVectorStoreRetriever,
) -> None:
    query = "Test query"
    relevant_documents = await time_weighted_retriever.ainvoke(query)
    want = [(doc, 0.5) for doc in _get_example_memories()]
    assert isinstance(relevant_documents, list)
    assert len(relevant_documents) == len(want)
    now = datetime.now()
    for doc in relevant_documents:
        # assert that the last_accessed_at is close to now.
        assert now - timedelta(hours=1) < doc.metadata["last_accessed_at"] <= now

    # assert that the last_accessed_at in the memory stream is updated.
    for d in time_weighted_retriever.memory_stream:
        assert now - timedelta(hours=1) < d.metadata["last_accessed_at"] <= now