커스텀 인스트루먼테이션

커스텀 인스트루먼테이션

애플리케이션 코드에 직접 인스트루먼테이션을 추가하면 애플리케이션이 트레이스하는 함수, 기록되는 입력·출력, 그리고 트레이스 계층 구조를 정밀하게 제어할 수 있어요. 핵심 인스트루먼테이션 접근 방식은 세 가지입니다:

• @traceable 데코레이터: 대부분의 경우에 권장돼요
• trace 컨텍스트 매니저: Python 전용
• RunTree API: 명시적이고 저수준의 제어

이 페이지에서는 다음 내용도 다룹니다:

• 커스텀 실행(run) ID 지정하기 — 실행 직후 피드백을 연결하거나 외부 시스템과 연관시키는 데 유용해요.
• 프로세스가 종료되기 전에 모든 트레이스가 제출되도록 보장하기.

LangChain(Python 또는 JS/TS)을 사용한다면 LangChain 전용 안내를 참고하세요.

LangSmith 통합이 내장된 LLM 프로바이더나 에이전트 프레임워크를 사용한다면 통합 개요를 대신 참고하세요.

출처: 문서

본문

사전 요구 사항

트레이싱 전에 다음 환경 변수를 설정하세요:

• LANGSMITH_TRACING=true: 트레이싱을 활성화해요. 코드를 변경하지 않고도 이 값을 켜고 꺼 트레이싱을 토글할 수 있어요.

  LANGSMITH_TRACING은 @traceable 데코레이터와 trace 컨텍스트 매니저를 제어해요. 환경 변수를 변경하지 않고 런타임에 @traceable을 재정의하려면 tracing_context(enabled=True/False)(Python)를 사용하거나 tracingEnabled를 traceable에 직접 전달하세요(JS/TS). RunTree 객체는 이러한 제어의 영향을 받지 않으며, post 시 항상 데이터를 LangSmith로 전송해요.

• LANGSMITH_API_KEY: 당신의 LangSmith API 키예요.

기본적으로 LangSmith는 트레이스를 default라는 프로젝트에 기록해요. 다른 프로젝트에 기록하려면 LANGSMITH_PROJECT를 설정하세요. 자세한 내용은 특정 프로젝트에 트레이스 기록을 참고하세요.

@traceable / traceable 사용하기

@traceable(Python), traceable(TypeScript), traceable(Kotlin) 또는 Tracing.traceFunction(Java)을 함수에 적용하면 해당 함수가 트레이스된 실행(run)이 됩니다. LangSmith는 중첩된 호출 전체에서 컨텍스트 전파를 자동으로 처리해요.

다음 예시는 간단한 파이프라인을 트레이스합니다: run_pipeline이 format_prompt를 호출해 메시지를 만들고, invoke_llm을 호출해 모델을 호출하며, parse_output을 호출해 결과를 추출해요.

각 함수는 개별적으로 트레이스되며, run_pipeline(역시 트레이스됨) 내부에서 호출되기 때문에 LangSmith는 자동으로 이를 자식 실행(run)으로 중첩해요. invoke_llm은 run_type="llm"을 사용해 LLM 호출로 표시하므로 LangSmith가 토큰 수와 대기 시간을 올바르게 렌더링할 수 있어요:

from langsmith import traceable
from openai import Client

openai = Client()

@traceable
def format_prompt(subject):
  return [
      {
          "role": "system",
          "content": "You are a helpful assistant.",
      },
      {
          "role": "user",
          "content": f"What's a good name for a store that sells {subject}?"
      }
  ]

@traceable(run_type="llm")
def invoke_llm(messages):
  return openai.chat.completions.create(
      messages=messages, model="gpt-5.4-mini", temperature=0
  )

@traceable
def parse_output(response):
  return response.choices[0].message.content

@traceable
def run_pipeline():
  messages = format_prompt("colorful socks")
  response = invoke_llm(messages)
  return parse_output(response)

run_pipeline()

import { traceable } from "langsmith/traceable";
import OpenAI from "openai";

const openai = new OpenAI();

const formatPrompt = traceable((subject: string) => {
  return [
    {
      role: "system" as const,
      content: "You are a helpful assistant.",
    },
    {
      role: "user" as const,
      content: `What's a good name for a store that sells ${subject}?`,
    },
  ];
},{ name: "formatPrompt" });

const invokeLLM = traceable(
  async ({ messages }: { messages: { role: string; content: string }[] }) => {
      return openai.chat.completions.create({
          model: "gpt-5.4-mini",
          messages: messages,
          temperature: 0,
      });
  },
  { run_type: "llm", name: "invokeLLM" }
);

const parseOutput = traceable(
  (response: any) => {
      return response.choices[0].message.content;
  },
  { name: "parseOutput" }
);

const runPipeline = traceable(
  async () => {
      const messages = await formatPrompt("colorful socks");
      const response = await invokeLLM({ messages });
      return parseOutput(response);
  },
  { name: "runPipeline" }
);

await runPipeline();

import com.langchain.smith.tracing.RunType
import com.langchain.smith.tracing.TraceConfig
import com.langchain.smith.tracing.Tracing
import com.openai.client.OpenAIClient
import com.openai.client.okhttp.OpenAIOkHttpClient
import com.openai.models.ChatModel
import com.openai.models.chat.completions.ChatCompletion
import com.openai.models.chat.completions.ChatCompletionCreateParams
import com.openai.models.chat.completions.ChatCompletionMessageParam
import com.openai.models.chat.completions.ChatCompletionSystemMessageParam
import com.openai.models.chat.completions.ChatCompletionUserMessageParam
import java.util.Arrays
import java.util.List
import java.util.function.Function

public class TraceablePipeline {
  public static void main(String[] args) {
    new TraceablePipelineRunner().run();
  }

  private static final class TraceablePipelineRunner {
    private final OpenAIClient openai = OpenAIOkHttpClient.fromEnv();

    private final Function<String, List<ChatCompletionMessageParam>> formatPrompt =
        Tracing.traceFunction(
            subject ->
                Arrays.asList(
                    ChatCompletionMessageParam.ofSystem(
                        ChatCompletionSystemMessageParam.builder()
                            .content("You are a helpful assistant.")
                            .build()),
                    ChatCompletionMessageParam.ofUser(
                        ChatCompletionUserMessageParam.builder()
                            .content("What's a good name for a store that sells " + subject + "?")
                            .build())),
            TraceConfig.builder().name("format_prompt").build());

    private final Function<List<ChatCompletionMessageParam>, ChatCompletion> invokeLlm =
        Tracing.traceFunction(
            messages ->
                openai.chat()
                    .completions()
                    .create(
                        ChatCompletionCreateParams.builder()
                            .model(ChatModel.GPT_5_5)
                            .messages(messages)
                            .build()),
            TraceConfig.builder().name("invoke_llm").runType(RunType.LLM).build());

    private final Function<ChatCompletion, String> parseOutput =
        Tracing.traceFunction(
            response -> response.choices().get(0).message().content().orElse(""),
            TraceConfig.builder().name("parse_output").build());

    private final Function<String, String> runPipeline =
        Tracing.traceFunction(
            subject -> parseOutput.apply(invokeLlm.apply(formatPrompt.apply(subject))),
            TraceConfig.builder().name("run_pipeline").build());

    void run() {
      runPipeline.apply("colorful socks");
    }
  }
}

이 예시의 공개 LangSmith 실행을 열어보세요.

import com.langchain.smith.tracing.RunType
import com.langchain.smith.tracing.TraceConfig
import com.langchain.smith.tracing.traceable
import com.openai.client.okhttp.OpenAIOkHttpClient
import com.openai.models.ChatModel
import com.openai.models.chat.completions.ChatCompletion
import com.openai.models.chat.completions.ChatCompletionCreateParams
import com.openai.models.chat.completions.ChatCompletionMessageParam
import com.openai.models.chat.completions.ChatCompletionSystemMessageParam
import com.openai.models.chat.completions.ChatCompletionUserMessageParam
import kotlin.jvm.optionals.getOrNull

val openai = OpenAIOkHttpClient.fromEnv()

val formatPrompt =
    traceable(
        { subject: String ->
            listOf(
                ChatCompletionMessageParam.ofSystem(
                    ChatCompletionSystemMessageParam.builder()
                        .content("You are a helpful assistant.")
                        .build(),
                ),
                ChatCompletionMessageParam.ofUser(
                    ChatCompletionUserMessageParam.builder()
                        .content("What's a good name for a store that sells $subject?")
                        .build(),
                ),
            )
        },
        TraceConfig.builder().name("format_prompt").build(),
    )

val invokeLlm =
    traceable(
        { messages: List<ChatCompletionMessageParam> ->
            openai.chat().completions().create(
                ChatCompletionCreateParams.builder()
                    .model(ChatModel.GPT_5_5)
                    .messages(messages)
                    .build(),
            )
        },
        TraceConfig.builder().name("invoke_llm").runType(RunType.LLM).build(),
    )

val parseOutput =
    traceable(
        { response: ChatCompletion ->
            response.choices()[0].message().content().getOrNull().orEmpty()
        },
        TraceConfig.builder().name("parse_output").build(),
    )

val runPipeline =
    traceable(
        { subject: String -> parseOutput(invokeLlm(formatPrompt(subject))) },
        TraceConfig.builder().name("run_pipeline").build(),
    )

println(runPipeline("colorful socks"))

UI에서는 format_prompt, invoke_llm, parse_output이 중첩된 자식 실행으로 있는 run_pipeline 트레이스를 확인할 수 있어요.

traceable로 동기 함수를 감쌀 때는(예: 앞선 예시의 formatPrompt) 호출 시 await 키워드를 사용해야 트레이스가 올바르게 기록됩니다.

trace 컨텍스트 매니저 사용하기 (Python 전용)

Python에서는 trace 컨텍스트 매니저를 사용해 트레이스를 LangSmith에 기록할 수 있어요. 이는 다음과 같은 상황에서 유용합니다:

1. 코드의 특정 블록에 대한 트레이스를 기록하려 할 때.
2. 트레이스의 입력, 출력 및 기타 속성을 제어하려 할 때.
3. 데코레이터나 래퍼를 사용할 수 없을 때.
4. 위 중 하나 또는 모두에 해당할 때.

컨텍스트 매니저는 traceable 데코레이터 및 wrap_openai 래퍼와 완벽하게 통합되므로 같은 애플리케이션에서 함께 사용할 수 있어요.

다음 예시는 세 가지를 모두 함께 사용하는 모습을 보여줍니다. wrap_openai는 OpenAI 클라이언트를 감싸 호출이 자동으로 트레이스되게 해요. my_tool은 @traceable을 run_type="tool"과 커스텀 name과 함께 사용해 트레이스에서 올바르게 나타나게 합니다. chat_pipeline 자체는 데코레이트되지 않으며, 대신 ls.trace가 호출을 감싸 프로젝트 이름과 입력을 명시적으로 전달하고 rt.end()로 출력을 수동으로 설정할 수 있게 해요:

import openai
import langsmith as ls
from langsmith.wrappers import wrap_openai

client = wrap_openai(openai.Client())

@ls.traceable(run_type="tool", name="Retrieve Context")
def my_tool(question: str) -> str:
    return "During this morning's meeting, we solved all world conflict."

def chat_pipeline(question: str):
    context = my_tool(question)
    messages = [
        { "role": "system", "content": "You are a helpful assistant. Please respond to the user's request only based on the given context." },
        { "role": "user", "content": f"Question: {question}\nContext: {context}"}
    ]
    chat_completion = client.chat.completions.create(
        model="gpt-5.4-mini", messages=messages
    )
    return chat_completion.choices[0].message.content

app_inputs = {"input": "Can you summarize this morning's meetings?"}

with ls.trace("Chat Pipeline", "chain", project_name="my_test", inputs=app_inputs) as rt:
    output = chat_pipeline("Can you summarize this morning's meetings?")
    rt.end(outputs={"output": output})

RunTree API 사용하기

트레이스를 LangSmith에 기록하는 또 다른 더 명시적인 방법은 RunTree API를 사용하는 것입니다. 이 API는 트레이싱을 더 세밀하게 제어할 수 있게 해줘요. 실행과 자식 실행을 수동으로 만들어 트레이스를 조립할 수 있습니다. 여전히 LANGSMITH_API_KEY를 설정해야 하지만, 이 방식에서는 LANGSMITH_TRACING이 필요하지 않아요.

이 방식은 대부분의 사용 사례에 권장되지 않습니다. 트레이스 컨텍스트를 수동으로 관리하는 것은 컨텍스트 전파를 자동으로 처리하는 @traceable에 비해 오류가 발생하기 쉬워요.

import openai
from langsmith.run_trees import RunTree

# This can be a user input to your app
question = "Can you summarize this morning's meetings?"

# Create a top-level run
pipeline = RunTree(
  name="Chat Pipeline",
  run_type="chain",
  inputs={"question": question}
)
pipeline.post()

# This can be retrieved in a retrieval step
context = "During this morning's meeting, we solved all world conflict."
messages = [
  { "role": "system", "content": "You are a helpful assistant. Please respond to the user's request only based on the given context." },
  { "role": "user", "content": f"Question: {question}\nContext: {context}"}
]

# Create a child run
child_llm_run = pipeline.create_child(
  name="OpenAI Call",
  run_type="llm",
  inputs={"messages": messages},
)
child_llm_run.post()

# Generate a completion
client = openai.Client()
chat_completion = client.chat.completions.create(
  model="gpt-5.4-mini", messages=messages
)

# End the runs and log them
child_llm_run.end(outputs=chat_completion)
child_llm_run.patch()
pipeline.end(outputs={"answer": chat_completion.choices[0].message.content})
pipeline.patch()

import OpenAI from "openai";
import { RunTree } from "langsmith";

// This can be a user input to your app
const question = "Can you summarize this morning's meetings?";

const pipeline = new RunTree({
  name: "Chat Pipeline",
  run_type: "chain",
  inputs: { question }
});
await pipeline.postRun();

// This can be retrieved in a retrieval step
const context = "During this morning's meeting, we solved all world conflict.";
const messages = [
  { role: "system", content: "You are a helpful assistant. Please respond to the user's request only based on the given context." },
  { role: "user", content: `Question: ${question}Context: ${context}` }
];

// Create a child run
const childRun = await pipeline.createChild({
  name: "OpenAI Call",
  run_type: "llm",
  inputs: { messages },
});
await childRun.postRun();

// Generate a completion
const client = new OpenAI();
const chatCompletion = await client.chat.completions.create({
  model: "gpt-5.4-mini",
  messages: messages,
});

// End the runs and log them
childRun.end(chatCompletion);
await childRun.patchRun();
pipeline.end({ outputs: { answer: chatCompletion.choices[0].message.content } });
await pipeline.patchRun();

Java 추가 예시를 보려면 공식 문서를 참고하세요 — Java 예시는 커스텀 루트 실행 ID와 전용 ExecutorService를 사용하며, 종료 시 executor를 shutdown하고 awaitTermination으로 대기해 프로세스가 끝나기 전에 백그라운드 실행 제출이 완료되도록 보장합니다. (Java/Kotlin 전체 예시 포함)

예시 사용법

앞서 설명한 유틸리티를 확장해 어떤 코드든 트레이스할 수 있어요. 다음 코드는 몇 가지 확장 예시를 보여줍니다.

클래스의 모든 public 메서드를 트레이스하기:

from typing import Any, Callable, Type, TypeVar

T = TypeVar("T")

def traceable_cls(cls: Type[T]) -> Type[T]:
    """Instrument all public methods in a class."""
    def wrap_method(name: str, method: Any) -> Any:
        if callable(method) and not name.startswith("__"):
            return traceable(name=f"{cls.__name__}.{name}")(method)
        return method

    # Handle __dict__ case
    for name in dir(cls):
        if not name.startswith("_"):
            try:
                method = getattr(cls, name)
                setattr(cls, name, wrap_method(name, method))
            except AttributeError:
                # Skip attributes that can't be set (e.g., some descriptors)
                pass

    # Handle __slots__ case
    if hasattr(cls, "__slots__"):
        for slot in cls.__slots__:  # type: ignore[attr-defined]
            if not slot.startswith("__"):
                try:
                    method = getattr(cls, slot)
                    setattr(cls, slot, wrap_method(slot, method))
                except AttributeError:
                    # Skip slots that don't have a value yet
                    pass

    return cls

@traceable_cls
class MyClass:
    def __init__(self, some_val: int):
        self.some_val = some_val

    def combine(self, other_val: int):
        return self.some_val + other_val

# See trace: https://smith.langchain.com/public/882f9ecf-5057-426a-ae98-0edf84fdcaf9/r
MyClass(13).combine(29)

커스텀 실행 ID 지정하기

기본적으로 LangSmith는 각 실행에 무작위 ID를 할당해요. 실행 ID를 미리 알아야 할 때(예: 실행 직후 피드백을 연결), LangSmith 실행을 외부 시스템의 ID와 연관시키거나, 결정적 ID로 실행을 멱등(idempotent)하게 만들 때 이를 재정의할 수 있습니다.

커스텀 실행 ID에는 UUID v7을 사용하세요. UUIDv7은 타임스탬프를 포함하므로 트레이스 내 실행의 시간 순서를 올바르게 유지합니다. LangSmith SDK는 uuid7 헬퍼를 export해요(Python v0.4.43+, JS v0.3.80+):

• Python: from langsmith import uuid7
• JS/TS: import { uuid7 } from 'langsmith'

모든 UUID v7 문자열이 허용됩니다 — 시스템이 이미 UUID v7 식별자를 사용한다면 SDK 헬퍼 또는 자체 헬퍼를 사용할 수 있어요.

다음 중 하나를 사용하세요:

• @traceable: @traceable 함수를 호출할 때 langsmith_extra 안에 run_id를 전달하세요(Python). TypeScript에서는 traceable에 전달하는 config 객체에 id를 전달해요:

from langsmith import traceable, uuid7

@traceable
def my_pipeline(question: str) -> str:
    return "answer"

run_id = uuid7()
my_pipeline("What is the capital of France?", langsmith_extra={"run_id": run_id})

# run_id can now be used to attach feedback, query the run, etc.

import { traceable } from "langsmith/traceable";
import { uuid7 } from "langsmith";

const runId = uuid7();

const myPipeline = traceable(
async (question: string) => {
    return "answer";
},
{ name: "my-pipeline", id: runId }
);

await myPipeline("What is the capital of France?");

// runId can now be used to attach feedback, query the run, etc.

• trace 컨텍스트 매니저(Python 전용): trace 컨텍스트 매니저 생성자에 run_id를 직접 전달하세요:

from langsmith import trace, uuid7

run_id = uuid7()

with trace("my-pipeline", run_id=run_id) as run:
    result = "answer"
    run.end(outputs={"result": result})

# run_id can now be used to attach feedback, query the run, etc.

종료 전에 모든 트레이스가 제출되도록 보장하기

LangSmith는 프로덕션 애플리케이션을 방해하지 않도록 백그라운드 스레드에서 트레이싱을 수행해요. 즉, 모든 트레이스가 LangSmith에 성공적으로 게시되기 전에 프로세스가 끝날 수 있습니다. 다음 옵션을 참고하세요:

• LangChain을 사용한다면 LangChain 트레이싱 가이드를 참고하세요.
• LangSmith SDK를 단독으로 사용한다면 종료 전에 flush 메서드를 사용할 수 있어요:

from langsmith import Client

client = Client()

@traceable(client=client)
async def my_traced_func():
# Your code here...
pass

try:
await my_traced_func()
finally:
await client.flush()

import { Client } from "langsmith";

const langsmithClient = new Client({});

const myTracedFunc = traceable(async () => {
// Your code here...
},{ client: langsmithClient });

try {
await myTracedFunc();
} finally {
await langsmithClient.flush();
}

관련 문서

• Observability concepts: 실행, 트레이스, LangSmith 데이터 모델에 대한 배경 지식
• Run (span) data format: dotted_order, trace_id, parent_run_id 등 실행 필드에 대한 스키마 참조
• SDK를 사용해 사용자 피드백 기록: 실행 ID를 미리 지정하는 일반적인 사용 사례
• 트레이스된 함수 내에서 현재 실행(span) 접근하기: 트레이스 내부에서 활성 실행을 읽거나 수정하기
• 특정 프로젝트에 트레이스 기록: default 대신 명명된 프로젝트로 트레이스 라우팅
• API로 트레이스하기: SDK의 저수준 REST API 대안
• Introduction to LangSmith Course의 Tracing Basics 동영상

더 알아보기 (Learn more)

• LangChain 트레이싱 가이드 — LangChain 사용 시 트레이싱 설정 방법
• 특정 프로젝트에 트레이스 기록
• SDK를 사용해 사용자 피드백 기록
• 관측성 개념 — 실행, 트레이스 및 데이터 모델