Source code for langchain_core.tracers.run_collector
"""一个收集所有嵌套运行的跟踪器列表。"""
from typing import Any, List, Optional, Union
from uuid import UUID
from langchain_core.tracers.base import BaseTracer
from langchain_core.tracers.schemas import Run
[docs]class RunCollectorCallbackHandler(BaseTracer):
"""收集所有嵌套运行的跟踪器列表。
此跟踪器对于检查和评估目的很有用。
参数
----------
example_id:Optional[Union[UUID, str]],默认值为None
要跟踪的示例的ID。可以是UUID或字符串。"""
name: str = "run-collector_callback_handler"
[docs] def __init__(
self, example_id: Optional[Union[UUID, str]] = None, **kwargs: Any
) -> None:
"""初始化RunCollectorCallbackHandler。
参数
----------
example_id:Optional[Union[UUID, str]],默认值为None
要被跟踪的示例的ID。可以是UUID或字符串。
"""
super().__init__(**kwargs)
self.example_id = (
UUID(example_id) if isinstance(example_id, str) else example_id
)
self.traced_runs: List[Run] = []
def _persist_run(self, run: Run) -> None:
"""将运行添加到traced_runs列表中以持久化。
参数
----------
run : Run
要持久化的运行。
"""
run_ = run.copy()
run_.reference_example_id = self.example_id
self.traced_runs.append(run_)