diff --git a/dev/.env.example b/dev/.env.example new file mode 100644 index 0000000..af56ba1 --- /dev/null +++ b/dev/.env.example @@ -0,0 +1,39 @@ +# AI Answers Plugin — Dev Server Config +# Copy this to .env and fill in your values +# .env is gitignored and never committed + +# Ollama endpoint (required) +LLM_URL=http://localhost:11434/v1/chat/completions + +# Default model +LLM_MODEL=qwen3.5:9b + +# Max response tokens +LLM_MAX_TOKENS=200 + +# Response temperature (0.0 - 2.0) +LLM_TEMPERATURE=0.2 + +# Bearer token for authenticated LLM endpoints +# Leave empty if no Bearer token is needed for your server +LLM_API_KEY= + +# Live SearXNG instance for real search results +# Leave empty to use mock results +SEARXNG_URL= + +# Valkey for streaming (required) +# Start with: docker run -d --name dev-valkey -p 6379:6379 valkey/valkey:9-alpine +VALKEY_HOST=localhost +VALKEY_PORT=6379 + +# Dev server host and port +DEV_HOST=127.0.0.1 +DEV_PORT=5000 + +# Plugin settings +LLM_INTERACTIVE=true +LLM_QUESTION_MARK_REQUIRED=false +LLM_TABS=general,science,it,news +LLM_CONTEXT_DEEP_COUNT=5 +LLM_CONTEXT_SHALLOW_COUNT=15 diff --git a/dev/dev.py b/dev/dev.py index 330af4d..3b24c2e 100644 --- a/dev/dev.py +++ b/dev/dev.py @@ -1,346 +1,1473 @@ +""" +dev.py — AI Answers Plugin Development Server +Provides a full-featured local dev environment for ollama_answers.py. + +Features: + - Mock SearXNG environment so the plugin loads without a real SearXNG install + - Optional live SearXNG backend (set SEARXNG_URL to proxy real searches) + - Debug console with request/response logging + - Plugin API explorer (all routes: /ai-stream, /ai-status, /ai-models, /ai-conversation, /ai-auxiliary-search) + - Dark UI matching SearXNG's default theme with real CSS vars + - Config panel: LLM endpoint, model, token limits, temperature + - Live query testing with streaming output in the browser + +Usage: + cd dev/ + python dev.py + + Or with a live SearXNG backend: + SEARXNG_URL=http://localhost:8080 python dev.py + +Environment variables (all optional): + LLM_URL Ollama endpoint (default: http://localhost:11434/v1/chat/completions) + LLM_MODEL Model name (default: qwen3.5:9b) + LLM_MAX_TOKENS Max tokens (default: 200) + LLM_TEMPERATURE Temperature (default: 0.2) + SEARXNG_URL Live SearXNG URL (default: None, uses mock results) + DEV_PORT Port to listen on (default: 5000) + DEV_HOST Host to bind to (default: 127.0.0.1) +""" + import sys import os -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import json +import time import logging +import hashlib +import secrets +import urllib.request +import urllib.parse from types import ModuleType -from flask import Flask, request, redirect -logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') +# ── path setup ────────────────────────────────────────────────────────────── +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +# ── .env loading ───────────────────────────────────────────────────────────── +try: + from dotenv import load_dotenv + # look for .env in the project root (one level up from dev/) + _env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env') + if os.path.exists(_env_path): + load_dotenv(_env_path) + print(f" Loaded .env from {_env_path}") +except ImportError: + pass # python-dotenv not installed, ignore + +# ── logging ────────────────────────────────────────────────────────────────── +logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s %(levelname)-8s %(name)s: %(message)s', + datefmt='%H:%M:%S' +) +logger = logging.getLogger('dev') + +# ── env defaults ───────────────────────────────────────────────────────────── os.environ.setdefault('LLM_URL', 'http://localhost:11434/v1/chat/completions') +os.environ.setdefault('LLM_MODEL', 'qwen3.5:9b') -# SearXNG module mocks -searx = ModuleType("searx") -searx_plugins = ModuleType("searx.plugins") -searx_results = ModuleType("searx.result_types") +SEARXNG_URL = os.getenv('SEARXNG_URL', '').rstrip('/') +DEV_PORT = int(os.getenv('DEV_PORT', 5000)) +DEV_HOST = os.getenv('DEV_HOST', '127.0.0.1') -class MockPlugin: +# ── SearXNG module mocks ───────────────────────────────────────────────────── +searx_mod = ModuleType("searx") +searx_plugins_mod = ModuleType("searx.plugins") +searx_results_mod = ModuleType("searx.result_types") +searx_network_mod = ModuleType("searx.network") + +class _MockPlugin: def __init__(self, cfg): self.active = getattr(cfg, 'active', True) -class MockPluginInfo: +class _MockPluginInfo: def __init__(self, **kwargs): self.meta = kwargs -class MockEngineResults: +class _MockEngineResults: def __init__(self): self.types = ModuleType("types") - self.types.Answer = lambda *args, **kwargs: kwargs.get('answer', args[0] if args else "") + self.types.Answer = lambda *a, **kw: kw.get('answer', a[0] if a else "") self._results = [] - def add(self, res): self._results.append(res) -searx_plugins.Plugin = MockPlugin -searx_plugins.PluginInfo = MockPluginInfo -searx_results.EngineResults = MockEngineResults +searx_plugins_mod.Plugin = _MockPlugin +searx_plugins_mod.PluginInfo = _MockPluginInfo +searx_results_mod.EngineResults = _MockEngineResults +searx_mod.settings = {'server': {'secret_key': 'dev-secret-key'}} -searx.settings = {'server': {'secret_key': 'demo-secret'}} -searx.network = ModuleType("searx.network") +# network mock — forwards real HTTP so Ollama calls work +import http.client, ssl as _ssl -sys.modules["searx"] = searx -sys.modules["searx.plugins"] = searx_plugins -sys.modules["searx.result_types"] = searx_results - -# Network module mock -searx_network = ModuleType("searx.network") -def mock_network_call(method, url, **kwargs): - import http.client, ssl, json +def _real_http(method, url, **kwargs): from urllib.parse import urlparse - - parsed = urlparse(url) - port = parsed.port or (443 if parsed.scheme=='https' else 80) - target = f"{parsed.hostname}:{port}" - - if parsed.scheme == 'https': - conn = http.client.HTTPSConnection(target, timeout=30, context=ssl.create_default_context()) + p = urlparse(url) + port = p.port or (443 if p.scheme == 'https' else 80) + path = p.path + (f'?{p.query}' if p.query else '') + if p.scheme == 'https': + conn = http.client.HTTPSConnection(p.hostname, port, timeout=30, + context=_ssl.create_default_context()) else: - conn = http.client.HTTPConnection(target, timeout=30) - + conn = http.client.HTTPConnection(p.hostname, port, timeout=30) headers = kwargs.get('headers', {}) - body = None - if kwargs.get('json'): - body = json.dumps(kwargs['json']) - elif kwargs.get('data'): - body = kwargs['data'] - - path = parsed.path - if parsed.query: - path += f"?{parsed.query}" - + body = json.dumps(kwargs['json']).encode() if kwargs.get('json') else kwargs.get('data') if kwargs.get('params'): - from urllib.parse import urlencode - query_str = urlencode(kwargs['params']) - if '?' in path: - path += f"&{query_str}" - else: - path += f"?{query_str}" - + qs = urllib.parse.urlencode(kwargs['params']) + path += ('&' if '?' in path else '?') + qs conn.request(method, path, body=body, headers=headers) return conn.getresponse() -def mock_stream(method, url, **kwargs): - res = mock_network_call(method, url, **kwargs) - - class MockResponse: - def __init__(self, r): - self.status_code = r.status - self.text = "Mock Response" # Stub - self._r = r - - def generator(): +class _MockHTTPResponse: + def __init__(self, r): + self.status_code = r.status + self._content = r.read() + self.text = self._content.decode('utf-8', errors='replace') + def json(self): + return json.loads(self.text) + +def _mock_get(url, **kw): + return _MockHTTPResponse(_real_http('GET', url, **kw)) + +def _mock_stream(method, url, **kw): + r = _real_http(method, url, **kw) + class _Resp: + status_code = r.status + text = "streaming" + def _gen(): while True: - chunk = res.read(128) + chunk = r.read(256) if not chunk: break yield chunk + return _Resp(), _gen() - return MockResponse(res), generator() +searx_network_mod.stream = _mock_stream +searx_network_mod.get = _mock_get -def mock_get(url, **kwargs): - import json - res = mock_network_call('GET', url, **kwargs) - - class MockResponse: - def __init__(self, r): - self.status_code = r.status - self._content = r.read() - self.text = self._content.decode('utf-8') - - def json(self): - return json.loads(self.text) - - return MockResponse(res) +sys.modules['searx'] = searx_mod +sys.modules['searx.plugins'] = searx_plugins_mod +sys.modules['searx.result_types'] = searx_results_mod +sys.modules['searx.network'] = searx_network_mod -searx_network.stream = mock_stream -searx_network.get = mock_get -sys.modules["searx.network"] = searx_network +# ── log capture handler for intent / score extraction ──────────────────────── +class _ScoreCaptureHandler(logging.Handler): + """Captures plugin log messages so dev routes can extract intent/scores.""" + def __init__(self): + super().__init__() + self.records = [] -from ollama_answers import SXNGPlugin + def emit(self, record): + self.records.append(self.format(record)) + + def clear(self): + self.records.clear() + +_score_handler = _ScoreCaptureHandler() +_score_handler.setFormatter(logging.Formatter('%(message)s')) +logging.getLogger('ollama_answers').addHandler(_score_handler) + +# ── load plugin ─────────────────────────────────────────────────────────────── +from flask import Flask, request, redirect, jsonify, Response from flask_babel import Babel +from ollama_answers import SXNGPlugin -app = Flask(__name__) +app = Flask(__name__) babel = Babel(app) -class MockConfig: +class _MockCfg: active = True -plugin = SXNGPlugin(MockConfig()) +plugin = SXNGPlugin(_MockCfg()) plugin.init(app) +# Override api_key from env if set +env_key = os.getenv('LLM_API_KEY', '').strip() +if env_key: + plugin.api_key = env_key -@app.route("/config", methods=["POST"]) -def update_config(): - url = request.form.get("url", "").strip() - bearer = request.form.get("bearer", "").strip() - model = request.form.get("model", "").strip() - query = request.form.get("q", "") - if url: - plugin.endpoint_url = url - plugin.api_key = bearer if bearer else "ollama" - if model: - plugin.model = model - redirect_q = f"?q={query}" if query else "" - return redirect(f"/{redirect_q}") +# Verify Valkey is reachable +try: + import valkey as _vk + _vk_host = os.getenv('VALKEY_HOST', 'localhost') + _vk_port = int(os.getenv('VALKEY_PORT', 6379)) + _test_vk = _vk.Valkey(host=_vk_host, port=_vk_port, + socket_connect_timeout=2) + _test_vk.ping() + print(f" Valkey: {_vk_host}:{_vk_port} ✓") +except Exception as e: + print(f""" + ⚠ Valkey not reachable at {_vk_host}:{_vk_port} + Streaming will not work without Valkey. + Start one with Docker: + docker run -d --name dev-valkey -p 6379:6379 valkey/valkey:9-alpine -@app.route("/search") -def mock_search(): - query = request.args.get("q", "") - format_type = request.args.get("format", "html") - - if format_type != "json": - return "Demo only supports JSON format", 400 - - results = [ - {"title": f"Result 1 for: {query}", "content": f"This is simulated content about {query}. It contains relevant information.", "url": f"https://example.com/1/{query.replace(' ', '-')}", "publishedDate": "2026-01-18"}, - {"title": f"Result 2 for: {query}", "content": f"Additional information regarding {query}. More context and details.", "url": f"https://example.com/2/{query.replace(' ', '-')}", "publishedDate": "2026-01-17"}, - {"title": f"Result 3 for: {query}", "content": f"Further reading on {query}. Expert analysis.", "url": f"https://example.com/3/{query.replace(' ', '-')}", "publishedDate": "2026-01-16"}, - ] - - return { - "results": results, - "infoboxes": [], - "answers": [], - "suggestions": [f"{query} explained", f"{query} tutorial"] + Or set VALKEY_HOST and VALKEY_PORT in your .env file. + Error: {e} +""") + +# ── request/response debug log storage ─────────────────────────────────────── +_DEV_SESSION = secrets.token_hex(12) + +_dev_log = [] # list of {ts, method, path, status, ms, req_body, resp_body} +MAX_LOG = 100 + +@app.before_request +def _log_req(): + request._dev_start = time.time() + request._dev_body = request.get_data(as_text=True) + +@app.after_request +def _log_resp(resp): + if request.path.startswith('/static'): + return resp + ms = round((time.time() - getattr(request, '_dev_start', time.time())) * 1000, 1) + entry = { + 'ts': time.strftime('%H:%M:%S'), + 'method': request.method, + 'path': request.full_path.rstrip('?'), + 'status': resp.status_code, + 'ms': ms, + 'req': getattr(request, '_dev_body', ''), + 'resp': resp.get_data(as_text=True)[:2000], } + _dev_log.insert(0, entry) + if len(_dev_log) > MAX_LOG: + _dev_log.pop() + return resp -@app.route("/") -def index(): - query = request.args.get("q", "why is the sky blue") - - class MockSearchQuery: - pageno = 1 - lang = 'en' - categories = ['general'] - MockSearchQuery.query = query - - class MockSearch: - search_query = MockSearchQuery() - class MockResultContainer: - def __init__(self): - self.answers = set() +# ── mock search results ─────────────────────────────────────────────────────── +MOCK_RESULTS = { + 'default': [ + {"title": "Wikipedia — General Knowledge", + "content": "A comprehensive reference covering the query topic in depth with cited sources.", + "url": "https://en.wikipedia.org/wiki/Main_Page", "publishedDate": "2026-01-15"}, + {"title": "NASA Science — Research Article", + "content": "Peer-reviewed research providing scientific context and empirical data on the topic.", + "url": "https://science.nasa.gov/", "publishedDate": "2026-01-10"}, + {"title": "Scientific American — Analysis", + "content": "In-depth analysis covering recent discoveries and expert commentary.", + "url": "https://scientificamerican.com/", "publishedDate": "2026-01-01"}, + {"title": "BBC Science — Overview", + "content": "Accessible explainer covering the fundamentals and current understanding.", + "url": "https://bbc.com/science", "publishedDate": "2025-12-20"}, + {"title": "MIT OpenCourseWare — Educational Resource", + "content": "Course materials and lecture notes providing structured educational content.", + "url": "https://ocw.mit.edu/", "publishedDate": "2025-12-15"}, + ] + [ + {"title": f"Source {i} — Additional Context", + "content": f"Supplementary information and supporting details from source {i}.", + "url": f"https://example.com/source-{i}", "publishedDate": "2025-12-01"} + for i in range(6, 16) + ] +} +def _get_results(query: str): + """Return real results from SearXNG if configured, else mock results.""" + if SEARXNG_URL: + try: + qs = urllib.parse.urlencode({'q': query, 'format': 'json', 'categories': 'general'}) + url = f"{SEARXNG_URL}/search?{qs}" + with urllib.request.urlopen(url, timeout=8) as r: + data = json.loads(r.read()) + results = data.get('results', []) + logger.info(f"Live SearXNG: {len(results)} results for '{query}'") + return results, data.get('infoboxes', []), data.get('answers', []) + except Exception as e: + logger.warning(f"Live SearXNG failed ({e}), falling back to mock results") + return MOCK_RESULTS['default'], [], [] + +# ── search route used by the demo UI ───────────────────────────────────────── +_last_render = {'html': '', 'query': '', 'scores': [], 'intent': '—', 'summary': ''} + +@app.route('/dev/search') +def dev_search(): + query = request.args.get('q', 'why is the sky blue') + + global _last_render + _last_render['intent'] = '…' + _last_render['query'] = query + + class _MockSearch: + class _MockSearchQuery: + pageno = 1 + lang = 'en-US' + categories = ['general'] + search_query = _MockSearchQuery() + search_query.query = query + + class _MockResultContainer: + def __init__(self, q): + self.answers = set() + self._q = q + self.infoboxes = [] def get_ordered_results(self): - base_results = [ - {"title": "Wikipedia", "content": "The sky appears blue due to Rayleigh scattering of sunlight. When sunlight enters the atmosphere, it collides with gas molecules and scatters in all directions. Blue light scatters more than other colors because it travels in shorter waves.", "url": "https://en.wikipedia.org/wiki/Rayleigh_scattering", "publishedDate": "2026-01-15"}, - {"title": "NASA Science", "content": "Shorter blue wavelengths scatter more than longer red wavelengths. This phenomenon, discovered by Lord Rayleigh in the 1870s, explains why we see a blue sky during the day.", "url": "https://science.nasa.gov/blue-sky", "publishedDate": "2026-01-10"}, - {"title": "Physics Today", "content": "The atmosphere acts as a filter, scattering blue light in all directions while letting other colors pass through more directly.", "url": "https://physicstoday.org/atmosphere", "publishedDate": "2026-01-01"}, - {"title": "Scientific American", "content": "At sunset, light travels through more atmosphere, scattering away the blue and leaving reds and oranges.", "url": "https://scientificamerican.com/sunset", "publishedDate": "2025-12-20"}, - {"title": "National Geographic", "content": "Ocean color also results from light scattering and absorption by water molecules.", "url": "https://nationalgeographic.com/ocean-blue", "publishedDate": "2025-12-15"}, - ] - broad_results = [ - {"title": "MIT OpenCourseWare: Atmospheric Physics", "content": "Course materials.", "url": "https://ocw.mit.edu/physics"}, - {"title": "NOAA: Understanding the Atmosphere", "content": "Educational resource.", "url": "https://noaa.gov/atmosphere"}, - {"title": "BBC Science: Why is the sky blue?", "content": "Explainer article.", "url": "https://bbc.com/science/sky"}, - {"title": "Khan Academy: Light and Color", "content": "Video lesson.", "url": "https://khanacademy.org/light"}, - {"title": "HowStuffWorks: Rayleigh Scattering", "content": "Detailed explanation.", "url": "https://howstuffworks.com/rayleigh"}, - {"title": "Physics Stack Exchange: Sky color discussion", "content": "Q&A thread.", "url": "https://physics.stackexchange.com/sky"}, - {"title": "Quora: Atmospheric optics explained", "content": "Community answers.", "url": "https://quora.com/atmosphere"}, - ] - if 'quantum' in query.lower(): - return [ - {"title": "IBM Quantum", "content": "Quantum computers rely on qubits, which can represent 0, 1, or both via superposition. They solve complex problems faster.", "url": "https://www.ibm.com/quantum", "publishedDate": "2026-01-15"}, - {"title": "Nature Physics", "content": "Entanglement allows qubits to be correlated instantly across distances. This is key for quantum cryptography and teleportation.", "url": "https://nature.com/articles/quantum", "publishedDate": "2026-01-10"}, - {"title": "Wikipedia", "content": "Quantum computing uses quantum mechanics. Major applications include drug discovery and materials science.", "url": "https://en.wikipedia.org/wiki/Quantum_computing", "publishedDate": "2025-12-01"} - ] + broad_results - return base_results + broad_results - result_container = MockResultContainer() + results, iboxes, _ = _get_results(self._q) + self.infoboxes = iboxes + return results - search = MockSearch() - plugin.post_search(None, search) - - injection_html = "" + result_container = _MockResultContainer(query) + + session_id = secrets.token_hex(12) + + # Clear prior conversation so each search starts fresh + try: + from ollama_answers import _get_valkey + vk = _get_valkey() + vk.delete(f"ai:conv:{session_id}") + except Exception: + pass + + class _MockRequest: + cookies = {'sxng_ai_session': session_id} + headers = {} + args = {} + script_root = '' + + class _MockForm(dict): + def get(self, key, default=None): + return default # always return html format + + form = _MockForm() + + _score_handler.clear() + search = _MockSearch() + try: + plugin.post_search(_MockRequest(), search) + except Exception as e: + logger.error(f"post_search failed: {e}", exc_info=True) + return f"

post_search error: {e}

" + + html_payload = "" if search.result_container.answers: - injection_html = list(search.result_container.answers)[0] - - bearer_display = plugin.api_key if plugin.api_key != "ollama" else "" - return f""" - - - - - AI Answers Demo - - - -
- ⚙ Ollama Configuration -
- -
- - -
-
- - -
- -
-
+ raw = list(search.result_container.answers)[0] + # Markup objects need str() to get the HTML + html_payload = str(raw) if raw else "" + logger.info(f"Got AI answer, length: {len(html_payload)}") + if html_payload: + import re as _re + text_preview = _re.sub(r'<[^>]+>', '', html_payload) + text_preview = ' '.join(text_preview.split())[:300] + logger.info(f"AI output preview: {text_preview}") + else: + logger.warning("post_search produced no answers") -
- - -
+ # Parse intent, per-URL TF-IDF scores, and rerank summary from captured log records +# Parse intent, per-URL TF-IDF scores, and rerank summary from captured log records + import re as _re + detected_intent = '—' -

Model: {plugin.model}

-
- {injection_html if injection_html else '

No response — check your Ollama endpoint and token above.

'} - - - """ + scores = [] + fallback_domains = set() + summary = '' -if __name__ == "__main__": - print("AI Answers - Demo\n") - print(f" Endpoint: {plugin.endpoint_url}") - print(f" Model: {plugin.model or 'N/A'}") - print(f" Mode: {'interactive' if plugin.interactive else 'simple'}") - print(f"\n http://localhost:5000/?q=why+is+the+sky+blue\n") - app.run(debug=False, port=5000) + for msg in _score_handler.records: + m = _re.match(r'AI Answers: \[(.+?)\] score=([\d.]+) chunks=(\d+)', msg) + + if m: + scores.append({ + 'url': m.group(1), + 'score': float(m.group(2)), + 'chunks': int(m.group(3)), + 'fallback': False + }) + + m = _re.match(r'AI Answers: falling back to snippet for \[\d+\] (.+)', msg) + if m: + fallback_domains.add(m.group(1)) + + m = _re.match(r'AI Answers: reranked (\d+) results, top score=([\d.]+)', msg) + if m: + summary = f"{m.group(1)} reranked · top {float(m.group(2)):.4f}" + + for s in scores: + domain = s['url'].replace('https://', '').replace('http://', '').split('/')[0] + + if domain in fallback_domains: + s['fallback'] = True + + scores.sort(key=lambda s: s['score'], reverse=True) + + # Refresh latest intent from Valkey + try: + from ollama_answers import _get_valkey + vk = _get_valkey() + latest_intent = vk.get("ai:last_intent") + + if latest_intent: + detected_intent = latest_intent + except Exception: + pass + + _last_render['html'] = html_payload + _last_render['query'] = query + _last_render['intent'] = detected_intent + _last_render['scores'] = scores + _last_render['summary'] = summary + + global _last_job + + _last_job['intent'] = detected_intent + _last_job['query'] = query + + logger.warning(f"RETURNING INTENT TO UI: {repr(_last_job.get('intent'))}") + + return jsonify({'ok': True, 'query': query}) + +@app.route('/dev/render') +def dev_render(): + html = _last_render.get('html', '') + return f""" + + + + + +
{html}
+
+ +""" + +# ── dev API log endpoint ────────────────────────────────────────────────────── +_last_job = {'job_id': None, 'intent': None, 'model': None, + 'query': None, 'start_ts': None} + +def _get_last_job_from_log(): + for entry in _dev_log: + if '/ai-stream' in entry.get('path', '') and entry.get('method') == 'POST': + try: + resp = json.loads(entry.get('resp', '{}')) + if resp.get('job_id'): + return resp['job_id'] + except Exception: + pass + return None + +@app.route('/dev/log') +def dev_log(): + return jsonify(_dev_log) + +@app.route('/dev/log/clear', methods=['POST']) +def dev_log_clear(): + _dev_log.clear() + return jsonify({'ok': True}) + +@app.route('/dev/last-job') +def dev_last_job(): + return jsonify(_last_job) + +@app.route('/dev/last-scores') +def dev_last_scores(): + return jsonify({ + 'scores': _last_render.get('scores', []), + 'intent': _last_render.get('intent', '—'), + 'query': _last_render.get('query', ''), + 'summary': _last_render.get('summary', ''), + }) + +@app.route('/dev/stream-watch') +def dev_stream_watch(): + job_id = request.args.get('job_id', '') + if not job_id: + return jsonify({'error': 'no job_id'}), 400 + + def generate(): + import valkey as _vk + host = os.getenv('VALKEY_HOST', 'localhost') + port = int(os.getenv('VALKEY_PORT', 6379)) + v = _vk.Valkey(host=host, port=port, + socket_connect_timeout=2, + decode_responses=True) + offset = 0 + max_polls = 600 + polls = 0 + while polls < max_polls: + polls += 1 + time.sleep(0.1) + try: + chunks = v.lrange(f"ai:job:{job_id}:chunks", offset, -1) + for chunk in chunks: + if chunk == '__DONE__': + yield "data: __DONE__\n\n" + return + if chunk.startswith('__ERROR__'): + yield f"data: __ERROR__{chunk[9:]}\n\n" + return + yield f"data: {json.dumps(chunk)}\n\n" + offset += len(chunks) + except Exception as e: + yield f"data: __ERROR__{e}\n\n" + return + yield "data: __DONE__\n\n" + + return Response(generate(), + mimetype='text/event-stream', + headers={ + 'Cache-Control': 'no-cache', + 'X-Accel-Buffering': 'no' + }) + +# ── dev config update ───────────────────────────────────────────────────────── +@app.route('/dev/config', methods=['POST']) +def dev_config(): + global SEARXNG_URL + data = request.json or {} + if 'llm_url' in data: + url = data.get('llm_url', '').strip() + if url: + plugin.endpoint_url = url + os.environ['LLM_URL'] = url + if 'model' in data and data['model']: + plugin.model = data['model'].strip() + os.environ['LLM_MODEL'] = plugin.model + if 'max_tokens' in data: + try: + plugin.max_tokens = max(1, int(data['max_tokens'])) + except (ValueError, TypeError): + pass + if 'temperature' in data: + try: + plugin.temperature = float(data['temperature']) + except (ValueError, TypeError): + pass + if 'api_key' in data: + plugin.api_key = data['api_key'].strip() or 'ollama' + if 'searxng_url' in data: + SEARXNG_URL = data['searxng_url'].strip().rstrip('/') + if 'valkey_host' in data or 'valkey_port' in data: + host = data.get('valkey_host', os.getenv('VALKEY_HOST', 'localhost')).strip() + port = int(data.get('valkey_port', os.getenv('VALKEY_PORT', 6379))) + os.environ['VALKEY_HOST'] = host + os.environ['VALKEY_PORT'] = str(port) + # Reset the valkey connection pool so it picks up new settings + import ollama_answers as _oa + _oa._VALKEY_POOL = None + return jsonify({ + 'llm_url': plugin.endpoint_url, + 'model': plugin.model, + 'max_tokens': plugin.max_tokens, + 'temperature': plugin.temperature, + 'api_key': plugin.api_key, + 'searxng_url': SEARXNG_URL or None, + }) + +@app.route('/dev/config', methods=['GET']) +def dev_config_get(): + return jsonify({ + 'llm_url': plugin.endpoint_url, + 'model': plugin.model, + 'max_tokens': plugin.max_tokens, + 'temperature': plugin.temperature, + 'api_key': plugin.api_key, + 'searxng_url': SEARXNG_URL or None, + 'interactive': plugin.interactive, + 'allowed_tabs': list(plugin.allowed_tabs), + 'valkey_host': os.getenv('VALKEY_HOST', 'localhost'), + 'valkey_port': int(os.getenv('VALKEY_PORT', 6379)), + }) + +# ── main dev UI ─────────────────────────────────────────────────────────────── +@app.route('/') +def index(): + return DEV_HTML + +DEV_HTML = """ + + + + +AI Answers — Dev Server + + + + +
+

✦ AI Answers Dev

+ Mock Results + Ollama: checking… + Valkey: checking… + +
+ +
+
+
+ + +
+
+
+

Enter a query above and click Search to test the plugin.

+
+ +
+
+ +
+
+
Config
+
Debug Log
+
Stream
+
API Explorer
+
+ + +
+
Plugin Config
+
loading…
+
Model—
+
Max Tokens—
+
Temperature—
+
Interactive—
+
Allowed Tabs—
+
SearXNG Backendmock
+
Valkey—
+ +
Edit Config
+
+
+ + +
+
+
+
+
+
+
+
+ + +
+
+ + +
+ +
+ + +
+
+ + + +
+
+
No requests yet
+
+
+ + +
+
+ Idle + + +
+
+ Shows token chunks arriving from Valkey in real time during streaming. +
+ +
+ Waiting for next search… +
+
+ First token: — + | + Total: — +
+
+ + +
+

Test plugin routes directly. Token is auto-injected.

+ +
+
+ GET + /ai-models + List Ollama models +
+
+ + 

+        

+      

+
+      

+        

+          POST
+          /ai-stream
+          Start AI stream
+        

+        

+          
+          
+          

+        

+      

+
+      

+        

+          GET
+          /ai-status/{job_id}
+          Poll stream status
+        

+        

+          

+          

+          
+          

+        

+      

+
+      

+        

+          GET
+          /ai-conversation
+          Get conversation
+        

+        

+          

+          
+          

+        

+      

+
+      

+        

+          DELETE
+          /ai-conversation
+          Clear conversation
+        

+        

+          

+          
+          

+        

+      

+
+    

+  

+

+
+
+
+
+"""
+
+# ── token endpoint (dev only) ─────────────────────────────────────────────────
+@app.route('/dev/token')
+def dev_token():
+    ts = str(time.time())
+    sig = hashlib.sha256(f"{ts}{plugin.secret}".encode()).hexdigest()
+    return jsonify({'token': f"{ts}.{sig}"})
+
+# ── ollama health check endpoint ──────────────────────────────────────────────
+@app.route('/dev/valkey-check')
+def dev_valkey_check():
+    try:
+        import valkey as _vk
+        host = os.getenv('VALKEY_HOST', 'localhost')
+        port = int(os.getenv('VALKEY_PORT', 6379))
+        v = _vk.Valkey(host=host, port=port, socket_connect_timeout=2)
+        v.ping()
+        return jsonify({'ok': True, 'host': host, 'port': port})
+    except Exception as e:
+        return jsonify({'ok': False, 'error': str(e)})
+
+@app.route('/dev/ollama-check')
+def dev_ollama_check():
+    try:
+        p = urllib.parse.urlparse(plugin.endpoint_url)
+        base = f"{p.scheme}://{p.netloc}"
+        url = f"{base}/api/tags"
+        req = urllib.request.Request(url, headers={'Authorization': f'Bearer {plugin.api_key}'})
+        with urllib.request.urlopen(req, timeout=3) as r:
+            data = json.loads(r.read())
+        models = len(data.get('models', []))
+        return jsonify({'ok': True, 'models': models})
+    except Exception as e:
+        return jsonify({'ok': False, 'error': str(e)})
+
+# ── entry point ───────────────────────────────────────────────────────────────
+if __name__ == '__main__':
+    print(f"""
+  ✦ AI Answers Dev Server
+  ─────────────────────────────────────────
+  UI:          http://{DEV_HOST}:{DEV_PORT}
+  LLM:         {plugin.endpoint_url}
+  Model:       {plugin.model}
+  Max tokens:  {plugin.max_tokens}
+  Temperature: {plugin.temperature}
+  SearXNG:     {SEARXNG_URL or 'mock results'}
+  ─────────────────────────────────────────
+""")
+    app.run(host=DEV_HOST, port=DEV_PORT, debug=False)
\ No newline at end of file