Function run_eval

openkb/skill/evaluator.py:389–462

Run trigger-accuracy + body-alignment evaluation. Args: skill_dir: `` /output/skills/ `` model: LiteLLM model string from KB config eval_set: pre-generated prompts; if None, generate fresh count: how many should-trigger + should-not prompts to generate

(
    skill_dir: Path,
    *,
    model: str,
    eval_set: list[EvalPrompt] | None = None,
    count: int = EVAL_DEFAULT_COUNT,
)

async def run_eval(
    skill_dir: Path,
    *,
    model: str,
    eval_set: list[EvalPrompt] | None = None,
    count: int = EVAL_DEFAULT_COUNT,
) -> EvalResult:
    """Run trigger-accuracy + body-alignment evaluation.

    Args:
        skill_dir: ``<kb>/output/skills/<name>``
        model: LiteLLM model string from KB config
        eval_set: pre-generated prompts; if None, generate fresh
        count: how many should-trigger + should-not prompts to generate
    """
    if eval_set is None:
        eval_set = await generate_eval_set(skill_dir, model=model, count=count)

    desc = _read_description(skill_dir)
    content = _skill_content_block(skill_dir)
    result = EvalResult(prompts=eval_set)

    # Run grading concurrently. Each prompt is independent — graders read
    # the same `desc`/`content` strings and produce results that are then
    # appended to `result` in eval_set order below, so concurrent
    # execution is correctness-preserving. A semaphore caps simultaneous
    # LLM calls to avoid hitting provider rate limits.
    sem = asyncio.Semaphore(EVAL_CONCURRENCY)

    async def _trigger(p: EvalPrompt) -> Literal["trigger", "no-trigger"]:
        async with sem:
            return await grade_one(desc, p.question, model=model)

    async def _coverage(
        p: EvalPrompt,
    ) -> tuple[Literal["supported", "unsupported", "ambiguous"], str]:
        async with sem:
            return await grade_coverage(content, p.question, model=model)

    trigger_tasks = [_trigger(p) for p in eval_set]
    # Body alignment only meaningful on questions the skill claims to
    # handle — for should-not questions the body is correctly empty of
    # relevant material.
    coverage_prompts = [p for p in eval_set if p.expected == "trigger"]
    coverage_tasks = [_coverage(p) for p in coverage_prompts]

    # return_exceptions=True so one failed grader doesn't discard the
    # other ~29 successful gradings. Errored prompts are surfaced
    # separately on the result and excluded from rate denominators.
    trigger_results, coverage_results = await asyncio.gather(
        asyncio.gather(*trigger_tasks, return_exceptions=True),
        asyncio.gather(*coverage_tasks, return_exceptions=True),
    )

    # Walk inputs in original order so `result.*` lists are deterministic
    # even though the gather() above completed out of order.
    for prompt, graded in zip(eval_set, trigger_results):
        if isinstance(graded, BaseException):