运行助手 (Run Helpers)

langsmith.run_helpers ¶

用于从函数创建运行树的装饰器。

函数	描述
`get_current_run_tree`	获取当前的运行树。
`get_tracing_context`	获取当前的追踪上下文。
`tracing_context`	为代码块设置追踪上下文。
`is_traceable_function`	检查一个函数是否被 @traceable 装饰器修饰。
`ensure_traceable`	确保一个函数是可追踪的。
`is_async`	检查函数或被包装的函数是否为异步函数。
`traceable`	使用 Langsmith 追踪一个函数。
`as_runnable`	将一个被 LangSmith @traceable 装饰器包装的函数转换为 Runnable。

LangSmithExtra ¶

基类：TypedDict

任何需要动态注入到运行中的附加信息。

name `instance-attribute` ¶

name: str | None

运行的可选名称。

reference_example_id `instance-attribute` ¶

reference_example_id: ID_TYPE | None

参考示例的可选 ID。

run_extra `instance-attribute` ¶

run_extra: dict | None

可选的额外运行信息。

parent `instance-attribute` ¶

parent: RunTree | str | Mapping | None

可选的父运行，可以是 RunTree、字符串或映射。

run_tree `instance-attribute` ¶

run_tree: RunTree | None

可选的运行树（已弃用）。

project_name `instance-attribute` ¶

project_name: str | None

项目的可选名称。

metadata `instance-attribute` ¶

metadata: dict[str, Any] | None

运行的可选元数据。

tags `instance-attribute` ¶

tags: list[str] | None

运行的可选标签列表。

run_id `instance-attribute` ¶

run_id: ID_TYPE | None

运行的可选 ID。

client `instance-attribute` ¶

client: Client | None

可选的 LangSmith 客户端。

on_end `instance-attribute` ¶

on_end: Callable[[RunTree], Any] | None

运行结束并发送后要调用的可选回调函数。

SupportsLangsmithExtra ¶

基类：Protocol, Generic[P, R]

此协议的实现接受一个可选的 langsmith_extra 参数。

参数	描述
`*args`	可变长度参数。
`langsmith_extra`	用于 Langsmith 的附加参数的可选字典。类型： `Optional[LangSmithExtra`
`**kwargs`	关键字参数。

返回	描述
`R`	可调用对象的返回类型。

方法	描述
`__call__`	当实例作为函数调用时，调用该实例。

call ¶

__call__(
    *args: args, langsmith_extra: LangSmithExtra | None = None, **kwargs: kwargs
) -> R

当实例作为函数调用时，调用该实例。

参数	描述
`*args`	可变长度参数列表。类型： `args` 默认值： `()`
`langsmith_extra`	包含 Langsmith 特定附加参数的可选字典。类型： `LangSmithExtra \| None` 默认值： `None`
`**kwargs`	任意关键字参数。类型： `kwargs` 默认值： `{}`

返回	描述
`R`	该方法的返回值。类型： `R`

trace ¶

在上下文中管理一个 LangSmith 运行。

此类既可以作为同步上下文管理器，也可以作为异步上下文管理器使用。

参数	描述
`name`	运行的名称。类型： `str`
`run_type`	运行的类型（例如，“chain”、“llm”、“tool”）。默认为“chain”。类型： `RUN_TYPE_T` 默认值： `'chain'`
`inputs`	运行的初始输入数据。默认为 None。类型： `Dict \| None` 默认值： `None`
`project_name`	与运行关联的项目名称。默认为 None。类型: `str \| None` 默认值: `None`
`parent`	父运行。可以是 RunTree、点分隔顺序字符串或追踪头。默认为 None。类型： `RunTree \| str \| Mapping \| None` 默认值： `None`
`tags`	运行的标签列表。默认为 None。类型： `List[str] \| None` 默认值： `None`
`metadata`	运行的附加元数据。默认为 None。类型： `Mapping[str, Any] \| None` 默认值： `None`
`client`	用于自定义设置的 LangSmith 客户端。默认为 None。类型： `Client \| None` 默认值： `None`
`run_id`	运行的预设标识符。默认为 None。类型： `ID_TYPE \| None` 默认值： `None`
`reference_example_id`	将运行与数据集示例关联。仅用于评估中的根运行。默认为 None。类型： `ID_TYPE \| None` 默认值： `None`
`要处理的异常`	要忽略的异常类型。默认为 None。类型： `Tuple[Type[BaseException], ...] \| None` 默认值： `None`
`extra`	要发送到 LangSmith 的额外数据。请改用“metadata”。默认为 None。类型： `Dict \| None` 默认值： `None`

示例

同步用法

.. code-block:: python

>>> with trace("My Operation", run_type="tool", tags=["important"]) as run:
...     result = "foo"  # Perform operation
...     run.metadata["some-key"] = "some-value"
...     run.end(outputs={"result": result})

异步用法

.. code-block:: python

>>> async def main():
...     async with trace("Async Operation", run_type="tool", tags=["async"]) as run:
...         result = "foo"  # Await async operation
...         run.metadata["some-key"] = "some-value"
...         # "end" just adds the outputs and sets error to None
...         # The actual patching of the run happens when the context exits
...         run.end(outputs={"result": result})
>>> asyncio.run(main())

处理特定异常

.. code-block:: python

>>> import pytest
>>> import sys
>>> with trace("Test", exceptions_to_handle=(pytest.skip.Exception,)):
...     if sys.platform == "win32": # Just an example
...         pytest.skip("Skipping test for windows")
...     result = "foo"  # Perform test operation

方法	描述
`__init__`	初始化 trace 上下文管理器。
`__enter__`	同步进入上下文管理器。
`__exit__`	同步退出上下文管理器。
`__aenter__`	异步进入上下文管理器。
`__aexit__`	异步退出上下文管理器。

init ¶

__init__(
    name: str,
    run_type: RUN_TYPE_T = "chain",
    *,
    inputs: dict | None = None,
    extra: dict | None = None,
    project_name: str | None = None,
    parent: RunTree | str | Mapping | Literal["ignore"] | None = None,
    tags: list[str] | None = None,
    metadata: Mapping[str, Any] | None = None,
    client: Client | None = None,
    run_id: ID_TYPE | None = None,
    reference_example_id: ID_TYPE | None = None,
    exceptions_to_handle: tuple[type[BaseException], ...] | None = None,
    attachments: Attachments | None = None,
    **kwargs: Any,
)

初始化 trace 上下文管理器。

如果传入了不支持的关键字参数，则发出警告。

enter ¶

__enter__() -> RunTree

同步进入上下文管理器。

返回	描述
`RunTree`	run_trees.RunTree：新创建的运行。

exit ¶

__exit__(
    exc_type: type[BaseException] | None = None,
    exc_value: BaseException | None = None,
    traceback: TracebackType | None = None,
) -> None

同步退出上下文管理器。

参数	描述
`exc_type`	发生的异常类型（如果有）。类型： `type[BaseException] \| None` 默认值： `None`
`exc_value`	发生的异常实例（如果有）。类型： `BaseException \| None` 默认值： `None`
`traceback`	与异常关联的回溯对象（如果有）。类型： `TracebackType \| None` 默认值： `None`

aenter `async` ¶

__aenter__() -> RunTree

异步进入上下文管理器。

返回	描述
`RunTree`	run_trees.RunTree：新创建的运行。

aexit `async` ¶

__aexit__(
    exc_type: type[BaseException] | None = None,
    exc_value: BaseException | None = None,
    traceback: TracebackType | None = None,
) -> None

异步退出上下文管理器。

参数	描述
`exc_type`	发生的异常类型（如果有）。类型： `type[BaseException] \| None` 默认值： `None`
`exc_value`	发生的异常实例（如果有）。类型： `BaseException \| None` 默认值： `None`
`traceback`	与异常关联的回溯对象（如果有）。类型： `TracebackType \| None` 默认值： `None`

get_current_run_tree ¶

get_current_run_tree() -> RunTree | None

获取当前的运行树。

get_tracing_context ¶

get_tracing_context(context: Context | None = None) -> dict[str, Any]

获取当前的追踪上下文。

tracing_context ¶

tracing_context(
    *,
    project_name: str | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    parent: RunTree | Mapping | str | Literal[False] | None = None,
    enabled: bool | Literal["local"] | None = None,
    client: Client | None = None,
    replicas: Sequence[WriteReplica] | None = None,
    distributed_parent_id: str | None = None,
    **kwargs: Any,
) -> Generator[None, None, None]

为代码块设置追踪上下文。

参数	描述
`project_name`	要将运行记录到的项目名称。默认为 None。类型: `str \| None` 默认值: `None`
`tags`	要添加到运行的标签。默认为 None。类型: `list[str] \| None` 默认值: `None`
`metadata`	要添加到运行的元数据。默认为 None。类型: `dict[str, Any] \| None` 默认值: `None`
`parent`	用于上下文的父运行。可以是 Run/RunTree 对象、请求头（用于分布式追踪）或点分隔顺序字符串。默认为 None。类型： `RunTree \| Mapping \| str \| Literal[False] \| None` 默认值： `None`
`client`	用于将运行记录到 LangSmith 的客户端。默认为 None，类型： `Client \| None` 默认值： `None`
`enabled`	追踪是否启用。默认为 None，表示将使用当前上下文值或环境变量。类型： `bool \| Literal['local'] \| None` 默认值： `None`
`replicas`	用于发送运行的 WriteReplica 字典序列。例如：[{"api_url": "https://api.example.com", "api_key": "key", "project_name": "proj"}] 或 [{"project_name": "my_experiment", "updates": {"reference_example_id": None}}] 类型： `Sequence[WriteReplica] \| None` 默认值： `None`
`distributed_parent_id`	用于分布式追踪的分布式父 ID。默认为 None。类型: `str \| None` 默认值: `None`

is_traceable_function ¶

is_traceable_function(func: Any) -> TypeGuard[SupportsLangsmithExtra[P, R]]

检查一个函数是否被 @traceable 装饰器修饰。

ensure_traceable ¶

ensure_traceable(
    func: Callable[P, R],
    *,
    name: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    tags: list[str] | None = None,
    client: Client | None = None,
    reduce_fn: Callable[[Sequence], dict | str] | None = None,
    project_name: str | None = None,
    process_inputs: Callable[[dict], dict] | None = None,
    process_outputs: Callable[..., dict] | None = None,
    process_chunk: Callable | None = None,
) -> SupportsLangsmithExtra[P, R]

确保一个函数是可追踪的。

is_async ¶

is_async(func: Callable) -> bool

检查函数或被包装的函数是否为异步函数。

traceable ¶

traceable(*args: Any, **kwargs: Any) -> Callable | Callable[[Callable], Callable]

使用 Langsmith 追踪一个函数。

参数	描述
`run_type`	要创建的运行（span）的类型。示例：llm, chain, tool, prompt, retriever 等。默认为“chain”。
`name`	运行的名称。默认为函数名称。
`metadata`	要添加到运行的元数据。默认为 None。
`tags`	要添加到运行的标签。默认为 None。
`client`	用于将运行记录到 LangSmith 的客户端。默认为 None，这将使用默认客户端。
`reduce_fn`	如果函数返回生成器，则用于聚合函数输出的函数。默认为 None，这意味着值将被记录为列表。注意：如果迭代器永不耗尽（例如，函数返回一个无限生成器），此函数将永远不会被调用，并且运行本身将卡在挂起状态。
`project_name`	要将运行记录到的项目名称。默认为 None，这将使用默认项目。
`process_inputs`	用于输入的自定义序列化/处理函数。默认为 None。
`process_outputs`	用于输出的自定义序列化/处理函数。默认为 None。
`dangerously_allow_filesystem`	是否允许文件系统访问以获取附件。默认为 False。引用本地文件路径的追踪将被上传到 LangSmith。通常，网络托管的应用程序不应使用此功能，因为引用的文件通常在用户的机器上，而不是在主机上。

返回	描述
`Callable \| Callable[[Callable], Callable]`	Union[Callable, Callable[[Callable], Callable]]：被装饰的函数。

注意

要求在环境中将 LANGSMITH_TRACING_V2 设置为 'true'。

示例

基本用法

.. code-block:: python

@traceable
def my_function(x: float, y: float) -> float:
    return x + y


my_function(5, 6)


@traceable
async def my_async_function(query_params: dict) -> dict:
    async with httpx.AsyncClient() as http_client:
        response = await http_client.get(
            "https://api.example.com/data",
            params=query_params,
        )
        return response.json()


asyncio.run(my_async_function({"param": "value"}))

使用生成器流式传输数据

.. code-block:: python

@traceable
def my_generator(n: int) -> Iterable:
    for i in range(n):
        yield i


for item in my_generator(5):
    print(item)

异步流式传输数据

.. code-block:: python

@traceable
async def my_async_generator(query_params: dict) -> Iterable:
    async with httpx.AsyncClient() as http_client:
        response = await http_client.get(
            "https://api.example.com/data",
            params=query_params,
        )
        for item in response.json():
            yield item


async def async_code():
    async for item in my_async_generator({"param": "value"}):
        print(item)


asyncio.run(async_code())

指定运行类型和名称

.. code-block:: python

@traceable(name="CustomName", run_type="tool")
def another_function(a: float, b: float) -> float:
    return a * b


another_function(5, 6)

使用自定义元数据和标签进行日志记录

.. code-block:: python

@traceable(
    metadata={"version": "1.0", "author": "John Doe"}, tags=["beta", "test"]
)
def tagged_function(x):
    return x**2


tagged_function(5)

指定自定义客户端和项目名称

.. code-block:: python

custom_client = Client(api_key="your_api_key")


@traceable(client=custom_client, project_name="My Special Project")
def project_specific_function(data):
    return data


project_specific_function({"data": "to process"})

手动传递 langsmith_extra

.. code-block:: python

@traceable
def manual_extra_function(x):
    return x**2


manual_extra_function(5, langsmith_extra={"metadata": {"version": "1.0"}})

as_runnable ¶

as_runnable(traceable_fn: Callable) -> Runnable

将一个被 LangSmith @traceable 装饰器包装的函数转换为 Runnable。

参数	描述
`traceable_fn`	被 @traceable 装饰器包装的函数。类型： `Callable`

返回	描述
`Runnable`	一个保持一致 LangSmith 追踪上下文的 Runnable 对象。类型： `Runnable`

引发	描述
`ImportError`	如果未安装 langchain 模块。
`ValueError`	如果提供的函数未被 @traceable 装饰器包装。

示例

@traceable ... def my_function(input_data): ... # 函数实现 ... pass runnable = as_runnable(my_function)

运行助手 (Run Helpers)

langsmith.run_helpers ¶

LangSmithExtra ¶

name instance-attribute ¶

reference_example_id instance-attribute ¶

run_extra instance-attribute ¶

parent instance-attribute ¶

run_tree instance-attribute ¶

project_name instance-attribute ¶

metadata instance-attribute ¶

tags instance-attribute ¶

run_id instance-attribute ¶

client instance-attribute ¶

on_end instance-attribute ¶

SupportsLangsmithExtra ¶

__call__ ¶

trace ¶

__init__ ¶

__enter__ ¶

__exit__ ¶

__aenter__ async ¶

__aexit__ async ¶

get_current_run_tree ¶

get_tracing_context ¶

tracing_context ¶

is_traceable_function ¶

ensure_traceable ¶

is_async ¶

traceable ¶

as_runnable ¶

name `instance-attribute` ¶

reference_example_id `instance-attribute` ¶

run_extra `instance-attribute` ¶

parent `instance-attribute` ¶

run_tree `instance-attribute` ¶

project_name `instance-attribute` ¶

metadata `instance-attribute` ¶

tags `instance-attribute` ¶

run_id `instance-attribute` ¶

client `instance-attribute` ¶

on_end `instance-attribute` ¶

call ¶

init ¶

enter ¶

exit ¶

aenter `async` ¶

aexit `async` ¶