Skip to content

dspy.LM

dspy.LM(model: str, model_type: Literal['chat', 'text'] = 'chat', temperature: float = 0.0, max_tokens: int = 1000, cache: bool = True, cache_in_memory: bool = True, callbacks: Optional[List[BaseCallback]] = None, num_retries: int = 8, provider=None, finetuning_model: Optional[str] = None, launch_kwargs: Optional[dict[str, Any]] = None, train_kwargs: Optional[dict[str, Any]] = None, **kwargs)

Bases: BaseLM

A language model supporting chat or text completion requests for use with DSPy modules.

Create a new language model instance for use with DSPy modules and programs.

Parameters:

Name Type Description Default
model str

The model to use. This should be a string of the form "llm_provider/llm_name" supported by LiteLLM. For example, "openai/gpt-4o".

 required
model_type Literal['chat', 'text']

The type of the model, either "chat" or "text".

 'chat'
temperature float

The sampling temperature to use when generating responses.

 0.0
max_tokens int

The maximum number of tokens to generate per response.

 1000
cache bool

Whether to cache the model responses for reuse to improve performance and reduce costs.

 True
cache_in_memory bool

To enable additional caching with LRU in memory.

 True
callbacks Optional[List[BaseCallback]]

A list of callback functions to run before and after each request.

 None
num_retries int

The number of times to retry a request if it fails transiently due to network error, rate limiting, etc. Requests are retried with exponential backoff.

 8
provider

The provider to use. If not specified, the provider will be inferred from the model.

 None
finetuning_model Optional[str]

The model to finetune. In some providers, the models available for finetuning is different from the models available for inference.

 None
Source code in dspy/clients/lm.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __init__(
    self,
    model: str,
    model_type: Literal["chat", "text"] = "chat",
    temperature: float = 0.0,
    max_tokens: int = 1000,
    cache: bool = True,
    cache_in_memory: bool = True,
    callbacks: Optional[List[BaseCallback]] = None,
    num_retries: int = 8,
    provider=None,
    finetuning_model: Optional[str] = None,
    launch_kwargs: Optional[dict[str, Any]] = None,
    train_kwargs: Optional[dict[str, Any]] = None,
    **kwargs,
):
    """
    Create a new language model instance for use with DSPy modules and programs.

    Args:
        model: The model to use. This should be a string of the form ``"llm_provider/llm_name"``
               supported by LiteLLM. For example, ``"openai/gpt-4o"``.
        model_type: The type of the model, either ``"chat"`` or ``"text"``.
        temperature: The sampling temperature to use when generating responses.
        max_tokens: The maximum number of tokens to generate per response.
        cache: Whether to cache the model responses for reuse to improve performance
               and reduce costs.
        cache_in_memory: To enable additional caching with LRU in memory.
        callbacks: A list of callback functions to run before and after each request.
        num_retries: The number of times to retry a request if it fails transiently due to
                     network error, rate limiting, etc. Requests are retried with exponential
                     backoff.
        provider: The provider to use. If not specified, the provider will be inferred from the model.
        finetuning_model: The model to finetune. In some providers, the models available for finetuning is different
            from the models available for inference.
    """
    # Remember to update LM.copy() if you modify the constructor!
    self.model = model
    self.model_type = model_type
    self.cache = cache
    self.cache_in_memory = cache_in_memory
    self.provider = provider or self.infer_provider()
    self.callbacks = callbacks or []
    self.history = []
    self.callbacks = callbacks or []
    self.num_retries = num_retries
    self.finetuning_model = finetuning_model
    self.launch_kwargs = launch_kwargs or {}
    self.train_kwargs = train_kwargs or {}

    # Handle model-specific configuration for different model families
    model_family = model.split("/")[-1].lower() if "/" in model else model.lower()

    # Match pattern: o[1,3] at the start, optionally followed by -mini and anything else
    model_pattern = re.match(r"^o([13])(?:-mini)?", model_family)

    if model_pattern:
        # Handle OpenAI reasoning models (o1, o3)
        assert (
            max_tokens >= 20_000 and temperature == 1.0
        ), "OpenAI's reasoning models require passing temperature=1.0 and max_tokens >= 20_000 to `dspy.LM(...)`"
        self.kwargs = dict(temperature=temperature, max_completion_tokens=max_tokens, **kwargs)
    else:
        self.kwargs = dict(temperature=temperature, max_tokens=max_tokens, **kwargs)

Functions

__call__(prompt=None, messages=None, **kwargs)

Source code in dspy/clients/base_lm.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@with_callbacks
def __call__(self, prompt=None, messages=None, **kwargs):
    response = self.forward(prompt=prompt, messages=messages, **kwargs)

    if kwargs.get("logprobs"):
        outputs = [
            {
                "text": c.message.content if hasattr(c, "message") else c["text"],
                "logprobs": c.logprobs if hasattr(c, "logprobs") else c["logprobs"],
            }
            for c in response.choices
        ]
    else:
        outputs = [c.message.content if hasattr(c, "message") else c["text"] for c in response.choices]

    if settings.disable_history:
        return outputs

    # Logging, with removed api key & where `cost` is None on cache hit.
    kwargs = {k: v for k, v in kwargs.items() if not k.startswith("api_")}
    entry = {
        "prompt": prompt,
        "messages": messages,
        "kwargs": kwargs,
        "response": response,
        "outputs": outputs,
        "usage": dict(response.usage),
        "cost": getattr(response, "_hidden_params", {}).get("response_cost"),
        "timestamp": datetime.datetime.now().isoformat(),
        "uuid": str(uuid.uuid4()),
        "model": self.model,
        "response_model": response.model,
        "model_type": self.model_type,
    }
    self.history.append(entry)
    self.update_global_history(entry)

    return outputs

copy(**kwargs)

Returns a copy of the language model with possibly updated parameters.

Source code in dspy/clients/base_lm.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def copy(self, **kwargs):
    """Returns a copy of the language model with possibly updated parameters."""

    import copy

    new_instance = copy.deepcopy(self)
    new_instance.history = []

    for key, value in kwargs.items():
        if hasattr(self, key):
            setattr(new_instance, key, value)
        if (key in self.kwargs) or (not hasattr(self, key)):
            new_instance.kwargs[key] = value

    return new_instance

dump_state()

Source code in dspy/clients/lm.py 
190
191
192
193
194
195
196
197
198
199
200
201
def dump_state(self):
    state_keys = [
        "model",
        "model_type",
        "cache",
        "cache_in_memory",
        "num_retries",
        "finetuning_model",
        "launch_kwargs",
        "train_kwargs",
    ]
    return {key: getattr(self, key) for key in state_keys} | self.kwargs

finetune(train_data: List[Dict[str, Any]], train_data_format: Optional[TrainDataFormat], train_kwargs: Optional[Dict[str, Any]] = None) -> TrainingJob

Source code in dspy/clients/lm.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def finetune(
    self,
    train_data: List[Dict[str, Any]],
    train_data_format: Optional[TrainDataFormat],
    train_kwargs: Optional[Dict[str, Any]] = None,
) -> TrainingJob:
    from dspy import settings as settings

    err = "Fine-tuning is an experimental feature."
    err += " Set `dspy.settings.experimental` to `True` to use it."
    assert settings.experimental, err

    err = f"Provider {self.provider} does not support fine-tuning."
    assert self.provider.finetunable, err

    def thread_function_wrapper():
        return self._run_finetune_job(job)

    thread = threading.Thread(target=thread_function_wrapper)
    train_kwargs = train_kwargs or self.train_kwargs
    model_to_finetune = self.finetuning_model or self.model
    job = self.provider.TrainingJob(
        thread=thread,
        model=model_to_finetune,
        train_data=train_data,
        train_data_format=train_data_format,
        train_kwargs=train_kwargs,
    )
    thread.start()

    return job

forward(prompt=None, messages=None, **kwargs)

Source code in dspy/clients/lm.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@with_callbacks
def forward(self, prompt=None, messages=None, **kwargs):
    # Build the request.
    cache = kwargs.pop("cache", self.cache)
    # disable cache will also disable in memory cache
    cache_in_memory = cache and kwargs.pop("cache_in_memory", self.cache_in_memory)
    messages = messages or [{"role": "user", "content": prompt}]
    kwargs = {**self.kwargs, **kwargs}

    # Make the request and handle LRU & disk caching.
    if cache_in_memory:
        completion = cached_litellm_completion if self.model_type == "chat" else cached_litellm_text_completion

        results = completion(
            request=dict(model=self.model, messages=messages, **kwargs),
            num_retries=self.num_retries,
        )
    else:
        completion = litellm_completion if self.model_type == "chat" else litellm_text_completion

        results = completion(
            request=dict(model=self.model, messages=messages, **kwargs),
            num_retries=self.num_retries,
            # only leverage LiteLLM cache in this case
            cache={"no-cache": not cache, "no-store": not cache},
        )

    if not getattr(results, "cache_hit", False) and dspy.settings.usage_tracker and hasattr(results, "usage"):
        settings.usage_tracker.add_usage(self.model, dict(results.usage))
    return results

infer_provider() -> Provider

Source code in dspy/clients/lm.py 
185
186
187
188
def infer_provider(self) -> Provider:
    if OpenAIProvider.is_provider_model(self.model):
        return OpenAIProvider()
    return Provider()

inspect_history(n: int = 1)

Source code in dspy/clients/base_lm.py 
114
115
def inspect_history(self, n: int = 1):
    _inspect_history(self.history, n)

kill(launch_kwargs: Optional[Dict[str, Any]] = None)

Source code in dspy/clients/lm.py 
133
134
def kill(self, launch_kwargs: Optional[Dict[str, Any]] = None):
    self.provider.kill(self, launch_kwargs)

launch(launch_kwargs: Optional[Dict[str, Any]] = None)

Source code in dspy/clients/lm.py 
130
131
def launch(self, launch_kwargs: Optional[Dict[str, Any]] = None):
    self.provider.launch(self, launch_kwargs)

update_global_history(entry)

Source code in dspy/clients/base_lm.py 
117
118
119
120
121
122
123
124
def update_global_history(self, entry):
    if settings.disable_history:
        return

    if len(GLOBAL_HISTORY) >= MAX_HISTORY_SIZE:
        GLOBAL_HISTORY.pop(0)

    GLOBAL_HISTORY.append(entry)