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)