put() — langchain Function Reference

Architecture documentation for the put() function in memory_assert.py from the langchain codebase.

Function python LangChainCore ApiManagement

Entity Profile

LangChainCore→ ApiManagement→ put() — langchain Function Reference

Dependency Diagram

graph TD
  0832e1df_4f2c_8c66_7089_7e38536724bf["put()"]
  0fad622c_f86b_840d_42b5_97c5a1c330f1["MemorySaverAssertImmutable"]
  0832e1df_4f2c_8c66_7089_7e38536724bf -->|defined in| 0fad622c_f86b_840d_42b5_97c5a1c330f1
  style 0832e1df_4f2c_8c66_7089_7e38536724bf fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/memory_assert.py lines 40–63

    def put(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        if self.put_sleep:
            time.sleep(self.put_sleep)
        # assert checkpoint hasn't been modified since last written
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        if saved := super().get(config):
            assert (
                self.serde.loads_typed(
                    self.storage_for_copies[thread_id][checkpoint_ns][saved["id"]]
                )
                == saved
            )
        self.storage_for_copies[thread_id][checkpoint_ns][checkpoint["id"]] = (
            self.serde.dumps_typed(copy_checkpoint(checkpoint))
        )
        # call super to write checkpoint
        return super().put(config, checkpoint, metadata, new_versions)