_DatasetRunContainer Class — langchain Architecture

Source Code

libs/langchain/langchain_classic/smith/evaluation/runner_utils.py lines 1094–1293

class _DatasetRunContainer:
    """A container to help manage the state of a eval run."""

    client: Client
    project: TracerSession
    wrapped_model: MCF
    examples: list[Example]
    configs: list[RunnableConfig]
    batch_evaluators: list[smith_eval_config.BATCH_EVALUATOR_LIKE] | None = None

    def _merge_test_outputs(
        self,
        batch_results: list,
        all_eval_results: dict[str, _RowResult],
    ) -> dict:
        results: dict = {}
        for example, output in zip(self.examples, batch_results, strict=False):
            row_result = all_eval_results.get(str(example.id), {})
            results[str(example.id)] = {
                "input": example.inputs,
                "feedback": row_result.get("feedback", []),
                "execution_time": row_result.get("execution_time"),
                "run_id": row_result.get("run_id"),
            }
            if isinstance(output, EvalError):
                results[str(example.id)]["Error"] = output.Error
            else:
                results[str(example.id)]["output"] = output
            if example.outputs:
                results[str(example.id)]["reference"] = example.outputs
        return results

    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

    def _collect_metrics(self) -> tuple[dict[str, _RowResult], dict[str, Run]]:
        all_eval_results: dict = {}
        all_runs: dict = {}
        for c in self.configs:
            for callback in cast("list", c["callbacks"]):
                if isinstance(callback, EvaluatorCallbackHandler):
                    eval_results = callback.logged_eval_results
                    for (_, example_id), v in eval_results.items():
                        all_eval_results.setdefault(str(example_id), {}).update(
                            {"feedback": v},
                        )
                elif isinstance(callback, LangChainTracer):
                    run = callback.latest_run
                    execution_time = (
                        (run.end_time - run.start_time).total_seconds()
                        if run and run.end_time
                        else None
                    )
                    run_id = str(run.id) if run else None
                    all_eval_results.setdefault(str(callback.example_id), {}).update(
                        {
                            "execution_time": execution_time,
                            "run_id": run_id,
                            "run": run,