RunTree#

class langsmith.run_trees.RunTree(
*,
id: UUID = None,
name: str,
start_time: datetime = None,
run_type: str = 'chain',
end_time: datetime | None = None,
extra: dict = None,
error: str | None = None,
serialized: dict | None = None,
events: list[dict] = None,
inputs: dict = None,
outputs: dict | None = None,
reference_example_id: UUID | None = None,
parent_run_id: UUID | None = None,
tags: list[str] | None = None,
attachments: dict[str, tuple[str, bytes] | Attachment | tuple[str, Path]] | dict[str, AttachmentInfo] = None,
parent_run: RunTree | None = None,
parent_dotted_order: str | None = None,
child_runs: list[RunTree] = None,
project_name: str = None,
project_id: UUID | None = None,
ls_client: Any | None = None,
dotted_order: str = '',
trace_id: UUID = '',
dangerously_allow_filesystem: bool | None = False,
replicas: Sequence[tuple[str, dict | None]] | None = None,
)[源]#

用于发布运行的具有反向引用的运行模式。

通过解析和验证关键字参数中的输入数据来创建新模型。

如果输入数据无法解析为有效模型,则引发 ValidationError。

属性

客户端

返回客户端。

元数据

检索元数据(如果有)。

修订ID

检索修订ID(如果有)。

方法

__init__(**data)

通过解析和验证关键字参数中的输入数据来创建新模型。

add_event(events)

将事件添加到事件列表。

add_inputs(inputs)

将给定输出更新插入到运行中。

add_metadata(metadata)

向运行添加元数据。

add_outputs(outputs)

将给定输出更新插入到运行中。

add_tags(tags)

向运行添加标签。

construct([_fields_set])

从受信任或预验证的数据创建新模型,设置 __dict__ 和 __fields_set__。

copy(*[, include, exclude, update, deep])

复制模型,可选择要包含、排除和更改的字段。

create_child(name[, run_type, run_id, ...])

向运行树添加一个子运行。

dict(*[, include, exclude, by_alias, ...])

生成模型的字典表示,可选指定要包含或排除的字段。

end(*[, outputs, error, end_time, events, ...])

设置运行和所有子运行的结束时间。

ensure_dotted_order(values)

确保运行的点状顺序。

from_dotted_order(dotted_order, **kwargs)

从提供的点状顺序创建新的“子”span。

from_headers(headers, **kwargs)

从提供的头信息创建新的“父”span。

from_orm(obj)

from_runnable_config(config, **kwargs)

从提供的可运行配置创建新的“子”span。

获取URL()

返回运行的URL。

infer_defaults(values)

为运行分配名称。

json(*[, include, exclude, by_alias, ...])

生成模型的 JSON 表示,includeexclude 参数与 dict() 相同。

parse_file(path, *[, content_type, ...])

parse_obj(obj)

parse_raw(b, *[, content_type, encoding, ...])

patch(*[, exclude_inputs])

在后台线程中将运行树修补到 API。

post([exclude_child_runs])

异步地将运行树发布到 API。

schema([by_alias, ref_template])

schema_json(*[, by_alias, ref_template])

set(*[, inputs, outputs, tags, metadata, ...])

设置运行的输入、输出、标签和元数据。

转为头信息()

将 RunTree 作为头信息字典返回。

update_forward_refs(**localns)

尝试根据此模型、globalns 和 localns 更新字段上的 ForwardRefs。

validate(value)

等待()

等待所有 _futures 完成。

参数:
  • id (UUID)

  • name (str)

  • start_time (datetime)

  • run_type (str)

  • end_time (datetime | None)

  • extra (dict)

  • error (str | None)

  • serialized (dict | None)

  • events (list[dict])

  • inputs (dict)

  • outputs (dict | None)

  • reference_example_id (UUID | None)

  • parent_run_id (UUID | None)

  • tags (list[str] | None)

  • attachments (dict[str, tuple[str, bytes] | Attachment | tuple[str, Path]] | dict[str, AttachmentInfo])

  • parent_run (RunTree | None)

  • parent_dotted_order (str | None)

  • child_runs (list[RunTree])

  • project_name (str)

  • project_id (UUID | None)

  • ls_client (Any | None)

  • dotted_order (str)

  • trace_id (UUID)

  • dangerously_allow_filesystem (bool | None)

  • replicas (Sequence[tuple[str, dict | None]] | None)

__init__(**data: Any) None#

通过解析和验证关键字参数中的输入数据来创建新模型。

如果输入数据无法解析为有效模型,则引发 ValidationError。

参数:

data (Any)

返回类型:

None

add_event(
events: RunEvent | Sequence[RunEvent] | Sequence[dict] | dict | str,
) None[源]#

将事件添加到事件列表。

参数:
  • (Union[ls_schemas.RunEvent (events) – Sequence[dict], dict, str]): 要添加的事件。它可以是单个事件、事件序列、字典序列、字典或字符串。

  • Sequence[ls_schemas.RunEvent] – Sequence[dict], dict, str]): 要添加的事件。它可以是单个事件、事件序列、字典序列、字典或字符串。

  • events (RunEvent | Sequence[RunEvent] | Sequence[dict] | dict | str)

返回类型:

None

:paramSequence[dict], dict, str]):

要添加的事件。它可以是单个事件、事件序列、字典序列、字典或字符串。

返回:

None

参数:

events (RunEvent | Sequence[RunEvent] | Sequence[dict] | dict | str)

返回类型:

None

add_inputs(inputs: dict[str, Any]) None[源]#

将给定输出更新插入到运行中。

参数:
  • outputs (Dict[str, Any]) – 包含要添加的输出的字典。

  • inputs (dict[str, Any])

返回:

None

返回类型:

None

add_metadata(
metadata: dict[str, Any],
) None[源]#

向运行添加元数据。

参数:

metadata (dict[str, Any])

返回类型:

None

add_outputs(outputs: dict[str, Any]) None[源]#

将给定输出更新插入到运行中。

参数:

outputs (Dict[str, Any]) – 包含要添加的输出的字典。

返回:

None

返回类型:

None

add_tags(
tags: Sequence[str] | str,
) None[源]#

向运行添加标签。

参数:

tags (Sequence[str] | str)

返回类型:

None

classmethod construct(
_fields_set: SetStr | None = None,
**values: Any,
) Model#

从受信任或预验证的数据创建新模型,设置 __dict__ 和 __fields_set__。默认值被遵守,但不会执行其他验证。行为如同设置了 Config.extra = ‘allow’,因为它会添加所有传递的值

参数:
  • _fields_set (SetStr | None)

  • values (Any)

返回类型:

Model

copy(
*,
include: AbstractSetIntStr | MappingIntStrAny | None = None,
exclude: AbstractSetIntStr | MappingIntStrAny | None = None,
update: DictStrAny | None = None,
deep: bool = False,
) Model#

复制模型,可选择要包含、排除和更改的字段。

参数:
  • include (AbstractSetIntStr | MappingIntStrAny | None) – 要包含在新模型中的字段

  • exclude (AbstractSetIntStr | MappingIntStrAny | None) – 要从新模型中排除的字段,与值一样,此参数优先于 include

  • update (DictStrAny | None) – 要在新模型中更改/添加的值。注意:在创建新模型之前不会验证数据:您应该信任此数据

  • deep (bool) – 设置为 True 以对模型进行深拷贝

  • self (Model)

返回:

新模型实例

返回类型:

Model

create_child(
name: str,
run_type: Literal['tool', 'chain', 'llm', 'retriever', 'embedding', 'prompt', 'parser'] = 'chain',
*,
run_id: UUID | str | None = None,
serialized: dict | None = None,
inputs: dict | None = None,
outputs: dict | None = None,
error: str | None = None,
reference_example_id: UUID | None = None,
start_time: datetime | None = None,
end_time: datetime | None = None,
tags: list[str] | None = None,
extra: dict | None = None,
attachments: dict[str, tuple[str, bytes] | Attachment | tuple[str, Path]] | None = None,
) RunTree[源]#

向运行树添加一个子运行。

参数:
  • name (str)

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

  • run_id (UUID | str | None)

  • serialized (dict | None)

  • inputs (dict | None)

  • outputs (dict | None)

  • error (str | None)

  • reference_example_id (UUID | None)

  • start_time (datetime | None)

  • end_time (datetime | None)

  • tags (list[str] | None)

  • extra (dict | None)

  • attachments (dict[str, tuple[str, bytes] | Attachment | tuple[str, Path]] | None)

返回类型:

RunTree

dict(
*,
include: AbstractSetIntStr | MappingIntStrAny | None = None,
exclude: AbstractSetIntStr | MappingIntStrAny | None = None,
by_alias: bool = False,
skip_defaults: bool | None = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
) DictStrAny#

生成模型的字典表示,可选指定要包含或排除的字段。

参数:
  • include (AbstractSetIntStr | MappingIntStrAny | None)

  • exclude (AbstractSetIntStr | MappingIntStrAny | None)

  • by_alias (bool)

  • skip_defaults (bool | None)

  • exclude_unset (bool)

  • exclude_defaults (bool)

  • exclude_none (bool)

返回类型:

DictStrAny

end(
*,
outputs: dict | None = None,
error: str | None = None,
end_time: datetime | None = None,
events: Sequence[RunEvent] | None = None,
metadata: dict[str, Any] | None = None,
) None[源]#

设置运行和所有子运行的结束时间。

参数:
  • outputs (dict | None)

  • error (str | None)

  • end_time (datetime | None)

  • events (Sequence[RunEvent] | None)

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

返回类型:

None

classmethod ensure_dotted_order(values: dict) dict[源]#

确保运行的点状顺序。

参数:

values (dict)

返回类型:

dict

classmethod from_dotted_order(
dotted_order: str,
**kwargs: Any,
) RunTree[源]#

从提供的点状顺序创建新的“子”span。

返回:

新的 span。

返回类型:

RunTree

参数:
  • dotted_order (str)

  • kwargs (Any)

classmethod from_headers(
headers: Mapping[str | bytes, str | bytes],
**kwargs: Any,
) RunTree | None[源]#

从提供的头信息创建新的“父”span。

从头信息中提取父 span 信息并创建新的 span。元数据和标签从 baggage header 中提取。点状顺序和跟踪 ID 从 trace header 中提取。

返回:

新的 span,如果

未找到父 span 信息,则为 None。

返回类型:

Optional[RunTree]

参数:
  • headers (Mapping[str | bytes, str | bytes])

  • kwargs (Any)

classmethod from_orm(obj: Any) Model#
参数:

obj (Any)

返回类型:

Model

classmethod from_runnable_config(
config: dict | None,
**kwargs: Any,
) RunTree | None[源]#

从提供的可运行配置创建新的“子”span。

需要安装 langchain。

返回:

新的 span,如果

未找到父 span 信息,则为 None。

返回类型:

Optional[RunTree]

参数:
  • config (dict | None)

  • kwargs (Any)

get_url() str[源]#

返回运行的URL。

返回类型:

str

classmethod infer_defaults(values: dict) dict[源]#

为运行分配名称。

参数:

values (dict)

返回类型:

dict

json(
*,
include: AbstractSetIntStr | MappingIntStrAny | None = None,
exclude: AbstractSetIntStr | MappingIntStrAny | None = None,
by_alias: bool = False,
skip_defaults: bool | None = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Callable[[Any], Any] | None = None,
models_as_dict: bool = True,
**dumps_kwargs: Any,
) str#

生成模型的 JSON 表示,includeexclude 参数与 dict() 相同。

encoder 是一个可选函数,作为 default 参数提供给 json.dumps(),其他参数与 json.dumps() 相同。

参数:
  • include (AbstractSetIntStr | MappingIntStrAny | None)

  • exclude (AbstractSetIntStr | MappingIntStrAny | None)

  • by_alias (bool)

  • skip_defaults (bool | None)

  • exclude_unset (bool)

  • exclude_defaults (bool)

  • exclude_none (bool)

  • encoder (Callable[[Any], Any] | None)

  • models_as_dict (bool)

  • dumps_kwargs (Any)

返回类型:

str

classmethod parse_file(
path: str | Path,
*,
content_type: str = None,
encoding: str = 'utf8',
proto: Protocol = None,
allow_pickle: bool = False,
) Model#
参数:
  • path (str | Path)

  • content_type (str)

  • encoding (str)

  • proto (Protocol)

  • allow_pickle (bool)

返回类型:

Model

classmethod parse_obj(obj: Any) Model#
参数:

obj (Any)

返回类型:

Model

classmethod parse_raw(
b: str | bytes,
*,
content_type: str = None,
encoding: str = 'utf8',
proto: Protocol = None,
allow_pickle: bool = False,
) Model#
参数:
  • b (str | bytes)

  • content_type (str)

  • encoding (str)

  • proto (Protocol)

  • allow_pickle (bool)

返回类型:

Model

patch(*, exclude_inputs: bool = False) None[源]#

在后台线程中将运行树修补到 API。

参数:

exclude_inputs (bool) – 是否从补丁请求中排除输入。

返回类型:

None

post(exclude_child_runs: bool = True) None[源]#

异步地将运行树发布到 API。

参数:

exclude_child_runs (bool)

返回类型:

None

classmethod schema(
by_alias: bool = True,
ref_template: str = '#/definitions/{model}',
) DictStrAny#
参数:
  • by_alias (bool)

  • ref_template (str)

返回类型:

DictStrAny

classmethod schema_json(
*,
by_alias: bool = True,
ref_template: str = '#/definitions/{model}',
**dumps_kwargs: Any,
) str#
参数:
  • by_alias (bool)

  • ref_template (str)

  • dumps_kwargs (Any)

返回类型:

str

set(
*,
inputs: ~collections.abc.Mapping[str,
~typing.Any] | None = <object object>,
outputs: ~collections.abc.Mapping[str,
~typing.Any] | None = <object object>,
tags: ~collections.abc.Sequence[str] | None = <object object>,
metadata: ~collections.abc.Mapping[str,
~typing.Any] | None = <object object>,
usage_metadata: ~langsmith.schemas.ExtractedUsageMetadata | None = <object object>,
) None[源]#

设置运行的输入、输出、标签和元数据。

如果执行此操作,它将覆盖 end() 方法的默认行为,从而忽略 @traceable 装饰器添加的新输出(否则这些输出将被添加)。

如果您的 LangChain 或 LangGraph 版本足够新,此操作还将覆盖 LangChainTracer 的默认行为。

参数:
  • inputs (Mapping[str, Any] | None) – 要设置的输入。

  • outputs (Mapping[str, Any] | None) – 要设置的输出。

  • tags (Sequence[str] | None) – 要设置的标签。

  • metadata (Mapping[str, Any] | None) – 要设置的元数据。

  • usage_metadata (ExtractedUsageMetadata | None) – 要设置的使用信息。

返回:

None

返回类型:

None

to_headers() dict[str, str][源]#

将 RunTree 作为头信息字典返回。

返回类型:

dict[str, str]

classmethod update_forward_refs(**localns: Any) None#

尝试根据此模型、globalns 和 localns 更新字段上的 ForwardRefs。

参数:

localns (Any)

返回类型:

None

classmethod validate(value: Any) Model#
参数:

value (Any)

返回类型:

Model

wait() None[源]#

等待所有 _futures 完成。

返回类型:

None