_run_batch_evaluators() — langchain Function Reference

Architecture documentation for the _run_batch_evaluators() function in runner_utils.py from the langchain codebase.

Function python LangChainCore Runnables called by 1

Entity Profile

LangChainCore→ Runnables→ _run_batch_evaluators() — langchain Function Reference

Dependency Diagram

graph TD
  31505a45_4c7a_ea20_3a86_a98f76b1d2b7["_run_batch_evaluators()"]
  3aaa6e94_b6a8_1c13_86d0_1709a1d93909["_DatasetRunContainer"]
  31505a45_4c7a_ea20_3a86_a98f76b1d2b7 -->|defined in| 3aaa6e94_b6a8_1c13_86d0_1709a1d93909
  7bd0a459_a7f0_719c_faf9_2cf0ffd65a8c["_collect_test_results()"]
  7bd0a459_a7f0_719c_faf9_2cf0ffd65a8c -->|calls| 31505a45_4c7a_ea20_3a86_a98f76b1d2b7
  style 31505a45_4c7a_ea20_3a86_a98f76b1d2b7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/smith/evaluation/runner_utils.py lines 1126–1149

    def _run_batch_evaluators(self, runs: dict[str, Run]) -> list[dict]:
        evaluators = self.batch_evaluators
        if not evaluators:
            return []
        runs_list = [runs[str(example.id)] for example in self.examples]
        aggregate_feedback = []
        with concurrent.futures.ThreadPoolExecutor() as executor:
            for evaluator in evaluators:
                try:
                    result = evaluator(runs_list, self.examples)
                    if isinstance(result, EvaluationResult):
                        result = result.model_dump()
                    aggregate_feedback.append(cast("dict", result))
                    executor.submit(
                        self.client.create_feedback,
                        **result,
                        run_id=None,
                        project_id=self.project.id,
                    )
                except Exception:
                    logger.exception(
                        "Error running batch evaluator %s", repr(evaluator)
                    )
        return aggregate_feedback