test_astream_log_deep_copies() — langchain Function Reference

Architecture documentation for the test_astream_log_deep_copies() function in test_runnable.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ test_astream_log_deep_copies() — langchain Function Reference

Dependency Diagram

graph TD
  97d69db6_8127_b4a0_e46a_04ffc2b73b5b["test_astream_log_deep_copies()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  97d69db6_8127_b4a0_e46a_04ffc2b73b5b -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  style 97d69db6_8127_b4a0_e46a_04ffc2b73b5b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 5416–5461

async def test_astream_log_deep_copies() -> None:
    """Verify that deep copies are used when using jsonpatch in astream log.

    jsonpatch re-uses objects in its API; e.g.,

    import jsonpatch
    obj1 = { "a": 1 }
    value = { "b": 2 }
    obj2 = { "a": 1, "value": value }

    ops = list(jsonpatch.JsonPatch.from_diff(obj1, obj2))
    assert id(ops[0]['value']) == id(value)

    This can create unexpected consequences for downstream code.
    """

    def _get_run_log(run_log_patches: Sequence[RunLogPatch]) -> RunLog:
        """Get run log."""
        run_log = RunLog(state=None)  # type: ignore[arg-type]
        for log_patch in run_log_patches:
            run_log += log_patch
        return run_log

    def add_one(x: int) -> int:
        """Add one."""
        return x + 1

    chain = RunnableLambda(add_one)
    chunks = []
    final_output: RunLogPatch | None = None
    async for chunk in chain.astream_log(1):
        chunks.append(chunk)
        final_output = chunk if final_output is None else final_output + chunk

    run_log = _get_run_log(chunks)
    state = run_log.state.copy()
    # Ignoring type here since we know that the state is a dict
    # so we can delete `id` for testing purposes
    state.pop("id")  # type: ignore[misc]
    assert state == {
        "final_output": 2,
        "logs": {},
        "streamed_output": [2],
        "name": "add_one",
        "type": "chain",
    }