From 84e212ae39290f06ab70924c371e23ff75378701 Mon Sep 17 00:00:00 2001 From: Justin Date: Thu, 20 Aug 2026 20:56:42 +0800 Subject: [PATCH 1/3] Fix Analyzer resume version handling and conclusion heartbeats --- loopai/skills/Analyzer/__init__.py | 2 + .../nodes/analyze_metric_report_node.py | 9 +-- .../skills/Analyzer/nodes/analyze_result.py | 11 +-- .../skills/Analyzer/nodes/draw_conclusion.py | 67 ++++++++++++++++--- loopai/skills/Analyzer/runner.py | 47 +++++++++---- 5 files changed, 105 insertions(+), 31 deletions(-) diff --git a/loopai/skills/Analyzer/__init__.py b/loopai/skills/Analyzer/__init__.py index bbfd34e..6d87b6a 100644 --- a/loopai/skills/Analyzer/__init__.py +++ b/loopai/skills/Analyzer/__init__.py @@ -153,6 +153,8 @@ def run( for control_key in ( "version_id", "run_id", + "new_version", + "force_new_version", "resume", "from_node", "checkpoint_path", diff --git a/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py b/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py index d84443c..2e964af 100644 --- a/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py +++ b/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py @@ -8,7 +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 langchain_openai import ChatOpenAI +from loopai.skills.Analyzer.utils.openai_compat_llm import OpenAICompatChat from loopai.schema.states import LoopAIState from loopai.logger import get_logger @@ -62,16 +62,17 @@ def _runtime_api_key(cfg: dict) -> str: ) -def init_model(state: LoopAIState) -> ChatOpenAI: +def init_model(state: LoopAIState) -> OpenAICompatChat: """ 初始化分析用模型。 使用 OpenAI-compatible / vLLM 风格接口。 """ cfg = _analyzer(state) - model = ChatOpenAI( + model = OpenAICompatChat( model=cfg["analyze_model_path"], - api_key=_runtime_api_key(cfg), base_url=cfg.get("analyze_base_url"), + api_key=_runtime_api_key(cfg), + max_tokens=int(cfg.get("analyze_max_tokens", 512) or 512), temperature=cfg.get("analyze_temperature", 0.0), top_p=cfg.get("analyze_top_p", 0.95), ) diff --git a/loopai/skills/Analyzer/nodes/analyze_result.py b/loopai/skills/Analyzer/nodes/analyze_result.py index 9776e5e..110a181 100644 --- a/loopai/skills/Analyzer/nodes/analyze_result.py +++ b/loopai/skills/Analyzer/nodes/analyze_result.py @@ -9,7 +9,7 @@ from loopai.skills.Analyzer.utils.stream import get_safe_stream_writer from loopai.skills.Analyzer.utils.stream import get_analyzer_resume_progress from loopai.common.prompts.prompt_loader import PromptLoader -from langchain_openai import ChatOpenAI +from loopai.skills.Analyzer.utils.openai_compat_llm import OpenAICompatChat from loopai.schema.states import LoopAIState from loopai.logger import get_logger from loopai.skills.Analyzer.history_compare import ( @@ -46,16 +46,17 @@ def _runtime_api_key(cfg: dict) -> str: or "EMPTY" ) -def init_model(state: LoopAIState) -> ChatOpenAI: +def init_model(state: LoopAIState) -> OpenAICompatChat: """ 使用标准 vLLM(OpenAI 兼容) 客户端 """ cfg = _analyzer(state) - model = ChatOpenAI( + model = OpenAICompatChat( model=cfg['analyze_model_path'], - api_key=_runtime_api_key(cfg), base_url=cfg['analyze_base_url'], + api_key=_runtime_api_key(cfg), + max_tokens=int(cfg.get("analyze_max_tokens", 512) or 512), temperature=cfg.get('analyze_temperature', 0.0), top_p=cfg.get('analyze_top_p', 0.95), ) @@ -63,7 +64,7 @@ def init_model(state: LoopAIState) -> ChatOpenAI: def _batch_one_with_heartbeat( - llm: ChatOpenAI, + llm: OpenAICompatChat, prompt: str, *, emit, diff --git a/loopai/skills/Analyzer/nodes/draw_conclusion.py b/loopai/skills/Analyzer/nodes/draw_conclusion.py index 14206b4..40647ff 100644 --- a/loopai/skills/Analyzer/nodes/draw_conclusion.py +++ b/loopai/skills/Analyzer/nodes/draw_conclusion.py @@ -2,11 +2,12 @@ import json import time import datetime +import threading from pathlib import Path from collections import Counter from typing import List, Dict, Any -from langchain_openai import ChatOpenAI +from loopai.skills.Analyzer.utils.openai_compat_llm import OpenAICompatChat from loopai.schema.states import LoopAIState from loopai.logger import get_logger @@ -110,7 +111,7 @@ def _runtime_api_key(cfg: dict) -> str: or "EMPTY" ) -def init_model(state: LoopAIState) -> ChatOpenAI: +def init_model(state: LoopAIState) -> OpenAICompatChat: """ 初始化模型 Args: @@ -124,16 +125,50 @@ def init_model(state: LoopAIState) -> ChatOpenAI: """ cfg = _analyzer(state) - model = ChatOpenAI( + model = OpenAICompatChat( model=cfg['analyze_model_path'], - api_key=_runtime_api_key(cfg), base_url=cfg['analyze_base_url'], + api_key=_runtime_api_key(cfg), + max_tokens=int(cfg.get("analyze_max_tokens", 512) or 512), temperature=cfg.get('analyze_temperature', 0.0), top_p=cfg.get('analyze_top_p', 0.95), ) return model +def _batch_one_with_heartbeat( + llm: OpenAICompatChat, + prompt: str, + *, + emit, + message: str, + start_progress: float, + end_progress: float, + data: Dict[str, Any] | None = None, +) -> str: + """Run one conclusion request while keeping UI progress alive.""" + stop_event = threading.Event() + + def heartbeat() -> None: + tick = 0 + while not stop_event.wait(1.0): + tick += 1 + progress = start_progress + (end_progress - start_progress) * min(0.85, tick / 20) + emit( + message, + progress=round(progress, 3), + data={**(data or {}), "waiting_seconds": tick, "heartbeat": True}, + ) + + thread = threading.Thread(target=heartbeat, daemon=True) + thread.start() + try: + return llm.batch([prompt])[0].content + finally: + stop_event.set() + thread.join(timeout=0.2) + + def try_read_oj_records(path_from_summary: str): """ 尝试读取 OJ 记录文件 @@ -603,7 +638,11 @@ def _emit(message, *, progress=None, data=None): llm = init_model(state) try: obtainer_prompt = build_obtainer_prompt(final_json, obtainer_stats) - obtainer_text = llm.batch([obtainer_prompt])[0].content + obtainer_text = _batch_one_with_heartbeat( + llm, obtainer_prompt, emit=_emit, + message="等待生成 Obtainer 报告", + start_progress=0.80, end_progress=0.95, + ) except Exception as exc: logger.error(f"生成 obtainer 报告时出错：{exc}") obtainer_text = "" @@ -694,7 +733,11 @@ def _emit(message, *, progress=None, data=None): logger.info("🤖 正在生成背景介绍……") try: bg_prompt = build_background_prompt(final_json) - background_text = llm.batch([bg_prompt])[0].content + background_text = _batch_one_with_heartbeat( + llm, bg_prompt, emit=_emit, + message="等待模型生成背景介绍", + start_progress=0.30, end_progress=0.38, + ) except Exception as e: logger.error(f"生成背景介绍时出错：{e}") background_text = "" @@ -736,7 +779,11 @@ def _emit(message, *, progress=None, data=None): logger.info("🤖 正在调用本地模型生成改进建议……") prompt = build_suggestion_prompt(final_json) try: - suggestion = llm.batch([prompt])[0].content + suggestion = _batch_one_with_heartbeat( + llm, prompt, emit=_emit, + message="等待模型生成改进建议", + start_progress=0.60, end_progress=0.70, + ) except Exception as e: logger.error(f"生成改进建议时出错：{e}") suggestion = "" @@ -765,7 +812,11 @@ def _emit(message, *, progress=None, data=None): logger.info("🤖 正在生成 obtainer 细粒度报告……") try: obtainer_prompt = build_obtainer_prompt(final_json, obtainer_stats) - obtainer_text = llm.batch([obtainer_prompt])[0].content + obtainer_text = _batch_one_with_heartbeat( + llm, obtainer_prompt, emit=_emit, + message="等待生成 Obtainer 报告", + start_progress=0.80, end_progress=0.95, + ) except Exception as e: logger.error(f"生成 obtainer 报告时出错：{e}") obtainer_text = "" diff --git a/loopai/skills/Analyzer/runner.py b/loopai/skills/Analyzer/runner.py index 981d592..4fc272a 100644 --- a/loopai/skills/Analyzer/runner.py +++ b/loopai/skills/Analyzer/runner.py @@ -120,6 +120,8 @@ def run_analyzer_standalone( checkpoint_path: Optional[str] = None, baseline_result_path: Optional[str] = None, analyze_batch_size: Optional[int] = None, + version_id: Optional[str] = None, + force_new_version: bool = False, emit_status: bool = True, **kwargs: Any, ) -> Dict[str, Any]: @@ -129,15 +131,24 @@ def run_analyzer_standalone( Runtime config is resolved in the skill layer so Codex/CLI can inject environment values before executing or resuming Analyzer steps. """ + runtime_kwargs = dict(kwargs) + for control_key in ("version_id", "run_id", "new_version", "force_new_version"): + runtime_kwargs.pop(control_key, None) runtime = resolve_analyzer_runtime_config( state, thread_id=thread_id, checkpoint_path=checkpoint_path, - **kwargs, + version_id=version_id, + **runtime_kwargs, ) + # A CLI checkpoint path is authoritative. In particular, resume may + # provide a version_id from the environment; do not replace the explicit + # SQLite path with the default output-derived path in that case. + explicit_checkpoint_path = bool(checkpoint_path) explicit_version = ( - kwargs.get("version_id") + version_id + or kwargs.get("version_id") or kwargs.get("run_id") or os.getenv("ANALYZER_VERSION_ID") or os.getenv("VERSION_ID") @@ -145,7 +156,9 @@ def run_analyzer_standalone( or ((state.get("analyzer") or {}).get("version_id") if isinstance(state, dict) else None) ) force_new_version = bool( - kwargs.get("new_version") or kwargs.get("force_new_version") + force_new_version + or kwargs.get("new_version") + or kwargs.get("force_new_version") ) if not force_new_version and not runtime.get("version_id"): @@ -171,9 +184,10 @@ def run_analyzer_standalone( runtime["version_id"] = "" if not force_new_version and runtime.get("version_id"): - runtime["checkpoint_path"] = get_version_checkpoint_path( - runtime["output_dir"], runtime["thread_id"], runtime["version_id"] - ) + if not explicit_checkpoint_path: + runtime["checkpoint_path"] = get_version_checkpoint_path( + runtime["output_dir"], runtime["thread_id"], runtime["version_id"] + ) if os.path.exists(runtime["checkpoint_path"]): candidate_state = load_analyzer_checkpoint( runtime["thread_id"], @@ -198,13 +212,17 @@ def run_analyzer_standalone( runtime["checkpoint_path"], version_id=runtime["version_id"], ) - resolve_analyzer_runtime_config( - state, + # Merge runtime values once so a caller-provided version_id cannot be + # passed twice through the explicit arguments and **kwargs. + resume_runtime_kwargs = dict(kwargs) + for control_key in ("version_id", "run_id", "new_version", "force_new_version"): + resume_runtime_kwargs.pop(control_key, None) + resume_runtime_kwargs.update( thread_id=runtime["thread_id"], checkpoint_path=runtime["checkpoint_path"], version_id=runtime["version_id"], - **kwargs, ) + resolve_analyzer_runtime_config(state, **resume_runtime_kwargs) if start_node is None: start_node = _resume_step_from_state(state) @@ -248,11 +266,12 @@ def run_analyzer_standalone( runtime["thread_id"], runtime["version_id"], ) - runtime["checkpoint_path"] = get_version_checkpoint_path( - output_dir, - runtime["thread_id"], - runtime["version_id"], - ) + if not explicit_checkpoint_path: + runtime["checkpoint_path"] = get_version_checkpoint_path( + output_dir, + runtime["thread_id"], + runtime["version_id"], + ) state["version_id"] = runtime["version_id"] state.setdefault("analyzer", {})["version_id"] = runtime["version_id"] state["analyzer"]["checkpoint_path"] = runtime["checkpoint_path"] From 2dfd5188c6ac15dd1ef3daa857b83f678482b855 Mon Sep 17 00:00:00 2001 From: Justin Date: Thu, 20 Aug 2026 21:02:45 +0800 Subject: [PATCH 2/3] Preserve Analyzer model output limits --- .../Analyzer/nodes/analyze_metric_report_node.py | 9 ++++----- loopai/skills/Analyzer/nodes/analyze_result.py | 11 +++++------ loopai/skills/Analyzer/nodes/draw_conclusion.py | 11 +++++------ 3 files changed, 14 insertions(+), 17 deletions(-) diff --git a/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py b/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py index 2e964af..d84443c 100644 --- a/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py +++ b/loopai/skills/Analyzer/nodes/analyze_metric_report_node.py @@ -8,7 +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.utils.openai_compat_llm import OpenAICompatChat +from langchain_openai import ChatOpenAI from loopai.schema.states import LoopAIState from loopai.logger import get_logger @@ -62,17 +62,16 @@ def _runtime_api_key(cfg: dict) -> str: ) -def init_model(state: LoopAIState) -> OpenAICompatChat: +def init_model(state: LoopAIState) -> ChatOpenAI: """ 初始化分析用模型。 使用 OpenAI-compatible / vLLM 风格接口。 """ cfg = _analyzer(state) - model = OpenAICompatChat( + model = ChatOpenAI( model=cfg["analyze_model_path"], - base_url=cfg.get("analyze_base_url"), api_key=_runtime_api_key(cfg), - max_tokens=int(cfg.get("analyze_max_tokens", 512) or 512), + base_url=cfg.get("analyze_base_url"), temperature=cfg.get("analyze_temperature", 0.0), top_p=cfg.get("analyze_top_p", 0.95), ) diff --git a/loopai/skills/Analyzer/nodes/analyze_result.py b/loopai/skills/Analyzer/nodes/analyze_result.py index 110a181..9776e5e 100644 --- a/loopai/skills/Analyzer/nodes/analyze_result.py +++ b/loopai/skills/Analyzer/nodes/analyze_result.py @@ -9,7 +9,7 @@ from loopai.skills.Analyzer.utils.stream import get_safe_stream_writer from loopai.skills.Analyzer.utils.stream import get_analyzer_resume_progress from loopai.common.prompts.prompt_loader import PromptLoader -from loopai.skills.Analyzer.utils.openai_compat_llm import OpenAICompatChat +from langchain_openai import ChatOpenAI from loopai.schema.states import LoopAIState from loopai.logger import get_logger from loopai.skills.Analyzer.history_compare import ( @@ -46,17 +46,16 @@ def _runtime_api_key(cfg: dict) -> str: or "EMPTY" ) -def init_model(state: LoopAIState) -> OpenAICompatChat: +def init_model(state: LoopAIState) -> ChatOpenAI: """ 使用标准 vLLM(OpenAI 兼容) 客户端 """ cfg = _analyzer(state) - model = OpenAICompatChat( + model = ChatOpenAI( model=cfg['analyze_model_path'], - base_url=cfg['analyze_base_url'], api_key=_runtime_api_key(cfg), - max_tokens=int(cfg.get("analyze_max_tokens", 512) or 512), + base_url=cfg['analyze_base_url'], temperature=cfg.get('analyze_temperature', 0.0), top_p=cfg.get('analyze_top_p', 0.95), ) @@ -64,7 +63,7 @@ def init_model(state: LoopAIState) -> OpenAICompatChat: def _batch_one_with_heartbeat( - llm: OpenAICompatChat, + llm: ChatOpenAI, prompt: str, *, emit, diff --git a/loopai/skills/Analyzer/nodes/draw_conclusion.py b/loopai/skills/Analyzer/nodes/draw_conclusion.py index 40647ff..1acb0ff 100644 --- a/loopai/skills/Analyzer/nodes/draw_conclusion.py +++ b/loopai/skills/Analyzer/nodes/draw_conclusion.py @@ -7,7 +7,7 @@ from collections import Counter from typing import List, Dict, Any -from loopai.skills.Analyzer.utils.openai_compat_llm import OpenAICompatChat +from langchain_openai import ChatOpenAI from loopai.schema.states import LoopAIState from loopai.logger import get_logger @@ -111,7 +111,7 @@ def _runtime_api_key(cfg: dict) -> str: or "EMPTY" ) -def init_model(state: LoopAIState) -> OpenAICompatChat: +def init_model(state: LoopAIState) -> ChatOpenAI: """ 初始化模型 Args: @@ -125,11 +125,10 @@ def init_model(state: LoopAIState) -> OpenAICompatChat: """ cfg = _analyzer(state) - model = OpenAICompatChat( + model = ChatOpenAI( model=cfg['analyze_model_path'], - base_url=cfg['analyze_base_url'], api_key=_runtime_api_key(cfg), - max_tokens=int(cfg.get("analyze_max_tokens", 512) or 512), + base_url=cfg['analyze_base_url'], temperature=cfg.get('analyze_temperature', 0.0), top_p=cfg.get('analyze_top_p', 0.95), ) @@ -137,7 +136,7 @@ def init_model(state: LoopAIState) -> OpenAICompatChat: def _batch_one_with_heartbeat( - llm: OpenAICompatChat, + llm: ChatOpenAI, prompt: str, *, emit, From 5a7ad7028e203d134b86437945abf51cb22c4753 Mon Sep 17 00:00:00 2001 From: Justin Date: Thu, 20 Aug 2026 21:28:22 +0800 Subject: [PATCH 3/3] Add adaptive Analyzer prompt retry --- .../skills/Analyzer/nodes/analyze_result.py | 73 ++++++++++++------- .../skills/Analyzer/nodes/draw_conclusion.py | 64 +++++++++++----- 2 files changed, 91 insertions(+), 46 deletions(-) diff --git a/loopai/skills/Analyzer/nodes/analyze_result.py b/loopai/skills/Analyzer/nodes/analyze_result.py index 9776e5e..3567cec 100644 --- a/loopai/skills/Analyzer/nodes/analyze_result.py +++ b/loopai/skills/Analyzer/nodes/analyze_result.py @@ -72,34 +72,53 @@ def _batch_one_with_heartbeat( end_progress: float, data: Dict[str, Any] | None = None, ) -> str: - resume_progress = get_analyzer_resume_progress() - if resume_progress: - start_progress = max(start_progress, min(resume_progress, end_progress)) - stop_event = threading.Event() - - def _heartbeat() -> None: - tick = 0 - while not stop_event.wait(1.0): - tick += 1 - wait_fraction = min(0.85, tick / 20) - progress = start_progress + (end_progress - start_progress) * wait_fraction - emit( - message, - progress=round(progress, 3), - data={ - **(data or {}), - "waiting_seconds": tick, - "heartbeat": True, - }, - ) - - heartbeat_thread = threading.Thread(target=_heartbeat, daemon=True) - heartbeat_thread.start() + def _is_timeout(exc: Exception) -> bool: + text = f"{type(exc).__name__}: {exc}".lower() + return isinstance(exc, TimeoutError) or any( + marker in text + for marker in ("timeout", "timed out", "read timed out", "524") + ) + + def _compact_prompt(value: str) -> str: + max_chars = 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 _run_once(request_prompt: str) -> str: + stop_event = threading.Event() + resume_progress = get_analyzer_resume_progress() + request_start = max(start_progress, min(resume_progress, end_progress)) if resume_progress else start_progress + + def _heartbeat() -> None: + tick = 0 + while not stop_event.wait(1.0): + tick += 1 + wait_fraction = min(0.85, tick / 20) + progress = request_start + (end_progress - request_start) * wait_fraction + emit(message, progress=round(progress, 3), data={ + **(data or {}), "waiting_seconds": tick, "heartbeat": True, + }) + + heartbeat_thread = threading.Thread(target=_heartbeat, daemon=True) + heartbeat_thread.start() + try: + return llm.batch([request_prompt])[0].content + finally: + stop_event.set() + heartbeat_thread.join(timeout=0.2) + try: - return llm.batch([prompt])[0].content - finally: - stop_event.set() - heartbeat_thread.join(timeout=0.2) + return _run_once(prompt) + except Exception as exc: + if not _is_timeout(exc): + raise + emit("模型请求超时，压缩输入后重试", progress=start_progress, data={ + **(data or {}), "retry": True, "input_compacted": True, + }) + return _run_once(_compact_prompt(prompt)) def pick_failure_examples(oj_records: List[Dict[str, Any]], top_k: int = 5) -> List[Dict[str, Any]]: diff --git a/loopai/skills/Analyzer/nodes/draw_conclusion.py b/loopai/skills/Analyzer/nodes/draw_conclusion.py index 1acb0ff..d4bda8e 100644 --- a/loopai/skills/Analyzer/nodes/draw_conclusion.py +++ b/loopai/skills/Analyzer/nodes/draw_conclusion.py @@ -145,27 +145,50 @@ def _batch_one_with_heartbeat( end_progress: float, data: Dict[str, Any] | None = None, ) -> str: - """Run one conclusion request while keeping UI progress alive.""" - stop_event = threading.Event() - - def heartbeat() -> None: - tick = 0 - while not stop_event.wait(1.0): - tick += 1 - progress = start_progress + (end_progress - start_progress) * min(0.85, tick / 20) - emit( - message, - progress=round(progress, 3), - data={**(data or {}), "waiting_seconds": tick, "heartbeat": True}, - ) + """Run normally, then retry once with a compact prompt after timeout.""" + def _is_timeout(exc: Exception) -> bool: + text = f"{type(exc).__name__}: {exc}".lower() + return isinstance(exc, TimeoutError) or any( + marker in text for marker in ("timeout", "timed out", "read timed out", "524") + ) + + def _compact_prompt(value: str) -> str: + max_chars = 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 _run_once(request_prompt: str) -> str: + stop_event = threading.Event() + + def heartbeat() -> None: + tick = 0 + while not stop_event.wait(1.0): + tick += 1 + progress = start_progress + (end_progress - start_progress) * min(0.85, tick / 20) + emit(message, progress=round(progress, 3), data={ + **(data or {}), "waiting_seconds": tick, "heartbeat": True, + }) + + thread = threading.Thread(target=heartbeat, daemon=True) + thread.start() + try: + return llm.batch([request_prompt])[0].content + finally: + stop_event.set() + thread.join(timeout=0.2) - thread = threading.Thread(target=heartbeat, daemon=True) - thread.start() try: - return llm.batch([prompt])[0].content - finally: - stop_event.set() - thread.join(timeout=0.2) + return _run_once(prompt) + except Exception as exc: + if not _is_timeout(exc): + raise + emit("模型请求超时，压缩输入后重试", progress=start_progress, data={ + **(data or {}), "retry": True, "input_compacted": True, + }) + return _run_once(_compact_prompt(prompt)) def try_read_oj_records(path_from_summary: str): @@ -741,6 +764,7 @@ def _emit(message, *, progress=None, data=None): logger.error(f"生成背景介绍时出错：{e}") background_text = "" final_json["background"] = background_text + _emit("背景介绍生成完成", progress=0.38) obtainer_stats = build_obtainer_stats(summary, oj_records, final_json) final_json["obtainer_stats"] = obtainer_stats @@ -832,5 +856,7 @@ def _emit(message, *, progress=None, data=None): logger.info(f"→ {obtainer_json_path}") logger.info(f"→ {obtainer_txt_path}") + _emit("Obtainer 报告生成完成", progress=0.95) + _emit("最终报告生成完成", progress=1.0) return state