RunLog Class — langchain Architecture

Source Code

libs/core/langchain_core/tracers/log_stream.py lines 168–226

class RunLog(RunLogPatch):
    """Run log."""

    state: RunState
    """Current state of the log, obtained from applying all ops in sequence."""

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

        Args:
            *ops: The operations to apply to the state.
            state: The initial state of the run log.
        """
        super().__init__(*ops)
        self.state = state

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

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

        Raises:
            TypeError: If the other object is not a `RunLog` or `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(self.state, other.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:
        return f"RunLog({pformat(self.state)})"

    @override
    def __eq__(self, other: object) -> bool:
        """Check if two `RunLog`s are equal.

        Args:
            other: The other `RunLog` to compare to.

        Returns:
            `True` if the `RunLog`s are equal, `False` otherwise.
        """
        # First compare that the state is the same
        if not isinstance(other, RunLog):
            return False
        if self.state != other.state:
            return False
        # Then compare that the ops are the same
        return super().__eq__(other)

    __hash__ = None