스레드 트레이스

W&B Weave _Threads_를 사용하면 LLM 애플리케이션의 멀티턴 대화를 추적하고 분석할 수 있습니다. Threads는 관련된 Calls를 공유된 thread_id로 묶으므로, 전체 세션을 시각화하고 각 턴에 걸친 대화 수준 메트릭을 추적할 수 있습니다. 스레드는 프로그래밍 방식으로 생성하고 Weave UI에서 시각화할 수 있습니다. Threads를 시작하려면 다음을 수행하세요:

Threads의 기본 사항을 먼저 익힙니다.
일반적인 사용 패턴과 실제 사용 사례를 보여주는 코드 샘플을 사용해 보세요.
- 기본 사용 예시

고급 활용 예시

사용 사례

다음과 같은 항목을 정리하고 분석하려는 경우 스레드가 유용합니다.

멀티턴 대화
세션 기반 워크플로
서로 관련된 오퍼레이션의 연속.

스레드를 사용하면 컨텍스트별로 Calls를 그룹화할 수 있어, 여러 step에 걸쳐 시스템이 어떻게 응답하는지 더 쉽게 이해할 수 있습니다. 예를 들어, 단일 사용자 세션, 에이전트의 의사결정 체인, 또는 인프라와 비즈니스 로직 계층에 걸친 복잡한 요청을 추적할 수 있습니다. 애플리케이션을 스레드와 턴으로 구조화하면 더 깔끔한 메트릭을 확보하고 Weave UI에서 더 나은 가시성을 얻을 수 있습니다. 모든 저수준 Op를 보는 대신, 중요한 상위 수준 step에 집중할 수 있습니다.

정의

스레드

_스레드(Thread)_는 공통된 대화 맥락을 공유하는 관련된 Call을 논리적으로 묶은 단위입니다. Thread는 다음과 같은 특징이 있습니다.

고유한 thread_id를 가집니다
하나 이상의 _turn_을 포함합니다
여러 Call에 걸쳐 맥락을 유지합니다
전체 사용자 세션 또는 상호작용 흐름을 나타냅니다

Turn

_Turn_은 Thread 내의 상위 수준 오퍼레이션으로, UI에서는 thread 뷰에서 개별 행으로 표시됩니다. 각 Turn은 다음과 같습니다.

대화 또는 워크플로에서 하나의 논리적 step을 나타냅니다
Turn은 thread 컨텍스트의 직접적인 하위 요소이며, 더 낮은 수준의 중첩된 call을 포함할 수 있습니다(thread 수준 통계에는 표시되지 않음).

Call

_Call_은 애플리케이션에서 @weave.op로 데코레이션된 함수의 실행을 의미합니다.

_Turn Calls_는 새로운 turn을 시작하는 최상위 오퍼레이션입니다.
_Nested Calls_는 turn 내의 하위 오퍼레이션입니다.

트레이스

_트레이스_는 단일 오퍼레이션의 전체 call 스택을 캡처합니다. 스레드는 동일한 논리적 대화나 세션에 속하는 트레이스를 함께 묶습니다. 다시 말해, 스레드는 여러 턴으로 구성되며 각 턴은 대화의 한 부분을 나타냅니다. Traces에 대한 자세한 내용은 Tracing overview를 참조하세요.

UI Overview

Weave 사이드바에서 Threads를 선택해 Threads list view로 이동합니다.

Threads 목록 뷰

프로젝트의 최근 Threads를 표시합니다
열에는 턴 수, 시작 시간, 마지막 업데이트 시간이 표시됩니다
행을 클릭하면 상세 드로어가 열립니다

스레드 세부 정보 드로어

아무 행이나 클릭하면 해당 행의 세부 정보 드로어가 열립니다.
스레드 내의 모든 턴을 표시합니다.
턴은 시작한 순서대로 나열됩니다(지속 시간이나 종료 시간이 아니라 시작 시간을 기준으로 함).
Call 수준 메타데이터(지연 시간, inputs, outputs)를 포함합니다.
로깅된 경우 메시지 콘텐츠 또는 구조화된 데이터를 선택적으로 표시합니다.
턴의 전체 실행을 보려면 스레드 세부 정보 드로어에서 해당 턴을 열 수 있습니다. 이렇게 하면 해당 턴 중 발생한 모든 중첩 오퍼레이션을 자세히 살펴볼 수 있습니다.
턴에 LLM calls에서 추출된 메시지가 포함되어 있으면 해당 메시지는 오른쪽 채팅 창에 표시됩니다. 이러한 메시지는 일반적으로 지원되는 인테그레이션(예: openai.ChatCompletion.create)에서 이루어진 calls에서 오며, 표시되려면 특정 기준을 충족해야 합니다. 자세한 내용은 채팅 뷰 동작을 참조하세요.

채팅 뷰 동작 방식

채팅 창에는 각 턴에서 발생한 LLM Calls에서 추출한 구조화된 메시지 데이터가 표시됩니다. 이 뷰는 상호작용을 대화 형식으로 렌더링해 보여줍니다.

무엇이 메시지로 간주되나요?

Weave는 한 turn 내에서 LLM 제공자와 직접 상호작용한 Calls(예: 프롬프트를 보내고 응답을 받는 경우)에서 메시지를 추출합니다. 다른 Calls 안에 더 깊게 중첩되지 않은 Calls만 메시지로 표시됩니다. 이렇게 하면 중간 step이나 내부 집계 로직이 중복 표시되는 것을 방지할 수 있습니다. 일반적으로 자동 패치된 서드파티 SDK는 다음과 같은 메시지를 생성합니다.

openai.ChatCompletion.create
anthropic.Anthropic.completion

메시지가 하나도 없으면 어떻게 되나요?

어떤 턴에서도 메시지가 생성되지 않으면 chat pane에 해당 턴의 빈 메시지 섹션이 표시됩니다. chat pane에는 같은 스레드의 다른 턴에서 온 메시지가 계속 표시될 수 있습니다.

Turn 및 Chat 상호작용

턴을 클릭하면 chat pane이 해당 턴의 메시지 위치로 스크롤됩니다(고정 동작).
chat pane을 스크롤하면 왼쪽 목록에서 해당 턴이 강조 표시됩니다.

트레이스 뷰로 이동하고 돌아가기

해당 항목을 클릭하면 턴의 전체 트레이스를 열 수 있습니다. 스레드 상세 뷰로 돌아갈 수 있도록 왼쪽 상단에 뒤로 가기 버튼이 표시됩니다. Weave는 화면 전환 시 UI 상태(예: 스크롤 위치)를 유지하지 않습니다.

SDK 사용

이 섹션의 각 예제는 애플리케이션에서 턴과 스레드를 구성하는 서로 다른 전략을 보여줍니다. 대부분의 예제에서는 스텁 함수 내부에 자체 LLM call 또는 시스템 동작을 구현해야 합니다.

세션이나 대화를 추적하려면 weave.thread() 컨텍스트 관리자를 사용하세요.
논리적 오퍼레이션에 @weave.op를 데코레이션해 턴 또는 중첩된 Calls로 추적하세요.
thread_id를 전달하면 Weave는 해당 블록의 모든 오퍼레이션을 같은 스레드로 묶는 데 사용합니다. thread_id를 생략하면 Weave가 고유한 값을 자동으로 생성합니다.

weave.thread()의 반환 값은 thread_id 속성을 가진 ThreadContext 객체이며, 이를 기록하거나 재사용하거나 다른 시스템에 전달할 수 있습니다. 중첩된 weave.thread() 컨텍스트는 같은 thread_id를 재사용하지 않는 한 항상 새 스레드를 시작합니다. 하위 컨텍스트가 종료되어도 상위 컨텍스트가 중단되거나 덮어써지지 않습니다. 따라서 앱 로직에 따라 분기된 스레드 구조나 계층형 스레드 오케스트레이션을 구현할 수 있습니다.

기본 스레드 생성

다음 코드 예제는 weave.thread()를 사용해 하나 이상의 오퍼레이션을 공유 thread_id로 묶는 방법을 보여줍니다. 이는 애플리케이션에서 스레드 사용을 시작하는 가장 간단한 방법입니다.

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:
        """중첩 call: 구현 세부 사항으로, 스레드 통계에 집계되지 않습니다."""
        context = self._retrieve_context(message)     # 또 다른 중첩 call
        intent = self._classify_intent(message)       # 또 다른 중첩 call
        response = self._call_llm(message, context)   # LLM call (중첩)
        return self._format_response(response)        # 마지막 중첩 call

    @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개의 턴 + 여러 중첩 call 생성
    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개의 call(중첩 포함)로 구성된 스레드

# 대안: 세션 추적을 위해 명시적 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

호출 깊이가 맞지 않는 수동 에이전트

이 예제에서는 스레드 컨텍스트를 적용하는 방식에 따라 call stack의 서로 다른 깊이에서 turn을 정의할 수 있음을 보여줍니다. 이 샘플은 두 개의 프로바이더(OpenAI 및 Anthropic)를 사용하며, 각 프로바이더는 turn 경계에 도달하기 전까지의 call 깊이가 서로 다릅니다. 모든 turn은 동일한 thread_id를 공유하지만, turn 경계는 프로바이더 로직에 따라 스택의 서로 다른 수준에 나타납니다. 이는 서로 다른 백엔드에 대해 호출을 다르게 트레이스해야 하면서도, 여전히 이를 동일한 스레드로 그룹화해야 할 때 유용합니다.

import weave
import random
import asyncio

class OpenAIProvider:
    """OpenAI 브랜치: 턴 경계까지 2단계 깊이의 call 체인"""

    @weave.op
    def route_to_openai(self, user_input: str, thread_id: str) -> str:
        """레벨 1: OpenAI 요청 라우팅 및 준비"""
        # 입력 검증, 라우팅 로직, 기본 전처리
        print(f"  L1: Routing to OpenAI for: {user_input}")

        # 여기가 턴 경계 - 스레드 컨텍스트로 래핑
        with weave.thread(thread_id):
            # 레벨 2 직접 호출 - call 체인 깊이 생성
            return self.execute_openai_call(user_input)

    @weave.op
    def execute_openai_call(self, user_input: str) -> str:
        """레벨 2: 턴 경계 - OpenAI API call 실행"""
        print(f"    L2: Executing OpenAI API call")
        response = f"OpenAI GPT-4 response: {user_input}"
        return response


class AnthropicProvider:
    """Anthropic 브랜치: 턴 경계까지 3단계 깊이의 call 체인"""

    @weave.op
    def route_to_anthropic(self, user_input: str, thread_id: str) -> str:
        """레벨 1: Anthropic 요청 라우팅 및 준비"""
        # 입력 검증, 라우팅 로직, 프로바이더 선택
        print(f"  L1: Routing to Anthropic for: {user_input}")

        # 레벨 2 호출 - call 체인 깊이 생성
        return self.authenticate_anthropic(user_input, thread_id)

    @weave.op
    def authenticate_anthropic(self, user_input: str, thread_id: str) -> str:
        """레벨 2: Anthropic 인증 및 설정 처리"""
        print(f"    L2: Authenticating with Anthropic")

        # 인증, 요청 속도 제한, 세션 관리
        auth_token = "anthropic_key_xyz_authenticated"

         # 여기가 턴 경계 - 레벨 3에서 스레드 컨텍스트로 래핑
        with weave.thread(thread_id):
            # 레벨 3 호출 - call 체인 추가 중첩
            return self.execute_anthropic_call(user_input, auth_token)

    @weave.op
    def execute_anthropic_call(self, user_input: str, auth_token: str) -> str:
        """레벨 3: 턴 경계 - Anthropic API call 실행"""
        print(f"      L3: Executing Anthropic API call with auth")
        response = f"Anthropic Claude response (auth: {auth_token[:15]}...): {user_input}"
        return response


class MultiProviderAgent:
    """call 체인 깊이가 서로 다른 프로바이더 간에 라우팅하는 메인 에이전트"""

    def __init__(self):
        self.openai_provider = OpenAIProvider()
        self.anthropic_provider = AnthropicProvider()

    def handle_conversation_turn(self, user_input: str, thread_id: str) -> str:
        """
        call 체인 깊이가 불균형한 프로바이더로 라우팅합니다.
        각 체인에서 서로 다른 중첩 레벨에 스레드 컨텍스트가 적용됩니다.
        """
        # 데모를 위해 프로바이더를 무작위로 선택
        use_openai = random.choice([True, False])

        if use_openai:
            print(f"Choosing OpenAI (2-level call chain)")
            # OpenAI: 레벨 1 → 레벨 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: 레벨 1 → 레벨 2 → 레벨 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"

    # call 체인 깊이가 서로 다른 멀티턴 대화
    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}")

        # 서로 다른 call 체인 깊이에서 동일한 thread_id 사용
        response = agent.handle_conversation_turn(user_input, conversation_id)
        print(f"Agent: {response}")

if __name__ == "__main__":
    asyncio.run(main())

# 예상 결과: 5개의 턴을 가진 단일 스레드
# - OpenAI 턴: call 체인의 레벨 2에 스레드 컨텍스트 적용
#   콜 스택: route_to_openai() → execute_openai_call() ← 여기에 스레드 컨텍스트
# - Anthropic 턴: call 체인의 레벨 3에 스레드 컨텍스트 적용
#   콜 스택: route_to_anthropic() → authenticate_anthropic() → execute_anthropic_call() ← 여기에 스레드 컨텍스트
# - 모든 턴이 thread_id 공유: "nested_depth_conversation_999"
# - 서로 다른 콜 스택 깊이에서 턴 경계 표시
# - call 체인의 보조 오퍼레이션은 턴이 아닌 중첩 calls로 추적

이전 세션 재개

이전에 시작한 세션을 재개하고 같은 스레드에 call을 계속 추가해야 하는 경우가 있습니다. 반면 기존 세션을 재개할 수 없어 대신 새 스레드를 시작해야 하는 경우도 있습니다. 선택적 스레드 재개를 구현할 때는 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}")

# call 추적을 위한 스레드 컨텍스트 설정
with weave.thread(thread_id) as thread_ctx:
    # 대화 이력 불러오기 또는 초기화
    history = load_history(thread_id)
    print(f"활성 스레드 ID: {thread_ctx.thread_id}")
    
    # 애플리케이션 로직을 여기에 작성하세요...

중첩된 스레드

이 예제는 서로 연계된 여러 스레드를 사용해 복잡한 애플리케이션을 구성하는 방법을 보여줍니다. 각 계층은 자체 스레드 컨텍스트에서 실행되므로 관심사를 깔끔하게 분리할 수 있습니다. 상위 애플리케이션 스레드는 공유 ThreadContext를 사용해 스레드 IDs를 설정함으로써 각 계층을 조율합니다. 시스템의 서로 다른 부분을 각각 독립적으로 분석하거나 모니터링하면서도, 이를 하나의 공유 세션에 연결해야 할 때 이 패턴을 사용하세요.

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 파라미터 드릴링 불필요
# - 앱/인프라/로직 레이어 독립적 모니터링
# - 스레드 컨텍스트를 통한 전역 조율

API 사양

엔드포인트

엔드포인트: 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 call 수
    start_time: datetime     # 이 스레드에서 turn call의 가장 이른 시작 시간
    last_updated: datetime   # 이 스레드에서 turn call의 가장 늦은 종료 시간

class ThreadsQueryRes:
    threads: List[ThreadSchema]

최근 활성 스레드 조회

이 예제는 가장 최근에 업데이트된 스레드 50개를 가져옵니다. 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}")

활동 수준에 따라 스레드 쿼리하기

이 예제는 턴 수를 기준으로 정렬해 가장 활동적인 스레드 20개를 가져옵니다.

# 가장 활발한 스레드 조회 (턴 수 기준)
response = client.threads_query(ThreadsQueryReq(
    project_id="my-project",
    sort_by=[SortBy(field="turn_count", direction="desc")],
    limit=20
))

최근 스레드만 쿼리하기

이 예제는 지난 24시간 이내에 시작된 스레드를 반환합니다. 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")]
))

시작하기

가이드

쿡북

레퍼런스

세부 정보 및 지원

사용 사례

정의

스레드

Turn

Call

트레이스

UI Overview

Threads 목록 뷰

스레드 세부 정보 드로어

채팅 뷰 동작 방식

무엇이 메시지로 간주되나요?

메시지가 하나도 없으면 어떻게 되나요?

Turn 및 Chat 상호작용

트레이스 뷰로 이동하고 돌아가기

SDK 사용

기본 스레드 생성

수동 에이전트 루프 구현

호출 깊이가 맞지 않는 수동 에이전트

이전 세션 재개

중첩된 스레드

API 사양

엔드포인트

요청 스키마

응답 스키마

최근 활성 스레드 조회

활동 수준에 따라 스레드 쿼리하기

최근 스레드만 쿼리하기

시작하기

가이드

쿡북

레퍼런스

세부 정보 및 지원

Documentation Index

​사용 사례

​정의

​스레드

​Turn

​Call

​트레이스

​UI Overview

​Threads 목록 뷰

​스레드 세부 정보 드로어

​채팅 뷰 동작 방식

​무엇이 메시지로 간주되나요?

​메시지가 하나도 없으면 어떻게 되나요?

​Turn 및 Chat 상호작용

​트레이스 뷰로 이동하고 돌아가기

​SDK 사용

​기본 스레드 생성

​수동 에이전트 루프 구현

​호출 깊이가 맞지 않는 수동 에이전트

​이전 세션 재개

​중첩된 스레드

​API 사양

​엔드포인트

​요청 스키마

​응답 스키마

​최근 활성 스레드 조회

​활동 수준에 따라 스레드 쿼리하기

​최근 스레드만 쿼리하기

사용 사례

정의

스레드

Turn

Call

트레이스

UI Overview

Threads 목록 뷰

스레드 세부 정보 드로어

채팅 뷰 동작 방식

무엇이 메시지로 간주되나요?

메시지가 하나도 없으면 어떻게 되나요?

Turn 및 Chat 상호작용

트레이스 뷰로 이동하고 돌아가기

SDK 사용

기본 스레드 생성

수동 에이전트 루프 구현

호출 깊이가 맞지 않는 수동 에이전트

이전 세션 재개

중첩된 스레드

API 사양

엔드포인트

요청 스키마

응답 스키마

최근 활성 스레드 조회

활동 수준에 따라 스레드 쿼리하기

최근 스레드만 쿼리하기