diff --git a/config/agent_fill_fields_template.txt b/config/agent_fill_fields_template.txt index 37fc981..494b1b9 100644 --- a/config/agent_fill_fields_template.txt +++ b/config/agent_fill_fields_template.txt @@ -13,51 +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`：厂商/组织（如 DeepSeek） - - `name`：产品名称（保留原始名称） - - `publish_date`：如内容提及系列发布时间可提取 - - `series`：所属系列（如 V4 系列） +**提取原则（重要）：** +- 优先提取**型号级精确参数**：参数量、上下文长度、能力指标、价格、发布日期等 +- 找不到型号级参数时，**品牌/系列级信息用于填充基础字段**：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": [], @@ -68,8 +78,8 @@ ``` **注意：** -- 严格按照API文档的字段定义填充数据 -- 不要编造或推测任何参数，只使用内容库中实际存在的信息 -- 如果某些字段无法从内容中提取，可以留空或填写默认值 +- `data_sources` 必须是数组，每项包含 `id`、`title`、`url`（原文链接）、`used_for`（该来源提供了哪些字段） +- 不要编造或推测任何参数，只使用内容中实际存在的信息 +- 如果某些字段无法从内容中提取，可以留空 - 不要执行任何提交操作，只生成并输出数据 - 确保输出的JSON格式正确，可以被程序解析 diff --git a/services/paramhub_client.py b/services/paramhub_client.py index ffb6909..8ab6f57 100644 --- a/services/paramhub_client.py +++ b/services/paramhub_client.py @@ -126,6 +126,30 @@ class ParamHubClient: 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: diff --git a/services/process_monitor.py b/services/process_monitor.py index 193c4b0..73d2237 100644 --- a/services/process_monitor.py +++ b/services/process_monitor.py @@ -211,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: # 构建任务文本 @@ -311,14 +316,19 @@ class ProcessMonitor: '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: # 构建任务文本 @@ -341,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, @@ -350,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 ) @@ -385,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 @@ -403,7 +477,12 @@ 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}' @@ -705,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', []) @@ -714,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]) @@ -843,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提交审核的智能体任务文本""" # 读取模板