diff --git a/examples/scripts/run_analyzer_standalone.py b/examples/scripts/run_analyzer_standalone.py index f93efed..6d98ad8 100644 --- a/examples/scripts/run_analyzer_standalone.py +++ b/examples/scripts/run_analyzer_standalone.py @@ -93,6 +93,12 @@ def _parse_args() -> argparse.Namespace: default=None, help="Number of failed samples per Analyzer model batch. Default: 20.", ) + parser.add_argument( + "--request-timeout-seconds", + type=float, + default=None, + help="Analyzer model client timeout. Default: 300 seconds.", + ) parser.add_argument("--resume", action="store_true", help="Resume using Configer/external runtime state.") parser.add_argument("--from-node", default=None, help="Force Analyzer to resume from a specific step.") parser.add_argument("--list-nodes", action="store_true", help="List standalone Analyzer steps and exit.") @@ -135,6 +141,7 @@ def main() -> None: version_id=args.version_id, baseline_result_path=args.baseline_result_path, analyze_batch_size=args.analyze_batch_size, + analyze_request_timeout_seconds=args.request_timeout_seconds, force_new_version=args.new_version, stream_stdout=args.stream_stdout, emit_status=False, diff --git a/loopai/skills/Analyzer/analyzer_agent.py b/loopai/skills/Analyzer/analyzer_agent.py index cd8550e..6ef834c 100644 --- a/loopai/skills/Analyzer/analyzer_agent.py +++ b/loopai/skills/Analyzer/analyzer_agent.py @@ -71,6 +71,9 @@ def check_required_fields(state: LoopAIState, runtime: Runtime[RuntimeContext]): has_result_path = ( bool(judger_cfg.get("output_result_path")) + or bool(judger_cfg.get("bench_result")) + or bool(judger_cfg.get("extra_bench_result")) + or bool(analyzer_cfg.get("eval_result_paths")) or bool(analyzer_cfg.get("eval_result_path")) ) if not has_result_path: @@ -82,6 +85,9 @@ def check_required_fields(state: LoopAIState, runtime: Runtime[RuntimeContext]): bool(judger_cfg.get("output_result_path")) or bool(judger_cfg.get("out_result_path")) or bool(judger_cfg.get("eval_result_path")) + or bool(judger_cfg.get("bench_result")) + or bool(judger_cfg.get("extra_bench_result")) + or bool(analyzer_cfg.get("eval_result_paths")) or bool(analyzer_cfg.get("eval_result_path")) ) if (not has_bench) and (not has_result_path): diff --git a/loopai/skills/Analyzer/bench_inputs.py b/loopai/skills/Analyzer/bench_inputs.py new file mode 100644 index 0000000..9a128be --- /dev/null +++ b/loopai/skills/Analyzer/bench_inputs.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List + + +def _source(name: Any, path: Any, task_type: Any = None) -> Dict[str, str]: + value = str(path or "").strip() + fallback_name = Path(value).stem if value else "bench" + return { + "bench_name": str(name or fallback_name), + "path": value, + "task_type": str(task_type or ""), + } + + +def _from_value(value: Any) -> List[Dict[str, str]]: + if isinstance(value, (str, Path)): + return [_source(None, value)] if str(value).strip() else [] + if isinstance(value, dict): + if any(key in value for key in ("output_result_path", "eval_result_path", "path")): + return [_source( + value.get("bench_name") or value.get("name"), + value.get("output_result_path") or value.get("eval_result_path") or value.get("path"), + value.get("task_type"), + )] + return [_source(name, path) for name, path in value.items() if path] + if isinstance(value, (list, tuple)): + sources: List[Dict[str, str]] = [] + for item in value: + sources.extend(_from_value(item)) + return sources + return [] + + +def resolve_eval_result_sources(state: Dict[str, Any], task_type: str) -> List[Dict[str, str]]: + """Resolve one or more Judger result files without breaking string input.""" + judger = state.get("judger") if isinstance(state.get("judger"), dict) else {} + analyzer = state.get("analyzer") if isinstance(state.get("analyzer"), dict) else {} + + sources: List[Dict[str, str]] = [] + for key in ("bench_result", "extra_bench_result"): + sources.extend(_from_value(judger.get(key))) + + compatible = [ + item for item in sources + if not item.get("task_type") or item.get("task_type") == task_type + ] + if sources: + sources = compatible + else: + sources = _from_value( + judger.get("output_result_paths") + or judger.get("output_result_path") + or analyzer.get("eval_result_paths") + or analyzer.get("eval_result_path") + ) + + deduplicated: List[Dict[str, str]] = [] + seen = set() + used_names = CounterLike() + for item in sources: + path = item.get("path", "") + if not path or path in seen: + continue + seen.add(path) + base_name = item.get("bench_name") or Path(path).stem + suffix = used_names.next(base_name) + item = dict(item) + item["bench_name"] = base_name if suffix == 1 else f"{base_name}-{suffix}" + deduplicated.append(item) + return deduplicated + + +class CounterLike: + def __init__(self) -> None: + self._counts: Dict[str, int] = {} + + def next(self, key: str) -> int: + self._counts[key] = self._counts.get(key, 0) + 1 + return self._counts[key] diff --git a/loopai/skills/Analyzer/bucket_strategy.py b/loopai/skills/Analyzer/bucket_strategy.py new file mode 100644 index 0000000..fea8b12 --- /dev/null +++ b/loopai/skills/Analyzer/bucket_strategy.py @@ -0,0 +1,763 @@ +from __future__ import annotations + +import math +import re +from collections import Counter, defaultdict +from typing import Any, Dict, Iterable, List, Tuple + + +_CODE_BUCKET_META = { + "code_output_contract": { + "label": "代码补全输出契约", + "severity": 1.20, + "transfer": 1.25, + "learnability_prior": 1.30, + "cost": 0.90, + "sample_direction": "函数签名/Docstring -> 仅输出完整可执行代码，不输出解释或 Markdown 围栏", + }, + "code_syntax_completion": { + "label": "Python 语法与补全完整性", + "severity": 1.15, + "transfer": 1.15, + "learnability_prior": 1.15, + "cost": 0.90, + "sample_direction": "缩进、括号、字符串、return 与完整函数体的短小 Python 补全样本", + }, + "code_interface_scope": { + "label": "函数接口、作用域与依赖", + "severity": 1.10, + "transfer": 1.10, + "learnability_prior": 1.00, + "cost": 1.00, + "sample_direction": "保持函数名和签名，补齐局部变量、标准库导入与辅助函数", + }, + "code_semantic_logic": { + "label": "语义逻辑与断言正确性", + "severity": 1.20, + "transfer": 1.15, + "learnability_prior": 0.75, + "cost": 1.15, + "sample_direction": "可执行但答案错误的短算法题，配套正常样例与反例断言", + }, + "code_boundary_robustness": { + "label": "边界条件与鲁棒性", + "severity": 1.05, + "transfer": 1.10, + "learnability_prior": 0.75, + "cost": 1.10, + "sample_direction": "空输入、单元素、重复值、极值和特殊字符等对比样本", + }, + "code_runtime_efficiency": { + "label": "运行时与效率", + "severity": 0.95, + "transfer": 0.85, + "learnability_prior": 0.55, + "cost": 1.30, + "sample_direction": "超时、递归深度和复杂度退化的成对优化样本", + }, +} + +_SQL_BUCKET_META = { + "sql_output_contract": { + "label": "SQL 输出契约", + "severity": 1.15, + "transfer": 1.20, + "learnability_prior": 1.25, + "cost": 0.90, + "sample_direction": "问题与 Schema -> 仅输出可执行 SQL，不输出解释或 Markdown", + }, + "sql_syntax": { + "label": "SQL 语法与结构", + "severity": 1.15, + "transfer": 1.10, + "learnability_prior": 1.10, + "cost": 0.90, + "sample_direction": "SELECT、JOIN、GROUP BY、子查询和聚合函数的短 SQL 修复样本", + }, + "sql_schema_linking": { + "label": "Schema Linking", + "severity": 1.20, + "transfer": 1.20, + "learnability_prior": 0.85, + "cost": 1.15, + "sample_direction": "问题实体与表、列、外键的显式对齐及易混淆 Schema 对比样本", + }, + "sql_semantic_logic": { + "label": "SQL 语义与结果正确性", + "severity": 1.20, + "transfer": 1.10, + "learnability_prior": 0.70, + "cost": 1.20, + "sample_direction": "可执行但结果错误的查询，覆盖过滤、聚合、排序与去重语义", + }, + "sql_type_value": { + "label": "类型、值与条件表达", + "severity": 1.00, + "transfer": 1.00, + "learnability_prior": 0.90, + "cost": 1.00, + "sample_direction": "日期、数值、NULL、字符串匹配和类型转换的边界样本", + }, + "sql_runtime_efficiency": { + "label": "SQL 运行时与效率", + "severity": 0.90, + "transfer": 0.80, + "learnability_prior": 0.50, + "cost": 1.30, + "sample_direction": "超时查询与等价高效查询的成对样本", + }, +} + +_GENERAL_BUCKET_META = { + "general_instruction_following": { + "label": "指令与输出格式遵循", + "severity": 1.20, + "transfer": 1.25, + "learnability_prior": 1.25, + "cost": 0.85, + "sample_direction": "覆盖格式、长度、结构和约束组合的指令遵循样本，答案需严格满足可验证要求", + }, + "general_relevance_intent": { + "label": "相关性与意图理解", + "severity": 1.10, + "transfer": 1.15, + "learnability_prior": 1.05, + "cost": 0.95, + "sample_direction": "相似意图辨析、答非所问纠正和用户目标对齐的对比样本", + }, + "general_factuality_grounding": { + "label": "事实性与知识依据", + "severity": 1.25, + "transfer": 1.15, + "learnability_prior": 0.75, + "cost": 1.20, + "sample_direction": "带可信来源或给定上下文的知识问答、幻觉纠正和事实核验样本", + }, + "general_reasoning_consistency": { + "label": "推理与一致性", + "severity": 1.20, + "transfer": 1.15, + "learnability_prior": 0.65, + "cost": 1.30, + "sample_direction": "多步推理、因果判断、前后一致性检查和反例验证样本", + }, + "general_completeness_coverage": { + "label": "完整性与要点覆盖", + "severity": 1.05, + "transfer": 1.10, + "learnability_prior": 1.00, + "cost": 0.95, + "sample_direction": "按评分要点覆盖关键信息、补全遗漏内容和避免回答截断的样本", + }, + "general_language_quality": { + "label": "表达与语言质量", + "severity": 0.90, + "transfer": 1.00, + "learnability_prior": 1.15, + "cost": 0.85, + "sample_direction": "流畅性、连贯性、简洁性、语法和结构化表达的改写对比样本", + }, + "general_safety_refusal": { + "label": "安全与拒答边界", + "severity": 1.30, + "transfer": 1.10, + "learnability_prior": 0.75, + "cost": 1.20, + "sample_direction": "合理拒答、不必要拒答和安全替代回答的边界对比样本", + }, +} + +_GENERAL_METHOD_REFERENCES = [ + { + "paper": "Holistic Evaluation of Language Models (HELM)", + "venue": "TMLR 2023", + "applied_to": "将通用文本质量拆成正确性、鲁棒性、安全性等可区分维度", + "url": "https://arxiv.org/abs/2211.09110", + }, + { + "paper": "Training Language Models to Follow Instructions with Human Feedback", + "venue": "NeurIPS 2022", + "applied_to": "把用户意图与指令遵循作为独立能力，而非普通内容错误", + "url": "https://proceedings.neurips.cc/paper_files/paper/2022/hash/b1efde53be364a73914f58805a001731-Abstract.html", + }, + { + "paper": "TruthfulQA: Measuring How Models Mimic Human Falsehoods", + "venue": "ACL 2022", + "applied_to": "把事实性和信息充分性从表面文本相似度中分离", + "url": "https://aclanthology.org/2022.acl-long.229/", + }, + { + "paper": "Skill-It! A Data-Driven Skills Framework for Understanding and Training Language Models", + "venue": "NeurIPS 2023", + "applied_to": "处理能力先后依赖，并为被前置失败遮蔽的能力保留探索预算", + "url": "https://proceedings.neurips.cc/paper_files/paper/2023/hash/70b8505ac79e3e131756f793cd80eb8d-Abstract-Conference.html", + }, + { + "paper": "DoReMi: Optimizing Data Mixtures Speeds Up Language Model Pretraining", + "venue": "NeurIPS 2023", + "applied_to": "不把观察频率直接等同于固定数据混合比例", + "url": "https://proceedings.neurips.cc/paper_files/paper/2023/hash/dcba6be91359358c2355cd920da3fcbd-Abstract-Conference.html", + }, + { + "paper": "LESS: Selecting Influential Data for Targeted Instruction Tuning", + "venue": "ICML 2024", + "applied_to": "后续以小规模试训的目标能力收益替代静态学习效率先验", + "url": "https://proceedings.mlr.press/v235/xia24c.html", + }, +] + +_UNKNOWN_BUCKET = "diagnostic_unknown" +_UNKNOWN_META = { + "label": "待诊断样本", + "severity": 0.0, + "transfer": 0.0, + "learnability_prior": 0.0, + "cost": 1.0, + "sample_direction": "补充执行日志或人工复核，不直接进入训练分桶", +} + +_PROSE_PREFIX_RE = re.compile( + r"^(?:to solve|let(?:'s| us)|here(?:'s| is| are)|the function|this function|" + r"we need|in this|first[, ]|sure[, ]|certainly[, ]|below is|based on|" + r"the provided|the task|the problem|an? (?:simple )?approach)", + re.IGNORECASE, +) + + +def _text(value: Any) -> str: + if value is None: + return "" + if isinstance(value, (dict, list, tuple)): + return str(value) + return str(value) + + +def _record_text(record: Dict[str, Any]) -> Tuple[str, str, str, str]: + judge = record.get("judge") if isinstance(record.get("judge"), dict) else {} + completion = _text( + record.get("completion") + or record.get("prediction") + or record.get("pred") + or record.get("generated_text") + or record.get("generated_ans") + or record.get("response") + or record.get("output") + or record.get("model_output") + ).strip() + result = " ".join( + _text(record.get(key)) + for key in ("result", "stdout", "stderr", "execution_result", "error") + if record.get(key) is not None + ).strip() + tags = " ".join(_text(tag) for tag in (judge.get("tags") or [])) + judge_text = " ".join( + _text(judge.get(key)) + for key in ("stage", "reason", "exception_type", "advice") + if judge.get(key) is not None + ) + return completion, result, tags, judge_text + + +def _normalize_task_type(task_type: Any) -> str: + normalized = str(task_type or "code").strip().lower().replace("-", "_") + if normalized in {"text2sql", "text_to_sql", "sql"}: + return "text2sql" + if normalized in {"code", "coding", "programming", "python"}: + return "code" + return "general" + + +def _first_text(record: Dict[str, Any], keys: Tuple[str, ...]) -> str: + for key in keys: + value = record.get(key) + if value not in (None, "", [], {}): + return _text(value).strip() + return "" + + +def _general_evidence(record: Dict[str, Any]) -> Tuple[str, str, str, str]: + judge = record.get("judge") if isinstance(record.get("judge"), dict) else {} + detail = record.get("metric_detail") if isinstance(record.get("metric_detail"), dict) else {} + question = _first_text(record, ("question", "prompt", "input", "query", "instruction")) + reference = _first_text(record, ( + "target", "reference", "ground_truth", "answer", "correct_answer", "gold", "solution", + )) + completion, result, tags, judge_text = _record_text(record) + top_level = " ".join( + _text(record.get(key)) + for key in ( + "error_type", "errors", "reason", "feedback", "critique", "category", + "match_type", "primary_metric", "failure_type", + ) + if record.get(key) not in (None, "", [], {}) + ) + detail_text = " ".join( + _text(detail.get(key)) + for key in ("match_type", "error_type", "reason", "feedback", "label", "category", "analysis") + if detail.get(key) not in (None, "", [], {}) + ) + judge_extra = " ".join( + _text(judge.get(key)) + for key in ( + "error_type", "primary_error", "secondary_error", "errors", "labels", + "dimensions", "feedback", "critique", + ) + if judge.get(key) not in (None, "", [], {}) + ) + evidence = " ".join((result, tags, judge_text, judge_extra, top_level, detail_text)).strip().lower() + return question, reference, completion, evidence + + +def _contains_any(text: str, tokens: Tuple[str, ...]) -> bool: + return any(token in text for token in tokens) + + +def _looks_like_refusal(completion: str) -> bool: + return bool(re.match( + r"^(?:i(?:'m| am) sorry|sorry|i cannot|i can't|i am unable|as an ai|" + r"抱歉|对不起|我不能|我无法|无法帮助|不能协助)", + completion.lstrip(), + re.IGNORECASE, + )) + + +def _violates_general_output_contract(question: str, reference: str, completion: str) -> bool: + prompt = question.lower() + stripped = completion.strip() + reference_stripped = reference.strip() + expects_json = ( + "json" in prompt + or "json 对象" in question + or reference_stripped.startswith(("{", "[")) + ) + if expects_json and stripped: + candidate = stripped + if candidate.startswith("```"): + candidate = re.sub(r"^```(?:json)?\s*|\s*```$", "", candidate, flags=re.IGNORECASE) + try: + import json + + json.loads(candidate) + except (TypeError, ValueError): + return True + if _contains_any(prompt, ("only answer", "answer only", "only output", "respond only")): + if "\n" in stripped or stripped.startswith(("Here", "Sure", "The answer")): + return True + if _contains_any(question, ("只回答", "仅回答", "只输出", "仅输出")): + if "\n" in stripped or stripped.startswith(("答案是", "下面", "当然")): + return True + if _contains_any(prompt, ("no markdown", "without markdown")) and "```" in stripped: + return True + if _contains_any(question, ("不要使用 Markdown", "不使用 Markdown")) and "```" in stripped: + return True + return False + + +def _looks_like_prose_or_wrapped_code(completion: str, *, sql: bool = False) -> bool: + stripped = completion.lstrip() + if not stripped: + return False + lowered = stripped.lower() + if lowered.startswith("```") or _PROSE_PREFIX_RE.search(stripped): + return True + if any(marker in stripped[:240] for marker in ("### ", "**", "---")): + return True + if sql: + return not re.match(r"^(?:select|with|insert|update|delete)\b", lowered) + return not re.match(r"^(?:async\s+def|def|class|from|import|@|#|[A-Za-z_]\w*\s*=|return)\b", stripped) + + +def classify_failure_bucket( + record: Dict[str, Any], + task_type: str = "code", +) -> Dict[str, Any]: + """Map one failed record to an actionable capability bucket. + + Runtime/parser evidence is intentionally preferred over the model's + fallback ``other`` label. The original label remains in the result for + auditability. + """ + judge = record.get("judge") if isinstance(record.get("judge"), dict) else {} + metric_detail = record.get("metric_detail") if isinstance(record.get("metric_detail"), dict) else {} + original_stage = _text( + judge.get("stage") + or record.get("error_type") + or record.get("match_type") + or metric_detail.get("match_type") + or "other" + ).lower() + completion, result, tags, judge_text = _record_text(record) + evidence = f"{result} {tags} {judge_text}".lower() + failed = not bool(record.get("passed", False)) + task_route = _normalize_task_type(task_type) + + if task_route == "general": + question, reference, completion, evidence = _general_evidence(record) + if failed and not completion.strip(): + return _classification( + "general_completeness_coverage", original_stage, 0.98, "回答为空，无法覆盖任务要求" + ) + if _contains_any(evidence, ( + "over-refusal", "overrefusal", "unnecessary refusal", "should not refuse", + "unsafe", "safety", "harmful", "toxicity", "toxic", "jailbreak", "bias", + "过度拒答", "不应拒答", "安全", "有害", "毒性", "偏见", + )): + return _classification( + "general_safety_refusal", original_stage, 0.92, "结构化评测证据指向安全或拒答边界问题" + ) + if _contains_any(evidence, ( + "instruction_following", "instruction following", "instruction", "format_error", "format error", "format", + "output format", "constraint violation", "did not follow", "length constraint", + "schema violation", "指令遵循", "未遵循", "格式错误", "输出格式", "违反约束", + )) or _violates_general_output_contract(question, reference, completion): + return _classification( + "general_instruction_following", original_stage, 0.92, "输出违反明确指令、格式或结构约束" + ) + if _contains_any(evidence, ( + "irrelevant", "off-topic", "off topic", "intent mismatch", "answer relevance", "relevance", "intent", + "does not answer", "wrong task", "答非所问", "无关", "偏题", "意图错误", "相关性", + )): + return _classification( + "general_relevance_intent", original_stage, 0.88, "评测证据指向答复相关性或意图理解错误" + ) + if _contains_any(evidence, ( + "hallucination", "factual", "factually incorrect", "unsupported", "fabricated", + "misinformation", "faithfulness", "groundedness", "unverified claim", + "事实错误", "幻觉", "虚构", "无依据", "知识错误", "事实性", "忠实度", + )): + return _classification( + "general_factuality_grounding", original_stage, 0.90, "评测证据指向事实性、幻觉或依据不足" + ) + if _contains_any(evidence, ( + "reasoning", "logical error", "invalid inference", "inconsistent", "contradiction", + "causal error", "calculation error", "推理错误", "逻辑错误", "前后矛盾", "因果错误", "计算错误", + )): + return _classification( + "general_reasoning_consistency", original_stage, 0.86, "评测证据指向推理链或前后一致性错误" + ) + if _contains_any(evidence, ( + "incomplete", "completeness", "missing key", "omission", "coverage", "insufficient", "partial answer", + "truncated", "not comprehensive", "不完整", "遗漏", "缺少要点", "覆盖不足", "回答截断", + )): + return _classification( + "general_completeness_coverage", original_stage, 0.88, "评测证据指向回答不完整或要点遗漏" + ) + if _contains_any(evidence, ( + "language_quality", "language quality", "fluency", "grammar", "style", "coherence", "readability", "verbosity", "verbose", + "repetition", "ambiguous", "语言质量", "语法问题", "表达", "不流畅", "啰嗦", "重复", "歧义", "连贯性", + )): + return _classification( + "general_language_quality", original_stage, 0.84, "评测证据指向语言、风格或表达质量问题" + ) + if failed and completion and _looks_like_refusal(completion): + return _classification( + "general_safety_refusal", original_stage, 0.76, "失败回答表现为拒答，需要复核是否属于过度拒答" + ) + if failed and reference and len(reference) >= 160 and len(completion) < max(24, int(len(reference) * 0.15)): + return _classification( + "general_completeness_coverage", original_stage, 0.62, "回答显著短于参考内容，疑似关键要点覆盖不足" + ) + return _classification(_UNKNOWN_BUCKET, original_stage, 0.30, "通用文本缺少可靠的结构化归因证据，进入诊断池") + + if task_route == "text2sql": + if failed and completion and _looks_like_prose_or_wrapped_code(completion, sql=True): + return _classification("sql_output_contract", original_stage, 0.96, "输出不是可直接执行的 SQL") + if any(token in evidence for token in ("no such table", "no such column", "unknown column", "schema", "foreign key", "sql_schema")): + return _classification("sql_schema_linking", original_stage, 0.94, "执行证据指向表、列或 Schema 对齐错误") + if any(token in evidence for token in ("syntax error", "parse error", "sql_syntax", "near \"")): + return _classification("sql_syntax", original_stage, 0.94, "SQL 解析或语法错误") + if any(token in evidence for token in ("timeout", "timed out", "too many", "sql_perf", "sql_timeout")): + return _classification("sql_runtime_efficiency", original_stage, 0.90, "查询超时或效率问题") + if any(token in evidence for token in ("datatype", "type mismatch", "null", "conversion", "sql_type")): + return _classification("sql_type_value", original_stage, 0.86, "类型、NULL 或值条件错误") + if any(token in evidence for token in ("wrong answer", "assert", "mismatch", "value", "sql_result")): + return _classification("sql_semantic_logic", original_stage, 0.84, "SQL 可执行但结果或语义不正确") + if original_stage in {"sql_schema"}: + return _classification("sql_schema_linking", original_stage, 0.78, "沿用细粒度 SQL stage") + if original_stage in {"sql_syntax", "syntax"}: + return _classification("sql_syntax", original_stage, 0.78, "沿用 SQL 语法 stage") + if original_stage in {"sql_type"}: + return _classification("sql_type_value", original_stage, 0.76, "沿用 SQL 类型 stage") + if original_stage in {"sql_timeout", "sql_perf", "timeout", "perf"}: + return _classification("sql_runtime_efficiency", original_stage, 0.76, "沿用 SQL 运行时 stage") + return _classification(_UNKNOWN_BUCKET, original_stage, 0.30, "现有证据不足，进入诊断池") + + if failed and completion and _looks_like_prose_or_wrapped_code(completion): + return _classification("code_output_contract", original_stage, 0.97, "代码任务输出了说明文字或 Markdown 包装") + if any(token in evidence for token in ( + "syntaxerror", "syntax error", "invalid syntax", "unterminated", "unexpected eof", + "indentationerror", "indentation error", "was never closed", "truncated", + )): + return _classification("code_syntax_completion", original_stage, 0.95, "执行器报告语法错误或补全不完整") + if any(token in evidence for token in ( + "nameerror", "not defined", "importerror", "modulenotfound", "missing function", + "entry point", "wrong signature", "argument", "scope", + )): + return _classification("code_interface_scope", original_stage, 0.92, "函数接口、名称、作用域或依赖不完整") + if any(token in evidence for token in ( + "timeout", "timed out", "memoryerror", "recursionerror", "recursion", "performance", "perf", + )): + return _classification("code_runtime_efficiency", original_stage, 0.90, "运行超时、内存或递归问题") + if any(token in evidence for token in ("edge case", "boundary", "empty input", "corner case")): + return _classification("code_boundary_robustness", original_stage, 0.82, "证据明确指向边界条件") + if any(token in evidence for token in ( + "assertionerror", "assert", "wrong answer", "expected", "actual", "valueerror", "value", + )): + return _classification("code_semantic_logic", original_stage, 0.84, "代码进入执行但结果或断言不正确") + if original_stage in {"syntax", "import", "compile"}: + return _classification("code_syntax_completion", original_stage, 0.76, "沿用代码语法 stage") + if original_stage in {"assert", "value", "logic", "type"}: + return _classification("code_semantic_logic", original_stage, 0.72, "沿用代码语义 stage") + if original_stage in {"timeout", "perf", "recursion"}: + return _classification("code_runtime_efficiency", original_stage, 0.76, "沿用代码运行时 stage") + return _classification(_UNKNOWN_BUCKET, original_stage, 0.30, "现有证据不足，进入诊断池") + + +def _classification(bucket: str, original_stage: str, confidence: float, reason: str) -> Dict[str, Any]: + meta = ( + _CODE_BUCKET_META.get(bucket) + or _SQL_BUCKET_META.get(bucket) + or _GENERAL_BUCKET_META.get(bucket) + or _UNKNOWN_META + ) + return { + "bucket": bucket, + "label": meta["label"], + "confidence": round(float(confidence), 4), + "reason": reason, + "original_stage": original_stage, + "reclassified_from_other": original_stage == "other" and bucket != _UNKNOWN_BUCKET, + } + + +def _project_with_floor_and_cap( + weights: Dict[str, float], + floor: float, + cap: float, +) -> Dict[str, float]: + keys = [key for key, value in weights.items() if value > 0] + if not keys: + return {} + floor = max(0.0, min(float(floor), 1.0 / len(keys))) + cap = max(1.0 / len(keys), min(1.0, float(cap))) + fixed: Dict[str, float] = {} + free = set(keys) + + for _ in range(len(keys) * 2 + 2): + remaining = max(0.0, 1.0 - sum(fixed.values())) + denominator = sum(weights[key] for key in free) + proposed = { + key: (remaining * weights[key] / denominator if denominator else remaining / max(len(free), 1)) + for key in free + } + violations = { + key: floor if value < floor else cap + for key, value in proposed.items() + if value < floor or value > cap + } + if not violations: + fixed.update(proposed) + break + key = max(violations, key=lambda item: abs(proposed[item] - violations[item])) + fixed[key] = violations[key] + free.remove(key) + if not free: + break + + total = sum(fixed.values()) + if total and not math.isclose(total, 1.0): + adjustable = max(fixed, key=fixed.get) + fixed[adjustable] += 1.0 - total + return {key: max(0.0, round(value, 6)) for key, value in fixed.items()} + + +def build_training_bucket_strategy( + records: Iterable[Dict[str, Any]], + task_type: str = "code", + *, + alpha: float = 1.0, + min_share: float = 0.05, + max_share: float = 0.50, +) -> Dict[str, Any]: + """Build an actionable, confidence-aware data allocation plan. + + ``learnability_prior`` is deliberately marked as a prior. A later + training round should replace it with measured metric gain per sample. + """ + task_route = _normalize_task_type(task_type) + failed = [record for record in records if isinstance(record, dict) and not record.get("passed", False)] + classifications = [classify_failure_bucket(record, task_route) for record in failed] + counts = Counter(item["bucket"] for item in classifications) + confidence_sum: Dict[str, float] = defaultdict(float) + examples: Dict[str, List[Dict[str, Any]]] = defaultdict(list) + domains: Dict[str, Counter] = defaultdict(Counter) + original_other = 0 + + for record, item in zip(failed, classifications): + confidence_sum[item["bucket"]] += item["confidence"] + if item["original_stage"] == "other": + original_other += 1 + domain = _first_text(record, ("domain", "subset", "source", "subject", "category")) + if not domain: + judge = record.get("judge") if isinstance(record.get("judge"), dict) else {} + domain = _text(judge.get("domain") or "unknown") + domains[item["bucket"]][domain] += 1 + if len(examples[item["bucket"]]) < 3: + completion, result, _, _ = _record_text(record) + examples[item["bucket"]].append({ + "task_id": record.get("task_id") or record.get("id") or record.get("sample_id"), + "original_stage": item["original_stage"], + "reason": item["reason"], + "result_head": result[:180], + "completion_head": completion.replace("\n", " ")[:180], + }) + + total_failed = len(failed) + if task_route == "text2sql": + meta_map = _SQL_BUCKET_META + elif task_route == "general": + meta_map = _GENERAL_BUCKET_META + else: + meta_map = _CODE_BUCKET_META + actionable_counts = {key: count for key, count in counts.items() if key != _UNKNOWN_BUCKET} + candidate_counts = dict(actionable_counts) + exploration_priors: Dict[str, float] = {} + # When almost every sample fails before executable code/SQL is produced, + # downstream semantic ability is censored rather than proven healthy. + # Reserve a small first-round exploration budget instead of allocating + # 100% to the visible prerequisite failure. + if total_failed: + if task_route == "text2sql" and counts.get("sql_output_contract", 0) / total_failed >= 0.50: + exploration_priors = { + "sql_syntax": 0.25, + "sql_schema_linking": 0.15, + "sql_semantic_logic": 0.15, + } + elif task_route == "code" and counts.get("code_output_contract", 0) / total_failed >= 0.50: + exploration_priors = { + "code_syntax_completion": 0.25, + "code_interface_scope": 0.15, + "code_semantic_logic": 0.15, + } + elif task_route == "general": + prerequisite_count = max( + counts.get("general_instruction_following", 0), + counts.get("general_completeness_coverage", 0), + counts.get("general_safety_refusal", 0), + ) + if prerequisite_count / total_failed >= 0.50: + exploration_priors = { + "general_relevance_intent": 0.15, + "general_factuality_grounding": 0.12, + "general_reasoning_consistency": 0.08, + } + for key in exploration_priors: + candidate_counts.setdefault(key, 0) + + power_denominator = sum(float(count) ** float(alpha) for count in actionable_counts.values()) + utility_weights: Dict[str, float] = {} + bucket_rows = [] + + for key, count in sorted(candidate_counts.items(), key=lambda item: (-item[1], item[0])): + meta = meta_map.get(key, _UNKNOWN_META) + observed_share = count / max(total_failed, 1) + confidence = confidence_sum[key] / count if count else 0.55 + if count: + need_signal = (observed_share + 1e-9) ** float(alpha) + allocation_basis = "observed_failure_and_priors" + else: + need_signal = exploration_priors.get(key, 0.0) + allocation_basis = "censored_capability_exploration" + utility = ( + need_signal + * confidence + * meta["severity"] + * meta["transfer"] + * meta["learnability_prior"] + / max(meta["cost"], 1e-6) + ) + utility_weights[key] = utility + bucket_rows.append({ + "bucket": key, + "label": meta["label"], + "count": count, + "observed_share": round(observed_share, 4), + "power_adjusted_share": round((count ** float(alpha)) / power_denominator, 4) + if power_denominator else 0.0, + "classification_confidence": round(confidence, 4), + "severity": meta["severity"], + "transfer_value": meta["transfer"], + "learnability_prior": meta["learnability_prior"], + "data_cost": meta["cost"], + "allocation_basis": allocation_basis, + "censored_by_upstream_failure": count == 0 and key in exploration_priors, + "sample_direction": meta["sample_direction"], + "examples": examples[key], + "domain_breakdown": [ + { + "domain": domain, + "count": domain_count, + "share_within_bucket": round(domain_count / max(count, 1), 4), + } + for domain, domain_count in domains[key].most_common(10) + ], + }) + + allocation = _project_with_floor_and_cap(utility_weights, min_share, max_share) + for row in bucket_rows: + row["recommended_share"] = allocation.get(row["bucket"], 0.0) + row["recommended_percent"] = round(row["recommended_share"] * 100, 2) + + unresolved = counts.get(_UNKNOWN_BUCKET, 0) + naive_alpha = 2.0 + original_stage_counts = Counter(item["original_stage"] for item in classifications) + naive_denominator = sum(float(count) ** naive_alpha for count in original_stage_counts.values()) + naive_other_share = ( + (float(original_stage_counts.get("other", 0)) ** naive_alpha) / naive_denominator + if naive_denominator else 0.0 + ) + unresolved_share = unresolved / max(total_failed, 1) + warnings = [] + if unresolved_share > 0.10: + warnings.append("待诊断样本超过失败样本的 10%，分桶预算置信度不足，应先补充执行证据或人工复核。") + if original_other: + warnings.append("原始 other 不参与训练预算；先用执行证据重分类，仍无法归因的样本进入诊断池。") + + return { + "strategy": "confidence_and_marginal_gain_aware", + "task_type": task_route, + "requested_task_type": str(task_type), + "failed_total": total_failed, + "parameters": { + "power_alpha": float(alpha), + "min_bucket_share": float(min_share), + "max_bucket_share": float(max_share), + "learnability": "prior_until_pilot_gain_is_available", + }, + "other_impact": { + "original_other_count": original_other, + "original_other_share": round(original_other / max(total_failed, 1), 4), + "naive_alpha_2_other_share": round(naive_other_share, 4), + "reclassified_from_other_count": sum( + 1 for item in classifications if item["reclassified_from_other"] + ), + "unresolved_count": unresolved, + "unresolved_share": round(unresolved_share, 4), + "training_allocation_share": 0.0, + }, + "buckets": bucket_rows, + "diagnostic_bucket": { + "bucket": _UNKNOWN_BUCKET, + "label": _UNKNOWN_META["label"], + "count": unresolved, + "recommended_share": 0.0, + "sample_direction": _UNKNOWN_META["sample_direction"], + "examples": examples[_UNKNOWN_BUCKET], + }, + "pilot_update_rule": ( + "每轮小规模补数后，以该桶目标指标增量/新增样本数更新 learnability，" + "下一轮按边际收益重新分配；不把当前错误占比永久固化为训练占比。" + ), + "methodology_references": _GENERAL_METHOD_REFERENCES if task_route == "general" else [], + "warnings": warnings, + } diff --git a/loopai/skills/Analyzer/cli.py b/loopai/skills/Analyzer/cli.py index 566b1fe..d98327e 100644 --- a/loopai/skills/Analyzer/cli.py +++ b/loopai/skills/Analyzer/cli.py @@ -50,6 +50,7 @@ def _parser() -> argparse.ArgumentParser: parser.add_argument("--checkpoint-path", default=None) parser.add_argument("--baseline-result-path", default=None) parser.add_argument("--analyze-batch-size", type=int, default=None) + parser.add_argument("--request-timeout-seconds", type=float, default=None) parser.add_argument("--resume", action="store_true") parser.add_argument("--new-version", action="store_true") parser.add_argument("--from-node", default=None) @@ -78,6 +79,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int: checkpoint_path=args.checkpoint_path, baseline_result_path=args.baseline_result_path, analyze_batch_size=args.analyze_batch_size, + analyze_request_timeout_seconds=args.request_timeout_seconds, version_id=args.version_id, force_new_version=args.new_version, emit_status=False, diff --git a/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py b/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py index d84443c..7b101e1 100644 --- a/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py +++ b/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py @@ -8,6 +8,7 @@ from loopai.common.event_tool import StreamEvent from loopai.skills.Analyzer.utils.stream import get_safe_stream_writer from loopai.common.prompts.prompt_loader import PromptLoader +from loopai.skills.Analyzer.bucket_strategy import build_training_bucket_strategy from langchain_openai import ChatOpenAI from loopai.schema.states import LoopAIState from loopai.logger import get_logger @@ -74,6 +75,7 @@ def init_model(state: LoopAIState) -> ChatOpenAI: base_url=cfg.get("analyze_base_url"), temperature=cfg.get("analyze_temperature", 0.0), top_p=cfg.get("analyze_top_p", 0.95), + timeout=float(cfg.get("analyze_request_timeout_seconds", 300)), ) return model @@ -365,6 +367,7 @@ def _build_obtainer_stats( match_type_pass_counter = {} domain_counter = {} field_presence_counter = {} + bucket_records = [] for idx, detail in enumerate(details): score = _normalize_detail_score(detail) @@ -384,6 +387,14 @@ def _build_obtainer_stats( "generated_ans": rec.get("generated_ans") or rec.get("completion") or rec.get("prediction"), } + bucket_record = dict(rec) + bucket_record["passed"] = score != 0.0 + bucket_record["metric_detail"] = detail + bucket_record["primary_metric"] = primary_metric_name + bucket_record.setdefault("generated_ans", sample["generated_ans"]) + bucket_record.setdefault("domain", domain) + bucket_records.append(bucket_record) + if isinstance(detail, dict): sample["match_type"] = detail.get("match_type") sample["extracted"] = detail.get("extracted") @@ -433,6 +444,14 @@ def _build_obtainer_stats( top_fields = sorted(field_presence_counter.items(), key=lambda x: x[1], reverse=True)[:20] top_domains = sorted(domain_counter.items(), key=lambda x: x[1], reverse=True)[:20] + analyzer_cfg = _analyzer(state) + allocation_plan = build_training_bucket_strategy( + bucket_records, + task_type="general", + alpha=float(analyzer_cfg.get("bucket_power_alpha", 1.0)), + min_share=float(analyzer_cfg.get("bucket_min_share", 0.05)), + max_share=float(analyzer_cfg.get("bucket_max_share", 0.50)), + ) return { "primary_metric": primary_metric_name, @@ -443,6 +462,11 @@ def _build_obtainer_stats( "field_presence_top": top_fields, "fail_bias_match_type": fail_bias_match_type[:15], "representative_failure_samples": representative_failure_samples, + "actionable_bucket_top": [ + [row["label"], row["count"]] + for row in allocation_plan.get("buckets", []) + ], + "allocation_plan": allocation_plan, } @@ -479,7 +503,7 @@ def build_prompt_for_data_plan(summary: Dict[str, Any]) -> str: loader = PromptLoader() template = loader("analyze_metric_report", "data_plan_user") - return template.format( + prompt = template.format( bench_name=summary["bench_name"], eval_type=summary["eval_type"], task_domain=summary["task_domain"], @@ -493,6 +517,16 @@ def build_prompt_for_data_plan(summary: Dict[str, Any]) -> str: quick_samples_json=json.dumps(summary["quick_samples"], ensure_ascii=False), summary_json=json.dumps(summary["summary_json"], ensure_ascii=False), ) + allocation_json = json.dumps(summary.get("allocation_plan") or {}, ensure_ascii=False) + return prompt + f""" + +【General Text 分桶约束】 +1. 必须优先使用 summary 中的 allocation_plan，不得直接按 primary metric 失败比例分配训练数据。 +2. 指令遵循、相关性、事实性、推理、完整性、语言质量和安全拒答是相互独立的能力桶。 +3. other/待诊断样本不进入训练预算，只能建议补充评测证据或人工复核。 +4. recommended_percent 仅表示第一轮先验预算；小规模试训后应按单位样本指标收益更新。 +allocation_plan={allocation_json} +""" def build_prompt_for_obtainer(summary: Dict[str, Any], obtainer_stats: Dict[str, Any]) -> str: """ @@ -502,7 +536,7 @@ def build_prompt_for_obtainer(summary: Dict[str, Any], obtainer_stats: Dict[str, loader = PromptLoader() template = loader("data_obtainer", "suggest_obtainer") - return template.format( + prompt = template.format( dataset_json=json.dumps({ "bench_name": summary["bench_name"], "eval_type": summary["eval_type"], @@ -514,6 +548,42 @@ def build_prompt_for_obtainer(summary: Dict[str, Any], obtainer_stats: Dict[str, summary_json=json.dumps(summary["summary_json"], ensure_ascii=False), obtainer_stats_json=json.dumps(obtainer_stats, ensure_ascii=False), ) + return prompt + """ + +【General Text 数据获取约束】 +1. 使用 allocation_plan.recommended_percent 生成能力级数据预算，再参考 domain_breakdown 选择内容领域。 +2. 不得把 primary_metric_failure、unknown 或 other 直接当作可采集的数据类型。 +3. 每个能力桶的数据建议必须对应 sample_direction，并说明验证该能力提升的指标。 +""" + + +def _render_allocation_plan(allocation_plan: Dict[str, Any]) -> str: + rows = allocation_plan.get("buckets") or [] + if not rows: + return "" + lines = [ + "【General Text 训练数据分桶建议】", + "错误出现频率只作为需求信号；以下比例同时考虑归因置信度、严重性、迁移价值、学习效率和数据成本。", + ] + for row in sorted(rows, key=lambda item: -item.get("recommended_share", 0.0)): + lines.append( + f"- {row.get('label')}：{row.get('recommended_percent', 0):.2f}% " + f"（观察 {row.get('count', 0)} 条，置信度 {row.get('classification_confidence', 0):.2f}）" + ) + lines.append(f" 样本方向：{row.get('sample_direction', '')}") + domains = row.get("domain_breakdown") or [] + if domains: + domain_text = "、".join( + f"{item.get('domain')} {item.get('count')} 条" for item in domains[:5] + ) + lines.append(f" 领域分布：{domain_text}") + other_impact = allocation_plan.get("other_impact") or {} + lines.append( + "- 待诊断样本：0.00% 训练预算" + f"（重分类后仍未解决 {other_impact.get('unresolved_count', 0)} 条）" + ) + lines.append(f"- 动态更新：{allocation_plan.get('pilot_update_rule', '')}") + return "\n".join(lines) def _invoke_prompt(llm, prompt): """ @@ -567,6 +637,9 @@ def _emit(message, *, progress=None, data=None): records = _load_records_from_alignment(metric_result) summary = _build_summary(state, metric_result, records) obtainer_stats = _build_obtainer_stats(state, metric_result, records, summary) + allocation_plan = obtainer_stats.get("allocation_plan") or {} + summary["allocation_plan"] = allocation_plan + _analyzer(state)["allocation_plan"] = allocation_plan _emit( "已构建 metric 摘要", @@ -606,6 +679,11 @@ def _emit(message, *, progress=None, data=None): data={"prompt_chars": len(obtainer_prompt or "")}, ) obtainer_text = _invoke_prompt(llm, obtainer_prompt) + allocation_text = _render_allocation_plan(allocation_plan) + if allocation_text: + report_text = f"{report_text.rstrip()}\n\n{allocation_text}\n" + data_plan_text = f"{data_plan_text.rstrip()}\n\n{allocation_text}\n" + obtainer_text = f"{obtainer_text.rstrip()}\n\n{allocation_text}\n" ts = time.strftime("%Y%m%d_%H%M%S") outdir = _ensure_analyzer_outdir(state) diff --git a/loopai/skills/Analyzer/nodes/analyze_result.py b/loopai/skills/Analyzer/nodes/analyze_result.py index 3567cec..6fb3270 100644 --- a/loopai/skills/Analyzer/nodes/analyze_result.py +++ b/loopai/skills/Analyzer/nodes/analyze_result.py @@ -58,6 +58,8 @@ def init_model(state: LoopAIState) -> ChatOpenAI: base_url=cfg['analyze_base_url'], temperature=cfg.get('analyze_temperature', 0.0), top_p=cfg.get('analyze_top_p', 0.95), + timeout=float(cfg.get('analyze_request_timeout_seconds', 300)), + max_retries=int(cfg.get('analyze_request_max_retries', 0)), ) return model @@ -80,15 +82,28 @@ def _is_timeout(exc: Exception) -> bool: ) def _compact_prompt(value: str) -> str: - max_chars = 12000 + max_chars = int((data or {}).get("retry_prompt_max_chars") or 12000) if len(value) <= max_chars: return value + "\n\nPlease answer concisely and keep the required output format." head = max_chars * 2 // 3 tail = max_chars - head return value[:head] + "\n...[Analyzer compact retry: middle evidence omitted]...\n" + value[-tail:] + def _chunk_text(chunk: Any) -> str: + content = getattr(chunk, "content", chunk) + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join( + str(item.get("text", "")) if isinstance(item, dict) else str(item) + for item in content + ) + return str(content or "") + def _run_once(request_prompt: str) -> str: stop_event = threading.Event() + started_at = time.monotonic() + first_chunk_seconds = None resume_progress = get_analyzer_resume_progress() request_start = max(start_progress, min(resume_progress, end_progress)) if resume_progress else start_progress @@ -105,7 +120,33 @@ def _heartbeat() -> None: heartbeat_thread = threading.Thread(target=_heartbeat, daemon=True) heartbeat_thread.start() try: - return llm.batch([request_prompt])[0].content + chunks = [] + for chunk in llm.stream(request_prompt): + if first_chunk_seconds is None: + first_chunk_seconds = round(time.monotonic() - started_at, 3) + chunks.append(_chunk_text(chunk)) + response = "".join(chunks) + emit("模型响应接收完成", progress=end_progress, data={ + **(data or {}), + "prompt_chars": len(request_prompt), + "response_chars": len(response), + "elapsed_seconds": round(time.monotonic() - started_at, 3), + "first_chunk_seconds": first_chunk_seconds, + "streaming": True, + }) + return response + except Exception as exc: + text = f"{type(exc).__name__}: {exc}".lower() + emit("模型请求失败", progress=start_progress, data={ + **(data or {}), + "prompt_chars": len(request_prompt), + "elapsed_seconds": round(time.monotonic() - started_at, 3), + "first_chunk_seconds": first_chunk_seconds, + "error_type": type(exc).__name__, + "timeout_scope": "upstream_gateway" if "524" in text else "client_or_provider", + "streaming": True, + }) + raise finally: stop_event.set() heartbeat_thread.join(timeout=0.2) @@ -115,9 +156,20 @@ def _heartbeat() -> None: except Exception as exc: if not _is_timeout(exc): raise - emit("模型请求超时，压缩输入后重试", progress=start_progress, data={ - **(data or {}), "retry": True, "input_compacted": True, - }) + error_text = f"{type(exc).__name__}: {exc}".lower() + emit( + "上游代理 524 超时，压缩输入后重试" + if "524" in error_text + else "模型请求超时，压缩输入后重试", + progress=start_progress, + data={ + **(data or {}), + "retry": True, + "input_compacted": True, + "original_prompt_chars": len(prompt), + "timeout_scope": "upstream_gateway" if "524" in error_text else "client_or_provider", + }, + ) return _run_once(_compact_prompt(prompt)) diff --git a/loopai/skills/Analyzer/nodes/draw_conclusion.py b/loopai/skills/Analyzer/nodes/draw_conclusion.py index d4bda8e..f009511 100644 --- a/loopai/skills/Analyzer/nodes/draw_conclusion.py +++ b/loopai/skills/Analyzer/nodes/draw_conclusion.py @@ -22,6 +22,7 @@ build_historical_comparison, render_historical_comparison_text, ) +from loopai.skills.Analyzer.bucket_strategy import build_training_bucket_strategy logger = get_logger() from collections import defaultdict from typing import List, Dict, Any @@ -131,6 +132,10 @@ def init_model(state: LoopAIState) -> ChatOpenAI: base_url=cfg['analyze_base_url'], temperature=cfg.get('analyze_temperature', 0.0), top_p=cfg.get('analyze_top_p', 0.95), + timeout=float(cfg.get('analyze_request_timeout_seconds', 300)), + # The adaptive retry below owns the retry policy so the same long + # prompt is not silently retried several times by the SDK first. + max_retries=int(cfg.get('analyze_request_max_retries', 0)), ) return model @@ -145,7 +150,7 @@ def _batch_one_with_heartbeat( end_progress: float, data: Dict[str, Any] | None = None, ) -> str: - """Run normally, then retry once with a compact prompt after timeout.""" + """Stream one normal request, then compact and retry once on timeout.""" def _is_timeout(exc: Exception) -> bool: text = f"{type(exc).__name__}: {exc}".lower() return isinstance(exc, TimeoutError) or any( @@ -153,15 +158,28 @@ def _is_timeout(exc: Exception) -> bool: ) def _compact_prompt(value: str) -> str: - max_chars = 12000 + max_chars = int((data or {}).get("retry_prompt_max_chars") or 12000) if len(value) <= max_chars: return value + "\n\nPlease answer concisely and keep the required output format." head = max_chars * 2 // 3 tail = max_chars - head return value[:head] + "\n...[Analyzer compact retry: middle evidence omitted]...\n" + value[-tail:] + def _chunk_text(chunk: Any) -> str: + content = getattr(chunk, "content", chunk) + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join( + str(item.get("text", "")) if isinstance(item, dict) else str(item) + for item in content + ) + return str(content or "") + def _run_once(request_prompt: str) -> str: stop_event = threading.Event() + started_at = time.monotonic() + first_chunk_seconds = None def heartbeat() -> None: tick = 0 @@ -175,7 +193,33 @@ def heartbeat() -> None: thread = threading.Thread(target=heartbeat, daemon=True) thread.start() try: - return llm.batch([request_prompt])[0].content + chunks = [] + for chunk in llm.stream(request_prompt): + if first_chunk_seconds is None: + first_chunk_seconds = round(time.monotonic() - started_at, 3) + chunks.append(_chunk_text(chunk)) + response = "".join(chunks) + emit("模型响应接收完成", progress=end_progress, data={ + **(data or {}), + "prompt_chars": len(request_prompt), + "response_chars": len(response), + "elapsed_seconds": round(time.monotonic() - started_at, 3), + "first_chunk_seconds": first_chunk_seconds, + "streaming": True, + }) + return response + except Exception as exc: + text = f"{type(exc).__name__}: {exc}".lower() + emit("模型请求失败", progress=start_progress, data={ + **(data or {}), + "prompt_chars": len(request_prompt), + "elapsed_seconds": round(time.monotonic() - started_at, 3), + "first_chunk_seconds": first_chunk_seconds, + "error_type": type(exc).__name__, + "timeout_scope": "upstream_gateway" if "524" in text else "client_or_provider", + "streaming": True, + }) + raise finally: stop_event.set() thread.join(timeout=0.2) @@ -185,8 +229,16 @@ def heartbeat() -> None: except Exception as exc: if not _is_timeout(exc): raise - emit("模型请求超时，压缩输入后重试", progress=start_progress, data={ + error_text = f"{type(exc).__name__}: {exc}".lower() + emit( + "上游代理 524 超时，压缩输入后重试" + if "524" in error_text + else "模型请求超时，压缩输入后重试", + progress=start_progress, + data={ **(data or {}), "retry": True, "input_compacted": True, + "original_prompt_chars": len(prompt), + "timeout_scope": "upstream_gateway" if "524" in error_text else "client_or_provider", }) return _run_once(_compact_prompt(prompt)) @@ -300,7 +352,12 @@ def enhance_stats_with_oj(records): return extras -def build_obtainer_stats(summary: dict, oj_records: list, final_json: dict) -> dict: +def build_obtainer_stats( + summary: dict, + oj_records: list, + final_json: dict, + strategy_config: dict | None = None, +) -> dict: """ 为 obtainer 提供更细粒度的数据缺口统计 """ @@ -328,7 +385,8 @@ def build_obtainer_stats(summary: dict, oj_records: list, final_json: dict) -> d fail_tags.update([str(t) for t in tags if t]) domain = ( - rec.get("domain") + rec.get("bench_name") + or rec.get("domain") or judge.get("domain") or rec.get("subset") or rec.get("source") @@ -374,6 +432,20 @@ def build_obtainer_stats(summary: dict, oj_records: list, final_json: dict) -> d "actual": ap.get("actual"), }) + strategy_config = strategy_config or {} + task_type = str( + strategy_config.get("analyze_task_type") + or summary.get("task_type") + or "code" + ) + allocation_plan = build_training_bucket_strategy( + oj_records or [], + task_type=task_type, + alpha=float(strategy_config.get("bucket_power_alpha", 1.0)), + min_share=float(strategy_config.get("bucket_min_share", 0.05)), + max_share=float(strategy_config.get("bucket_max_share", 0.50)), + ) + return { "failed_total": len(failed), "passed_total": len(passed), @@ -382,6 +454,11 @@ def build_obtainer_stats(summary: dict, oj_records: list, final_json: dict) -> d "domain_top": domain_counter.most_common(20), "fail_bias_tags": fail_bias_tags[:15], "representative_failure_samples": sample_briefs, + "actionable_bucket_top": [ + [row["label"], row["count"]] + for row in allocation_plan.get("buckets", []) + ], + "allocation_plan": allocation_plan, } @@ -490,11 +567,19 @@ def build_obtainer_prompt(final_json: dict, obtainer_stats: dict) -> str: summary_json = json.dumps(final_json.get("summary", {}), ensure_ascii=False, indent=2) obtainer_stats_json = json.dumps(obtainer_stats, ensure_ascii=False, indent=2) - return tpl.format( + prompt = tpl.format( dataset_json=dataset_json, summary_json=summary_json, obtainer_stats_json=obtainer_stats_json, ) + return prompt + """ + +【分桶预算约束】 +1. 优先使用细粒度统计中的 allocation_plan，而不是直接按 failure_stage_top 等比例分桶。 +2. other/待诊断样本不进入训练预算；只能建议补日志、重分类或人工复核。 +3. recommended_percent 是第一轮先验预算，不是永久比例；需说明小规模试训后按单位样本收益更新。 +4. 不得因为某类错误出现得多，就默认它需要同比例训练数据。 +""" def build_background_prompt(final_json: dict) -> str: """ 构建背景介绍的提示文本：只介绍数据集本身 @@ -549,7 +634,10 @@ def pct(x, y, digits=2): lines.append(f"本次评测共 {t} 个样本，其中通过 {p} 个，样本正确率 {pass_rate_str}。") if top_cnt > 0: - lines.append(f"最主要的失败类型是 “{top_name}”，共有 {top_cnt} 次。") + if top_name == "other" and (final_json.get("obtainer_stats") or {}).get("allocation_plan"): + lines.append(f"原始聚合中 “other” 共有 {top_cnt} 次；该标签仅表示尚未归因，不能直接作为训练分桶。") + else: + lines.append(f"最主要的失败类型是 “{top_name}”，共有 {top_cnt} 次。") task_type = final_json.get("dataset", {}).get("task_type", "code") if task_type == "text2sql": @@ -566,7 +654,25 @@ def pct(x, y, digits=2): tags_str = ", ".join([f"{k}:{v}" for k, v in list(tags.items())[:8]]) lines.append(f"常见标签 Top：{tags_str}。") - lines.append("整体来看，建议优先修复最常见错误并优化边界测试。") + allocation_plan = (final_json.get("obtainer_stats") or {}).get("allocation_plan") or {} + bucket_rows = allocation_plan.get("buckets") or [] + if bucket_rows: + lines.append("") + lines.append("【训练数据分桶建议】") + lines.append("错误占比仅作为需求信号，以下比例同时考虑归因置信度、迁移价值和学习效率先验：") + for row in sorted(bucket_rows, key=lambda item: -item.get("recommended_share", 0)): + lines.append( + f"- {row.get('label')}：{row.get('recommended_percent', 0):.2f}% " + f"（观察到 {row.get('count', 0)} 条）" + ) + other_impact = allocation_plan.get("other_impact") or {} + lines.append( + f"- 待诊断/other：0.00%（原始 {other_impact.get('original_other_count', 0)} 条，" + f"重分类后仍未解决 {other_impact.get('unresolved_count', 0)} 条）" + ) + lines.append(f"- 更新规则：{allocation_plan.get('pilot_update_rule', '')}") + + lines.append("整体来看，应先修复阻断后续评测的前置能力，再通过小规模试训的单位样本收益动态调整分桶，而不是直接照搬错误占比。") comparison = final_json.get("historical_comparison") if comparison: lines.append(render_historical_comparison_text(comparison)) @@ -606,6 +712,30 @@ def make_obtainer_human_text(obtainer_stats: dict, llm_text: str = "") -> str: ) lines.append("") + allocation_plan = obtainer_stats.get("allocation_plan") or {} + bucket_rows = allocation_plan.get("buckets") or [] + if bucket_rows: + lines.append("建议训练数据分桶（第一轮先验）：") + for row in sorted(bucket_rows, key=lambda item: -item.get("recommended_share", 0)): + basis = ( + "观察证据" + if row.get("allocation_basis") == "observed_failure_and_priors" + else "下游能力被前置失败遮蔽，预留探索预算" + ) + lines.append( + f"- {row.get('label')}：{row.get('recommended_percent', 0):.2f}% " + f"（观察 {row.get('count', 0)} 条，置信度 {row.get('classification_confidence', 0):.2f}，{basis}）" + ) + lines.append(f" 样本方向：{row.get('sample_direction')}") + other_impact = allocation_plan.get("other_impact") or {} + lines.append( + "- 待诊断样本：0.00% 训练预算" + f"（原始 other {other_impact.get('original_other_count', 0)} 条，" + f"重分类后未解决 {other_impact.get('unresolved_count', 0)} 条）" + ) + lines.append(f"动态更新：{allocation_plan.get('pilot_update_rule', '')}") + lines.append("") + if llm_text: lines.append("【模型生成的 Obtainer 建议】") lines.append(llm_text.strip()) @@ -744,6 +874,7 @@ def _emit(message, *, progress=None, data=None): "field_schema": field_schema, "example": example, "total_samples": final_json["totals"]["total_samples"], + "benches": summary.get("bench_summaries") or {}, } final_json["samples"] = samples @@ -766,8 +897,14 @@ def _emit(message, *, progress=None, data=None): final_json["background"] = background_text _emit("背景介绍生成完成", progress=0.38) - obtainer_stats = build_obtainer_stats(summary, oj_records, final_json) + obtainer_stats = build_obtainer_stats( + summary, + oj_records, + final_json, + strategy_config=_analyzer(state), + ) final_json["obtainer_stats"] = obtainer_stats + state.setdefault("analyzer", {})["allocation_plan"] = obtainer_stats.get("allocation_plan", {}) baseline_result_path = _analyzer(state).get("baseline_result_path") if baseline_result_path: @@ -778,6 +915,7 @@ def _emit(message, *, progress=None, data=None): # ===== 写入 JSON 报告 ===== final_json_path = os.path.join(outdir, f"final_report_{run_ts}.json") + state["analyzer"]["analyze_output_final_report_json_path"] = final_json_path _emit("写入最终 JSON 报告", progress=0.4 , data={"path": final_json_path}) with open(final_json_path, "w", encoding="utf-8") as f: f.write(json.dumps(final_json, ensure_ascii=False, indent=2)) @@ -788,6 +926,7 @@ def _emit(message, *, progress=None, data=None): # ===== 写入文本报告（包含背景介绍）===== final_txt_path = os.path.join(outdir, f"final_report_{run_ts}.txt") + state["analyzer"]["analyze_output_final_report_text_path"] = final_txt_path _emit("写入文本报告", progress=0.5 , data={"path": final_txt_path}) with open(final_txt_path, "w", encoding="utf-8") as f: f.write(make_human_text(final_json, background=background_text)) diff --git a/loopai/skills/Analyzer/nodes/eval_model.py b/loopai/skills/Analyzer/nodes/eval_model.py index 0d22655..4dda113 100644 --- a/loopai/skills/Analyzer/nodes/eval_model.py +++ b/loopai/skills/Analyzer/nodes/eval_model.py @@ -22,6 +22,7 @@ get_analyzer_resume_progress, get_safe_stream_writer, ) +from loopai.skills.Analyzer.bench_inputs import resolve_eval_result_sources # ===== PromptLoader 单例 & 模板缓存 ===== _PROMPT_LOADER: PromptLoader | None = None _TEMPLATE_CACHE: dict[tuple[str, str], str] = {} @@ -276,6 +277,7 @@ def init_model(state: LoopAIState) -> BaseChatModel: max_tokens=cfg.get("analyze_max_tokens", 512), temperature=cfg.get("analyze_temperature", 0.0), top_p=cfg.get("analyze_top_p", 0.95), + request_timeout=float(cfg.get("analyze_request_timeout_seconds", 300)), ) def build_judge_prompt_generic(task: str, evidence: Dict[str, Any]) -> str: @@ -514,7 +516,13 @@ def build_evidence_for_record(rec: Dict[str, Any], task_type: str) -> Dict[str, return build_evidence_for_record_sql(rec) return build_evidence_for_record_code(rec) -def _build_and_write_summary(rows: List[Dict[str, Any]], outdir: Path, run_ts: str, task_type: str = "code"): +def _build_and_write_summary( + rows: List[Dict[str, Any]], + outdir: Path, + run_ts: str, + task_type: str = "code", + eval_result_sources: Optional[List[Dict[str, str]]] = None, +): """ 根据 rows 生成 summary 的函数 Args: @@ -539,6 +547,9 @@ def _build_and_write_summary(rows: List[Dict[str, Any]], outdir: Path, run_ts: s task_pass_map = defaultdict(bool) task_ids = set() + bench_stats: Dict[str, Dict[str, Any]] = defaultdict( + lambda: {"total_samples": 0, "passed_samples": 0, "failure_stage_distribution": Counter()} + ) def bin_loc(n: int) -> str: """ @@ -564,11 +575,15 @@ def bin_kw(n: int) -> str: for rec in rows: total_samples += 1 + bench_name = str(rec.get("bench_name") or "default") + bench_stats[bench_name]["total_samples"] += 1 if rec.get("passed"): passed_samples += 1 + bench_stats[bench_name]["passed_samples"] += 1 else: stg = (rec.get("judge") or {}).get("stage", "other") stage_counter[stg] += 1 + bench_stats[bench_name]["failure_stage_distribution"][stg] += 1 tags = (rec.get("judge") or {}).get("tags") or [] for t in tags: tag_counter[t] += 1 @@ -608,6 +623,7 @@ def bin_kw(n: int) -> str: tid = rec.get("task_id") if tid is not None: + tid = f"{bench_name}::{tid}" task_ids.add(tid) if rec.get("passed"): task_pass_map[tid] = True @@ -621,8 +637,20 @@ def bin_kw(n: int) -> str: else: pass_at_k_task = {} + bench_summaries = {} + for bench_name, values in sorted(bench_stats.items()): + bench_total = int(values["total_samples"]) + bench_passed = int(values["passed_samples"]) + bench_summaries[bench_name] = { + "total_samples": bench_total, + "passed_samples": bench_passed, + "pass_rate_samples": round(bench_passed / max(bench_total, 1), 4), + "failure_stage_distribution": dict(values["failure_stage_distribution"]), + } + summary = { "run_ts": run_ts, + "task_type": task_type, "results_file": None, "total_samples": total_samples, "passed_samples": passed_samples, @@ -638,6 +666,8 @@ def bin_kw(n: int) -> str: "expected_top10": expected_top.most_common(10), }, "tag_top10": tag_counter.most_common(10), + "bench_summaries": bench_summaries, + "eval_result_sources": eval_result_sources or [], } os.makedirs(outdir, exist_ok=True) @@ -649,6 +679,13 @@ def bin_kw(n: int) -> str: lines = [] lines.append(f"评测时间：{run_ts}") lines.append(f"样本正确率：{passed_samples}/{total_samples}（{summary['pass_rate_samples'] * 100:.2f}%）") + if len(bench_summaries) > 1: + lines.append("分 Bench 结果：") + for bench_name, values in bench_summaries.items(): + lines.append( + f" - {bench_name}: {values['passed_samples']}/{values['total_samples']} " + f"（{values['pass_rate_samples'] * 100:.2f}%）" + ) if pass_at_k_task: lines.append("Pass@k(任务口径)： " + ", ".join([f"Pass@{k}={v * 100:.2f}%" for k, v in pass_at_k_task.items()])) lines.append("主要错因(stage)分布：") @@ -834,29 +871,47 @@ def eval_model_node(state: LoopAIState): judge = LLMJudge(task=task_type) - # 读取评测结果（JSONL） - judger_cfg = state.get("judger") or {} + # 读取一个或多个同类型 Bench 结果。单个字符串路径保持兼容。 analyzer_cfg = state.get("analyzer") or {} - eval_result_path = judger_cfg.get("output_result_path") - if not eval_result_path: - eval_result_path = analyzer_cfg.get("eval_result_path") - - if not eval_result_path: + eval_result_sources = resolve_eval_result_sources(state, task_type) + if not eval_result_sources: raise ValueError( - "Missing analyzer.eval_result_path. " - "Please provide analyzer.eval_result_path " - "or run judger to generate output_result_path." + f"No Analyzer-compatible result file found for task_type={task_type}. " + "Please provide analyzer.eval_result_path(s) or run Judger with at least " + "one bench of the same task_type." ) state.setdefault("analyzer", {}) - state["analyzer"]["eval_result_path"] = eval_result_path - - with open(eval_result_path, 'r', encoding='utf-8') as f: - lines = [ln for ln in f if ln.strip()] - result_content = [json.loads(ln) for ln in lines] + state["analyzer"]["eval_result_path"] = eval_result_sources[0]["path"] + state["analyzer"]["eval_result_paths"] = [item["path"] for item in eval_result_sources] + state["analyzer"]["eval_result_sources"] = eval_result_sources + + result_content = [] + for source in eval_result_sources: + source_path = source["path"] + with open(source_path, "r", encoding="utf-8") as f: + for line in f: + if not line.strip(): + continue + record = json.loads(line) + if isinstance(record, dict): + record.setdefault("bench_name", source["bench_name"]) + result_content.append(record) + if writer: + writer(StreamEvent( + current="analyzer.eval_model", + progress=0.03, + message=f"已读取 {len(eval_result_sources)} 个 Bench", + data={ + "bench_count": len(eval_result_sources), + "bench_names": [item["bench_name"] for item in eval_result_sources], + "total_samples": len(result_content), + }, + ).json()) # 仅失败样本做判因 - failed_results = [r for r in result_content if not r.get("passed")] + failed_positions = [index for index, record in enumerate(result_content) if not record.get("passed")] + failed_results = [result_content[index] for index in failed_positions] total_failed = len(failed_results) batch_checkpoint_path = _batch_checkpoint_path(state) # 初始化 LLM @@ -1104,13 +1159,19 @@ def _stable_key(i: int): data=None ).json()) - logger.info(f" 判因完成：共 {len(failed_results)} 条") + for index, position in enumerate(failed_positions): + result_content[position] = failed_results[index] + + logger.info( + f" 判因完成：失败样本 {len(failed_results)} 条，完整样本 {len(result_content)} 条，" + f"Bench {len(eval_result_sources)} 个" + ) ts = time.strftime("%Y%m%d_%H%M%S") out_dir = _ensure_analyzer_outdir(state) out_jsonl_path = out_dir / f"oj_records_enriched_{ts}.jsonl" with open(out_jsonl_path, "w", encoding="utf-8") as f: - for r in failed_results: + for r in result_content: f.write(json.dumps(r, ensure_ascii=False) + "\n") state.setdefault("analyzer", {}) state["analyzer"]["analyze_output_result_path"] = str(out_jsonl_path.resolve()) @@ -1125,7 +1186,11 @@ def _stable_key(i: int): ).json()) try: summary_json_path, summary_txt_path = _build_and_write_summary( - failed_results, _ensure_analyzer_outdir(state), ts, task_type=task_type + result_content, + _ensure_analyzer_outdir(state), + ts, + task_type=task_type, + eval_result_sources=eval_result_sources, ) with open(summary_json_path, "r", encoding="utf-8") as f: sdata = json.load(f) diff --git a/loopai/skills/Analyzer/pipeline_runner.py b/loopai/skills/Analyzer/pipeline_runner.py index d6fb2b3..35a7426 100644 --- a/loopai/skills/Analyzer/pipeline_runner.py +++ b/loopai/skills/Analyzer/pipeline_runner.py @@ -18,10 +18,24 @@ "finish", ) +GENERAL_ANALYZER_PIPELINE_STEPS = ( + "metric_recommend", + "metric_score", + "analyze_metric_report", + "finish", +) + +_ALL_ANALYZER_PIPELINE_STEPS = tuple(dict.fromkeys( + ANALYZER_PIPELINE_STEPS + GENERAL_ANALYZER_PIPELINE_STEPS +)) + _STEP_PROGRESS_RANGES = { "eval_model": (0.00, 0.45), "analyze_result": (0.45, 0.75), "draw_conclusion": (0.75, 0.95), + "metric_recommend": (0.00, 0.15), + "metric_score": (0.15, 0.55), + "analyze_metric_report": (0.55, 0.95), "finish": (0.95, 1.00), } @@ -33,6 +47,12 @@ "AnalyzerAgent.analyze_result_node": "analyze_result", "draw_conclusion_node": "draw_conclusion", "AnalyzerAgent.draw_conclusion_node": "draw_conclusion", + "metric_recommend_node": "metric_recommend", + "AnalyzerAgent.metric_recommend_node": "metric_recommend", + "metric_score_node": "metric_score", + "AnalyzerAgent.metric_score_node": "metric_score", + "analyze_metric_report_node": "analyze_metric_report", + "AnalyzerAgent.analyze_metric_report_node": "analyze_metric_report", "finish_node": "finish", "AnalyzerAgent.finish_node": "finish", } @@ -42,14 +62,14 @@ def normalize_analyzer_step(step_name: Optional[str]) -> Optional[str]: if not step_name: return None step_name = str(step_name) - if step_name in ANALYZER_PIPELINE_STEPS: + if step_name in _ALL_ANALYZER_PIPELINE_STEPS: return step_name if step_name in _STEP_ALIASES: return _STEP_ALIASES[step_name] for alias, step in _STEP_ALIASES.items(): if alias in step_name: return step - for step in ANALYZER_PIPELINE_STEPS: + for step in _ALL_ANALYZER_PIPELINE_STEPS: if step in step_name: return step return step_name @@ -365,25 +385,33 @@ def load_analyzer_checkpoint( return load_analyzer_state_from_configer(task_id=thread_id) -def _start_index(step_name: str) -> int: +def _pipeline_steps_for_state(state: Dict[str, Any]) -> tuple[str, ...]: + task_type = str((state.get("analyzer") or {}).get("analyze_task_type") or "code").lower() + if task_type in {"code", "coding", "programming", "python", "text2sql", "text_to_sql", "sql"}: + return ANALYZER_PIPELINE_STEPS + return GENERAL_ANALYZER_PIPELINE_STEPS + + +def _start_index(step_name: str, pipeline_steps: tuple[str, ...] = ANALYZER_PIPELINE_STEPS) -> int: step_name = normalize_analyzer_step(step_name) - if step_name not in ANALYZER_PIPELINE_STEPS: - available = ", ".join(ANALYZER_PIPELINE_STEPS) + if step_name not in pipeline_steps: + available = ", ".join(pipeline_steps) raise ValueError(f"Unknown Analyzer step: {step_name}. Available steps: {available}") - return ANALYZER_PIPELINE_STEPS.index(step_name) + return pipeline_steps.index(step_name) def _resume_step_from_state(state: Dict[str, Any]) -> str: + pipeline_steps = _pipeline_steps_for_state(state) current = normalize_analyzer_step(state.get("current")) last_completed = normalize_analyzer_step(state.get("last_completed")) - if last_completed in ANALYZER_PIPELINE_STEPS: - if current == last_completed or current not in ANALYZER_PIPELINE_STEPS: - next_index = min(_start_index(last_completed) + 1, len(ANALYZER_PIPELINE_STEPS) - 1) - return ANALYZER_PIPELINE_STEPS[next_index] - if current in ANALYZER_PIPELINE_STEPS: + if last_completed in pipeline_steps: + if current == last_completed or current not in pipeline_steps: + next_index = min(_start_index(last_completed, pipeline_steps) + 1, len(pipeline_steps) - 1) + return pipeline_steps[next_index] + if current in pipeline_steps: return current - return ANALYZER_PIPELINE_STEPS[0] + return pipeline_steps[0] def _is_finished(state: Dict[str, Any]) -> bool: @@ -420,6 +448,15 @@ def _run_step( if step_name == "draw_conclusion": from loopai.skills.Analyzer.nodes.draw_conclusion import draw_conclusion_node return draw_conclusion_node(state) + if step_name == "metric_recommend": + from loopai.skills.Analyzer.nodes.metric_recommend_node import metric_recommend_node + return metric_recommend_node(state) + if step_name == "metric_score": + from loopai.skills.Analyzer.nodes.metric_score_node import metric_score_node + return metric_score_node(state) + if step_name == "analyze_metric_report": + from loopai.skills.Analyzer.nodes.analyze_metric_report_node import analyze_metric_report_node + return analyze_metric_report_node(state) raise ValueError(f"Unknown executable Analyzer step: {step_name}") finally: reset_analyzer_stream_writer(token) @@ -451,16 +488,18 @@ def run_analyzer_pipeline( if baseline_result_path: state.setdefault("analyzer", {})["baseline_result_path"] = baseline_result_path + pipeline_steps = _pipeline_steps_for_state(state) + if from_node is not None: start_step = normalize_analyzer_step(from_node) elif resume: start_step = _resume_step_from_state(state) else: - start_step = ANALYZER_PIPELINE_STEPS[0] + start_step = pipeline_steps[0] if start_step is None: - start_step = ANALYZER_PIPELINE_STEPS[0] - start_at = _start_index(start_step) + start_step = pipeline_steps[0] + start_at = _start_index(start_step, pipeline_steps) initial_resume_progress = ( resume_progress if resume and from_node is None else 0.0 ) @@ -512,7 +551,7 @@ def checkpoint_progress( ) return state - for step_name in ANALYZER_PIPELINE_STEPS[start_at:]: + for step_name in pipeline_steps[start_at:]: state["current"] = step_name step_resume_progress = ( resume_progress diff --git a/loopai/skills/Analyzer/runtime_config.py b/loopai/skills/Analyzer/runtime_config.py index e44d506..542459c 100644 --- a/loopai/skills/Analyzer/runtime_config.py +++ b/loopai/skills/Analyzer/runtime_config.py @@ -228,6 +228,14 @@ def resolve_analyzer_runtime_config( state.get("output_dir") if isinstance(state, dict) else None, "./outputs", ) + request_timeout_seconds = float(_first_non_empty( + kwargs.get("analyze_request_timeout_seconds"), + os.getenv("ANALYZER_REQUEST_TIMEOUT_SECONDS"), + analyzer.get("analyze_request_timeout_seconds"), + 300, + )) + if request_timeout_seconds <= 0: + raise ValueError("analyze_request_timeout_seconds must be greater than 0") require_api_key = kwargs.get("require_api_key") needs_llm = bool(require_api_key) if require_api_key is not None else bool(model or base_url) @@ -260,6 +268,7 @@ def resolve_analyzer_runtime_config( analyzer["db_path"] = db_path if output_dir: analyzer["output_dir"] = output_dir + analyzer["analyze_request_timeout_seconds"] = request_timeout_seconds if task_id and version_id and output_dir: analyzer["runtime_output_dir"] = str( Path(str(output_dir)) @@ -278,6 +287,7 @@ def resolve_analyzer_runtime_config( "db_path": db_path, "analyzer_model": model, "analyzer_base_url": base_url, + "analyze_request_timeout_seconds": request_timeout_seconds, "has_analyzer_api_key": bool(api_key), "api_key_source": ( "kwargs" diff --git a/loopai/skills/Analyzer/utils/openai_compat_llm.py b/loopai/skills/Analyzer/utils/openai_compat_llm.py index 7428cb1..d9a3d17 100644 --- a/loopai/skills/Analyzer/utils/openai_compat_llm.py +++ b/loopai/skills/Analyzer/utils/openai_compat_llm.py @@ -32,6 +32,7 @@ class OpenAICompatChat(BaseChatModel): max_tokens: int = 512 temperature: float = 0.0 top_p: float = 0.95 + request_timeout: float = 300.0 @property def _llm_type(self) -> str: @@ -92,7 +93,7 @@ def _generate( self.base_url.rstrip("/") + "/chat/completions", json=payload, headers=headers, - timeout=120, + timeout=self.request_timeout, ) resp.raise_for_status() data = resp.json() diff --git a/skills/Analyzer/BUCKET_STRATEGY.md b/skills/Analyzer/BUCKET_STRATEGY.md new file mode 100644 index 0000000..6b3cfa7 --- /dev/null +++ b/skills/Analyzer/BUCKET_STRATEGY.md @@ -0,0 +1,85 @@ +# Analyzer 三类任务分桶策略 + +本文档简要说明 Analyzer 当前对 Code、Text2SQL 和 General Text 三类任务的失败样本分桶与训练数据分配策略。 + +## 1. 共同原则 + +1. **三条路线独立分桶**：Code、Text2SQL 和 General Text 使用不同的能力桶，不把不同任务的错误混在一起统计。 +2. **错误占比不等于训练占比**：错误数量只表示缺陷被观察到的频率，训练投入还要考虑归因置信度、严重程度、迁移价值、学习效率和数据成本。 +3. **优先重分类 `other`**：先利用执行日志、评测标签和错误原因把 `other` 归入具体能力桶；仍无法归因的样本进入“待诊断样本”，训练预算为 0。 +4. **保留探索预算**：如果大量样本在前置环节失败，后续能力实际上没有被充分测试，系统会给被遮蔽的能力保留少量探索数据。 +5. **小规模试训后动态更新**：首轮使用学习效率先验；后续按“目标指标增量 / 新增样本数”更新学习效率，再调整下一轮比例。 + +每个桶的基础权重为： + +```text +weight_i = error_share_i^alpha + * confidence_i + * severity_i + * transfer_i + * learnability_i + / data_cost_i +``` + +权重归一化后得到建议训练比例。当前默认 `alpha=1.0`，单个有效桶的建议比例通常限制在 5% 至 50%，避免单一错误完全挤占训练数据。 + +## 2. Code 分桶 + +| 能力桶 | 典型问题 | 推荐数据方向 | +| --- | --- | --- | +| 代码补全输出契约 | 输出解释、Markdown 围栏、缺少完整函数 | 函数签名或 Docstring 到完整可执行代码的严格格式样本 | +| Python 语法与补全完整性 | 缩进、括号、字符串、`return` 或函数体不完整 | 短小 Python 语法修复与代码补全样本 | +| 函数接口、作用域与依赖 | 函数名或签名错误、变量未定义、缺少导入 | 接口保持、局部变量、标准库和辅助函数样本 | +| 语义逻辑与断言正确性 | 代码可执行但结果错误 | 带正常断言和反例断言的短算法样本 | +| 边界条件与鲁棒性 | 空输入、单元素、重复值、极值处理错误 | 边界条件与对比样本 | +| 运行时与效率 | 超时、递归过深、复杂度退化 | 低效与高效实现的成对优化样本 | + +当“代码补全输出契约”占失败样本 50% 以上时，系统仍会为语法、接口和语义能力保留探索预算，避免因为代码尚不可执行就误判这些能力没有问题。 + +## 3. Text2SQL 分桶 + +| 能力桶 | 典型问题 | 推荐数据方向 | +| --- | --- | --- | +| SQL 输出契约 | 输出解释、Markdown 或不可直接执行的 SQL | 问题与 Schema 到纯 SQL 的严格输出样本 | +| SQL 语法与结构 | `SELECT`、`JOIN`、聚合、子查询结构错误 | 短 SQL 修复与结构化组合样本 | +| Schema Linking | 表、列、实体或外键匹配错误 | 问题实体与 Schema 显式对齐、易混 Schema 对比样本 | +| SQL 语义与结果正确性 | SQL 可执行但过滤、聚合、排序或去重错误 | 可执行但结果错误的查询纠正样本 | +| 类型、值与条件表达 | 日期、数值、`NULL`、字符串和类型转换错误 | 值匹配、条件表达和类型边界样本 | +| SQL 运行时与效率 | 查询超时或不必要的高复杂度 | 等价 SQL 的低效与高效实现对比样本 | + +当“SQL 输出契约”占失败样本 50% 以上时，系统会额外探索 SQL 语法、Schema Linking 和语义正确性，避免前置格式错误遮蔽真实查询能力。 + +## 4. General Text 分桶 + +| 能力桶 | 典型问题 | 推荐数据方向 | +| --- | --- | --- | +| 指令与输出格式遵循 | 未满足格式、长度、结构或显式约束 | 多约束指令遵循和可验证格式样本 | +| 相关性与意图理解 | 答非所问、偏离用户目标、误解意图 | 相似意图辨析和目标对齐样本 | +| 事实性与知识依据 | 事实错误、幻觉、缺少给定依据 | 有可信来源或给定上下文的事实核验样本 | +| 推理与一致性 | 推理跳步、因果错误、前后矛盾 | 多步推理、一致性检查和反例验证样本 | +| 完整性与要点覆盖 | 漏答关键点、回答截断、信息不足 | 按评分要点补全和覆盖关键信息的样本 | +| 表达与语言质量 | 不流畅、不连贯、冗余或语法较差 | 流畅性、简洁性和结构化表达的改写样本 | +| 安全与拒答边界 | 应拒绝未拒绝，或对正常请求过度拒绝 | 合理拒答、不必要拒答和安全替代回答样本 | + +General Text 优先使用 Judger 提供的结构化标签和理由归因；仅在证据明确时使用空回答、格式违规或明显拒答等规则回退。普通 exact-match 失败不会被强行猜测为某个能力桶，而是进入待诊断样本。 + +General Text 采用两级统计：先确定上述能力桶的训练比例，再在每个能力桶内部统计领域分布，例如知识问答、摘要、写作或对话，避免把“领域”和“能力缺陷”混为一类。 + +## 5. 输出与使用 + +分桶结果写入： + +```text +state["analyzer"]["allocation_plan"] +``` + +报告中会同时给出：原始错误占比、重分类结果、建议训练比例、样本构造方向、代表性失败案例和待诊断样本告警。该比例适合作为下一轮数据补充的初始方案，不应被视为固定不变的长期数据配方。 + +## 6. 参考方法 + +- HELM（TMLR 2023）：将通用能力拆为正确性、鲁棒性与安全性等维度。 +- InstructGPT（NeurIPS 2022）：将指令遵循和用户意图对齐作为独立能力。 +- TruthfulQA（ACL 2022）：区分事实性问题与表面文本相似度。 +- Skill-It!（NeurIPS 2023）：考虑能力之间的先后依赖和被遮蔽能力的探索。 +- DoReMi（NeurIPS 2023）：根据训练价值动态优化数据混合，而非固定按频率分配。 +- LESS（ICML 2024）：使用目标任务收益选择更有影响力的训练数据。 diff --git a/skills/Analyzer/BUCKET_STRATEGY.pdf b/skills/Analyzer/BUCKET_STRATEGY.pdf new file mode 100644 index 0000000..c31441e Binary files /dev/null and b/skills/Analyzer/BUCKET_STRATEGY.pdf differ diff --git a/skills/Analyzer/SKILL.md b/skills/Analyzer/SKILL.md index fdeeb22..e54ec30 100644 --- a/skills/Analyzer/SKILL.md +++ b/skills/Analyzer/SKILL.md @@ -68,6 +68,7 @@ Supported options: - `--print-result` - `--list-nodes` - `--stream-stdout` +- `--request-timeout-seconds` (default: `300`) ## Environment Variables Runtime configuration should come from environment/system runtime where possible: @@ -79,6 +80,7 @@ Runtime configuration should come from environment/system runtime where possible - `DB_PATH` - `ANALYZER_CHECKPOINT_PATH` - `ANALYZER_VERSION_ID` / `VERSION_ID` +- `ANALYZER_REQUEST_TIMEOUT_SECONDS` Config JSON should not store API keys. Runtime API key/model/base URL should be placed under the task/system config when available: @@ -135,10 +137,14 @@ Analyzer output files and event files are version-scoped: //analyzer// ``` -Standalone function pipeline remains: +Standalone selects the pipeline from `analyzer.analyze_task_type`: `eval_model -> analyze_result -> draw_conclusion -> finish` +General text uses its existing metric pipeline: + +`metric_recommend -> metric_score -> analyze_metric_report -> finish` + The in-memory state still carries: - `state["current"]` @@ -151,6 +157,71 @@ Set `baseline_result_path` to enable Historical Comparison. Current results come Analyzer preserves the `historical_comparison` field and appends a `Historical Comparison` section to report/final_report outputs when available. Missing or unreadable baseline files produce a warning instead of failing the main flow. +## Multiple Benches + +Analyzer can consume two or more Judger results from `judger.bench_result` and +`judger.extra_bench_result`. Results with the same `task_type` are merged into +one analysis run while `summary["bench_summaries"]` preserves per-bench sample +counts, pass rates, and failure distributions. A single string +`analyzer.eval_result_path` remains supported. Standalone callers may also use: + +```json +{ + "analyzer": { + "analyze_task_type": "code", + "eval_result_path": ["humaneval.jsonl", "mbpp.jsonl"] + } +} +``` + +Do not combine `code`, `text2sql`, and general-text results in one Analyzer +route; each task type keeps its own analysis rules. + +## Data Bucket Strategy + +The final report includes `obtainer_stats.allocation_plan`. Analyzer first +reclassifies fallback `other` records with runtime/parser evidence, then +computes a first-round data budget from observed need, classification +confidence, severity, transfer value, learnability prior, and data cost. +`other`/unresolved records receive zero training allocation and enter a +diagnostic queue. Each actionable bucket is capped by default at 50%, and the +plan explicitly requires pilot-training gains to update later rounds. + +Analyzer keeps three independent bucket routes: + +- Code: output contract, syntax/completion, interface/scope, semantic logic, + boundary robustness, and runtime efficiency. +- Text2SQL: SQL output contract, syntax, schema linking, semantic correctness, + type/value handling, and runtime efficiency. +- General Text: instruction/format following, relevance/intent, factuality and + grounding, reasoning consistency, completeness/coverage, language quality, + and safety/refusal boundaries. + +General Text uses structured evaluator labels and reasons first. Empty answers, +verifiable format violations, and obvious refusal patterns provide deterministic +fallback evidence. Generic exact-match failures are not guessed into factuality +or reasoning; unresolved samples enter the zero-budget diagnostic queue. The +plan allocates by capability first and reports the observed domain distribution +inside each capability bucket. + +The General Text design follows these established ideas without claiming to +reimplement the full paper algorithms: + +- [HELM](https://arxiv.org/abs/2211.09110) (TMLR 2023): multi-dimensional model evaluation. +- [InstructGPT](https://proceedings.neurips.cc/paper_files/paper/2022/hash/b1efde53be364a73914f58805a001731-Abstract.html) (NeurIPS 2022): user intent and instruction following. +- [TruthfulQA](https://aclanthology.org/2022.acl-long.229/) (ACL 2022): truthfulness separated from informativeness. +- [Skill-It!](https://proceedings.neurips.cc/paper_files/paper/2023/hash/70b8505ac79e3e131756f793cd80eb8d-Abstract-Conference.html) (NeurIPS 2023): prerequisite and ordered skill acquisition. +- [DoReMi](https://proceedings.neurips.cc/paper_files/paper/2023/hash/dcba6be91359358c2355cd920da3fcbd-Abstract-Conference.html) (NeurIPS 2023): adaptive data mixtures instead of raw-frequency mixing. +- [LESS](https://proceedings.mlr.press/v235/xia24c.html) (ICML 2024): targeted data selection and empirical influence/utility. + +## Model Request Timeout + +Analyzer model requests use a 300-second client timeout by default. Conclusion +requests stream response chunks so long output does not need to wait for the +entire completion before the connection becomes active. A provider/proxy `524` +may still enforce its own shorter gateway limit; in that case Analyzer records +the elapsed time and prompt length, then retries once with compact evidence. + ## Stream Runtime Analyzer standalone follows the same base event writer style as Judger: @@ -158,12 +229,6 @@ Analyzer standalone follows the same base event writer style as Judger: from loopai.common.event_tool import StreamEvent, get_event_writer ``` -Analyzer-specific stdout/state-message/redaction helpers live in: - -```text -loopai/skills/Analyzer/event_tool.py -``` - Events are written to: ```text