Weave Service API로 트레이싱하기

이 문서는 대화형 노트북입니다. 로컬에서 실행하거나 다음 링크를 사용하세요.

이 노트북에서는 W&B Weave Service API로 트레이스를 로깅하여 다음 작업을 수행하는 방법을 알아봅니다.

단순한 LLM Call과 응답을 모의로 만들어 Weave에 로깅합니다.
더 복잡한 LLM Call과 응답을 모의로 만들어 Weave에 로깅합니다.
로깅된 트레이스에 샘플 조회 쿼리를 실행합니다.

로깅된 트레이스 보기 이 가이드의 코드를 실행하면서 생성된 모든 Weave 트레이스를 보려면 Weave 프로젝트(team_id\project_id로 지정)의 Traces 탭으로 이동한 다음 트레이스 이름을 선택하세요.

시작하기 전에 사전 요구 사항을 완료하세요.

사전 요구 사항: 변수 및 엔드포인트 설정

트레이스를 로깅하기 전에 Service API 엔드포인트를 설정하고 W&B 자격 증명으로 인증해야 합니다. 다음 코드는 Service API에 액세스할 때 사용하는 URL 엔드포인트를 설정합니다.

또한 다음 변수를 설정해야 합니다.

project_id: 트레이스를 로깅할 Weights & Biases 프로젝트 이름입니다.
team_id: CoreWeave Forge 팀 이름입니다.
wandb_token: W&B API 키입니다.

import datetime
import json

import requests

# 헤더 및 URL
headers = {"Content-Type": "application/json"}
url_start = "https://trace.wandb.ai/call/start"
url_end = "https://trace.wandb.ai/call/end"
url_stream_query = "https://trace.wandb.ai/calls/stream_query"

# W&B 변수
team_id = ""
project_id = ""
wandb_token = ""

단순한 트레이스

다음 섹션에서는 단순한 트레이스를 만드는 방법을 단계별로 설명합니다.

단순한 트레이스 시작하기
단순한 트레이스 종료하기

단순한 트레이스 시작하기

다음 코드는 샘플 LLM Call payload_start를 생성하고, url_start 엔드포인트를 사용해 Weave에 로깅합니다. payload_start 객체는 Why is the sky blue?라는 쿼리로 OpenAI의 gpt-4o를 호출하는 상황을 시뮬레이션합니다. 코드가 성공적으로 실행되면 트레이스가 시작되었음을 알리는 메시지가 출력됩니다.

Call started. ID: 01939cdc-38d2-7d61-940d-dcca0a56c575, Trace ID: 01939cdc-38d2-7d61-940d-dcd0e76c5f34
python
## ------------
## 트레이스 시작
## ------------
payload_start = {
    "start": {
        "project_id": f"{team_id}/{project_id}",
        "op_name": "simple_trace",
        "started_at": datetime.datetime.now().isoformat(),
        "inputs": {
            # 트레이스를 펼쳤을 때 채팅 UI가 표시되도록 하려면 이 "messages" 형식을 사용하세요.
            "messages": [{"role": "user", "content": "Why is the sky blue?"}],
            "model": "gpt-4o",
        },
        "attributes": {},
    }
}
response = requests.post(
    url_start, headers=headers, json=payload_start, auth=("api", wandb_token)
)
if response.status_code == 200:
    data = response.json()
    call_id = data.get("id")
    trace_id = data.get("trace_id")
    print(f"Call started. ID: {call_id}, Trace ID: {trace_id}")
else:
    print("Start request failed with status:", response.status_code)
    print(response.text)
    exit()

단순한 트레이스 종료하기

다음 코드는 단순한 트레이스를 완료하기 위해 샘플 LLM Call payload_end를 생성하고, url_end 엔드포인트를 통해 Weave에 로깅합니다. payload_end 객체는 Why is the sky blue?라는 쿼리에 대한 OpenAI gpt-4o의 응답을 모방합니다. 이 객체는 Weave 대시보드의 트레이스 뷰에 가격 요약 정보와 Chat Completion이 표시되도록 형식이 지정되어 있습니다. 코드가 성공적으로 실행되면 트레이스가 완료되었음을 알리는 메시지가 출력됩니다:

Call ended.
python
## ------------
## 트레이스 종료
## ------------
payload_end = {
    "end": {
        "project_id": f"{team_id}/{project_id}",
        "id": call_id,
        "ended_at": datetime.datetime.now().isoformat(),
        "output": {
            # 펼친 트레이스의 채팅 UI에 완료 결과를 표시하려면 이 "choices" 형식을 사용하세요.
            "choices": [
                {
                    "message": {
                        "content": "It’s due to Rayleigh scattering, where shorter blue wavelengths of sunlight scatter in all directions."
                    }
                },
            ]
        },
        # 트레이스 table에 가격 요약 정보를 표시하려면 summary를 다음과 같은 형식으로 작성하세요.
        "summary": {
            "usage": {
                "gpt-4o": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30,
                    "requests": 1,
                }
            }
        },
    }
}
response = requests.post(
    url_end, headers=headers, json=payload_end, auth=("api", wandb_token)
)
if response.status_code == 200:
    print("Call ended.")
else:
    print("End request failed with status:", response.status_code)
    print(response.text)

복잡한 트레이스

단순한 단일 단계 트레이스를 로깅해 보았으니, 이제 다음 섹션에서는 여러 오퍼레이션으로 구성된 RAG 조회처럼 하위 span이 포함된 더 복잡한 트레이스를 만드는 방법을 살펴봅니다.

복잡한 트레이스 시작하기
RAG 문서 조회를 위한 하위 span 추가하기
LLM 완료 호출을 위한 하위 span 추가하기
복잡한 트레이스 종료하기

복잡한 트레이스 시작하기

다음 코드는 여러 span으로 구성된 더 복잡한 트레이스를 만드는 방법을 보여줍니다. RAG(Retrieval-Augmented Generation) 조회에 이어 LLM Call을 수행하는 경우를 예로 들 수 있습니다. 코드의 첫 부분에서는 오퍼레이션을 나타내는 상위 트레이스(payload_parent_start)를 초기화합니다. 여기서 오퍼레이션은 사용자 쿼리 Can you summarize the key points of this document?를 처리합니다. payload_parent_start 객체는 다단계 워크플로의 첫 단계를 모방하며, url_start 엔드포인트를 통해 Weave에 오퍼레이션을 로깅합니다. 코드가 성공적으로 실행되면 부모 호출이 로깅되었음을 알리는 메시지가 출력됩니다.

Parent call started. ID: 01939d26-0844-7c43-94bb-cdc471b6d65f, Trace ID: 01939d26-0844-7c43-94bb-cdd97dc296c8
python
## ------------
## 트레이스 시작(부모)
## ------------

# 부모 호출: 시작
payload_parent_start = {
    "start": {
        "project_id": f"{team_id}/{project_id}",
        "op_name": "complex_trace",
        "started_at": datetime.datetime.now().isoformat(),
        "inputs": {"question": "Can you summarize the key points of this document?"},
        "attributes": {},
    }
}
response = requests.post(
    url_start, headers=headers, json=payload_parent_start, auth=("api", wandb_token)
)
if response.status_code == 200:
    data = response.json()
    parent_call_id = data.get("id")
    trace_id = data.get("trace_id")
    print(f"Parent call started. ID: {parent_call_id}, Trace ID: {trace_id}")
else:
    print("Parent start request failed with status:", response.status_code)
    print(response.text)
    exit()

RAG 문서 조회용 하위 span 추가

다음 코드는 이전 단계에서 시작한 복합 상위 트레이스에 하위 span을 추가하는 방법을 보여 줍니다. 이 단계에서는 워크플로의 RAG 문서 조회 하위 오퍼레이션을 모델링합니다. 하위 트레이스는 payload_child_start 객체로 시작하며, 이 객체에는 다음 항목이 포함됩니다.

trace_id: 이 하위 span을 상위 트레이스에 연결합니다.
parent_id: 하위 span을 상위 오퍼레이션에 연결합니다.
inputs: 검색 쿼리를 로깅합니다. 예: "This is a search query of the documents I'm looking for."

url_start 엔드포인트를 성공적으로 호출하면 자식 호출이 시작되고 완료되었다는 메시지가 출력됩니다.

Child call started. ID: 01939d32-23d6-75f2-9128-36a4a806f179
Child call ended.
python
## ------------
## 자식 span:
## 예: RAG 문서 조회
## ------------

# 자식 call: 시작
payload_child_start = {
    "start": {
        "project_id": f"{team_id}/{project_id}",
        "op_name": "rag_document_lookup",
        "trace_id": trace_id,
        "parent_id": parent_call_id,
        "started_at": datetime.datetime.now().isoformat(),
        "inputs": {
            "document_search": "This is a search query of the documents I'm looking for."
        },
        "attributes": {},
    }
}
response = requests.post(
    url_start, headers=headers, json=payload_child_start, auth=("api", wandb_token)
)
if response.status_code == 200:
    data = response.json()
    child_call_id = data.get("id")
    print(f"Child call started. ID: {child_call_id}")
else:
    print("Child start request failed with status:", response.status_code)
    print(response.text)
    exit()

# 자식 call: 종료
payload_child_end = {
    "end": {
        "project_id": f"{team_id}/{project_id}",
        "id": child_call_id,
        "ended_at": datetime.datetime.now().isoformat(),
        "output": {
            "document_results": "This will be the RAG'd document text which will be returned from the search query."
        },
        "summary": {},
    }
}
response = requests.post(
    url_end, headers=headers, json=payload_child_end, auth=("api", wandb_token)
)
if response.status_code == 200:
    print("Child call ended.")
else:
    print("Child end request failed with status:", response.status_code)
    print(response.text)

LLM 완료 호출에 대한 하위 span 추가

다음 코드는 복잡한 상위 트레이스에 LLM 완료 호출을 나타내는 하위 span을 하나 더 추가하는 방법을 보여줍니다. 이 단계에서는 이전 RAG 오퍼레이션에서 검색한 문서 컨텍스트를 바탕으로 LLM이 응답을 생성하는 과정을 모델링합니다. LLM 완료 트레이스는 payload_child_start 객체로 시작하며, 이 객체에는 다음 항목이 포함됩니다.

trace_id: 이 하위 span을 상위 트레이스에 연결합니다.
parent_id: 하위 span을 상위 워크플로에 연결합니다.
inputs: 사용자 쿼리와 추가된 문서 컨텍스트를 포함해 LLM에 전달되는 입력 메시지를 로깅합니다.
model: 오퍼레이션에 사용할 모델(gpt-4o)을 지정합니다.

성공하면 LLM 하위 span 트레이스가 시작되었다는 메시지가 출력됩니다.

Child call started. ID: 0245acdf-83a9-4c90-90df-dcb2b89f234a

오퍼레이션이 완료되면 payload_child_end 객체가 LLM이 생성한 응답을 output 필드에 로깅하고 트레이스를 종료합니다. 또한 사용량 요약 정보도 함께 로깅합니다. 성공하면 LLM 하위 span 트레이스가 시작되고 종료되었음을 알리는 메시지가 출력됩니다.

Child call started. ID: 0245acdf-83a9-4c90-90df-dcb2b89f234a
Child call ended.
python
## ------------
## 자식 span:
## LLM 완성 호출 생성
## ------------

# 자식 호출: 시작
payload_child_start = {
    "start": {
        "project_id": f"{team_id}/{project_id}",
        "op_name": "llm_completion",
        "trace_id": trace_id,
        "parent_id": parent_call_id,
        "started_at": datetime.datetime.now().isoformat(),
        "inputs": {
            "messages": [
                {
                    "role": "user",
                    "content": "With the following document context, could you help me answer:\n Can you summarize the key points of this document?\n [+ appended document context]",
                }
            ],
            "model": "gpt-4o",
        },
        "attributes": {},
    }
}
response = requests.post(
    url_start, headers=headers, json=payload_child_start, auth=("api", wandb_token)
)
if response.status_code == 200:
    data = response.json()
    child_call_id = data.get("id")
    print(f"Child call started. ID: {child_call_id}")
else:
    print("Child start request failed with status:", response.status_code)
    print(response.text)
    exit()

# 자식 호출: 종료
payload_child_end = {
    "end": {
        "project_id": f"{team_id}/{project_id}",
        "id": child_call_id,
        "ended_at": datetime.datetime.now().isoformat(),
        "output": {
            "choices": [
                {"message": {"content": "This is the response generated by the LLM."}},
            ]
        },
        "summary": {
            "usage": {
                "gpt-4o": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30,
                    "requests": 1,
                }
            }
        },
    }
}
response = requests.post(
    url_end, headers=headers, json=payload_child_end, auth=("api", wandb_token)
)
if response.status_code == 200:
    print("Child call ended.")
else:
    print("Child end request failed with status:", response.status_code)
    print(response.text)

복잡한 트레이스 종료

다음 코드는 상위 트레이스를 마무리하여 워크플로가 완료되었음을 표시하는 방법을 보여줍니다. 이 단계에서는 모든 하위 span(예: RAG 조회 및 LLM 완료)의 결과를 집계하고 최종 출력과 메타데이터를 로깅합니다. 트레이스는 payload_parent_end 객체로 마무리되며, 이 객체에는 다음 항목이 포함됩니다.

id: 처음 상위 트레이스를 시작할 때 받은 parent_call_id입니다.
output: 워크플로의 최종 출력입니다.
summary: 워크플로의 사용량 데이터를 종합합니다.
prompt_tokens: 모든 프롬프트에 사용된 총 토큰 수입니다.
completion_tokens: 모든 응답에서 생성된 총 토큰 수입니다.
total_tokens: 워크플로 전체의 토큰 수 합계입니다.
requests: 전송된 총 요청 수입니다(이 예시에서는 1).

성공하면 다음과 같이 출력됩니다.

Parent call ended.
python
## ------------
## 트레이스 종료
## ------------

# 부모 호출: 종료
payload_parent_end = {
    "end": {
        "project_id": f"{team_id}/{project_id}",
        "id": parent_call_id,
        "ended_at": datetime.datetime.now().isoformat(),
        "output": {
            "choices": [
                {"message": {"content": "This is the response generated by the LLM."}},
            ]
        },
        "summary": {
            "usage": {
                "gpt-4o": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30,
                    "requests": 1,
                }
            }
        },
    }
}
response = requests.post(
    url_end, headers=headers, json=payload_parent_end, auth=("api", wandb_token)
)
if response.status_code == 200:
    print("Parent call ended.")
else:
    print("Parent end request failed with status:", response.status_code)
    print(response.text)

조회 쿼리 실행하기

Weave에 트레이스를 로깅했다면 Service API를 사용해 프로그래밍 방식으로 트레이스를 쿼리할 수 있습니다. 다음 코드는 이전 예시에서 생성한 트레이스 중 inputs.model 필드가 gpt-4o인 트레이스만 필터링해 쿼리하는 방법을 보여줍니다. query_payload 객체에는 다음 항목이 포함됩니다.

project_id: 쿼리할 팀과 프로젝트를 식별합니다.
filter: 쿼리가 트레이스 루트(최상위 트레이스)만 반환하도록 합니다.
query: $expr 연산자를 사용해 필터 로직을 정의합니다.
- $getField: inputs.model 필드를 가져옵니다.
- $literal: inputs.model이 "gpt-4o"인 트레이스를 찾습니다.
limit: 쿼리 결과를 최대 10,000개로 제한합니다.
offset: 첫 번째 결과부터 쿼리를 시작합니다.
sort_by: started_at 타임스탬프를 기준으로 결과를 내림차순 정렬합니다.
include_feedback: 결과에서 피드백 데이터를 제외합니다.

쿼리가 성공하면 응답에 쿼리 매개변수와 일치하는 트레이스 데이터가 포함됩니다.

{'id': '01939cf3-541f-76d3-ade3-50cfae068b39', 'project_id': 'cool-new-team/uncategorized', 'op_name': 'simple_trace', 'display_name': None, 'trace_id': '01939cf3-541f-76d3-ade3-50d5cfabe2db', 'parent_id': None, 'started_at': '2024-12-06T17:10:12.590000Z', 'attributes': {}, 'inputs': {'messages': [{'role': 'user', 'content': 'Why is the sky blue?'}], 'model': 'gpt-4o'}, 'ended_at': '2024-12-06T17:47:08.553000Z', 'exception': None, 'output': {'choices': [{'message': {'content': 'It’s due to Rayleigh scattering, where shorter blue wavelengths of sunlight scatter in all directions.'}}]}, 'summary': {'usage': {'gpt-4o': {'prompt_tokens': 10, 'completion_tokens': 20, 'requests': 1, 'total_tokens': 30}}, 'weave': {'status': 'success', 'trace_name': 'simple_trace', 'latency_ms': 2215963}}, 'wb_user_id': 'VXNlcjoyMDk5Njc0', 'wb_run_id': None, 'deleted_at': None}
python
query_payload = {
    "project_id": f"{team_id}/{project_id}",
    "filter": {"trace_roots_only": True},
    "query": {
        "$expr": {"$eq": [{"$getField": "inputs.model"}, {"$literal": "gpt-4o"}]}
    },
    "limit": 10000,
    "offset": 0,
    "sort_by": [{"field": "started_at", "direction": "desc"}],
    "include_feedback": False,
}
response = requests.post(
    url_stream_query, headers=headers, json=query_payload, auth=("api", wandb_token)
)
if response.status_code == 200:
    print("Query successful!")
    try:
        data = response.json()
        print(data)
    except json.JSONDecodeError as e:
        # 대체 방식으로 디코딩
        json_objects = response.text.strip().split("\n")
        parsed_data = [json.loads(obj) for obj in json_objects]
        print(parsed_data)
else:
    print(f"Query failed with status code: {response.status_code}")
    print(response.text)

​사전 요구 사항: 변수 및 엔드포인트 설정

​단순한 트레이스

​단순한 트레이스 시작하기

​단순한 트레이스 종료하기

​복잡한 트레이스

​복잡한 트레이스 시작하기

​RAG 문서 조회용 하위 span 추가

​LLM 완료 호출에 대한 하위 span 추가

​복잡한 트레이스 종료

​조회 쿼리 실행하기

사전 요구 사항: 변수 및 엔드포인트 설정

단순한 트레이스

단순한 트레이스 시작하기

단순한 트레이스 종료하기

복잡한 트레이스

복잡한 트레이스 시작하기

RAG 문서 조회용 하위 span 추가

LLM 완료 호출에 대한 하위 span 추가

복잡한 트레이스 종료

조회 쿼리 실행하기