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