test_invoke() — langchain Function Reference

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

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_invoke() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/langchain/tests/unit_tests/retrievers/test_time_weighted_retriever.py lines 144–159

def test_invoke(
    time_weighted_retriever: TimeWeightedVectorStoreRetriever,
) -> None:
    query = "Test query"
    relevant_documents = time_weighted_retriever.invoke(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