在使用 LangSmith SDK、LangGraph 和 LangChain 进行追踪时，追踪应自动传播正确的上下文，以便在父追踪中执行的代码能在 UI 的预期位置呈现。

如果您看到子运行（child run）进入了单独的追踪（并出现在顶层），可能是由以下已知的"边缘情况"之一引起的。

Python

以下概述了使用 Python 构建时导致"分离"追踪的常见原因。

使用 asyncio 进行上下文传播

在 Python 版本 < 3.11 中使用异步调用（尤其是流式处理）时，您可能会遇到追踪嵌套问题。这是因为 Python 的 asyncio 仅在 3.11 版本中才完全支持传递上下文。

原因

LangChain 和 LangSmith SDK 使用 contextvars 隐式传播追踪信息。在 Python 3.11 及更高版本中，这可以无缝工作。然而，在早期版本（3.8、3.9、3.10）中，asyncio 任务缺乏适当的 contextvar 支持，这可能导致追踪断开。

解决方法

1. 升级 Python 版本（推荐） 如果可能，请升级到 Python 3.11 或更高版本，以获得自动的上下文传播。

2. 手动传播上下文 如果无法升级，您需要手动传播追踪上下文。具体方法取决于您的设置：

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 直接传递运行树（run tree）：

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

使用线程进行上下文传播

通常，在启动追踪后，希望在一个追踪内对子任务应用一些并行处理。Python 标准库的 ThreadPoolExecutor 默认会破坏追踪。

原因

Python 的 contextvars 在新线程中开始时是空的。以下是两种处理方法来保持追踪的连续性：

解决方法

1. 使用 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()

2. 手动提供父运行树

或者，您可以手动将父运行树传递给内部函数：

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 参数将其传递给内部函数。

这两种方法都能确保内部函数调用即使在单独的线程中执行，也能正确地聚合在初始的追踪栈下。