跟踪嵌套疑难解答
使用 LangSmith SDK、LangGraph 和 LangChain 进行跟踪时，跟踪应自动传播正确的上下文，以便在父跟踪中执行的代码呈现在 UI 中的预期位置。
如果您看到子运行转到单独的跟踪（并显示在顶层），则可能是由以下已知的“边缘情况”之一引起的。
蟒
下面概述了使用 python 构建时出现“拆分”跟踪的常见原因。
使用 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 直接传入 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 的 stdlibThreadPoolExecutor默认情况下，中断跟踪。
为什么
Python 的 contextvar 在新线程中开始时为空。以下是处理保持跟踪连续性的两种方法：
解决方法
-
使用 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() -
手动提供父运行树
或者，您可以手动将父 run tree 传递给内部函数：
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参数。
这两种方法都可以确保内部函数调用在初始跟踪堆栈下正确聚合，即使在单独的线程中执行也是如此。