dspy.BaseLM¶

@with_callbacks
def __call__(self, prompt=None, *, messages=None, **kwargs):
    """Return legacy outputs, or one lm15.Response for an explicit Request."""
    from dspy.clients.execution import execute, finalize, prepare

    call = prepare(self, prompt, messages, kwargs)
    if not call.managed:
        warn_legacy_lm()
    warn_openai_messages(self, messages)
    return finalize(self, call, execute(self, call))

@with_callbacks
async def acall(self, prompt=None, *, messages=None, **kwargs):
    """Async equivalent of __call__, with the same execution ownership."""
    import asyncio

    from dspy.clients.execution import aexecute, finalize, prepare

    # Canonical media snapshots may read local files. Context variables
    # propagate to the worker; callbacks/finalization stay on the caller.
    call = await asyncio.to_thread(prepare, self, prompt, messages, kwargs, asynchronous=True)
    if not call.managed:
        warn_legacy_lm()
    warn_openai_messages(self, messages)
    return finalize(self, call, await aexecute(self, call))

async def aforward(self, prompt=None, messages=None, **kwargs):
    """Asynchronously return an OpenAI-shaped provider response."""
    raise NotImplementedError("Subclasses must implement this method.")

def copy(self, **kwargs):
    """Return a copy of the language model with updated parameters.

    The default implementation makes a shallow runtime copy. Provider
    clients, sessions, and local model handles are preserved by reference.
    DSPy-owned mutable state is isolated for `history`, the `callbacks`
    list, and the `kwargs` dict. Other attributes are shared by reference.
    Subclasses with additional mutable DSPy-owned state should override this
    method.

    Args:
        **kwargs: Attribute or request-parameter updates to apply to the
            copy. For example, `lm.copy(rollout_id=1, temperature=1.0)`
            returns an LM whose requests use a different rollout ID at
            non-zero temperature to bypass cache collisions.

    Returns:
        A copied LM instance.
    """

    new_instance = copy_module.copy(self)
    new_instance.history = []
    new_instance.callbacks = list(getattr(self, "callbacks", []) or [])
    new_instance.kwargs = dict(getattr(self, "kwargs", {}) or {})

    for key, value in kwargs.items():
        if hasattr(new_instance, key):
            setattr(new_instance, key, value)
        if (key in new_instance.kwargs) or (not hasattr(self, key)):
            if value is None:
                new_instance.kwargs.pop(key, None)
            else:
                new_instance.kwargs[key] = value
    if hasattr(new_instance, "_warned_zero_temp_rollout"):
        new_instance._warned_zero_temp_rollout = False

    return new_instance

def dump_state(self) -> dict[str, Any]:
    """Return a sanitized reconstruction state for this LM.

    Subclasses whose state is captured by `BaseLM.__init__` can use this
    default. Subclasses with extra persistent state should override both
    `dump_state` and `load_state`.

    Returns:
        A dictionary that can be passed to `BaseLM.load_state`. The state
        excludes API keys.
    """
    filtered_kwargs = {key: value for key, value in self.kwargs.items() if key not in ("api_key", LM_CLASS_STATE_KEY)}
    return {
        LM_CLASS_STATE_KEY: f"{type(self).__module__}.{type(self).__qualname__}",
        "model": self.model,
        "model_type": self.model_type,
        "cache": self.cache,
        "num_retries": getattr(self, "num_retries", 3),
        **filtered_kwargs,
    }

def forward(self, prompt=None, messages=None, **kwargs):
    """Return an OpenAI-shaped provider response for a legacy LM call."""
    raise NotImplementedError("Subclasses must implement this method.")

def inspect_history(self, n: int = 1, file: "TextIO | None" = None) -> None:
    pretty_print_history(self.history, n, file=file)

@classmethod
def load_state(cls, state: dict[str, Any], *, allow_custom_lm_class: bool = False) -> "BaseLM":
    """Reconstruct an LM from `dump_state` output.

    Legacy states without a class marker load as `dspy.LM`. Custom LM
    classes must be importable by their module-qualified class path and are
    only loaded when `allow_custom_lm_class=True`.

    Args:
        state: Serialized LM state produced by `dump_state`.
        allow_custom_lm_class: If True, allow importing and loading custom
            `BaseLM` subclasses recorded in `state`. Enable only for trusted
            state.

    Returns:
        The reconstructed LM instance.

    Raises:
        ValueError: If `state` references a custom LM class and
            `allow_custom_lm_class` is False.
        ImportError: If the serialized LM class cannot be imported.
        TypeError: If the serialized class is not a `BaseLM` subclass.
    """
    state = dict(state)
    class_path = state.pop(LM_CLASS_STATE_KEY, None)

    if cls is BaseLM:
        if class_path == "dspy.clients.typesafe.TypeSafe":
            from dspy.clients.typesafe import TypeSafe

            return TypeSafe.load_state(state)
        if class_path is None:
            # Legacy saved programs did not record the concrete LM class.
            from dspy.clients.lm import LM

            return LM(**state)

        if class_path != _BUILTIN_LM_CLASS_PATH and not allow_custom_lm_class:
            raise ValueError(
                f"Refusing to import custom serialized LM class `{class_path}`. "
                "Pass allow_unsafe_lm_state=True when loading trusted files to enable custom LM classes."
            )

        lm_cls = _import_lm_class(class_path)
        if not issubclass(lm_cls, BaseLM):
            raise TypeError(f"Serialized LM class `{class_path}` must be a subclass of dspy.BaseLM.")
        if "allow_custom_lm_class" in inspect.signature(lm_cls.load_state).parameters:
            return lm_cls.load_state(state, allow_custom_lm_class=allow_custom_lm_class)
        return lm_cls.load_state(state)

    return cls(**state)

def update_history(self, entry):
    record_history(self, entry)