#!/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) _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 = _wiki_cfg.get('wiki', {}).get('dir', '/app/agentexample/qaagent67wiki/wiki') # 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 prompt = f"""你是一个海事领域问答助手。请严格基于以下参考资料回答问题。 资料来自多个来源(检索文档、知识图谱、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"""你是一个海事领域 wiki 知识库问答助手。 基于以下已编译的 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 maritime 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', } _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] = {'rag': None, 'wiki': None, 'question': question} 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'} t1 = threading.Thread(target=run_rag) t2 = threading.Thread(target=run_wiki) t3 = threading.Thread(target=run_baseline) t1.start() t2.start() t3.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 = r['rag'] is not None and r['wiki'] is not None # Save to history when both are 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'], 'rag': r['rag'], 'wiki': r['wiki'], } threading.Thread(target=save_record, args=(record,)).start() return jsonify({ 'status': 'done' if done else 'processing', 'rag': r['rag'], 'wiki': r['wiki'], 'baseline': r.get('baseline'), }) @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 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
LambdaRAG (Unified) Classify + Route + Retrieve
Ask to see RAG response
Wiki Mode Compile + Query
Ask to see Wiki response
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://0.0.0.0:8080") app.run(host='0.0.0.0', port=8080, debug=False, threaded=True)