RunLogPatch Class — langchain Architecture

Architecture documentation for the RunLogPatch class in log_stream.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  7aee39c0_a5c9_c6ea_388c_4792f8844146["RunLogPatch"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146["RunLogPatch"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146 -->|extends| 7aee39c0_a5c9_c6ea_388c_4792f8844146
  bd3b753a_7a64_0c77_8d7a_afcce4658003["log_stream.py"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146 -->|defined in| bd3b753a_7a64_0c77_8d7a_afcce4658003
  9812ec43_2e3a_60f3_00bf_e523da5c69c5["__init__()"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146 -->|method| 9812ec43_2e3a_60f3_00bf_e523da5c69c5
  0c27761a_14a2_d901_a67d_8ec1e4619066["__add__()"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146 -->|method| 0c27761a_14a2_d901_a67d_8ec1e4619066
  6b646557_7807_d523_55fe_52da66eb42d6["__repr__()"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146 -->|method| 6b646557_7807_d523_55fe_52da66eb42d6
  92fc9fc1_436f_0d70_6f9c_f76218fb0638["__eq__()"]
  7aee39c0_a5c9_c6ea_388c_4792f8844146 -->|method| 92fc9fc1_436f_0d70_6f9c_f76218fb0638

Relationship Graph

Source Code

libs/core/langchain_core/tracers/log_stream.py lines 115–165

class RunLogPatch:
    """Patch to the run log."""

    ops: list[dict[str, Any]]
    """List of `JSONPatch` operations, which describe how to create the run state
    from an empty dict.

    This is the minimal representation of the log, designed to be serialized as JSON and
    sent over the wire to reconstruct the log on the other side. Reconstruction of the
    state can be done with any JSONPatch-compliant library, see https://jsonpatch.com
    for more information.
    """

    def __init__(self, *ops: dict[str, Any]) -> None:
        """Create a RunLogPatch.

        Args:
            *ops: The operations to apply to the state.
        """
        self.ops = list(ops)

    def __add__(self, other: RunLogPatch | Any) -> RunLog:
        """Combine two `RunLogPatch` instances.

        Args:
            other: The other `RunLogPatch` to combine with.

        Raises:
            TypeError: If the other object is not a `RunLogPatch`.

        Returns:
            A new `RunLog` representing the combination of the two.
        """
        if type(other) is RunLogPatch:
            ops = self.ops + other.ops
            state = jsonpatch.apply_patch(None, copy.deepcopy(ops))
            return RunLog(*ops, state=state)

        msg = f"unsupported operand type(s) for +: '{type(self)}' and '{type(other)}'"
        raise TypeError(msg)

    @override
    def __repr__(self) -> str:
        # 1:-1 to get rid of the [] around the list
        return f"RunLogPatch({pformat(self.ops)[1:-1]})"

    @override
    def __eq__(self, other: object) -> bool:
        return isinstance(other, RunLogPatch) and self.ops == other.ops

    __hash__ = None  # type: ignore[assignment]