#!/usr/bin/env python3 """ Maritime QA Comparison Demo RAG vs Wiki side-by-side comparison With server-side conversation history persistence """ import json import os import sys import time import threading from datetime import datetime from flask import Flask, request, jsonify, render_template_string import yaml # Load paths from agent config _agent_cfg_path = os.environ.get('AGENT_CONFIG', '/app/agentexample/qaagent67/agent-config.yml') _wiki_cfg_path = os.environ.get('WIKI_CONFIG', '/app/agentexample/qaagent67wiki/agent-config.yml') _cfg = {} _wiki_cfg = {} if os.path.exists(_agent_cfg_path): with open(_agent_cfg_path) as f: _cfg = yaml.safe_load(f) if os.path.exists(_wiki_cfg_path): with open(_wiki_cfg_path) as f: _wiki_cfg = yaml.safe_load(f) # ═══ instance.yml overlay (domain-specific overrides) ═══ def _deep_merge(base, override): import copy result = copy.deepcopy(base) for k, v in (override or {}).items(): if k in result and isinstance(result[k], dict) and isinstance(v, dict): result[k] = _deep_merge(result[k], v) else: result[k] = copy.deepcopy(v) return result _instance_cfg_path = os.environ.get('INSTANCE_CONFIG', '') if not _instance_cfg_path: _base = (_cfg.get('knowledge', {}) or {}).get('baseDir', '') if _base: _cand = os.path.join(_base, 'instance.yml') if os.path.isfile(_cand): _instance_cfg_path = _cand if _instance_cfg_path and os.path.isfile(_instance_cfg_path): with open(_instance_cfg_path) as _f: _instance = yaml.safe_load(_f) or {} _cfg = _deep_merge(_cfg, _instance) print('[config] loaded instance overlay:', _instance_cfg_path) _knowledge = _cfg.get('knowledge', {}) KNOWLEDGE_BASE = _knowledge.get('baseDir', os.environ.get('KNOWLEDGE_DIR', '/data/knowledge/maritime')) sys.path.insert(0, KNOWLEDGE_BASE) try: from search_unified import search except ImportError: from search_engine import search from lambdagent.providers import create_provider app = Flask(__name__) HISTORY_FILE = os.environ.get('HISTORY_FILE', '/app/qademo/conversation_history.json') # WIKI_DIR — prefer instance.yml overlay (_cfg.wiki.dir), then standalone wiki config, then default WIKI_DIR = ( (_cfg.get('wiki', {}) or {}).get('dir', '') or (_wiki_cfg.get('wiki', {}) or {}).get('dir', '') or '/app/agentexample/qaagent67wiki/wiki' ) # ═══ Demo UI config (domain-specific, loaded from instance.yml) ═══ _demo = _cfg.get('demo', {}) or {} DEMO_TITLE = _demo.get('title', 'Knowledge QA Demo') DEMO_TAGLINE = _demo.get('tagline', 'RAG vs Wiki | Qwen2.5:32B local') DEMO_PLACEHOLDER = _demo.get('placeholder', 'Please enter a question...') DEMO_DOMAIN = _demo.get('domain', 'knowledge') DEMO_EXAMPLES = _demo.get('examples', []) # list of {q, label} DEMO_PORT = int(os.environ.get('PORT') or _demo.get('port', 8080)) DEMO_BIND = os.environ.get('BIND_HOST') or _demo.get('bind') or '0.0.0.0' # 左右面板默认选项: 海事域左边默认 LambdaRAG,其他域默认 Lite _is_maritime = 'maritime' in DEMO_DOMAIN.lower() or '海事' in DEMO_TITLE DEMO_LEFT_DEFAULT = _demo.get('leftDefault', 'rag' if _is_maritime else 'lite') DEMO_RIGHT_DEFAULT = _demo.get('rightDefault', 'wiki') def _html_attr_esc(s): """Escape for HTML attribute in double quotes.""" return (str(s).replace('&', '&').replace('<', '<').replace('>', '>') .replace('"', '"').replace("'", ''')) def _html_text_esc(s): return (str(s).replace('&', '&').replace('<', '<').replace('>', '>')) _OLD_TITLE_TITLE = 'Maritime QA Demo' _OLD_TITLE_H1 = '

Maritime QA Demo

' _OLD_TAGLINE = '

RAG vs Wiki | 1326 docs | Qwen2.5:32B local

' _OLD_PLACEHOLDER = 'placeholder="Please enter a maritime question..."' _OLD_EXAMPLES = ( ' Ship entry/exit reporting system\n' ' Crew competency certificates\n' ' Qiongzhou Strait routing system\n' ' Maritime law comparison\n' ' Crew development trends' ) def _apply_demo_config(html): html = html.replace(_OLD_TITLE_TITLE, '' + _html_text_esc(DEMO_TITLE) + '') html = html.replace(_OLD_TITLE_H1, '

' + _html_text_esc(DEMO_TITLE) + '

') html = html.replace(_OLD_TAGLINE, '

' + _html_text_esc(DEMO_TAGLINE) + '

') html = html.replace(_OLD_PLACEHOLDER, 'placeholder="' + _html_attr_esc(DEMO_PLACEHOLDER) + '"') if DEMO_EXAMPLES: lines = [] for e in DEMO_EXAMPLES: q = str(e.get('q', '')) label = str(e.get('label', q)) q_esc = _html_attr_esc(q) label_esc = _html_text_esc(label) lines.append(' ' + label_esc + '') html = html.replace(_OLD_EXAMPLES, '\n'.join(lines)) # Inject default panel selections as JS init_script = ( '\n\n' ) html = html.replace('', init_script + '') return html # LLM provider (lazy init) _provider = None def get_provider(): global _provider if _provider is None: _provider = create_provider('ollama', model='qwen2.5-32b', base_url='http://127.0.0.1:8000/v1', timeout=120) return _provider # ── Conversation history persistence ── _history_lock = threading.Lock() def load_history(): if os.path.exists(HISTORY_FILE): with open(HISTORY_FILE, 'r', encoding='utf-8') as f: return json.load(f) return [] def save_record(record): with _history_lock: history = load_history() history.append(record) with open(HISTORY_FILE, 'w', encoding='utf-8') as f: json.dump(history, f, ensure_ascii=False, indent=2) # ── RAG answer ── def rag_answer(question, top_k=5): t0 = time.time() provider = get_provider() results = search(question, top_k=top_k) # Build enriched context from fusion results enriched = results[0].get('_enriched_context', '') if results else '' if not enriched: parts = [] for i, r in enumerate(results): parts.append('[doc' + str(i+1) + ': ' + r['source'] + ']\n' + r['text']) enriched = '\n\n'.join(parts) enriched = enriched[:3500] # fit within 8192 context window _sys = _cfg.get('systemPrompt', '').strip() _sys_short = _sys[:500] if _sys else 'You are a ' + DEMO_DOMAIN + ' knowledge QA assistant.' prompt = f"""{_sys_short} 请严格基于以下参考资料回答问题。资料来自多个来源(检索文档、知识图谱、Wiki知识),请综合利用。 如果资料中没有相关信息,请说明"资料中未找到相关信息"。 回答要具体,引用原文,标注来源 [来源: 文件名]。 {enriched} ## 问题 {question} ## 回答""" answer = provider.chat([{'role': 'user', 'content': prompt}]) elapsed = time.time() - t0 # Extract unified search metadata query_type = results[0].get('_query_type', 'unknown') if results else 'unknown' strategy = results[0].get('_strategy', 'unknown') if results else 'unknown' search_time = results[0].get('_total_time', 0) if results else 0 return { 'answer': answer, 'sources': [r['source'] for r in results], 'elapsed': round(elapsed, 1), 'chunks_used': len(results), 'mode': 'LambdaRAG', 'query_type': query_type, 'strategy': strategy, 'search_time': search_time, } # ── Wiki answer ── def wiki_answer(question, top_k=5): t0 = time.time() provider = get_provider() from pathlib import Path wiki_path = Path(WIKI_DIR) wiki_pages = [] for subdir in ['sources', 'entities', 'topics', 'analyses']: d = wiki_path / subdir if d.exists(): for f in d.glob('*.md'): try: content = f.read_text(encoding='utf-8') q_chars = set(question) match_score = sum(1 for c in q_chars if c in content) if match_score > len(question) * 0.3: wiki_pages.append({ 'name': f.stem, 'content': content[:1500], 'score': match_score, }) except: pass wiki_pages.sort(key=lambda x: x['score'], reverse=True) wiki_pages = wiki_pages[:3] if wiki_pages: wiki_context = '\n\n'.join([ f'[Wiki: {p["name"]}]\n{p["content"]}' for p in wiki_pages ]) source_type = 'wiki' else: results = search(question, top_k=top_k) compile_context = '\n\n'.join([ f'[doc: {r["source"]}]\n{r["text"]}' for r in results ]) compile_prompt = f"""请阅读以下文档片段,提炼出与问题相关的核心知识点。 用结构化的方式组织,标注来源。 文档: {compile_context} 问题: {question} 请输出结构化的知识摘要:""" compiled = provider.chat([{'role': 'user', 'content': compile_prompt}]) wiki_context = compiled source_type = 'compiled' safe_name = question[:30].replace('/', '_').replace(' ', '_') analysis_path = wiki_path / 'analyses' / f'{safe_name}.md' analysis_path.parent.mkdir(parents=True, exist_ok=True) with open(analysis_path, 'w', encoding='utf-8') as f: f.write(f'# {question}\n\n{compiled}\n') answer_prompt = f"""You are a {DEMO_DOMAIN} wiki knowledge QA assistant. 基于以下已编译的 wiki 知识回答问题。回答要具体,标注来源。 ## Wiki 知识 {wiki_context} ## 问题 {question} ## 回答""" answer = provider.chat([{'role': 'user', 'content': answer_prompt}]) elapsed = time.time() - t0 return { 'answer': answer, 'source_type': source_type, 'wiki_pages_used': len(wiki_pages), 'elapsed': round(elapsed, 1), 'mode': 'Wiki', } # ── Async results store ── # -- BM25 Baseline answer (for comparison) -- def bm25_baseline_answer(question, top_k=5): t0 = time.time() provider = get_provider() try: from search_engine import search as bm25_only results = bm25_only(question, top_k=top_k) except Exception: results = [] context_parts = [] for i, r in enumerate(results): context_parts.append('[doc' + str(i+1) + ': ' + r['source'] + ']' + chr(10) + r['text']) context = (chr(10)*2).join(context_parts) prompt = ('You are a ' + DEMO_DOMAIN + ' QA assistant. Answer based on docs only.' + chr(10) + '## Docs' + chr(10) + context + chr(10) + '## Question' + chr(10) + question + chr(10) + '## Answer') answer = provider.chat([{'role': 'user', 'content': prompt}]) elapsed = time.time() - t0 return { 'answer': answer, 'sources': [r['source'] for r in results], 'elapsed': round(elapsed, 1), 'chunks_used': len(results), 'mode': 'BM25 Baseline', } # -- Direct LLM answer (no retrieval, pure parametric knowledge) -- def direct_llm_answer(question): t0 = time.time() provider = get_provider() prompt = ('You are a ' + DEMO_DOMAIN + ' QA assistant. ' 'Answer the following question using ONLY your own knowledge. ' 'If you are not sure, say so.\n\n' '## Question\n' + question + '\n\n## Answer') answer = provider.chat([{'role': 'user', 'content': prompt}]) elapsed = time.time() - t0 return { 'answer': answer, 'elapsed': round(elapsed, 1), 'mode': 'Direct LLM', 'chunks_used': 0, 'sources': [], } # ── Lite answer (BM25 + Wiki combined — qaagent67lite strategy) ── def lite_answer(question, top_k=5): """qaagent67lite: BM25 关键词检索 + Wiki 编译知识融合""" t0 = time.time() provider = get_provider() from pathlib import Path # Path 1: BM25 bm25_results = [] try: from search_engine import search as bm25_search bm25_results = bm25_search(question, top_k=top_k) except Exception: pass bm25_context = '\n\n'.join([ f'[BM25 doc: {r["source"]}]\n{r["text"]}' for r in bm25_results[:5] ]) if bm25_results else '' # Path 2: Wiki wiki_path = Path(WIKI_DIR) wiki_pages = [] for subdir in ['entities', 'topics', 'sources', 'analyses']: d = wiki_path / subdir if d.exists(): for f in d.glob('*.md'): try: content = f.read_text(encoding='utf-8') q_chars = set(question) match_score = sum(1 for c in q_chars if c in content) if match_score > len(question) * 0.3: wiki_pages.append({ 'name': f.stem, 'subdir': subdir, 'content': content[:1500], 'score': match_score, }) except Exception: pass wiki_pages.sort(key=lambda x: x['score'], reverse=True) wiki_pages = wiki_pages[:3] wiki_context = '\n\n'.join([ f'[Wiki/{p["subdir"]}: {p["name"]}]\n{p["content"]}' for p in wiki_pages ]) if wiki_pages else '' # Merge contexts merged = '' if wiki_context: merged += '## Wiki 编译知识\n' + wiki_context + '\n\n' if bm25_context: merged += '## BM25 检索片段\n' + bm25_context if not merged.strip(): return { 'answer': '知识库中未找到相关信息。', 'elapsed': round(time.time() - t0, 1), 'mode': 'Lite (BM25+Wiki)', 'bm25_chunks': 0, 'wiki_pages_used': 0, } _sys = _cfg.get('systemPrompt', '').strip() _sys_short = _sys[:500] if _sys else 'You are a ' + DEMO_DOMAIN + ' knowledge QA assistant.' prompt = f"""{_sys_short} 请基于以下两种来源回答问题。Wiki 编译知识是经过整理的结构化知识(优先参考),BM25 检索片段是原始文档片段(用于补充细节)。 如果两种来源有矛盾,以 Wiki 知识为准。标注信息来源。 {merged} ## 问题 {question} ## 回答""" answer = provider.chat([{'role': 'user', 'content': prompt}]) elapsed = time.time() - t0 return { 'answer': answer, 'elapsed': round(elapsed, 1), 'mode': 'Lite (BM25+Wiki)', 'bm25_chunks': len(bm25_results), 'wiki_pages_used': len(wiki_pages), 'sources': [r['source'] for r in bm25_results[:3]] + [p['name'] for p in wiki_pages], } _results = {} # ── API routes ── @app.route('/api/ask', methods=['POST']) def api_ask(): data = request.json question = data.get('question', '').strip() if not question: return jsonify({'error': 'question is required'}), 400 req_id = str(int(time.time() * 1000)) _results[req_id] = {'lite': None, 'rag': None, 'wiki': None, 'question': question} def run_lite(): try: _results[req_id]['lite'] = lite_answer(question) except Exception as e: _results[req_id]['lite'] = {'answer': f'Error: {e}', 'elapsed': 0, 'mode': 'Lite'} def run_rag(): try: _results[req_id]['rag'] = rag_answer(question) except Exception as e: _results[req_id]['rag'] = {'answer': f'Error: {e}', 'elapsed': 0, 'mode': 'RAG'} def run_wiki(): try: _results[req_id]['wiki'] = wiki_answer(question) except Exception as e: _results[req_id]['wiki'] = {'answer': f'Error: {e}', 'elapsed': 0, 'mode': 'Wiki'} def run_baseline(): try: _results[req_id]['baseline'] = bm25_baseline_answer(question) except Exception as e: _results[req_id]['baseline'] = {'answer': 'Error: ' + str(e), 'elapsed': 0, 'mode': 'BM25 Baseline'} def run_direct(): try: _results[req_id]['direct'] = direct_llm_answer(question) except Exception as e: _results[req_id]['direct'] = {'answer': 'Error: ' + str(e), 'elapsed': 0, 'mode': 'Direct LLM'} t0 = threading.Thread(target=run_lite) t1 = threading.Thread(target=run_rag) t2 = threading.Thread(target=run_wiki) t3 = threading.Thread(target=run_baseline) t4 = threading.Thread(target=run_direct) t0.start() t1.start() t2.start() t3.start() t4.start() return jsonify({'req_id': req_id, 'status': 'processing'}) @app.route('/api/result/') def api_result(req_id): if req_id not in _results: return jsonify({'error': 'not found'}), 404 r = _results[req_id] # Done when all 5 modes have results done = all(r.get(k) is not None for k in ('lite', 'rag', 'wiki', 'baseline', 'direct')) # Save to history when all done if done and not r.get('_saved'): r['_saved'] = True record = { 'id': req_id, 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'question': r['question'], 'lite': r['lite'], 'rag': r['rag'], 'wiki': r.get('wiki'), 'baseline': r.get('baseline'), 'direct': r.get('direct'), } threading.Thread(target=save_record, args=(record,)).start() return jsonify({ 'status': 'done' if done else 'processing', 'lite': r.get('lite'), 'rag': r.get('rag'), 'wiki': r.get('wiki'), 'baseline': r.get('baseline'), 'direct': r.get('direct'), }) @app.route('/api/history') def api_history(): history = load_history() history.reverse() # newest first return jsonify(history[:50]) # last 50 @app.route('/api/history/export') def api_history_export(): history = load_history() return jsonify({ 'exported_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'total_records': len(history), 'records': history, }) @app.route('/api/wiki/index') def api_wiki_index(): """List all wiki pages grouped by category""" from pathlib import Path wiki_path = Path(WIKI_DIR) result = {} for subdir in ['sources', 'entities', 'topics', 'analyses']: d = wiki_path / subdir pages = [] if d.exists(): for f in sorted(d.glob('*.md')): stat = f.stat() pages.append({ 'name': f.stem, 'path': f'{subdir}/{f.name}', 'size': stat.st_size, 'modified': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M'), }) result[subdir] = pages # Also read index.md idx_path = wiki_path / 'index.md' result['index_content'] = idx_path.read_text(encoding='utf-8') if idx_path.exists() else '' return jsonify(result) @app.route('/api/wiki/page/') def api_wiki_page(page_path): """Read a single wiki page""" from pathlib import Path wiki_path = Path(WIKI_DIR) full_path = wiki_path / page_path if not full_path.exists() or not str(full_path).startswith(str(wiki_path)): return jsonify({'error': 'not found'}), 404 content = full_path.read_text(encoding='utf-8') return jsonify({'path': page_path, 'content': content, 'size': len(content)}) @app.route('/api/rag/search') def api_rag_search(): """Search RAG chunks""" query = request.args.get('q', '').strip() top_k = int(request.args.get('k', 10)) if not query: return jsonify({'error': 'q parameter required'}), 400 results = search(query, top_k=top_k) return jsonify({'query': query, 'total_results': len(results), 'results': results}) @app.route('/api/rag/stats') def api_rag_stats(): """RAG index statistics""" import json as json_mod from pathlib import Path idx_path = Path(_knowledge.get('indexFile', KNOWLEDGE_BASE + '/rag_index.json')) if not idx_path.exists(): return jsonify({'error': 'index not built'}), 404 with open(idx_path) as f: data = json_mod.load(f) # Sample chunks for preview sample_chunks = data['chunks'][:20] for c in sample_chunks: pass # show full text return jsonify({ 'total_chunks': data['total_chunks'], 'total_files': data['total_files'], 'chunk_size': data['chunk_size'], 'keyword_index_size': data.get('keyword_index_size', 0), 'sample_chunks': sample_chunks, }) @app.route('/wiki') def wiki_page(): return render_template_string(WIKI_HTML) @app.route('/rag') def rag_page(): return render_template_string(RAG_HTML) @app.route('/') def index(): return _apply_demo_config(render_template_string(HTML_TEMPLATE)) HTML_TEMPLATE = r""" Maritime QA Demo

Maritime QA Demo

RAG vs Wiki | 1326 docs | Qwen2.5:32B local

Ship entry/exit reporting system Crew competency certificates Qiongzhou Strait routing system Maritime law comparison Crew development trends
-
Ask to see response
-
Ask to see comparison
History
Export
No history yet

All records are saved on the server. Click to export.

""" WIKI_HTML = r""" Wiki Browser

Wiki Browser

QA Demo | RAG Chunks

Wiki Index

Loading...
""" RAG_HTML = r""" RAG Chunks Browser

RAG Chunks Browser

QA Demo | Wiki
-
Total Chunks
-
Source Files
-
Chunk Size
-
Keywords
Sample Chunks (from index)
""" if __name__ == '__main__': print("Loading search index...") search("test", top_k=1) # warm up print("Index loaded. Starting server...") print("Access: http://" + DEMO_BIND + ":" + str(DEMO_PORT)) app.run(host=DEMO_BIND, port=DEMO_PORT, debug=False, threaded=True)