더 이상 실행을 추적하고 싶지 않다면 LANGSMITH_TRACING 환경 변수를 제거할 수 있습니다. 이는 RunTree 객체나 API 사용자에게는 영향을 미치지 않습니다. 이들은 저수준 기능으로 설계되어 추적 토글의 영향을 받지 않기 때문입니다.
LangChain(Python 또는 JS/TS)을 사용하는 경우, 이 섹션을 건너뛰고 LangChain 전용 안내로 바로 이동하실 수 있습니다. @traceable / traceable 사용하기
LangSmith는 Python의 @traceable 데코레이터와 TypeScript의 traceable 함수를 통해 기존 코드를 최소한으로 수정하면서 추적을 로깅할 수 있게 해줍니다.
@traceable 또는 traceable을 사용하더라도, 추적을 LangSmith에 로깅하려면 LANGSMITH_TRACING 환경 변수를 'true'로 설정해야 합니다. 이를 통해 코드를 변경하지 않고도 추적을 켜고 끌 수 있습니다.또한 API 키를 LANGSMITH_API_KEY 환경 변수로 설정해야 합니다(자세한 내용은 설정 참조).기본적으로 추적은 default라는 이름의 프로젝트에 로깅됩니다. 다른 프로젝트에 추적을 로깅하려면 이 섹션을 참조하세요.@traceable 데코레이터는 LangSmith Python SDK에서 추적을 로깅하는 간단한 방법입니다. 어떤 함수든 @traceable로 데코레이트하기만 하면 됩니다.
동기 함수를 traceable로 래핑할 때(예: 아래 예제의 formatPrompt), 추적이 올바르게 로깅되도록 호출 시 await 키워드를 사용해야 한다는 점에 유의하세요.
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-4o-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()
trace 컨텍스트 매니저 사용하기 (Python 전용)
Python에서는 trace 컨텍스트 매니저를 사용하여 LangSmith에 추적을 로깅할 수 있습니다. 이는 다음과 같은 상황에서 유용합니다:
- 특정 코드 블록에 대한 추적을 로깅하고 싶을 때.
- 추적의 입력, 출력 및 기타 속성을 제어하고 싶을 때.
- 데코레이터나 래퍼를 사용하는 것이 실용적이지 않을 때.
- 위의 경우 중 일부 또는 전부.
컨텍스트 매니저는 traceable 데코레이터 및 wrap_openai 래퍼와 원활하게 통합되므로, 동일한 애플리케이션에서 함께 사용할 수 있습니다.
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-4o-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이 필요하지 않습니다.
이 방법은 추적 컨텍스트를 전파하는 과정에서 실수하기 쉽기 때문에 권장되지 않습니다.
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-4o-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()
사용 예제
위의 유틸리티를 확장하여 모든 코드를 편리하게 추적할 수 있습니다. 다음은 몇 가지 확장 예제입니다:
클래스의 모든 공개 메서드 추적하기:
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)
종료 전 모든 추적이 제출되도록 보장하기
LangSmith의 추적은 프로덕션 애플리케이션을 방해하지 않도록 백그라운드 스레드에서 수행됩니다. 이는 모든 추적이 LangSmith에 성공적으로 게시되기 전에 프로세스가 종료될 수 있음을 의미합니다. 애플리케이션을 종료하기 전에 모든 추적이 제출되도록 보장하는 몇 가지 옵션이 있습니다.
LangSmith SDK 사용하기
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()
LangChain 사용하기
LangChain을 사용하는 경우, LangChain 추적 가이드를 참조하세요.
비디오 튜토리얼을 선호하신다면, LangSmith 소개 과정의 추적 기초 비디오를 확인하세요.
