init() — langchain Function Reference

Architecture documentation for the __init__() function in evaluation.py from the langchain codebase.

Function python Observability Tracers

Entity Profile

Observability→ Tracers→ __init__() — langchain Function Reference

Dependency Diagram

graph TD
  cbca36a7_cf61_b4f4_35e4_e5b9a1bc5b1e["__init__()"]
  d98d30f4_d5fd_24fc_54d0_e2f82eecc3cd["EvaluatorCallbackHandler"]
  cbca36a7_cf61_b4f4_35e4_e5b9a1bc5b1e -->|defined in| d98d30f4_d5fd_24fc_54d0_e2f82eecc3cd
  style cbca36a7_cf61_b4f4_35e4_e5b9a1bc5b1e fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/tracers/evaluation.py lines 72–116

    def __init__(
        self,
        evaluators: Sequence[langsmith.RunEvaluator],
        client: langsmith.Client | None = None,
        example_id: UUID | str | None = None,
        skip_unfinished: bool = True,  # noqa: FBT001,FBT002
        project_name: str | None = "evaluators",
        max_concurrency: int | None = None,
        **kwargs: Any,
    ) -> None:
        """Create an EvaluatorCallbackHandler.

        Args:
            evaluators: The run evaluators to apply to all top level runs.
            client: The LangSmith client instance to use for evaluating the runs.

                If not specified, a new instance will be created.
            example_id: The example ID to be associated with the runs.
            skip_unfinished: Whether to skip unfinished runs.
            project_name: The LangSmith project name to be organize eval chain runs
                under.
            max_concurrency: The maximum number of concurrent evaluators to run.
        """
        super().__init__(**kwargs)
        self.example_id = (
            UUID(example_id) if isinstance(example_id, str) else example_id
        )
        self.client = client or langchain_tracer.get_client()
        self.evaluators = evaluators
        if max_concurrency is None:
            self.executor = _get_executor()
        elif max_concurrency > 0:
            self.executor = ThreadPoolExecutor(max_workers=max_concurrency)
            weakref.finalize(
                self,
                lambda: cast("ThreadPoolExecutor", self.executor).shutdown(wait=True),
            )
        else:
            self.executor = None
        self.futures = weakref.WeakSet[Future[None]]()
        self.skip_unfinished = skip_unfinished
        self.project_name = project_name
        self.logged_eval_results = {}
        self.lock = threading.Lock()
        _TRACERS.add(self)