diff --git a/.github/workflows/themefinder-eval.yml b/.github/workflows/themefinder-eval.yml index 509fa34dc..8db1ec176 100644 --- a/.github/workflows/themefinder-eval.yml +++ b/.github/workflows/themefinder-eval.yml @@ -71,6 +71,8 @@ jobs: LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }} LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }} LANGFUSE_BASE_URL: ${{ secrets.LANGFUSE_BASE_URL }} + THEMEFINDER_EVAL_DATASET_SOURCE: langfuse + THEMEFINDER_EVAL_ARTEFACT_STORE: langfuse HF_TOKEN: ${{ secrets.HF_TOKEN }} ENVIRONMENT: production DATASET: ${{ github.event.inputs.dataset || 'gambling_XS' }} diff --git a/README.md b/README.md index 994e1118d..939eb3fa8 100644 --- a/README.md +++ b/README.md @@ -146,3 +146,10 @@ The workspace settings are configured to: - Enable TypeScript support in Svelte files You can override these settings in your User Settings if you prefer different personal configurations. See the [VSCode settings documentation](https://code.visualstudio.com/docs/getstarted/settings) for more information on the settings hierarchy. + +### Running Evals + +Component-level evals live in `themefinder/evals/pipelines/`, where there is one dir per component. Each dir holds a DVC pipeline that runs an eval for that specific component. To run a specific component's eval pipeline: +1. `cd` to the relevant dir +2. Set the desired parameters in the `params.yaml` file +3. Run `uv run --package themefinder --extra dev dvc repro` to run the pipeline in a version-aware fashion (i.e. only running the stages whose dependencies have changed since their last run). If you want to run the whole pipeline regardless of version changes, run `uv run --package themefinder --extra dev dvc repro --force` \ No newline at end of file diff --git a/themefinder/evals/pipelines/mapping/.dvc/.gitignore b/themefinder/evals/pipelines/mapping/.dvc/.gitignore new file mode 100644 index 000000000..d7d9b4800 --- /dev/null +++ b/themefinder/evals/pipelines/mapping/.dvc/.gitignore @@ -0,0 +1,2 @@ +/config.local +/cache diff --git a/themefinder/evals/pipelines/mapping/.dvc/config b/themefinder/evals/pipelines/mapping/.dvc/config new file mode 100644 index 000000000..e69de29bb diff --git a/themefinder/evals/pipelines/mapping/.dvcignore b/themefinder/evals/pipelines/mapping/.dvcignore new file mode 100644 index 000000000..519730552 --- /dev/null +++ b/themefinder/evals/pipelines/mapping/.dvcignore @@ -0,0 +1,3 @@ +# Add patterns of files dvc should ignore, which could improve +# the performance. Learn more at +# https://dvc.org/doc/user-guide/dvcignore diff --git a/themefinder/evals/pipelines/mapping/download_data.py b/themefinder/evals/pipelines/mapping/download_data.py new file mode 100644 index 000000000..0f3ff8a71 --- /dev/null +++ b/themefinder/evals/pipelines/mapping/download_data.py @@ -0,0 +1,253 @@ +"""Data acquisition for theme mapping evaluation. + +Fetches theme-mapping evaluation datasets either from Langfuse (preferred) +or from the local on-disk fallback (see `datasets.load_local_data`), and +normalises both sources into a common item shape that `evaluate.py` can +consume without needing to know where the data came from. +""" + +import json +import sys +from datetime import date +from pathlib import Path + +# Add the evals/ directory to the path so sibling modules (langfuse_utils, +# datasets) can be imported when this script is run directly from within +# pipelines/mapping/. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent)) + +import langfuse_utils +import pandas as pd +from datasets import DatasetConfig, load_local_data + + +def _normalise_langfuse_item(item) -> dict: + """Convert a Langfuse dataset item into the common item shape. + + Args: + item: Langfuse dataset item (has .input, .expected_output, + .metadata, .id) + + Returns: + Dict with question_part, responses_df, question, topics_df, + expected_output, and the original Langfuse item (kept so callers + can still create trace-linked runs via `langfuse_utils.dataset_item_trace`). + """ + responses_df = pd.DataFrame(item.input["responses"]) + question = item.input["question"] + topics_df = pd.DataFrame(item.input["topics"]) + topics_df = topics_df.rename(columns={"topic_id": "topic_id", "topic": "topic"}) + + return { + "question_part": item.metadata.get("question_part", item.id), + "responses_df": responses_df, + "question": question, + "topics_df": topics_df, + "expected_output": item.expected_output, + "langfuse_item": item, + } + + +def _normalise_local_item(item: dict) -> dict: + """Convert a local dataset item dict into the common item shape. + + Args: + item: Dict as returned by `datasets.load_local_data` + + Returns: + Dict with question_part, responses_df, question, topics_df, + expected_output, and `langfuse_item` set to None. + """ + responses_df = pd.DataFrame(item["input"]["responses"]) + question = item["input"]["question"] + topics_df = pd.DataFrame(item["input"]["topics"]) + + return { + "question_part": item.get("metadata", {}).get("question_part", "unknown"), + "responses_df": responses_df, + "question": question, + "topics_df": topics_df, + "expected_output": item["expected_output"], + "langfuse_item": None, + } + + +def fetch_langfuse_dataset_items( + ctx: langfuse_utils.LangfuseContext, config: DatasetConfig +) -> list[dict] | None: + """Fetch and normalise dataset items from Langfuse. + + Args: + ctx: LangfuseContext with an active client + config: DatasetConfig identifying the dataset to fetch + + Returns: + List of normalised items, or None if the dataset could not be + retrieved from Langfuse (callers should fall back to local data). + """ + try: + dataset = ctx.client.get_dataset(config.name) + except Exception as e: + print( + f"Dataset {config.name} not found in Langfuse, falling back to local: {e}" + ) + return None + + return [_normalise_langfuse_item(item) for item in dataset.items] + + +def load_local_dataset_items( + config: DatasetConfig, question_num: int | None = None +) -> list[dict]: + """Load and normalise dataset items from the local fallback data. + + Args: + config: DatasetConfig identifying the dataset to load + question_num: Optional specific question number (1-3) to filter to + + Returns: + List of normalised items + """ + data_items = load_local_data(config) + + if question_num is not None: + data_items = [ + item + for item in data_items + if f"part_{question_num}" + in item.get("metadata", {}).get("question_part", "") + ] + + return [_normalise_local_item(item) for item in data_items] + + +def get_dataset_items( + ctx: langfuse_utils.LangfuseContext, + config: DatasetConfig, + question_num: int | None = None, +) -> list[dict]: + """Get normalised mapping-eval dataset items, preferring Langfuse. + + Args: + ctx: LangfuseContext (may or may not be enabled) + config: DatasetConfig identifying the dataset to load + question_num: Optional specific question number (1-3). Only applied + to the local fallback - Langfuse datasets are used in full. + + Returns: + List of normalised items, each with keys: question_part, + responses_df, question, topics_df, expected_output, langfuse_item + (None for locally-sourced items). + """ + if ctx.is_enabled: + items = fetch_langfuse_dataset_items(ctx, config) + if items is not None: + return items + + return load_local_dataset_items(config, question_num) + + +def write_dataset_items_to_local(config: DatasetConfig, items: list[dict]) -> Path: + """Persist normalised dataset items to `evals/data//`. + + Writes each question part in the same layout expected by + `datasets.load_local_data` (`inputs//{question.json, + responses.jsonl}` and `outputs/mapping/// + {themes.json, mapping.jsonl}`), so the dataset - regardless of whether it + was originally fetched from Langfuse or local disk - is always available + as a local, version-controllable fallback under + `config.local_path` (a subdirectory of `evals/data/` named after the + dataset). + + Args: + config: DatasetConfig identifying the dataset (its `local_path` is + `evals/data/`) + items: Normalised dataset items, as returned by `get_dataset_items` + + Returns: + The directory the dataset was written to (`config.local_path`) + """ + output_dir = config.local_path + date_str = date.today().isoformat() + + for item in items: + question_part = item["question_part"] + + inputs_dir = output_dir / "inputs" / question_part + inputs_dir.mkdir(parents=True, exist_ok=True) + + with open(inputs_dir / "question.json", "w") as f: + json.dump({"question_text": item["question"]}, f, indent=4) + + with open(inputs_dir / "responses.jsonl", "w") as f: + for _, row in item["responses_df"][["response_id", "response"]].iterrows(): + f.write( + json.dumps( + { + "response_id": row["response_id"], + "response": row["response"], + } + ) + + "\n" + ) + + outputs_dir = output_dir / "outputs" / "mapping" / date_str / question_part + outputs_dir.mkdir(parents=True, exist_ok=True) + + with open(outputs_dir / "themes.json", "w") as f: + json.dump(item["topics_df"].to_dict(orient="records"), f, indent=4) + + mappings = item["expected_output"].get("mappings", {}) + with open(outputs_dir / "mapping.jsonl", "w") as f: + for response_id, labels in mappings.items(): + f.write( + json.dumps({"response_id": response_id, "labels": labels}) + "\n" + ) + + return output_dir + + +if __name__ == "__main__": + import argparse + + import dotenv + + dotenv.load_dotenv() + + parser = argparse.ArgumentParser( + description="Download/inspect theme mapping evaluation data" + ) + parser.add_argument( + "--dataset", + default="gambling_XS", + help="Dataset identifier (e.g., gambling_XS)", + ) + parser.add_argument( + "--question", type=int, default=None, help="Specific question number (1-3)" + ) + args = parser.parse_args() + + dataset_config = DatasetConfig(dataset=args.dataset, stage="mapping") + langfuse_ctx = langfuse_utils.get_langfuse_context( + session_id="download_data_check", + eval_type="mapping", + metadata={"dataset": args.dataset}, + tags=[args.dataset], + ) + + dataset_items = get_dataset_items(langfuse_ctx, dataset_config, args.question) + source = ( + "Langfuse" if dataset_items and dataset_items[0]["langfuse_item"] else "local" + ) + print( + f"Loaded {len(dataset_items)} item(s) for dataset '{args.dataset}' from {source} source" + ) + for dataset_item in dataset_items: + print( + f" {dataset_item['question_part']}: " + f"{len(dataset_item['responses_df'])} responses, " + f"{len(dataset_item['topics_df'])} topics" + ) + + written_dir = write_dataset_items_to_local(dataset_config, dataset_items) + print(f"Wrote dataset to {written_dir}") diff --git a/themefinder/evals/pipelines/mapping/dvc.yaml b/themefinder/evals/pipelines/mapping/dvc.yaml new file mode 100644 index 000000000..faae2daa9 --- /dev/null +++ b/themefinder/evals/pipelines/mapping/dvc.yaml @@ -0,0 +1,30 @@ +stages: + download_data: + cmd: uv run download_data.py --dataset ${dataset} + deps: + - download_data.py + - ../../datasets.py + - ../../langfuse_utils.py + params: + - dataset + outs: + - ../../data/${dataset} + evaluate: + cmd: uv run evaluate.py --dataset ${dataset} --model ${model} --output scores.json + deps: + - evaluate.py + - download_data.py + - ../../datasets.py + - ../../langfuse_utils.py + - ../../utils_gateway.py + - ../../evaluators.py + - ../../metrics.py + - ../../prompts.py + - ../../../src/themefinder + - ../../data/${dataset} + params: + - dataset + - model + outs: + - scores.json: + cache: false diff --git a/themefinder/evals/pipelines/mapping/evaluate.py b/themefinder/evals/pipelines/mapping/evaluate.py new file mode 100644 index 000000000..7e826e402 --- /dev/null +++ b/themefinder/evals/pipelines/mapping/evaluate.py @@ -0,0 +1,260 @@ +"""Theme mapping response generation and evaluation. + +Runs the theme mapping task (assigning themes to responses) against +evaluation data - fetched via `download_data.py` - and scores the results, +with Langfuse dataset/experiment support when configured. +""" + +import argparse +import asyncio +import json +import os +import sys +from datetime import datetime +from pathlib import Path + +import dotenv + +# Add the evals/ directory to the path so sibling modules (langfuse_utils, +# utils_gateway, datasets, evaluators, metrics) can be imported when this +# script is run directly from within pipelines/mapping/. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent)) + +import langfuse_utils +import utils_gateway +from datasets import DatasetConfig +from download_data import get_dataset_items +from evaluators import mapping_f1_evaluator +from metrics import calculate_mapping_metrics +from themefinder.llm import OpenAILLM + +from themefinder import theme_mapping + + +async def evaluate_mapping( + dataset: str = "gambling_XS", + question_num: int | None = None, + model: str | None = None, + llm: OpenAILLM | None = None, + langfuse_ctx: langfuse_utils.LangfuseContext | None = None, +) -> dict: + """Run mapping evaluation. + + Args: + dataset: Dataset identifier (e.g., "gambling_S", "healthcare_M") + question_num: Optional specific question number (1-3) to evaluate + model: Optional gateway model name to use for theme mapping (falls + back to AUTO_EVAL_4_1_SWEDEN_DEPLOYMENT env var if not set). + Ignored if `llm` is provided. + llm: Optional pre-configured LLM instance (for benchmark runs) + langfuse_ctx: Optional pre-configured Langfuse context (for benchmark runs) + + Returns: + Dict containing evaluation scores + """ + dotenv.load_dotenv() + + config = DatasetConfig(dataset=dataset, stage="mapping") + + # Use provided context or create new one + owns_context = langfuse_ctx is None + if langfuse_ctx is None: + session_id = f"{config.name.replace('/', '_')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + langfuse_ctx = langfuse_utils.get_langfuse_context( + session_id=session_id, + eval_type="mapping", + metadata={"dataset": dataset}, + tags=[dataset], + ) + + # Use provided LLM or create new one + if llm is None: + base_url, api_key = utils_gateway.gateway_credentials() + llm = OpenAILLM( + model=model or os.getenv("AUTO_EVAL_4_1_SWEDEN_DEPLOYMENT"), + request_kwargs={"temperature": 0}, + base_url=base_url, + api_key=api_key, + ) + + # Fetch data - download_data.py handles the Langfuse vs local branching + # and normalises both sources into a common item shape. + dataset_items = get_dataset_items(langfuse_ctx, config, question_num) + + if langfuse_ctx.is_enabled and dataset_items and dataset_items[0]["langfuse_item"]: + result = await _run_with_langfuse(langfuse_ctx, dataset_items, llm) + else: + result = await _run_local_fallback(dataset_items, llm) + + # Only flush if we created the context + if owns_context: + langfuse_utils.flush(langfuse_ctx) + return result + + +async def _run_with_langfuse(ctx, dataset_items: list[dict], llm) -> dict: + """Run evaluation with manual dataset iteration for proper trace control. + + Args: + ctx: LangfuseContext + dataset_items: Normalised dataset items (see `download_data.py`), + each carrying its original Langfuse item under "langfuse_item" + llm: LangChain LLM instance + + Returns: + Dict containing evaluation scores + """ + all_scores = {} + + for item in dataset_items: + langfuse_item = item["langfuse_item"] + + # Create trace for this item with full metadata + with langfuse_utils.dataset_item_trace(ctx, langfuse_item, ctx.session_id) as ( + trace, + trace_id, + ): + # Run theme mapping + result_df, unprocessable_df = await theme_mapping( + responses_df=item["responses_df"][["response_id", "response"]], + llm=llm, + question=item["question"], + refined_themes_df=item["topics_df"], + ) + if not unprocessable_df.empty: + print( + f" Warning: {len(unprocessable_df)} responses could not be processed" + ) + + # Build labels map + labels = dict( + zip( + result_df["response_id"].astype(str), + result_df["labels"].tolist(), + ) + ) + output = {"labels": labels} + + # Update trace with output + if trace: + trace.update(output=output) + + # Run evaluator and attach score + eval_result = mapping_f1_evaluator( + output=output, + expected_output=item["expected_output"], + ) + + if trace_id and ctx.client: + ctx.client.create_score( + trace_id=trace_id, + name=eval_result.name, + value=eval_result.value, + data_type="NUMERIC", + ) + + # Collect for return + item_key = item["question_part"] + all_scores[f"{item_key}_f1"] = eval_result.value + + # Include pipeline output for disk persistence + all_scores[f"{item_key}_output"] = output + + print(f"Mapping Eval Results: {ctx.session_id}") + return all_scores + + +async def _run_local_fallback(dataset_items: list[dict], llm) -> dict: + """Run evaluation without Langfuse (local development). + + Args: + dataset_items: Normalised dataset items (see `download_data.py`) + llm: LangChain LLM instance + + Returns: + Dict containing evaluation scores + """ + all_scores = {} + + for item in dataset_items: + question_part = item["question_part"] + responses_df = item["responses_df"] + expected_mappings = item["expected_output"]["mappings"] + + result, unprocessable_df = await theme_mapping( + responses_df=responses_df[["response_id", "response"]], + llm=llm, + question=item["question"], + refined_themes_df=item["topics_df"][["topic_id", "topic"]], + ) + if not unprocessable_df.empty: + print( + f" Warning: {len(unprocessable_df)} responses could not be processed" + ) + + # Merge for comparison + responses_df["topics"] = ( + responses_df["response_id"].astype(str).map(expected_mappings) + ) + responses_df = responses_df.merge( + result[["response_id", "labels"]], "inner", on="response_id" + ) + + mapping_metrics = calculate_mapping_metrics( + df=responses_df, column_one="topics", column_two="labels" + ) + print(f"Theme Mapping ({question_part}): \n {mapping_metrics}") + + # Collect scores with question prefix + for key, value in mapping_metrics.items(): + if isinstance(value, (int, float)): + all_scores[f"{question_part}_{key}"] = value + + return all_scores + + +def write_results(results: dict, output_path: Path) -> None: + """Persist evaluation results to a JSON file for DVC to track as a metric. + + Args: + results: Dict of evaluation scores (as returned by `evaluate_mapping`) + output_path: Path to write the JSON results to + """ + output_path.parent.mkdir(parents=True, exist_ok=True) + with open(output_path, "w") as f: + json.dump(results, f, indent=2, default=str) + + +if __name__ == "__main__": + import nest_asyncio + + nest_asyncio.apply() + + parser = argparse.ArgumentParser(description="Run theme mapping evaluation") + parser.add_argument( + "--dataset", + default="gambling_XS", + help="Dataset identifier (e.g., gambling_XS)", + ) + parser.add_argument( + "--question", type=int, default=None, help="Specific question number (1-3)" + ) + parser.add_argument( + "--model", + default=None, + help="Gateway model name to use for theme mapping " + "(defaults to AUTO_EVAL_4_1_SWEDEN_DEPLOYMENT env var)", + ) + parser.add_argument( + "--output", + default="scores.json", + help="Path to write evaluation results (JSON)", + ) + args = parser.parse_args() + + scores = asyncio.run( + evaluate_mapping( + dataset=args.dataset, question_num=args.question, model=args.model + ) + ) + write_results(scores, Path(args.output)) diff --git a/themefinder/evals/pipelines/mapping/params.yaml b/themefinder/evals/pipelines/mapping/params.yaml new file mode 100644 index 000000000..9285c9f74 --- /dev/null +++ b/themefinder/evals/pipelines/mapping/params.yaml @@ -0,0 +1,2 @@ +dataset: gambling_XS +model: gpt-4.1-sweden-2025-03 diff --git a/uv.lock b/uv.lock index cecfa4c79..d76252ce1 100644 --- a/uv.lock +++ b/uv.lock @@ -3930,4 +3930,4 @@ dependencies = [ sdist = { url = "https://files.pythonhosted.org/packages/10/9a/2fef89272d98b799e4daa50201c5582ec76bdd4e92a1a7e3deb74c52b7fa/zc_lockfile-4.0.tar.gz", hash = "sha256:d3ab0f53974296a806db3219b9191ba0e6d5cbbd1daa2e0d17208cb9b29d2102", size = 10956, upload-time = "2025-09-18T07:32:34.412Z" } wheels = [ { url = "https://files.pythonhosted.org/packages/3b/7f/3a614b65bc4b181578b1d50a78663ee02d5d2d3b859712f3d3597c8afe6f/zc_lockfile-4.0-py3-none-any.whl", hash = "sha256:aa3aa295257bebaa09ea9ad5cb288bf9f98f88de6932f96b6659f62715d83581", size = 9143, upload-time = "2025-09-18T07:32:33.517Z" }, -] +] \ No newline at end of file