trace#

class langsmith.run_helpers.trace(name: str, run_type: Literal['tool', 'chain', 'llm', 'retriever', 'embedding', 'prompt', 'parser'] = '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: UUID | str | None = None, reference_example_id: UUID | str | None = None, exceptions_to_handle: Tuple[Type[BaseException], ...] | None = None, attachments: Dict[str, Tuple[str, bytes] | Attachment | Tuple[str, Path]] | None = None, **kwargs: Any)[source]#

在上下文中管理 LangSmith 运行。

此类可以用作同步和异步上下文管理器。

参数:
  • name (str) – 运行的名称。

  • run_type (ls_client.RUN_TYPE_T, 可选) – 运行类型 (例如, “chain”, “llm”, “tool”)。默认为 “chain”。

  • inputs (Optional[Dict], 可选) – 运行的初始输入数据。默认为 None。

  • project_name (Optional[str], 可选) – 与运行关联的项目名称。默认为 None。

  • parent (Optional[Union[run_trees.RunTree, str, Mapping]], 可选) – 父运行。可以是 RunTree、点分隔的顺序字符串或跟踪标头。默认为 None。

  • tags (Optional[List[str]], 可选) – 运行的标签列表。默认为 None。

  • metadata (Optional[Mapping[str, Any]], 可选) – 运行的附加元数据。默认为 None。

  • client (Optional[ls_client.Client], 可选) – LangSmith 客户端,用于自定义设置。默认为 None。

  • run_id (Optional[ls_client.ID_TYPE], 可选) – 运行的预设标识符。默认为 None。

  • reference_example_id (Optional[ls_client.ID_TYPE], 可选) – 将运行与数据集示例关联。仅用于评估中的根运行。默认为 None。

  • exceptions_to_handle (Optional[Tuple[Type[BaseException], ...]], 可选) – 要忽略的异常类型。默认为 None。

  • extra (Optional[Dict], 可选) – 要发送到 LangSmith 的额外数据。请改用 “metadata”。默认为 None。

  • attachments (Optional[schemas.Attachments])

  • kwargs (Any)

示例

同步用法

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

异步用法

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

处理特定异常

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

初始化跟踪上下文管理器。

如果传递了不支持的 kwargs,则发出警告。

方法

__init__(name[, run_type, inputs, extra, ...])

初始化跟踪上下文管理器。

__init__(name: str, run_type: Literal['tool', 'chain', 'llm', 'retriever', 'embedding', 'prompt', 'parser'] = '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: UUID | str | None = None, reference_example_id: UUID | str | None = None, exceptions_to_handle: Tuple[Type[BaseException], ...] | None = None, attachments: Dict[str, Tuple[str, bytes] | Attachment | Tuple[str, Path]] | None = None, **kwargs: Any)[source]#

初始化跟踪上下文管理器。

如果传递了不支持的 kwargs,则发出警告。

参数:
  • name (str)

  • run_type (Literal['tool', 'chain', 'llm', 'retriever', 'embedding', 'prompt', 'parser'])

  • inputs (Dict | None)

  • extra (Dict | None)

  • project_name (str | None)

  • parent (RunTree | str | Mapping | Literal['ignore'] | None)

  • tags (List[str] | None)

  • metadata (Mapping[str, Any] | None)

  • client (Client | None)

  • run_id (UUID | str | None)

  • reference_example_id (UUID | str | None)

  • exceptions_to_handle (Tuple[Type[BaseException], ...] | None)

  • attachments (Dict[str, Tuple[str, bytes] | Attachment | Tuple[str, Path]] | None)

  • kwargs (Any)