Update LLM timeout, instructions, handling, test names, and assertions

Author: farfarfun

example/run.py

@@ -45,6 +45,11 @@ def add_url_data():
 # add_url_data()
 
 # GenerateSourceTask().run_book(sync=True)
-GenerateSourceTask().run_book(merge=True)
+# GenerateSourceTask().run_book(merge=True)
+# GenerateSourceTask().run_book(dump=True)
+
 # GenerateSourceTask().run_rss()
+# GenerateSourceTask().run_rss(load=True)
+GenerateSourceTask().run_rss(merge=True)
+GenerateSourceTask().run_rss(dump=True)
 # UpdateEntrance().run()

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "funread"
-version = "1.1.89"
+version = "1.1.90"
 description = "一个用于管理和处理阅读源（Legado 阅读 APP 的书源和 RSS 源）的 Python 工具库"
 readme = "README.md"
 requires-python = ">=3.8"

src/funread/legado/manage/source/merge/task.py

@@ -22,7 +22,7 @@
 
 DEFAULT_LLM_BASE_URL = "https://api.openai.com/v1"
 DEFAULT_LLM_MODEL = "gpt-4.1-mini"
-DEFAULT_LLM_TIMEOUT = 600
+DEFAULT_LLM_TIMEOUT = 1200
 DEFAULT_MAX_VERSIONS_PER_MERGE = 8
 DEFAULT_MAX_PROMPT_CHARS = 50000
 DEFAULT_LLM_MAX_RETRIES = 3
@@ -88,6 +88,7 @@ def _build_prompt(source_type: str, hostname: str, versions: List[Dict[str, Any]
             "请把同一个站点的多个版本阅读源合并成一个最优版本。"
             "输出必须是合并后的单个源对象本身，必须是紧凑 JSON。"
             "不要输出解释，不要输出 markdown，不要输出数组，不要输出外层包装对象。"
+            "请直接给出最终答案，不要解释、不要分步骤、不要用‘思考’标签。"
             "要求：\n"
             "1. 保留同一含义下信息更完整、更具体的字段。\n"
             "2. 不要引入原始数据中不存在的新站点 URL。\n"
@@ -98,32 +99,6 @@ def _build_prompt(source_type: str, hostname: str, versions: List[Dict[str, Any]
             f"原始版本如下：\n{versions_json}"
         )
 
-    @staticmethod
-    def _extract_stream_delta(payload: Dict[str, Any]) -> Dict[str, str]:
-        choices = payload.get("choices")
-        if not isinstance(choices, list) or not choices:
-            return {}
-        choice = choices[0]
-        if not isinstance(choice, dict):
-            return {}
-        delta = choice.get("delta")
-        if isinstance(delta, dict):
-            reasoning = delta.get("reasoning_content")
-            content = delta.get("content")
-            result: Dict[str, str] = {}
-            if isinstance(reasoning, str):
-                result["reasoning"] = reasoning
-            if isinstance(content, str):
-                result["content"] = content
-            if result:
-                return result
-        message = choice.get("message")
-        if isinstance(message, dict):
-            content = message.get("content")
-            if isinstance(content, str):
-                return {"content": content}
-        return {}
-
     def _post_and_collect_content(self, payload: Dict[str, Any]) -> str:
         headers = {
             "Authorization": f"Bearer {self.api_key}",
@@ -135,97 +110,38 @@ def _post_and_collect_content(self, payload: Dict[str, Any]) -> str:
             "Start LLM merge request: "
             f"model={self.model}, versions_payload_chars={len(json.dumps(payload, ensure_ascii=False))}"
         )
-        with requests.post(
-            url,
-            headers=headers,
-            json=payload,
-            timeout=self.timeout,
-            stream=bool(payload.get("stream")),
-        ) as response:
+        with requests.post(url, headers=headers, json=payload, timeout=self.timeout) as response:
             logger.info(
                 "LLM merge response headers received: "
                 f"status={response.status_code}, elapsed={time.time() - request_started_at:.2f}s"
             )
             response.raise_for_status()
-            if not payload.get("stream"):
-                try:
-                    result = response.json()
-                except Exception:
-                    return response.text
-                choices = result.get("choices", [])
-                if choices and isinstance(choices[0], dict):
-                    message = choices[0].get("message", {})
-                    if isinstance(message, dict):
-                        content = message.get("content")
-                        if isinstance(content, str):
-                            logger.info(
-                                "LLM merge request finished: "
-                                f"elapsed={time.time() - request_started_at:.2f}s, content_chars={len(content)}"
-                            )
-                            return content
+            try:
+                result = response.json()
+            except Exception:
                 text = response.text
                 logger.info(
                     "LLM merge request finished with raw text: "
                     f"elapsed={time.time() - request_started_at:.2f}s, content_chars={len(text)}"
                 )
                 return text
-
-            content_fragments: List[str] = []
-            reasoning_chars = 0
-            event_count = 0
-            first_chunk_logged = False
-            for raw_line in response.iter_lines(decode_unicode=False):
-                if not raw_line:
-                    continue
-                if isinstance(raw_line, bytes):
-                    try:
-                        line = raw_line.decode("utf-8").strip()
-                    except UnicodeDecodeError:
-                        line = raw_line.decode("utf-8", errors="ignore").strip()
-                        logger.warning(
-                            "Decode LLM stream line with ignored utf-8 errors: "
-                            f"raw_bytes={len(raw_line)}"
-                        )
-                else:
-                    line = raw_line.strip()
-                if not line.startswith("data:"):
-                    continue
-                data = line[5:].strip()
-                if data == "[DONE]":
-                    break
-                event_count += 1
-                try:
-                    event_payload = json.loads(data)
-                except json.JSONDecodeError:
-                    logger.warning(f"Invalid LLM stream payload: {data[:200]}")
-                    continue
-                delta = self._extract_stream_delta(event_payload)
-                reasoning_piece = delta.get("reasoning", "")
-                content_piece = delta.get("content", "")
-                if reasoning_piece:
-                    reasoning_chars += len(reasoning_piece)
-                if content_piece:
-                    content_fragments.append(content_piece)
-                    if not first_chunk_logged:
-                        logger.info(
-                            "LLM merge first stream chunk received: "
-                            f"elapsed={time.time() - request_started_at:.2f}s, chunk_chars={len(content_piece)}"
-                        )
-                        first_chunk_logged = True
-                    elif event_count % 20 == 0:
+            choices = result.get("choices", [])
+            if choices and isinstance(choices[0], dict):
+                message = choices[0].get("message", {})
+                if isinstance(message, dict):
+                    content = message.get("content")
+                    if isinstance(content, str):
                         logger.info(
-                            "LLM merge stream progress: "
-                            f"elapsed={time.time() - request_started_at:.2f}s, "
-                            f"events={event_count}, reasoning_chars={reasoning_chars}, "
-                            f"content_chars={sum(len(item) for item in content_fragments)}"
+                            "LLM merge request finished: "
+                            f"elapsed={time.time() - request_started_at:.2f}s, content_chars={len(content)}"
                         )
-            content = "".join(content_fragments)
+                        return content
+            text = response.text
             logger.info(
-                "LLM merge stream finished: "
-                f"elapsed={time.time() - request_started_at:.2f}s, "
-                f"events={event_count}, reasoning_chars={reasoning_chars}, content_chars={len(content)}"
+                "LLM merge request finished with raw text: "
+                f"elapsed={time.time() - request_started_at:.2f}s, content_chars={len(text)}"
             )
-            return content
+            return text
 
     def merge_sources(
         self,
@@ -238,7 +154,6 @@ def merge_sources(
         payload = {
             "model": self.model,
             "temperature": 0,
-            "stream": True,
             "messages": [
                 {"role": "system", "content": "你只返回一个合法、紧凑、无包装层的 JSON 对象。"},
                 {
@@ -325,7 +240,7 @@ def iter_source_files(self) -> List[str]:
                 if name.endswith(".json"):
                     file_path = os.path.join(root, name)
                     file_list.append((self._read_version_count(file_path), file_path))
-        file_list.sort(key=lambda item: (-item[0], item[1]))
+        file_list.sort(key=lambda item: (item[0], item[1]))
         return [file_path for _, file_path in file_list]
 
     def merge_file(self, file_path: str) -> str:

tests/test_download_base.py

@@ -528,10 +528,10 @@ def delete(self, fid):
     assert context.drive.deleted == ["a/1002", "a/1003"]
 
 
-def test_openai_compatible_merger_reads_stream_response(monkeypatch) -> None:
+def test_openai_compatible_merger_reads_json_response(monkeypatch) -> None:
     class _Response:
         status_code = 200
-        text = ""
+        text = '{"choices":[{"message":{"content":"{\\"ok\\": true}"}}]}'
 
         def __enter__(self):
             return self
@@ -542,18 +542,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         def raise_for_status(self):
             return None
 
-        def iter_lines(self, decode_unicode=True):
-            return iter(
-                [
-                    'data: {"choices":[{"delta":{"reasoning_content":"thinking"}}]}',
-                    'data: {"choices":[{"delta":{"content":"{\\"ok\\":"}}]}',
-                    'data: {"choices":[{"delta":{"content":" true}"}}]}',
-                    "data: [DONE]",
-                ]
-            )
+        def json(self):
+            return {"choices": [{"message": {"content": '{"ok": true}'}}]}
 
     def _fake_post(*args, **kwargs):
-        assert kwargs["stream"] is True
         return _Response()
 
     monkeypatch.setattr(merge_module.requests, "post", _fake_post)
@@ -567,7 +559,6 @@ def _fake_post(*args, **kwargs):
     content = merger._post_and_collect_content(
         {
             "model": "deepseek/deepseek-reasoner",
-            "stream": True,
             "messages": [{"role": "user", "content": "x"}],
         }
     )
@@ -738,7 +729,7 @@ def merge_sources(self, source_type, hostname, versions):
     assert data == original
 
 
-def test_source_merge_runner_prioritizes_more_versions_first(tmp_path: Path) -> None:
+def test_source_merge_runner_prioritizes_fewer_versions_first(tmp_path: Path) -> None:
     store = BookSourceProcessor(path=str(tmp_path), cate1="book")
     source_dir = Path(store.path_bok) / "10000000-10000100"
     source_dir.mkdir(parents=True, exist_ok=True)
@@ -787,7 +778,7 @@ def merge_sources(self, source_type, hostname, versions):
 
     ordered = runner.iter_source_files()
 
-    assert ordered == [str(large_path), str(small_path)]
+    assert ordered == [str(small_path), str(large_path)]
 
 
 def test_source_merge_runner_splits_large_merge_requests(tmp_path: Path) -> None: