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, callbacks: Optional[List[BaseCallback]] = None, num_retries: int = 8, provider=None, finetuning_model: Optional[str] = None, launch_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
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 
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
def __init__(
    self,
    model: str,
    model_type: Literal["chat", "text"] = "chat",
    temperature: float = 0.0,
    max_tokens: int = 1000,
    cache: 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,
    **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.
        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.provider = provider or self.infer_provider()
    self.callbacks = callbacks or []
    self.kwargs = dict(temperature=temperature, max_tokens=max_tokens, **kwargs)
    self.history = []
    self.callbacks = callbacks or []
    self.num_retries = num_retries
    self.finetuning_model = finetuning_model
    self.launch_kwargs = launch_kwargs

    # TODO(bug): Arbitrary model strings could include the substring "o1-".
    # We should find a more robust way to check for the "o1-" family models.
    if "o1-" in model:
        assert (
            max_tokens >= 5000 and temperature == 1.0
        ), "OpenAI's o1-* models require passing temperature=1.0 and max_tokens >= 5000 to `dspy.LM(...)`"

Functions

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

Source code in dspy/clients/lm.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 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
129
130
131
132
@with_callbacks
def __call__(self, prompt=None, messages=None, **kwargs):
    # Build the request.
    cache = kwargs.pop("cache", self.cache)
    messages = messages or [{"role": "user", "content": prompt}]
    kwargs = {**self.kwargs, **kwargs}

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

    response = completion(
        request=dict(model=self.model, messages=messages, **kwargs),
        num_retries=self.num_retries,
    )
    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"]]

    # 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 = dict(prompt=prompt, messages=messages, kwargs=kwargs, response=response)
    entry = dict(**entry, outputs=outputs, usage=dict(response["usage"]))
    entry = dict(**entry, cost=response.get("_hidden_params", {}).get("response_cost"))
    entry = dict(
        **entry,
        timestamp=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/lm.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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

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

Source code in dspy/clients/lm.py 
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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def finetune(
    self,
    train_data: List[Dict[str, Any]],
    train_kwargs: Optional[Dict[str, Any]] = None,
    data_format: Optional[DataFormat] = 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

    # Perform data validation before starting the thread to fail early
    train_kwargs = train_kwargs or {}
    if not data_format:
        adapter = self.infer_adapter()
        data_format = infer_data_format(adapter)
    validate_data_format(data=train_data, data_format=data_format)

    # TODO(PR): We can quickly add caching, but doing so requires
    # adding functions that just call other functions as we had in the last
    # iteration, unless people have other ideas.
    def thread_function_wrapper():
        return self._run_finetune_job(job)

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

    return job

infer_adapter() -> Adapter

Source code in dspy/clients/lm.py 
207
208
209
210
211
212
213
214
215
216
217
def infer_adapter(self) -> Adapter:
    import dspy

    if dspy.settings.adapter:
        return dspy.settings.adapter

    model_type_to_adapter = {
        "chat": dspy.ChatAdapter(),
    }
    model_type = self.model_type
    return model_type_to_adapter[model_type]

infer_provider() -> Provider

Source code in dspy/clients/lm.py 
200
201
202
203
204
205
def infer_provider(self) -> Provider:
    if OpenAIProvider.is_provider_model(self.model):
        return OpenAIProvider()
    # TODO(PR): Keeping this function here will require us to import all
    # providers in this file. Is this okay?
    return Provider()

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

Source code in dspy/clients/lm.py 
138
139
140
def kill(self, launch_kwargs: Optional[Dict[str, Any]] = None):
    launch_kwargs = launch_kwargs or self.launch_kwargs
    self.provider.kill(self.model, launch_kwargs)

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

Source code in dspy/clients/lm.py 
134
135
136
def launch(self, launch_kwargs: Optional[Dict[str, Any]] = None):
    launch_kwargs = launch_kwargs or self.launch_kwargs
    self.provider.launch(self.model, launch_kwargs)