dspy.LM

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

基类: BaseLM

支持聊天或文本完成请求的语言模型，用于 DSPy 模块。

创建一个新的语言模型实例，用于 DSPy 模块和程序。

参数

名称	类型	描述	默认值
model	str	要使用的模型。这应该是一个 LiteLLM 支持的、形式为 "llm_provider/llm_name" 的字符串。例如，"openai/gpt-4o"。	必需
model_type	Literal['chat', 'text']	模型类型，可以是 "chat" 或 "text"。	'chat'
temperature	float	生成响应时使用的采样温度。	0.0
max_tokens	int	每次响应生成的最大 token 数。	4000
cache	bool	是否缓存模型响应以便重用，从而提高性能并降低成本。	True
cache_in_memory	已弃用	启用额外的基于内存 LRU 的缓存。	True
callbacks	Optional[List[BaseCallback]]	在每次请求之前和之后运行的回调函数列表。	None
num_retries	int	如果请求因网络错误、速率限制等暂时性原因失败，重试的次数。请求将采用指数退避策略重试。	3
provider		要使用的提供程序。如果未指定，提供程序将从模型中推断。	None
finetuning_model	Optional[str]	用于微调的模型。在某些提供程序中，可用于微调的模型与可用于推理的模型不同。	None

源代码位于 dspy/clients/lm.py

def __init__(
    self,
    model: str,
    model_type: Literal["chat", "text"] = "chat",
    temperature: float = 0.0,
    max_tokens: int = 4000,
    cache: bool = True,
    cache_in_memory: bool = True,
    callbacks: Optional[List[BaseCallback]] = None,
    num_retries: int = 3,
    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 (deprecated): 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.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,4] at the start, optionally followed by -mini and anything else
    model_pattern = re.match(r"^o([134])(?:-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)

函数

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

源代码位于 dspy/clients/base_lm.py

@with_callbacks
def __call__(self, prompt=None, messages=None, **kwargs):
    response = self.forward(prompt=prompt, messages=messages, **kwargs)
    outputs = self._process_lm_response(response, prompt, messages, **kwargs)

    return outputs

acall(prompt=None, messages=None, **kwargs) async

源代码位于 dspy/clients/base_lm.py

@with_callbacks
async def acall(self, prompt=None, messages=None, **kwargs):
    response = await self.aforward(prompt=prompt, messages=messages, **kwargs)
    outputs = self._process_lm_response(response, prompt, messages, **kwargs)
    return outputs

aforward(prompt=None, messages=None, **kwargs) async

源代码位于 dspy/clients/lm.py

async def aforward(self, prompt=None, messages=None, **kwargs):
    # Build the request.
    cache = kwargs.pop("cache", self.cache)
    enable_memory_cache = kwargs.pop("cache_in_memory", self.cache_in_memory)

    messages = messages or [{"role": "user", "content": prompt}]
    kwargs = {**self.kwargs, **kwargs}

    completion = alitellm_completion if self.model_type == "chat" else alitellm_text_completion
    completion, litellm_cache_args = self._get_cached_completion_fn(completion, cache, enable_memory_cache)

    results = await completion(
        request=dict(model=self.model, messages=messages, **kwargs),
        num_retries=self.num_retries,
        cache=litellm_cache_args,
    )

    if any(c.finish_reason == "length" for c in results["choices"]):
        logger.warning(
            f"LM response was truncated due to exceeding max_tokens={self.kwargs['max_tokens']}. "
            "You can inspect the latest LM interactions with `dspy.inspect_history()`. "
            "To avoid truncation, consider passing a larger max_tokens when setting up dspy.LM. "
            f"You may also consider increasing the temperature (currently {self.kwargs['temperature']}) "
            " if the reason for truncation is repetition."
        )

    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

copy(**kwargs)

返回语言模型的副本，可能包含更新的参数。

源代码位于 dspy/clients/base_lm.py

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()

源代码位于 dspy/clients/lm.py

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

源代码位于 dspy/clients/lm.py

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 = 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)

源代码位于 dspy/clients/lm.py

def forward(self, prompt=None, messages=None, **kwargs):
    # Build the request.
    cache = kwargs.pop("cache", self.cache)
    enable_memory_cache = kwargs.pop("cache_in_memory", self.cache_in_memory)

    messages = messages or [{"role": "user", "content": prompt}]
    kwargs = {**self.kwargs, **kwargs}

    completion = litellm_completion if self.model_type == "chat" else litellm_text_completion
    completion, litellm_cache_args = self._get_cached_completion_fn(completion, cache, enable_memory_cache)

    results = completion(
        request=dict(model=self.model, messages=messages, **kwargs),
        num_retries=self.num_retries,
        cache=litellm_cache_args,
    )

    if any(c.finish_reason == "length" for c in results["choices"]):
        logger.warning(
            f"LM response was truncated due to exceeding max_tokens={self.kwargs['max_tokens']}. "
            "You can inspect the latest LM interactions with `dspy.inspect_history()`. "
            "To avoid truncation, consider passing a larger max_tokens when setting up dspy.LM. "
            f"You may also consider increasing the temperature (currently {self.kwargs['temperature']}) "
            " if the reason for truncation is repetition."
        )

    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

源代码位于 dspy/clients/lm.py

def infer_provider(self) -> Provider:
    if OpenAIProvider.is_provider_model(self.model):
        return OpenAIProvider()
    return Provider()

inspect_history(n: int = 1)

源代码位于 dspy/clients/base_lm.py

def inspect_history(self, n: int = 1):
    _inspect_history(self.history, n)

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

源代码位于 dspy/clients/lm.py

def kill(self, launch_kwargs: Optional[Dict[str, Any]] = None):
    self.provider.kill(self, launch_kwargs)

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

源代码位于 dspy/clients/lm.py

def launch(self, launch_kwargs: Optional[Dict[str, Any]] = None):
    self.provider.launch(self, launch_kwargs)

reinforce(train_kwargs) -> ReinforceJob

源代码位于 dspy/clients/lm.py

def reinforce(self, train_kwargs) -> ReinforceJob:
    # TODO(GRPO Team): Should we return an initialized job here?
    from dspy import settings as settings

    err = f"Provider {self.provider} does not implement the reinforcement learning interface."
    assert self.provider.reinforceable, err

    job = self.provider.ReinforceJob(lm=self, train_kwargs=train_kwargs)
    job.initialize()
    return job

update_global_history(entry)

源代码位于 dspy/clients/base_lm.py

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)