dspy.Predict
dspy.Predict
dspy.Predict는 언어 모델(LM)을 이용해 입력을 출력으로 매핑하는 가장 기본적인 DSPy 모듈입니다. 시그니처 하나만 주면 프롬프트를 직접 짜지 않아도 입력→출력 변환을 수행합니다.
출처: 문서
본문
dspy.Predict(signature: str | type[Signature], callbacks: list[BaseCallback] | None = None, **config)
- Bases:
Module(callbacks=None),Parameter
언어 모델을 이용해 입력을 출력으로 매핑하는 기본적인 DSPy 모듈입니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
signature |
str | type[Signature] |
작업(task)을 설명하는 입력/출력 시그니처. | required |
callbacks |
list[BaseCallback] | None |
계측(instrumentation)을 위한 콜백의 선택적 리스트. | None |
**config |
기본 언어 모델로 전달되는 키워드 인자. 모듈을 호출할 때 config 딕셔너리를 넘기면 단일 호출에 한해 이 값들을 덮어쓸 수 있습니다. 예: |
{} |
predict = dspy.Predict("q -> a", rollout_id=1, temperature=1.0)
predict(q="What is 1 + 52?", config={"rollout_id": 2, "temperature": 1.0})
소스 코드는 dspy/predict/predict.py에 있습니다.
def __init__(self, signature: str | type[Signature], callbacks: list[BaseCallback] | None = None, **config):
super().__init__(callbacks=callbacks)
self.stage = random.randbytes(8).hex()
self.signature = ensure_signature(signature)
self.fields = {}
self.config = config
self.reset()
Methods
__call__(*args, **kwargs)
소스 코드는 dspy/predict/predict.py에 있습니다.
def __call__(self, *args, **kwargs):
if args:
raise ValueError(self._get_positional_args_error_message())
return super().__call__(**kwargs)
acall(*args, **kwargs) (async)
소스 코드는 dspy/predict/predict.py에 있습니다.
async def acall(self, *args, **kwargs):
if args:
raise ValueError(self._get_positional_args_error_message())
return await super().acall(**kwargs)
aforward(**kwargs) (async)
소스 코드는 dspy/predict/predict.py에 있습니다.
async def aforward(self, **kwargs):
lm, config, signature, demos, kwargs = self._forward_preprocess(**kwargs)
adapter = resolve_adapter(lm, settings.adapter or ChatAdapter(), signature, self.fields, self.signature)
if self._should_stream():
with settings.context(caller_predict=self):
completions = await adapter.acall(lm, lm_kwargs=config, signature=signature, demos=demos, inputs=kwargs)
else:
with settings.context(send_stream=None):
completions = await adapter.acall(lm, lm_kwargs=config, signature=signature, demos=demos, inputs=kwargs)
return self._forward_postprocess(completions, signature, **kwargs)
batch(examples, num_threads=None, max_errors=None, return_failed_examples=False, provide_traceback=None, disable_progress_bar=False, timeout=120, straggler_limit=3)
Parallel 모듈을 이용해 dspy.Example 인스턴스 리스트를 병렬로 처리합니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
examples |
list[Example] |
처리할 dspy.Example 인스턴스 리스트. |
required |
num_threads |
int | None |
병렬 처리에 사용할 스레드의 수. | None |
max_errors |
int | None |
실행을 멈추기 전에 허용할 최대 에러 수. None이면 dspy.settings.max_errors를 따릅니다. |
None |
return_failed_examples |
bool |
실패한 example과 예외를 반환할지 여부. | False |
provide_traceback |
bool | None |
에러 로그에 traceback 정보를 포함할지 여부. | None |
disable_progress_bar |
bool |
진행률 표시줄 표시 여부. | False |
timeout |
int |
지연(straggler) 작업을 재제출하기 전 대기 시간(초). 0이면 비활성화. | 120 |
straggler_limit |
int |
남은 작업이 이 수 이하일 때만 지연 작업을 검사. | 3 |
Returns: 결과 리스트, 그리고 선택적으로 실패한 example과 예외.
소스 코드는 dspy/primitives/module.py에 있습니다.
def batch(
self,
examples: list[Example],
num_threads: int | None = None,
max_errors: int | None = None,
return_failed_examples: bool = False,
provide_traceback: bool | None = None,
disable_progress_bar: bool = False,
timeout: int = 120,
straggler_limit: int = 3,
) -> list[Example] | tuple[list[Example], list[Example], list[Exception]]:
"""
Processes a list of dspy.Example instances in parallel using the Parallel module.
Args:
examples: List of dspy.Example instances to process.
num_threads: Number of threads to use for parallel processing.
max_errors: Maximum number of errors allowed before stopping execution.
If ``None``, inherits from ``dspy.settings.max_errors``.
return_failed_examples: Whether to return failed examples and exceptions.
provide_traceback: Whether to include traceback information in error logs.
disable_progress_bar: Whether to display the progress bar.
timeout: Seconds before a straggler task is resubmitted. Set to 0 to disable.
straggler_limit: Only check for stragglers when this many or fewer tasks remain.
Returns:
List of results, and optionally failed examples and exceptions.
"""
# Create a list of execution pairs (self, example)
exec_pairs = [(self, example.inputs()) for example in examples]
# Create an instance of Parallel
parallel_executor = Parallel(
num_threads=num_threads,
max_errors=max_errors,
return_failed_examples=return_failed_examples,
provide_traceback=provide_traceback,
disable_progress_bar=disable_progress_bar,
timeout=timeout,
straggler_limit=straggler_limit,
)
# Execute the forward method of Parallel
if return_failed_examples:
results, failed_examples, exceptions = parallel_executor.forward(exec_pairs)
return results, failed_examples, exceptions
else:
results = parallel_executor.forward(exec_pairs)
return results
deepcopy()
모듈을 깊은 복사(deep copy)합니다.
기본 파이썬 deepcopy를 약간 조정한 것으로, self.parameters()만 깊은 복사를 하고 나머지 속성은 얕은 복사(shallow copy)를 수행합니다.
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def deepcopy(self):
"""Deep copy the module.
This is a tweak to the default python deepcopy that only deep copies `self.parameters()`, and for other
attributes, we just do the shallow copy.
"""
try:
# If the instance itself is copyable, we can just deep copy it.
# Otherwise we will have to create a new instance and copy over the attributes one by one.
return copy.deepcopy(self)
except Exception:
pass
# Create an empty instance.
new_instance = self.__class__.__new__(self.__class__)
# Set attribuetes of the copied instance.
for attr, value in self.__dict__.items():
if isinstance(value, BaseModule):
setattr(new_instance, attr, value.deepcopy())
else:
try:
# Try to deep copy the attribute
setattr(new_instance, attr, copy.deepcopy(value))
except Exception:
logging.warning(
f"Failed to deep copy attribute '{attr}' of {self.__class__.__name__}, "
"falling back to shallow copy or reference copy."
)
try:
# Fallback to shallow copy if deep copy fails
setattr(new_instance, attr, copy.copy(value))
except Exception:
# If even the shallow copy fails, we just copy over the reference.
setattr(new_instance, attr, value)
return new_instance
dump_state(json_mode=True)
소스 코드는 dspy/predict/predict.py에 있습니다.
def dump_state(self, json_mode=True):
state_keys = ["traces", "train"]
state = {k: getattr(self, k) for k in state_keys}
state["demos"] = []
for demo in self.demos:
demo = demo.copy()
for field in demo:
# FIXME: Saving BaseModels as strings in examples doesn't matter because you never re-access as an object
demo[field] = serialize_object(demo[field])
if isinstance(demo, dict) or not json_mode:
state["demos"].append(demo)
else:
state["demos"].append(demo.toDict())
state["signature"] = self.signature.dump_state()
if self.fields:
DecisionState(self.signature, self.fields)
state["fields"] = copy.deepcopy(self.fields)
state["lm"] = self.lm.dump_state() if self.lm else None
return state
forward(**kwargs)
소스 코드는 dspy/predict/predict.py에 있습니다.
def forward(self, **kwargs):
lm, config, signature, demos, kwargs = self._forward_preprocess(**kwargs)
adapter = resolve_adapter(lm, settings.adapter or ChatAdapter(), signature, self.fields, self.signature)
if self._should_stream():
with settings.context(caller_predict=self):
completions = adapter(lm, lm_kwargs=config, signature=signature, demos=demos, inputs=kwargs)
else:
with settings.context(send_stream=None):
completions = adapter(lm, lm_kwargs=config, signature=signature, demos=demos, inputs=kwargs)
return self._forward_postprocess(completions, signature, **kwargs)
get_config()
소스 코드는 dspy/predict/predict.py에 있습니다.
def get_config(self):
return self.config
get_lm()
이 모듈의 predictor들이 사용하는 언어 모델을 반환합니다.
모든 모듈의 말단(leaf)이 같은 LM을 사용한다면 그 언어 모델을 반환합니다. 서로 다른 LM이 여러 개 사용되고 있으면 에러를 발생시킵니다.
Returns: 이 모듈의 predictor들이 사용하는 언어 모델 인스턴스.
Raises: ValueError — 이 모듈의 predictor들이 서로 다른 언어 모델을 사용하는 경우.
소스 코드는 dspy/primitives/module.py에 있습니다.
def get_lm(self):
"""Get the language model used by this module's predictors.
Returns the language model if every module leaf uses the same LM. Raises an error if multiple
different LMs are in use.
Returns:
The language model instance used by this module's predictors.
Raises:
ValueError: If multiple different language models are being
used by the predictors in this module.
"""
all_used_lms = [param.lm for _, param in self.named_parameters() if isinstance(param, Module)]
if len(set(all_used_lms)) == 1:
return all_used_lms[0]
raise ValueError("Multiple LMs are being used in the module. There's no unique LM to return.")
inspect_history(n=1, file=None) -> None
이 모듈의 LM 호출 기록을 표시합니다.
이 모듈이 만든 가장 최근 언어 모델 호출을 보기 좋게 출력합니다. 디버깅과 모듈 동작을 이해하는 데 유용합니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n |
int |
표시할 최근 기록 항목 수. 기본값은 1. | 1 |
file |
TextIO | None |
출력을 작성할 선택적 file-like 객체. 제공되면 ANSI 색상 코드가 자동으로 꺼집니다. 기본값은 None(stdout으로 출력). |
None |
소스 코드는 dspy/primitives/module.py에 있습니다.
def inspect_history(self, n: int = 1, file: "TextIO | None" = None) -> None:
"""Display the LM call history for this module.
Prints a formatted view of the most recent language model calls
made by this module, useful for debugging and understanding
the module's behavior.
Args:
n: The number of recent history entries to display.
Defaults to 1.
file: An optional file-like object to write output to. When
provided, ANSI color codes are automatically disabled.
Defaults to `None` (prints to stdout).
"""
pretty_print_history(self.history, n, file=file)
load(path, allow_pickle=False, allow_unsafe_lm_state=False)
저장된 모듈을 불러옵니다. 기존 프로그램의 상태만이 아니라 프로그램 전체를 불러오고 싶다면 dspy.load도 확인해 보세요.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path |
str |
저장된 상태 파일의 경로. .json 또는 .pkl 파일이어야 합니다. |
required |
allow_pickle |
bool |
True면 .pkl 파일을 허용합니다. 이는 임의 코드를 실행할 수 있어 위험하며, 파일 출처를 확신하고 신뢰할 수 있는 환경에서만 사용해야 합니다. |
False |
allow_unsafe_lm_state |
bool |
True면 불러온 상태에서 안전하지 않은 LM 엔드포인트 키(예: api_base, base_url, model_list)를 유지하고 커스텀 LM 클래스의 import를 허용합니다. 신뢰할 수 있는 파일에 대해서만 활성화하세요. |
False |
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def load(self, path, allow_pickle=False, allow_unsafe_lm_state=False):
"""Load the saved module. You may also want to check out dspy.load, if you want to
load an entire program, not just the state for an existing program.
Args:
path (str): Path to the saved state file, which should be a .json or a .pkl file
allow_pickle (bool): If True, allow loading .pkl files, which can run arbitrary code.
This is dangerous and should only be used if you are sure about the source of the file and in a trusted environment.
allow_unsafe_lm_state (bool): If True, preserves unsafe LM endpoint keys (e.g.,
`api_base`, `base_url`, and `model_list`) from loaded state and allows importing custom LM classes.
Enable only for trusted files.
"""
path = Path(path)
if path.suffix == ".json":
with open(path, "rb") as f:
state = orjson.loads(f.read())
elif path.suffix == ".pkl":
if not allow_pickle:
raise ValueError("Loading .pkl files can run arbitrary code, which may be dangerous. Prefer "
"saving with .json files if possible. Set `allow_pickle=True` "
"if you are sure about the source of the file and in a trusted environment.")
with open(path, "rb") as f:
state = cloudpickle.load(f)
else:
raise ValueError(f"`path` must end with `.json` or `.pkl`, but received: {path}")
dependency_versions = get_dependency_versions()
saved_dependency_versions = state["metadata"]["dependency_versions"]
for key, saved_version in saved_dependency_versions.items():
if dependency_versions[key] != saved_version:
logger.warning(
f"There is a mismatch of {key} version between saved model and current environment. "
f"You saved with `{key}=={saved_version}`, but now you have "
f"`{key}=={dependency_versions[key]}`. This might cause errors or performance downgrade "
"on the loaded model, please consider loading the model in the same environment as the "
"saving environment."
)
self.load_state(state, allow_unsafe_lm_state=allow_unsafe_lm_state)
load_state(state, *, allow_unsafe_lm_state=False) -> Predict
Predict 객체의 저장된 상태를 불러옵니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state |
dict |
Predict 객체의 저장된 상태. |
required |
allow_unsafe_lm_state |
bool |
True면 직렬화된 LM 상태에서 api_base, base_url, model_list를 유지하고 커스텀 LM 클래스의 import를 허용합니다. 신뢰할 수 있는 파일을 불러올 때만 활성화하세요. |
False |
Returns: 메서드 체이닝을 위한 자기 자신(self).
소스 코드는 dspy/predict/predict.py에 있습니다.
def load_state(self, state: dict, *, allow_unsafe_lm_state: bool = False) -> "Predict":
"""Load the saved state of a `Predict` object.
Args:
state: The saved state of a `Predict` object.
allow_unsafe_lm_state: If True, preserves `api_base`, `base_url`, and `model_list` from
serialized LM state and allows importing custom LM classes. Enable only when loading trusted files.
Returns:
Self to allow method chaining.
"""
restored_signature = self.signature.load_state(state["signature"])
restored_fields = copy.deepcopy(state.get("fields", {}))
DecisionState(restored_signature, restored_fields)
excluded_keys = ["signature", "extended_signature", "lm", "fields"]
for name, value in state.items():
# `excluded_keys` are fields that go through special handling.
if name not in excluded_keys:
setattr(self, name, value)
self.signature = restored_signature
self.fields = restored_fields
sanitized_lm_state = _sanitize_lm_state(state["lm"], allow_unsafe_lm_state) if state["lm"] else None
self.lm = (
BaseLM.load_state(sanitized_lm_state, allow_custom_lm_class=allow_unsafe_lm_state)
if sanitized_lm_state
else None
)
if "extended_signature" in state: # legacy, up to and including 2.5, for CoT.
raise NotImplementedError("Loading extended_signature is no longer supported in DSPy 2.6+")
return self
map_named_predictors(func)
이 모듈의 모든 named predictor에 함수를 적용합니다.
모듈의 모든 Predict 인스턴스를 순회하며 주어진 함수를 적용하고, 각 predictor를 함수의 반환값으로 교체합니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
func |
Predict 인스턴스를 받아 새 Predict 인스턴스(또는 호환 객체)를 반환하는 콜러블. |
required |
Returns: 메서드 체이닝을 위한 Module(자기 자신).
>>> import dspy
>>> class MyProgram(dspy.Module):
... def __init__(self):
... super().__init__()
... self.qa = dspy.Predict("question -> answer")
...
>>> program = MyProgram()
>>> program.map_named_predictors(lambda p: p)
소스 코드는 dspy/primitives/module.py에 있습니다.
def map_named_predictors(self, func):
"""Apply a function to all named predictors in this module.
This method iterates through all Predict instances in the module
and applies the given function to each, replacing the original
predictor with the function's return value.
Args:
func: A callable that takes a Predict instance and returns
a new Predict instance (or compatible object).
Returns:
Module: Returns self for method chaining.
Examples:
>>> import dspy
>>> class MyProgram(dspy.Module):
... def __init__(self):
... super().__init__()
... self.qa = dspy.Predict("question -> answer")
...
>>> program = MyProgram()
>>> program.map_named_predictors(lambda p: p)
"""
for name, predictor in self.named_predictors():
set_attribute_by_name(self, name, func(predictor))
return self
named_parameters()
named_parameters는 PyTorch와 달리 (비재귀적인) 파라미터 리스트도 처리합니다.
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def named_parameters(self):
"""
Unlike PyTorch, handles (non-recursive) lists of parameters too.
"""
import dspy
from dspy.predict.parameter import Parameter
visited = set()
named_parameters = []
def add_parameter(param_name, param_value):
if isinstance(param_value, Parameter):
if id(param_value) not in visited:
visited.add(id(param_value))
named_parameters.append((param_name, param_value))
elif isinstance(param_value, dspy.Module):
# When a sub-module is pre-compiled, keep it frozen.
if not getattr(param_value, "_compiled", False):
for sub_name, param in param_value.named_parameters():
add_parameter(f"{param_name}.{sub_name}", param)
if isinstance(self, Parameter):
add_parameter("self", self)
for name, value in self.__dict__.items():
if isinstance(value, Parameter):
add_parameter(name, value)
elif isinstance(value, dspy.Module):
# When a sub-module is pre-compiled, keep it frozen.
if not getattr(value, "_compiled", False):
for sub_name, param in value.named_parameters():
add_parameter(f"{name}.{sub_name}", param)
elif isinstance(value, (list, tuple)):
for idx, item in enumerate(value):
add_parameter(f"{name}[{idx}]", item)
elif isinstance(value, dict):
for key, item in value.items():
add_parameter(f"{name}['{key}']", item)
return named_parameters
named_predictors()
이 모듈의 모든 named Predict 모듈을 반환합니다.
모든 파라미터를 순회하며 dspy.Predict 인스턴스인 것들을 이름과 함께 반환합니다.
Returns: list[tuple[str, Predict]] — name은 속성 경로, predictor는 Predict 인스턴스인 (name, predictor) 튜플 리스트.
>>> import dspy
>>> class MyProgram(dspy.Module):
... def __init__(self):
... super().__init__()
... self.qa = dspy.Predict("question -> answer")
... self.summarize = dspy.Predict("text -> summary")
...
>>> program = MyProgram()
>>> for name, p in program.named_predictors():
... print(name)
qa
summarize
소스 코드는 dspy/primitives/module.py에 있습니다.
def named_predictors(self):
"""Return all named Predict modules in this module.
Iterates through all parameters and returns those that are instances
of ``dspy.Predict``, along with their names.
Returns:
list[tuple[str, Predict]]: A list of (name, predictor) tuples
where name is the attribute path and predictor is the
Predict instance.
Examples:
>>> import dspy
>>> class MyProgram(dspy.Module):
... def __init__(self):
... super().__init__()
... self.qa = dspy.Predict("question -> answer")
... self.summarize = dspy.Predict("text -> summary")
...
>>> program = MyProgram()
>>> for name, p in program.named_predictors():
... print(name)
qa
summarize
"""
from dspy.predict.predict import Predict
return [(name, param) for name, param in self.named_parameters() if isinstance(param, Predict)]
named_sub_modules(type_=None, skip_compiled=False)
모듈의 모든 하위 모듈과 그 이름을 찾습니다.
self.children[4]['key'].sub_module이 하위 모듈이라면 이름은 children[4]['key'].sub_module이 됩니다. 하지만 하위 모듈이 서로 다른 경로로 접근 가능한 경우 그 중 하나의 경로만 반환됩니다.
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def named_sub_modules(self, type_=None, skip_compiled=False) -> Generator[tuple[str, "BaseModule"], None, None]:
"""Find all sub-modules in the module, as well as their names.
Say `self.children[4]['key'].sub_module` is a sub-module. Then the name will be
`children[4]['key'].sub_module`. But if the sub-module is accessible at different
paths, only one of the paths will be returned.
"""
if type_ is None:
type_ = BaseModule
queue = deque([("self", self)])
seen = {id(self)}
def add_to_queue(name, item):
if id(item) not in seen:
seen.add(id(item))
queue.append((name, item))
while queue:
name, item = queue.popleft()
if isinstance(item, type_):
yield name, item
if isinstance(item, BaseModule):
if skip_compiled and getattr(item, "_compiled", False):
continue
for sub_name, sub_item in item.__dict__.items():
add_to_queue(f"{name}.{sub_name}", sub_item)
elif isinstance(item, (list, tuple)):
for i, sub_item in enumerate(item):
add_to_queue(f"{name}[{i}]", sub_item)
elif isinstance(item, dict):
for key, sub_item in item.items():
add_to_queue(f"{name}[{key}]", sub_item)
parameters()
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def parameters(self):
return [param for _, param in self.named_parameters()]
predictors()
이 모듈의 모든 Predict 모듈을 반환합니다.
Returns: list[Predict] — 이 모듈의 모든 Predict 인스턴스 리스트.
>>> import dspy
>>> class MyProgram(dspy.Module):
... def __init__(self):
... super().__init__()
... self.qa = dspy.Predict("question -> answer")
...
>>> program = MyProgram()
>>> len(program.predictors())
소스 코드는 dspy/primitives/module.py에 있습니다.
def predictors(self):
"""Return all Predict modules in this module.
Returns:
list[Predict]: A list of all Predict instances in this module.
Examples:
>>> import dspy
>>> class MyProgram(dspy.Module):
... def __init__(self):
... super().__init__()
... self.qa = dspy.Predict("question -> answer")
...
>>> program = MyProgram()
>>> len(program.predictors())
"""
return [param for _, param in self.named_predictors()]
reset()
소스 코드는 dspy/predict/predict.py에 있습니다.
def reset(self):
self.lm = None
self.traces = []
self.train = []
self.demos = []
reset_copy()
모듈을 깊은 복사하고 모든 파라미터를 초기화합니다.
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def reset_copy(self):
"""Deep copy the module and reset all parameters."""
new_instance = self.deepcopy()
for param in new_instance.parameters():
param.reset()
return new_instance
save(path, save_program=False, modules_to_serialize=None)
모듈을 저장합니다.
모듈을 디렉토리 또는 파일에 저장합니다. 두 가지 모드가 있습니다:
save_program=False: 파일 확장자에 따라 모듈의 상태만 json 또는 pickle 파일로 저장.save_program=True: cloudpickle로 모듈 전체를 디렉토리에 저장. 모델의 상태와 구조(architecture)가 모두 포함됩니다.
save_program=True이고 modules_to_serialize가 주어지면 cloudpickle의 register_pickle_by_value로 해당 모듈들을 직렬화에 등록합니다. 이렇게 하면 cloudpickle이 참조(reference)가 아니라 값(by value)으로 모듈을 직렬화해서, 저장된 프로그램과 함께 모듈이 완전히 보존됩니다. 프로그램과 함께 직렬화해야 하는 커스텀 모듈이 있을 때 유용합니다. None이면 직렬화에 등록되는 모듈이 없습니다.
또한 의존성 버전도 저장하므로, 불러올 때 중요한 의존성 또는 DSPy 버전의 불일치가 있는지 확인할 수 있습니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path |
str |
save_program=False일 때는 .json 또는 .pkl 파일 경로, save_program=True일 때는 디렉토리 경로. |
required |
save_program |
bool |
True면 cloudpickle로 모듈 전체를 디렉토리에 저장. | False |
modules_to_serialize |
list |
cloudpickle의 register_pickle_by_value로 직렬화할 모듈 리스트. None이면 등록하지 않음. |
None |
소스 코드는 dspy/primitives/base_module.py에 있습니다.
def save(self, path, save_program=False, modules_to_serialize=None):
"""Save the module.
Save the module to a directory or a file. There are two modes:
- `save_program=False`: Save only the state of the module to a json or pickle file, based on the value of
the file extension.
- `save_program=True`: Save the whole module to a directory via cloudpickle, which contains both the state and
architecture of the model.
If `save_program=True` and `modules_to_serialize` are provided, it will register those modules for serialization
with cloudpickle's `register_pickle_by_value`. This causes cloudpickle to serialize the module by value rather
than by reference, ensuring the module is fully preserved along with the saved program. This is useful
when you have custom modules that need to be serialized alongside your program. If None, then no modules
will be registered for serialization.
We also save the dependency versions, so that the loaded model can check if there is a version mismatch on
critical dependencies or DSPy version.
Args:
path (str): Path to the saved state file, which should be a .json or .pkl file when `save_program=False`,
and a directory when `save_program=True`.
save_program (bool): If True, save the whole module to a directory via cloudpickle, otherwise only save
the state.
modules_to_serialize (list): A list of modules to serialize with cloudpickle's `register_pickle_by_value`.
If None, then no modules will be registered for serialization.
"""
metadata = {}
metadata["dependency_versions"] = get_dependency_versions()
path = Path(path)
if save_program:
if path.suffix:
raise ValueError(
f"`path` must point to a directory without a suffix when `save_program=True`, but received: {path}"
)
if path.exists() and not path.is_dir():
raise NotADirectoryError(f"The path '{path}' exists but is not a directory.")
if not path.exists():
# Create the directory (and any parent directories)
path.mkdir(parents=True)
logger.warning("Loading untrusted .pkl files can run arbitrary code, which may be dangerous. To avoid "
'this, prefer saving using json format using module.save("module.json").')
try:
with serialize_by_value(modules_to_serialize), open(path / "program.pkl", "wb") as f:
cloudpickle.dump(self, f)
except Exception as e:
raise RuntimeError(
f"Saving failed with error: {e}. Please remove the non-picklable attributes from your DSPy program, "
"or consider using state-only saving by setting `save_program=False`."
)
with open(path / "metadata.json", "wb") as f:
f.write(orjson.dumps(metadata, option=orjson.OPT_INDENT_2 | orjson.OPT_APPEND_NEWLINE))
return
if path.suffix == ".json":
state = self.dump_state()
state["metadata"] = metadata
try:
with open(path, "wb") as f:
f.write(orjson.dumps(state, option=orjson.OPT_INDENT_2 | orjson.OPT_APPEND_NEWLINE))
except Exception as e:
raise RuntimeError(
f"Failed to save state to {path} with error: {e}. Your DSPy program may contain non "
"json-serializable objects, please consider saving the state in .pkl by using `path` ending "
"with `.pkl`, or saving the whole program by setting `save_program=True`."
)
elif path.suffix == ".pkl":
logger.warning("Loading untrusted .pkl files can run arbitrary code, which may be dangerous. To avoid "
'this, prefer saving using json format using module.save("module.json").')
state = self.dump_state(json_mode=False)
state["metadata"] = metadata
with open(path, "wb") as f:
cloudpickle.dump(state, f)
else:
raise ValueError(f"`path` must end with `.json` or `.pkl` when `save_program=False`, but received: {path}")
set_lm(lm)
이 모듈의 모든 predictor에 언어 모델을 설정합니다.
이 메서드는 모듈의 모든 말단(leaf) 모듈에 언어 모델을 재귀적으로 설정합니다.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lm |
모든 predictor에 사용할 언어 모델 인스턴스. | required |
>>> import dspy
>>> lm = dspy.LM("openai/gpt-4o-mini")
>>> program = dspy.Predict("question -> answer")
>>> program.set_lm(lm)
소스 코드는 dspy/primitives/module.py에 있습니다.
def set_lm(self, lm):
"""Set the language model for all predictors in this module.
This method recursively sets the language model on every leaf module
in this module.
Args:
lm: The language model instance to use for all predictors.
Examples:
>>> import dspy
>>> lm = dspy.LM("openai/gpt-4o-mini")
>>> program = dspy.Predict("question -> answer")
>>> program.set_lm(lm)
"""
for _, param in self.named_parameters():
if isinstance(param, Module):
param.lm = lm
update_config(**kwargs)
소스 코드는 dspy/predict/predict.py에 있습니다.
def update_config(self, **kwargs):
self.config = {**self.config, **kwargs}