메인 콘텐츠로 건너뛰기
Weave Calls Screenshot
Weave Calls Screenshot
Weave Calls Screenshot
Calls는 Weave 의 핵심 빌드 블록입니다. 이는 다음을 포함한 단일 함수 실행을 나타냅니다:
  • Inputs (인수)
  • Outputs (반환값)
  • 메타데이터 (기간, 예외, LLM 사용량 등)
Calls는 OpenTelemetry 데이터 모델의 span과 유사합니다. Call은 다음이 가능합니다:
  • Trace에 속함 (동일한 실행 컨텍스트 내의 calls 집합)
  • 부모 및 자식 Calls를 가짐 (트리 구조 형성)

Creating Calls

Weave 에서 Calls를 생성하는 세 가지 주요 방법이 있습니다:

1. LLM 라이브러리의 자동 추적

Weave 는 openai, anthropic, cohere, mistral과 같은 일반적인 LLM 라이브러리에 대한 호출을 자동으로 추적합니다. 프로그램 시작 시 weave.init('project_name')을 호출하기만 하면 됩니다:
weave.initautopatch_settings 인수를 사용하여 Weave 의 기본 추적 동작을 제어할 수 있습니다.
import weave

from openai import OpenAI
client = OpenAI()

# Weave Tracing 초기화
weave.init('intro-example')

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {
            "role": "user",
            "content": "How are you?"
        }
    ],
    temperature=0.8,
    max_tokens=64,
    top_p=1,
)
Call의 summary 사전(dictionary)에 메트릭이나 다른 사후 호출 값을 저장할 수 있습니다. 실행 중에 call.summary를 수정하면 추가한 모든 값이 호출이 끝날 때 Weave 가 계산한 요약 데이터와 병합됩니다.

2. 함수 데코레이팅 및 래핑

하지만 LLM 애플리케이션에는 추적하고 싶은 추가 로직(전처리/후처리, 프롬프트 등)이 있는 경우가 많습니다.
Weave 를 사용하면 @weave.op 데코레이터를 사용하여 이러한 호출을 수동으로 추적할 수 있습니다. 예:
import weave

# Weave Tracing 초기화
weave.init('intro-example')

# 함수 데코레이팅
@weave.op
def my_function(name: str):
    return f"Hello, {name}!"

# 함수 호출 -- Weave가 자동으로 입력과 출력을 추적합니다
print(my_function("World"))
클래스의 메소드도 추적할 수 있습니다.

동기 및 비동기 제너레이터 함수 추적

Weave 는 깊게 중첩된 패턴을 포함하여 동기 및 비동기 제너레이터 함수의 추적을 지원합니다.
제너레이터는 값을 느리게(lazily) 생성하므로, 제너레이터가 완전히 소비될 때(예: 리스트로 변환할 때)만 출력이 로그에 기록됩니다. trace에 출력이 캡처되도록 하려면 제너레이터를 완전히 소비하세요(예: list() 사용).
from typing import Generator
import weave

weave.init("my-project")

# 이 함수는 단순한 동기 제너레이터를 사용합니다.
# Weave는 호출과 입력(`x`)을 추적하지만, 
# 출력 값은 제너레이터가 소비될 때(예: `list()`를 통해)만 캡처됩니다.
@weave.op
def basic_gen(x: int) -> Generator[int, None, None]:
    yield from range(x)

# 제너레이터 파이프라인 내에서 사용되는 일반 동기 함수입니다.
# 이 호출들도 Weave에 의해 독립적으로 추적됩니다.
@weave.op
def inner(x: int) -> int:
    return x + 1

# 다른 추적된 함수(`inner`)를 호출하는 동기 제너레이터입니다.
# 생성된 각 값은 `inner`에 대한 별도의 추적된 호출에서 나옵니다.
@weave.op
def nested_generator(x: int) -> Generator[int, None, None]:
    for i in range(x):
        yield inner(i)

# 위 제너레이터를 구성하는 더 복잡한 제너레이터입니다.
# 여기서 추적하면 계층적인 call 트리가 생성됩니다:
# - `deeply_nested_generator` (부모)
#   - `nested_generator` (자식)
#     - `inner` (손자)
@weave.op
def deeply_nested_generator(x: int) -> Generator[int, None, None]:
    for i in range(x):
        for j in nested_generator(i):
            yield j

# Weave가 출력을 캡처하려면 제너레이터가 *소비*되어야 합니다.
# 이는 동기 및 비동기 제너레이터 모두에 해당됩니다.
res = deeply_nested_generator(4)
list(res)  # 모든 중첩된 호출과 yield의 추적을 트리거합니다.
Weave에서 제너레이터 함수 추적하기.

실행 중에 call 오브젝트 핸들 가져오기

가끔 Call 오브젝트 자체에 대한 핸들을 가져오는 것이 유용할 때가 있습니다. 결과와 Call 오브젝트를 모두 반환하는 op.call 메소드를 호출하여 이를 수행할 수 있습니다. 예:
result, call = my_function.call("World")
그런 다음 call을 사용하여 추가 속성을 설정, 업데이트 또는 가져올 수 있습니다 (피드백에 사용할 호출 ID를 가져오는 데 가장 흔히 사용됨).
op이 클래스의 메소드인 경우, 인스턴스를 op의 첫 번째 인수로 전달해야 합니다 (아래 예시 참조).
# 인스턴스를 첫 번째 인수로 전달하는 것에 주의하세요.
print(instance.my_method.call(instance, "World"))
import weave

# Weave Tracing 초기화
weave.init("intro-example")

class MyClass:
    # 메소드 데코레이팅
    @weave.op
    def my_method(self, name: str):
        return f"Hello, {name}!"

instance = MyClass()

# 메소드 호출 -- Weave가 자동으로 입력과 출력을 추적합니다
instance.my_method.call(instance, "World")

실행 시 call 표시 이름 설정

가끔 호출의 표시 이름을 오버라이드하고 싶을 수 있습니다. 다음 네 가지 방법 중 하나로 이를 수행할 수 있습니다:
  1. op을 호출할 때 표시 이름을 변경합니다:
result = my_function("World", __weave={"display_name": "My Custom Display Name"})
__weave 사전을 사용하면 호출 표시 이름을 설정하며, 이는 Op 표시 이름보다 우선합니다.
  1. 호출별로 표시 이름을 변경합니다. 이는 Call 오브젝트를 반환하는 Op.call 메소드를 사용하며, 그런 다음 Call.set_display_name을 사용하여 표시 이름을 설정할 수 있습니다.
result, call = my_function.call("World")
call.set_display_name("My Custom Display Name")
  1. 특정 Op의 모든 Calls에 대해 표시 이름을 변경합니다:
@weave.op(call_display_name="My Custom Display Name")
def my_function(name: str):
    return f"Hello, {name}!"
  1. call_display_nameCall 오브젝트를 받아 문자열을 반환하는 함수일 수도 있습니다. 함수가 호출될 때 Call 오브젝트가 자동으로 전달되므로, 함수의 이름, 호출 입력, 필드 등을 기반으로 동적으로 이름을 생성할 수 있습니다.
  2. 일반적인 유스 케이스 중 하나는 함수 이름에 타임스탬프를 추가하는 것입니다.
    from datetime import datetime
    
    @weave.op(call_display_name=lambda call: f"{call.func_name}__{datetime.now()}")
    def func():
        return ...
    
  3. .attributes를 사용하여 커스텀 메타데이터를 기록할 수도 있습니다.
    def custom_attribute_name(call):
        model = call.attributes["model"]
        revision = call.attributes["revision"]
        now = call.attributes["date"]
    
        return f"{model}__{revision}__{now}"
    
    @weave.op(call_display_name=custom_attribute_name)
    def func():
        return ...
    
    with weave.attributes(
        {
            "model": "finetuned-llama-3.1-8b",
            "revision": "v0.1.2",
            "date": "2024-08-01",
        }
    ):
        func()  # 표시 이름은 "finetuned-llama-3.1-8b__v0.1.2__2024-08-01"이 됩니다.
    
        with weave.attributes(
            {
                "model": "finetuned-gpt-4o",
                "revision": "v0.1.3",
                "date": "2024-08-02",
            }
        ):
            func()  # 표시 이름은 "finetuned-gpt-4o__v0.1.3__2024-08-02"가 됩니다.
    
기술 참고: “Calls”는 “Ops”에 의해 생성됩니다. Op은 @weave.op으로 데코레이팅된 함수 또는 메소드입니다. 기본적으로 Op의 이름은 함수 이름이며, 연관된 호출들도 동일한 표시 이름을 갖게 됩니다. 위의 예시는 특정 Op의 모든 Calls에 대해 표시 이름을 오버라이드하는 방법을 보여줍니다. 가끔 사용자는 Op 자체의 이름을 오버라이드하고 싶어 합니다. 이는 다음 두 가지 방법 중 하나로 가능합니다:
  1. 호출이 로깅되기 전에 Op의 name 속성을 설정합니다.
my_function.name = "My Custom Op Name"
  1. op 데코레이터에서 name 옵션을 설정합니다.
@weave.op(name="My Custom Op Name")

병렬(멀티 스레드) 함수 호출 추적

기본적으로 병렬 호출은 Weave 에서 모두 별도의 루트 호출로 나타납니다. 동일한 부모 op 아래에 올바르게 중첩되게 하려면 ThreadPoolExecutor를 사용하세요.다음 코드 샘플은 ThreadPoolExecutor 사용법을 보여줍니다. 첫 번째 함수인 funcx를 받아 x+1을 반환하는 간단한 op입니다. 두 번째 함수인 outer는 입력 리스트를 받는 또 다른 op입니다. outer 내부에서 ThreadPoolExecutorexc.map(func, inputs)를 사용하면 func에 대한 각 호출이 여전히 동일한 부모 trace 컨텍스트를 유지합니다.
import weave

@weave.op
def func(x):
    return x+1

@weave.op
def outer(inputs):
    with weave.ThreadPoolExecutor() as exc:
        exc.map(func, inputs)

# Weave 프로젝트 이름 업데이트
client = weave.init('my-weave-project')
outer([1,2,3,4,5])
Weave UI에서 이는 단일 부모 호출과 5개의 중첩된 자식 호출을 생성하므로, 증분 작업이 병렬로 실행되더라도 완전한 계층적 trace를 얻을 수 있습니다.Trace UI, outer에 대한 단일 부모 호출과 5개의 중첩된 자식 호출을 보여줍니다.

3. 수동 Call 추적

API를 직접 사용하여 수동으로 Calls를 생성할 수도 있습니다.
import weave

# Weave Tracing 초기화
client = weave.init('intro-example')

def my_function(name: str):
    # 호출 시작
    call = client.create_call(op="my_function", inputs={"name": name})

    # ... 함수 코드 ...

    # 호출 종료
    client.finish_call(call, output="Hello, World!")

# 함수 호출
print(my_function("World"))

4. 클래스 및 오브젝트 메소드 추적

클래스 및 오브젝트 메소드도 추적할 수 있습니다.
weave.op을 사용하여 클래스의 모든 메소드를 추적합니다.
import weave

# Weave Tracing 초기화
weave.init("intro-example")

class MyClass:
    # 메소드 데코레이팅
    @weave.op
    def my_method(self, name: str):
        return f"Hello, {name}!"

instance = MyClass()

# 메소드 호출 -- Weave가 자동으로 입력과 출력을 추적합니다
print(instance.my_method("World"))

Viewing Calls

웹 앱에서 호출을 보려면 다음 단계를 따르세요:
  1. 프로젝트의 Traces 탭으로 이동합니다.
  2. 목록에서 보려는 호출을 찾습니다.
  3. 호출을 클릭하여 상세 페이지를 엽니다.
상세 페이지에는 호출의 입력, 출력, 런타임 및 추가 메타데이터가 표시됩니다.Web App에서 Call 보기

weave.Markdown으로 렌더링된 traces 커스터마이징

weave.Markdown을 사용하면 원본 데이터를 잃지 않고 trace 정보가 표시되는 방식을 커스터마이징할 수 있습니다. 이를 통해 기본 데이터 구조를 보존하면서 입력과 출력을 읽기 쉬운 형식의 콘텐츠 블록으로 렌더링할 수 있습니다.
@weave.op 데코레이터에서 postprocess_inputspostprocess_output 함수를 사용하여 trace 데이터를 포맷팅하세요. 다음 코드 샘플은 포스트프로세서를 사용하여 Weave 에서 이모지와 더 읽기 쉬운 포맷팅으로 호출을 렌더링합니다:
import weave

def postprocess_inputs(query) -> weave.Markdown:
    search_box = f"""
**Search Query:**
``+`
{query}
``+`
"""
    return {"search_box": weave.Markdown(search_box),
            "query": query}

def postprocess_output(docs) -> weave.Markdown:
    formatted_docs = f"""
# {docs[0]["title"]}

{docs[0]["content"]}

[Read more]({docs[0]["url"]})

---

# {docs[1]["title"]}

{docs[1]["content"]}

[Read more]({docs[1]["url"]})
"""
    return weave.Markdown(formatted_docs)

@weave.op(
    postprocess_inputs=postprocess_inputs,
    postprocess_output=postprocess_output,
)
def rag_step(query):
    # S&P 500 기업들에 대한 예시 신문 기사들
    docs = [
        {
            "title": "OpenAI",
            "content": "OpenAI is a company that makes AI models.",
            "url": "https://www.openai.com",
        },
        {
            "title": "Google",
            "content": "Google is a company that makes search engines.",
            "url": "https://www.google.com",
        },
    ]
    return docs

if __name__ == "__main__":
    weave.init('markdown_renderers')
    rag_step("Tell me about OpenAI")
다음 스크린샷에서 각각 포맷팅되지 않은 출력과 포맷팅된 출력 사이의 차이점을 확인할 수 있습니다. 코드 샘플을 사용하여 Weave UI에서 렌더링된 호출.

Updating Calls

Calls는 생성된 후에는 대부분 불변이지만, 지원되는 몇 가지 변형 작업이 있습니다: 호출 상세 페이지로 이동하여 UI에서 이러한 모든 작업을 수행할 수 있습니다:
Web App에서 Call 업데이트

표시 이름 설정

호출의 표시 이름을 설정하려면 Call.set_display_name() 메소드를 사용할 수 있습니다.
import weave

# 클라이언트 초기화
client = weave.init("your-project-name")

# ID로 특정 호출 가져오기
call = client.get_call("call-uuid-here")

# 호출의 표시 이름 설정
call.set_display_name("My Custom Display Name")
실행 시 호출의 표시 이름을 설정할 수도 있습니다.

피드백 추가

자세한 내용은 Feedback Documentation을 참조하세요.

Call 삭제

Python API를 사용하여 Call을 삭제하려면 Call.delete 메소드를 사용할 수 있습니다.
import weave

# 클라이언트 초기화
client = weave.init("your-project-name")

# ID로 특정 호출 가져오기
call = client.get_call("call-uuid-here")

# 호출 삭제
call.delete()

여러 Calls 삭제

Python API를 사용하여 여러 Calls를 일괄 삭제하려면 Call ID 목록을 delete_calls()에 전달하세요.
import weave

# 클라이언트 초기화
client = weave.init("my-project")

# 클라이언트에서 모든 호출 가져오기
all_calls = client.get_calls()

# 처음 1000개의 Call 오브젝트 리스트 가져오기
first_1000_calls = all_calls[:1000]

# 처음 1000개의 Call ID 리스트 가져오기
first_1000_calls_ids = [c.id for c in first_1000_calls]

# ID로 처음 1000개의 Call 오브젝트 삭제
client.delete_calls(call_ids=first_1000_calls_ids)

Querying and exporting Calls

많은 호출들의 스크린샷
프로젝트의 /calls 페이지(“Traces” 탭)에는 프로젝트의 모든 Calls에 대한 테이블 뷰가 포함되어 있습니다. 여기에서 다음을 수행할 수 있습니다:
  • 정렬
  • 필터링
  • 내보내기
Calls Table View
내보내기 모달(위 그림 참조)을 사용하면 다양한 형식으로 데이터를 내보낼 수 있을 뿐만 아니라, 선택한 호출에 해당하는 Python 및 CURL 코드를 보여줍니다! 가장 쉽게 시작하는 방법은 UI에서 뷰를 구성한 다음, 생성된 코드 조각을 통해 내보내기 API에 대해 자세히 알아보는 것입니다.
Python API를 사용하여 호출을 가져오려면 client.get_calls 메소드를 사용할 수 있습니다:
import weave

# 클라이언트 초기화
client = weave.init("your-project-name")

# 호출 가져오기
calls = client.get_calls(filter=...)

Call 스키마

필드 전체 목록은 스키마를 참조하세요.
속성타입설명
idstring (uuid)호출의 고유 식별자
project_idstring (선택사항)연관된 프로젝트 식별자
op_namestring연산의 이름 (참조일 수 있음)
display_namestring (선택사항)호출에 대한 사용자 친화적인 이름
trace_idstring (uuid)이 호출이 속한 trace의 식별자
parent_idstring (uuid)부모 호출의 식별자
started_atdatetime호출이 시작된 시점의 타임스탬프
attributesDict[str, Any]호출에 대한 사용자 정의 메타데이터 (실행 중 읽기 전용)
inputsDict[str, Any]호출에 대한 입력 파라미터
ended_atdatetime (선택사항)호출이 종료된 시점의 타임스탬프
exceptionstring (선택사항)호출 실패 시 에러 메시지
outputAny (선택사항)호출의 결과
summaryOptional[SummaryMap]실행 후 요약 정보. 실행 중에 이를 수정하여 커스텀 메트릭을 기록할 수 있습니다.
wb_user_idOptional[str]연관된 Weights & Biases 사용자 ID
wb_run_idOptional[str]연관된 Weights & Biases run ID
deleted_atdatetime (선택사항)호출 삭제 시점의 타임스탬프 (해당하는 경우)
위 테이블은 Weave 에서 Call의 주요 속성을 개략적으로 설명합니다. 각 속성은 함수 호출을 추적하고 관리하는 데 중요한 역할을 합니다:
  • id, trace_id, parent_id 필드는 시스템 내에서 호출을 정리하고 관련짓는 데 도움이 됩니다.
  • 시간 정보(started_at, ended_at)를 통해 성능 분석이 가능합니다.
  • attributesinputs 필드는 호출에 대한 컨텍스트를 제공합니다. Attributes는 호출이 시작되면 고정되므로 호출 전에 weave.attributes로 설정해야 합니다. outputsummary는 결과를 캡처하며, 실행 중에 summary를 업데이트하여 추가 메트릭을 로깅할 수 있습니다.
  • Weights & Biases와의 인테그레이션은 wb_user_idwb_run_id를 통해 용이해집니다.
이러한 포괄적인 속성 세트를 통해 프로젝트 전체에서 함수 호출에 대한 상세한 추적 및 분석이 가능합니다. 계산된 필드:
  • 비용
  • 기간
  • 상태

Saved views

Trace 테이블 구성, 필터 및 정렬을 _저장된 뷰_로 저장하여 선호하는 설정에 빠르게 엑세스할 수 있습니다. UI 및 Python SDK를 통해 저장된 뷰를 구성하고 엑세스할 수 있습니다. 자세한 내용은 Saved Views를 참조하세요.

Traces 테이블에서 W&B run 보기

Weave 를 사용하면 코드의 함수 호출을 추적하고 이들이 실행된 W&B runs에 직접 링크할 수 있습니다. @weave.op()으로 함수를 추적하고 wandb.init() 컨텍스트 내에서 호출하면, Weave 는 자동으로 해당 trace를 W&B run과 연결합니다. 연관된 runs에 대한 링크는 Traces 테이블에 표시됩니다.
다음 Python 코드는 wandb.init() 컨텍스트 내에서 실행될 때 추적된 작업이 어떻게 W&B runs에 연결되는지 보여줍니다. 이러한 traces는 Weave UI에 나타나며 해당 run과 연관됩니다.
import wandb
import weave

def example_wandb(projname):
    # projname을 entity와 project로 분리
    entity, project = projname.split("/", 1)

    # 추적을 위한 Weave 컨텍스트 초기화
    weave.init(projname)

    # 추적 가능한 작업 정의
    @weave.op()
    def say(message: str) -> str:
        return f"I said: {message}"

    # 첫 번째 W&B run
    with wandb.init(
        entity=entity,
        project=project,
        notes="Experiment 1",
        tags=["baseline", "paper1"],
    ) as run:
        say("Hello, world!")
        say("How are you!")
        run.log({"messages": 2})

    # 두 번째 W&B run
    with wandb.init(
        entity=entity,
        project=project,
        notes="Experiment 2",
        tags=["baseline", "paper1"],
    ) as run:
        say("Hello, world from experiment 2!")
        say("How are you!")
        run.log({"messages": 2})

if __name__ == "__main__":
    # 실제 W&B username/project로 교체하세요
    example_wandb("your-username/your-project")
코드 샘플 사용 방법:
  1. 터미널에서 의존성 설치:
pip install wandb weave
  1. W&B 로그인:
wandb login
  1. 스크립트에서 your-username/your-project를 실제 W&B entity/project로 변경합니다.
  2. 스크립트 실행:
python weave_trace_with_wandb.py
  1. https://weave.wandb.ai에 접속하여 프로젝트를 선택합니다.
  2. Traces 탭에서 trace 결과를 확인합니다. 연관된 runs에 대한 링크가 Traces 테이블에 표시됩니다.

Configure autopatching

기본적으로 Weave 는 openai, anthropic, cohere, mistral과 같은 일반적인 LLM 라이브러리에 대한 호출을 자동으로 패치하고 추적합니다.
autopatch_settings 인수는 지원이 중단될 예정입니다. 암시적 패치를 비활성화하려면 implicitly_patch_integrations=False를 사용하거나, 인테그레이션별로 설정을 구성하려면 patch_openai(settings={...})와 같은 특정 패치 함수를 호출하세요.

모든 autopatching 비활성화

weave.init(..., implicitly_patch_integrations=False)

특정 인테그레이션 활성화

import weave

weave.init(..., implicitly_patch_integrations=False)

# 그런 다음 원하는 인테그레이션만 수동으로 패치합니다
weave.integrations.patch_anthropic()
weave.integrations.patch_cohere()

입력 및 출력 후처리

패치 함수에 설정을 전달하여 입력 및 출력(예: PII 데이터) 처리 방식을 커스터마이징할 수 있습니다:
import weave.integrations

def redact_inputs(inputs: dict) -> dict:
    if "email" in inputs:
        inputs["email"] = "[REDACTED]"
    return inputs

weave.init(...)
weave.integrations.patch_openai(
    settings={
        "op_settings": {"postprocess_inputs": redact_inputs}
    }
)
자세한 내용은 How to use Weave with PII data를 참조하세요.

FAQs

대규모 trace가 잘리는 현상을 어떻게 방지하나요?

자세한 내용은 문제 해결 가이드Trace 데이터가 잘림 섹션을 참조하세요.

추적을 어떻게 비활성화하나요?

환경 변수

프로그램 전체에 대해 무조건적으로 추적을 비활성화하고 싶은 경우, 환경 변수 WEAVE_DISABLED=true를 설정할 수 있습니다.

클라이언트 초기화

때로는 특정 조건에 따라 특정 초기화에 대해서만 추적을 활성화하고 싶을 수 있습니다. 이 경우 init 설정에서 disabled 플래그로 클라이언트를 초기화할 수 있습니다.
import weave

# 클라이언트 초기화
client = weave.init(..., settings={"disabled": True})

컨텍스트 매니저

마지막으로, 애플리케이션 로직에 따라 단일 함수에 대해서만 추적을 비활성화하고 싶을 수 있습니다. 이 경우 weave.trace.context.call_context에서 임포트할 수 있는 with set_tracing_enabled(False) 컨텍스트 매니저를 사용할 수 있습니다.
import weave
from weave.trace.context.call_context import set_tracing_enabled

client = weave.init(...)

@weave.op
def my_op():
    ...

with set_tracing_enabled(False):
    my_op()

Call에 대한 정보를 어떻게 캡처하나요?

일반적으로 op을 직접 호출합니다:
@weave.op
def my_op():
    ...

my_op()
하지만 op에서 call 메소드를 호출하여 호출 오브젝트에 직접 엑세스할 수도 있습니다:
@weave.op
def my_op():
    ...

output, call = my_op.call()
여기서 call 오브젝트는 입력, 출력 및 기타 메타데이터를 포함하여 호출에 대한 모든 정보를 담고 있습니다.
@weave.op
def my_op():
    ...

output, call = my_op.call()