v2.0.1 修复搜索并发冲突与提交失败问题

- 搜索服务改用 namespace 隔离浏览器实例（参考 webtest-agent 方案），解决定时任务与手动处理并发调用 agent-browser 互相踢掉导致搜索结果为0的问题 - agent-browser 调用加瞬时错误自动重试，使用 /tmp/xdg-rt 目录 - 步骤4 提示词放宽：品牌/系列相关页面也纳入提取，无精确型号时兜底选最相关内容 - 步骤4 大模型返回空时自动兜底，不再直接跳过导致会话卡死 - 会话收尾修复：失败/无数据时会话状态正确标记，并自动移除待处理产品 - 处理入口统一：定时任务/单产品/批量处理均改为 process_monitor 大模型流程（防重） - paramhub_client 增加登录态自动恢复：401 或连接失败时自动重新登录重试 - 全流程实测通过：deepseek-v4-flash-0731 → review_id 19ed2daaabfe
2026-08-13 18:25:22 +08:00
parent 23991c3552
commit a0b870ee98
9 changed files with 280 additions and 160 deletions
@@ -28,6 +28,16 @@
 ### 3. 整理产品参数
 根据获取到的内容，提取并整理产品的各项参数，严格按照API文档中定义的字段格式填充。

+**提取原则：**
+- 优先从内容中提取精确型号对应的参数（参数量、上下文长度、性能指标等）
+- 如果找不到型号级参数，**品牌/系列级信息也可用于填充基础字段**：
+  - `organization`：厂商/组织（如 DeepSeek）
+  - `name`：产品名称（保留原始名称）
+  - `publish_date`：如内容提及系列发布时间可提取
+  - `series`：所属系列（如 V4 系列）
+- 不要编造或推测任何参数，只使用内容中实际存在的信息
+- 找不到的字段留空即可
+
 ### 4. 格式检查
 对生成的数据进行以下检查：
 - 必填字段是否齐全（name必须有值）
@@ -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**，除非所有内容都与该产品完全无关
@@ -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:
@@ -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,
@@ -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,
@@ -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,10 +120,7 @@ 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)}")
@@ -115,14 +129,8 @@ class ParamHubClient:
    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 +151,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 +198,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()
@@ -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):
        """执行处理流程"""
@@ -253,15 +260,57 @@ 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'] = '大模型未找到相关数据且无兜底内容'
                    else:
                        self._fail_step(session_id, 4, f"大模型调用失败: {agent_result.get('error', '未知错误')}")
                        result['message'] = f'大模型调用失败: {agent_result.get("error")}'
@@ -361,6 +410,35 @@ class ProcessMonitor:
                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]
@@ -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}
@@ -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)}")