Update core/pipeline_agent.py

core/pipeline_agent.py

@@ -14,13 +14,9 @@ import logging
 import concurrent.futures
 from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type
 
-# 基础日志配置
+# 基础日志配置：自带时间戳和日志级别
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler(sys.stdout)])
 
-# 辅助函数：专门用于向 Hugging Face 容器输出强制刷新的彩色/醒目日志
-def hf_log(msg):
-    print(f"[HF Container Log] {msg}", flush=True)
-
 try:
     from langdetect import detect
     LANGDETECT_AVAILABLE = True
@@ -74,7 +70,7 @@ class SSciPipelineAgent:
         except: 
             self.tokenizer = tiktoken.get_encoding("gpt2")
 
-        hf_log("✅ SSciPipelineAgent 初始化完成。三维检索节点与大语言模型已就绪。")
+        logging.info("✅ SSciPipelineAgent 初始化完成。三维检索节点与大语言模型已就绪。")
 
     def _extract_json_from_text(self, text):
         if not text: 
@@ -100,7 +96,7 @@ class SSciPipelineAgent:
             else:
                 return json.loads(text[start_list:end_list+1])
         except Exception as e:
-            hf_log(f"❌ JSON 提取失败: {e}")
+            logging.error(f"❌ JSON 提取失败: {e}")
             return None
 
     def export_state(self):
@@ -124,7 +120,7 @@ class SSciPipelineAgent:
             self.citation_map = state_dict.get("citation_map", {})
             self.used_citations = set(state_dict.get("used_citations", []))
             self.global_memory = state_dict.get("global_memory", self.global_memory)
-        hf_log(f"🔄 状态恢复成功: DS API={self.ds_calls}, 现有文献={len(self.citation_map)}")
+        logging.info(f"🔄 状态恢复成功: DS API={self.ds_calls}, 现有文献={len(self.citation_map)}")
 
     def _calc_paragraph_count(self, target_words, lang_hint="en"):
         words_per_para = 100 if lang_hint in ["ru", "kk"] else 130
@@ -143,7 +139,7 @@ class SSciPipelineAgent:
         resp = requests.get(url, headers=headers, params=params, timeout=timeout)
         
         if resp.status_code == 429:
-            hf_log(f"⚠️ [API 限流] 触发 429 (Too Many Requests)。启动指数退避。URL: {url}")
+            logging.warning(f"⚠️ [API 限流] 触发 429 (Too Many Requests)。启动指数退避。URL: {url}")
             time.sleep(random.uniform(2.0, 5.0))
             resp.raise_for_status()
             
@@ -151,7 +147,7 @@ class SSciPipelineAgent:
             return resp
             
         if resp.status_code != 200:
-            hf_log(f"❌ [API 错误] 状态码: {resp.status_code} | URL: {url}")
+            logging.error(f"❌ [API 错误] 状态码: {resp.status_code} | URL: {url}")
             resp.raise_for_status()
         return resp
 
@@ -174,13 +170,13 @@ class SSciPipelineAgent:
             hypo_matches = re.findall(r'(H\d+[:\s].*?)(?=\n|$)', text, re.IGNORECASE)
             if hypo_matches: 
                 self.global_memory["hypotheses"] = "\n".join(hypo_matches[:10])
-                hf_log(f"🧠 [记忆更新] 从 {section_name} 中提取了假设 H1-Hn")
+                logging.info(f"🧠 [记忆更新] 从 {section_name} 中提取了假设 H1-Hn")
                 
         if any(kw in lower_sec for kw in ["method", "measure", "variable"]):
             var_matches = re.findall(r'([A-Z][a-z]+(?:\s+[A-Z][a-z]+)*)\s+(?:was|is|refers to)\s+([^\.]+\.)', text)
             if var_matches: 
                 self.global_memory["variables"] = "\n".join([f"- {v[0]}: {v[1][:100]}" for v in var_matches[:8]])
-                hf_log(f"🧠 [记忆更新] 从 {section_name} 中提取了核心变量定义")
+                logging.info(f"🧠 [记忆更新] 从 {section_name} 中提取了核心变量定义")
                 
         if "model" in lower_sec or "figure" in lower_sec:
             model_match = re.search(r'(Figure\s*\d+[^\.]+\.)', text, re.IGNORECASE)
@@ -261,7 +257,7 @@ class SSciPipelineAgent:
 
     def call_deepseek(self, system_prompt, user_prompt, max_tokens=8000, temp=0.3, model_name="deepseek-reasoner"):
         tokens_est = self._count_tokens(user_prompt)
-        hf_log(f"🚀 [DeepSeek API - 静态请求] 模型: {model_name} | MaxTokens: {max_tokens} | 输入估算: {tokens_est} tokens")
+        logging.info(f"🚀 [DeepSeek 静态请求] 模型: {model_name} | MaxTokens: {max_tokens} | 输入估算: {tokens_est} tokens")
         
         for attempt in range(3):
             with self.lock: 
@@ -270,7 +266,7 @@ class SSciPipelineAgent:
             if attempt == 2 and model_name == "deepseek-reasoner":
                 current_model = "deepseek-chat"
                 temp = 0.3 
-                hf_log("⚠️ 强制降级为 deepseek-chat 执行本节点！")
+                logging.warning("⚠️ 强制降级为 deepseek-chat 执行本节点！")
                 
             payload = {
                 "model": current_model, 
@@ -286,7 +282,7 @@ class SSciPipelineAgent:
             try:
                 with requests.post(self.deepseek_api_url, headers={"Authorization": f"Bearer {self.ds_key}"}, json=payload, stream=True, timeout=self.timeouts["standard"]) as response:
                     if response.status_code != 200:
-                        hf_log(f"❌ 尝试 {attempt+1}/3 失败: 状态码 {response.status_code} - {response.text[:200]}")
+                        logging.error(f"❌ 尝试 {attempt+1}/3 失败: 状态码 {response.status_code} - {response.text[:200]}")
                         if attempt == 2: 
                             raise requests.exceptions.HTTPError("API 熔断")
                         time.sleep(3 ** attempt)
@@ -303,17 +299,17 @@ class SSciPipelineAgent:
                                 pass
                 
                 elapsed = time.time() - start_time
-                hf_log(f"✅ [DeepSeek API - 返回] 耗时: {elapsed:.2f}s | 输出长度: {len(full_content)} 字符")
+                logging.info(f"✅ [静态请求返回] 耗时: {elapsed:.2f}s | 输出长度: {len(full_content)} 字符")
                 return full_content.strip()
             except Exception as e:
-                hf_log(f"❌ [网络异常] 尝试 {attempt+1}: {e}")
+                logging.error(f"❌ [网络异常] 尝试 {attempt+1}: {e}")
                 if attempt == 2: 
                     raise e
                 time.sleep(3 ** attempt)
 
     def stream_deepseek(self, system_prompt, user_prompt, max_tokens=8000, model_name="deepseek-reasoner"):
         tokens_est = self._count_tokens(user_prompt)
-        hf_log(f"🌊 [DeepSeek API - 流式生成] 模型: {model_name} | 输入估算: {tokens_est} tokens")
+        logging.info(f"🌊 [DeepSeek 流式请求] 模型: {model_name} | 输入估算: {tokens_est} tokens")
         
         temp = 0.3 if model_name == "deepseek-chat" else 0.2
         for attempt in range(3):
@@ -323,7 +319,7 @@ class SSciPipelineAgent:
             if attempt >= 1 and model_name == "deepseek-reasoner":
                 current_model = "deepseek-chat"
                 temp = 0.3  
-                hf_log("⚠️ 推理节点异常，强制降级至 deepseek-chat 并重试。")
+                logging.warning("⚠️ 推理节点异常，强制降级至 deepseek-chat 并重试。")
 
             payload = {
                 "model": current_model, 
@@ -334,14 +330,13 @@ class SSciPipelineAgent:
             if current_model != "deepseek-reasoner": 
                 payload["temperature"] = temp
                 
-            think_heartbeat = 0
             has_started_reasoning = False
             start_time = time.time()
             
             try:
                 with requests.post(self.deepseek_api_url, headers={"Authorization": f"Bearer {self.ds_key}"}, json=payload, stream=True, timeout=self.timeouts["stream"]) as response:
                     if response.status_code != 200:
-                        hf_log(f"❌ 流式尝试 {attempt+1}/3 失败: 状态码 {response.status_code}")
+                        logging.error(f"❌ 流式尝试 {attempt+1}/3 失败: 状态码 {response.status_code}")
                         if attempt == 2: 
                             response.raise_for_status()
                         time.sleep(2)
@@ -359,14 +354,13 @@ class SSciPipelineAgent:
                                     if not has_started_reasoning:
                                         yield "\n<think>\n"
                                         has_started_reasoning = True
-                                        hf_log(f"🤔 [R1 推理启动] 开始输出深度思考块...")
+                                        logging.info(f"🤔 [R1 推理启动] 开始在终端流式输出思考过程...")
                                         
-                                    think_heartbeat += 1
-                                    
-                                    # 针对 Hugging Face 容器底层的实时心跳探针，直接 flush 到 stdout
-                                    if think_heartbeat % 25 == 0:
-                                        elapsed = time.time() - start_time
-                                        hf_log(f"⏳ [DeepSeek 心跳] {current_model} 思考中... ({elapsed:.1f}秒, 接收 {think_heartbeat} 块)")
+                                    # ---------------------------------------------------------
+                                    # 🔥 核心修改点：直接向终端输出内容并强制刷新，实现无延迟的打字机效果
+                                    # ---------------------------------------------------------
+                                    sys.stdout.write(reasoning_content)
+                                    sys.stdout.flush()
                                     
                                     yield reasoning_content
                                     continue
@@ -375,19 +369,34 @@ class SSciPipelineAgent:
                                     yield "\n</think>\n\n"
                                     has_started_reasoning = False
                                     elapsed = time.time() - start_time
-                                    hf_log(f"💡 [R1 推理结束] 思考完成，总耗时 {elapsed:.1f}秒。开始流式输出正文...")
+                                    
+                                    # 思考结束，终端换行，并记录日志
+                                    sys.stdout.write("\n")
+                                    sys.stdout.flush()
+                                    logging.info(f"💡 [R1 推理结束] 耗时 {elapsed:.1f}秒。开始在终端流式输出正文...")
 
                                 if content: 
+                                    # 🔥 正文内容同样直接推送到终端流并强制刷新
+                                    sys.stdout.write(content)
+                                    sys.stdout.flush()
+                                    
                                     yield content
                                     
                             except Exception: 
                                 pass
-                hf_log(f"🏁 [流式生成结束] 总耗时 {(time.time() - start_time):.1f}秒。")
+                
+                # 整个流结束，终端换行
+                sys.stdout.write("\n")
+                sys.stdout.flush()
+                logging.info(f"🏁 [流式生成结束] 总耗时 {(time.time() - start_time):.1f}秒。")
                 return 
             except GeneratorExit: 
                 raise
             except Exception as e:
-                hf_log(f"❌ [流式中断]: {e}")
+                # 异常中断时也要确保换行，防止日志排版错乱
+                sys.stdout.write("\n")
+                sys.stdout.flush()
+                logging.error(f"❌ [流式中断]: {e}")
                 if attempt == 2: 
                     raise
                 time.sleep(2)
@@ -425,7 +434,7 @@ class SSciPipelineAgent:
 
         for attempt in range(2):
             try:
-                hf_log("📄 [本地解析] 正在榨取用户上传的核心文献...")
+                logging.info("📄 [本地解析] 正在榨取用户上传的核心文献...")
                 response = self.call_deepseek(sys_prompt, user_prompt, temp=0.1, model_name="deepseek-chat") 
                 papers = self._extract_json_from_text(response)
                 if not papers: 
@@ -467,7 +476,7 @@ class SSciPipelineAgent:
                         self.used_citations.add(cite_tag) 
                         
                     rag_context.append(f"{cite_tag} [USER CORE PAPER] Title: {title}. Core Thesis: {core_thesis}. Abstract: {abstract}")
-                hf_log(f"✅ [本地解析] 成功提取 {len(papers)} 篇本地文献元数据。")
+                logging.info(f"✅ [本地解析] 成功提取 {len(papers)} 篇本地文献元数据。")
                 return "\n".join(rag_context)
             except Exception as e: 
                 time.sleep(2)
@@ -514,7 +523,6 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
         raise ValueError("Outline Generation Failed.")
 
     def prefetch_jit_literature(self, topic, keywords, structure):
-        """Pre-fetch JIT: 全局并发预抓取所有方法与结果章节所需支撑文献"""
         jit_rag_texts = []
         queries_to_fetch = set()
         
@@ -529,7 +537,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
         if not queries_to_fetch:
             return ""
 
-        hf_log(f"🔍 [Pre-fetch JIT] 生成了 {len(queries_to_fetch)} 个定向检索词: {queries_to_fetch}")
+        logging.info(f"🔍 [Pre-fetch JIT] 生成了 {len(queries_to_fetch)} 个定向检索词: {queries_to_fetch}")
         
         with concurrent.futures.ThreadPoolExecutor(max_workers=self.lit_max_workers) as executor:
             futures = []
@@ -546,7 +554,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
                             self.citation_map[cite_tag] = {"in_text": res["in_text"], "ref": res["ref"], "id": res["id"]}
                             jit_rag_texts.append(f"{cite_tag} {res['rag_text']}")
 
-        hf_log(f"✅ [Pre-fetch JIT] 成功抓取并注入了 {len(jit_rag_texts)} 篇底层支撑文献。")
+        logging.info(f"✅ [Pre-fetch JIT] 成功抓取并注入了 {len(jit_rag_texts)} 篇底层支撑文献。")
         return "\n".join(jit_rag_texts)
 
     def generate_search_queries(self, topic, keywords):
@@ -567,7 +575,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
         time.sleep(random.uniform(0.5, 1.5)) 
         headers = {"x-api-key": str(self.s2_key).strip()} if self.s2_key else {}
         results = []
-        hf_log(f"📡 [S2 Worker] 正在 Semantic Scholar 检索: '{term}' (offset={offset})")
+        logging.info(f"📡 [S2 Worker] 正在 Semantic Scholar 检索: '{term}' (offset={offset})")
         try:
             resp = self._requests_get_with_retry(
                 f"{self.s2_api_url}/paper/search", 
@@ -576,7 +584,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
             )
             if resp.status_code == 400: return results
             papers = resp.json().get("data", [])
-            hf_log(f"✅ [S2 Worker] '{term}' 检索完成，捕获 {len(papers)} 条原始数据")
+            logging.info(f"✅ [S2 Worker] '{term}' 检索完成，捕获 {len(papers)} 条原始数据")
             for paper in papers:
                 title = paper.get("title")
                 year = paper.get("year")
@@ -617,7 +625,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
                     "source": "s2"
                 })
         except Exception as e:
-            hf_log(f"⚠️ S2 Worker 异常 (term: {term}): {e}")
+            logging.warning(f"⚠️ S2 Worker 异常 (term: {term}): {e}")
         return results
 
     def _oa_worker(self, term, page, filter_str):
@@ -625,7 +633,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
         headers = {"User-Agent": f"mailto:ssci_researcher_{random.randint(10000, 99999)}@gmail.com"}
         if self.openalex_key: headers["Authorization"] = f"Bearer {self.openalex_key}"
         results = []
-        hf_log(f"🌐 [OA Worker] 正在 OpenAlex 检索: '{term}' (page={page})")
+        logging.info(f"🌐 [OA Worker] 正在 OpenAlex 检索: '{term}' (page={page})")
         try:
             resp = self._requests_get_with_retry(
                 self.openalex_api_url, 
@@ -633,7 +641,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
                 params={"search": term, "per-page": self.lit_per_page, "page": page, "filter": filter_str}
             )
             raw_items = resp.json().get("results", [])
-            hf_log(f"✅ [OA Worker] '{term}' 检索完成，捕获 {len(raw_items)} 条原始数据")
+            logging.info(f"✅ [OA Worker] '{term}' 检索完成，捕获 {len(raw_items)} 条原始数据")
             for item in raw_items:
                 title = item.get("title")
                 if not title or not self._is_english_dominant(title): continue
@@ -676,7 +684,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
                     "source": "oa"
                 })
         except Exception as e:
-            hf_log(f"⚠️ OA Worker 异常 (term: {term}): {e}")
+            logging.warning(f"⚠️ OA Worker 异常 (term: {term}): {e}")
         return results
 
     def fetch_references_parallel(self, topic, keywords, min_required=50, progress_callback=None):
@@ -693,7 +701,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
             if expansion_phase == 0: 
                 search_terms = self.generate_search_queries(topic, keywords)
             else:
-                hf_log(f"🔄 [文献检索] 阶段 {expansion_phase}: 数量不足 ({len(self.citation_map)}/{min_required})，正在扩大检索范围...")
+                logging.info(f"🔄 [文献检索] 阶段 {expansion_phase}: 数量不足 ({len(self.citation_map)}/{min_required})，正在扩大检索范围...")
                 sys_prompt = "You are an expert academic librarian. We must broaden the search scope."
                 user_prompt = f"Topic: {topic}\nKeywords: {keywords}\nUsed Queries: {list(all_used_terms)}\nGenerate 3 NEW, BROADER English search queries. Return ONLY a JSON array of strings."
                 response = self.call_deepseek(sys_prompt, user_prompt, temp=0.5, model_name="deepseek-chat")
@@ -707,7 +715,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
             if not current_batch_terms: continue
             for t in current_batch_terms: all_used_terms.add(t)
 
-            hf_log(f"🚀 [并发下发] 启动扩张阶段 {expansion_phase+1}, 关键词: {current_batch_terms}")
+            logging.info(f"🚀 [并发下发] 启动扩张阶段 {expansion_phase+1}, 关键词: {current_batch_terms}")
             
             with concurrent.futures.ThreadPoolExecutor(max_workers=self.lit_max_workers) as executor:
                 futures = []
@@ -743,7 +751,7 @@ You MUST output your response EXACTLY using the following two delimiters. Do not
                     if len(self.citation_map) >= self.lit_target_papers:
                         break
 
-        hf_log(f"🎯 [并发检索结束] 成功入库并排重了 {len(self.citation_map)} 篇核心文献。")
+        logging.info(f"🎯 [并发检索结束] 成功入库并排重了 {len(self.citation_map)} 篇核心文献。")
         return "\n".join(llm_rag_context)
 
     def resolve_citations(self, text, quarantine=False):