#!/usr/bin/env python3 """ LambdaRAG v3.1: Intent-Entity-Fusion-Slot Pipeline Changes from v3: P0-1: Fix graph keywords error (ensure all paths return keywords) P0-2: Scoring-based intent classification (not first-match) P0-3: Query logging to feedback/query_log.jsonl P0-4: Missing query logging to feedback/missing_queries.jsonl P1-1: Corrective RAG — retrieval quality check + reformulate P1-2: Compare skips entity enhancement P1-5: Temporal time expansion ("近五年" → absolute years) Architecture: λ = IntentEntityExtract(q) >> EntityEnhancedFusionSearch(q, entities, intent) >> [Optional] RetrievalQualityCheck → Reformulate+Retry >> SlotFillFromResults(intent, entities, results) >> BuildEnrichedContext(slots, results, relations) >> LogQuery """ import json import os import re import sys import time import urllib.request from collections import defaultdict from datetime import datetime from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed # ── Imports ── try: from search_engine import search as _bm25_search_fn except ImportError: _bm25_search_fn = None try: from search_engine_v2 import _vector_search as _vector_search_fn, _embed_query except ImportError: _vector_search_fn = None _embed_query = None try: from graph_engine import get_graph except ImportError: get_graph = None try: from config import cfg WIKI_DIR = str(cfg.wiki_dir) VLLM_URL = cfg.vllm_url WEIGHT_PROFILES = cfg.weight_profiles DEFAULT_WEIGHTS = cfg.default_weights RRF_K = cfg.rrf_k PROMPTS = cfg.raw.get('prompts', {}) FEEDBACK_DIR = cfg.base_dir / 'feedback' except ImportError: WIKI_DIR = os.environ.get('WIKI_DIR', str(Path(__file__).parent.parent / 'wiki')) VLLM_URL = os.environ.get('VLLM_URL', 'http://127.0.0.1:8000/v1/chat/completions') WEIGHT_PROFILES = { 'person': {'bm25': 0.8, 'vector': 0.5, 'graph': 0.3, 'wiki': 0.5}, 'process': {'bm25': 0.7, 'vector': 0.5, 'graph': 0.3, 'wiki': 0.6}, 'fact': {'bm25': 1.0, 'vector': 0.5, 'graph': 0.1, 'wiki': 0.1}, 'compare': {'bm25': 0.5, 'vector': 0.8, 'graph': 0.4, 'wiki': 0.7}, 'temporal': {'bm25': 1.0, 'vector': 0.2, 'graph': 0.05, 'wiki': 0.2}, 'data': {'bm25': 1.0, 'vector': 0.3, 'graph': 0.05, 'wiki': 0.1}, } DEFAULT_WEIGHTS = {'bm25': 0.7, 'vector': 0.5, 'graph': 0.3, 'wiki': 0.3} RRF_K = 60 PROMPTS = {} FEEDBACK_DIR = Path(os.environ.get('KNOWLEDGE_DIR', '.')) / 'feedback' # Ensure weights for all types for t, w in [('person', {'bm25':0.8,'vector':0.5,'graph':0.3,'wiki':0.5}), ('process', {'bm25':0.7,'vector':0.5,'graph':0.3,'wiki':0.6}), ('data', {'bm25':1.0,'vector':0.3,'graph':0.05,'wiki':0.1})]: if t not in WEIGHT_PROFILES: WEIGHT_PROFILES[t] = w CURRENT_YEAR = datetime.now().year def _llm_call(prompt, max_tokens=500): body = json.dumps({ "model": "qwen2.5-32b", "messages": [{"role": "user", "content": prompt}], "temperature": 0.0, "max_tokens": max_tokens, }).encode('utf-8') req = urllib.request.Request(VLLM_URL, data=body, headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) return data["choices"][0]["message"]["content"].strip() # ══════════════════════════════════════════ # P0-3 / P0-4: Feedback Logging # ══════════════════════════════════════════ def _log_query(query, intent, strategy, elapsed, path_counts, has_result): """P0-3: Append query record to feedback/query_log.jsonl""" try: FEEDBACK_DIR.mkdir(parents=True, exist_ok=True) record = { "ts": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"), "query": query[:200], "intent": intent, "strategy": strategy, "elapsed": elapsed, "path_counts": path_counts, "has_result": has_result, } with open(FEEDBACK_DIR / "query_log.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") except: pass # Logging should never crash the pipeline def _log_missing(query, intent): """P0-4: Log queries with no useful results""" try: FEEDBACK_DIR.mkdir(parents=True, exist_ok=True) record = { "ts": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"), "query": query[:200], "intent": intent, } with open(FEEDBACK_DIR / "missing_queries.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") except: pass # ══════════════════════════════════════════ # P0-2: Scoring-based Intent Classification # ══════════════════════════════════════════ # Intent keyword rules with scores (higher = stronger signal) _INTENT_RULES = { 'person': [ (r'谁负责|找谁|联系方式|打什么电话|哪位老师|联系人|哪个部门管辖|主管机关', 5), ], 'process': [ (r'怎么办|操作步骤|怎么报销|审批流程|怎么预约|怎么退回|怎么申请|办理流程', 5), (r'能不能报销|可以报销吗|是否可以|能否|需要.*吗|要不要', 4), (r'报销|预约|退回|建账|扣税|签字|审批|发放|申请|备案|登记', 2), ], 'data': [ (r'及格率|通过率|排名|统计数据|人次.*统计|吞吐量|收费.*标准.*表|费用.*表', 5), (r'清单.*内容|哪些.*机构|标准.*表', 3), ], 'fact': [ (r'\d{4}年.*多少|数量.*多少|人数|统计|税率|额度|识别号|有效期', 4), (r'标准.*多少|是什么|是多少|包括哪些|有哪些', 2), ], 'compare': [ (r'异同|比较|区别|对比|不同|相同|差异', 5), ], 'relation': [ (r'隶属|上级|下级|依据|谁.*管|属于|归.*管|由谁制定|哪个.*认可', 4), ], 'synthesis': [ (r'总结|综合|梳理|分析.*体系|框架|概述|全面|整体', 4), ], 'temporal': [ (r'最新|修订|什么时候|截止|几月份|何时.*施行|变化.*趋势', 4), (r'近.*年|历史|演变', 3), ], } def _score_intent(query): """P0-2: Score-based classification — highest score wins""" q = query.lower() scores = defaultdict(float) for intent, rules in _INTENT_RULES.items(): for pattern, weight in rules: if re.search(pattern, q): scores[intent] += weight if not scores: return None # Return highest scoring intent best = max(scores.items(), key=lambda x: x[1]) return best[0] if best[1] >= 2 else None # Minimum threshold of 2 # ══════════════════════════════════════════ # Step 1: Intent + Entity Extraction # ══════════════════════════════════════════ def intent_entity_extract(query): """ Classify intent (scoring-based) + extract entities. Fast path for most intents, LLM only for person/process or ambiguous. """ # P0-2: Score-based classification fast_intent = _score_intent(query) # Fast path: rule-classified non-person/process → skip LLM if fast_intent and fast_intent not in ('person', 'process'): entities = _rule_based_entities(query) return { 'intent': fast_intent, 'entities': entities, 'keywords': list(entities.values())[:5], 'original_query': query, } # LLM extraction for: unknown / person / process try: prompt = f"""分析以下问题,输出JSON(不要其他文字): {{ "intent": "person/process/fact/data/compare/temporal 之一", "entities": {{"关键实体名": "值"}}, "keywords": ["扩展关键词"] }} 问题:{query}""" result = _llm_call(prompt, max_tokens=200) match = re.search(r'\{[\s\S]*\}', result) if match: parsed = json.loads(match.group()) intent = fast_intent or parsed.get('intent', 'fact') entities = parsed.get('entities', {}) keywords = parsed.get('keywords', []) return { 'intent': intent, 'entities': entities, 'keywords': keywords, 'original_query': query, } except Exception as e: print(f"[intent_entity] LLM error: {e}") return { 'intent': fast_intent or 'fact', 'entities': _rule_based_entities(query), 'keywords': [], 'original_query': query, } def _rule_based_entities(query): """Extract entities from query using simple rules (no LLM)""" entities = {} quoted = re.findall(r'[《「](.+?)[》」]', query) for q in quoted: entities[q] = q nouns = re.findall(r'[\u4e00-\u9fff]{2,8}', query) seen = set() for n in nouns: if n not in seen and len(n) >= 2: seen.add(n) entities[n] = n if len(entities) >= 5: break return entities # ══════════════════════════════════════════ # P1-5: Temporal Time Expansion # ══════════════════════════════════════════ def _expand_temporal_query(query): """Expand vague time words to absolute years""" expanded = query year = CURRENT_YEAR expanded = re.sub(r'近五年|近5年', f'{year-4}年至{year}年', expanded) expanded = re.sub(r'近三年|近3年', f'{year-2}年至{year}年', expanded) expanded = re.sub(r'近年来|近年', f'{year-2}年至{year}年', expanded) expanded = re.sub(r'最新', f'{year}年最新', expanded) return expanded # ══════════════════════════════════════════ # Step 2: Entity-Enhanced 4-Path Fusion Search # ══════════════════════════════════════════ def entity_enhanced_search(query_info, top_k=5): query = query_info['original_query'] intent = query_info['intent'] entities = query_info.get('entities', {}) keywords = query_info.get('keywords', []) weights = WEIGHT_PROFILES.get(intent, DEFAULT_WEIGHTS) # P1-2: Compare skips entity enhancement (entities dilute precise matching) if intent == 'compare': enhanced_query = query else: entity_terms = ' '.join(str(v) for v in entities.values() if v) keyword_terms = ' '.join(keywords[:5]) enhanced_query = f"{query} {entity_terms} {keyword_terms}".strip() # P1-5: Temporal time expansion if intent == 'temporal': enhanced_query = _expand_temporal_query(enhanced_query) entity_names = [str(v) for v in entities.values() if v and len(str(v)) > 1] path_results = {} with ThreadPoolExecutor(max_workers=4) as pool: futures = { pool.submit(_path_bm25, enhanced_query, 10): 'bm25', pool.submit(_path_vector, query, 10): 'vector', pool.submit(_path_graph, query, entity_names, 10): 'graph', pool.submit(_path_wiki, query, entity_names, 5): 'wiki', } for future in as_completed(futures): name = futures[future] try: path_results[name] = future.result(timeout=15) except Exception as e: print(f"[{name}] Error: {e}") path_results[name] = [] fused = _weighted_rrf_fuse(path_results, weights) if intent == 'temporal': fused = _time_filter(fused, query) if intent == 'data': fused = _boost_table_chunks(fused) return fused, path_results, weights # ══════════════════════════════════════════ # P1-1: Corrective RAG — Retrieval Quality Check # ══════════════════════════════════════════ def _check_retrieval_quality(query, fused_results, threshold=0.3): """ Quick check: do top results actually contain query-relevant content? Returns True if quality is acceptable, False if reformulation needed. Uses simple keyword overlap (no LLM call for speed). """ if not fused_results: return False # Check if top-3 results contain at least some query keywords query_chars = set(re.findall(r'[\u4e00-\u9fff]', query)) if not query_chars: return True top_text = ' '.join(r.get('text', '')[:300] for r in fused_results[:3]) overlap = sum(1 for c in query_chars if c in top_text) ratio = overlap / len(query_chars) return ratio >= threshold def _reformulate_query(query): """Simple query reformulation: extract core nouns""" nouns = re.findall(r'[\u4e00-\u9fff]{2,6}', query) if len(nouns) >= 2: return ' '.join(nouns[:4]) return query # ══════════════════════════════════════════ # 4-Path Retrieval Functions # ══════════════════════════════════════════ def _path_bm25(query, top_k=10): if _bm25_search_fn: try: results = _bm25_search_fn(query, top_k=top_k) for r in results: r['_path'] = 'bm25' return results except Exception as e: print(f"[bm25] Error: {e}") return [] def _path_vector(query, top_k=10): if _vector_search_fn: try: results = _vector_search_fn(query, top_k=top_k) for r in results: r['_path'] = 'vector' return results except Exception as e: print(f"[vector] Error: {e}") return [] def _path_graph(query, entity_names=None, top_k=10): if not get_graph: return [] try: graph = get_graph(wiki_dir=WIKI_DIR) seeds = entity_names or [] if not seeds: seeds = graph.find_entities(query, max_results=5) if not seeds: return [] expand = graph.expand_query(query, seed_entities=seeds, max_hops=2) expanded_query = expand.get('expanded_query', query) results = [] if _bm25_search_fn: r = _bm25_search_fn(expanded_query, top_k=top_k // 2) if r: results.extend(r) if _vector_search_fn: r = _vector_search_fn(expanded_query, top_k=top_k // 2) if r: results.extend(r) for rel in expand.get('relations', [])[:5]: results.append({ 'source': f"relation:{rel.get('subject','')}→{rel.get('object','')}", 'chunk_id': 0, 'text': f"{rel.get('subject','')} {rel.get('relation','')} {rel.get('object','')}。{rel.get('description', '')}", 'score': 0.5, }) for r in results: r['_path'] = 'graph' if results: results[0]['_relation_context'] = expand.get('relation_context', '') return results except Exception as e: print(f"[graph] Error: {e}") return [] def _path_wiki(query, entity_names=None, top_k=5): wiki_path = Path(WIKI_DIR) if not wiki_path.exists(): return [] pages = [] for subdir in ['sources', 'entities', 'topics', 'analyses']: d = wiki_path / subdir if not d.exists(): continue for f in d.glob('*.md'): try: content = f.read_text(encoding='utf-8') score = 0 if entity_names: for e in entity_names: if e in f.stem or e in content[:500]: score += 10 score += sum(1 for c in set(query) if c in content[:500]) if score > len(query) * 0.2: pages.append({ 'source': f'wiki:{subdir}/{f.stem}', 'chunk_id': 0, 'text': content[:800], 'score': score / 100.0, '_path': 'wiki', }) except: pass pages.sort(key=lambda x: -x['score']) return pages[:top_k] def _weighted_rrf_fuse(path_results, weights, k=RRF_K): scores = defaultdict(float) chunk_map = {} path_tags = defaultdict(set) for path_name, results in path_results.items(): w = weights.get(path_name, 0.5) if w == 0: continue for rank, r in enumerate(results): key = f"{r.get('source', '')}#{r.get('chunk_id', 0)}" scores[key] += w / (k + rank + 1) if key not in chunk_map: chunk_map[key] = r.copy() path_tags[key].add(path_name) sorted_keys = sorted(scores.items(), key=lambda x: -x[1]) fused = [] for key, score in sorted_keys: r = chunk_map[key] r['score'] = round(score, 4) r['_fusion_paths'] = list(path_tags[key]) fused.append(r) return fused def _boost_table_chunks(results): table_results = [] other_results = [] for r in results: text = r.get('text', '') if 'TABLE_START' in text or (text.count('|') > 10 and '---' in text): r['score'] = r.get('score', 0) + 0.1 table_results.append(r) else: other_results.append(r) return table_results + other_results def _time_filter(results, query): scored = [] for r in results: boost = 0 text = r.get('text', '') + r.get('source', '') if any(k in text for k in ['最新修订', '现行有效', '起施行']): boost += 2 src_years = re.findall(r'(20\d{2})', r.get('source', '')) if src_years: boost += (max(int(y) for y in src_years) - 2000) * 0.1 scored.append((r, r.get('score', 0) + boost)) scored.sort(key=lambda x: -x[1]) return [r for r, _ in scored] # ══════════════════════════════════════════ # Step 3: Slot Fill from Search Results # ══════════════════════════════════════════ def slot_fill(query_info, fused_results): intent = query_info['intent'] entities = query_info.get('entities', {}) query = query_info['original_query'] # Fast path: fact/compare/temporal/synthesis skip slot fill if intent in ('fact', 'compare', 'temporal', 'synthesis'): return {} context = '\n'.join([r.get('text', '')[:500] for r in fused_results[:5]]) if not context.strip(): return {} if intent == 'person' and 'person_extract' in PROMPTS: prompt = PROMPTS['person_extract'].format(context=context[:2500], question=query) elif intent == 'process' and 'process_extract' in PROMPTS: prompt = PROMPTS['process_extract'].format(context=context[:2500], question=query) elif intent == 'fact' and 'fact_extract' in PROMPTS: prompt = PROMPTS['fact_extract'].format(context=context[:2500], question=query) elif intent in ('relation', 'synthesis') and 'fact_extract' in PROMPTS: prompt = PROMPTS['fact_extract'].format(context=context[:2500], question=query) elif intent == 'data': entity_desc = ', '.join(f'{k}: {v}' for k, v in entities.items()) if entities else '' prompt = f"""基于以下知识库内容(可能包含表格),精确提取问题要求的数据。 仅使用知识库信息,不编造。 知识库内容: {context[:2500]} 问题:{query} 查找目标:{entity_desc} 输出JSON: {{"answer_summary": "直接回答(含具体数字)", "data_values": {{"指标名": "数值"}}, "table_source": "数据来源文件名", "source": "信息来源"}}""" else: entity_desc = ', '.join(f'{k}: {v}' for k, v in entities.items()) if entities else '' prompt = f"""基于以下知识库内容,提取关键信息。仅使用知识库信息,不编造。输出JSON。 知识库内容: {context[:2000]} 问题:{query} {{"answer_summary": "一句话回答", "key_values": {{"指标": "值"}}, "conditions": "适用条件", "policy_basis": "政策依据", "source": "来源文件名"}}""" try: result = _llm_call(prompt, max_tokens=500) match = re.search(r'\{[\s\S]*\}', result) if match: return json.loads(match.group()) except Exception as e: print(f"[slot_fill] Error: {e}") return {} # ══════════════════════════════════════════ # Step 4: Build Enriched Context # ══════════════════════════════════════════ def build_enriched_context(query_info, fused_results, slots, top_k=5): context_parts = [] relation_context = "" wiki_summaries = [] if slots: slots_text = "## 提取的关键信息\n" for k, v in slots.items(): if v and k != 'source': if isinstance(v, dict): for sk, sv in v.items(): if sv: slots_text += f"- {sk}: {sv}\n" else: slots_text += f"- {k}: {v}\n" context_parts.append(slots_text) for r in fused_results[:top_k]: source = r.get('source', '') text = r.get('text', '') if source.startswith('wiki:'): wiki_summaries.append(f"[Wiki: {source[5:]}]\n{text[:600]}") elif source.startswith('relation:'): if not relation_context: relation_context = "## 实体关系\n" relation_context += f"- {text}\n" else: context_parts.append(f"[doc: {source}]\n{text}") if '_relation_context' in r and r['_relation_context']: if not relation_context: relation_context = "## 实体关系\n" relation_context += r['_relation_context'] + "\n" enriched = "" if relation_context: enriched += relation_context + "\n" if wiki_summaries: enriched += "## Wiki 知识\n" + "\n".join(wiki_summaries) + "\n\n" enriched += "\n".join(context_parts) return enriched # ══════════════════════════════════════════ # Main search function # ══════════════════════════════════════════ def search(query, top_k=5): """ LambdaRAG v3.1 Pipeline: 1. Intent + Entity extraction (scoring-based, fast) 2. Entity-enhanced 4-path fusion search 2b. [P1-1] Retrieval quality check → reformulate if needed 3. Slot filling (person/process/data only) 4. Build enriched context 5. Log query + log missing """ t0 = time.time() # Step 1: Intent + Entity query_info = intent_entity_extract(query) intent = query_info['intent'] # Step 2: Entity-enhanced fusion search fused, path_results, weights = entity_enhanced_search(query_info, top_k) # Step 2b: P1-1 Corrective RAG — check retrieval quality if not _check_retrieval_quality(query, fused): # Reformulate and retry once alt_query = _reformulate_query(query) if alt_query != query: alt_info = {**query_info, 'original_query': alt_query} alt_fused, alt_paths, _ = entity_enhanced_search(alt_info, top_k) if _check_retrieval_quality(query, alt_fused): fused = alt_fused path_results = alt_paths # Step 3: Slot fill slots = slot_fill(query_info, fused) # Step 4: Build enriched context enriched = build_enriched_context(query_info, fused, slots, top_k) elapsed = time.time() - t0 # Metadata strategy = f"v3.1({intent}|e={len(query_info.get('entities',{}))}|s={len(slots)})" pc = {k: len(v) for k, v in path_results.items()} if fused: fused[0]['_query_type'] = intent fused[0]['_entities'] = query_info.get('entities', {}) fused[0]['_slots'] = slots fused[0]['_weights'] = weights fused[0]['_strategy'] = strategy fused[0]['_total_time'] = round(elapsed, 3) fused[0]['_path_counts'] = pc fused[0]['_enriched_context'] = enriched[:3500] # Step 5: P0-3 + P0-4 Logging has_result = bool(fused) _log_query(query, intent, strategy, round(elapsed, 2), pc, has_result) if not has_result: _log_missing(query, intent) return fused[:top_k] # ══════════════════════════════════════════ # CLI test # ══════════════════════════════════════════ if __name__ == '__main__': query = sys.argv[1] if len(sys.argv) > 1 else '横向经费报销接待餐费的标准' print(f"Query: {query}") print("=" * 60) info = intent_entity_extract(query) print(f"Intent: {info['intent']}") print(f"Entities: {info['entities']}") print(f"Keywords: {info['keywords']}") print(f"\nSearching...") results = search(query, top_k=5) if results: meta = results[0] print(f"Time: {meta.get('_total_time', '?')}s") print(f"Strategy: {meta.get('_strategy', '?')}") print(f"Slots: {json.dumps(meta.get('_slots', {}), ensure_ascii=False)[:200]}") print(f"Path counts: {meta.get('_path_counts', {})}") for i, r in enumerate(results): paths = r.get('_fusion_paths', []) print(f"\n[{i+1}] score={r.get('score','?')} paths={paths} {r.get('source','?')[:60]}") # Show feedback log if FEEDBACK_DIR.exists(): log = FEEDBACK_DIR / "query_log.jsonl" if log.exists(): lines = log.read_text().strip().split('\n') print(f"\nFeedback log: {len(lines)} entries")