_copy_run() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2 called by 1

Entity Profile

LangChainCore→ Runnables→ _copy_run() — langchain Function Reference

Dependency Diagram

graph TD
  2edec481_9714_8233_fe69_21ac5be56870["_copy_run()"]
  1c72b396_ff94_2618_42c3_843ec828f717["FakeTracer"]
  2edec481_9714_8233_fe69_21ac5be56870 -->|defined in| 1c72b396_ff94_2618_42c3_843ec828f717
  07bf4724_e010_ed55_b883_d074a448c2b8["_persist_run()"]
  07bf4724_e010_ed55_b883_d074a448c2b8 -->|calls| 2edec481_9714_8233_fe69_21ac5be56870
  d5b7b6c0_f6b6_2489_1752_29e8d4da7b57["_replace_uuid()"]
  2edec481_9714_8233_fe69_21ac5be56870 -->|calls| d5b7b6c0_f6b6_2489_1752_29e8d4da7b57
  07f9dcf9_f8c2_f4d9_fb48_0b6a09055b88["_replace_message_id()"]
  2edec481_9714_8233_fe69_21ac5be56870 -->|calls| 07f9dcf9_f8c2_f4d9_fb48_0b6a09055b88
  style 2edec481_9714_8233_fe69_21ac5be56870 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 139–162

    def _copy_run(self, run: Run) -> Run:
        if run.dotted_order:
            levels = run.dotted_order.split(".")
            processed_levels = []
            for level in levels:
                timestamp, run_id = level.split("Z")
                new_run_id = self._replace_uuid(UUID(run_id))
                processed_level = f"{timestamp}Z{new_run_id}"
                processed_levels.append(processed_level)
            new_dotted_order = ".".join(processed_levels)
        else:
            new_dotted_order = None
        update_dict = {
            "id": self._replace_uuid(run.id),
            "parent_run_id": (
                self.uuids_map[run.parent_run_id] if run.parent_run_id else None
            ),
            "child_runs": [self._copy_run(child) for child in run.child_runs],
            "trace_id": self._replace_uuid(run.trace_id) if run.trace_id else None,
            "dotted_order": new_dotted_order,
            "inputs": self._replace_message_id(run.inputs),
            "outputs": self._replace_message_id(run.outputs),
        }
        return pydantic_copy(run, update=update_dict)