추적 중첩 문제 해결

​

Python

​

asyncio를 사용한 컨텍스트 전파

​

원인

​

해결 방법

1. Python 버전 업그레이드(권장) 가능하다면 자동 컨텍스트 전파를 위해 Python 3.11 이상으로 업그레이드하세요.
2. 수동 컨텍스트 전파 업그레이드가 불가능한 경우 추적 컨텍스트를 수동으로 전파해야 합니다. 방법은 설정에 따라 다릅니다: a) LangGraph 또는 LangChain 사용 부모 config를 자식 호출에 전달합니다:
   Copy

   import asyncio
   from langchain_core.runnables import RunnableConfig, RunnableLambda
   
   @RunnableLambda
   async def my_child_runnable(
       inputs: str,
       # config 인수(아래 parent_runnable에 있음)는 선택 사항입니다
   ):
       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 직접 사용 실행 트리를 직접 전달합니다:
   Copy

   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,
       # 실행 트리는 데코레이터에 의해 자동으로 채워질 수 있습니다
       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 결합 수동 전달을 위해 기술을 조합합니다:
   Copy

   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로 데코레이션된 함수는 "config"를 통해 전달될 때 RunnableConfig를 직접 받을 수 있습니다
       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):
       # 추적 컨텍스트를 수동으로 설정할 수 있습니다
       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())
   

​

스레딩을 사용한 컨텍스트 전파

​

원인

​

해결 방법

1. LangSmith의 ContextThreadPoolExecutor 사용 LangSmith는 컨텍스트 전파를 자동으로 처리하는 ContextThreadPoolExecutor를 제공합니다:
   Copy

   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()
   

2. 부모 실행 트리 수동 제공 또는 부모 실행 트리를 내부 함수에 수동으로 전달할 수 있습니다:
   Copy

   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()