diff --git a/requirements.txt b/requirements.txt index e1361623a..c6ab1894f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -50,6 +50,7 @@ rich rouge scikit-image scikit-learn +selfies sentence_transformers sentencepiece setuptools diff --git a/tests/test_scimif.py b/tests/test_scimif.py new file mode 100644 index 000000000..fc43a22a8 --- /dev/null +++ b/tests/test_scimif.py @@ -0,0 +1,67 @@ +import json +import unittest + +from PIL import Image + +from vlmeval.dataset.scimif import SciMIF +from vlmeval.dataset.utils.scimif_eval import evaluate_record, summarize_results + + +class TestSciMIF(unittest.TestCase): + + def test_convert_text_only_record(self): + converted = SciMIF._convert_record( + { + 'sample_id': 'physics_0', + 'subject': 'physics', + 'edit_question': 'Return a velocity in m/s.', + 'instruction_list': [], + }, 0) + + self.assertIsNone(converted['image']) + self.assertEqual(json.loads(converted['image_path']), []) + self.assertEqual(converted['question'], 'Return a velocity in m/s.') + + def test_convert_image_record(self): + converted = SciMIF._convert_record( + { + 'sample_id': 'geography_0', + 'subject': 'geography', + 'edit_question': 'Describe the image.', + 'image': [Image.new('RGB', (2, 2), color='white')], + 'image_path': ['images/geography/example.png'], + 'instruction_list': [], + }, 0) + + self.assertGreater(len(json.loads(converted['image'])[0]), 64) + self.assertEqual(json.loads(converted['image_path']), ['geography/example.png']) + + def test_instruction_evaluation_and_summary(self): + item = { + 'index': + '0', + 'subject': + 'physics', + 'edit_question': + 'Give the final velocity in m/s.', + 'prediction': + 'The final answer is 3 m/s.', + 'instruction_list': [ + { + 'instruction_name': 'physics_unit_consistency', + 'source': 'core_task', + 'required_parameters': 'm/s', + }, + ], + } + result = evaluate_record(item) + summary = summarize_results([{**item, **result}]) + + self.assertEqual(result['instruction_score'], 1.0) + self.assertEqual(result['strict_score'], 1.0) + self.assertEqual(summary[0]['instruction_accuracy'], 1.0) + self.assertEqual(result['instruction_results'][0]['source'], 'core_task') + + +if __name__ == '__main__': + unittest.main() diff --git a/vlmeval/dataset/__init__.py b/vlmeval/dataset/__init__.py index 2b71263d4..c475b8e78 100644 --- a/vlmeval/dataset/__init__.py +++ b/vlmeval/dataset/__init__.py @@ -123,6 +123,7 @@ from .robospatialbench import RoboSpatialBench from .sarena import SArena from .scidocbench import SciDocBench +from .scimif import SciMIF from .sfebench import SFE from .SGI_Bench_1_0.deep_research import SGI_Bench_Deep_Research from .SGI_Bench_1_0.dry_experiment import SGI_Bench_Dry_Experiment @@ -322,7 +323,7 @@ def evaluate(self, eval_file, **judge_kwargs): SciDocBench, OmniMat, MMRarebenchDiagnosis, MMRarebenchTreatment, MMRarebenchCrossmodal, MMRarebenchExamination, MRareBenchDiagnosis, MRareBenchEvidenceVerif, MolRecBenchWildDataset, BabyVision, WildprobeDataset, - PerceptionBench, SUPERChemDataset, C4Bench, + PerceptionBench, SUPERChemDataset, C4Bench, SciMIF, ] # add by EASI team diff --git a/vlmeval/dataset/scimif.py b/vlmeval/dataset/scimif.py new file mode 100644 index 000000000..38af664c7 --- /dev/null +++ b/vlmeval/dataset/scimif.py @@ -0,0 +1,233 @@ +import io +import json +import os.path as osp +import re +from typing import Any + +import pandas as pd +from PIL import Image + +from vlmeval.smp import dump, encode_image_to_base64, get_intermediate_file_path, get_logger, load +from vlmeval.utils import track_progress_rich +from .image_base import ImageBaseDataset +from .utils import DEBUG_MESSAGE, build_judge +from .utils.scimif_eval import evaluate_record, summarize_results + +logger = get_logger(__name__) + + +class _JudgeClient: + + def __init__(self, judge): + self.judge = judge + + def __call__(self, prompt: str) -> str: + result = self.judge.generate(prompt) + fail_message = getattr(self.judge, 'fail_msg', '') + if not result or (fail_message and fail_message in result): + raise RuntimeError('The judge model failed to return a response.') + return str(result) + + +def _evaluate_scimif_row(item, llm_client, judge_model): + return evaluate_record(item, llm_client=llm_client, judge_model=judge_model) + + +class SciMIF(ImageBaseDataset): + """SciMIF benchmark loaded from its Hugging Face dataset repository.""" + + TYPE = 'VQA' + MODALITY = 'IMAGE' + DEFAULT_JUDGE_MODEL = 'gpt-4.1' + + HF_REPO_ID = 'Sheryle7436/SciMIF' + HF_CONFIG = 'default' + HF_SPLIT = 'test' + + @classmethod + def supported_datasets(cls): + return ['SciMIF'] + + def __init__(self, dataset='SciMIF', skip_noimg=False): + # SciMIF contains both multimodal and text-only samples. Text-only + # samples must remain in the benchmark. + super().__init__(dataset=dataset, skip_noimg=skip_noimg) + + @staticmethod + def _to_pil_image(value: Any) -> Image.Image: + """Convert a decoded or non-decoded Hugging Face image to PIL.""" + + if isinstance(value, Image.Image): + return value + + if isinstance(value, dict): + image_bytes = value.get('bytes') + image_path = value.get('path') + + if image_bytes is not None: + if isinstance(image_bytes, memoryview): + image_bytes = image_bytes.tobytes() + with Image.open(io.BytesIO(image_bytes)) as image: + return image.copy() + + if image_path: + with Image.open(image_path) as image: + return image.copy() + + if isinstance(value, str): + with Image.open(value) as image: + return image.copy() + + raise TypeError(f'Unsupported SciMIF image value: {type(value)!r}') + + @staticmethod + def _as_list(value: Any) -> list: + if value is None: + return [] + if isinstance(value, (list, tuple)): + return [item for item in value if item is not None] + return [value] + + @classmethod + def _convert_record(cls, record: dict, index: int) -> dict: + sample_id = str(record.get('sample_id') or f'SciMIF_{index}') + + images = cls._as_list(record.get('image')) + encoded_images = [encode_image_to_base64(cls._to_pil_image(image)) for image in images] + + image_paths = [ + str(path).removeprefix('images/') for path in cls._as_list(record.get('image_path')) + if str(path).strip() and str(path).strip() != '[]' + ] + if len(image_paths) != len(encoded_images): + image_paths = [f'{sample_id}_{image_index}.jpg' for image_index in range(len(encoded_images))] + + answer = record.get('answer') + has_answer = answer is not None and str(answer).strip() != '' + if answer is None: + answer = '' + elif not isinstance(answer, str): + answer = json.dumps(answer, ensure_ascii=False) + + return { + 'index': index, + 'id': record.get('id'), + 'sample_id': sample_id, + 'split': cls.HF_SPLIT, + 'category': record.get('subject', ''), + 'subject': record.get('subject', ''), + 'task': record.get('task', ''), + 'question': record.get('edit_question', ''), + 'edit_question': record.get('edit_question', ''), + 'original_question': record.get('original_question', ''), + 'answer': answer, + 'has_answer': has_answer, + 'choose_instruction': json.dumps(record.get('choose_instruction') or [], ensure_ascii=False), + 'instruction_list': json.dumps(record.get('instruction_list') or [], ensure_ascii=False), + # ImageBaseDataset parses JSON lists and writes decoded images to + # $LMUData/images/SciMIF when build_prompt() is called. + # Keep text-only samples truly empty. ImageBaseDataset interprets + # short non-empty strings (such as "[]") as references to another + # sample's image, which is not the meaning here. + 'image': json.dumps(encoded_images) if encoded_images else None, + 'image_path': json.dumps(image_paths, ensure_ascii=False), + } + + def load_data(self, dataset): + if dataset != 'SciMIF': + raise ValueError(f'Unsupported dataset name: {dataset!r}') + + try: + from datasets import load_dataset + except ImportError as exc: + raise ImportError('Loading SciMIF requires the `datasets` package. ' + 'Install it with `pip install datasets`.') from exc + + hf_dataset = load_dataset( + self.HF_REPO_ID, + self.HF_CONFIG, + split=self.HF_SPLIT, + ) + + rows = [self._convert_record(record, index) for index, record in enumerate(hf_dataset)] + return pd.DataFrame(rows) + + def build_prompt(self, line): + if isinstance(line, int): + line = self.data.iloc[line] + + image_value = line.get('image') + has_image = (bool(image_value) + if isinstance(image_value, str) else isinstance(image_value, list) and len(image_value) > 0) + image_paths = self.dump_image(line) if has_image else [] + messages = [dict(type='image', value=image_path) for image_path in image_paths] + messages.append(dict(type='text', value=line['question'])) + return messages + + @classmethod + def evaluate(cls, eval_file, **judge_kwargs): + data = load(eval_file) + if not isinstance(data, pd.DataFrame): + data = pd.DataFrame(data) + if 'prediction' not in data: + raise ValueError('SciMIF evaluation requires a `prediction` column.') + + judge_options = dict(judge_kwargs) + nproc = judge_options.pop('nproc', 4) + judge_name = judge_options.pop('model', cls.DEFAULT_JUDGE_MODEL) + judge_options.pop('use_verifier', None) + judge_options.pop('use_vllm', None) + safe_judge_name = re.sub(r'[^A-Za-z0-9_.-]+', '_', str(judge_name)) + + detail_file = get_intermediate_file_path(eval_file, f'_{safe_judge_name}_details', 'xlsx') + score_file = get_intermediate_file_path(eval_file, f'_{safe_judge_name}_score', 'csv') + tmp_file = get_intermediate_file_path(eval_file, f'_{safe_judge_name}_tmp', 'pkl') + + records = data.to_dict(orient='records') + keys = [str(record.get('index', position)) for position, record in enumerate(records)] + cached = load(tmp_file) if osp.exists(tmp_file) else {} + if not isinstance(cached, dict): + cached = {} + + pending_records = [] + pending_keys = [] + for key, record in zip(keys, records): + if key not in cached: + pending_keys.append(key) + pending_records.append(record) + + if pending_records: + judge_options.setdefault('temperature', 0) + judge_options.setdefault('timeout', 300) + judge_options.setdefault('max_tokens', 1024) + judge = build_judge(model=judge_name, **judge_options) + assert judge.working(), ('SciMIF instruction evaluation requires a working judge API.\n' + DEBUG_MESSAGE) + llm_client = _JudgeClient(judge) + tasks = [dict(item=record, llm_client=llm_client, judge_model=judge_name) for record in pending_records] + new_results = track_progress_rich( + _evaluate_scimif_row, + tasks, + nproc=nproc, + chunksize=nproc, + keys=pending_keys, + save=tmp_file, + ) + cached.update(dict(zip(pending_keys, new_results))) + else: + logger.info(f'Reused all {len(cached)} cached SciMIF evaluation results.') + + evaluated_records = [] + for key, record in zip(keys, records): + evaluation = cached[key] + evaluated_records.append({**record, **evaluation}) + + details = pd.DataFrame(evaluated_records) + details['instruction_results'] = details['instruction_results'].map( + lambda value: json.dumps(value, ensure_ascii=False)) + dump(details, detail_file) + + summary = pd.DataFrame(summarize_results(evaluated_records)) + dump(summary, score_file) + logger.info(f'SciMIF detailed results saved to {detail_file}.') + logger.info(f'SciMIF scores saved to {score_file}.') + return summary diff --git a/vlmeval/dataset/utils/scimif/__init__.py b/vlmeval/dataset/utils/scimif/__init__.py new file mode 100644 index 000000000..a24edf8b4 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/__init__.py @@ -0,0 +1,21 @@ +from .analysis_method_checking import evaluate_method_constraint +from .analysis_step_checking import evaluate_analysis_steps +from .chemistry_count_atom_checking import evaluate_atom_count +from .chemistry_count_bond_checking import evaluate_bond_count +from .chemistry_count_group_checking import evaluate_group_count +from .chemistry_format_validation import evaluate_molecular_format +from .geography_format_geocoding_validation import evaluate_geography_address +from .life_format_entity_relationship_validation import evaluate_entity_relationship +from .life_sequence_length_checking import evaluate_sequence_length +from .materials_format_characterization_technique_validation import \ + evaluate_characterization_technique +from .materials_property_prediction_checking import evaluate_property_prediction +from .options_matching import evaluate_options_constraint +from .unit_matching import evaluate_unit_consistency + +__all__ = [ + 'evaluate_unit_consistency', 'evaluate_method_constraint', 'evaluate_options_constraint', 'evaluate_analysis_steps', + 'evaluate_atom_count', 'evaluate_bond_count', 'evaluate_group_count', 'evaluate_molecular_format', + 'evaluate_geography_address', 'evaluate_entity_relationship', 'evaluate_sequence_length', + 'evaluate_characterization_technique', 'evaluate_property_prediction' +] diff --git a/vlmeval/dataset/utils/scimif/analysis_method_checking.py b/vlmeval/dataset/utils/scimif/analysis_method_checking.py new file mode 100644 index 000000000..90f4b44c6 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/analysis_method_checking.py @@ -0,0 +1,52 @@ +import re +from typing import Any, Dict + +from .extraction_utils import extract_method + + +def evaluate_method_constraint(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + llm_client=None, + **kwargs) -> Dict[str, Any]: + if not required_parameters or not response: + return {'passed': False, 'score': 0.0, 'detail': 'Missing required_parameters response is empty'} + method_text = required_parameters.strip() + for phrase in [ + 'specific chemical fomulas or laws', 'specific physical fomulas or laws', + 'specific geographical fomulas or laws', 'specific biological fomulas or laws', + 'specific materials science fomulas or laws' + ]: + method_text = re.sub(re.escape(phrase) + '\\s*[:：]?\\s*', '', method_text, flags=re.I) + method_text = method_text.strip() + if not method_text: + return {'passed': False, 'score': 0.0, 'detail': 'Unable to parse required_parameters formula/law'} + extracted, extract_src = extract_method(response, method_text, llm_client=llm_client, item=item) + if not extracted: + return {'passed': False, 'score': 0.0, 'detail': f"formula/law: '{method_text}'"} + patterns = _build_method_patterns(method_text) + matched = any((re.search(p, extracted, re.IGNORECASE | re.DOTALL) for p in patterns)) + if not matched: + matched = any((re.search(p, response, re.IGNORECASE | re.DOTALL) for p in patterns)) + if matched: + return {'passed': True, 'score': 1.0, 'detail': f'method; extracted({extract_src}): {extracted}'} + return {'passed': False, 'score': 0.0, 'detail': f'method required ; extracted({extract_src}): {extracted}'} + + +def _build_method_patterns(method_text: str) -> list: + patterns = [] + patterns.append(re.escape(method_text)) + if '=' in method_text: + parts = method_text.split('=', 1) + if len(parts) == 2: + patterns.append(re.escape(parts[0].strip()) + '\\s*=\\s*' + re.escape(parts[1].strip())) + formula_aliases = { + 'F=ma': ['F\\s*=\\s*ma', 'Newton'], + 'PV=nRT': ['PV\\s*=\\s*nRT', 'ideal gas'], + 'E=mc²': ['E\\s*=\\s*mc', 'mass-energy'] + } + for k, aliases in formula_aliases.items(): + if k in method_text or any((a.lower() in method_text.lower() for a in aliases)): + patterns.extend(aliases) + return patterns diff --git a/vlmeval/dataset/utils/scimif/analysis_step_checking.py b/vlmeval/dataset/utils/scimif/analysis_step_checking.py new file mode 100644 index 000000000..8c567e817 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/analysis_step_checking.py @@ -0,0 +1,120 @@ +import json +import re +from typing import Any, Dict + + +def evaluate_analysis_steps(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + instruction_description: str = '', + llm_client=None, + **kwargs) -> Dict[str, Any]: + if not response: + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + if llm_client is not None: + return _llm_judge(response, item, instruction_name, required_parameters, instruction_description, llm_client) + return _heuristic_judge(response, item, instruction_name, required_parameters, instruction_description) + + +def _llm_judge(response: str, item: Dict[str, Any], instruction_name: str, required_parameters: str, + instruction_description: str, llm_client) -> Dict[str, Any]: + try: + prompt = _build_judge_prompt(response, item, instruction_name, required_parameters, instruction_description) + if hasattr(llm_client, 'chat'): + result = llm_client.chat(prompt) + elif hasattr(llm_client, 'complete'): + result = llm_client.complete(prompt) + else: + result = llm_client(prompt) + if not isinstance(result, str): + result = getattr(result, 'content', None) or getattr(result, 'text', None) or str(result) + return _parse_llm_judge_result(result or '') + except Exception as e: + return {'passed': False, 'score': 0.0, 'detail': f'LLM-judge error: {e}'} + + +def _parse_llm_judge_result(result: str) -> Dict[str, Any]: + json_match = re.search('```(?:json)?\\s*(\\{[\\s\\S]*?\\})\\s*```', result) + if json_match: + try: + data = json.loads(json_match.group(1)) + return _result_from_parsed_json(data, result) + except json.JSONDecodeError: + pass + json_match = re.search('\\{[^{}]*\\"score\\"[^{}]*\\}', result) + if json_match: + try: + data = json.loads(json_match.group(0)) + return _result_from_parsed_json(data, result) + except json.JSONDecodeError: + pass + frac_match = re.search('score\\s*[=:]\\s*(\\d+)\\s*/\\s*(\\d+)', result, re.I) + if frac_match: + num, den = (int(frac_match.group(1)), int(frac_match.group(2))) + score = num / den if den > 0 else 0.0 + return {'passed': score >= 1.0, 'score': score, 'detail': f'LLM judge: {num}/{den} steps matched'} + dec_match = re.search('score\\s*[=:]\\s*([\\d.]+)', result, re.I) + if dec_match: + score = float(dec_match.group(1)) + return {'passed': score >= 1.0, 'score': min(1.0, max(0.0, score)), 'detail': f'LLM-judge: score={score}'} + passed = 'yes' in result.lower() or 'true' in result.lower() + return { + 'passed': passed, + 'score': 1.0 if passed else 0.0, + 'detail': f"LLM judge: {'passed' if passed else 'failed'}" + } + + +def _result_from_parsed_json(data: dict, raw_result: str) -> Dict[str, Any]: + score = data.get('score') + if score is not None: + if isinstance(score, (list, tuple)) and len(score) >= 2: + num, den = (int(score[0]), int(score[1])) + s = num / den if den > 0 else 0.0 + elif isinstance(score, str) and '/' in score: + parts = score.split('/') + if len(parts) == 2: + num, den = (int(parts[0].strip()), int(parts[1].strip())) + s = num / den if den > 0 else 0.0 + else: + s = float(score) + else: + s = float(score) + s = min(1.0, max(0.0, s)) + else: + matched = data.get('matched', data.get('matched_count', 0)) + total = data.get('total', data.get('total_steps', 1)) + s = matched / total if total > 0 else 0.0 + detail_parts = [] + if 'required_steps' in data: + detail_parts.append(f"requiredsteps: {len(data['required_steps'])}") + if 'matched' in data: + detail_parts.append(f"match: {data['matched']}") + if 'total' in data: + detail_parts.append(f"total: {data['total']}") + detail = 'LLM-judge: ' + ', '.join(detail_parts) if detail_parts else f'LLM-judge: score={s:.2f}' + return {'passed': s >= 1.0, 'score': s, 'detail': detail} + + +def _heuristic_judge(response: str, item: Dict[str, Any], instruction_name: str, required_parameters: str, + instruction_description: str) -> Dict[str, Any]: + step_indicators = [ + 'step\\s*\\d+', 'steps?\\s*\\d+', 'first', 'second', 'then', 'next', 'finally', '①|②|③|④|⑤', '1\\.|2\\.|3\\.', + '→|⇒|->' + ] + has_steps = any((re.search(p, response, re.I) for p in step_indicators)) + sentences = [s.strip() for s in response.replace('\n', '.').split('.') if len(s.strip()) > 10] + has_multiple = len(sentences) >= 2 + passed = has_steps or has_multiple + return { + 'passed': passed, + 'score': 1.0 if passed else 0.0, + 'detail': 'Detected a multi-step response' if passed else 'No multi-step response detected' + } + + +def _build_judge_prompt(response: str, item: Dict[str, Any], instruction_name: str, required_parameters: str, + instruction_description: str) -> str: + question = item.get('edit_question', item.get('original_question', '')) + return f"""You are an expert evaluator for scientific questions. Evaluate whether the model's response satisfies the instruction for "reasoning steps" or "reasoning process".\n\n## Task\n1. **First output your reasoning** (brief): Extract the required step framework from the question and required_parameters; extract the actual steps from the model's response.\n2. **Then output a structured conclusion**: Compare each required step against the response, determine if it is mentioned, and compute the score.\n\n## Input\n\n**Instruction type**: {instruction_name}\n**Instruction description**: {instruction_description}\n**Required parameters/steps**: {required_parameters}\n\n**Question (edit_question)**:\n{question[:1500]}\n\n**Model response**:\n{response[:3000]}\n\n## Output Requirements\n1. First write your **reasoning**: List the steps you extracted from required_parameters and edit_question (required_steps), and the steps you extracted from the response (response_steps).\n2. Then write your **conclusion**: Output a JSON block at the end in the following format (do not omit):\n\n```json\n{{\n "required_steps": ["step 1 description", "step 2 description", ...],\n "response_steps": ["step 1 from response", "step 2 from response", ...],\n "matched": ,\n "total": ,\n "score": \n}}\n```\n\n**Scoring rule**: score = number of correctly mentioned steps / total required steps, full score is 1.0. A step counts as mentioned if the response contains corresponding content.""" # noqa: E501 diff --git a/vlmeval/dataset/utils/scimif/chemistry_count_atom_checking.py b/vlmeval/dataset/utils/scimif/chemistry_count_atom_checking.py new file mode 100644 index 000000000..258a0adba --- /dev/null +++ b/vlmeval/dataset/utils/scimif/chemistry_count_atom_checking.py @@ -0,0 +1,187 @@ +import re +from typing import Any, Dict, Optional, Tuple + +try: + import rdkit + HAS_RDKIT = rdkit is not None +except ImportError: + HAS_RDKIT = False +from .extraction_utils import extract_molecule +from .rdkit_utils import mol_from_smiles_lenient + +_SELFIES_ELEMENT_TOKENS = frozenset( + {'H', 'B', 'C', 'N', 'O', 'F', 'P', 'S', 'Cl', 'Br', 'I', 'Si', 'As', 'Se', 'Sb', 'Te', 'Po'}) +_SMILES_ELEMENT_RE = re.compile( + 'Ac|Ag|Al|Am|Ar|As|At|Au|Ba|Be|Bh|Bi|Bk|Br|Ca|Cd|Ce|Cf|Cl|Cm|Cn|Co|Cr|Cs|Cu|Db|Ds|Dy|Er|Es|Eu|Fe|Fl|Fm|Fr|Ga|Gd|Ge|He|Hf|Hg|Ho|Hs|In|Ir|Kr|La|Li|Lr|Lu|Lv|Mc|Md|Mg|Mn|Mo|Mt|Na|Nb|Nd|Ne|Nh|Ni|No|Np|Os|Pa|Pb|Pd|Pm|Po|Pr|Pt|Pu|Ra|Rb|Re|Rf|Rg|Rh|Rn|Ru|Sb|Sc|Se|Sg|Si|Sm|Sn|Sr|Ta|Tb|Tc|Te|Th|Ti|Tl|Tm|Ts|Xe|Yb|Zn|Zr|B|C|N|O|F|P|S|I|K|V|Y|W|H|U|b|c|n|o|p|s' # noqa: E501 +) + + +def _count_atoms_from_selfies_brackets(mol_str: str) -> Optional[Dict[str, int]]: + counts: Dict[str, int] = {} + for inner in re.findall('\\[([^\\]]+)\\]', mol_str): + inner = inner.strip() + if inner.startswith(('Branch', 'Ring', 'Expl', 'Pad')): + continue + if inner.startswith('=') and len(inner) >= 2: + sym = inner[1:] + if sym in _SELFIES_ELEMENT_TOKENS: + counts[sym] = counts.get(sym, 0) + 1 + continue + if inner in _SELFIES_ELEMENT_TOKENS: + counts[inner] = counts.get(inner, 0) + 1 + return counts if counts else None + + +def evaluate_atom_count(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + llm_client=None, + **kwargs) -> Dict[str, Any]: + if not required_parameters or not response: + return {'passed': False, 'score': 0.0, 'detail': 'Missing required_parameters response is empty'} + edit_question = item.get('edit_question', '') or '' if item else '' + requirement_text = f"{required_parameters or ''}\n{edit_question}" + target_counts = _parse_atom_requirements(requirement_text) + target_total = _parse_total_atom_requirement(requirement_text) + if not target_counts and target_total is None: + return { + 'passed': False, + 'score': 0.0, + 'detail': f'Evaluator unable to parseatom :{required_parameters}', + 'skipped': True + } + mol_str, fmt, extract_src = extract_molecule(response, llm_client=llm_client, item=item) + if not mol_str: + return {'passed': False, 'score': 0.0, 'detail': 'molecule (SMILES/SELFIES)'} + if HAS_RDKIT: + actual, parse_note = _count_atoms_rdkit(mol_str, fmt) + else: + actual, parse_note = (_count_atoms_fallback(mol_str, fmt), 'not installed RDKit, atom token') + if actual is None: + reason = parse_note or 'RDKit cannot molecule' + return {'passed': False, 'score': 0.0, 'detail': f'moleculeParsing failed; {reason}; extracted: {mol_str}'} + mismatches = [] + for elem, count in target_counts.items(): + got = actual.get(elem, 0) + if got != count: + mismatches.append(f'{elem}: required {count}, actual {got}') + if target_total is not None: + got_total = sum(actual.values()) + if got_total != target_total: + mismatches.append(f'atom : required{target_total}, actual {got_total}') + passed = len(mismatches) == 0 + base_detail = 'atom count required' if passed else '; '.join(mismatches) + extra = f'; {parse_note}' if parse_note else '' + detail = f'{base_detail}; extracted({extract_src}): {mol_str}{extra}' + return {'passed': passed, 'score': 1.0 if passed else 0.0, 'detail': detail} + + +def _parse_atom_requirements(text: str) -> Dict[str, int]: + elem_map = { + 'hydrogen': 'H', + 'boron': 'B', + 'carbon': 'C', + 'nitrogen': 'N', + 'oxygen': 'O', + 'fluorine': 'F', + 'silicon': 'Si', + 'phosphorus': 'P', + 'sulfur': 'S', + 'chlorine': 'Cl', + 'arsenic': 'As', + 'selenium': 'Se', + 'bromine': 'Br', + 'antimony': 'Sb', + 'tellurium': 'Te', + 'iodine': 'I', + 'polonium': 'Po' + } + valid_symbols = set(elem_map.values()) + result: Dict[str, int] = {} + pattern = '(\\d+)\\s+([A-Za-z]{1,12})\\s+atoms?\\b' + for m in re.finditer(pattern, text or '', re.I): + num, raw_name = (int(m.group(1)), m.group(2)) + name = raw_name.lower() + elem = elem_map.get(name) + if elem is None: + symbol = raw_name[0].upper() + raw_name[1:].lower() + elem = symbol if symbol in valid_symbols else None + if elem: + result[elem] = num + return result + + +def _parse_total_atom_requirement(text: str) -> Optional[int]: + for m in re.finditer('(?:exactly\\s+)?(\\d+)\\s+atoms?\\b', text or '', re.I): + prefix = (text or '')[max(0, m.start() - 20):m.start()] + if re.search('[A-Za-z]+\\s*$', prefix) and (not re.search('(?:exactly|total|contains?)\\s*$', prefix, re.I)): + continue + return int(m.group(1)) + return None + + +def _decode_selfies_for_rdkit(mol_str: str) -> Tuple[Optional[str], Optional[Dict[str, int]], str]: + try: + import selfies + smiles = selfies.decoder(mol_str) + if smiles is None: + return (None, None, 'SELFIES decoder None') + return (smiles, None, '') + except ImportError: + naive = _count_atoms_from_selfies_brackets(mol_str) + if naive: + return (None, naive, 'not installed selfies, token ( : pip install selfies)') + return (None, None, 'not installed selfies , cannot SELFIES. : pip install selfies') + except Exception as e: + return (None, None, f'SELFIES decoder error: {e}') + + +def _count_atoms_rdkit(mol_str: str, fmt: str) -> Tuple[Optional[Dict[str, int]], str]: + smiles = mol_str + if (fmt or '').lower() == 'selfies': + sm, naive, note = _decode_selfies_for_rdkit(mol_str) + if naive is not None: + return (naive, note) + if sm is None: + if mol_from_smiles_lenient(mol_str) is None: + return (None, note) + smiles = mol_str + else: + smiles = sm + try: + mol = mol_from_smiles_lenient(smiles) + if mol is None: + sm_preview = smiles if len(smiles) <= 220 else smiles[:220] + '...' + return (None, f'RDKit MolFromSmiles failed( sanitize=False); decoded SMILES:{sm_preview!r}') + counts: Dict[str, int] = {} + for atom in mol.GetAtoms(): + sym = atom.GetSymbol() + counts[sym] = counts.get(sym, 0) + 1 + return (counts, '') + except Exception as e: + return (None, f'error:{e}') + + +def _count_atoms_fallback(mol_str: str, fmt: str = '') -> Optional[Dict[str, int]]: + if (fmt or '').lower() == 'selfies': + return _count_atoms_from_selfies_brackets(mol_str) + counts: Dict[str, int] = {} + bracket_spans = [] + for match in re.finditer('\\[([^\\]]+)\\]', mol_str): + bracket_spans.append(match.span()) + inner = re.sub('^\\d+', '', match.group(1)) + atom = re.match('([A-Z][a-z]?|[bcnops])', inner) + if atom: + symbol = atom.group(1) + symbol = symbol.capitalize() if symbol in 'bcnops' else symbol + counts[symbol] = counts.get(symbol, 0) + 1 + chars = list(mol_str) + for start, end in bracket_spans: + chars[start:end] = ' ' * (end - start) + unbracketed = ''.join(chars) + for match in _SMILES_ELEMENT_RE.finditer(unbracketed): + symbol = match.group(0) + symbol = symbol.capitalize() if symbol in 'bcnops' else symbol + counts[symbol] = counts.get(symbol, 0) + 1 + return counts if counts else None diff --git a/vlmeval/dataset/utils/scimif/chemistry_count_bond_checking.py b/vlmeval/dataset/utils/scimif/chemistry_count_bond_checking.py new file mode 100644 index 000000000..d82cb6598 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/chemistry_count_bond_checking.py @@ -0,0 +1,106 @@ +import re +from typing import Any, Dict + +try: + from rdkit import Chem + from rdkit.Chem import Lipinski + HAS_RDKIT = True +except ImportError: + HAS_RDKIT = False +from .extraction_utils import extract_molecule +from .rdkit_utils import mol_from_smiles_lenient + + +def evaluate_bond_count(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + llm_client=None, + **kwargs) -> Dict[str, Any]: + if not required_parameters or not response: + return {'passed': False, 'score': 0.0, 'detail': 'Missing required_parameters response is empty'} + edit_question = item.get('edit_question', '') or '' if item else '' + target = _parse_bond_requirements(f"{required_parameters or ''}\n{edit_question}") + if not target: + return { + 'passed': False, + 'score': 0.0, + 'detail': f'Evaluator unable to parsechemical bond :{required_parameters}', + 'skipped': True + } + if not HAS_RDKIT: + return { + 'passed': False, + 'score': 0.0, + 'detail': 'RDKit not installed, cannot reliably molecule chemical bond count', + 'skipped': True + } + mol_str, fmt, extract_src = extract_molecule(response, llm_client=llm_client, item=item) + if not mol_str: + return {'passed': False, 'score': 0.0, 'detail': 'molecule'} + actual = _count_bonds_rdkit(mol_str, fmt) + if actual is None: + return {'passed': False, 'score': 0.0, 'detail': f'moleculeParsing failed; extracted: {mol_str}'} + mismatches = [] + for bond_type, count in target.items(): + got = actual.get(bond_type, 0) + if got != count: + mismatches.append(f'{bond_type}: required {count}, actual {got}') + passed = len(mismatches) == 0 + base_detail = 'chemical bond count required' if passed else '; '.join(mismatches) + detail = f'{base_detail}; extracted({extract_src}): {mol_str}' + return {'passed': passed, 'score': 1.0 if passed else 0.0, 'detail': detail} + + +def _parse_bond_requirements(text: str) -> Dict[str, int]: + result: Dict[str, int] = {} + pattern = '(?:exactly\\s+)?(\\d+)\\s+(rotatable|single|double|triple|aromatic)\\s+bonds?\\b' + for m in re.finditer(pattern, text or '', re.I): + result[m.group(2).lower()] = int(m.group(1)) + total_patterns = [ + '(?:exactly\\s+)?(\\d+)\\s+(?:total\\s+)?chemical\\s+bonds?\\b', + '(?:exactly\\s+)?(\\d+)\\s+total\\s+(?:explicit\\s+heavy-atom\\s+)?bonds?\\b', + '(?:exactly\\s+)?(\\d+)\\s+explicit\\s+heavy-atom\\s+bonds?\\b' + ] + for total_pattern in total_patterns: + m = re.search(total_pattern, text or '', re.I) + if m: + result['total'] = int(m.group(1)) + break + return result + + +def _count_bonds_rdkit(mol_str: str, fmt: str) -> Dict[str, int]: + try: + if (fmt or '').lower() == 'selfies': + try: + import selfies + decoded = selfies.decoder(mol_str) + if decoded: + mol_str = decoded + except Exception: + pass + mol = mol_from_smiles_lenient(mol_str) + if mol is None: + return None + single = double = triple = aromatic = 0 + for bond in mol.GetBonds(): + bt = bond.GetBondType() + if bt == Chem.BondType.SINGLE: + single += 1 + elif bt == Chem.BondType.DOUBLE: + double += 1 + elif bt == Chem.BondType.TRIPLE: + triple += 1 + elif bt == Chem.BondType.AROMATIC: + aromatic += 1 + return { + 'total': int(mol.GetNumBonds()), + 'single': single, + 'double': double, + 'triple': triple, + 'aromatic': aromatic, + 'rotatable': int(Lipinski.NumRotatableBonds(mol)) + } + except Exception: + return None diff --git a/vlmeval/dataset/utils/scimif/chemistry_count_group_checking.py b/vlmeval/dataset/utils/scimif/chemistry_count_group_checking.py new file mode 100644 index 000000000..fbb9bacf8 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/chemistry_count_group_checking.py @@ -0,0 +1,126 @@ +import re +from typing import Any, Dict, Iterable + +try: + from rdkit import Chem + HAS_RDKIT = True +except ImportError: + HAS_RDKIT = False +from .extraction_utils import extract_molecule +from .rdkit_utils import mol_from_smiles_lenient + +FUNCTIONAL_GROUP_SMARTS = { + 'hydroxyl': '[OX2H;!$(O[C,S,P]=O)]', + 'carboxyl': '[CX3](=O)[OX2H1]', + 'carbonyl': '[CX3]=[OX1]', + 'amine': '[NX3;!$(N[CX3](=O))]', + 'amide': '[CX3](=O)[NX3]', + 'ester': '[CX3](=O)[OX2][#6;!$(C=O)]', + 'ether': '[OD2;!$([O][C,S,P]=O)]([#6])[#6]', + 'aldehyde': '[CX3H1](=O)[#6,#1]', + 'ketone': '[#6][CX3](=O)[#6]', + 'benzene': 'c1ccccc1', + 'sulfone': '[SX4](=[OX1])(=[OX1])', + 'sulfoxide': '[SX3](=[OX1])([#6])[#6]', + 'sulfide': '[SX2]([#6])[#6]', + 'disulfide': '[SX2]-[SX2]', + 'nitro': '[NX3+](=[OX1])[O-]', + 'nitrile': '[CX2]#[NX1]', + 'halo': '[#6]-[F,Cl,Br,I]', + 'anhydride': '[CX3](=O)[OX2][CX3](=O)', + 'borane': '[BX3;H1,H2,H3]', + 'thiol': '[SX2H]' +} +GROUP_PATTERNS = [('carboxyl', 'carboxyl(?:ic\\s+acid)?'), ('hydroxyl', 'hydroxyl|hydroxy|alcohol'), + ('amine', 'amine|amino'), ('amide', 'amide'), ('aldehyde', 'aldehyde'), ('ketone', 'ketone'), + ('benzene', 'benzene(?:\\s+ring)?|phenyl'), ('ester', 'ester'), ('ether', 'ether'), + ('sulfone', 'sulfone'), ('sulfoxide', 'sulfoxide'), ('disulfide', 'disulfide'), + ('sulfide', 'sulfide|thioether'), ('nitro', 'nitro'), ('nitrile', 'nitrile|cyano'), + ('halo', 'halo|halide'), ('anhydride', 'anhydride'), ('borane', 'borane'), ('carbonyl', 'carbonyl'), + ('thiol', 'thiol|sulfhydryl')] + + +def evaluate_group_count(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + llm_client=None, + **kwargs) -> Dict[str, Any]: + if not required_parameters or not response: + return {'passed': False, 'score': 0.0, 'detail': 'Missing required_parameters response is empty'} + edit_question = item.get('edit_question', '') or '' if item else '' + target = _parse_group_requirements(f"{required_parameters or ''}\n{edit_question}") + if not target: + return { + 'passed': False, + 'score': 0.0, + 'detail': f'Evaluator unable to parsefunctional group :{required_parameters}', + 'skipped': True + } + if not HAS_RDKIT: + return { + 'passed': False, + 'score': 0.0, + 'detail': 'RDKit not installed, cannot reliably functional group', + 'skipped': True + } + mol_str, fmt, extract_src = extract_molecule(response, llm_client=llm_client, item=item) + if not mol_str: + return {'passed': False, 'score': 0.0, 'detail': 'molecule'} + actual = _count_groups_rdkit(mol_str, fmt, target.keys()) + if actual is None: + return {'passed': False, 'score': 0.0, 'detail': f'moleculeParsing failed; extracted: {mol_str}'} + mismatches = [] + for group, requirement in target.items(): + count = requirement['count'] + mode = requirement['mode'] + got = actual.get(group, 0) + ok = mode == 'exactly' and got == count or (mode == 'at_least' and got >= count) or (mode == 'at_most' + and got <= count) + if not ok: + mode_zh = {'exactly': 'exactly', 'at_least': 'at least', 'at_most': 'at most'}[mode] + mismatches.append(f'{group}: required{mode_zh} {count}, actual {got}') + passed = len(mismatches) == 0 + base_detail = 'functional group count required' if passed else '; '.join(mismatches) + detail = f'{base_detail}; extracted({extract_src}): {mol_str}' + return {'passed': passed, 'score': 1.0 if passed else 0.0, 'detail': detail} + + +def _parse_group_requirements(text: str) -> Dict[str, Dict[str, Any]]: + result: Dict[str, Dict[str, Any]] = {} + blob = text or '' + for canonical, group_pattern in GROUP_PATTERNS: + pattern = f'(?:(exactly|at\\s+least|at\\s+most)\\s+)?(\\d+)\\s+(?:{group_pattern})s?(?:\\s+(?:functional\\s+)?groups?|\\s+rings?)?\\b' # noqa: E501 + for m in re.finditer(pattern, blob, re.I): + qualifier = (m.group(1) or '').lower().replace(' ', '_') + if qualifier not in {'at_least', 'at_most'}: + qualifier = 'exactly' + result[canonical] = {'count': int(m.group(2)), 'mode': qualifier} + break + return result + + +def _count_groups_rdkit(mol_str: str, fmt: str, groups: Iterable[str]) -> Dict[str, int]: + try: + smiles = mol_str + if (fmt or '').lower() == 'selfies': + try: + import selfies + decoded = selfies.decoder(mol_str) + if decoded: + smiles = decoded + except Exception: + smiles = mol_str + mol = mol_from_smiles_lenient(smiles) + if mol is None: + return None + result = {} + for g in groups: + smarts = FUNCTIONAL_GROUP_SMARTS.get(g) + if smarts: + pat = Chem.MolFromSmarts(smarts) + if pat: + result[g] = len(mol.GetSubstructMatches(pat, uniquify=True)) + return result + except Exception: + return None diff --git a/vlmeval/dataset/utils/scimif/chemistry_format_validation.py b/vlmeval/dataset/utils/scimif/chemistry_format_validation.py new file mode 100644 index 000000000..82fb28ea1 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/chemistry_format_validation.py @@ -0,0 +1,158 @@ +import re +from typing import Any, Dict + +try: + import rdkit + HAS_RDKIT = rdkit is not None +except ImportError: + HAS_RDKIT = False +try: + import selfies + HAS_SELFIES = True +except ImportError: + HAS_SELFIES = False +from .extraction_utils import extract_molecule +from .rdkit_utils import mol_from_smiles_lenient + + +def evaluate_molecular_format(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + llm_client=None, + **kwargs) -> Dict[str, Any]: + if not response: + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + fmt = _detect_required_format(required_parameters, item) + if not fmt: + fmt = 'SMILES' + if fmt.upper() == 'SELFIES': + return _validate_selfies(response, llm_client=llm_client, item=item) + if fmt.upper() == 'SMILES': + return _validate_smiles(response, llm_client=llm_client, item=item) + if fmt.upper() == 'IUPAC': + return _validate_iupac(response, llm_client=llm_client, item=item) + return {'passed': False, 'score': 0.0, 'detail': f':{fmt}'} + + +def _detect_required_format(required_parameters: str, item: Dict) -> str: + text = (required_parameters or '') + ' ' + (item.get('edit_question', '') or '') + text_lower = text.lower() + if 'selfies' in text_lower: + return 'SELFIES' + if 'smiles' in text_lower: + return 'SMILES' + if 'iupac' in text_lower: + return 'IUPAC' + return '' + + +def _validate_selfies(response: str, llm_client=None, item=None) -> Dict[str, Any]: + mol_str, _, extract_src = extract_molecule(response, llm_client=llm_client, item=item) + if not mol_str: + return {'passed': False, 'score': 0.0, 'detail': 'Not found SELFIES string'} + if not _is_valid_selfies_candidate(mol_str): + return {'passed': False, 'score': 0.0, 'detail': f'valid SELFIES ; extracted({extract_src}): {mol_str}'} + if HAS_SELFIES: + try: + smiles = selfies.decoder(mol_str) + if smiles and HAS_RDKIT: + mol = mol_from_smiles_lenient(smiles) + if mol is not None: + return { + 'passed': True, + 'score': 1.0, + 'detail': f'SELFIES The format is valid; extracted({extract_src}): {mol_str}' + } + except Exception: + return { + 'passed': False, + 'score': 0.0, + 'detail': f'SELFIES Parsing failed; extracted({extract_src}): {mol_str}' + } + if re.match('^(\\[[^\\]]+\\])+$', mol_str): + return {'passed': True, 'score': 1.0, 'detail': f'SELFIES ; extracted({extract_src}): {mol_str}'} + return { + 'passed': False, + 'score': 0.0, + 'detail': f'SELFIES The format is invalid; extracted({extract_src}): {mol_str}' + } + + +def _is_valid_selfies_candidate(s: str) -> bool: + if not s or not s.strip().startswith('['): + return False + low = s.strip().lower() + if low in {'molecule', 'smiles', 'selfies', 'chemical', 'answer'}: + return False + return bool(re.match('^(\\[[^\\]]+\\])+$', s.strip())) + + +def _validate_smiles(response: str, llm_client=None, item=None) -> Dict[str, Any]: + mol_str, _, extract_src = extract_molecule(response, llm_client=llm_client, item=item) + if not mol_str: + return {'passed': False, 'score': 0.0, 'detail': 'Not found SMILES string'} + if HAS_RDKIT: + mol = mol_from_smiles_lenient(mol_str) + if mol is not None: + return { + 'passed': True, + 'score': 1.0, + 'detail': f'SMILES The format is valid; extracted({extract_src}): {mol_str}' + } + return { + 'passed': False, + 'score': 0.0, + 'detail': f'RDKit Unable to parse SMILES; extracted({extract_src}): {mol_str}' + } + if re.fullmatch('[A-Za-z0-9@+\\-\\[\\]\\(\\)=#$\\\\/%.:*]+', mol_str) and len(mol_str) >= 3: + return {'passed': True, 'score': 1.0, 'detail': f'SMILES ; extracted({extract_src}): {mol_str}'} + return { + 'passed': False, + 'score': 0.0, + 'detail': f'SMILES The format is invalid; extracted({extract_src}): {mol_str}' + } + + +def _is_valid_smiles_candidate(s: str) -> bool: + if not s or len(s) < 3: + return False + low = s.strip().lower() + blocklist = {'molecule', 'smiles', 'selfies', 'chemical', 'answer', 'structure', 'compound', 'format'} + if low in blocklist: + return False + if not re.fullmatch('[A-Za-z0-9@+\\-\\[\\]\\(\\)=#$\\\\/%.:*]+', s): + return False + if HAS_RDKIT: + return mol_from_smiles_lenient(s) is not None + return bool(re.search('(?:Cl|Br|Si|Se|Te|[BCNOFPSIbcnosp])', s)) + + +def _validate_iupac(response: str, llm_client=None, item=None) -> Dict[str, Any]: + extracted, extract_src = _extract_iupac(response, llm_client=llm_client, item=item) + if not extracted: + return {'passed': False, 'score': 0.0, 'detail': 'Not found IUPAC'} + if re.search('\\b(ethane|methane|propane|butane|pentane|hexane|benzene|ethanol)\\b', extracted, re.I): + return {'passed': True, 'score': 1.0, 'detail': f'Detected IUPAC ; extracted({extract_src}): {extracted}'} + if re.search('[a-z]+\\-\\d\\-[a-z]+|[a-z]+\\d+[a-z]*', extracted): + return {'passed': True, 'score': 1.0, 'detail': f'IUPAC ; extracted({extract_src}): {extracted}'} + return { + 'passed': False, + 'score': 0.0, + 'detail': f'Did not detect a valid IUPAC ; extracted({extract_src}): {extracted}' + } + + +def _extract_iupac(response: str, llm_client=None, item=None) -> tuple: + m = re.search('\\b([a-z]+\\-\\d\\-[a-z]+|[a-z]+\\d+[a-z]*|[A-Za-z]+(?:ane|ene|ol|one|al|oic)\\b)', response, re.I) + if m: + return (m.group(1), 'regex') + if llm_client: + from .extraction_utils import _extract_via_llm + extracted = _extract_via_llm( + response, + llm_client, + prompt_addendum='Extract the IUPAC chemical name from the response. Return only the name, nothing else.') + if extracted: + return (extracted, 'llm') + return ('', '') diff --git a/vlmeval/dataset/utils/scimif/extraction_utils.py b/vlmeval/dataset/utils/scimif/extraction_utils.py new file mode 100644 index 000000000..a2af9e116 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/extraction_utils.py @@ -0,0 +1,229 @@ +import json +import re +from typing import Any, Dict, Iterable, List, Optional, Tuple + +_MOLECULE_BLOCKLIST = frozenset({ + 'molecule', 'smiles', 'selfies', 'iupac', 'chemical', 'answer', 'formula', 'structure', 'compound', + 'representation', 'format', 'output', 'result', 'here', 'is', 'the', 'below', 'above', 'following', 'see' +}) +_SELFIES_FULL_RE = re.compile('(?:\\[[A-Za-z0-9_@+\\-=#$\\\\/%.:*]+\\])+') +_ELEMENT_SYMBOLS = 'Ac|Ag|Al|Am|Ar|As|At|Au|Ba|Be|Bh|Bi|Bk|Br|Ca|Cd|Ce|Cf|Cl|Cm|Cn|Co|Cr|Cs|Cu|Db|Ds|Dy|Er|Es|Eu|Fe|Fl|Fm|Fr|Ga|Gd|Ge|He|Hf|Hg|Ho|Hs|In|Ir|Kr|La|Li|Lr|Lu|Lv|Mc|Md|Mg|Mn|Mo|Mt|Na|Nb|Nd|Ne|Nh|Ni|No|Np|Os|Pa|Pb|Pd|Pm|Po|Pr|Pt|Pu|Ra|Rb|Re|Rf|Rg|Rh|Rn|Ru|Sb|Sc|Se|Sg|Si|Sm|Sn|Sr|Ta|Tb|Tc|Te|Th|Ti|Tl|Tm|Ts|Xe|Yb|Zn|Zr|B|C|N|O|F|P|S|I|K|V|Y|W|H|U|b|c|n|o|p|s' # noqa: E501 +_SMILES_TOKEN_RE = re.compile( # noqa: E501 + f'\\[[^\\[\\]\\s]+\\]|{_ELEMENT_SYMBOLS}|\\d+|[@+\\-\\(\\)=#$\\\\/%.:*]') +_FINAL_ANSWER_MARKER_RE = re.compile( + 'final[\\s_-]+(?:numerical[\\s_-]+)?answer\\s*[:：]?|the\\s+answer\\s+is\\s*(?:[:：]|\\.(?=\\s|$))?|(? bool: + if not _SELFIES_FULL_RE.fullmatch(s): + return False + return bool(re.search('\\[(?:[=#]?)(?:Cl|Br|Si|Se|Te|[BCNOFPSIbcnosp])', s)) + + +def _looks_like_smiles_without_rdkit(s: str) -> bool: + tokens = _SMILES_TOKEN_RE.findall(s) + if not tokens or ''.join(tokens) != s: + return False + return any((t.startswith('[') or re.fullmatch(_ELEMENT_SYMBOLS, t) for t in tokens)) + + +def _is_valid_molecule_candidate(s: str) -> bool: + if not s: + return False + s = s.strip().strip('"\'`') + low = s.lower() + if low in _MOLECULE_BLOCKLIST: + return False + if _looks_like_selfies(s): + return True + if not re.fullmatch('[A-Za-z0-9@+\\-\\[\\]\\(\\)=#$\\\\/%.:*]+', s): + return False + if not re.search('(?:Cl|Br|Si|Se|Te|Na|Li|Mg|Ca|Al|[BCNOFPSIbcnosp])', s): + return False + try: + from .rdkit_utils import mol_from_smiles_lenient + return mol_from_smiles_lenient(s) is not None + except Exception: + return _looks_like_smiles_without_rdkit(s) + + +def _walk_json_strings(obj: Any) -> Iterable[str]: + if isinstance(obj, dict): + for key in ('answer', 'smiles', 'selfies', 'molecule', 'result'): + if key in obj: + yield from _walk_json_strings(obj[key]) + for key, value in obj.items(): + if str(key).lower() not in {'answer', 'smiles', 'selfies', 'molecule', 'result'}: + yield from _walk_json_strings(value) + elif isinstance(obj, list): + for value in obj: + yield from _walk_json_strings(value) + elif isinstance(obj, str): + yield obj + + +def _json_string_candidates(text: str) -> List[str]: + out: List[str] = [] + snippets = [(text or '').strip()] + snippets.extend((m.group(1).strip() for m in re.finditer('```(?:json)?\\s*([\\s\\S]*?)```', text or '', re.I))) + for snippet in snippets: + try: + obj = json.loads(snippet) + except Exception: + continue + out.extend(_walk_json_strings(obj)) + return out + + +def _segment_molecule_candidates(text: str) -> List[str]: + candidates: List[str] = [] + candidates.extend(_json_string_candidates(text)) + stripped = text.strip().strip('"\'`').rstrip('.,;:') + if stripped and (not re.search('\\s', stripped)): + candidates.append(stripped) + fence_matches = list(re.finditer('```(?:smiles|selfies|text)?\\s*([\\s\\S]*?)```', text, re.I)) + for m in reversed(fence_matches): + block = m.group(1).strip() + if block: + candidates.append(block) + label_re = re.compile( + '(?:canonical\\s+)?(?:SMILES|SELFIES)\\s*[:：=]\\s*(?:`([^`]+)`|\\"([^\\"]+)\\"|\'([^\']+)\'|([^\\s,;]+))', re.I) + for m in reversed(list(label_re.finditer(text))): + value = next((g for g in m.groups() if g), '').strip() + if value: + candidates.append(value) + selfies_matches = _SELFIES_FULL_RE.findall(text) + candidates.extend(sorted(selfies_matches, key=len, reverse=True)) + token_re = re.compile('[A-Za-z0-9@+\\-\\[\\]\\(\\)=#$\\\\/%.:*]{3,}') + candidates.extend(sorted(token_re.findall(text), key=len, reverse=True)) + return candidates + + +def _molecule_candidates(response: str) -> List[str]: + text = response or '' + segments: List[str] = [] + markers = list(_FINAL_ANSWER_MARKER_RE.finditer(text)) + if markers: + tail = text[markers[-1].end():].strip() + if tail: + segments.append(tail) + segments.append(text) + candidates: List[str] = [] + for segment in segments: + candidates.extend(_segment_molecule_candidates(segment)) + out: List[str] = [] + seen = set() + for raw in candidates: + value = raw.strip().strip('"\'`').rstrip('.,;:') + if value and value not in seen: + seen.add(value) + out.append(value) + return out + + +def extract_molecule(response: str, llm_client=None, item: Optional[Dict] = None) -> Tuple[str, str, str]: + for candidate in _molecule_candidates(response): + if not _is_valid_molecule_candidate(candidate): + continue + fmt = 'selfies' if _looks_like_selfies(candidate) else 'smiles' + return (candidate, fmt, 'structured/regex') + if llm_client: + prompt_addendum = ('Extract the chemical molecule representation (SMILES or SELFIES format) from the response. ' + 'Return only the molecule string, nothing else.') + extracted = _extract_via_llm( + response, + llm_client, + prompt_addendum=prompt_addendum, + ) + if extracted and _is_valid_molecule_candidate(extracted): + fmt = 'selfies' if extracted.strip().startswith('[') else 'smiles' + return (extracted.strip(), fmt, 'llm') + return ('', '', '') + + +def extract_method(response: str, + required_parameters: str, + llm_client=None, + item: Optional[Dict] = None) -> Tuple[str, str]: + if llm_client: + extracted = _extract_via_llm( + response, + llm_client, + prompt_addendum=( + f'The required formula or law is: {required_parameters}. ' + 'Extract only the formula, law, or scientific method that the model response actually uses. ' + 'Do not infer or supply a method that is not explicitly present in the response. ' + 'If no method is present, return NOT_FOUND. Otherwise, return only the extracted ' + 'formula/method string, with no explanation.'), + ) + if extracted and extracted.strip().upper() != 'NOT_FOUND': + return (extracted.strip(), 'llm') + method_str = _extract_method_regex(response, required_parameters) + if method_str: + return (method_str, 'regex') + return ('', '') + + +def _extract_molecule_regex(response: str) -> Tuple[str, str]: + selfies_matches = re.findall('(?:\\[[^\\]]+\\])+', response) + if selfies_matches: + full = max(selfies_matches, key=len) + if full and '[' in full and (']' in full) and _is_valid_molecule_candidate(full): + return (full, 'selfies') + for m in re.finditer('[C][^\\s\\[\\]]{5,}', response): + cand = m.group(0) + if re.match('^[CBNOSPFI\\[\\]ClBr@=\\-\\(\\)\\\\\\/\\.#+\\d]+$', cand): + return (cand, 'smiles') + if selfies_matches: + full = max(selfies_matches, key=len) + if full and '[' in full and (']' in full): + return (full, 'selfies') + return ('', '') + + +def _extract_method_regex(response: str, required_parameters: str) -> str: + text = required_parameters.strip() + for phrase in [ + 'specific chemical fomulas or laws', 'specific physical fomulas or laws', + 'specific geographical fomulas or laws', 'specific biological fomulas or laws', + 'specific materials science fomulas or laws' + ]: + text = re.sub(re.escape(phrase) + '\\s*[:：]?\\s*', '', text, flags=re.I) + text = text.strip() + if not text: + return '' + if re.search(re.escape(text), response, re.I): + return text + if '=' in text: + parts = text.split('=', 1) + if len(parts) == 2: + pat = re.escape(parts[0].strip()) + '\\s*=\\s*' + re.escape(parts[1].strip()) + m = re.search(pat, response, re.I) + if m: + return m.group(0) + return '' + + +def _extract_via_llm(response: str, llm_client, prompt_addendum: str) -> str: + prompt = ('Extract the requested content from the following model response.\n\n' + f'{prompt_addendum}\n\nModel response:\n{response[:2500]}\n\n' + 'Extracted content (only the extracted string, no explanation):') + try: + if hasattr(llm_client, 'chat'): + result = llm_client.chat(prompt) + elif hasattr(llm_client, 'complete'): + result = llm_client.complete(prompt) + else: + result = llm_client(prompt) + if not isinstance(result, str): + result = getattr(result, 'content', None) or getattr(result, 'text', None) or str(result) + result = result or '' + lines = result.strip().split('\n') + for line in lines: + line = line.strip().strip('"\'`') + if line and len(line) > 2: + return line + return result.strip().strip('"\'`') if result else '' + except Exception: + return '' diff --git a/vlmeval/dataset/utils/scimif/general_checking.py b/vlmeval/dataset/utils/scimif/general_checking.py new file mode 100644 index 000000000..5eeb36929 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/general_checking.py @@ -0,0 +1,983 @@ +import ast +import csv +import io +import json +import re +import xml.etree.ElementTree as ET +from html.parser import HTMLParser +from typing import Any, Dict, List, Optional, Tuple + +EN_NUM = { + 'zero': 0, + 'one': 1, + 'two': 2, + 'three': 3, + 'four': 4, + 'five': 5, + 'six': 6, + 'seven': 7, + 'eight': 8, + 'nine': 9, + 'ten': 10 +} +_NUMBER_TOKEN_RE = re.compile('-?\\d+(?:,\\d{3})*(?:\\.\\d+)?(?:[eE][+-]?\\d+)?') +_SCI_NUMBER_TOKEN_RE = re.compile('-?\\d+(?:,\\d{3})*(?:\\.\\d+)?[eE][+-]?\\d+') +_SUPERSCRIPT_TO_ASCII = str.maketrans({ + '⁰': '0', + '¹': '1', + '²': '2', + '³': '3', + '⁴': '4', + '⁵': '5', + '⁶': '6', + '⁷': '7', + '⁸': '8', + '⁹': '9', + '⁺': '+', + '⁻': '-' +}) + + +def _sentence_has_final_answer_phrase(s: str) -> bool: + if not s or not s.strip(): + return False + if re.search('final(?:\\s+numerical)?\\s+answer|final_answer|final_numerical_answer', s, re.I): + return True + return bool(re.search('\\b(?:the\\s+)?answer\\s+is\\b|\\bfinal\\s+conclusion\\b|\\bresult\\s+is\\b', s, re.I)) + + +def _normalize_latex_scientific_for_eval(text: str) -> str: + if not text: + return text + s = text.replace('$', ' ') + s = re.sub('\\\\(?:text|mathrm|textrm)\\s*\\{e([+-])\\}\\s*(\\d+)', 'e\\1\\2', s, flags=re.I) + s = re.sub('\\\\(?:text|mathrm|textrm)\\s*\\{[eE]\\}', 'e', s) + s = re.sub('(-?\\d+(?:\\.\\d+)?)\\s+e\\s*([+-]?\\d+)', '\\1e\\2', s, flags=re.I) + s = re.sub('(-?\\d+(?:,\\d{3})*(?:\\.\\d+)?)\\s*(?:\\\\times|×|[xX])\\s*10\\s*\\^\\s*\\{?\\s*([+-]?\\d+)\\s*\\}?', + '\\1e\\2', s) + + def replace_unicode_exponent(match: re.Match) -> str: + exponent = match.group(2).translate(_SUPERSCRIPT_TO_ASCII) + return f'{match.group(1)}e{exponent}' + + s = re.sub('(-?\\d+(?:,\\d{3})*(?:\\.\\d+)?)\\s*(?:\\\\times|×|[xX])\\s*10([⁰¹²³⁴⁵⁶⁷⁸⁹⁺⁻]+)', + replace_unicode_exponent, s) + return s + + +def _split_for_final_answer_scope(response: str) -> List[str]: + return [p.strip() for p in re.split('[。!?\\n]|(? str: + s = re.sub('\\^\\{[^}]*\\}', '^', s) + s = re.sub('_\\{[^}]*\\}', '_', s) + return s + + +def _should_skip_numeric_token_at(text: str, start: int) -> bool: + if start <= 0: + return False + prev = text[start - 1] + if prev in '^_' or prev.isalpha(): + return True + if start >= 2 and text[start - 2] in '^_' and (text[start - 1] == '{'): + return True + return False + + +def _last_number_token_in_response(response: str) -> Optional[str]: + t = _mask_latex_sup_sub_for_numeric_scan(_normalize_latex_scientific_for_eval(response)) + last: Optional[str] = None + for m in _NUMBER_TOKEN_RE.finditer(t): + if _should_skip_numeric_token_at(t, m.start()): + continue + last = m.group(0) + return last + + +def _last_scientific_number_token_in_response(response: str) -> Optional[str]: + t = _mask_latex_sup_sub_for_numeric_scan(_normalize_latex_scientific_for_eval(response)) + matches = list(_SCI_NUMBER_TOKEN_RE.finditer(t)) + if not matches: + return None + return matches[-1].group(0) + + +def _decimal_digits_in_token(num: str) -> int: + clean_num = num.replace(',', '') + if 'e' in clean_num.lower(): + base = clean_num.lower().split('e')[0] + return len(base.split('.')[-1]) if '.' in base else 0 + return len(clean_num.split('.')[-1]) if '.' in clean_num else 0 + + +def _parse_decimal_places_hint(text: str) -> Optional[int]: + if not text or not str(text).strip(): + return None + t = str(text).strip() + low = t.lower() + m = re.search('(\\d+)\\s*decimal\\s*places?', low) or re.search('(\\d+)\\s*decimal\\b', low) + if m: + return int(m.group(1)) + m = re.search('\\b(zero|one|two|three|four|five|six|seven|eight|nine|ten)\\s+decimal\\s*places?\\b', low) + if m: + return EN_NUM.get(m.group(1)) + if re.search('nearest\\s+integer|0\\s+decimal\\s*places?|no\\s+decimal', low): + return 0 + m = re.search('^\\s*(\\d+)\\s*$', t) + if m: + return int(m.group(1)) + return None + + +def _resolve_decimal_required_digits(required_parameters: str, edit_question: str) -> Tuple[Optional[int], str]: + rp = (required_parameters or '').strip() + eq = (edit_question or '').strip() + for label, blob in (('required_parameters', rp), ('edit_question', eq)): + d = _parse_decimal_places_hint(blob) + if d is not None: + return (d, label) + return (None, '') + + +def _requires_whole_response_numeric_format(edit_question: str) -> bool: + low = re.sub('\\s+', ' ', (edit_question or '').lower()) + patterns = ( + '\\ball intermediate and final numerical (?:values|results|quantities)\\b', + '\\ball intermediate (?:calculations|steps|values|results).{0,100}\\bfinal (?:answer|result)\\b', + '\\ball numerical (?:quantities|values|results|answers).{0,80}\\b(?:output|response|calculation|calculations|reasoning|solution)\\b', # noqa: E501 + '\\bevery (?:numeric|numerical) (?:value|quantity|result).{0,80}\\b(?:output|response|calculation|calculations|reasoning|solution)\\b', # noqa: E501 + '\\b(?:entire|whole) (?:output|response).{0,80}\\b(?:decimal|precision|scientific notation|scientific annotation)\\b' # noqa: E501 + ) + return any((re.search(pattern, low) for pattern in patterns)) + + +def _extract_final_answer_region(response: str) -> Tuple[Optional[str], str]: + try: + parsed = json.loads(response) + except (TypeError, ValueError, json.JSONDecodeError): + parsed = None + if isinstance(parsed, dict): + for key, value in parsed.items(): + normalized_key = re.sub('[\\s-]+', '_', str(key).strip().lower()) + if normalized_key in {'final_answer', 'final_numerical_answer'}: + if isinstance(value, str): + return (value, f'JSON field {key!r}') + return (json.dumps(value, ensure_ascii=False), f'JSON field {key!r}') + marker_re = re.compile( + '\\bfinal(?:\\s+numerical)?\\s+answer\\b|\\bfinal_(?:numerical_)?answer\\b|\\bthe\\s+answer\\s+is\\b|\\bresult\\s+is\\b', # noqa: E501 + re.I) + matches = list(marker_re.finditer(response)) + if matches: + last = matches[-1] + return (response[last.start():], 'final-answer') + return (None, '') + + +def _numeric_tokens(text: str) -> List[str]: + normalized = _mask_latex_sup_sub_for_numeric_scan(_normalize_latex_scientific_for_eval(text)) + tokens: List[str] = [] + for match in _NUMBER_TOKEN_RE.finditer(normalized): + if _should_skip_numeric_token_at(normalized, match.start()): + continue + line_start = normalized.rfind('\n', 0, match.start()) + 1 + prefix = normalized[line_start:match.start()] + suffix = normalized[match.end():] + structural_prefix = re.sub('[\\s#>*_`-]+', '', prefix) + if not structural_prefix and re.match('\\s*[.)]', suffix): + continue + tokens.append(match.group(0)) + return tokens + + +def _format_bad_tokens(tokens: List[str], limit: int = 5) -> str: + shown = ', '.join((repr(token) for token in tokens[:limit])) + if len(tokens) > limit: + shown += f'total{len(tokens)}' + return shown + + +def check_decimal_format(response: str, required_parameters: str, item: Dict = None, **kwargs) -> Dict: + if not isinstance(response, str) or not response.strip(): + return {'score': 0.0, 'detail': 'The response is empty'} + eq = item.get('edit_question', '') or '' if item else '' + required_digits, src = _resolve_decimal_required_digits(required_parameters or '', eq) + if required_digits is None: + return { + 'score': 0.0, + 'detail': f'Unable to parsedecimal placesrequired。required_parameters={required_parameters!r}' + } + if _requires_whole_response_numeric_format(eq): + target_text = response + scope = 'question requiredcheckthe entire response ( intermediate calculations)' + tokens = _numeric_tokens(target_text) + else: + target_text, located_by = _extract_final_answer_region(response) + if target_text is not None: + scope = f'checkfinal answer ({located_by}）' + tokens = _numeric_tokens(target_text) + else: + scope = 'question required Not foundanswer , checkresponse' + all_tokens = _numeric_tokens(response) + tokens = all_tokens[-1:] if all_tokens else [] + if not tokens: + return {'score': 0.0, 'detail': f'{scope}, Not detected'} + bad_tokens = [token for token in tokens if _decimal_digits_in_token(token) != required_digits] + score = 0.0 if bad_tokens else 1.0 + requirement_detail = (f'{required_digits}required ({src}）' + if score else f'does not satisfy{required_digits}required:{_format_bad_tokens(bad_tokens)}') + return {'score': score, 'detail': f'{scope}；check {len(tokens)},{requirement_detail}'} + + +def check_scientific_format(response: str, required_parameters: str = None, item: Dict = None, **kwargs) -> Dict: + if not isinstance(response, str) or not response.strip(): + return {'score': 0.0, 'detail': 'The response is empty'} + eq = item.get('edit_question', '') or '' if item else '' + if _requires_whole_response_numeric_format(eq): + target_text = response + scope = 'question requiredcheckthe entire response ( intermediate calculations)' + tokens = _numeric_tokens(target_text) + else: + target_text, located_by = _extract_final_answer_region(response) + if target_text is not None: + scope = f'checkfinal answer ({located_by}）' + tokens = _numeric_tokens(target_text) + else: + scope = 'question required Not foundanswer , checkresponse' + all_tokens = _numeric_tokens(response) + tokens = all_tokens[-1:] if all_tokens else [] + if not tokens: + return {'score': 0.0, 'detail': f'{scope}, Not detected'} + bad_tokens = [token for token in tokens if _SCI_NUMBER_TOKEN_RE.fullmatch(token) is None] + score = 0.0 if bad_tokens else 1.0 + format_detail = 'scientific notation' if score else f'scientific notation:{_format_bad_tokens(bad_tokens)}' + return {'score': score, 'detail': f'{scope}；check {len(tokens)},{format_detail}'} + + +def is_box_wrapped(text: str) -> bool: + lines = text.splitlines() + if len(lines) < 3: + return False + top = lines[0] + bottom = lines[-1] + middle = lines[1:-1] + if not (top.startswith('┌') and top.endswith('┐')): + return False + if not (bottom.startswith('└') and bottom.endswith('┘')): + return False + for line in middle: + if not (line.startswith('│') and line.endswith('│')): + return False + return True + + +def is_latex_box_wrapped(text: str) -> bool: + if not text: + return False + return bool(re.search('\\\\(?:boxed|fbox)\\s*\\{', text, re.I)) + + +def check_wrap_up(response: str, required_parameters: str, item: Dict = None, **kwargs) -> Dict: + if not isinstance(response, str) or not response.strip(): + return {'score': 0.0, 'detail': 'The response is empty'} + wrapper = required_parameters.strip() + res = response.strip() + success = False + if not wrapper: + if is_box_wrapped(res): + return {'score': 1.0, 'detail': 'box (┌─┐ │ │ └─┘)'} + if is_latex_box_wrapped(res): + return {'score': 1.0, 'detail': 'Detected a LaTeX box wrapper'} + if res.startswith('```') and res.endswith('```') and (len(res) >= 6): + return {'score': 1.0, 'detail': '``` ```'} + if res.startswith('[') and res.endswith(']'): + return {'score': 1.0, 'detail': 'Detected square-bracket wrapping'} + if res.startswith('(') and res.endswith(')'): + return {'score': 1.0, 'detail': '( )'} + if res.startswith('{') and res.endswith('}'): + return {'score': 1.0, 'detail': '{ }'} + return {'score': 0.0, 'detail': 'No supported wrapper was detected'} + if wrapper.startswith('```'): + success = res.startswith(wrapper) and res.endswith('```') + elif wrapper == '[ ]': + success = res.startswith('[') and res.endswith(']') + elif wrapper == '( )': + success = res.startswith('(') and res.endswith(')') + elif wrapper == '{ }': + success = res.startswith('{') and res.endswith('}') + elif wrapper == 'box': + success = is_box_wrapped(res) + elif wrapper.lower() in ('boxed', '\\boxed', 'latex_boxed', 'fbox', '\\fbox'): + success = is_latex_box_wrapped(res) + elif wrapper.startswith('<') and wrapper.endswith('>'): + tag = re.escape(wrapper.strip('<>')) + success = bool(re.match(f'^<{tag}[^>]*>.*?$', res, re.DOTALL)) + else: + success = res.startswith(wrapper) and res.endswith(wrapper) + if success: + return {'score': 1.0, 'detail': f'{wrapper}'} + return {'score': 0.0, 'detail': f'{wrapper}correct'} + + +_CASE_WORD_RE = re.compile('(?:all\\s+)?(?:lower|upper)[\\s\\-]?case|uppercase|lowercase', re.I) +_ANSWER_LOC_RE = re.compile( + '(final\\s*numerical\\s*answer|final\\s*answer|the\\s+answer\\s+is|final_answer|final_numerical_answer)', re.I) + + +def _normalize_case_text(text: str) -> str: + t = (text or '').lower() + t = t.replace('lower-case', 'lowercase').replace('upper-case', 'uppercase') + t = t.replace('lower case', 'lowercase').replace('upper case', 'uppercase') + return t + + +def _case_windows(text: str, radius: int = 160) -> str: + wins = [] + for m in _CASE_WORD_RE.finditer(text or ''): + a = max(0, m.start() - radius) + b = min(len(text), m.end() + radius) + wins.append(text[a:b]) + return ' || '.join(wins) if wins else text or '' + + +def infer_casing_scope(edit_question: str, required_parameters: str = '') -> str: + full = f"{edit_question or ''}\n{required_parameters or ''}" + q = _normalize_case_text(full) + if not _CASE_WORD_RE.search(q): + return 'skip' + ctx = _case_windows(q) + if re.search( + 'output\\s+answer\\s+format|entire\\s+output\\s+answer|final\\s+output\\s+answer|(?:final\\s+)?answer\\s+format|entire\\s+answer\\b|final\\s+answer.{0,60}(?:lower|upper)case|(?:lower|upper)case.{0,40}final\\s+answer|answer\\s+must\\s+be\\s+(?:written\\s+|formatted\\s+|provided\\s+)?in\\s+all\\s+(?:lower|upper)case|format(?:ted)?\\s+your\\s+answer.{0,60}(?:lower|upper)case|answer\\s+as\\b.{0,80}(?:lower|upper)case|label\\s+values?.{0,40}(?:lower|upper)case|values?\\s+in\\s+(?:all\\s+)?(?:lower|upper)case|string\\s+values?.{0,40}(?:lower|upper)case|final\\s+list.{0,40}(?:lower|upper)case|(?:lower|upper)case.{0,40}final\\s+list|output\\s+must\\s+be\\s+the\\s+name.{0,40}(?:lower|upper)case|in\\s+all\\s+(?:lower|upper)case\\s+letters', # noqa: E501 + ctx, + re.S): + if re.search( + 'entire\\s+(?:response|output|reply)\\b(?!\\s+answer)|presented\\s+entirely\\s+in\\s+(?:all\\s+)?(?:lower|upper)case|your\\s+entire\\s+output\\s+must\\s+be\\s+in\\s+all\\s+(?:lower|upper)case', # noqa: E501 + ctx, + re.S): + return 'whole' + return 'answer' + if re.search( + '(?:entire\\s+)?(?:response|output|reply)\\b.{0,50}(?:all\\s+)?(?:lower|upper)case|(?:lower|upper)case.{0,50}(?:entire\\s+)?(?:response|output|reply)\\b|presented\\s+entirely\\s+in\\s+(?:all\\s+)?(?:lower|upper)case|your\\s+entire\\s+output\\s+must\\s+be\\s+in\\s+all\\s+(?:lower|upper)case|reponse\\s+must\\s+be\\s+output\\s+in\\s+all\\s+(?:lower|upper)case|output\\s+must\\s+be\\s+in\\s+all\\s+(?:lower|upper)case|output\\s+(?:should|must)\\s+be\\s+in\\s+(?:all\\s+)?(?:lower|upper)case|ouput\\s+must\\s+be\\s+in\\s+(?:lower|upper)case', # noqa: E501 + ctx, + re.S): + if re.search('output\\s+answer', ctx): + return 'answer' + return 'whole' + if re.search('\\banswer\\b.{0,80}(?:lower|upper)case|(?:lower|upper)case.{0,80}\\banswer\\b', ctx, re.S): + return 'answer' + if re.search('(?:lower|upper)case', ctx): + return 'answer' + return 'skip' + + +def _collect_json_string_values(obj: Any) -> List[str]: + vals: List[str] = [] + + def walk(x: Any) -> None: + if isinstance(x, dict): + for v in x.values(): + walk(v) + elif isinstance(x, list): + for v in x: + walk(v) + elif isinstance(x, str): + vals.append(x) + + walk(obj) + return vals + + +def extract_answer_span_for_casing(response: str) -> Tuple[Optional[str], str]: + if not isinstance(response, str) or not response.strip(): + return (None, 'empty') + text = response.strip() + m = re.search('\\\\(?:boxed|fbox)\\s*\\{([^{}]*(?:\\{[^{}]*\\}[^{}]*)*)\\}', text, re.I) + if m and m.group(1).strip(): + return (m.group(1).strip(), 'boxed') + m = _ANSWER_LOC_RE.search(text) + if m: + tail = text[m.end():].lstrip(' \t:：-') + block = re.split('\\n\\s*\\n', tail, maxsplit=1)[0].strip() + if not block: + line = tail.split('\n', 1)[0].strip() + block = line + if block: + return (block, 'answer_phrase') + cleaned = re.sub('^```[a-zA-Z]*\\s*', '', text) + cleaned = re.sub('\\s*```$', '', cleaned).strip() + try: + obj = json.loads(cleaned) + vals = _collect_json_string_values(obj) + if vals: + return (' '.join(vals), 'json_values') + except Exception: + pass + fence = re.search('```(?:json)?\\s*(\\{.*?\\}|\\[.*?\\])\\s*```', text, re.I | re.S) + if fence: + try: + obj = json.loads(fence.group(1)) + vals = _collect_json_string_values(obj) + if vals: + return (' '.join(vals), 'json_values') + except Exception: + pass + return (None, 'unresolved') + + +def _letters_case_ok(text: str, mode: str) -> bool: + letters = [c for c in text if 'A' <= c <= 'Z' or 'a' <= c <= 'z'] + if not letters: + return False + if mode == 'upper': + return all(('A' <= c <= 'Z' for c in letters)) + return all(('a' <= c <= 'z' for c in letters)) + + +def _check_casing(response: str, mode: str, item: Dict = None, required_parameters: str = '', **kwargs) -> Dict: + if not isinstance(response, str) or not response.strip(): + return {'score': 0.0, 'detail': 'The content is empty', 'skipped': False} + eq = '' + rp = required_parameters or '' + if item and isinstance(item, dict): + eq = item.get('edit_question', '') or '' + if not rp: + rp = kwargs.get('required_parameters', '') or '' + rp = required_parameters or kwargs.get('required_parameters', '') or rp + scope = infer_casing_scope(eq, rp) + if scope == 'skip': + return {'score': 0.0, 'detail': 'lowercase , skipped instruction ( )', 'skipped': True, 'casing_scope': 'skip'} + if scope == 'whole': + target, src = (response, 'whole_response') + else: + target, src = extract_answer_span_for_casing(response) + if target is None: + return { + 'score': 0.0, + 'detail': f'casing_scope=answer, answer ({src}）', + 'skipped': False, + 'casing_scope': 'answer' + } + ok = _letters_case_ok(target, mode) + label = 'uppercase' if mode == 'upper' else 'lowercase' + if ok: + return { + 'score': 1.0, + 'detail': f'casing_scope={scope}, src={src},{label}', + 'skipped': False, + 'casing_scope': scope + } + return { + 'score': 0.0, + 'detail': f'casing_scope={scope}, src={src}, some letters are not {label}', + 'skipped': False, + 'casing_scope': scope + } + + +def check_uppercase(response: str, item: Dict = None, required_parameters: str = '', **kwargs) -> Dict: + return _check_casing(response, 'upper', item=item, required_parameters=required_parameters, **kwargs) + + +def check_lowercase(response: str, item: Dict = None, required_parameters: str = '', **kwargs) -> Dict: + return _check_casing(response, 'lower', item=item, required_parameters=required_parameters, **kwargs) + + +_FINAL_ANSWER_RE = re.compile( + 'final[\\s_-]+(?:numerical[\\s_-]+)?answer\\s*[:：]?|the\\s+answer\\s+is\\s*(?:[:：]|\\.(?=\\s|$))?|(? str: + q = (edit_question or '').lower() + strong_whole_patterns = [ + 'entire\\s+(?:final\\s+)?(?:response|output|model\\s+output|reply)', + 'format\\s+your\\s+entire\\s+(?:response|output|reply)', + 'all\\s+(?:of\\s+the\\s+)?(?:output|content)\\s+must\\s+be\\s+(?:formatted\\s+)?(?:as|in)', 'output\\s+only\\b', + 'no\\s+additional\\s+(?:text|content|explanation)' + ] + if any((re.search(p, q, re.I) for p in strong_whole_patterns)): + return 'whole' + if re.search('\\bfinal(?:[\\s-]+\\w+){0,4}[\\s-]+(?:answer|output)\\b', q, re.I): + return 'final' + weak_whole = '(?:your|the)\\s+(?:response|reply)\\s+must\\s+be\\s+(?:valid\\s+)?(?:formatted\\s+)?(?:as|in)\\s+(?:a\\s+)?(?:json|list|tuple|dictionary|markdown|html|xml|csv)\\b' # noqa: E501 + if re.search(weak_whole, q, re.I): + return 'whole' + return 'final' + + +def _strip_outer_code_fence(text: str) -> str: + s = (text or '').strip() + m = re.fullmatch('```(?:[A-Za-z0-9_+.-]+)?\\s*\\n?([\\s\\S]*?)\\n?```', s) + return m.group(1).strip() if m else s + + +def _extract_last_code_fence(text: str) -> Optional[str]: + matches = list(re.finditer('```(?:[A-Za-z0-9_+.-]+)?\\s*\\n?([\\s\\S]*?)\\n?```', text or '')) + return matches[-1].group(1).strip() if matches else None + + +def _extract_balanced_braces(text: str, open_pos: int) -> Optional[str]: + depth = 0 + for i in range(open_pos, len(text)): + if text[i] == '{': + depth += 1 + elif text[i] == '}': + depth -= 1 + if depth == 0: + return text[open_pos + 1:i].strip() + return None + + +def _unwrap_answer_container(text: str) -> str: + s = _strip_outer_code_fence(text) + boxed = list(re.finditer('\\\\(?:boxed|fbox)\\s*\\{', s, re.I)) + if boxed: + open_pos = s.find('{', boxed[-1].start()) + inner = _extract_balanced_braces(s, open_pos) + if inner: + s = inner + s = s.strip() + if s.startswith('\\[') and s.endswith('\\]'): + s = s[2:-2].strip() + if s.startswith('$') and s.endswith('$') and (len(s) >= 2): + s = s[1:-1].strip() + return s + + +def extract_format_target(response: str, edit_question: str) -> Tuple[str, str, str]: + scope = infer_format_scope(edit_question) + if scope == 'whole': + return (_strip_outer_code_fence(response), scope, 'whole_response') + matches = list(_FINAL_ANSWER_RE.finditer(response or '')) + if matches: + tail = (response or '')[matches[-1].end():].strip() + tail = re.sub( + '^(?:\\((?:tuple|list|dictionary|dict|json|html|xml|csv|markdown|answer)[^()\\n]{0,40}\\))?\\s*[:：]?\\s*\\*{0,2}\\s*', # noqa: E501 + '', + tail, + flags=re.I).strip() + if tail: + return (_unwrap_answer_container(tail), scope, 'answer_phrase') + boxed = list(re.finditer('\\\\(?:boxed|fbox)\\s*\\{', response or '', re.I)) + if boxed: + open_pos = (response or '').find('{', boxed[-1].start()) + inner = _extract_balanced_braces(response or '', open_pos) + if inner: + return (_unwrap_answer_container(inner), scope, 'boxed_answer') + fenced = _extract_last_code_fence(response or '') + if fenced: + return (_unwrap_answer_container(fenced), scope, 'last_code_fence') + markup = re.search('(<([A-Za-z_][\\w:.-]*)\\b[^>]*>[\\s\\S]*)\\s*$', (response or '').strip(), re.I) + if markup: + return (markup.group(1).strip(), scope, 'final_markup') + blocks = [p.strip() for p in re.split('\\n\\s*\\n', response or '') if p.strip()] + target = blocks[-1] if blocks else (response or '').strip() + return (_unwrap_answer_container(target), scope, 'last_block') + + +def _literal_value(text: str): + cleaned = _unwrap_answer_container(text) + try: + return json.loads(cleaned) + except Exception: + try: + return ast.literal_eval(cleaned) + except Exception: + return None + + +def _valid_list_target(text: str) -> bool: + value = _literal_value(text) + if isinstance(value, list): + return True + cleaned = _unwrap_answer_container(text) + if re.search('\\\\begin\\{(?:itemize|enumerate)\\}[\\s\\S]*\\\\item\\b[\\s\\S]*\\\\end\\{(?:itemize|enumerate)\\}', + cleaned): + return True + return bool(re.search('^\\s*(?:[-*+]|\\d+[.)])\\s+\\S+', cleaned, re.M)) + + +def _valid_tuple_target(text: str) -> bool: + value = _literal_value(text) + if isinstance(value, tuple): + return True + cleaned = _unwrap_answer_container(text) + return bool(re.fullmatch('\\(\\s*[^(),]+(?:\\s*,\\s*[^(),]+)+\\s*\\)', cleaned, re.S)) + + +def _extract_required_dict_keys(edit_question: str) -> List[str]: + snippets = re.findall('\\{[^{}]{1,1000}\\}', edit_question or '', re.S) + for snippet in reversed(snippets): + value = _literal_value(snippet) + if isinstance(value, dict) and value: + return [str(k) for k in value.keys()] + return [] + + +def _valid_dictionary_target(text: str, edit_question: str) -> Tuple[bool, List[str]]: + value = _literal_value(text) + if not isinstance(value, dict): + return (False, []) + required_keys = _extract_required_dict_keys(edit_question) + return (all((k in value for k in required_keys)), required_keys) + + +_VOID_HTML_TAGS = frozenset( + {'area', 'base', 'br', 'col', 'embed', 'hr', 'img', 'input', 'link', 'meta', 'source', 'track', 'wbr'}) + + +class _HTMLStructureParser(HTMLParser): + + def __init__(self): + super().__init__(convert_charrefs=True) + self.stack: List[str] = [] + self.roots: List[str] = [] + self.invalid = False + self.text_outside = False + + def handle_starttag(self, tag, attrs): + tag = tag.lower() + if not self.stack: + self.roots.append(tag) + if tag not in _VOID_HTML_TAGS: + self.stack.append(tag) + + def handle_startendtag(self, tag, attrs): + if not self.stack: + self.roots.append(tag.lower()) + + def handle_endtag(self, tag): + tag = tag.lower() + if not self.stack or self.stack[-1] != tag: + self.invalid = True + return + self.stack.pop() + + def handle_data(self, data): + if data.strip() and (not self.stack): + self.text_outside = True + + +def _expected_root_tag(edit_question: str, language: str) -> Optional[str]: + keyword_pos = (edit_question or '').lower().rfind(language.lower()) + scope = (edit_question or '')[keyword_pos:] if keyword_pos >= 0 else edit_question or '' + tags = re.findall('<([A-Za-z_][\\w:.-]*)\\b[^>]*>', scope) + return tags[0].lower() if tags else None + + +def _valid_html_target(text: str, edit_question: str) -> Tuple[bool, Optional[str]]: + cleaned = _unwrap_answer_container(text) + parser = _HTMLStructureParser() + try: + parser.feed(cleaned) + parser.close() + except Exception: + return (False, None) + expected = _expected_root_tag(edit_question, 'html') + ok = bool(parser.roots) and (not parser.stack) and (not parser.invalid) and (not parser.text_outside) + if expected: + ok = ok and parser.roots[0] == expected + return (ok, expected) + + +def _valid_xml_target(text: str, edit_question: str) -> Tuple[bool, Optional[str]]: + cleaned = _unwrap_answer_container(text) + expected = _expected_root_tag(edit_question, 'xml') + try: + root = ET.fromstring(cleaned) + except ET.ParseError: + return (False, expected) + actual = root.tag.split('}')[-1].lower() + return (not expected or actual == expected, expected) + + +def check_json_format(response: str, item: Dict = None, **kwargs) -> Dict: + if not response: + return {'score': 0.0, 'detail': 'Empty response'} + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + cleaned = _strip_outer_code_fence(target) + try: + json.loads(cleaned) + return {'score': 1.0, 'detail': f'scope={scope}, src={src}, JSON'} + except Exception as e: + return { + 'score': 0.0, + 'detail': f'scope={scope}, src={src}, JSON Parsing failed: {str(e)} | cleaned={cleaned[:100]}' + } + + +def check_list_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + ok = _valid_list_target(target) + return { + 'score': 1.0 if ok else 0.0, + 'detail': f"scope={scope}, src={src}, {('Detected a valid' if ok else 'valid')}" + } + + +def check_tuple_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + ok = _valid_tuple_target(target) + return { + 'score': 1.0 if ok else 0.0, + 'detail': f"scope={scope}, src={src}, {('Detected a valid' if ok else 'valid')}" + } + + +def check_dictionary_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + ok, keys = _valid_dictionary_target(target, q) + key_note = f', required_keys={keys}' if keys else '' + return { + 'score': 1.0 if ok else 0.0, + 'detail': f"scope={scope}, src={src}{key_note}, {('The format is valid' if ok else 'valid')}" + } + + +def check_markdown_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + patterns = ['^#{1,6}\\s+', '^\\s*[-*+]\\s+', '^\\s*\\d+\\.\\s+', '```', '\\[.+?\\]\\(.+?\\)'] + for p in patterns: + if re.search(p, target, re.MULTILINE): + return {'score': 1.0, 'detail': f'scope={scope}, src={src}, Detected Markdown :{p}'} + return {'score': 0.0, 'detail': f'scope={scope}, src={src}, Not detected Markdown'} + + +def check_html_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + ok, expected = _valid_html_target(target, q) + return { + 'score': 1.0 if ok else 0.0, + 'detail': + f"scope={scope}, src={src}, expected_root={expected or 'any'}, {('HTML valid' if ok else 'HTML invalid')}" + } + + +def check_xml_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + ok, expected = _valid_xml_target(target, q) + return { + 'score': 1.0 if ok else 0.0, + 'detail': + f"scope={scope}, src={src}, expected_root={expected or 'any'}, {('XML valid' if ok else 'XML invalid')}" + } + + +def check_csv_format(response: str, item: Dict = None, **kwargs) -> Dict: + q = (item or {}).get('edit_question', '') + target, scope, src = extract_format_target(response, q) + try: + rows = [row for row in csv.reader(io.StringIO(target)) if any((cell.strip() for cell in row))] + except csv.Error as e: + return {'score': 0.0, 'detail': f'scope={scope}, src={src}, CSV Parsing failed: {e}'} + if not rows: + return {'score': 0.0, 'detail': f'scope={scope}, src={src}, CSV The content is empty'} + counts = [len(row) for row in rows] + if len(set(counts)) == 1 and counts[0] > 1: + return {'score': 1.0, 'detail': f'scope={scope}, src={src}, CSV passed, total{len(rows)},{counts[0]}'} + line_match = re.search('\\b(\\d+|one|two|three|four|five)\\s+lines?\\b', q, re.I) + if line_match and len(set(counts)) == 1 and (counts[0] == 1): + raw_n = line_match.group(1).lower() + required_lines = int(raw_n) if raw_n.isdigit() else EN_NUM.get(raw_n) + if required_lines == len(rows): + return {'score': 1.0, 'detail': f'scope={scope}, src={src}, CSV passed, total{len(rows)}'} + return {'score': 0.0, 'detail': f'scope={scope}, src={src}, CSV inconsistent Not detected'} + + +def _normalize_for_match(s: str) -> str: + return re.sub('\\s+', ' ', re.sub('[^a-z0-9\\u4e00-\\u9fff]+', ' ', s.lower())).strip() + + +def _extract_gt_blockquote_options(edit_question: str) -> List[str]: + q = edit_question + low = q.lower() + idx = low.find('choice list') + if idx != -1: + q = q[idx:] + out: List[str] = [] + for line in q.splitlines(): + s = line.strip() + if not s.startswith('>'): + continue + rest = s[1:].strip() + if rest: + out.append(rest) + return out + + +def _extract_choose_options_from_question(edit_question: str) -> Tuple[List[str], str]: + gt_opts = _extract_gt_blockquote_options(edit_question) + if gt_opts: + return (gt_opts, 'CHOICE LIST (> )') + q_lower = edit_question.lower() + keywords = ['choice list', 'choices', 'choice', 'options', 'option'] + start_idx = None + for kw in keywords: + i = q_lower.find(kw) + if i != -1: + start_idx = i + len(kw) + break + tail_original = edit_question[start_idx:] if start_idx is not None else edit_question + parts_original = [p.strip() for p in re.split('[。!?]|(?= 1: + contents.append(val) + if contents: + return (contents, 'question choice/options') + for m in option_pat.finditer(edit_question): + val = m.group(2).strip() + if val: + contents.append(val) + if contents: + return (contents, 'question option') + loose_gt: List[str] = [] + for line in edit_question.splitlines(): + s = line.strip() + if s.startswith('>'): + rest = s[1:].strip() + if rest: + loose_gt.append(rest) + if loose_gt: + return (loose_gt, 'question >') + return ([], '') + + +def check_choose_from(response: str, required_parameters: str = None, item: Dict = None, **kwargs) -> Dict: + edit_question = item.get('edit_question', '') if item else '' + if not response or not edit_question: + return {'score': 0.0, 'detail': 'question'} + res_norm = _normalize_for_match(response) + option_texts, scope_note = _extract_choose_options_from_question(edit_question) + if option_texts: + for opt in option_texts: + opt_stripped = opt.strip() + if opt_stripped and opt_stripped in response: + return { + 'score': 1.0, + 'detail': f"option ({scope_note}): {opt[:120]}{('...' if len(opt) > 120 else '')}" + } + o_norm = _normalize_for_match(opt) + if len(o_norm) < 2: + continue + if o_norm in res_norm: + return { + 'score': 1.0, + 'detail': f"option ({scope_note}): {opt[:120]}{('...' if len(opt) > 120 else '')}" + } + return {'score': 0.0, 'detail': f'response option ({scope_note}); option ={len(option_texts)}'} + q_lower = edit_question.lower() + letters = set(re.findall('\\b([a-z])[\\.\\)]', q_lower)) + nums = set(re.findall('\\b(\\d+)[\\.\\)]', q_lower)) + res_lower = response.lower().strip() + for ch in sorted(letters): + if re.search(f'(?:^|\\s){re.escape(ch)}(?:\\s|[\\.\\)]|$)', res_lower): + return {'score': 1.0, 'detail': f'option :{ch}(question option , )'} + for n in sorted(nums, key=len, reverse=True): + if re.search(f'(?:^|\\s){re.escape(n)}(?:\\s|[\\.\\)]|$)', res_lower): + return {'score': 1.0, 'detail': f'option :{n}(question option , )'} + return {'score': 0.0, 'detail': 'question option, response'} + + +def check_judge(response: str, item: Dict = None, **kwargs) -> Dict: + res = response.strip().lower() + match = re.search('\\b(yes|no|true|false)\\b', res) + if match: + return {'score': 1.0, 'detail': f'Detected :{match.group()}'} + return {'score': 0.0, 'detail': 'Not detected (Yes/No/True/False)'} + + +def check_number_response(response: str, required_parameters: str, item: Dict = None, **kwargs) -> Dict: + if not isinstance(response, str) or not response.strip(): + return {'score': 0.0, 'detail': 'The response is empty'} + m = re.search('\\d+', str(required_parameters)) + if not m: + return {'score': 0.0, 'detail': f'Unable to parse required_parameters: {required_parameters}'} + n = int(m.group()) + if n <= 2: + return {'score': 0.0, 'detail': f'N > 2, actual:{n}'} + llm_client = kwargs.get('llm_client') + judge_model = kwargs.get('judge_model') + if llm_client is None: + return {'score': 0.0, 'detail': 'llm_client, cannot LLM-judge'} + edit_question = item.get('edit_question', '') if isinstance(item, dict) else '' + prompt = f'You are a strict evaluator.\n\nDecide whether the model output contains EXACTLY {n} distinct response types/categories.\n\nGuidelines:\n- A "response type/category" means a clearly separable class of outputs.\n- Count distinct categories, not wording variations.\n\nReturn ONLY in JSON format:\n{{"answer": "YES or NO", "reason": "explain how many categories you found and why"}}\n\nQuestion:\n{edit_question[:1200]}\n\nModel output:\n{response[:3000]}\n' # noqa: E501 + try: + if hasattr(llm_client, 'chat') and hasattr(getattr(llm_client.chat, 'completions', None), 'create'): + resp = llm_client.chat.completions.create(model=judge_model, + messages=[{ + 'role': 'system', + 'content': 'You are a strict evaluator.' + }, { + 'role': 'user', + 'content': prompt + }], + temperature=0) + text = (resp.choices[0].message.content or '').strip() + elif hasattr(llm_client, 'complete'): + text = str(llm_client.complete(prompt)).strip() + else: + text = str(llm_client(prompt)).strip() + if '{' in text: + start = text.index('{') + end = text.rindex('}') + 1 + result = json.loads(text[start:end]) + answer = str(result.get('answer', '')).lower() + reason = result.get('reason', '') + if 'yes' in answer: + return {'score': 1.0, 'detail': f'LLM-judge: YES (N={n}) | {reason}'} + elif 'no' in answer: + return {'score': 0.0, 'detail': f'LLM-judge: NO (N={n}) | {reason}'} + return {'score': 0.0, 'detail': f'LLM-judge Unable to parse:{text[:200]}'} + except Exception as e: + return {'score': 0.0, 'detail': f'LLM-judge error: {e}'} + + +def _split_sentences(text: str) -> List[str]: + parts = re.split('(?<=[。!?]|(? Dict: + if not isinstance(response, str) or not response.strip(): + return {'score': 0.0, 'detail': 'The model response is empty'} + pos_match = re.search('\\b(beginning|middle|end)\\b', required_parameters.lower()) + if not pos_match: + return {'score': 0.0, 'detail': 'required (beginning/middle/end)'} + target_pos = pos_match.group() + sentences = _split_sentences(response) + if not sentences: + return {'score': 0.0, 'detail': 'cannot'} + answer_phrase_re = re.compile( + '(final\\s*numerical\\s*answer|final\\s*answer|final\\s+analysis|final_answer|final_numerical_answer|(?:the\\s+)?answer\\s+is|final\\s+conclusion|result\\s+is)', # noqa: E501 + re.I) + idx = None + for i, sent in enumerate(sentences): + if answer_phrase_re.search(sent): + idx = i + break + if idx is None: + return { + 'score': + 0.0, + 'detail': + 'Not foundanswer ( final answer / final numerical answer / final analysis / ' + 'The answer is / answer is / final conclusion / result is / )' + } + n = len(sentences) + ratio = idx / max(n - 1, 1) if n > 1 else 0.5 + if target_pos == 'beginning' and ratio < 0.2 or (target_pos == 'middle' + and 0.2 <= ratio <= 0.8) or (target_pos == 'end' and ratio > 0.8): + return {'score': 1.0, 'detail': f'answer{idx + 1}/{n},{target_pos}( ≈{ratio:.2f})'} + return {'score': 0.0, 'detail': f'answer{idx + 1}/{n}, .required:{target_pos}, ≈{ratio:.2f}'} diff --git a/vlmeval/dataset/utils/scimif/geography_format_geocoding_validation.py b/vlmeval/dataset/utils/scimif/geography_format_geocoding_validation.py new file mode 100644 index 000000000..4d6be26bc --- /dev/null +++ b/vlmeval/dataset/utils/scimif/geography_format_geocoding_validation.py @@ -0,0 +1,129 @@ +import json +import re +from typing import Any, Dict, List, Optional + +_PLACEHOLDER_SUBSTRINGS = ('specific geographical address format', 'specific format of geographical') + + +def evaluate_geography_address(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not response or not str(response).strip(): + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + rp = (required_parameters or '').strip() + eq = item.get('edit_question', '') or '' if item else '' + if _is_placeholder_required_parameters(rp): + rp = '' + schema_keys = _extract_address_schema_keys(rp, eq) + if schema_keys: + obj = _try_parse_json_object_from_response(response) + if obj is not None: + missing = [k for k in schema_keys if not _dict_has_schema_key(obj, k)] + if not missing: + return {'passed': True, 'score': 1.0, 'detail': f'JSON field:{schema_keys}'} + return {'passed': False, 'score': 0.0, 'detail': f'JSON field{missing}(required :{schema_keys}）'} + ok, why = _heuristic_address_match(response) + if ok: + return {'passed': True, 'score': 1.0, 'detail': f'{why}; JSON field{schema_keys}'} + return {'passed': False, 'score': 0.0, 'detail': f'Not found JSON passed ; required field:{schema_keys}'} + ok, why = _heuristic_address_match(response) + if ok: + return {'passed': True, 'score': 1.0, 'detail': why} + return {'passed': False, 'score': 0.0, 'detail': 'The output does not satisfy required'} + + +def _heuristic_address_match(response: str) -> tuple: + level_indicators = [ + '\\d+°\\d+[\'\\"]?\\s*[NS]\\s*\\d+°\\d+[\'\\"]?\\s*[EW]', '(?:Province|City|District|County|Street|Road|Ave)', + '[A-Za-z\\s]+,\\s*[A-Za-z\\s]+,\\s*[A-Za-z\\s]+', '\\d+\\s*(?:km|miles?)\\s*(?:north|south|east|west)' + ] + for p in level_indicators: + if re.search(p, response, re.I): + return (True, 'The output satisfies ( )') + parts = re.split('[,，/\\-]\\s*', response) + if len([p for p in parts if len(p.strip()) > 2]) >= 2: + return (True, 'The output contains ( )') + return (False, '') + + +def _is_placeholder_required_parameters(rp: str) -> bool: + if not rp: + return True + low = rp.lower() + return any((s in low for s in _PLACEHOLDER_SUBSTRINGS)) + + +def _extract_address_schema_keys(rp: str, eq: str) -> List[str]: + for blob in (rp, eq): + keys = _extract_keys_from_text(blob or '') + if keys: + return keys + return [] + + +def _extract_keys_from_text(text: str) -> List[str]: + if not text.strip(): + return [] + i = text.find('{') + j = text.rfind('}') + if i != -1 and j > i: + snippet = text[i:j + 1] + try: + d = json.loads(snippet) + if isinstance(d, dict) and d: + return list(d.keys()) + except json.JSONDecodeError: + pass + keys = re.findall('"([^"]+)"\\s*:', snippet) + if not keys: + keys = re.findall("'([^']+)'\\s*:", snippet) + if keys: + return list(dict.fromkeys(keys)) + low = text.lower() + for prefix in ('geographical hierarchy format:', 'geographical address format:', 'address format:', + 'hierarchy format:', 'hierarchy:', 'format:'): + if prefix in low: + idx = low.find(prefix) + rest = text[idx + len(prefix):].strip() + rest = rest.split('\n')[0] + parts = re.split('[,，;]', rest) + out: List[str] = [] + for p in parts: + p = p.strip().strip('"\'') + if p and len(p) < 100: + out.append(p) + if out: + return out + return [] + + +def _try_parse_json_object_from_response(response: str) -> Optional[Dict[str, Any]]: + t = response.strip() + t = re.sub('^```[a-zA-Z]*\\s*', '', t) + t = re.sub('\\s*```\\s*$', '', t).strip() + i = t.find('{') + j = t.rfind('}') + if i == -1 or j <= i: + return None + try: + obj = json.loads(t[i:j + 1]) + return obj if isinstance(obj, dict) else None + except json.JSONDecodeError: + return None + + +def _norm_key(s: str) -> str: + s = s.strip().lower() + s = s.replace('-', '_') + s = re.sub('[\\s/]+', '_', s) + return s + + +def _dict_has_schema_key(obj: Dict[str, Any], expected: str) -> bool: + ne = _norm_key(expected) + for k in obj.keys(): + if _norm_key(k) == ne: + return True + return False diff --git a/vlmeval/dataset/utils/scimif/life_format_entity_relationship_validation.py b/vlmeval/dataset/utils/scimif/life_format_entity_relationship_validation.py new file mode 100644 index 000000000..347aafcb7 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/life_format_entity_relationship_validation.py @@ -0,0 +1,24 @@ +import re +from typing import Any, Dict + + +def evaluate_entity_relationship(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not response: + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + tuple_pattern = '\\([^)]+\\)' + tuples_found = re.findall(tuple_pattern, response) + valid_tuples = [] + for t in tuples_found: + inner = t[1:-1] + parts = [p.strip().strip('"\'') for p in re.split('[,，]', inner)] + if 2 <= len(parts) <= 5 and all((len(p) > 0 for p in parts)): + valid_tuples.append(t) + if len(valid_tuples) >= 1: + return {'passed': True, 'score': 1.0, 'detail': f'The output contains {len(valid_tuples)}'} + if re.search('\\([^,]+,\\s*[^,]+,\\s*[^)]+\\)', response): + return {'passed': True, 'score': 1.0, 'detail': 'The output satisfies (subject, relation, object)'} + return {'passed': False, 'score': 0.0, 'detail': 'The output does not contain a valid'} diff --git a/vlmeval/dataset/utils/scimif/life_sequence_length_checking.py b/vlmeval/dataset/utils/scimif/life_sequence_length_checking.py new file mode 100644 index 000000000..901041d2b --- /dev/null +++ b/vlmeval/dataset/utils/scimif/life_sequence_length_checking.py @@ -0,0 +1,24 @@ +import re +from typing import Any, Dict + + +def evaluate_sequence_length(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not response: + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + params_lower = (required_parameters or '').lower() + answer = item.get('answer', '') + numbers = re.findall('\\b(\\d+)\\b', response) + if numbers: + answer_nums = re.findall('\\b(\\d+)\\b', str(answer)) + if answer_nums and numbers: + if any((n in answer_nums for n in numbers)): + return {'passed': True, 'score': 1.0, 'detail': 'numeric value answer is consistent'} + return {'passed': True, 'score': 1.0, 'detail': 'The output contains numeric value'} + length_indicators = ['longest', 'shortest', 'length', 'bp', 'nt', 'amino acid', 'ORF'] + if any((ind in response.lower() or ind in params_lower for ind in length_indicators)): + return {'passed': True, 'score': 1.0, 'detail': 'The output discusses sequence length'} + return {'passed': False, 'score': 0.0, 'detail': 'The output does not satisfy the sequence-length constraint'} diff --git a/vlmeval/dataset/utils/scimif/materials_format_characterization_technique_validation.py b/vlmeval/dataset/utils/scimif/materials_format_characterization_technique_validation.py new file mode 100644 index 000000000..3fc34cb7a --- /dev/null +++ b/vlmeval/dataset/utils/scimif/materials_format_characterization_technique_validation.py @@ -0,0 +1,43 @@ +import re +from typing import Any, Dict, List + +DEFAULT_TECHNIQUES = [ + 'SEM', 'TEM', 'XRD', 'AFM', 'XPS', 'FTIR', 'Raman', 'DSC', 'TGA', 'BET', 'UV-Vis', 'NMR', 'EDX', 'EDS', 'STEM', + 'HRTEM', 'SAED', 'DTA', 'TMA', 'DMA', 'ICP', 'XRF', 'SIMS', 'ESCA' +] + + +def evaluate_characterization_technique(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not response: + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + techniques = _parse_technique_list(required_parameters) + if not techniques: + techniques = DEFAULT_TECHNIQUES + mentioned = [] + for t in techniques: + if re.search('\\b' + re.escape(t) + '\\b', response, re.I): + mentioned.append(t) + if not mentioned: + return {'passed': False, 'score': 0.0, 'detail': 'The output does not contain a valid'} + invalid = [] + tech_pattern = '\\b(SEM|TEM|XRD|AFM|XPS|FTIR|Raman|DSC|TGA|BET|NMR|EDX|EDS|HRTEM)\\b' + for m in re.finditer(tech_pattern, response, re.I): + if m.group(1).upper() not in [x.upper() for x in techniques]: + invalid.append(m.group(1)) + if invalid and (not mentioned): + return {'passed': False, 'score': 0.0, 'detail': f'The output contains :{invalid}'} + return {'passed': True, 'score': 1.0, 'detail': f'The output contains a valid :{mentioned}'} + + +def _parse_technique_list(required_parameters: str) -> List[str]: + if not required_parameters: + return [] + text = required_parameters.strip() + for sep in [',', ';', '、', '|']: + if sep in text: + return [t.strip() for t in text.split(sep) if t.strip()] + return [text] if text else [] diff --git a/vlmeval/dataset/utils/scimif/materials_property_prediction_checking.py b/vlmeval/dataset/utils/scimif/materials_property_prediction_checking.py new file mode 100644 index 000000000..a3de9bedb --- /dev/null +++ b/vlmeval/dataset/utils/scimif/materials_property_prediction_checking.py @@ -0,0 +1,27 @@ +import re +from typing import Any, Dict + + +def evaluate_property_prediction(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not response: + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + params_lower = (required_parameters or '').lower() + is_discrete = 'discrete' in params_lower or 'classification' in params_lower or 'category' in params_lower + is_continuous = 'continuous' in params_lower or 'numeric value' in params_lower + if is_discrete: + labels = re.findall('\\b(Yes|No|High|Low|Medium|Class \\d+|Category \\d+)\\b', response, re.I) + if labels: + return {'passed': True, 'score': 1.0, 'detail': 'The output contains a discrete label'} + if re.search('[A-Za-z]{3,}', response) and (not re.search('\\d+\\.\\d+', response)): + return {'passed': True, 'score': 1.0, 'detail': 'The output is label-like'} + return {'passed': False, 'score': 0.0, 'detail': 'A discrete label is required'} + if is_continuous: + numbers = re.findall('\\b\\d+\\.?\\d*\\b', response) + if numbers: + return {'passed': True, 'score': 1.0, 'detail': 'The output contains a numeric value'} + return {'passed': False, 'score': 0.0, 'detail': 'A numeric value is required'} + return {'passed': True, 'score': 1.0, 'detail': 'No property type was specified'} diff --git a/vlmeval/dataset/utils/scimif/options_matching.py b/vlmeval/dataset/utils/scimif/options_matching.py new file mode 100644 index 000000000..398ec9f76 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/options_matching.py @@ -0,0 +1,169 @@ +import json +import re +from typing import Any, Dict, List, Optional, Tuple + +_JSON_SCHEMA_KEYS = frozenset({'lighting_condition', 'platform', 'view_direction', 'weather'}) + + +def evaluate_options_constraint(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not response or not str(response).strip(): + return {'passed': False, 'score': 0.0, 'detail': 'response is empty'} + eq = item.get('edit_question', '') or '' if item else '' + labels, label_src = _resolve_option_labels(required_parameters or '', eq) + if not labels: + return { + 'passed': False, + 'score': 0.0, + 'detail': + f'cannot required_parameters edit_question validlabel ; required_parameters={required_parameters!r}' + } + response_clean = response.strip() + invalid_found: List[str] = [] + passed = _check_options_in_response(response_clean, labels, invalid_found) + src_note = f' (labelsource: {label_src})' + if passed: + return {'passed': True, 'score': 1.0, 'detail': f'The output satisfiesoption :{labels}{src_note}'} + return { + 'passed': + False, + 'score': + 0.0, + 'detail': (f'The output contains option:{invalid_found}' + if invalid_found else f'The output does not satisfyoption: {labels}') + src_note + } + + +def _parse_labels(required_parameters: str) -> List[str]: + text = required_parameters.strip() + for sep in [',', ';', '|']: + if sep in text: + parts = [p.strip().strip('"\'') for p in text.split(sep)] + return [p for p in parts if p] + return [text] if text else [] + + +def _resolve_option_labels(required_parameters: str, edit_question: str) -> Tuple[List[str], str]: + rp = (required_parameters or '').strip() + if rp: + labels = _parse_labels(rp) + if labels: + return (labels, 'required_parameters') + labels = _parse_labels_from_edit_question(edit_question or '') + if labels: + return (labels, 'edit_question') + return ([], '') + + +def _parse_labels_from_edit_question(text: str) -> List[str]: + if not text or not text.strip(): + return [] + seen = set() + out: List[str] = [] + + def add(s: str) -> None: + s = (s or '').strip() + if not s or len(s) > 200: + return + low = s.lower() + if low in seen: + return + seen.add(low) + out.append(s) + + for m in re.finditer('\\[([^\\]]+)\\]', text, re.DOTALL): + inner = m.group(1).strip() + if not inner: + continue + try: + parsed = json.loads('[' + inner + ']') + if isinstance(parsed, list): + for x in parsed: + if isinstance(x, str): + add(x) + continue + except json.JSONDecodeError: + pass + for part in re.split(',', inner): + part = part.strip().strip('"\'') + if part: + add(part) + for line in text.splitlines(): + line = line.strip() + if '|' not in line or '"' not in line: + continue + for q in re.findall('"([^"]+)"', line): + if q in _JSON_SCHEMA_KEYS: + continue + if len(q) > 120: + continue + add(q) + low = text.lower() + if 'choice list' in low: + idx = low.find('choice list') + tail = text[idx:] + for ln in tail.splitlines(): + s = ln.strip() + if not s.startswith('>'): + continue + rest = s[1:].strip() + if rest and len(rest) < 500: + add(rest) + return out + + +def _try_parse_json_object_from_response(response: str) -> Optional[Dict[str, Any]]: + t = response.strip() + t = re.sub('^```[a-zA-Z]*\\s*', '', t) + t = re.sub('\\s*```\\s*$', '', t).strip() + i = t.find('{') + j = t.rfind('}') + if i == -1 or j <= i: + return None + try: + obj = json.loads(t[i:j + 1]) + return obj if isinstance(obj, dict) else None + except json.JSONDecodeError: + return None + + +def _check_options_in_response(response: str, labels: List[str], invalid_found: List[str]) -> bool: + labels_lower = {s.lower() for s in labels} + obj = _try_parse_json_object_from_response(response) + if obj is not None: + str_vals = [v for v in obj.values() if isinstance(v, str)] + if str_vals: + ok = True + for k, v in obj.items(): + if not isinstance(v, str): + continue + if v.lower() in labels_lower or v in labels: + continue + invalid_found.append(f'{k}={v!r}') + ok = False + return ok + parts = re.split('[,;]\\s*', response) + answer_like = [p.strip().strip('"\'') for p in parts if p.strip() and len(p.strip()) < 50] + if len(answer_like) >= 3 and (not ('{' in response and '"' in response)): + for p in answer_like: + if p.lower() not in labels_lower and (not p.isdigit()): + invalid_found.append(p) + return len(invalid_found) == 0 + words = re.findall('"([^"]+)"', response) or re.findall('\\b(Yes|No|Maybe|Unknown)\\b', response, re.I) + if not words: + words = re.findall('\\b([A-Za-z][a-z]{1,25})\\b', response) + for w in words: + if w in _JSON_SCHEMA_KEYS: + continue + if w.lower() in labels_lower or w in labels: + continue + if w.isdigit() or len(w) > 40: + continue + if w.lower() in {'the', 'and', 'or', 'is', 'are', 'to', 'of', 'in', 'a', 'an', 'json'}: + continue + invalid_found.append(w) + has_valid = any((re.search(re.escape(label), response, re.I) for label in labels)) + return has_valid and len(invalid_found) == 0 diff --git a/vlmeval/dataset/utils/scimif/rdkit_utils.py b/vlmeval/dataset/utils/scimif/rdkit_utils.py new file mode 100644 index 000000000..6e2bc68b4 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/rdkit_utils.py @@ -0,0 +1,45 @@ +from __future__ import annotations +from typing import Any, Optional + + +def suppress_rdkit_logs() -> None: + try: + from rdkit import RDLogger + RDLogger.DisableLog('rdApp.*') + return + except Exception: + pass + try: + from rdkit import rdBase + rdBase.DisableLog('rdApp.*') + except Exception: + pass + + +def enable_rdkit_logs() -> None: + try: + from rdkit import RDLogger + RDLogger.EnableLog('rdApp.*') + return + except Exception: + pass + try: + from rdkit import rdBase + rdBase.EnableLog('rdApp.*') + except Exception: + pass + + +def mol_from_smiles_lenient(smiles: str) -> Optional[Any]: + from rdkit import Chem + if not smiles or not str(smiles).strip(): + return None + suppress_rdkit_logs() + try: + mol = Chem.MolFromSmiles(smiles) + if mol is not None: + return mol + mol = Chem.MolFromSmiles(smiles, sanitize=False) + return mol + finally: + enable_rdkit_logs() diff --git a/vlmeval/dataset/utils/scimif/unit_matching.py b/vlmeval/dataset/utils/scimif/unit_matching.py new file mode 100644 index 000000000..911f3a667 --- /dev/null +++ b/vlmeval/dataset/utils/scimif/unit_matching.py @@ -0,0 +1,43 @@ +import re +from typing import Any, Dict + + +def evaluate_unit_consistency(response: str, + item: Dict[str, Any], + instruction_name: str, + required_parameters: str = '', + **kwargs) -> Dict[str, Any]: + if not required_parameters or not response: + return {'passed': False, 'score': 0.0, 'detail': 'Missing required_parameters response is empty'} + units_text = required_parameters.strip() + for phrase in [ + 'specific chemical stoichiometric units', 'specific physical stoichiometric units', + 'specific geographical units', 'specific biological units', 'specific materials units' + ]: + units_text = re.sub(re.escape(phrase) + '\\s*[:：]?\\s*', '', units_text, flags=re.I) + units_text = units_text.strip() or required_parameters.strip() + unit_patterns = _extract_unit_patterns(units_text) + matched = False + for pattern in unit_patterns: + if re.search(pattern, response, re.IGNORECASE | re.DOTALL): + matched = True + break + if matched: + return {'passed': True, 'score': 1.0, 'detail': f'The output contains unit:{units_text}'} + return {'passed': False, 'score': 0.0, 'detail': f"The output does not contain unit '{units_text}'"} + + +def _extract_unit_patterns(units_text: str) -> list: + common_units = [ + 'mol/L', 'mol·L⁻¹', 'mol\\s*/\\s*L', 'mol\\s*·\\s*L', 'kg', 'g', 'mg', 'm/s', 'm/s²', 'km', 'm', 'cm', 'mm', + 'Pa', 'kPa', 'MPa', 'GPa', 'J', 'kJ', 'eV', '°C', 'K', '°F', 'cells/mL', 'cells/\\s*mL', '%', 'percent' + ] + patterns = [] + clean = units_text.strip() + if clean: + escaped = re.escape(clean) + patterns.append(escaped) + for u in common_units: + if re.search(u, units_text, re.I): + patterns.append(u) + return patterns if patterns else [re.escape(units_text)] diff --git a/vlmeval/dataset/utils/scimif_eval.py b/vlmeval/dataset/utils/scimif_eval.py new file mode 100644 index 000000000..646e2d3e0 --- /dev/null +++ b/vlmeval/dataset/utils/scimif_eval.py @@ -0,0 +1,271 @@ +import inspect +import json +from importlib import import_module +from typing import Any, Dict, Iterable, List, Optional + +SCIENCE_MAP = { + 'chemistry_unit_consistency': ('unit_matching', 'evaluate_unit_consistency'), + 'physics_unit_consistency': ('unit_matching', 'evaluate_unit_consistency'), + 'geography_unit_consistency': ('unit_matching', 'evaluate_unit_consistency'), + 'biology_unit_consistency': ('unit_matching', 'evaluate_unit_consistency'), + 'material_unit_consistency': ('unit_matching', 'evaluate_unit_consistency'), + 'chemistry_molecular_format_validity': ('chemistry_format_validation', 'evaluate_molecular_format'), + 'chemistry_entity_option_constraint': ('options_matching', 'evaluate_options_constraint'), + 'chemistry_atom_count_constraint': ('chemistry_count_atom_checking', 'evaluate_atom_count'), + 'chemistry_atom_bond_constraint': ('chemistry_count_bond_checking', 'evaluate_bond_count'), + 'chemistry_atom_group_constraint': ('chemistry_count_group_checking', 'evaluate_group_count'), + 'chemistry_method_constraint': ('analysis_method_checking', 'evaluate_method_constraint'), + 'chemistry_reaction_steps_constraint': ('analysis_step_checking', 'evaluate_analysis_steps'), + 'chemistry_analysis_steps_constraint': ('analysis_step_checking', 'evaluate_analysis_steps'), + 'physics_method_constraint': ('analysis_method_checking', 'evaluate_method_constraint'), + 'physics_analysis_steps_constraint': ('analysis_step_checking', 'evaluate_analysis_steps'), + 'geography_method_constraint': ('analysis_method_checking', 'evaluate_method_constraint'), + 'geography_analysis_steps_constraint': ('analysis_step_checking', 'evaluate_analysis_steps'), + 'biology_method_constraint': ('analysis_method_checking', 'evaluate_method_constraint'), + 'biology_analysis_steps_constraint': ('analysis_step_checking', 'evaluate_analysis_steps'), + 'material_method_constraint': ('analysis_method_checking', 'evaluate_method_constraint'), + 'material_analysis_steps_constraint': ('analysis_step_checking', 'evaluate_analysis_steps'), + 'geography_address_format_validity': ('geography_format_geocoding_validation', 'evaluate_geography_address'), + 'geography_scene_option_constraint': ('options_matching', 'evaluate_options_constraint'), + 'biology_entity_relationship_format_validity': + ('life_format_entity_relationship_validation', 'evaluate_entity_relationship'), + 'biology_sequence_length_constraint': ('life_sequence_length_checking', 'evaluate_sequence_length'), + 'material_characterization_technique_format_constraint': + ('materials_format_characterization_technique_validation', 'evaluate_characterization_technique'), + 'material_property_prediction_constraint': + ('materials_property_prediction_checking', 'evaluate_property_prediction'), +} + +LEGACY_SCIENCE_ALIASES = { + 'life_unit_consistency': + SCIENCE_MAP['biology_unit_consistency'], + 'life_method_constraint': + SCIENCE_MAP['biology_method_constraint'], + 'life_analysis_steps_constraint': + SCIENCE_MAP['biology_analysis_steps_constraint'], + 'life_entity_relationship_format_validity': (SCIENCE_MAP['biology_entity_relationship_format_validity']), + 'life_sequence_length_constraint': + SCIENCE_MAP['biology_sequence_length_constraint'], + 'materials_unit_consistency': + SCIENCE_MAP['material_unit_consistency'], + 'materials_method_constraint': + SCIENCE_MAP['material_method_constraint'], + 'materials_analysis_steps_constraint': + SCIENCE_MAP['material_analysis_steps_constraint'], + 'materials_characterization_technique_format_constraint': + (SCIENCE_MAP['material_characterization_technique_format_constraint']), + 'materials_property_prediction_constraint': (SCIENCE_MAP['material_property_prediction_constraint']), +} + +GENERAL_MAP = { + 'general_decimal_annotation': ('general_checking', 'check_decimal_format'), + 'general_scientific_annotation': ('general_checking', 'check_scientific_format'), + 'general_wrap_up': ('general_checking', 'check_wrap_up'), + 'general_all_uppercase': ('general_checking', 'check_uppercase'), + 'general_all_lowercase': ('general_checking', 'check_lowercase'), + 'general_json_constraint': ('general_checking', 'check_json_format'), + 'general_list_constraint': ('general_checking', 'check_list_format'), + 'general_tuple_constraint': ('general_checking', 'check_tuple_format'), + 'general_dictionary_constraint': ('general_checking', 'check_dictionary_format'), + 'general_markdown_constraint': ('general_checking', 'check_markdown_format'), + 'general_html_constraint': ('general_checking', 'check_html_format'), + 'general_xml_constraint': ('general_checking', 'check_xml_format'), + 'general_csv_constraint': ('general_checking', 'check_csv_format'), + 'general_choose_from': ('general_checking', 'check_choose_from'), + 'general_judge': ('general_checking', 'check_judge'), + 'general_number_response': ('general_checking', 'check_number_response'), + 'general_response_structure': ('general_checking', 'check_response_structure'), +} + +INSTRUCTION_EVALUATOR_MAP = { + **SCIENCE_MAP, + **LEGACY_SCIENCE_ALIASES, + **GENERAL_MAP, +} + +LLM_EXTRACTION_INSTRUCTIONS = { + 'chemistry_molecular_format_validity', + 'chemistry_atom_count_constraint', + 'chemistry_atom_bond_constraint', + 'chemistry_atom_group_constraint', + 'chemistry_method_constraint', + 'physics_method_constraint', + 'geography_method_constraint', + 'biology_method_constraint', + 'material_method_constraint', + 'life_method_constraint', + 'materials_method_constraint', +} + + +def parse_instruction_list(value: Any) -> List[Dict[str, Any]]: + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + if value is None: + return [] + if isinstance(value, str): + text = value.strip() + if not text: + return [] + try: + parsed = json.loads(text) + except json.JSONDecodeError as exc: + raise ValueError('SciMIF instruction_list must contain valid JSON.') from exc + if not isinstance(parsed, list): + raise ValueError('SciMIF instruction_list must decode to a list.') + return [item for item in parsed if isinstance(item, dict)] + raise TypeError(f'Unsupported SciMIF instruction_list type: {type(value)!r}') + + +def get_evaluator(instruction_name: str): + target = INSTRUCTION_EVALUATOR_MAP.get(instruction_name) + if target is None: + return None + module_name, function_name = target + module = import_module(f'.scimif.{module_name}', package=__package__) + return getattr(module, function_name) + + +def evaluate_single_instruction(response: str, + item: Dict[str, Any], + instruction: Dict[str, Any], + llm_client=None, + judge_model: Optional[str] = None) -> Dict[str, Any]: + instruction_name = instruction.get('instruction_name', '') + evaluator = get_evaluator(instruction_name) + if evaluator is None: + return { + 'score': 0.0, + 'detail': f'Unsupported instruction: {instruction_name}', + 'skipped': True, + } + + call_kwargs = { + 'required_parameters': instruction.get('required_parameters') or '', + 'instruction_description': '', + 'edit_question': item.get('edit_question', ''), + 'reference_answer': item.get('answer', ''), + 'item': item, + 'instruction_name': instruction_name, + } + if ('analysis_step' in instruction_name or 'reaction_steps' in instruction_name + or instruction_name == 'general_number_response' or instruction_name in LLM_EXTRACTION_INSTRUCTIONS): + call_kwargs['llm_client'] = llm_client + call_kwargs['judge_model'] = judge_model + + try: + signature = inspect.signature(evaluator) + accepts_varkw = any(parameter.kind == inspect.Parameter.VAR_KEYWORD + for parameter in signature.parameters.values()) + if accepts_varkw: + filtered_kwargs = call_kwargs + else: + filtered_kwargs = {key: value for key, value in call_kwargs.items() if key in signature.parameters} + result = evaluator(response, **filtered_kwargs) + if not isinstance(result, dict): + return { + 'score': 0.0, + 'detail': f'Unexpected evaluator result type: {type(result)!r}', + 'skipped': False, + } + try: + score = float(result.get('score', 0.0)) + except (TypeError, ValueError): + score = 0.0 + return { + **result, + 'score': min(1.0, max(0.0, score)), + 'detail': str(result.get('detail', '')), + 'skipped': bool(result.get('skipped', False)), + } + except Exception as exc: + return { + 'score': 0.0, + 'detail': f'Evaluator error: {exc}', + 'skipped': False, + } + + +def evaluate_record(item: Dict[str, Any], llm_client=None, judge_model: Optional[str] = None) -> Dict[str, Any]: + response_value = item.get('prediction', item.get('response', '')) + response = '' if response_value is None else str(response_value) + instructions = parse_instruction_list(item.get('instruction_list')) + results = [] + + for instruction in instructions: + evaluation = evaluate_single_instruction( + response=response, + item=item, + instruction=instruction, + llm_client=llm_client, + judge_model=judge_model, + ) + result = { + 'instruction_name': instruction.get('instruction_name', ''), + 'source': instruction.get('source', ''), + 'required_parameters': instruction.get('required_parameters') or '', + 'score': evaluation['score'], + 'detail': evaluation['detail'], + 'skipped': evaluation['skipped'], + } + if 'casing_scope' in evaluation: + result['casing_scope'] = evaluation['casing_scope'] + results.append(result) + + evaluated = [result for result in results if not result['skipped']] + instruction_score = (sum(result['score'] for result in evaluated) / len(evaluated) if evaluated else 0.0) + strict_score = float(bool(evaluated) and all(result['score'] >= 1.0 for result in evaluated)) + + return { + 'instruction_results': results, + 'instruction_score': instruction_score, + 'strict_score': strict_score, + 'evaluated_instructions': len(evaluated), + 'skipped_instructions': len(results) - len(evaluated), + } + + +def summarize_results(records: Iterable[Dict[str, Any]], + subjects: Optional[Iterable[str]] = None) -> List[Dict[str, Any]]: + record_list = list(records) + if subjects is None: + subject_names = sorted({str(record.get('subject', '')) for record in record_list}) + else: + subject_names = list(subjects) + + groups = [('overall', record_list)] + groups.extend(( + subject, + [record for record in record_list if str(record.get('subject', '')) == subject], + ) for subject in subject_names if subject) + + summary = [] + for group_name, group_records in groups: + instruction_results = [ + result for record in group_records for result in record.get('instruction_results', []) + if not result.get('skipped', False) + ] + instruction_score_sum = sum(float(result.get('score', 0.0)) for result in instruction_results) + instruction_accuracy = instruction_score_sum / len(instruction_results) if instruction_results else 0.0 + sample_score_sum = sum(float(record.get('instruction_score', 0.0)) for record in group_records) + sample_accuracy = sample_score_sum / len(group_records) if group_records else 0.0 + strict_score_sum = sum(float(record.get('strict_score', 0.0)) for record in group_records) + strict_accuracy = strict_score_sum / len(group_records) if group_records else 0.0 + + source_accuracy = {} + for source in ('original', 'core_task', 'added_general'): + source_results = [result for result in instruction_results if result.get('source', '') == source] + source_score_sum = sum(float(result.get('score', 0.0)) for result in source_results) + source_accuracy[f'{source}_accuracy'] = source_score_sum / len(source_results) if source_results else 0.0 + + summary.append({ + 'split': group_name, + 'samples': len(group_records), + 'instructions': len(instruction_results), + 'skipped': sum(int(record.get('skipped_instructions', 0)) for record in group_records), + 'instruction_accuracy': instruction_accuracy, + 'sample_accuracy': sample_accuracy, + 'strict_accuracy': strict_accuracy, + **source_accuracy, + }) + return summary