메인 콘텐츠로 건너뛰기
이 노트북은 대화형입니다. 로컬에서 실행하거나 아래 링크를 사용하세요:

​
Weave와 통합하기: 프로덕션 대시보드

GenAI 도구 생태계는 빠르게 변화하고 있으며, 새로운 프레임워크, 도구, 애플리케이션이 계속해서 등장하고 있습니다. Weave는 GenAI 모니터링과 평가를 위한 원스톱 솔루션을 목표로 합니다. 이는 때때로 기존 플랫폼과 통합하거나 프로젝트나 조직의 특정 요구사항에 맞게 Weave를 확장해야 함을 의미하기도 합니다. 이 쿡북에서는 Weave의 강력한 API와 함수를 활용하여, Weave의 Traces 뷰를 확장하는 형태로 프로덕션 모니터링용 커스텀 대시보드를 만드는 방법을 보여줍니다. 여기에서는 다음에 중점을 둡니다:
  • Weave에서 traces, 비용, 피드백 및 기타 지표 가져오기
  • 사용자 피드백과 비용 분포에 대한 집계 보기 생성
  • 시간에 따른 토큰 사용량과 지연 시간 시각화 생성
streamlit을 설치하고 이 프로덕션 대시보드 스크립트를 실행하여, 자신의 Weave 프로젝트에서 이 대시보드를 바로 사용해 볼 수 있습니다! Weave를 사용한 프로덕션 대시보드 예시

​
1. 설정

이 튜토리얼을 따라 하려면 다음 패키지들만 설치하면 됩니다: 
!pip install streamlit pandas plotly weave

​
2. 구현

​
2.1 Weave 클라이언트 초기화 및 비용 정의

먼저, Weave 클라이언트를 초기화하고 각 모델에 대한 비용을 추가하는 함수를 정의합니다.
  • 여러 표준 모델에 대한 기본 비용을 이미 포함해 두었지만, 사용자가 직접 커스텀 비용과 커스텀 모델을 쉽게 추가할 수 있도록 설계되어 있습니다. 아래에서는 일부 모델에 커스텀 비용을 추가하고, 나머지 모델에는 기본 비용을 사용하는 방법을 보여줍니다.
  • 비용은 Weave에서 각 호출마다 추적되는 토큰 수를 기반으로 계산됩니다. 많은 LLM 벤더 라이브러리의 경우 토큰 사용량을 자동으로 추적하지만, 어떤 호출이든 커스텀 토큰 수를 직접 반환할 수도 있습니다. 커스텀 모델에 대한 토큰 수와 비용 계산을 정의하는 방법은 이 쿡북을 참고하세요 - custom cost cookbook.
PROJECT_NAME = "wandb-smle/weave-cookboook-demo"
python
import weave

MODEL_NAMES = [
    # 모델 이름, 프롬프트 비용, 완성 비용
    ("gpt-4o-2024-05-13", 0.03, 0.06),
    ("gpt-4o-mini-2024-07-18", 0.03, 0.06),
    ("gemini/gemini-1.5-flash", 0.00025, 0.0005),
    ("gpt-4o-mini", 0.03, 0.06),
    ("gpt-4-turbo", 0.03, 0.06),
    ("claude-3-haiku-20240307", 0.01, 0.03),
    ("gpt-4o", 0.03, 0.06),
]

def init_weave_client(project_name):
    try:
        client = weave.init(project_name)
        for model, prompt_cost, completion_cost in MODEL_NAMES:
            client.add_cost(
                llm_id=model,
                prompt_token_cost=prompt_cost,
                completion_token_cost=completion_cost,
            )
    except Exception as e:
        print(f"프로젝트 '{project_name}'에 대한 Weave 클라이언트 초기화 실패: {e}")
        return None
    else:
        return client

client = init_weave_client(PROJECT_NAME)

​
2.2 Weave에서 호출 데이터 가져오기

Weave에서 호출 데이터를 가져오려면 두 가지 방법이 있습니다:
  1. 호출별로 데이터 가져오기
  2. 고수준 API 사용

​
2.2.1 호출 단위로 데이터 가져오기

Weave에서 데이터에 접근하는 첫 번째 방법은 필터링된 호출 목록을 가져온 뒤 필요한 데이터를 호출 단위로 추출하는 것입니다. 이를 위해 calls_query_stream API를 사용해 Weave에서 호출 데이터를 가져올 수 있습니다:
  • calls_query_stream API: 이 API를 사용하면 Weave에서 호출 데이터를 가져올 수 있습니다.
  • filter dictionary: 이 dictionary에는 호출 데이터를 가져오기 위한 필터 매개변수가 포함됩니다. 자세한 내용은 여기를 참고하세요.
  • expand_columns list: 이 list에는 호출 데이터에서 확장할 컬럼이 포함됩니다.
  • sort_by list: 이 list에는 호출 데이터를 정렬하기 위한 매개변수가 포함됩니다.
  • include_costs boolean: 이 boolean은 호출 데이터에 비용 정보를 포함할지 여부를 나타냅니다.
  • include_feedback boolean: 이 boolean은 호출 데이터에 피드백 정보를 포함할지 여부를 나타냅니다.
import itertools
from datetime import datetime, timedelta

import pandas as pd

def fetch_calls(client, project_id, start_time, trace_roots_only, limit):
    filter_params = {
        "project_id": project_id,
        "filter": {"started_at": start_time, "trace_roots_only": trace_roots_only},
        "expand_columns": ["inputs.example", "inputs.model"],
        "sort_by": [{"field": "started_at", "direction": "desc"}],
        "include_costs": True,
        "include_feedback": True,
    }
    try:
        calls_stream = client.server.calls_query_stream(filter_params)
        calls = list(
            itertools.islice(calls_stream, limit)
        )  # calls 수가 너무 많을 경우 가져올 수를 제한
        print(f"Fetched {len(calls)} calls.")
    except Exception as e:
        print(f"Error fetching calls: {e}")
        return []
    else:
        return calls

calls = fetch_calls(client, PROJECT_NAME, datetime.now() - timedelta(days=1), True, 100)
python
# 원시 데이터는 Call 객체의 리스트입니다
pd.DataFrame([call.dict() for call in calls]).head(3)
Weave에서 반환된 결과를 사용하면 호출을 처리하는 작업은 매우 쉽습니다. 관련 정보를 추출해 딕셔너리 리스트에 저장한 다음, 이 리스트를 pandas DataFrame으로 변환하여 반환합니다. 
import json
from datetime import datetime

import pandas as pd

def process_calls(calls):
    records = []
    for call in calls:
        feedback = call.summary.get("weave", {}).get("feedback", [])
        thumbs_up = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👍"
        )
        thumbs_down = sum(
            1
            for item in feedback
            if isinstance(item, dict) and item.get("payload", {}).get("emoji") == "👎"
        )
        latency = call.summary.get("weave", {}).get("latency_ms", 0)

        records.append(
            {
                "Call ID": call.id,
                "Trace ID": call.trace_id,  # 트레이스를 검색하는 데 사용할 수 있는 고유 ID
                "Display Name": call.display_name,  # UI 또는 프로그래밍 방식으로 설정할 수 있는 선택 사항 이름
                "Latency (ms)": latency,
                "Thumbs Up": thumbs_up,
                "Thumbs Down": thumbs_down,
                "Started At": pd.to_datetime(getattr(call, "started_at", datetime.min)),
                "Inputs": json.dumps(call.inputs, default=str),
                "Outputs": json.dumps(call.output, default=str),
            }
        )
    return pd.DataFrame(records)
python
df_calls = process_calls(calls)
df_calls.head(3)

​
2.2.2 고수준 API 사용

모든 호출을 일일이 살펴보는 대신, Weave는 모델 비용, 피드백, 기타 지표에 직접 접근할 수 있는 고수준 API도 제공합니다. 예를 들어 비용의 경우, query_costs API를 사용하여 프로젝트에서 사용 중인 모든 LLM의 비용을 조회합니다: 
# 비용 API를 사용하여 비용 가져오기
costs = client.query_costs()
df_costs = pd.DataFrame([cost.dict() for cost in costs])
df_costs["total_cost"] = (
    df_costs["prompt_token_cost"] + df_costs["completion_token_cost"]
)

# 고유한 llm_id마다 첫 번째 행만 표시
df_costs

​
2.4 입력 수집 및 시각화 생성

이제 Plotly를 사용해 시각화를 생성해 보겠습니다. 이는 가장 기본적인 대시보드이지만, 원하는 대로 자유롭게 커스터마이즈할 수 있습니다! 더 복잡한 예시는 여기에 있는 Streamlit 예제를 참고하세요. 
import plotly.express as px
import plotly.graph_objects as go

def plot_feedback_pie_chart(thumbs_up, thumbs_down):
    fig = go.Figure(
        data=[
            go.Pie(
                labels=["Thumbs Up", "Thumbs Down"],
                values=[thumbs_up, thumbs_down],
                marker={"colors": ["#66b3ff", "#ff9999"]},
                hole=0.3,
            )
        ]
    )
    fig.update_traces(textinfo="percent+label", hoverinfo="label+percent")
    fig.update_layout(showlegend=False, title="Feedback Summary")
    return fig

def plot_model_cost_distribution(df):
    fig = px.bar(
        df,
        x="llm_id",
        y="total_cost",
        color="llm_id",
        title="Cost Distribution by Model",
    )
    fig.update_layout(xaxis_title="Model", yaxis_title="Cost (USD)")
    return fig

# 모든 플롯의 소스 코드를 참조하세요
python
plot_feedback_pie_chart(df_calls["Thumbs Up"].sum(), df_calls["Thumbs Down"].sum())
python
plot_model_cost_distribution(df_costs)

​
결론

이 쿠크북에서는 Weave의 API와 함수를 사용해 커스텀 프로덕션 모니터링 대시보드를 만드는 방법을 보여주었습니다. 현재 Weave는 데이터를 쉽게 입력하고 커스텀 프로세스를 위해 데이터를 추출할 수 있도록, 빠른 통합을 우선시합니다.
  • 데이터 입력:
    • 프레임워크에 구애받지 않는 트레이싱: @weave-op() 데코레이터를 사용하거나 CSV에서 호출을 가져올 수 있습니다(관련 import 쿠크북 참고).
    • 다양한 프로그래밍 프레임워크와 언어에서 Weave로 로그를 전송할 수 있는 서비스 API 엔드포인트. 자세한 내용은 여기를 참고하세요.
  • 데이터 출력:
    • CSV, TSV, JSONL, JSON 형식으로 데이터를 쉽게 다운로드할 수 있습니다. 자세한 내용은 여기를 참고하세요.
    • 데이터에 대한 프로그래밍 방식 접근을 통해 손쉽게 내보낼 수 있습니다. 이 쿠크북에서 설명한 것처럼 export 패널의 “Use Python” 섹션을 참고하세요. 자세한 내용은 여기를 참고하세요.
이 커스텀 대시보드는 Weave의 기본 Traces 뷰를 확장하여, 프로덕션 환경에서 LLM 애플리케이션을 맞춤형으로 모니터링할 수 있게 해 줍니다. 더 복잡한 대시보드를 보고 싶다면, 자신만의 Weave 프로젝트 URL을 추가할 수 있는 Streamlit 예제를 이 리포지토리에서 확인해 보세요.