""" Phase 1: 历史招投标相似度分析 对比新招标文件与各历史招标文件的相似度， 决定每组历史数据对最终投标文件的影响权重。 用法: from phase1_parsing.analyze_similarity import calculate_similarity results = calculate_similarity(new_tender_text, historical_groups) """ import sys import os import json import re sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from llm.llm_utils import LLM def extract_tender_features(text: str) -> dict: """从招标文本中提取关键特征用于相似度比较""" features = {} # 项目类型判断 type_keywords = { "物业管理": ["物业", "保洁", "保安", "绿化", "维修", "会务", "客服"], "工程装修": ["装修", "改造", "修缮", "施工", "工程"], "IT服务": ["软件", "系统", "信息化", "开发", "运维", "网络安全"], "物资采购": ["采购", "设备", "货物", "供应", "物资"], "咨询服务": ["咨询", "评估", "审计", "监理", "设计"], "餐饮服务": ["餐饮", "食堂", "供餐", "配餐"], } matched_types = [] for ttype, kws in type_keywords.items(): if any(kw in text[:8000] for kw in kws): matched_types.append(ttype) features["project_types"] = matched_types # 采购人（同一采购人很可能项目相似） m = re.search(r"采购人[：:]\s*(.{2,30})", text[:5000]) if m: features["buyer"] = m.group(1).strip() # 预算规模 m = re.search(r"预算[^\n]{0,10}[：:]\s*([\d,.]+\s*万)", text[:10000]) if m: budget_str = m.group(1).strip().replace(",", "") try: budget_val = float(budget_str.replace("万", "")) features["budget"] = budget_val if budget_val > 1000: features["budget_level"] = "大型" elif budget_val > 200: features["budget_level"] = "中型" else: features["budget_level"] = "小型" except ValueError: pass # 服务期限 m = re.search(r"服务期限[：:](.{0,40})", text[:5000]) if m: features["service_period"] = m.group(1).strip() # 地域 m = re.search(r"([一-鿿]{2,8}(?:区|市|县))", text[:3000]) if m: features["region"] = m.group(1).strip() # 关键需求关键词密度 demand_kw = ["服务方案", "质量标准", "人员配置", "管理制度", "应急预案", "考核", "培训", "设备", "安全", "合同"] found_kw = [kw for kw in demand_kw if kw in text] features["demand_keywords"] = found_kw return features def calculate_similarity( new_tender_text: str, historical_groups: list, use_llm: bool = True, ) -> list: """ 计算新招标文件与每组历史招标文件的相似度。 Args: new_tender_text: 新招标文件全文 historical_groups: [ { "group_id": "project_A", "name": "项目名称", "tender_text": "历史招标文本", "bid_text": "对应投标文本", "tender_file": "原始PDF路径", "bid_file": "原始PDF路径", }, ... ] use_llm: 是否用 LLM 做语义级评分（否则只用关键词匹配） Returns: [ { "group_id": "project_A", "name": "项目名称", "similarity_score": 0.85, # 0~1 "similarity_level": "高", # 高/中/低 "similarity_reason": "同为政府机关物业管理...", "new_tender_features": {...}, "history_tender_features": {...}, "bid_structure": {...}, # 对应投标的目录结构 "weight": 0.6, # 归一化后的权重 }, ... ] """ new_features = extract_tender_features(new_tender_text) results = [] for group in historical_groups: hist_features = extract_tender_features(group.get("tender_text", "")) if use_llm: score, reason = _llm_similarity_score( new_tender_text[:8000], group.get("tender_text", "")[:8000], ) else: score, reason = _keyword_similarity(new_features, hist_features) # 提取投标目录 from phase1_parsing.extract_bid_toc import extract_toc bid_toc = extract_toc(group.get("bid_text", "")) # 判断相似等级 if score >= 0.7: level = "高" elif score >= 0.4: level = "中" else: level = "低" results.append({ "group_id": group["group_id"], "name": group.get("name", group["group_id"]), "similarity_score": round(score, 4), "similarity_level": level, "similarity_reason": reason, "new_tender_features": new_features, "history_tender_features": hist_features, "bid_toc": bid_toc, "bid_text_len": len(group.get("bid_text", "")), "tender_text_len": len(group.get("tender_text", "")), }) # 归一化权重：相似度越高权重越大 scores = [r["similarity_score"] for r in results] total = sum(scores) or 1 for r in results: r["weight"] = round(r["similarity_score"] / total, 4) results.sort(key=lambda x: x["similarity_score"], reverse=True) return results def _keyword_similarity(new_feat: dict, hist_feat: dict) -> tuple: """基于关键词匹配的相似度计算（快速但粗糙）""" score = 0.0 reasons = [] # 项目类型匹配 new_types = set(new_feat.get("project_types", [])) hist_types = set(hist_feat.get("project_types", [])) if new_types and hist_types: overlap = new_types & hist_types if overlap: score += 0.3 reasons.append(f"类型匹配: {overlap}") # 采购人匹配 if (new_feat.get("buyer") and hist_feat.get("buyer") and new_feat["buyer"] == hist_feat["buyer"]): score += 0.25 reasons.append("同一采购人") # 预算规模匹配 if new_feat.get("budget_level") and hist_feat.get("budget_level"): if new_feat["budget_level"] == hist_feat["budget_level"]: score += 0.15 reasons.append(f"预算规模相当({new_feat['budget_level']})") # 需求关键词重叠 new_kw = set(new_feat.get("demand_keywords", [])) hist_kw = set(hist_feat.get("demand_keywords", [])) if new_kw and hist_kw: overlap = new_kw & hist_kw if overlap: kw_score = len(overlap) / max(len(new_kw), 1) * 0.2 score += kw_score reasons.append(f"需求关键词重叠: {overlap}") # 地域匹配 if (new_feat.get("region") and hist_feat.get("region") and new_feat["region"] == hist_feat["region"]): score += 0.1 reasons.append("同区域") score = min(score, 1.0) reason = "；".join(reasons) if reasons else "无明显相似特征" return score, reason def _llm_similarity_score(new_excerpt: str, hist_excerpt: str) -> tuple: """用 LLM 评估两个招标文件的相似度（语义级）""" try: llm = LLM() system = ( "你是一个招标文件比对专家。比较两份招标文件，输出相似度评分(0~1)和理由。\n" "输出 JSON: {\"score\": 0.85, \"reason\": \"同类型项目，服务内容重叠度高\"}" ) user = ( "请比较以下两份招标文件的相似度，考虑项目类型、服务内容、预算规模、采购人等因素：\n\n" f"【新招标文件】\n{new_excerpt[:4000]}\n\n" f"【历史招标文件】\n{hist_excerpt[:4000]}" ) result = llm._call(system, user, max_tokens=1024, temperature=0.2, response_format={"type": "json_object"}) parsed = json.loads(result) return float(parsed.get("score", 0.5)), parsed.get("reason", "") except Exception as e: return 0.5, f"LLM评分异常: {e}" if __name__ == "__main__": import sys sys.stdout.reconfigure(encoding="utf-8") # 简单测试 text1 = "项目名称：松江区机关事务管理局物业管理服务 预算金额：1739.51万元 服务内容：保洁、保安、绿化、会务" text2 = "项目名称：嘉定区机关事务管理局物业管理服务 预算金额：1200万元 服务内容：保洁、保安、绿化" groups = [{ "group_id": "test", "name": "测试项目", "tender_text": text2, "bid_text": "第一章 投标人资格...", }] results = calculate_similarity(text1, groups, use_llm=False) for r in results: print(f"{r['name']}: 相似度 {r['similarity_score']} ({r['similarity_level']})") print(f" 理由: {r['similarity_reason']}") print(f" 权重: {r['weight']}")