kb_workflow/.env.example

@@ -12,3 +12,6 @@ export KB_BASE_PATH="~/work/knowledge-base"
 # 可选: Git 配置
 # export KB_GIT_USER="Your Name"
 # export KB_GIT_EMAIL="your@email.com"
+
+# 可选: 工作流行为配置
+# export KB_OVERWRITE_ARCHIVES="false"  # 非交互模式下是否覆盖已存在的归档文件（true/false）

kb_workflow/deep_analyzer/__init__.py

@@ -90,15 +90,19 @@ class SimpleAIClient:
 
     def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
         import requests
+        import sys
+
         base_url = base_url or "https://open.bigmodel.cn/api/coding/paas/v4"
         url = f"{base_url}/chat/completions"
         headers = {"Authorization": f"Bearer {api_key}"}
-        # GLM-4.5/4.6/4.7/5 默认启用thinking mode，添加 disable_reasoning 参数获取正常响应
+        # GLM-4.5/4.6/4.7/5 默认启用thinking mode，使用 thinking.type 参数禁用
         payload = {
             "model": model,
             "messages": [{"role": "user", "content": prompt}],
             "max_tokens": max_tokens,
-            "disable_reasoning": True  # 禁用思考模式，获取直接响应
+            "thinking": {
+                "type": "disabled"  # 禁用思考模式，获取直接响应
+            }
         }
         response = requests.post(url, headers=headers, json=payload, timeout=600)
         response.raise_for_status()
@@ -110,9 +114,9 @@ class SimpleAIClient:
             # 如果仍然为空，检查是否有reasoning_content（用于调试）
             reasoning = message.get('reasoning_content', '')
             if reasoning:
-                print(f"[DEBUG] Zhipu API返回空的content，但有reasoning_content。")
-                print(f"[DEBUG] 这可能表示disable_reasoning参数未生效或模型不支持。")
-                print(f"[DEBUG] reasoning_content前200字符: {reasoning[:200]}")
+                print(f"[DEBUG] Zhipu API返回空的content，但有reasoning_content。", file=sys.stderr)
+                print(f"[DEBUG] 这可能表示thinking.type参数未生效或模型不支持。", file=sys.stderr)
+                print(f"[DEBUG] reasoning_content前200字符: {reasoning[:200]}", file=sys.stderr)
             raise ValueError(f"Zhipu API返回空内容。请检查API配置和模型参数。响应数据: {data}")
 
         return content

kb_workflow/pipeline/config.py

@@ -58,6 +58,11 @@ class KBSettings(BaseSettings):
         default=SecretStr(""), description="B站 Cookie (for video download)"
     )
 
+    # 工作流行为配置
+    overwrite_archives: bool = Field(
+        default=False, description="覆盖已存在的归档文件（非交互模式）"
+    )
+
 
 @lru_cache
 def get_settings() -> KBSettings:

kb_workflow/pipeline/processors/archive.py

@@ -46,6 +46,8 @@ class ArchiveProcessor(BaseProcessor):
             Archive 处理结果
         """
         try:
+            import sys
+
             # Step 1: 检查是否已完成（幂等执行）
             if self.state_manager:
                 state = self.state_manager.load()
@@ -73,7 +75,9 @@ class ArchiveProcessor(BaseProcessor):
                 platform_dir, input_data.content_id
             )
             if existing_file:
-                should_overwrite = self._ask_overwrite(existing_file)
+                # 检测是否为非交互式环境
+                is_non_interactive = not sys.stdin.isatty()
+                should_overwrite = self._ask_overwrite(existing_file, is_non_interactive)
                 if not should_overwrite:
                     logger.info(f"User chose not to overwrite, skipping archive")
                     return ArchiveOutput(
@@ -143,17 +147,28 @@ class ArchiveProcessor(BaseProcessor):
         matches = list(platform_dir.glob(pattern))
         return matches[0] if matches else None
 
-    def _ask_overwrite(self, existing_file: Path) -> bool:
+    def _ask_overwrite(self, existing_file: Path, is_non_interactive: bool = False) -> bool:
         """询问用户是否覆盖已存在的文件
 
         使用 rich.prompt.Confirm 模式（与 Pipeline._ask_rerun_step 相同）
 
         Args:
             existing_file: 已存在的文件路径
+            is_non_interactive: 是否为非交互式环境（stdin不是tty）
 
         Returns:
             True 如果用户选择覆盖
         """
+        # 非交互式模式：使用配置决定
+        if is_non_interactive:
+            if self.settings.overwrite_archives:
+                logger.info(f"Non-interactive mode: Overwriting existing archive (KB_OVERWRITE_ARCHIVES=true): {existing_file.name}")
+                return True
+            else:
+                logger.info(f"Non-interactive mode: Skipping existing archive (KB_OVERWRITE_ARCHIVES=false): {existing_file.name}")
+                return False
+
+        # 交互式模式：询问用户
         try:
             from rich.console import Console
             from rich.prompt import Confirm