동시성 & 성능

동시성 & 성능 (Concurrency & Performance)

평가 케이스가 병렬로 실행되는 방식을 제어해요.

기본적으로 Pydantic Evals는 처리량을 최대화하기 위해 모든 케이스를 동시에 실행해요. max_concurrency 매개변수로 이 동작을 제어할 수 있어요.

출처: 문서

본문

기본 사용법

from pydantic_evals import Case, Dataset


def my_task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='concurrency_demo', cases=[Case(inputs='test1'), Case(inputs='test2')])

# 모든 케이스를 동시에 실행 (기본)
report = dataset.evaluate_sync(my_task)

# 동시 케이스를 5개로 제한
report = dataset.evaluate_sync(my_task, max_concurrency=5)

# 순차 실행 (한 번에 하나씩)
report = dataset.evaluate_sync(my_task, max_concurrency=1)

동시성을 제한해야 할 때

속도 제한 (Rate Limiting)

많은 API에는 동시 요청을 제한하는 속도 제한이 있어요:

from pydantic_evals import Case, Dataset


async def my_llm_task(inputs: str) -> str:
    return f'LLM Result: {inputs}'


dataset = Dataset(name='rate_limit_demo', cases=[Case(inputs='test1')])

# API가 초당 10개 요청을 허용한다면
report = dataset.evaluate_sync(
    my_llm_task,
    max_concurrency=10,
)

리소스 제약

시스템 리소스에 과부하가 걸리지 않도록 동시성을 제한해요:

from pydantic_evals import Case, Dataset


def heavy_computation(inputs: str) -> str:
    return f'Heavy: {inputs}'


def db_query_task(inputs: str) -> str:
    return f'DB: {inputs}'


dataset = Dataset(name='resource_constraints', cases=[Case(inputs='test1')])

# 메모리 집약적 작업
report = dataset.evaluate_sync(
    heavy_computation,
    max_concurrency=2,  # 한 번에 2개만
)

# 데이터베이스 커넥션 풀 한계
report = dataset.evaluate_sync(
    db_query_task,
    max_concurrency=5,  # 커넥션 풀 크기에 맞춤
)

디버깅

명확한 오류 트레이스를 보려면 순차로 실행해요:

from pydantic_evals import Case, Dataset


def my_task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='debug_demo', cases=[Case(inputs='test1')])

# 디버깅이 더 쉬움
report = dataset.evaluate_sync(
    my_task,
    max_concurrency=1,
)

성능 비교

성능 차이를 보여주는 예시예요:

concurrency_example.py

import asyncio

from pydantic_evals import Case, Dataset

# 여러 테스트 케이스가 있는 데이터셋 만들기
dataset = Dataset(
    name='performance_comparison',
    cases=[
        Case(
            name=f'case_{i}',
            inputs=i,
            expected_output=i * 2,
        )
        for i in range(10)
    ]
)


async def slow_task(input_value: int) -> int:
    """Simulates a slow operation (e.g., API call)."""
    await asyncio.sleep(0.1)  # 케이스당 100ms
    return input_value * 2


# 무제한 동시성: 총 ~0.1s (모든 케이스가 병렬 실행)
report = dataset.evaluate_sync(slow_task)

# 제한된 동시성: 총 ~0.5s (한 번에 2개, 5배치)
report = dataset.evaluate_sync(slow_task, max_concurrency=2)

# 순차: 총 ~1.0s (한 번에 하나씩, 10케이스)
report = dataset.evaluate_sync(slow_task, max_concurrency=1)

평가자와의 동시성

작업 실행과 평가자 실행 모두 기본적으로 동시에 일어나요:

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import LLMJudge


def my_task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(
    name='evaluator_concurrency',
    cases=[Case(inputs=f'test{i}') for i in range(40)],  # 40개 케이스
    evaluators=[
        LLMJudge(rubric='Quality check'),  # API 호출 발생
    ],
)

# 작업과 평가자 모두 제어된 동시성으로 실행
report = dataset.evaluate_sync(
    my_task,
    max_concurrency=10,
)

평가자가 비싸다면(예: LLMJudge), 동시성을 제한하면 다음을 관리하는 데 도움이 돼요:

  • API 속도 제한
  • 비용 (동시 API 호출 감소)
  • 메모리 사용량

비동기 vs 동기

동기와 비동기 평가 모두 동시성 제어를 지원해요:

동기 API

from pydantic_evals import Case, Dataset


def my_task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='sync_demo', cases=[Case(inputs='test1')])

# 내부적으로 제어된 동시성으로 비동기 작업 실행
report = dataset.evaluate_sync(my_task, max_concurrency=10)

비동기 API

from pydantic_evals import Case, Dataset


async def my_task(inputs: str) -> str:
    return f'Result: {inputs}'


async def run_evaluation():
    dataset = Dataset(name='async_demo', cases=[Case(inputs='test1')])
    # 동일한 동작이지만 비동기 컨텍스트에서
    report = await dataset.evaluate(my_task, max_concurrency=10)
    return report

동시성 모니터링

설정을 최적화하려면 실행을 추적해요:

import time

from pydantic_evals import Case, Dataset


def task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='monitoring', cases=[Case(inputs=f'test{i}') for i in range(10)])

t0 = time.time()
report = dataset.evaluate_sync(task, max_concurrency=10)
duration = time.time() - t0

num_cases = len(report.cases) + len(report.failures)
avg_duration = duration / num_cases

print(f'Total: {duration:.2f}s')
#> Total: 0.01s
print(f'Cases: {num_cases}')
#> Cases: 10
print(f'Avg per case: {avg_duration:.2f}s')
#> Avg per case: 0.00s
print(f'Effective concurrency: ~{num_cases * avg_duration / duration:.1f}')
#> Effective concurrency: ~1.0

속도 제한 처리

속도 제한에 걸리면 평가가 실패해요. 재시도 전략을 사용해요:

from pydantic_evals import Case, Dataset


def task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='rate_limit_handling', cases=[Case(inputs='test1')])

# 속도 제한을 피하려고 동시성 감소
report = dataset.evaluate_sync(
    task,
    max_concurrency=5,  # 속도 제한 아래 유지
)

일시적인 실패 처리는 재시도 전략을 참고해요.

다음 단계

더 알아보기 (Learn more)