트레이스 중첩 문제 해결하기

LangSmith SDK, LangGraph, LangChain으로 트레이싱할 때는 트레이싱이 올바른 컨텍스트를 자동으로 전파해서, 부모 트레이스 안에서 실행된 코드가 UI의 예상 위치에 렌더링되는 것이 정상입니다.

만약 어떤 자식 런이 별도의 트레이스로 분리되어(최상위 레벨에 나타나) 보인다면, 다음에 소개하는 알려진 "엣지 케이스" 중 하나가 원인일 수 있어요.

출처: 문서

본문

Python

다음은 파이썬으로 빌드할 때 "분리된(split)" 트레이스가 생기는 흔한 원인들을 정리한 내용이에요.

asyncio를 사용한 컨텍스트 전파

Python 버전이 3.11 미만인 경우 비동기 호출(특히 스트리밍)을 사용할 때 트레이스 중첩에 문제가 생길 수 있어요. 이는 Python의 asyncio가 컨텍스트 전달을 완전히 지원하게 된 것이 3.11 버전부터이기 때문이에요.

왜 그런가요?

LangChain과 LangSmith SDK는 contextvars를 사용해서 트레이싱 정보를 암묵적으로 전파해요. Python 3.11 이상에서는 이것이 자연스럽게 동작해요. 하지만 이전 버전(3.8, 3.9, 3.10)에서는 asyncio 태스크가 contextvar를 제대로 지원하지 않아 트레이스가 끊어질 수 있어요.

해결 방법

Python 버전 업그레이드 (권장) 가능하다면 Python 3.11 이상으로 업그레이드하면 컨텍스트가 자동으로 전파돼요.

수동 컨텍스트 전파 업그레이드가 어렵다면 트레이싱 컨텍스트를 직접 전파해야 해요. 방법은 설정에 따라 달라져요:

a) LangGraph 또는 LangChain 사용 시 부모 config를 자식 호출에 전달하세요:

import asyncio
from langchain_core.runnables import RunnableConfig, RunnableLambda

@RunnableLambda
async def my_child_runnable(
    inputs: str,
    # The config arg (present in parent_runnable below) is optional
):
    yield "A"
    yield "response"

@RunnableLambda
async def parent_runnable(inputs: str, config: RunnableConfig):
    async for chunk in my_child_runnable.astream(inputs, config):
        yield chunk

async def main():
    return [val async for val in parent_runnable.astream("call")]

asyncio.run(main())

b) LangSmith를 직접 사용할 때 런 트리를 직접 전달하세요:

import asyncio
import langsmith as ls

@ls.traceable
async def my_child_function(inputs: str):
    yield "A"
    yield "response"

@ls.traceable
async def parent_function(
    inputs: str,
    # The run tree can be auto-populated by the decorator
    run_tree: ls.RunTree,
):
    async for chunk in my_child_function(inputs, langsmith_extra={"parent": run_tree}):
        yield chunk

async def main():
    return [val async for val in parent_function("call")]

asyncio.run(main())

c) 데코레이터 코드와 LangGraph/LangChain 결합 수동 핸드오프를 위해 여러 기법을 조합해 사용하세요:

import asyncio
import langsmith as ls
from langchain_core.runnables import RunnableConfig, RunnableLambda

@RunnableLambda
async def my_child_runnable(inputs: str):
    yield "A"
    yield "response"

@ls.traceable
async def my_child_function(inputs: str, run_tree: ls.RunTree):
    with ls.tracing_context(parent=run_tree):
        async for chunk in my_child_runnable.astream(inputs):
            yield chunk

@RunnableLambda
async def parent_runnable(inputs: str, config: RunnableConfig):
    # @traceable decorated functions can directly accept a RunnableConfig when passed in via "config"
    async for chunk in my_child_function(inputs, langsmith_extra={"config": config}):
        yield chunk

@ls.traceable
async def parent_function(inputs: str, run_tree: ls.RunTree):
    # You can set the tracing context manually
    with ls.tracing_context(parent=run_tree):
        async for chunk in parent_runnable.astream(inputs):
            yield chunk

async def main():
    return [val async for val in parent_function("call")]

asyncio.run(main())

threading을 사용한 컨텍스트 전파

트레이싱을 시작한 뒤, 단일 트레이스 안에서 자식 태스크에 병렬 처리를 적용하고 싶은 경우가 흔해요. Python 표준 라이브러리의 ThreadPoolExecutor는 기본적으로 트레이싱을 깨뜨려요.

왜 그런가요?

Python의 contextvars는 새 스레드 안에서는 빈 상태로 시작해요. 트레이스 연속성을 유지하는 두 가지 방법이 있어요.

해결 방법

LangSmith의 ContextThreadPoolExecutor 사용하기

LangSmith는 컨텍스트 전파를 자동으로 처리하는 ContextThreadPoolExecutor를 제공해요:

from langsmith.utils import ContextThreadPoolExecutor
from langsmith import traceable

@traceable
def outer_func():
    with ContextThreadPoolExecutor() as executor:
        inputs = [1, 2]
        r = list(executor.map(inner_func, inputs))

@traceable
def inner_func(x):
    print(x)

outer_func()

부모 런 트리를 직접 전달하기

또는 부모 런 트리를 내부 함수에 직접 전달할 수도 있어요:

from langsmith import traceable, get_current_run_tree
from concurrent.futures import ThreadPoolExecutor

@traceable
def outer_func():
    rt = get_current_run_tree()
    with ThreadPoolExecutor() as executor:
        r = list(
            executor.map(
                lambda x: inner_func(x, langsmith_extra={"parent": rt}), [1, 2]
            )
        )

@traceable
def inner_func(x):
    print(x)

outer_func()

이 방법에서는 get_current_run_tree()로 현재 런 트리를 가져와서 langsmith_extra 파라미터를 통해 내부 함수에 전달해요.

두 방법 모두, 내부 함수 호출이 별도의 스레드에서 실행되더라도 초기 트레이스 스택 아래에 올바르게 집계되도록 보장해요.

트레이스 중첩 문제 해결하기