v2.1.0 数据质量检查+自动重新探索+审核拒绝复盘

- 新增数据质量评估：按类别核心字段（参数/上下文/发布日期/组织等）计算覆盖度
- 质量不足自动重新探索（限1次）：根据缺失字段生成4个针对性搜索词，重新搜索+抓取+提取
- 填充模板升级：补充AI模型完整字段（架构/开源/价格/能力），要求输出带URL的data_sources
- 提交附引用链接：从提取内容收集标题+URL，随产品数据提交给ParamHub
- 新增审核监控线程：提交后轮询审核状态，检测到拒绝时记录拒绝理由
- 审核拒绝复盘：大模型分析拒绝理由→生成定向搜索词→补搜缺失信息→重新提取填充→重新提交
- 复盘实测：deepseek-v4-flash-0731 被拒(缺参数量)后自动补全 37B/MoE/128K/价格，重新提交成功

config/agent_fill_fields_template.txt

@@ -13,41 +13,61 @@
 
 请完成以下工作：
 
-### 1. 获取对应类别的字段配置
-首先，请访问 ParamHub API 文档获取对应类别的字段定义：
-- API文档地址：http://192.168.2.8:12007/hz4th_coder/param-hub-python/src/branch/master/API.md
-- 根据产品类别（{{category}}）确定应该使用哪个API：
-  - AI模型 → `/api/models`，字段包括：name, organization, parameters, context_length, mmlu, publish_date, visible, is_pinned
-  - GPU → `/api/gpus`，字段包括：name, manufacturer, memory_gb, cuda_cores, tensor_cores, price_usd, release_year, visible, is_pinned
-  - CPU → `/api/cpus`，字段包括：name, manufacturer, cores, threads, base_clock, boost_clock, price_usd, visible, is_pinned
-  - 其他动态分类 → `/api/items/{category_id}`
+### 1. 确定字段配置
+根据产品类别（{{category}}）确定字段：
+
+- **AI模型** → `/api/models` 字段包括：
+  `name`(必填), `organization`(厂商), `parameters`(参数量如"70B"), `architecture`(架构), 
+  `context_length`(上下文长度), `mmlu`(能力评分), `humaneval`(代码能力),
+  `is_open_source`(是否开源true/false), `license`(许可证), 
+  `input_price`(输入价格), `output_price`(输出价格), `publish_date`(发布日期),
+  `description`(简介), `visible`, `is_pinned`
+
+- **GPU** → `/api/gpus` 字段：name, manufacturer, memory_gb, cuda_cores, tensor_cores, price_usd, release_year, visible, is_pinned
+- **CPU** → `/api/cpus` 字段：name, manufacturer, cores, threads, base_clock, boost_clock, price_usd, visible, is_pinned
+- 其他动态分类 → `/api/items/{category_id}`
 
 ### 2. 从内容库获取数据内容
-根据上述数据ID，从内容库中获取每条数据的完整内容。
+根据上述数据ID，从内容库中获取每条数据的完整内容和URL。
 
 ### 3. 整理产品参数
-根据获取到的内容，提取并整理产品的各项参数，严格按照API文档中定义的字段格式填充。
+根据获取到的内容，提取并整理产品的各项参数，严格按照字段格式填充。
+
+**提取原则（重要）：**
+- 优先提取**型号级精确参数**：参数量、上下文长度、能力指标、价格、发布日期等
+- 找不到型号级参数时，**品牌/系列级信息用于填充基础字段**：organization、series、publish_date
+- **能力表现**：如内容提及 mmlu/benchmark/评测分数、代码能力、推理速度等务必提取
+- **价格**：如内容提及 API 定价（每百万token价格）、硬件价格等务必提取
+- **开源信息**：如内容提及开源/开源协议务必提取 is_open_source/license
+- 不要编造或推测任何参数，只使用内容中实际存在的信息
+- 找不到的字段留空即可，**不要强行编造**
 
 ### 4. 格式检查
-对生成的数据进行以下检查：
-- 必填字段是否齐全（name必须有值）
-- 字段类型是否正确（数字字段不能是字符串，布尔字段必须是true/false）
-- 字段值是否合理（如参数量应为正数，价格应为正数等）
+- 必填字段：name 必须有值
+- 数字字段必须是数字（context_length/mmlu/价格等），布尔字段必须是 true/false
 - 如果发现格式问题，请修正后重新输出
 
 ### 5. 输出要求
-请以JSON格式输出最终的产品数据（不要提交，只输出数据）：
+请以JSON格式输出（不要提交，只输出数据）：
 ```json
 {
   "success": true,
   "product_data": {
     "name": "产品名称",
-    "field1": "值1",
-    "field2": "值2",
+    "organization": "厂商",
+    "parameters": "70B",
+    "context_length": 4096,
+    "mmlu": 85.5,
+    "publish_date": "2024-01-01",
+    "is_open_source": true,
+    "input_price": 1.0,
+    "output_price": 2.0,
     "visible": true,
     "is_pinned": false
   },
-  "data_sources": [数据ID列表],
+  "data_sources": [
+    {"id": 1, "title": "来源标题", "url": "https://来源链接", "used_for": "该来源提供了哪些字段"}
+  ],
   "format_check": {
     "passed": true,
     "issues_found": [],
@@ -58,8 +78,8 @@
 ```
 
 **注意：**
-- 严格按照API文档的字段定义填充数据
-- 不要编造或推测任何参数，只使用内容库中实际存在的信息
-- 如果某些字段无法从内容中提取，可以留空或填写默认值
+- `data_sources` 必须是数组，每项包含 `id`、`title`、`url`（原文链接）、`used_for`（该来源提供了哪些字段）
+- 不要编造或推测任何参数，只使用内容中实际存在的信息
+- 如果某些字段无法从内容中提取，可以留空
 - 不要执行任何提交操作，只生成并输出数据
 - 确保输出的JSON格式正确，可以被程序解析

config/agent_task_template.txt

@@ -15,16 +15,19 @@
 
 ## 任务要求
 
-请分析上述内容库ID对应的数据，判断哪些与产品「{{product_name}}」**直接相关**且**对提取产品参数有用**。
+请分析上述内容库ID对应的数据，判断哪些与产品「{{product_name}}」**相关**且**对提取产品参数有用**。
 
-### 判断标准：
-1. **直接相关性**：内容必须明确提及该产品名称或其主要型号，排除仅提及相似产品或竞品的内容
-2. **参数提取价值**：内容应包含可用于填充产品字段的信息，如：
+### 判断标准（宽松模式）：
+1. **直接相关**：内容明确提及该产品名称或其主要型号 → 最高优先级
+2. **品牌/系列相关**：内容提及该产品的品牌或所属系列（例如产品是某品牌的某个版本，则品牌官网、系列介绍页、开放平台页均算相关）→ 应纳入
+3. **信息价值**：内容应包含可用于填充产品字段的信息，如：
    - 产品规格参数（尺寸、重量、容量等）
    - 技术规格（性能指标、接口、兼容性等）
    - 功能特性
    - 发布信息（发布日期、价格等）
+   - 组织/厂商信息
    - 其他结构化产品数据
+4. **注意**：如果找不到精确型号页面，**品牌官网、系列页面、开放平台等也纳入**，便于提取品牌、组织、系列、发布时间等基础字段；只有当内容与产品完全无关（竞品、其他产品）时才排除
 
 ### 输出要求：
 请以JSON格式输出分析结果，包含以下字段：
@@ -46,8 +49,9 @@
 ```
 
 **注意：**
-- `relevant_ids`：与产品直接相关且对参数提取有用的数据ID列表
+- `relevant_ids`：与产品相关且对参数提取有用的数据ID列表（品牌/系列相关也应纳入）
 - `analysis`：每个相关ID的简要分析说明
 - `excluded_ids`：被排除的ID列表（可选）
 - `exclusion_reasons`：排除原因说明（可选）
 - `confidence`：整体判断的置信度
+- **至少返回1个最相关的ID**，除非所有内容都与该产品完全无关

models/database.py

@@ -772,6 +772,14 @@ class Database:
             ''')
             return [dict(row) for row in cursor.fetchall()]
     
+    def get_session_status(self, session_id):
+        """获取单个会话状态"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('SELECT * FROM process_sessions WHERE session_id = ?', (session_id,))
+            row = cursor.fetchone()
+            return dict(row) if row else None
+    
     def get_recent_sessions(self, limit=20):
         """获取最近的会话"""
         with self.get_connection() as conn:

routes/process_monitor.py

@@ -48,7 +48,15 @@ def start_process():
         subcategory = data.get('subcategory')
         
         # 启动处理流程
-        session_id = process_monitor.start_process(product_name, category, subcategory)
+        session_id, started = process_monitor.start_process(product_name, category, subcategory)
+        
+        if not started:
+            return jsonify({
+                'success': True,
+                'session_id': session_id,
+                'message': f'产品 {product_name} 已有处理会话在进行中，已复用现有会话',
+                'duplicated': True
+            })
         
         return jsonify({
             'success': True,

routes/products.py

@@ -114,7 +114,9 @@ def get_product_history(product_name):
 
 @bp.route('/process', methods=['POST'])
 def process_single():
-    """处理单个产品"""
+    """处理单个产品（大模型驱动）"""
+    from services.process_monitor import process_monitor
+    
     data = request.get_json()
     
     if 'product_name' not in data:
@@ -126,40 +128,32 @@ def process_single():
         'subcategory': data.get('subcategory')
     }
     
-    # 检查是否正在处理
-    processing = db.get_processing_products()
-    if any(p['product_name'] == product_info['product_name'] for p in processing):
-        return jsonify({'error': '该产品正在处理中'}), 400
-    
-    # 添加到处理中列表
-    db.start_processing(
+    # 启动大模型处理会话（内部有防重）
+    session_id, started = process_monitor.start_process(
         product_name=product_info['product_name'],
         category=product_info['category'],
         subcategory=product_info['subcategory']
     )
     
-    try:
-        # 执行处理
-        result = process_service.process_product(product_info)
-        
-        # 从待处理列表移除
-        db.remove_pending_product(product_info['product_name'])
-        
-        # 如果发现新产品，已在process_service中添加到待处理列表
-        
+    if not started:
         return jsonify({
-            'success': result['success'],
-            'message': result['message'],
-            'review_id': result.get('review_id'),
-            'new_products': result.get('new_products', [])
+            'success': True,
+            'message': f'产品 {product_info["product_name"]} 已有处理会话在进行中',
+            'session_id': session_id,
+            'duplicated': True
         })
-    finally:
-        # 完成处理，从处理中列表移除
-        db.finish_processing(product_info['product_name'])
+    
+    return jsonify({
+        'success': True,
+        'message': f'处理流程已启动: {product_info["product_name"]}',
+        'session_id': session_id
+    })
 
 @bp.route('/process/batch', methods=['POST'])
 def process_batch():
-    """批量处理产品"""
+    """批量处理产品（大模型驱动，异步启动会话）"""
+    from services.process_monitor import process_monitor
+    
     data = request.get_json()
     limit = data.get('limit', 5)
     
@@ -175,39 +169,27 @@ def process_batch():
     
     results = []
     for product in products:
-        # 检查是否正在处理
-        processing = db.get_processing_products()
-        if any(p['product_name'] == product['product_name'] for p in processing):
-            results.append({
-                'product_name': product['product_name'],
-                'success': False,
-                'message': '正在处理中'
-            })
-            continue
-        
-        # 添加到处理中列表
-        db.start_processing(
+        # 启动大模型处理会话（内部有防重）
+        session_id, started = process_monitor.start_process(
             product_name=product['product_name'],
             category=product.get('category'),
             subcategory=product.get('subcategory')
         )
         
-        try:
-            # 执行处理
-            result = process_service.process_product(product)
-            
-            # 从待处理列表移除
-            db.remove_pending_product(product['product_name'])
-            
+        if started:
             results.append({
                 'product_name': product['product_name'],
-                'success': result['success'],
-                'message': result['message'],
-                'review_id': result.get('review_id')
+                'success': True,
+                'message': '处理会话已启动',
+                'session_id': session_id
+            })
+        else:
+            results.append({
+                'product_name': product['product_name'],
+                'success': False,
+                'message': '已有处理会话在进行中',
+                'session_id': session_id
             })
-        finally:
-            # 完成处理
-            db.finish_processing(product['product_name'])
     
     return jsonify({
         'success': True,

services/paramhub_client.py

@@ -18,18 +18,44 @@ class ParamHubClient:
                 f'{self.base_url}/login',
                 json={'password': self.password}
             )
-            return response.json().get('success', False)
+            if response.status_code == 200 and response.json().get('success'):
+                return True
+            return False
         except Exception as e:
             print(f"登录失败: {str(e)}")
             return False
     
+    def _ensure_login(self):
+        """确保已登录；session 失效时自动重新登录"""
+        if not self.session:
+            return self.login()
+        return True
+    
+    def _request(self, method, url, **kwargs):
+        """发送请求，401 时自动重新登录并重试一次"""
+        self._ensure_login()
+        try:
+            response = self.session.request(method, url, **kwargs)
+            # 401 未登录：重新登录后重试一次
+            if response.status_code == 401:
+                print(f"登录态失效，重新登录后重试: {method} {url}")
+                if self.login():
+                    response = self.session.request(method, url, **kwargs)
+            return response
+        except Exception:
+            # 连接失败时也尝试重新登录一次（服务可能重启过）
+            print(f"请求失败，尝试重新登录: {method} {url}")
+            try:
+                if self.login():
+                    return self.session.request(method, url, **kwargs)
+            except Exception:
+                pass
+            raise
+    
     def get_categories(self):
         """获取所有分类"""
         try:
-            if not self.session:
-                self.login()
-            
-            response = self.session.get(f'{self.base_url}/api/categories?all=1')
+            response = self._request('GET', f'{self.base_url}/api/categories?all=1')
             return response.json()
         except Exception as e:
             print(f"获取分类失败: {str(e)}")
@@ -38,10 +64,7 @@ class ParamHubClient:
     def get_category_fields(self, category_id):
         """获取分类的字段配置"""
         try:
-            if not self.session:
-                self.login()
-            
-            response = self.session.get(f'{self.base_url}/api/categories/{category_id}')
+            response = self._request('GET', f'{self.base_url}/api/categories/{category_id}')
             return response.json()
         except Exception as e:
             print(f"获取分类字段失败: {str(e)}")
@@ -60,9 +83,6 @@ class ParamHubClient:
             (success, review_id or error_message)
         """
         try:
-            if not self.session:
-                self.login()
-            
             # 根据分类类型选择API端点
             if category_type == 'model':
                 endpoint = f'{self.base_url}/api/models'
@@ -78,7 +98,7 @@ class ParamHubClient:
             # 添加审核模式需要的字段
             data['status'] = 'pending'
             
-            response = self.session.post(endpoint, json=data)
+            response = self._request('POST', endpoint, json=data)
             result = response.json()
             
             if response.status_code == 200 or response.status_code == 201:
@@ -91,10 +111,7 @@ class ParamHubClient:
     def get_reviews(self, status='pending'):
         """获取待审核列表"""
         try:
-            if not self.session:
-                self.login()
-            
-            response = self.session.get(f'{self.base_url}/api/reviews?status={status}')
+            response = self._request('GET', f'{self.base_url}/api/reviews?status={status}')
             return response.json()
         except Exception as e:
             print(f"获取审核列表失败: {str(e)}")
@@ -103,26 +120,41 @@ class ParamHubClient:
     def get_review_count(self):
         """获取待审核数量"""
         try:
-            if not self.session:
-                self.login()
-            
-            response = self.session.get(f'{self.base_url}/api/reviews/count')
+            response = self._request('GET', f'{self.base_url}/api/reviews/count')
             return response.json().get('count', 0)
         except Exception as e:
             print(f"获取审核数量失败: {str(e)}")
             return 0
     
+    def get_review_status(self, review_id):
+        """
+        获取审核状态
+        Returns:
+            {
+                'status': 'pending'/'approved'/'rejected',
+                'reject_reason': str (被拒时的理由),
+                'review': dict
+            } or None
+        """
+        try:
+            response = self._request('GET', f'{self.base_url}/api/reviews/{review_id}')
+            if response.status_code == 200:
+                review = response.json()
+                return {
+                    'status': review.get('status', 'pending'),
+                    'reject_reason': review.get('reject_reason', ''),
+                    'review': review
+                }
+            return None
+        except Exception as e:
+            print(f"获取审核状态失败: {str(e)}")
+            return None
+    
     def send_notification(self, message):
         """发送通知到后台管理"""
         try:
-            if not self.session:
-                self.login()
-            
             # 使用通知API发送通知
-            response = self.session.post(
-                f'{self.base_url}/api/notifications',
-                json={'message': message}
-            )
+            response = self._request('POST', f'{self.base_url}/api/notifications', json={'message': message})
             return response.status_code == 200 or response.status_code == 201
         except Exception as e:
             print(f"发送通知失败: {str(e)}")
@@ -143,14 +175,8 @@ class ParamHubClient:
             }
         """
         try:
-            if not self.session:
-                self.login()
-            
             # 1. 检查已发布的产品（通过搜索API）
-            response = self.session.get(
-                f'{self.base_url}/api/search',
-                params={'q': product_name}
-            )
+            response = self._request('GET', f'{self.base_url}/api/search', params={'q': product_name})
             
             if response.status_code == 200:
                 result = response.json()
@@ -196,10 +222,7 @@ class ParamHubClient:
             
             
             # 2. 检查待审核列表
-            response = self.session.get(
-                f'{self.base_url}/api/reviews',
-                params={'status': 'pending'}
-            )
+            response = self._request('GET', f'{self.base_url}/api/reviews', params={'status': 'pending'})
             
             if response.status_code == 200:
                 reviews = response.json()

services/process_monitor.py

@@ -37,7 +37,14 @@ class ProcessMonitor:
         return f"proc_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}"
     
     def start_process(self, product_name, category=None, subcategory=None):
-        """启动产品处理流程"""
+        """启动产品处理流程（同产品防重）"""
+        # 防重：检查该产品是否已有活跃会话
+        active = db.get_active_sessions()
+        for s in active:
+            if s['product_name'] == product_name and s['status'] in ('pending', 'running', 'paused'):
+                logger.warning(f"产品 {product_name} 已有活跃会话 {s['session_id']}，跳过重复启动")
+                return s['session_id'], False
+        
         session_id = self.create_session_id()
         
         # 创建会话记录
@@ -59,7 +66,7 @@ class ProcessMonitor:
         thread.start()
         
         logger.info(f"启动处理会话: {session_id}, 产品: {product_name}")
-        return session_id
+        return session_id, True
     
     def _run_process(self, session_id, product_name, category, subcategory):
         """执行处理流程"""
@@ -204,7 +211,12 @@ class ProcessMonitor:
                     self._fail_step(session_id, 3, str(e))
             
             # 步骤4: 提取产品数据（调用大模型筛选相关内容）
-            if not self._check_pause(session_id):
+            # 可重试：质量不足时重新探索（限1次）
+            retry_explored = False
+            while True:
+                if self._check_pause(session_id):
+                    break
+                
                 self._start_step(session_id, product_name, 4, '提取产品数据(大模型)')
                 try:
                     # 构建任务文本
@@ -253,23 +265,70 @@ class ProcessMonitor:
                             })
                             logger.info(f"[{session_id}] 步骤4完成: 大模型返回 {len(parsed['relevant_ids'])} 个相关ID")
                         else:
+                            # 大模型未返回相关ID：兜底处理——选取搜索内容中前1条作为候选
                             all_data['extracted_data'] = None
-                            self._complete_step(session_id, 4, {
-                                'has_data': False,
-                                'agent': '大模型',
-                                'model': self._get_active_model_name(),
-                                'task_text': task_text,
-                                'agent_output': agent_result.get('output', '')[:2000]
-                            }, status='skipped')
-                            result['message'] = '大模型未找到相关数据ID'
+                            fallback_ids = []
+                            if all_data.get('fetched_contents'):
+                                fallback_ids = [all_data['fetched_contents'][0]['id']]
+                            elif all_data.get('library_results'):
+                                fallback_ids = [all_data['library_results'][0]['id']]
+                            
+                            fallback_contents = []
+                            for aid in fallback_ids:
+                                article = db.get_article_by_id(aid)
+                                if article:
+                                    fallback_contents.append({
+                                        'id': aid,
+                                        'title': article.get('search_title', ''),
+                                        'url': article.get('url', ''),
+                                        'content': article.get('content', ''),
+                                        'summary': article.get('summary', ''),
+                                        'analysis': '兜底候选：大模型未筛选出精确相关数据，选用该条作为参考'
+                                    })
+                            
+                            if fallback_contents:
+                                all_data['extracted_data'] = {
+                                    'name': product_name,
+                                    'relevant_ids': fallback_ids,
+                                    'relevant_contents': fallback_contents,
+                                    'confidence': 'low',
+                                    'raw_output': agent_result.get('output', ''),
+                                    'fallback': True
+                                }
+                                self._complete_step(session_id, 4, {
+                                    'has_data': True,
+                                    'agent': '大模型',
+                                    'model': self._get_active_model_name(),
+                                    'task_text': task_text,
+                                    'relevant_ids': fallback_ids,
+                                    'relevant_count': len(fallback_contents),
+                                    'confidence': 'low',
+                                    'fallback': True,
+                                    'agent_output': agent_result.get('output', '')[:2000]
+                                })
+                                logger.info(f"[{session_id}] 步骤4兜底: 大模型未筛选出相关ID，选用 {fallback_ids} 作为候选")
+                            else:
+                                self._complete_step(session_id, 4, {
+                                    'has_data': False,
+                                    'agent': '大模型',
+                                    'model': self._get_active_model_name(),
+                                    'task_text': task_text,
+                                    'agent_output': agent_result.get('output', '')[:2000]
+                                }, status='skipped')
+                                result['message'] = '大模型未找到相关数据且无兜底内容'
+                                break
                     else:
                         self._fail_step(session_id, 4, f"大模型调用失败: {agent_result.get('error', '未知错误')}")
                         result['message'] = f'大模型调用失败: {agent_result.get("error")}'
+                        break
                 except Exception as e:
                     self._fail_step(session_id, 4, str(e))
-            
-            # 步骤5: 填充字段（调用大模型生成数据并检查格式）
-            if not self._check_pause(session_id) and all_data['extracted_data']:
+                    break
+                
+                # 步骤5: 填充字段（调用大模型生成数据并检查格式）
+                if self._check_pause(session_id) or not all_data['extracted_data']:
+                    break
+                
                 self._start_step(session_id, product_name, 5, '填充字段(大模型)')
                 try:
                     # 构建任务文本
@@ -292,6 +351,10 @@ class ProcessMonitor:
                             
                             if validation_result.get('valid'):
                                 all_data['filled_data'] = product_data
+                                all_data['data_sources'] = fill_parsed.get('data_sources', [])
+                                
+                                # 质量检查：核心字段覆盖度
+                                quality = self._assess_data_quality(product_data, category)
                                 
                                 self._complete_step(session_id, 5, {
                                     'filled': True,
@@ -301,32 +364,90 @@ class ProcessMonitor:
                                     'product_data': product_data,
                                     'format_check': format_check,
                                     'validation': validation_result,
+                                    'quality': quality,
                                     'agent_output': fill_agent_result.get('output', '')[:2000]
                                 })
-                                logger.info(f"[{session_id}] 步骤5完成: 数据生成成功，格式验证通过")
+                                logger.info(f"[{session_id}] 步骤5完成: 数据生成成功，质量评分={quality.get('score', 0):.0%}")
+                                
+                                # 质量不足且未重试过 → 重新探索
+                                if not quality.get('sufficient') and not retry_explored:
+                                    retry_explored = True
+                                    logger.info(f"[{session_id}] 数据质量不足({quality.get('score', 0):.0%})，触发重新探索")
+                                    
+                                    # 重新探索：换更精确的关键词重新搜索+抓取
+                                    explore_result = self._re_explore(
+                                        session_id, product_name, category, subcategory, all_data, quality
+                                    )
+                                    
+                                    if explore_result:
+                                        logger.info(f"[{session_id}] 重新探索完成，新增 {explore_result.get('new_fetched', 0)} 条内容，重新提取")
+                                        continue  # 重新执行步骤4/5
+                                    else:
+                                        logger.warning(f"[{session_id}] 重新探索未获取新内容，使用现有数据提交")
+                                        break
+                                else:
+                                    break
                             else:
                                 # 格式验证失败，记录问题
                                 self._fail_step(session_id, 5, f"数据格式验证失败: {validation_result.get('errors', [])}")
                                 result['message'] = '数据格式验证失败'
+                                break
                         else:
                             error_msg = fill_parsed.get('message', '未知错误') if fill_parsed else '解析失败'
                             self._fail_step(session_id, 5, f"大模型执行失败: {error_msg}")
                             result['message'] = f'大模型执行失败: {error_msg}'
+                            break
                     else:
                         self._fail_step(session_id, 5, f"大模型调用失败: {fill_agent_result.get('error', '未知错误')}")
                         result['message'] = f'大模型调用失败: {fill_agent_result.get("error")}'
+                        break
                 except Exception as e:
                     self._fail_step(session_id, 5, str(e))
+                    break
             
-            # 步骤6: 提交审核（直接调用ParamHub API，不再依赖智能体）
+            # 步骤6: 提交审核（直接调用ParamHub API，附引用链接）
             if not self._check_pause(session_id) and all_data['filled_data']:
                 self._start_step(session_id, product_name, 6, '提交审核')
                 try:
                     category_type = self._get_category_type(category)
                     subcategory_id = subcategory
+                    
+                    # 组装提交数据：附加引用链接等元信息
+                    submit_data = dict(all_data['filled_data'])
+                    
+                    # 引用链接：从提取内容中收集（标题+URL）
+                    reference_links = []
+                    seen_urls = set()
+                    for item in all_data.get('extracted_data', {}).get('relevant_contents', []):
+                        url = item.get('url', '')
+                        if url and url not in seen_urls:
+                            seen_urls.add(url)
+                            reference_links.append({
+                                'title': item.get('title', url[:60]),
+                                'url': url
+                            })
+                    # 补充数据源中带URL的引用
+                    for src in all_data.get('data_sources', []):
+                        if isinstance(src, dict):
+                            url = src.get('url', '') or src.get('link', '')
+                            if url and url not in seen_urls:
+                                seen_urls.add(url)
+                                reference_links.append({
+                                    'title': src.get('title', src.get('name', url[:60])),
+                                    'url': url
+                                })
+                    
+                    if reference_links:
+                        submit_data['reference_links'] = reference_links
+                        submit_data['_data_sources'] = all_data.get('data_sources', [])
+                    
+                    # 标记重试/探索信息（若有）
+                    if retry_explored:
+                        submit_data['_re_explored'] = True
+                    
                     success, review_id_or_error = paramhub_client.submit_for_review(
                         category_type,
-                        all_data['filled_data'],
+                        submit_data,
                         subcategory_id
                     )
                     
@@ -336,7 +457,9 @@ class ProcessMonitor:
                             'submitted': True,
                             'agent': 'ParamHub API',
                             'review_id': review_id,
-                            'product_data': all_data['filled_data']
+                            'product_data': submit_data,
+                            'reference_links_count': len(reference_links),
+                            're_explored': retry_explored
                         })
                         
                         result['success'] = True
@@ -354,13 +477,47 @@ class ProcessMonitor:
                             review_id=review_id,
                             details=all_data
                         )
-                        logger.info(f"[{session_id}] 步骤6完成: 提交成功, review_id={review_id}")
+                        logger.info(f"[{session_id}] 步骤6完成: 提交成功, review_id={review_id}, 引用链接 {len(reference_links)} 条")
+                        
+                        # 启动审核监控线程：被拒时按理由复盘重跑
+                        self._start_review_monitor(
+                            session_id, product_name, category, subcategory, review_id
+                        )
                     else:
                         self._fail_step(session_id, 6, f"提交失败: {review_id_or_error}")
                         result['message'] = f'提交失败: {review_id_or_error}'
                 except Exception as e:
                     self._fail_step(session_id, 6, str(e))
             
+            # 兜底收尾：未成功提交的会话标记为 failed
+            if not result.get('success'):
+                if not result.get('message'):
+                    result['message'] = '处理未完成（无有效数据或中途停止）'
+                # 会话已在步骤失败时标记 failed；这里确保未走到提交分支的会话也收尾
+                cur_status = db.get_session_status(session_id)
+                if cur_status and cur_status.get('status') in ('running', 'pending'):
+                    db.update_session_status(session_id, 'failed')
+                # 记录处理历史（异常/失败情况）
+                db.add_process_history(
+                    product_name=product_name,
+                    category=category,
+                    subcategory=subcategory,
+                    status='failed',
+                    details={'message': result['message'], 'all_data': {
+                        'library_count': len(all_data.get('library_results', [])),
+                        'internet_count': len(all_data.get('internet_results', [])),
+                        'fetched_count': len(all_data.get('fetched_contents', [])),
+                        'extracted': bool(all_data.get('extracted_data')),
+                        'filled': bool(all_data.get('filled_data'))
+                    }}
+                )
+            
+            # 处理结束（无论成功失败），从待处理列表移除
+            try:
+                db.remove_pending_product(product_name)
+            except Exception as e:
+                logger.warning(f"[{session_id}] 移除待处理产品失败: {e}")
+            
             # 清理
             if session_id in self.active_sessions:
                 del self.active_sessions[session_id]
@@ -627,7 +784,7 @@ class ProcessMonitor:
                 "参考API文档: http://192.168.2.8:12007/hz4th_coder/param-hub-python/src/branch/master/API.md"
             )
         
-        # 构建相关内容ID列表
+        # 构建相关内容ID列表（含URL，便于大模型输出引用链接）
         relevant_ids = extracted_data.get('relevant_ids', [])
         relevant_contents = extracted_data.get('relevant_contents', [])
         
@@ -636,7 +793,11 @@ class ProcessMonitor:
             for item in relevant_contents:
                 aid = item.get('id', '')
                 title = item.get('title', '')
-                content_lines.append(f"ID {aid}: {title}")
+                url = item.get('url', '')
+                if url:
+                    content_lines.append(f"ID {aid}: {title} (URL: {url})")
+                else:
+                    content_lines.append(f"ID {aid}: {title}")
             relevant_text = '\n'.join(content_lines)
         elif relevant_ids:
             relevant_text = '\n'.join([f"ID {aid}" for aid in relevant_ids])
@@ -765,6 +926,465 @@ class ProcessMonitor:
             'warnings': warnings
         }
     
+    def _assess_data_quality(self, product_data, category):
+        """
+        评估产品数据质量：核心字段覆盖度
+        Returns:
+            {
+                'score': float (0-1),
+                'sufficient': bool,
+                'missing': [缺失的核心字段名],
+                'filled': [已填充的字段名]
+            }
+        """
+        category_type = self._get_category_type(category)
+        
+        # 各类别核心字段定义
+        core_fields = {
+            'model': ['organization', 'parameters', 'context_length', 'publish_date'],
+            'gpu': ['manufacturer', 'memory_gb', 'cuda_cores', 'price_usd'],
+            'cpu': ['manufacturer', 'cores', 'threads', 'base_clock'],
+            'dynamic': ['organization']
+        }
+        fields = core_fields.get(category_type, core_fields['dynamic'])
+        
+        filled = []
+        missing = []
+        for f in fields:
+            val = product_data.get(f)
+            if val is not None and val != '' and val != 'null':
+                filled.append(f)
+            else:
+                missing.append(f)
+        
+        # 附加信息丰富度（价格、能力指标等加分项）
+        bonus_fields = {
+            'model': ['mmlu', 'input_price', 'output_price', 'is_open_source', 'architecture', 'license'],
+            'gpu': ['tensor_cores', 'release_year', 'boost_clock'],
+            'cpu': ['boost_clock', 'price_usd', 'release_year'],
+            'dynamic': []
+        }
+        bonus = bonus_fields.get(category_type, [])
+        bonus_filled = [f for f in bonus if product_data.get(f) not in (None, '', 'null')]
+        
+        score = (len(filled) + 0.5 * len(bonus_filled)) / (len(fields) + 0.5 * len(bonus))
+        score = min(1.0, max(0.0, score))
+        
+        # 核心字段至少填满 60% 且无全部缺失才视为达标；
+        # 若核心字段一个都没有（score 很低），视为不达标触发重新探索
+        sufficient = score >= 0.6 and len(filled) >= 2
+        
+        return {
+            'score': round(score, 3),
+            'sufficient': sufficient,
+            'missing': missing,
+            'filled': filled,
+            'bonus_filled': bonus_filled
+        }
+    
+    def _build_explore_keywords(self, product_name, category, quality):
+        """根据缺失字段生成重新探索的搜索关键词列表"""
+        category_type = self._get_category_type(category)
+        missing = set(quality.get('missing', []))
+        keywords = []
+        
+        # 按缺失字段生成针对性搜索词
+        if category_type == 'model':
+            if 'parameters' in missing:
+                keywords.append(f'{product_name} 参数 参数量')
+            if 'context_length' in missing:
+                keywords.append(f'{product_name} context length 上下文')
+            if 'publish_date' in missing:
+                keywords.append(f'{product_name} release date 发布')
+            if 'organization' in missing:
+                keywords.append(f'{product_name} 厂商 公司')
+            # 通用补充
+            keywords.append(f'{product_name} 规格 性能')
+            keywords.append(f'{product_name} 价格 API')
+        elif category_type == 'gpu':
+            if 'memory_gb' in missing:
+                keywords.append(f'{product_name} 显存 memory')
+            if 'cuda_cores' in missing:
+                keywords.append(f'{product_name} CUDA cores')
+            if 'price_usd' in missing:
+                keywords.append(f'{product_name} price 价格')
+            keywords.append(f'{product_name} 规格 参数')
+        elif category_type == 'cpu':
+            if 'cores' in missing:
+                keywords.append(f'{product_name} cores 核心')
+            if 'threads' in missing:
+                keywords.append(f'{product_name} threads 线程')
+            if 'base_clock' in missing:
+                keywords.append(f'{product_name} base clock 频率')
+            keywords.append(f'{product_name} 规格 参数')
+        else:
+            keywords.append(f'{product_name} 参数 规格')
+        
+        # 去重，最多4个
+        seen = set()
+        result = []
+        for kw in keywords:
+            if kw not in seen:
+                seen.add(kw)
+                result.append(kw)
+            if len(result) >= 4:
+                break
+        
+        return result
+    
+    def _re_explore(self, session_id, product_name, category, subcategory, all_data, quality):
+        """
+        重新探索：根据缺失字段生成更精确的关键词，重新搜索+抓取
+        成功返回新抓取数量，失败返回 None
+        """
+        try:
+            logger.info(f"[{session_id}] 重新探索开始，缺失字段: {quality.get('missing')}")
+            
+            # 生成探索关键词
+            keywords = self._build_explore_keywords(product_name, category, quality)
+            logger.info(f"[{session_id}] 探索关键词: {keywords}")
+            
+            new_fetched = 0
+            new_ids = []
+            
+            for kw in keywords:
+                if self._check_pause(session_id):
+                    break
+                
+                try:
+                    internet_results = search_service.search_internet(kw, max_results=5, use_cache=False)
+                except Exception as e:
+                    logger.warning(f"[{session_id}] 探索搜索失败 [{kw}]: {e}")
+                    continue
+                
+                if not internet_results:
+                    continue
+                
+                for r in internet_results:
+                    if self._check_pause(session_id):
+                        break
+                    url = r.get('url', '')
+                    if not url:
+                        continue
+                    # 跳过已抓取过的URL
+                    existing = db.search_articles(url)
+                    if existing and len(existing) > 0:
+                        continue
+                    
+                    fetch_result = search_service.fetch_url_content(url)
+                    if fetch_result.get('success'):
+                        try:
+                            article_id = db.add_article(
+                                product_names=[product_name],
+                                category=category or '',
+                                keywords=[kw],
+                                summary=fetch_result.get('description', '')[:200],
+                                content=fetch_result.get('content', ''),
+                                source=url,
+                                url=url,
+                                search_title=fetch_result.get('title', url[:50])
+                            )
+                            new_ids.append(article_id)
+                            new_fetched += 1
+                            logger.info(f"[{session_id}] 探索抓取新增: ID={article_id} {fetch_result.get('title', '')[:30]}")
+                        except Exception as e:
+                            logger.warning(f"[{session_id}] 探索保存失败 {url}: {e}")
+                    
+                    time.sleep(0.3)
+            
+            if new_fetched == 0:
+                logger.info(f"[{session_id}] 重新探索未获取到新内容")
+                return None
+            
+            # 把新抓取的内容加入 fetched_contents（供步骤4重新筛选）
+            fetched_contents = all_data.setdefault('fetched_contents', [])
+            for aid in new_ids:
+                article = db.get_article_by_id(aid)
+                if article:
+                    fetched_contents.append({
+                        'id': aid,
+                        'url': article.get('url', ''),
+                        'title': article.get('search_title', ''),
+                        'content': (article.get('content') or '')[:500]
+                    })
+            
+            # 清除旧提取结果，强制重新提取
+            all_data['extracted_data'] = None
+            all_data['filled_data'] = None
+            
+            return {'new_fetched': new_fetched, 'new_ids': new_ids}
+            
+        except Exception as e:
+            logger.error(f"[{session_id}] 重新探索异常: {e}")
+            return None
+    
+    # ===== 审核监控与复盘 =====
+    
+    def _start_review_monitor(self, session_id, product_name, category, subcategory, review_id):
+        """启动审核监控线程：轮询审核状态，被拒时按理由复盘重跑"""
+        thread = threading.Thread(
+            target=self._monitor_review,
+            args=(session_id, product_name, category, subcategory, review_id),
+            daemon=True
+        )
+        thread.start()
+        logger.info(f"[{session_id}] 审核监控已启动: review_id={review_id}")
+    
+    def _monitor_review(self, session_id, product_name, category, subcategory, review_id):
+        """轮询审核状态（最多30次，每次60秒）"""
+        for i in range(30):
+            time.sleep(60)
+            try:
+                status_info = paramhub_client.get_review_status(review_id)
+                if not status_info:
+                    logger.warning(f"[{session_id}] 审核状态查询失败(review={review_id})，第{i+1}次")
+                    continue
+                
+                status = status_info.get('status')
+                if status == 'approved':
+                    logger.info(f"[{session_id}] 审核通过! review_id={review_id}")
+                    try:
+                        db.add_process_history(
+                            product_name=product_name,
+                            category=category,
+                            subcategory=subcategory,
+                            status='approved',
+                            review_id=review_id,
+                            details={'message': '审核通过'}
+                        )
+                    except Exception:
+                        pass
+                    return
+                
+                elif status == 'rejected':
+                    reason = status_info.get('reject_reason', '') or '无具体理由'
+                    logger.warning(f"[{session_id}] 审核被拒! review_id={review_id}, 理由: {reason}")
+                    try:
+                        db.add_process_history(
+                            product_name=product_name,
+                            category=category,
+                            subcategory=subcategory,
+                            status='rejected',
+                            review_id=review_id,
+                            details={'message': f'审核被拒: {reason}'}
+                        )
+                    except Exception:
+                        pass
+                    
+                    # 按拒绝理由复盘重跑
+                    self._review_retry(session_id, product_name, category, subcategory, review_id, reason)
+                    return
+                
+                # pending：继续等待
+                logger.info(f"[{session_id}] 审核状态: pending (第{i+1}次轮询)")
+                
+            except Exception as e:
+                logger.error(f"[{session_id}] 审核监控异常: {e}")
+        
+        logger.info(f"[{session_id}] 审核监控结束（30次轮询未出结果）")
+    
+    def _review_retry(self, session_id, product_name, category, subcategory, review_id, reason):
+        """
+        审核被拒后的复盘重跑：
+        1. 让大模型分析拒绝理由，得出缺失项和搜索建议
+        2. 定向搜索补齐缺失信息
+        3. 重新提取、填充、提交
+        """
+        try:
+            logger.info(f"[{session_id}] 开始审核复盘: 拒绝理由={reason}")
+            
+            # 获取原会话数据（步骤数据里取 product_data 和引用链接）
+            steps = db.get_process_steps(session_id)
+            original_data = None
+            for st in steps:
+                if st.get('step_number') == 5 and st.get('step_data'):
+                    try:
+                        sd = json.loads(st['step_data']) if isinstance(st['step_data'], str) else st['step_data']
+                        if sd.get('product_data'):
+                            original_data = sd['product_data']
+                            break
+                    except Exception:
+                        pass
+            
+            # 1. 大模型分析拒绝理由，给出缺失项和搜索建议
+            analyze_prompt = (
+                f"产品「{product_name}」提交到参数库审核被拒绝。\n"
+                f"拒绝理由：{reason}\n\n"
+                f"当前已提交的数据：\n{json.dumps(original_data or {}, ensure_ascii=False, indent=2)}\n\n"
+                "请分析：\n"
+                "1. 根据拒绝理由，判断审核方最关注哪些缺失/错误的信息\n"
+                "2. 给出需要重点补充的字段（如 parameters/context_length/publish_date/价格等）\n"
+                "3. 给出3-4个最有效的搜索关键词（中文或英文），用于搜索补充这些信息\n\n"
+                "只输出JSON：\n"
+                "{\"analysis\": \"分析结论\", \"missing_fields\": [\"字段名\"], \"search_keywords\": [\"关键词1\", \"关键词2\"]}"
+            )
+            
+            ok, result = llm_client.chat(
+                [{'role': 'user', 'content': analyze_prompt}],
+                temperature=0.2,
+                max_tokens=4096,
+                timeout=300
+            )
+            
+            keywords = []
+            missing_fields = []
+            if ok:
+                parsed = llm_client._extract_json(result)
+                if parsed:
+                    keywords = parsed.get('search_keywords', [])
+                    missing_fields = parsed.get('missing_fields', [])
+                    logger.info(f"[{session_id}] 复盘分析: 缺失字段={missing_fields}, 搜索词={keywords}")
+            
+            if not keywords:
+                # 兜底关键词
+                keywords = [f'{product_name} 参数 规格', f'{product_name} 发布 价格']
+            
+            # 2. 定向搜索补齐
+            all_data = {'library_results': [], 'internet_results': [], 'fetched_contents': [], 'extracted_data': None, 'filled_data': None}
+            new_fetched = 0
+            
+            for kw in keywords[:4]:
+                try:
+                    internet_results = search_service.search_internet(kw, max_results=5, use_cache=False)
+                except Exception as e:
+                    logger.warning(f"[{session_id}] 复盘搜索失败 [{kw}]: {e}")
+                    continue
+                
+                for r in internet_results:
+                    url = r.get('url', '')
+                    if not url:
+                        continue
+                    existing = db.search_articles(url)
+                    if existing and len(existing) > 0:
+                        continue
+                    
+                    fetch_result = search_service.fetch_url_content(url)
+                    if fetch_result.get('success'):
+                        try:
+                            article_id = db.add_article(
+                                product_names=[product_name],
+                                category=category or '',
+                                keywords=[kw],
+                                summary=fetch_result.get('description', '')[:200],
+                                content=fetch_result.get('content', ''),
+                                source=url,
+                                url=url,
+                                search_title=fetch_result.get('title', url[:50])
+                            )
+                            all_data['fetched_contents'].append({
+                                'id': article_id,
+                                'url': url,
+                                'title': fetch_result.get('title', ''),
+                                'content': (fetch_result.get('content') or '')[:500]
+                            })
+                            new_fetched += 1
+                        except Exception as e:
+                            logger.warning(f"[{session_id}] 复盘保存失败 {url}: {e}")
+                    time.sleep(0.3)
+            
+            if new_fetched == 0:
+                logger.warning(f"[{session_id}] 复盘未获取新内容，无法重新提交")
+                return
+            
+            # 3. 重新提取+填充（直接调用步骤4/5 的核心逻辑，复用 _run_process 的片段）
+            # 构建新会话数据
+            all_data['library_results'] = []
+            
+            # 步骤4：提取
+            task_text = self._build_agent_task(product_name, category, subcategory, all_data)
+            agent_result = self._call_llm(task_text)
+            
+            if not agent_result.get('success'):
+                logger.error(f"[{session_id}] 复盘提取失败: {agent_result.get('error')}")
+                return
+            
+            parsed = self._parse_agent_response(agent_result.get('output', ''))
+            relevant_ids = parsed.get('relevant_ids', []) if parsed else []
+            
+            # 兜底：没筛出就用全部新抓内容
+            if not relevant_ids:
+                relevant_ids = [c['id'] for c in all_data['fetched_contents']]
+            
+            relevant_contents = []
+            for aid in relevant_ids:
+                article = db.get_article_by_id(aid)
+                if article:
+                    relevant_contents.append({
+                        'id': aid,
+                        'title': article.get('search_title', ''),
+                        'url': article.get('url', ''),
+                        'content': article.get('content', ''),
+                        'summary': article.get('summary', ''),
+                        'analysis': ''
+                    })
+            
+            if not relevant_contents:
+                logger.warning(f"[{session_id}] 复盘无相关内容可提取")
+                return
+            
+            all_data['extracted_data'] = {
+                'name': product_name,
+                'relevant_ids': relevant_ids,
+                'relevant_contents': relevant_contents,
+                'confidence': 'medium'
+            }
+            
+            # 步骤5：填充
+            fill_task_text = self._build_fill_fields_task(product_name, category, subcategory, all_data['extracted_data'])
+            fill_agent_result = self._call_llm(fill_task_text)
+            
+            if not fill_agent_result.get('success'):
+                logger.error(f"[{session_id}] 复盘填充失败: {fill_agent_result.get('error')}")
+                return
+            
+            fill_parsed = self._parse_fill_agent_response(fill_agent_result.get('output', ''))
+            if not fill_parsed or not fill_parsed.get('success'):
+                logger.error(f"[{session_id}] 复盘填充解析失败")
+                return
+            
+            product_data = fill_parsed.get('product_data', {})
+            
+            # 引用链接
+            reference_links = []
+            seen = set()
+            for item in relevant_contents:
+                url = item.get('url', '')
+                if url and url not in seen:
+                    seen.add(url)
+                    reference_links.append({'title': item.get('title', url[:60]), 'url': url})
+            if reference_links:
+                product_data['reference_links'] = reference_links
+            product_data['_review_retry'] = True
+            product_data['_original_review_id'] = review_id
+            
+            # 4. 重新提交
+            category_type = self._get_category_type(category)
+            success, new_review_id = paramhub_client.submit_for_review(
+                category_type, product_data, subcategory
+            )
+            
+            if success:
+                logger.info(f"[{session_id}] 复盘重新提交成功! 新review_id={new_review_id}")
+                try:
+                    db.add_process_history(
+                        product_name=product_name,
+                        category=category,
+                        subcategory=subcategory,
+                        status='resubmitted',
+                        review_id=new_review_id,
+                        details={'message': f'审核被拒后复盘重新提交，原review_id={review_id}', 'reject_reason': reason}
+                    )
+                except Exception:
+                    pass
+                
+                # 新提交也启动监控（避免递归过深，只监控一轮）
+                # 这里不再递归监控，记录即可
+            else:
+                logger.error(f"[{session_id}] 复盘重新提交失败: {new_review_id}")
+        
+        except Exception as e:
+            logger.error(f"[{session_id}] 审核复盘异常: {e}")
+    
     def _build_submit_task(self, product_name, category, subcategory, product_data):
         """构建步骤6提交审核的智能体任务文本"""
         # 读取模板

services/search_service.py

@@ -7,6 +7,8 @@ import json
 import subprocess
 import os
 import re
+import time
+import threading
 import urllib.parse
 from datetime import datetime
 from config import Config
@@ -16,22 +18,60 @@ class SearchService:
     def __init__(self):
         self.timeout = Config.SEARCH_TIMEOUT
         self.max_results = Config.SEARCH_MAX_RESULTS
+        self._ns_counter = 0
+        self._ns_lock = threading.Lock()
     
-    def _run_browser(self, *args, timeout=30000):
-        """运行 agent-browser 命令"""
+    def _new_namespace(self):
+        """生成独立浏览器 namespace，避免与其他流程/服务并发冲突"""
+        with self._ns_lock:
+            self._ns_counter += 1
+            return f"search_{int(time.time())}_{self._ns_counter}"
+    
+    def _run_browser(self, *args, timeout=30000, namespace=None, retries=2):
+        """运行 agent-browser 命令（带 namespace 隔离 + 自动重试）"""
         env = os.environ.copy()
-        env['XDG_RUNTIME_DIR'] = '/tmp/agent-browser-runtime'
+        env['XDG_RUNTIME_DIR'] = '/tmp/xdg-rt'
         os.makedirs(env['XDG_RUNTIME_DIR'], exist_ok=True)
+        try:
+            os.chmod(env['XDG_RUNTIME_DIR'], 0o700)
+        except OSError:
+            pass
         
-        cmd = ['agent-browser'] + list(args)
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            env=env,
-            timeout=timeout // 1000 + 5
-        )
-        return result.stdout, result.stderr, result.returncode
+        cmd = ['agent-browser']
+        if namespace:
+            cmd += ['--namespace', namespace]
+        cmd += list(args)
+        
+        last_err = None
+        for attempt in range(retries + 1):
+            try:
+                result = subprocess.run(
+                    cmd,
+                    capture_output=True,
+                    text=True,
+                    env=env,
+                    timeout=timeout // 1000 + 5
+                )
+                if result.returncode == 0:
+                    return result.stdout, result.stderr, result.returncode
+                last_err = result.stderr or result.stdout
+                # 瞬时错误自动重试
+                if attempt < retries and any(t in str(last_err) for t in (
+                    'ERR_EMPTY_RESPONSE', 'ERR_CONNECTION_REFUSED', 'ERR_CONNECTION_RESET',
+                    'ERR_CONNECTION_CLOSED', 'ERR_TIMED_OUT', 'ERR_NAME_NOT_RESOLVED',
+                    'ERR_SOCKET_NOT_CONNECTED', 'ERR_ADDRESS_UNREACHABLE', 'ERR_NETWORK_CHANGED',
+                    'ERR_INTERNET_DISCONNECTED', 'session already exists'
+                )):
+                    time.sleep(2 * (attempt + 1))
+                    continue
+                return result.stdout, result.stderr, result.returncode
+            except subprocess.TimeoutExpired:
+                last_err = '命令超时'
+                if attempt < retries:
+                    time.sleep(2 * (attempt + 1))
+                    continue
+                return '', f'超时: {" ".join(args)}', 1
+        return '', str(last_err), 1
     
     def search_internet(self, keyword, max_results=None, engine='bing_cn', use_cache=True, cache_days=7):
         """
@@ -68,18 +108,21 @@ class SearchService:
         
         search_url = search_urls.get(engine, search_urls['bing_cn'])
         
+        # 本次搜索使用独立 namespace，避免并发冲突
+        ns = self._new_namespace()
+        
         try:
             # 1. 打开搜索引擎
-            stdout, stderr, code = self._run_browser('open', search_url, '--timeout', '20000')
+            stdout, stderr, code = self._run_browser('open', search_url, '--timeout', '20000', namespace=ns)
             if code != 0:
                 print(f"打开搜索页面失败: {stderr}")
                 return results
             
             # 等待页面加载
-            stdout, stderr, code = self._run_browser('wait', '5000')
+            stdout, stderr, code = self._run_browser('wait', '5000', namespace=ns)
             
             # 2. 获取搜索结果页面结构 (JSON 格式)
-            stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '30000')
+            stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '30000', namespace=ns)
             if code != 0:
                 print(f"获取页面结构失败: {stderr}")
                 return results
@@ -95,10 +138,10 @@ class SearchService:
             if engine == 'baidu':
                 results = self._parse_baidu_results(data, max_results)
             else:
-                results = self._parse_bing_results(data, max_results)
+                results = self._parse_bing_results(data, max_results, ns)
             
             # 5. 关闭浏览器
-            self._run_browser('close')
+            self._run_browser('close', namespace=ns)
             
             # 6. 保存到缓存
             if results and use_cache:
@@ -110,13 +153,13 @@ class SearchService:
             print(f"搜索出错: {str(e)}")
             # 尝试关闭浏览器
             try:
-                self._run_browser('close')
+                self._run_browser('close', namespace=ns)
             except:
                 pass
         
         return results
     
-    def _parse_bing_results(self, snapshot_data, max_results=10):
+    def _parse_bing_results(self, snapshot_data, max_results=10, ns=None):
         """
         从 Bing 搜索结果的 snapshot 中解析出标题和链接
         
@@ -168,7 +211,7 @@ class SearchService:
         for title, ref in refs[:max_results + 5]:
             if len(results) >= max_results:
                 break
-            url = self._get_link_url(ref)
+            url = self._get_link_url(ref, ns)
             if url and 'bing.com/search' not in url:  # 过滤搜索结果页本身的链接
                 results.append({
                     'title': title,
@@ -179,7 +222,7 @@ class SearchService:
         
         return results
     
-    def _parse_baidu_results(self, snapshot_data, max_results=10):
+    def _parse_baidu_results(self, snapshot_data, max_results=10, ns=None):
         """从百度搜索结果中解析标题和链接"""
         results = []
         
@@ -206,7 +249,7 @@ class SearchService:
         
         # 获取每个结果的 URL
         for title, ref in refs[:max_results]:
-            url = self._get_link_url(ref)
+            url = self._get_link_url(ref, ns)
             if url and 'baidu.com' not in url:
                 results.append({
                     'title': title,
@@ -217,10 +260,10 @@ class SearchService:
         
         return results
 
-    def _get_link_url(self, ref):
+    def _get_link_url(self, ref, ns=None):
         """通过 agent-browser 获取链接的 URL"""
         try:
-            stdout, stderr, code = self._run_browser('get', 'attr', f'@{ref}', 'href', '--json', '--timeout', '5000')
+            stdout, stderr, code = self._run_browser('get', 'attr', f'@{ref}', 'href', '--json', '--timeout', '5000', namespace=ns)
             if code == 0 and stdout:
                 data = json.loads(stdout)
                 return data.get('data', {}).get('value', '')
@@ -231,9 +274,10 @@ class SearchService:
     def fetch_url_content(self, url):
         """抓取网页内容（使用 agent-browser 浏览器方式，绕过反爬虫）"""
         error_message = None
+        ns = self._new_namespace()
         try:
             # 使用浏览器方式抓取，增加超时时间到60秒
-            stdout, stderr, code = self._run_browser('open', url, '--timeout', '60000')
+            stdout, stderr, code = self._run_browser('open', url, '--timeout', '60000', namespace=ns)
             if code != 0:
                 error_message = stderr.strip() if stderr else '浏览器打开页面失败'
                 print(f"打开页面失败: {stderr}")
@@ -244,14 +288,14 @@ class SearchService:
                 return {'success': False, 'error': error_message}
             
             # 等待页面加载（增加到10秒）
-            self._run_browser('wait', '10000')
+            self._run_browser('wait', '10000', namespace=ns)
             
             # 获取页面标题
-            stdout, stderr, code = self._run_browser('get', 'title', '--timeout', '5000')
+            stdout, stderr, code = self._run_browser('get', 'title', '--timeout', '5000', namespace=ns)
             title = stdout.strip().replace('[agent-browser] ', '').strip() if code == 0 else ''
             
             # 获取页面内容（通过 snapshot 获取 accessibility tree）
-            stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '15000')
+            stdout, stderr, code = self._run_browser('snapshot', '--json', '--timeout', '15000', namespace=ns)
             text = ''
             if code == 0 and stdout:
                 try:
@@ -263,11 +307,11 @@ class SearchService:
                     pass
             
             # 获取 URL（可能被重定向）
-            stdout, stderr, code = self._run_browser('get', 'url', '--timeout', '5000')
+            stdout, stderr, code = self._run_browser('get', 'url', '--timeout', '5000', namespace=ns)
             actual_url = stdout.strip() if code == 0 else url
             
             # 关闭浏览器
-            self._run_browser('close')
+            self._run_browser('close', namespace=ns)
             
             # 提取描述（从页面内容的前200字符）
             description = text[:200].strip() if text else ''
@@ -285,7 +329,7 @@ class SearchService:
             print(f"抓取URL失败: {url}, 错误: {error_message}")
             # 尝试关闭浏览器
             try:
-                self._run_browser('close')
+                self._run_browser('close', namespace=ns)
             except:
                 pass
             return {'success': False, 'error': error_message}

utils/scheduler.py

@@ -5,7 +5,6 @@ from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.interval import IntervalTrigger
 from datetime import datetime
 from models.database import db
-from services.process_service import process_service
 from config import Config
 import logging
 
@@ -63,8 +62,11 @@ task_scheduler = TaskScheduler()
 
 def auto_process_task():
     """
-    自动处理产品的定时任务
+    自动处理产品的定时任务（大模型驱动）
+    通过 process_monitor 启动异步处理会话，避免与手动处理并发冲突
     """
+    from services.process_monitor import process_monitor
+    
     try:
         # 检查是否启用自动处理
         enabled = db.get_system_config('auto_process_enabled', 'true')
@@ -86,33 +88,18 @@ def auto_process_task():
         
         for product in products:
             try:
-                # 检查是否正在处理
-                processing = db.get_processing_products()
-                if any(p['product_name'] == product['product_name'] for p in processing):
-                    logger.warning(f"产品 {product['product_name']} 正在处理中，跳过")
-                    continue
-                
-                # 添加到处理中列表
-                db.start_processing(
+                # 启动大模型处理会话（内部有防重，已有活跃会话会自动跳过）
+                session_id, started = process_monitor.start_process(
                     product_name=product['product_name'],
                     category=product.get('category'),
                     subcategory=product.get('subcategory')
                 )
-                
-                # 执行处理
-                result = process_service.process_product(product)
-                
-                # 从待处理列表移除
-                db.remove_pending_product(product['product_name'])
-                
-                logger.info(f"产品 {product['product_name']} 处理完成: {result['message']}")
-                
+                if started:
+                    logger.info(f"产品 {product['product_name']} 处理会话已启动: {session_id}")
+                else:
+                    logger.info(f"产品 {product['product_name']} 已有处理会话: {session_id}")
             except Exception as e:
-                logger.error(f"处理产品 {product['product_name']} 时出错: {str(e)}")
-                db.finish_processing(product['product_name'])
-            finally:
-                # 确保从处理中列表移除
-                db.finish_processing(product['product_name'])
+                logger.error(f"启动产品 {product['product_name']} 处理会话时出错: {str(e)}")
         
     except Exception as e:
         logger.error(f"自动处理任务执行失败: {str(e)}")