evaluate_chunking

Instructions

Implementation Reference

• src/chunktuner/mcp/service.py:79-126 (handler)

  Core implementation of evaluate_chunking_impl - handles dry-run cost estimates and full evaluation with embeddings. Validates path, strategies, ingests docs, and runs evaluation via Evaluator.

  def evaluate_chunking_impl(
      path: str,
      use_case: str,
      *,
      content_type: str | None = None,
      strategies: list[str] | None = None,
      max_docs: int = 20,
      top_k: int = 5,
      dry_run: bool = False,
      embedding_model: str | None = None,
  ) -> dict:
      p = require_under_base(path)
      if not p.exists():
          raise ValueError("path does not exist")
      if strategies is not None:
          _validate_strategies(strategies)
      names = strategies or ["fixed_tokens", "recursive_character"]
      if dry_run:
          fi = FileIngestor(root=p.parent if p.is_file() else p)
          docs = fi.ingest_path(p) if p.is_file() else fi.ingest_dir(p)
          docs = docs[:max_docs]
          grid: dict[str, list[dict]] = {
              n: build_full_registry().get(n).default_param_grid() for n in names
          }
          est = CostEstimator().estimate(
              docs, names, grid, embedding_model or "text-embedding-3-small"
          )
          return est.model_dump()
      embed = (
          LiteLLMEmbeddingFunction(embedding_model) if embedding_model else DummyEmbeddingFunction()
      )
      fi = FileIngestor(root=p.parent if p.is_file() else p)
      docs = fi.ingest_path(p) if p.is_file() else fi.ingest_dir(p)
      docs = docs[:max_docs]
      ds = trivial_dataset_for_docs(docs)
      reg = build_full_registry()
      ev = Evaluator(embed, top_k=top_k)
      scorer = ScoreCalculator(cast(UseCase, use_case))
      results = []
      for n in names:
          strat = reg.get(n)
          for params in strat.default_param_grid():
              cfg = ChunkConfig(name=n, params=dict(params))
              results.append(ev.evaluate(strat, cfg, docs, ds, scorer=scorer))
      return {
          "dataset_summary": {"queries": len(ds.queries)},
          "results": [r.model_dump() for r in results],
      }

• src/chunktuner/mcp/tools.py:59-88 (registration)

  MCP tool registration of 'evaluate_chunking' as a FastMCP tool with parameters: path, use_case, content_type, strategies, max_docs, top_k, dry_run, embedding_model.

  @mcp.tool()
  def evaluate_chunking(
      path: str,
      use_case: str = "rag_qa",
      content_type: str | None = None,
      strategies: list[str] | None = None,
      max_docs: int = 20,
      top_k: int = 5,
      dry_run: bool = False,
      embedding_model: str | None = None,
  ) -> dict:
      """Dry-run cost estimate or full evaluation (``DummyEmbeddingFunction`` if no model)."""
      t0 = time.perf_counter()
      with _eval_sem:
          try:
              out = evaluate_chunking_impl(
                  path,
                  use_case,
                  content_type=content_type,
                  strategies=strategies,
                  max_docs=max_docs,
                  top_k=top_k,
                  dry_run=dry_run,
                  embedding_model=embedding_model,
              )
              _log_tool("evaluate_chunking", t0, True)
              return out
          except Exception:
              _log_tool("evaluate_chunking", t0, False)
              raise

• src/chunktuner/api/routes.py:27-36 (schema)

  Pydantic schema EvaluateBody defining input validation for evaluate_chunking (path, use_case, content_type, strategies, max_docs, top_k, dry_run, embedding_model).

  class EvaluateBody(BaseModel):
      path: str
      use_case: str = "rag_qa"
      content_type: str | None = None
      strategies: list[str] | None = None
      max_docs: int = 20
      top_k: int = 5
      dry_run: bool = False
      embedding_model: str | None = None

• src/chunktuner/api/routes.py:66-80 (registration)

  FastAPI route registration for POST /evaluate_chunking, delegates to evaluate_chunking_impl service.

  @router.post("/evaluate_chunking")
  def evaluate_chunking(body: EvaluateBody) -> dict:
      try:
          return evaluate_chunking_impl(
              body.path,
              body.use_case,
              content_type=body.content_type,
              strategies=body.strategies,
              max_docs=body.max_docs,
              top_k=body.top_k,
              dry_run=body.dry_run,
              embedding_model=body.embedding_model,
          )
      except ValueError as e:
          raise HTTPException(status_code=400, detail=str(e)) from e

• src/chunktuner/api/security.py:16-28 (helper)

  Path validation helper require_under_base used by evaluate_chunking_impl to ensure path is under CHUNK_TUNER_BASE_DIR.

  def require_under_base(path_str: str) -> Path:
      """Resolve ``path_str`` to an absolute path and ensure it stays under the workspace base."""
      base = _resolved_base_dir()
      candidate = Path(path_str).expanduser()
      if not candidate.is_absolute():
          candidate = (base / candidate).resolve()
      else:
          candidate = candidate.resolve()
      try:
          candidate.relative_to(base)
      except ValueError as e:
          raise ValueError(f"path must be under CHUNK_TUNER_BASE_DIR ({base}): {candidate}") from e
      return candidate