dspy.Parallel

dspy.Parallel

dspy.Parallel은 (module, example) 쌍을 병렬로, 멀티스레드로 실행하기 위한 유틸리티 클래스입니다. Example, dict, tuple, list 같은 다양한 example 형식을 지원하고, 견고한 에러 처리와 선택적인 진행률 표시를 제공하며, 필요하면 실패한 example과 예외까지 반환해 줍니다.

출처: 문서

본문

dspy.Parallel(
    num_threads: int | None = None,
    max_errors: int | None = None,
    access_examples: bool = True,
    return_failed_examples: bool = False,
    provide_traceback: bool | None = None,
    disable_progress_bar: bool = False,
    timeout: int = 120,
    straggler_limit: int = 3,
)

(module, example) 쌍을 병렬·멀티스레드로 실행하는 유틸리티 클래스입니다. Example, dict, tuple, list 같은 다양한 example 형식을 지원하고, 견고한 에러 처리와 선택적인 진행률 표시를 제공하며, 필요하면 실패한 example과 예외까지 반환합니다.

Parameters:

Example

import dspy
from dspy import Parallel
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

examples = [
    {"question": "What is the capital of Spain?"},
    {"question": "What is 3 * 4?"},
    {"question": "Who wrote Hamlet?"},
]

module = dspy.Predict("question->answer")
exec_pairs = [(module, example) for example in examples]
parallel = Parallel(num_threads=3, disable_progress_bar=False)
results = parallel(exec_pairs)
for i, result in enumerate(results):
    print(f"Result {i+1}: {result.answer}")

# Expected Output:
# Result 1: Madrid
# Result 2: 12
# Result 3: William Shakespeare

소스 코드는 dspy/predict/parallel.py에 있습니다.

def __init__(
    self,
    num_threads: int | None = None,
    max_errors: int | None = None,
    access_examples: bool = True,
    return_failed_examples: bool = False,
    provide_traceback: bool | None = None,
    disable_progress_bar: bool = False,
    timeout: int = 120,
    straggler_limit: int = 3,
):
    """
    (module, example) 쌍을 병렬·멀티스레드로 실행하는 유틸리티 클래스.
    다양한 example 형식(예: `Example`, dict, tuple, list)을 지원하고,
    견고한 에러 처리, 선택적인 진행률 표시를 제공하며,
    필요하면 실패한 example과 예외까지 반환한다.

    Args:
        num_threads (Optional[int]): 사용할 스레드의 수. 기본값은 `settings.num_threads`.
        max_errors (Optional[int]): 예외를 던지기 전에 허용할 최대 에러 수. 기본값은 `settings.max_errors`.
        access_examples (bool): `Example` 객체를 `.inputs()`로 언팩할지 여부. 기본값은 True.
        return_failed_examples (bool): 실패한 example을 반환할지 여부. 기본값은 False.
        provide_traceback (Optional[bool]): traceback을 제공할지 여부. 기본값은 None.
        disable_progress_bar (bool): 진행률 표시줄을 끌지 여부. 기본값은 False.

    Example:
        ...(위의 예시 코드와 동일)...
    """

    super().__init__()
    self.num_threads = num_threads or settings.num_threads
    self.max_errors = settings.max_errors if max_errors is None else max_errors
    self.access_examples = access_examples
    self.return_failed_examples = return_failed_examples
    self.provide_traceback = provide_traceback
    self.disable_progress_bar = disable_progress_bar
    self.timeout = timeout
    self.straggler_limit = straggler_limit

    self.error_count = 0
    self.error_lock = threading.Lock()
    self.cancel_jobs = threading.Event()
    self.failed_examples = []
    self.exceptions = []

Methods

__call__(*args: Any, **kwargs: Any) -> Any

소스 코드는 dspy/predict/parallel.py에 있습니다.

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    return self.forward(*args, **kwargs)

forward(exec_pairs: list[tuple[Any, Example]], num_threads: int | None = None) -> list[Any]

소스 코드는 dspy/predict/parallel.py에 있습니다.

def forward(self, exec_pairs: list[tuple[Any, Example]], num_threads: int | None = None) -> list[Any]:
    num_threads = num_threads if num_threads is not None else self.num_threads

    executor = ParallelExecutor(
        num_threads=num_threads,
        max_errors=self.max_errors,
        provide_traceback=self.provide_traceback,
        disable_progress_bar=self.disable_progress_bar,
        timeout=self.timeout,
        straggler_limit=self.straggler_limit,
    )

    def process_pair(pair):
        result = None
        module, example = pair

        if isinstance(example, Example):
            if self.access_examples:
                result = module(**example.inputs())
            else:
                result = module(example)
        elif isinstance(example, dict):
            result = module(**example)
        elif isinstance(example, list) and module.__class__.__name__ == "Parallel":
            result = module(example)
        elif isinstance(example, tuple):
            result = module(*example)
        else:
            raise ValueError(
                f"Invalid example type: {type(example)}, only supported types are Example, dict, list and tuple"
            )
        return result

    # Execute the processing function over the execution pairs
    results = executor.execute(process_pair, exec_pairs)

    # Populate failed examples and exceptions from the executor
    if self.return_failed_examples:
        for failed_idx in executor.failed_indices:
            if failed_idx < len(exec_pairs):
                _, original_example = exec_pairs[failed_idx]
                self.failed_examples.append(original_example)
                if exception := executor.exceptions_map.get(failed_idx):
                    self.exceptions.append(exception)

        return results, self.failed_examples, self.exceptions
    else:
        return results

더 알아보기 (Learn more)

• Predict
• Example
• ReAct