test_inmemory_upsert() — langchain Function Reference

Architecture documentation for the test_inmemory_upsert() function in test_in_memory.py from the langchain codebase.

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ test_inmemory_upsert() — langchain Function Reference

Dependency Diagram

graph TD
  6d9d0ddf_e5ee_e4a7_d49a_49236b7b059d["test_inmemory_upsert()"]
  a974d690_d5fa_ba65_671d_ce8278eefe7d["test_in_memory.py"]
  6d9d0ddf_e5ee_e4a7_d49a_49236b7b059d -->|defined in| a974d690_d5fa_ba65_671d_ce8278eefe7d
  style 6d9d0ddf_e5ee_e4a7_d49a_49236b7b059d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/vectorstores/test_in_memory.py lines 154–179

async def test_inmemory_upsert() -> None:
    """Test upsert documents."""
    embedding = DeterministicFakeEmbedding(size=2)
    store = InMemoryVectorStore(embedding=embedding)

    # Check sync version
    store.add_documents([Document(page_content="foo", id="1")])
    assert sorted(store.store.keys()) == ["1"]

    # Check async version
    await store.aadd_documents([Document(page_content="bar", id="2")])
    assert sorted(store.store.keys()) == ["1", "2"]

    # update existing document
    await store.aadd_documents(
        [Document(page_content="baz", id="2", metadata={"metadata": "value"})]
    )
    item = store.store["2"]

    baz_vector = embedding.embed_query("baz")
    assert item == {
        "id": "2",
        "text": "baz",
        "vector": baz_vector,
        "metadata": {"metadata": "value"},
    }