recommend_config
Run a tuner to evaluate and rank chunking strategies for your RAG corpus, returning recommendations based on your use case and content.
Instructions
Run tuner and return ranked Recommendation (uses dummy embeddings by default).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| path | Yes | ||
| use_case | No | rag_qa | |
| content_type | No | ||
| strategies | No | ||
| max_docs | No | ||
| top_k | No | ||
| embedding_model | No |
Implementation Reference
- src/chunktuner/mcp/service.py:129-165 (handler)Core implementation of recommend_config: validates path/strategies, creates embedding function, ingests docs, builds AutoTuner, calls tuner.recommend(), and returns ranked Recommendation.
def recommend_config_impl( path: str, use_case: str, *, content_type: str | None = None, strategies: list[str] | None = None, max_docs: int = 20, top_k: int = 5, 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) 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] uc = cast(UseCase, use_case) tuner = AutoTuner( build_full_registry(), Evaluator(embed, top_k=top_k), ScoreCalculator(uc), ) strat_names = strategies or ["fixed_tokens", "recursive_character"] rec = tuner.recommend( docs, uc, strategies=strat_names, max_docs=max_docs, baseline=True, content_type=content_type, ) return rec.model_dump() - src/chunktuner/mcp/tools.py:90-117 (handler)MCP tool registration as a FastMCP @tool decorator: wraps recommend_config_impl with semaphore protection and logging.
@mcp.tool() def recommend_config( 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, embedding_model: str | None = None, ) -> dict: """Run tuner and return ranked ``Recommendation`` (uses dummy embeddings by default).""" t0 = time.perf_counter() with _eval_sem: try: out = recommend_config_impl( path, use_case, content_type=content_type, strategies=strategies, max_docs=max_docs, top_k=top_k, embedding_model=embedding_model, ) _log_tool("recommend_config", t0, True) return out except Exception: _log_tool("recommend_config", t0, False) raise - src/chunktuner/mcp/tools.py:25-117 (registration)register_tools() function registers all MCP tools on the FastMCP instance via decorator pattern.
def register_tools(mcp: object) -> None: from mcp.server.fastmcp import FastMCP if not isinstance(mcp, FastMCP): raise TypeError("Expected FastMCP instance") @mcp.tool() def list_strategies(content_type: str | None = None) -> list[dict]: """List registered chunking strategies, optionally filtered by content type.""" t0 = time.perf_counter() try: out = list_strategies_impl(content_type) _log_tool("list_strategies", t0, True) return out except Exception: _log_tool("list_strategies", t0, False) raise @mcp.tool() def preview_chunks( text: str, strategy_name: str, config: dict | None = None, ) -> list[dict]: """Chunk inline text with one strategy + params (no embeddings).""" t0 = time.perf_counter() try: out = preview_chunks_impl(text, strategy_name, config or {}) _log_tool("preview_chunks", t0, True) return out except Exception: _log_tool("preview_chunks", t0, False) raise @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 @mcp.tool() def recommend_config( 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, embedding_model: str | None = None, ) -> dict: """Run tuner and return ranked ``Recommendation`` (uses dummy embeddings by default).""" t0 = time.perf_counter() with _eval_sem: try: out = recommend_config_impl( path, use_case, content_type=content_type, strategies=strategies, max_docs=max_docs, top_k=top_k, embedding_model=embedding_model, ) _log_tool("recommend_config", t0, True) return out except Exception: _log_tool("recommend_config", t0, False) raise - src/chunktuner/api/routes.py:38-46 (schema)Pydantic request body model for the /recommend_config HTTP endpoint (path, use_case, content_type, strategies, max_docs, top_k, embedding_model).
class RecommendConfigBody(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 embedding_model: str | None = None - src/chunktuner/api/routes.py:83-96 (registration)FastAPI HTTP POST endpoint at /recommend_config, delegates to recommend_config_impl.
@router.post("/recommend_config") def recommend_config(body: RecommendConfigBody) -> dict: try: return recommend_config_impl( body.path, body.use_case, content_type=body.content_type, strategies=body.strategies, max_docs=body.max_docs, top_k=body.top_k, embedding_model=body.embedding_model, ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from e