实现动态压缩输入以及分阶段保存

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",

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]]:

loopai/skills/Analyzer/nodes/draw_conclusion.py

@@ -2,6 +2,7 @@
 import json
 import time
 import datetime
+import threading
 from pathlib import Path
 from collections import Counter
 from typing import List, Dict, Any
@@ -134,6 +135,62 @@ def init_model(state: LoopAIState) -> ChatOpenAI:
     return model
 
 
+def _batch_one_with_heartbeat(
+    llm: ChatOpenAI,
+    prompt: str,
+    *,
+    emit,
+    message: str,
+    start_progress: float,
+    end_progress: float,
+    data: Dict[str, Any] | None = None,
+) -> str:
+    """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)
+
+    try:
+        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):
     """
     尝试读取 OJ 记录文件
@@ -603,7 +660,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,11 +755,16 @@ 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 = ""
     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
@@ -736,7 +802,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 +835,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 = ""
@@ -782,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

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,23 +131,34 @@ 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")
         or (state.get("version_id") if isinstance(state, dict) else None)
         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"]