import json import time import uuid from typing import Any from backend.config import get_generator_config, get_llm_config from backend.difficulty_validator import validate_hard_sample from backend.extract_schema import ( build_extract_template, format_extract_field_lines, normalize_extract_result, ) from backend.llm_client import call_llm CLASSIFY_CATEGORIES = [ "公司公告", "市场行情", "宏观经济", "投资建议", "风险提示", "财报解读", "政策监管", "其他", ] GENERATED_SAMPLES: dict[str, dict[str, Any]] = {} DIFFICULTY_LABELS = { "easy": "简单", "medium": "中等", "hard": "困难", } HARD_CLASSIFY_EXAMPLE = """ 参考困难样本（表面像财报解读，标准答案是风险提示）： { "text": "公司在业绩说明会上表示，一季度营收同比增 18%、海外订单恢复较快，但管理层同时强调，汇率波动与原材料涨价仍可能拖累下半年毛利率，并已提请投资者关注相关不确定性。", "expected_category": "风险提示" } """ HARD_EXTRACT_EXAMPLE = """ 参考困难样本（主体用指代、含条件推测、7 字段中 6 项有值）： { "text": "华东某龙头光伏企业在上游硅料反弹背景下，虽然组件出货量创同期新高，但管理层在电话会中提示，二季度单瓦毛利较上年同期可能收窄 3 至 5 个百分点，并建议投资者审慎看待后续抢装退潮带来的价格竞争。", "expected_result": { "主体机构": [], "时间周期": ["二季度", "较上年同期"], "金额数值": [], "财务指标": ["单瓦毛利可能收窄3至5个百分点", "出货量创同期新高"], "业务事件": ["电话会", "硅料反弹", "抢装退潮"], "风险因素": ["上游硅料反弹", "价格竞争", "毛利收窄"], "行业板块": ["光伏"] } } """ HARD_MATCH_EXAMPLE = """ 参考困难不匹配样本（同公司、同主题，但事件本质不同，模型易误判为匹配）： { "text_a": "上市公司公告称，公司拟以集中竞价方式回购公司股份，回购资金总额不超过 2 亿元，回购股份将用于员工持股计划，实施期限自股东大会审议通过之日起 12 个月内完成。", "text_b": "同一家公司披露，控股股东拟通过大宗交易减持公司股份，减持比例不超过总股本的 1%，已预先披露减持计划，自公告之日起 6 个月内实施完毕。", "expected_is_match": false } """ def _parse_json_object(raw_output: str) -> dict[str, Any]: text = raw_output.strip() if text.startswith("```"): first_newline = text.find("\n") if first_newline != -1: text = text[first_newline + 1 :] if text.rstrip().endswith("```"): text = text.rstrip()[:-3].rstrip() data = json.loads(text) if not isinstance(data, dict): raise ValueError("模型返回的不是 JSON 对象") return data def _text_length_rule(task: str, difficulty: str) -> str: if task == "classify": if difficulty == "easy": return "text 长度 25-60 字" if difficulty == "medium": return "text 长度 45-90 字" return "text 长度 80-160 字，必须含转折与多重信息" if task == "extract": if difficulty == "easy": return "text 长度 30-70 字" if difficulty == "medium": return "text 长度 50-100 字" return "text 长度 80-150 字，必须多层嵌套" if difficulty == "easy": return "text_a / text_b 各 15-45 字" if difficulty == "medium": return "text_a / text_b 各 30-70 字" return "text_a / text_b 各 45-100 字，必须高度相关但难判断" def _difficulty_rules(task: str, difficulty: str) -> str: if task == "classify": if difficulty == "easy": return """ - 文本只表达一种类别特征，类别信号非常明确 - 读者几乎不需要推理 - 禁止混合多个类别特征 """ if difficulty == "medium": return """ - 文本包含 2 个以上信息点，但主类别仍可唯一判断 - 需要读完整句才能确定，不能只看关键词 - 不允许出现两个类别都同样合理的歧义 """ return """ - 必须制造真实分类歧义：至少 2 个类别在表面都“说得通” - 标准答案唯一，只能依靠 subtle 线索区分（文本功能 > 关键词） - 必须使用转折结构：但 / 然而 / 同时 / 尽管 / 虽然…却 - 推荐组合：财报解读 + 风险提示、公司公告 + 政策监管、市场行情 + 投资建议 - 禁止短句、禁止标题式新闻、禁止类别信号只偏向一个类别 - 要让初级标注员第一次读也拿不准，资深标注员能依据“核心意图”定案 """ + HARD_CLASSIFY_EXAMPLE if task == "extract": if difficulty == "easy": return """ - 公司名、日期、金额、指标等关键信息都明确出现 - 字段可以直接抄写，不需要推理 """ if difficulty == "medium": return """ - 至少 4 类字段有值，部分信息需要轻微改写后抽取 - 允许出现代词，但 expected_result 必须写真实实体 """ return """ - 文本至少 80 字，expected_result 至少 5 个字段非空 - 主体机构禁止在 text 中完整直述（用“某龙头”“该公司”“头部酒企”等指代） - 必须含条件/推测：若、如果、可能、或将在、预计可能 - 金额与比例可分散在从句中，需理解后归一化写入字段 - 至少 2 个字段容易漏抽（如隐含行业、嵌套风险、间接时间） - expected_result 必须完整准确，字段名必须使用中文 """ + HARD_EXTRACT_EXAMPLE if difficulty == "easy": return """ - 两段文本表达同一事件，只是同义改写 - 是否匹配应非常明显 """ if difficulty == "medium": return """ - 核心事件相同，但表述更概括或更具体 - 需要理解语义，不能只看关键词重叠 """ return """ - 每段至少 45 字，必须难到让模型容易犯错 - expected_is_match=false：同公司/同主题/高词重叠，但核心事件或关键数字不同 - expected_is_match=true：表述路径完全不同，但指向同一事实（不同语序、不同信息顺序） - 禁止仅靠同义词替换；false 样本要像“回购 vs 减持”，true 样本要像“发行成功 vs 簿记完成” """ + HARD_MATCH_EXAMPLE def _build_classify_prompt(difficulty: str) -> str: categories = "、".join(CLASSIFY_CATEGORIES) return f"""你是金融文本分类测试数据生成器。 请生成 1 条 {DIFFICULTY_LABELS[difficulty]} 难度的中文金融文本测试样本。 可选类别（expected_category 必须从中选择）： {categories} 难度要求： {_difficulty_rules("classify", difficulty)} 输出 JSON 对象，字段： - text：{_text_length_rule("classify", difficulty)} - expected_category：标准类别 只返回 JSON，不要输出解释。""" def _build_extract_prompt(difficulty: str) -> str: field_lines = format_extract_field_lines() empty_template = json.dumps(build_extract_template(), ensure_ascii=False, indent=2) return f"""你是金融文本信息抽取测试数据生成器。 请生成 1 条 {DIFFICULTY_LABELS[difficulty]} 难度的中文金融文本测试样本，用于公告解读、财报分析等场景。 需要支持的抽取字段（字段名必须使用中文）： {field_lines} 难度要求： {_difficulty_rules("extract", difficulty)} 输出 JSON 对象，字段： - text：{_text_length_rule("extract", difficulty)} - expected_result：标准答案 JSON，结构如下： {empty_template} 只返回 JSON，不要输出解释。""" def _build_match_prompt(difficulty: str) -> str: return f"""你是金融文本匹配测试数据生成器。 请生成 1 条 {DIFFICULTY_LABELS[difficulty]} 难度的中文金融文本匹配样本。 难度要求： {_difficulty_rules("match", difficulty)} 输出 JSON 对象，字段： - text_a：{_text_length_rule("match", difficulty)} - text_b：{_text_length_rule("match", difficulty)} - expected_is_match：true 或 false 只返回 JSON，不要输出解释。""" def _build_generation_prompt(task: str, difficulty: str) -> str: if task == "classify": return _build_classify_prompt(difficulty) if task == "extract": return _build_extract_prompt(difficulty) if task == "match": return _build_match_prompt(difficulty) raise ValueError(f"未知任务：{task}") def _normalize_extract_result(result: Any) -> dict[str, list[Any]]: return normalize_extract_result(result) def _validate_generated_sample(task: str, difficulty: str, sample: dict[str, Any]) -> dict[str, Any]: if task == "classify": text = str(sample.get("text", "")).strip() category = str(sample.get("expected_category", "")).strip() min_len = 60 if difficulty == "hard" else 12 if len(text) < min_len: raise ValueError(f"生成的 text 过短（{difficulty} 至少 {min_len} 字）") if category not in CLASSIFY_CATEGORIES: raise ValueError(f"expected_category 不在允许类别中：{category}") payload = {"text": text, "expected_category": category} if difficulty == "hard": validate_hard_sample("classify", payload, expected_category=category) return payload if task == "extract": text = str(sample.get("text", "")).strip() min_len = 80 if difficulty == "hard" else 12 if len(text) < min_len: raise ValueError(f"生成的 text 过短（{difficulty} 至少 {min_len} 字）") expected_result = _normalize_extract_result(sample.get("expected_result")) if not any(expected_result.values()): raise ValueError("expected_result 不能全部为空") payload = {"text": text, "expected_result": expected_result} if difficulty == "hard": validate_hard_sample( "extract", payload, expected_result=expected_result, ) return payload if task == "match": text_a = str(sample.get("text_a", "")).strip() text_b = str(sample.get("text_b", "")).strip() min_len = 45 if difficulty == "hard" else 8 if len(text_a) < min_len or len(text_b) < min_len: raise ValueError(f"生成的 text_a 或 text_b 过短（{difficulty} 至少 {min_len} 字）") if sample.get("expected_is_match") not in {True, False}: raise ValueError("expected_is_match 必须是 true 或 false") payload = { "text_a": text_a, "text_b": text_b, "expected_is_match": sample["expected_is_match"], } if difficulty == "hard": validate_hard_sample( "match", payload, expected_is_match=sample["expected_is_match"], ) return payload raise ValueError(f"未知任务：{task}") def list_generated_samples( task: str, difficulty: str | None = None, ) -> list[dict[str, Any]]: prefix = f"gen_{task}_" samples = [ sample for sample in GENERATED_SAMPLES.values() if str(sample.get("id", "")).startswith(prefix) and (not difficulty or sample.get("difficulty") == difficulty) ] samples.sort(key=lambda item: item.get("generated_at", 0), reverse=True) return samples def get_generated_sample(sample_id: str) -> dict[str, Any] | None: return GENERATED_SAMPLES.get(sample_id) def _make_sample_id(task: str, difficulty: str) -> str: suffix = uuid.uuid4().hex[:8] return f"gen_{task}_{difficulty}_{suffix}" def _generation_params(difficulty: str, generator_config: dict[str, Any], llm_config: dict[str, Any]) -> dict[str, Any]: if difficulty == "hard": return { "temperature": float( generator_config.get( "hard_temperature", generator_config.get("temperature", 0.85), ) ), "max_retries": int(generator_config.get("hard_max_retries", 5)), "timeout_seconds": float( generator_config.get( "hard_timeout_seconds", generator_config.get("timeout_seconds", 90), ) ), } return { "temperature": float(generator_config.get("temperature", 0.85)), "max_retries": int(generator_config.get("max_retries", 3)), "timeout_seconds": float(generator_config.get("timeout_seconds", 90)), } def generate_sample(task: str, difficulty: str) -> dict[str, Any]: if difficulty not in DIFFICULTY_LABELS: raise ValueError(f"未知难度：{difficulty}") prompt = _build_generation_prompt(task, difficulty) generator_config = get_generator_config() llm_config = get_llm_config() params = _generation_params(difficulty, generator_config, llm_config) last_error = "生成失败" max_retries = params["max_retries"] timeout_seconds = params["timeout_seconds"] for attempt in range(max_retries): try: raw_output = call_llm( prompt, model=generator_config.get("model", llm_config["model"]), temperature=params["temperature"], timeout=timeout_seconds, ) except Exception as exc: last_error = f"大模型调用失败：{exc}" if attempt + 1 >= max_retries: break prompt = ( _build_generation_prompt(task, difficulty) + f"\n\n上次调用失败（{timeout_seconds:.0f}s 超时或网络异常），请重新生成。错误：{last_error}" ) continue try: parsed = _parse_json_object(raw_output) payload = _validate_generated_sample(task, difficulty, parsed) sample = { "id": _make_sample_id(task, difficulty), "task": task, "difficulty": difficulty, "generated": True, "generated_at": int(time.time()), **payload, } GENERATED_SAMPLES[sample["id"]] = sample return sample except (ValueError, json.JSONDecodeError) as exc: last_error = str(exc) prompt = ( _build_generation_prompt(task, difficulty) + "\n\n上次生成不合格，请重新生成，并严格遵守 JSON 格式和难度要求。错误：" + last_error ) raise ValueError(f"测试数据生成失败：{last_error}")