Documentation Index
Fetch the complete documentation index at: https://translations.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
W&B Weave _Threads_를 사용하면 LLM 애플리케이션에서 멀티 턴 대화를 추적하고 분석할 수 있습니다. Threads는 관련 호출을 공통 thread_id 아래에 그룹화하여 전체 세션을 시각화하고, 턴 단위가 아닌 대화 단위 메트릭을 추적할 수 있도록 합니다. 스레드는 코드에서 생성할 수 있으며, Weave UI에서 시각화할 수 있습니다.
Threads를 사용하려면 다음을 수행하세요:
- Threads의 기본 개념을 먼저 이해하세요.
- 일반적인 사용 패턴과 실제 사용 사례를 보여주는 코드 샘플을 실행해 보세요.
Thread는 다음과 같은 항목을 구성하고 분석할 때 유용합니다:
- 다중 턴 대화
- 세션 기반 워크플로
- 서로 관련된 연산의 일련의 과정
Thread를 사용하면 호출을 컨텍스트별로 그룹화할 수 있어, 여러 단계를 거치며 시스템이 어떻게 응답하는지 더 쉽게 파악할 수 있습니다. 예를 들어 단일 사용자 세션, 에이전트의 의사 결정 체인, 인프라와 비즈니스 로직 계층 전반에 걸친 복잡한 요청을 추적할 수 있습니다.
애플리케이션을 Thread와 turn 단위로 구조화하면 더 깔끔한 지표와 Weave UI에서의 가시성을 확보할 수 있습니다. 모든 저수준 연산을 일일이 보는 대신, 중요한 고수준 단계에 집중할 수 있습니다.
_Thread_는 공통된 대화 맥락을 공유하는 관련 호출들을 논리적으로 묶은 것입니다. Thread는 다음과 같은 특징이 있습니다.
- 고유한
thread_id를 가집니다
- 하나 이상의 _turns_를 포함합니다
- 호출 간 맥락을 유지합니다
- 전체 사용자 세션 또는 상호작용 흐름을 나타냅니다
_Turn_은 Thread 내의 상위 수준 작업으로, UI의 thread view에서 개별 행으로 표시됩니다. 각 Turn은 다음과 같습니다.
- 대화 또는 워크플로우에서 하나의 논리적 단계를 나타냅니다.
- Thread 컨텍스트의 직접적인 자식이며, 중첩된 더 하위 수준의 call을 포함할 수 있습니다(이러한 call은 thread 수준 통계에는 표시되지 않습니다).
_Call_은 애플리케이션에서 @weave.op로 데코레이션된 함수가 실행된 모든 경우를 의미합니다.
- _Turn calls_는 새로운 턴을 시작하는 최상위 연산입니다.
- _Nested calls_는 하나의 턴 내에 존재하는 하위 수준 연산입니다.
_Trace_는 단일 연산에 대한 전체 호출 스택을 캡처합니다. Thread는 동일한 논리적 대화 또는 세션에 속하는 여러 Trace를 함께 그룹화합니다. 즉, 하나의 Thread는 여러 번의 턴(turn)으로 구성되며, 각 턴은 대화의 한 부분을 나타냅니다. Trace에 대한 자세한 내용은 Tracing 개요를 참고하세요.
Weave 사이드바에서 Threads를 선택하여 Threads 목록 보기로 이동하세요.
- 현재 프로젝트의 최근 Threads를 표시합니다
- 열에는 턴 수, 시작 시간, 마지막 업데이트 시간이 포함됩니다
- 행을 클릭하여 상세 드로어를 엽니다
- 아무 행이나 클릭하면 해당 행의 상세 드로어가 열립니다.
- 하나의 스레드 내 모든 턴을 표시합니다.
- 턴은 시작된 순서대로 나열됩니다(지속 시간이나 종료 시간이 아니라 시작 시간을 기준으로 정렬).
- 콜 단위 메타데이터(지연 시간, 입력, 출력)를 포함합니다.
- 로그된 경우, 선택적으로 메시지 내용 또는 구조화된 데이터를 표시합니다.
- 턴의 전체 실행 내용을 보려면 스레드 상세 드로어에서 해당 턴을 열 수 있습니다. 이렇게 하면 해당 턴에서 발생한 모든 중첩 연산을 자세히 살펴볼 수 있습니다.
- 턴에 LLM 호출에서 추출된 메시지가 포함된 경우, 오른쪽 채팅 패널에 표시됩니다. 이 메시지는 일반적으로 지원되는 통합(
openai.ChatCompletion.create 등)에서 수행된 호출에서 오며, 표시되기 위해 특정 조건을 충족해야 합니다. 자세한 내용은 채팅 뷰 동작을 참조하세요.
채팅 패널에는 각 턴마다 수행된 LLM 호출에서 추출한 구조화된 메시지 데이터가 표시됩니다. 이 뷰는 해당 상호작용을 대화형으로 표현해 보여줍니다.
메시지는 한 턴(turn) 내에서 LLM 제공자와의 직접적인 상호작용(예: 프롬프트를 보내고 응답을 받는 것)을 나타내는 호출에서 추출됩니다. 다른 호출 안에 중첩되어 있지 않은 호출만 메시지로 표시됩니다. 이렇게 하면 중간 단계나 합쳐진 내부 로직이 중복해서 표시되는 것을 방지할 수 있습니다.
일반적으로 메시지는 다음과 같이 자동 패치된 서드파티 SDK에서 생성됩니다:
openai.ChatCompletion.createanthropic.Anthropic.completion
어떤 턴에서 메시지가 전혀 생성되지 않으면, 해당 턴에 대해 채팅 창에는 비어 있는 메시지 영역이 표시됩니다. 하지만 채팅 창에는 여전히 같은 스레드의 다른 턴에서 나온 메시지들이 표시될 수 있습니다.
- 턴을 클릭하면 채팅 창이 해당 턴의 메시지 위치로 스크롤되며 고정(pinning)됩니다.
- 채팅 창을 스크롤하면 왼쪽 목록의 해당 턴이 강조 표시됩니다.
턴의 전체 트레이스를 열려면 해당 턴을 클릭합니다.
스레드 상세 보기로 돌아갈 수 있도록 왼쪽 상단에 뒤로 가기 버튼이 표시됩니다. 전환 시 스크롤 위치와 같은 UI 상태는 유지되지 않습니다.
이 섹션의 각 예제는 애플리케이션에서 턴(turn)과 스레드(thread)를 구성하는 다양한 전략을 보여줍니다. 대부분의 예제에서는 스텁 함수 내부에 직접 LLM 호출이나 시스템 동작을 구현해야 합니다.
- 세션이나 대화를 추적하려면
weave.thread() 컨텍스트 매니저를 사용하십시오.
- 논리 연산을
@weave.op으로 데코레이트하여 턴 또는 중첩 호출로 추적하십시오.
thread_id를 전달하면 Weave는 해당 블록의 모든 연산을 동일한 스레드로 묶어서 관리합니다. thread_id를 생략하면 Weave가 자동으로 고유한 값을 생성합니다.
weave.thread()의 반환 값은 thread_id 속성을 가진 ThreadContext 객체이며, 이를 로그로 남기거나 재사용하거나 다른 시스템에 전달할 수 있습니다.
중첩된 weave.thread() 컨텍스트는 동일한 thread_id를 재사용하지 않는 한 항상 새 스레드를 시작합니다. 하위 컨텍스트를 종료해도 상위 컨텍스트는 중단되거나 덮어써지지 않습니다. 이를 통해 애플리케이션 로직에 따라 분기된 스레드 구조나 계층형 스레드 오케스트레이션을 구현할 수 있습니다.
다음 코드는 weave.thread()를 사용하여 하나 이상의 연산을 공통 thread_id로 묶는 방법을 보여줍니다. 이는 애플리케이션에서 Threads 사용을 시작하는 가장 간단한 방법입니다.
import weave
@weave.op
def say_hello(name: str) -> str:
return f"Hello, {name}!"
# 새 스레드 컨텍스트 시작
with weave.thread() as thread_ctx:
print(f"Thread ID: {thread_ctx.thread_id}")
say_hello("Bill Nye the Science Guy")
@weave.op 데코레이터와 weave.thread() 컨텍스트 관리를 사용해 대화형 에이전트를 수동으로 정의하는 방법을 보여줍니다. process_user_message를 호출할 때마다 스레드에 새로운 턴이 생성됩니다. 직접 에이전트 루프를 구현하고 컨텍스트와 중첩 처리 방식을 완전히 제어하고 싶을 때 이 패턴을 사용할 수 있습니다.
단기 상호작용에는 자동 생성된 스레드 ID를 사용하고, 세션 간에 스레드 컨텍스트를 유지하려면 사용자 정의 세션 ID(예: user_session_123)를 전달하세요.
import weave
class ConversationAgent:
@weave.op
def process_user_message(self, message: str) -> str:
"""
턴 수준 작업: 하나의 대화 턴을 나타냅니다.
스레드 통계에는 이 함수만 집계됩니다.
"""
# 사용자 메시지 저장
# 중첩 호출을 통해 AI 응답 생성
response = self._generate_response(message)
# 어시스턴트 응답 저장
return response
@weave.op
def _generate_response(self, message: str) -> str:
"""중첩 호출: 구현 세부 사항으로, 스레드 통계에 집계되지 않습니다."""
context = self._retrieve_context(message) # 또 다른 중첩 호출
intent = self._classify_intent(message) # 또 다른 중첩 호출
response = self._call_llm(message, context) # LLM 호출 (중첩)
return self._format_response(response) # 최종 중첩 호출
@weave.op
def _retrieve_context(self, message: str) -> str:
# 벡터 DB 조회, 지식 베이스 쿼리 등
return "retrieved_context"
@weave.op
def _classify_intent(self, message: str) -> str:
# 의도 분류 로직
return "general_inquiry"
@weave.op
def _call_llm(self, message: str, context: str) -> str:
# OpenAI/Anthropic 등 API 호출
return "llm_response"
@weave.op
def _format_response(self, response: str) -> str:
# 응답 포맷팅 로직
return f"Formatted: {response}"
# 사용법: 스레드 컨텍스트가 자동으로 설정됨
agent = ConversationAgent()
# 스레드 컨텍스트 설정 - process_user_message 호출마다 하나의 턴이 됨
with weave.thread() as thread_ctx: # thread_id 자동 생성
print(f"Thread ID: {thread_ctx.thread_id}")
# process_user_message 호출마다 1개의 턴 + 여러 중첩 호출 생성
agent.process_user_message("Hello, help with setup") # 턴 1
agent.process_user_message("What languages do you recommend?") # 턴 2
agent.process_user_message("Explain Python vs JavaScript") # 턴 3
# 결과: 3개의 턴을 가진 스레드, 총 약 15~20개의 호출 (중첩 포함)
# 대안: 세션 추적을 위해 명시적 thread_id 사용
session_id = "user_session_123"
with weave.thread(session_id) as thread_ctx:
print(f"Session Thread ID: {thread_ctx.thread_id}") # "user_session_123"
agent.process_user_message("Continue our previous conversation") # 이 세션의 턴 1
agent.process_user_message("Can you summarize what we discussed?") # 이 세션의 턴 2
thread_id를 공유하지만, 턴 경계는 프로바이더 로직에 따라 스택의 서로 다른 수준에 나타납니다. 이는 서로 다른 백엔드에 대해 호출을 각기 다르게 추적해야 하지만, 여전히 동일한 스레드로 묶어야 할 때 유용합니다.
import weave
import random
import asyncio
class OpenAIProvider:
"""OpenAI 브랜치: 전환 경계까지 2단계 깊이의 호출 체인"""
@weave.op
def route_to_openai(self, user_input: str, thread_id: str) -> str:
"""Level 1: OpenAI 요청 라우팅 및 준비"""
# 입력 유효성 검사, 라우팅 로직, 기본 전처리
print(f" L1: Routing to OpenAI for: {user_input}")
# 전환 경계 - 스레드 컨텍스트로 래핑
with weave.thread(thread_id):
# Level 2 직접 호출 - 호출 체인 깊이 생성
return self.execute_openai_call(user_input)
@weave.op
def execute_openai_call(self, user_input: str) -> str:
"""Level 2: 전환 경계 - OpenAI API 호출 실행"""
print(f" L2: Executing OpenAI API call")
response = f"OpenAI GPT-4 response: {user_input}"
return response
class AnthropicProvider:
"""Anthropic 브랜치: 전환 경계까지 3단계 깊이의 호출 체인"""
@weave.op
def route_to_anthropic(self, user_input: str, thread_id: str) -> str:
"""Level 1: Anthropic 요청 라우팅 및 준비"""
# 입력 유효성 검사, 라우팅 로직, 공급자 선택
print(f" L1: Routing to Anthropic for: {user_input}")
# Level 2 호출 - 호출 체인 깊이 생성
return self.authenticate_anthropic(user_input, thread_id)
@weave.op
def authenticate_anthropic(self, user_input: str, thread_id: str) -> str:
"""Level 2: Anthropic 인증 및 설정 처리"""
print(f" L2: Authenticating with Anthropic")
# 인증, 속도 제한, 세션 관리
auth_token = "anthropic_key_xyz_authenticated"
# 전환 경계 - Level 3에서 스레드 컨텍스트로 래핑
with weave.thread(thread_id):
# Level 3 호출 - 호출 체인 추가 중첩
return self.execute_anthropic_call(user_input, auth_token)
@weave.op
def execute_anthropic_call(self, user_input: str, auth_token: str) -> str:
"""Level 3: 전환 경계 - Anthropic API 호출 실행"""
print(f" L3: Executing Anthropic API call with auth")
response = f"Anthropic Claude response (auth: {auth_token[:15]}...): {user_input}"
return response
class MultiProviderAgent:
"""서로 다른 호출 체인 깊이를 가진 공급자 간에 라우팅하는 메인 에이전트"""
def __init__(self):
self.openai_provider = OpenAIProvider()
self.anthropic_provider = AnthropicProvider()
def handle_conversation_turn(self, user_input: str, thread_id: str) -> str:
"""
호출 체인 깊이가 다른 공급자로 라우팅합니다.
스레드 컨텍스트는 각 체인의 서로 다른 중첩 레벨에서 적용됩니다.
"""
# 데모용으로 공급자를 무작위 선택
use_openai = random.choice([True, False])
if use_openai:
print(f"Choosing OpenAI (2-level call chain)")
# OpenAI: Level 1 → Level 2 (전환 경계)
response = self.openai_provider.route_to_openai(user_input, thread_id)
return f"[OpenAI Branch] {response}"
else:
print(f"Choosing Anthropic (3-level call chain)")
# Anthropic: Level 1 → Level 2 → Level 3 (전환 경계)
response = self.anthropic_provider.route_to_anthropic(user_input, thread_id)
return f"[Anthropic Branch] {response}"
async def main():
agent = MultiProviderAgent()
conversation_id = "nested_depth_conversation_999"
# 호출 체인 깊이가 다른 멀티턴 대화
conversation_turns = [
"What's deep learning?",
"Explain neural network backpropagation",
"How do attention mechanisms work?",
"What's the transformer architecture?",
"Compare CNNs vs RNNs"
]
print(f"Starting conversation: {conversation_id}")
for i, user_input in enumerate(conversation_turns, 1):
print(f"\\n--- Turn {i} ---")
print(f"User: {user_input}")
# 서로 다른 호출 체인 깊이에서 동일한 thread_id 사용
response = agent.handle_conversation_turn(user_input, conversation_id)
print(f"Agent: {response}")
if __name__ == "__main__":
asyncio.run(main())
# 예상 결과: 5개의 턴을 가진 단일 스레드
# - OpenAI 턴: 호출 체인의 Level 2에서 스레드 컨텍스트
# 호출 스택: route_to_openai() → execute_openai_call() ← 여기에 스레드 컨텍스트
# - Anthropic 턴: 호출 체인의 Level 3에서 스레드 컨텍스트
# 호출 스택: route_to_anthropic() → authenticate_anthropic() → execute_anthropic_call() ← 여기에 스레드 컨텍스트
# - 모든 턴이 thread_id 공유: "nested_depth_conversation_999"
# - 서로 다른 호출 스택 깊이에서 표시된 전환 경계
# - 호출 체인의 보조 작업은 턴이 아닌 중첩 호출로 추적
thread_id 파라미터를 None으로 두지 마십시오. 이렇게 하면 스레드 그룹화가 완전히 비활성화됩니다. 대신 항상 유효한 스레드 ID를 제공해야 합니다. 새 스레드를 만들어야 한다면 generate_id()와 같은 함수를 사용해 고유한 식별자를 생성하십시오.
thread_id가 지정되지 않으면 Weave의 내부 구현은 랜덤 UUID v7을 자동으로 생성합니다. 이 동작을 직접 구현한 generate_id() 함수에서 동일하게 구현하거나, 원하는 임의의 고유 문자열 값을 사용할 수 있습니다.
import weave
import uuidv7
import argparse
def generate_id():
"""UUID v7을 사용하여 고유한 스레드 ID를 생성합니다."""
return str(uuidv7.uuidv7())
@weave.op
def load_history(session_id):
"""주어진 세션의 대화 기록을 불러옵니다."""
# 구현 내용을 여기에 작성하세요
return []
# 세션 재개를 위한 커맨드라인 인수 파싱
parser = argparse.ArgumentParser()
parser.add_argument("--session-id", help="재개할 기존 세션 ID")
args = parser.parse_args()
# 스레드 ID 결정: 기존 세션 재개 또는 새 세션 생성
if args.session_id:
thread_id = args.session_id
print(f"세션 재개 중: {thread_id}")
else:
thread_id = generate_id()
print(f"새 세션 시작 중: {thread_id}")
# 호출 추적을 위한 스레드 컨텍스트 설정
with weave.thread(thread_id) as thread_ctx:
# 대화 기록 불러오기 또는 초기화
history = load_history(thread_id)
print(f"활성 스레드 ID: {thread_ctx.thread_id}")
# 애플리케이션 로직을 여기에 작성하세요...
ThreadContext를 사용해 스레드 ID를 설정하여 이러한 계층을 조율합니다. 시스템의 각 부분을 독립적으로 분석하거나 모니터링하면서도 공유 세션에 연결해 두고 싶을 때 이 패턴을 사용하십시오.
import weave
from contextlib import contextmanager
from typing import Dict
# 중첩 스레드 조율을 위한 전역 스레드 컨텍스트
class ThreadContext:
def __init__(self):
self.app_thread_id = None
self.infra_thread_id = None
self.logic_thread_id = None
def setup_for_request(self, request_id: str):
self.app_thread_id = f"app_{request_id}"
self.infra_thread_id = f"{self.app_thread_id}_infra"
self.logic_thread_id = f"{self.app_thread_id}_logic"
# 전역 인스턴스
thread_ctx = ThreadContext()
class InfrastructureLayer:
"""전용 스레드에서 모든 인프라 작업을 처리합니다"""
@weave.op
def authenticate_user(self, user_id: str) -> Dict:
# 인증 로직...
return {"user_id": user_id, "authenticated": True}
@weave.op
def call_payment_gateway(self, amount: float) -> Dict:
# 결제 처리...
return {"status": "approved", "amount": amount}
@weave.op
def update_inventory(self, product_id: str, quantity: int) -> Dict:
# 재고 관리...
return {"product_id": product_id, "updated": True}
def execute_operations(self, user_id: str, order_data: Dict) -> Dict:
"""전용 스레드 컨텍스트에서 모든 인프라 작업을 실행합니다"""
with weave.thread(thread_ctx.infra_thread_id):
auth_result = self.authenticate_user(user_id)
payment_result = self.call_payment_gateway(order_data["amount"])
inventory_result = self.update_inventory(order_data["product_id"], order_data["quantity"])
return {
"auth": auth_result,
"payment": payment_result,
"inventory": inventory_result
}
class BusinessLogicLayer:
"""전용 스레드에서 비즈니스 로직을 처리합니다"""
@weave.op
def validate_order(self, order_data: Dict) -> Dict:
# 유효성 검사 로직...
return {"valid": True}
@weave.op
def calculate_pricing(self, order_data: Dict) -> Dict:
# 가격 계산...
return {"total": order_data["amount"], "tax": order_data["amount"] * 0.08}
@weave.op
def apply_business_rules(self, order_data: Dict) -> Dict:
# 비즈니스 규칙...
return {"rules_applied": ["standard_processing"], "priority": "normal"}
def execute_logic(self, order_data: Dict) -> Dict:
"""전용 스레드 컨텍스트에서 모든 비즈니스 로직을 실행합니다"""
with weave.thread(thread_ctx.logic_thread_id):
validation = self.validate_order(order_data)
pricing = self.calculate_pricing(order_data)
rules = self.apply_business_rules(order_data)
return {"validation": validation, "pricing": pricing, "rules": rules}
class OrderProcessingApp:
"""메인 애플리케이션 오케스트레이터"""
def __init__(self):
self.infra = InfrastructureLayer()
self.business = BusinessLogicLayer()
@weave.op
def process_order(self, user_id: str, order_data: Dict) -> Dict:
"""메인 주문 처리 - 앱 스레드의 턴이 됩니다"""
# 전용 스레드에서 중첩 작업 실행
infra_results = self.infra.execute_operations(user_id, order_data)
logic_results = self.business.execute_logic(order_data)
# 최종 오케스트레이션
return {
"order_id": f"order_12345",
"status": "completed",
"infra_results": infra_results,
"logic_results": logic_results
}
# 전역 스레드 컨텍스트 조율을 활용한 사용 예시
def handle_order_request(request_id: str, user_id: str, order_data: Dict):
# 해당 요청에 대한 스레드 컨텍스트 설정
thread_ctx.setup_for_request(request_id)
# 앱 스레드 컨텍스트에서 실행
with weave.thread(thread_ctx.app_thread_id):
app = OrderProcessingApp()
result = app.process_order(user_id, order_data)
return result
# 사용 예시
order_result = handle_order_request(
request_id="req_789",
user_id="user_001",
order_data={"product_id": "laptop", "quantity": 1, "amount": 1299.99}
)
# 예상 스레드 구조:
#
# 앱 스레드: app_req_789
# └── 턴: process_order() ← 메인 오케스트레이션
#
# 인프라 스레드: app_req_789_infra
# ├── 턴: authenticate_user() ← 인프라 작업 1
# ├── 턴: call_payment_gateway() ← 인프라 작업 2
# └── 턴: update_inventory() ← 인프라 작업 3
#
# 로직 스레드: app_req_789_logic
# ├── 턴: validate_order() ← 비즈니스 로직 작업 1
# ├── 턴: calculate_pricing() ← 비즈니스 로직 작업 2
# └── 턴: apply_business_rules() ← 비즈니스 로직 작업 3
#
# 장점:
# - 스레드 간 명확한 관심사 분리
# - 스레드 ID를 파라미터로 전달할 필요 없음
# - 앱/인프라/로직 레이어의 독립적인 모니터링
# - 스레드 컨텍스트를 통한 전역 조율
POST /threads/query
class ThreadsQueryReq:
project_id: str
limit: Optional[int] = None
offset: Optional[int] = None
sort_by: Optional[list[SortBy]] = None # 지원 필드: thread_id, turn_count, start_time, last_updated
sortable_datetime_after: Optional[datetime] = None # 세분화 최적화를 통한 스레드 필터링
sortable_datetime_before: Optional[datetime] = None # 세분화 최적화를 통한 스레드 필터링
class ThreadSchema:
thread_id: str # 스레드의 고유 식별자
turn_count: int # 이 스레드의 turn 호출 횟수
start_time: datetime # 이 스레드에서 turn 호출의 가장 이른 시작 시간
last_updated: datetime # 이 스레드에서 turn 호출의 가장 늦은 종료 시간
class ThreadsQueryRes:
threads: List[ThreadSchema]
my-project를 실제 프로젝트 ID로 바꾸세요.
# 가장 최근에 활성화된 스레드 가져오기
response = client.threads_query(ThreadsQueryReq(
project_id="my-project",
sort_by=[SortBy(field="last_updated", direction="desc")],
limit=50
))
for thread in response.threads:
print(f"Thread {thread.thread_id}: {thread.turn_count} turns, last active {thread.last_updated}")
# 가장 활발한 스레드 가져오기 (턴 수 기준)
response = client.threads_query(ThreadsQueryReq(
project_id="my-project",
sort_by=[SortBy(field="turn_count", direction="desc")],
limit=20
))
timedelta의 days 값을 조정해 시간 범위를 변경할 수 있습니다.
from datetime import datetime, timedelta
# 지난 24시간 내에 시작된 스레드 가져오기
yesterday = datetime.now() - timedelta(days=1)
response = client.threads_query(ThreadsQueryReq(
project_id="my-project",
sortable_datetime_after=yesterday,
sort_by=[SortBy(field="start_time", direction="desc")]
))
플랫폼
W&B Models
W&B Weave
W&B Inference
W&B 교육
릴리스 노트
