Util

트레이스 유틸리티 모듈은 고급 추적 시나리오를 위한 헬퍼 함수와 클래스를 제공합니다.

​

ThreadPoolExecutor

스레드 경계를 넘어 트레이스 컨텍스트를 보존하는 컨텍스트 인식 ThreadPoolExecutor입니다.

from weave import ThreadPoolExecutor

@weave.op
def process_item(item: str) -> str:
    return item.upper()

@weave.op
def process_batch(items: List[str]) -> List[str]:
    with ThreadPoolExecutor(max_workers=4) as executor:
        # Each item is processed in parallel while maintaining trace context
        results = list(executor.map(process_item, items))
    return results

​

주요 기능

• 트레이스에서 부모-자식 관계 보존
• 스레드 경계를 넘어 트레이스 컨텍스트 유지
• 표준 ThreadPoolExecutor API와 호환

​

중첩 추적 예제

@weave.op
def fetch_data(url: str) -> dict:
    # Simulated API call
    return {"data": f"from {url}"}

@weave.op
def process_urls(urls: List[str]) -> List[dict]:
    with ThreadPoolExecutor(max_workers=10) as executor:
        # All fetch_data calls will be properly nested under process_urls
        results = list(executor.map(fetch_data, urls))
    return results

# Usage
urls = ["http://api1.com", "http://api2.com", "http://api3.com"]
data = process_urls(urls)

​

attributes

호출에 속성을 추가하기 위한 컨텍스트 관리자.

from weave import attributes

@weave.op
def my_function():
    return "result"

# Add metadata to the call
with attributes({'environment': 'production', 'version': '1.0.0'}):
    my_function()

​

중첩된 속성

속성은 중첩될 수 있으며 병합됩니다:

with attributes({'env': 'prod'}):
    with attributes({'region': 'us-east-1'}):
        # Call will have both attributes
        my_function()

​

set_tracing_enabled

추적을 일시적으로 비활성화하기 위한 컨텍스트 관리자.

from weave.trace.context.call_context import set_tracing_enabled

@weave.op
def traced_function():
    return "This will be traced"

# Temporarily disable tracing
with set_tracing_enabled(False):
    traced_function()  # This call won't be traced

# Tracing is automatically re-enabled outside the context
traced_function()  # This will be traced

​

get_current_call

op 내에서 현재 실행 중인 호출을 가져옵니다.

from weave.trace.context import get_current_call

@weave.op
def my_function():
    call = get_current_call()
    if call:
        print(f"Current call ID: {call.id}")
        print(f"Current call inputs: {call.inputs}")

​

요약 유틸리티

​

add_summary

호출에 사용자 정의 요약 메트릭을 추가합니다.

@weave.op
def evaluate_model(prompt: str) -> str:
    result = model.generate(prompt)
    
    # Add custom metrics to the call summary
    call = get_current_call()
    if call:
        call.summary.update({
            'token_count': len(result.split()),
            'confidence': 0.95
        })
    
    return result

이러한 유틸리티는 Weave의 추적 시스템과 원활하게 작동하도록 설계되었습니다. 항상 weave.init()가 이러한 기능을 사용하기 전에 호출되었는지 확인하세요.