diff --git a/.github/workflows/rag-benchmark.yml b/.github/workflows/rag-benchmark.yml new file mode 100644 index 00000000..bf2076bd --- /dev/null +++ b/.github/workflows/rag-benchmark.yml @@ -0,0 +1,326 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +name: RAG Benchmark + +on: + workflow_dispatch: + schedule: + # cron day-of-month steps restart every month, so this runs on the 1st, 15th and 29th at 04:00 UTC + - cron: '0 4 */14 * *' + +permissions: + contents: read + +concurrency: + group: rag-benchmark-${{ github.head_ref || github.run_id }} + cancel-in-progress: true + + +jobs: + changes: + runs-on: ubuntu-latest + permissions: + contents: read + pull-requests: read + + outputs: + src: ${{ steps.changes.outputs.src }} + + steps: + - uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36 # v3.0.2 + id: changes + continue-on-error: true + with: + filters: | + src: + - 'main.py' + - 'main_em.py' + - 'config.cpu.yaml' + - 'config.gpu.yaml' + - 'context_chat_backend/**' + - 'appinfo/**' + - 'example.env' + - 'requirements.txt' + - 'benchmark/**' + - '.github/workflows/rag-benchmark.yml' + - 'Dockerfile' + - 'dockerfile_scripts/**' + + benchmark: + runs-on: ubuntu-24.04 + + needs: changes + if: needs.changes.outputs.src != 'false' + + # hard cap so a stuck run fails fast instead of burning the 6h default + timeout-minutes: 180 + + strategy: + fail-fast: false + matrix: + benchmark: [ 'multihop-rag', 'frames' ] + + name: RAG Benchmark - ${{ matrix.benchmark }} + + env: + PGSQL_PORT: 4445 + CCB_DB_URL: postgresql+psycopg://root:rootpassword@localhost:4445/nextcloud + NC_SERVER_VERSION: stable34 + NC_PHP_VERSION: '8.2' + # queries per benchmark leg, stratified by question type; 0 = all of them + # (2255 for multihop-rag after null queries are dropped, 824 for frames) + BENCHMARK_QUERY_LIMIT: 0 + # wall-clock cap for the query loop, leaving room for setup and indexing + # within the 180 min job timeout; at ~2.3s/query the full multihop-rag set + # needs ~86 min + BENCHMARK_BUDGET_SECONDS: 7200 + # indexing attempts, 10s apart (45 min); FRAMES indexes ~2.5k documents + BENCHMARK_INDEX_ATTEMPTS: 270 + + services: + postgres: + image: pgvector/pgvector:pg17 + ports: + - 4445:5432/tcp + env: + POSTGRES_USER: root + POSTGRES_PASSWORD: rootpassword + POSTGRES_DB: nextcloud + options: --health-cmd pg_isready --health-interval 5s --health-timeout 2s --health-retries 5 --name postgres --hostname postgres + + steps: + - name: Checkout server + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + with: + repository: nextcloud/server + ref: ${{ env.NC_SERVER_VERSION }} + submodules: 'recursive' + persist-credentials: false + + - name: Set up php ${{ env.NC_PHP_VERSION }} + uses: shivammathur/setup-php@9e72090525849c5e82e596468b86eb55e9cc5401 # v2 + with: + php-version: ${{ env.NC_PHP_VERSION }} + tools: phpunit + extensions: mbstring, iconv, fileinfo, intl, sqlite, pdo_mysql, pdo_sqlite, pgsql, pdo_pgsql, gd, zip + + - name: Checkout context_chat php app + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + with: + repository: nextcloud/context_chat + path: apps/context_chat + persist-credentials: false + + - name: Checkout backend + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + with: + path: context_chat_backend/ + persist-credentials: false + + - name: Checkout app_api + uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 + with: + repository: nextcloud/app_api + ref: ${{ env.NC_SERVER_VERSION }} + path: apps/app_api + persist-credentials: false + + - name: Get app version + id: appinfo + uses: skjnldsv/xpath-action@7e6a7c379d0e9abc8acaef43df403ab4fc4f770c # master + with: + filename: context_chat_backend/appinfo/info.xml + expression: "/info/version/text()" + + - name: Set up Nextcloud + run: | + sleep 25 + mkdir data + ./occ maintenance:install \ + --verbose \ + --database=pgsql \ + --database-name=nextcloud \ + --database-host=127.0.0.1 \ + --database-port=$PGSQL_PORT \ + --database-user=root \ + --database-pass=rootpassword \ + --admin-user admin \ + --admin-pass password + composer run serve & + + - name: Enable context_chat, app_api and testing + run: ./occ app:enable -vvv -f context_chat app_api testing + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3 + + - name: Build the context_chat_backend cpu image + uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v6 + with: + context: context_chat_backend + push: false + platforms: linux/amd64 + # use local tag so image is not pulled from remote + tags: ghcr.io/ccb-benchmark:local + target: runtime-cpu + load: true + # the gha backend keys on scope, not on the tag, and the default scope + # ("buildkit") is shared with the integration tests, whose builds would + # otherwise overwrite this one's cache on every run + cache-from: type=gha,scope=rag-benchmark + cache-to: type=gha,mode=max,scope=rag-benchmark + + # the preparation scripts only use the standard library, so the runner's + # default python is enough and no backend dependencies are needed here + - name: Download and prepare corpus and queries + run: | + python3 context_chat_backend/benchmark/prepare_dataset.py \ + ${{ matrix.benchmark }} \ + --out benchmark_data \ + --limit ${{ env.BENCHMARK_QUERY_LIMIT }} + + - name: Upload benchmark corpus to Nextcloud + run: | + mkdir -p data/admin/files/benchmark + cp -r benchmark_data/docs/. data/admin/files/benchmark/ + ./occ files:scan admin + + - name: Start backend container + run: | + cd context_chat_backend + echo "NEXTCLOUD_URL=http://localhost:8080" >> example.env + cd .. + # create it up front so the bind mount is not root-owned-only and the log + # files stay readable from the runner after the job + mkdir -p context_chat_backend/persistent_storage/logs + chmod -R 777 context_chat_backend/persistent_storage + docker run --net host --name context_chat_backend \ + --env-file context_chat_backend/example.env \ + -v $(pwd)/context_chat_backend/persistent_storage:/app/persistent_storage \ + -e APP_PERSISTENT_STORAGE=/app/persistent_storage \ + -e EXTERNAL_DB="$CCB_DB_URL" \ + -d ghcr.io/ccb-benchmark:local + + - name: Register backend + run: | + timeout 10 ./occ app_api:daemon:register --net host manual_install "Manual Install" manual-install http localhost http://localhost:8080 + timeout 180 ./occ app_api:app:register context_chat_backend manual_install \ + --json-info "{\"appid\":\"context_chat_backend\",\"name\":\"Context Chat Backend\",\"daemon_config_name\":\"manual_install\",\"version\":\"${{ fromJson(steps.appinfo.outputs.result).version }}\",\"secret\":\"12345\",\"port\":10034,\"scopes\":[],\"system_app\":0}" \ + --force-scopes --wait-finish + + - name: Run cron jobs to populate indexing queue + run: | + while true; do php cron.php; sleep 10; done & + sleep 30 + + - name: Wait for indexing to complete + run: | + # eligible_files_count is cached in appconfig for an hour, so a value read + # before the corpus upload would let the gate pass on a partial index + ./occ config:app:delete context_chat eligible_files_count || true + ./occ config:app:delete context_chat eligible_files_count_time || true + + success=0 + for i in $(seq 1 ${{ env.BENCHMARK_INDEX_ATTEMPTS }}); do + stats=$(./occ context_chat:stats --json 2>/dev/null || echo '{}') + # jq -e fails on null/false, so malformed stats are reported instead of + # silently becoming empty strings that bash then treats as 0 + if ! total=$(echo "$stats" | jq -e '.eligible_files_count // 0') \ + || ! indexed=$(echo "$stats" | jq -e '.vectordb_document_counts.files__default // 0') \ + || ! queued=$(echo "$stats" | jq -e '[(.queued_documents_counts // {}) | to_entries[].value] | add // 0'); then + echo "Attempt $i: could not read stats, retrying. Raw output: $stats" + sleep 10 + continue + fi + echo "Attempt $i: eligible=$total indexed=$indexed queued=$queued" + + diff=$((total - indexed)) + threshold=$((total * 3 / 100)) + if [ "$total" -gt 0 ] && [ "$queued" -eq 0 ] && [ "$diff" -le "$threshold" ]; then + echo "Indexing complete (diff=$diff, threshold=$threshold)" + success=1 + break + fi + + # Check if backend is still alive; inspect prints exactly "true"/"false" + # and nothing at all when the container is gone + ccb_alive=$(docker inspect -f '{{.State.Running}}' context_chat_backend 2>/dev/null || true) + if [ "$ccb_alive" != "true" ]; then + echo "Error: Context Chat Backend container is not running. Exiting." + docker logs --tail 50 context_chat_backend 2>&1 || true + exit 1 + fi + + sleep 10 + done + + if [ $success -ne 1 ]; then + echo "Indexing did not complete within the allotted time" + exit 1 + fi + + # Retrieval-only: context_chat:search runs the vector search without invoking a + # text2text provider, which is what the retrieval metrics actually measure. + # maybe add LLM-as-judge answer correctness eval using llm2 + a very small model + # prompt it with the query + retrieved context + ground-truth answer, score pass/fail + - name: Run benchmark queries and compute retrieval metrics + run: | + ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' > worker_logs 2>&1 & + + python3 context_chat_backend/benchmark/run_benchmark.py \ + --queries benchmark_data/queries.json \ + --results benchmark_results.json \ + --occ ./occ \ + --user admin \ + --budget-seconds ${{ env.BENCHMARK_BUDGET_SECONDS }} + + - name: Show benchmark results + if: always() + run: | + echo "=== ${{ matrix.benchmark }} results ===" + cat benchmark_results.json 2>/dev/null || echo "No results file" + + - name: Upload benchmark results + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + if: always() + with: + name: benchmark-results-${{ matrix.benchmark }} + path: benchmark_results.json + if-no-files-found: warn + + - name: Show container logs + if: always() + run: | + # supervisord routes both the main app and the embedding server to the + # container's stdout/stderr, so this covers each of them + docker logs context_chat_backend || echo "No container logs" + + - name: Show main app JSON logs + if: always() + run: | + sudo tail -v -n +1 context_chat_backend/persistent_storage/logs/ccb.log* || echo "No logs in logs directory" + + - name: Show embedding server JSON logs + if: always() + run: | + sudo tail -v -n +1 context_chat_backend/persistent_storage/logs/em_server.log* || echo "No logs in logs directory" + + - name: Show task processing worker logs + if: always() + run: | + cat worker_logs || echo "No worker logs" + + summary: + permissions: + contents: none + runs-on: ubuntu-latest-low + needs: [changes, benchmark] + + if: always() + + name: rag-benchmark + + steps: + - name: Summary status + run: if ${{ needs.changes.outputs.src != 'false' && needs.benchmark.result != 'success' }}; then exit 1; fi diff --git a/.nextcloudignore b/.nextcloudignore index f4ef4093..ce832100 100644 --- a/.nextcloudignore +++ b/.nextcloudignore @@ -25,6 +25,7 @@ /README.md /readme.markdown /screenshots +/benchmark /examples /docs /src diff --git a/benchmark/prepare_dataset.py b/benchmark/prepare_dataset.py new file mode 100755 index 00000000..2910cb52 --- /dev/null +++ b/benchmark/prepare_dataset.py @@ -0,0 +1,405 @@ +#!/usr/bin/env python3 +# +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# +"""Build a retrieval corpus and query set for the Context Chat RAG benchmark. + +Writes ``/docs/.txt`` for every corpus document and ``/queries.json`` +holding ``[{"query": str, "stratum": str, "evidence_list": [doc_id, ...]}, ...]``. + +Queries without resolvable evidence are dropped: they can never contribute a hit and +would silently deflate recall. +""" + +from __future__ import annotations + +import argparse +import ast +import csv +import hashlib +import json +import random +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from collections import defaultdict +from pathlib import Path + +# The datasets library is deliberately not used: it pulls in pyarrow, whose wheels +# declare no numpy bound but require numpy >= 2 at runtime, which conflicts with the +# numpy 1.x resolved for the backend's own requirements. Both datasets are published +# as plain JSON/TSV, so the stdlib is enough. +HF_BASE = 'https://huggingface.co/datasets' +# pinned to a revision so the query set cannot drift between scheduled runs; bump +# these deliberately when picking up an upstream dataset change +MULTIHOP_REVISION = '71ac0d0bd1f951d2d6b70311f7d2ae404e1ffa82' +FRAMES_REVISION = '58d9fb6330f3ab1316d1eca12e5e8ef23dcc22ef' +MULTIHOP_CORPUS_URL = f'{HF_BASE}/yixuantt/MultiHopRAG/resolve/{MULTIHOP_REVISION}/corpus.json' +MULTIHOP_QUERIES_URL = f'{HF_BASE}/yixuantt/MultiHopRAG/resolve/{MULTIHOP_REVISION}/MultiHopRAG.json' +FRAMES_URL = f'{HF_BASE}/google/frames-benchmark/resolve/{FRAMES_REVISION}/test.tsv' + +# The extracts API only allows the 20-title batch when restricted to lead sections +# (exintro); whole-article extracts force exlimit=1. The bulk of the FRAMES corpus is +# therefore fetched as lead sections, with full articles pulled only for the titles +# the dataset links to a specific section - see the note in build_frames(). +WIKI_API = 'https://en.wikipedia.org/w/api.php' +USER_AGENT = 'nextcloud-context-chat-benchmark/1.0 (https://github.com/nextcloud/context_chat_backend)' +# the extracts API caps whole-article requests at one title per call, so we fetch +# lead sections, which allows the full 20-title batch limit +WIKI_BATCH = 20 +WIKI_RETRIES = 4 +SEED = 42 + + +def http_get(url: str) -> bytes: + """Fetch a URL with retries, following HF's CDN redirects.""" + request = urllib.request.Request(url, headers={'User-Agent': USER_AGENT}) # noqa: S310 - fixed https hosts + last_error: Exception | None = None + for attempt in range(WIKI_RETRIES): + try: + with urllib.request.urlopen(request, timeout=120) as response: # noqa: S310 + return response.read() + except (urllib.error.URLError, TimeoutError) as error: + last_error = error + time.sleep(2 ** attempt) + raise RuntimeError(f'Failed to download {url}') from last_error + + +def doc_id_for(value: str) -> str: + """Stable, filesystem-safe document id.""" + return hashlib.md5(value.encode('utf-8')).hexdigest()[:16] # noqa: S324 - not security relevant + + +def stratified_sample(records: list[dict], limit: int) -> list[dict]: + """Round-robin over strata so every question type keeps its share of the sample.""" + if limit <= 0 or limit >= len(records): + return sorted(records, key=lambda r: r['query']) + + buckets: dict[str, list[dict]] = defaultdict(list) + for record in records: + buckets[record['stratum']].append(record) + + rng = random.Random(SEED) # noqa: S311 - sampling, not cryptography + for bucket in buckets.values(): + bucket.sort(key=lambda r: r['query']) + rng.shuffle(bucket) + + picked: list[dict] = [] + names = sorted(buckets) + while len(picked) < limit: + progressed = False + for name in names: + if buckets[name] and len(picked) < limit: + picked.append(buckets[name].pop()) + progressed = True + if not progressed: + break + + return sorted(picked, key=lambda r: r['query']) + + +def write_corpus(docs_dir: Path, documents: dict[str, str]) -> None: + docs_dir.mkdir(parents=True, exist_ok=True) + for doc_id, content in sorted(documents.items()): + (docs_dir / f'{doc_id}.txt').write_text(content, encoding='utf-8') + + +def finish(out: Path, documents: dict[str, str], records: list[dict], limit: int) -> None: + records = stratified_sample(records, limit) + # only keep documents reachable as evidence plus the full distractor corpus; + # the corpus is always written in full so retrieval stays realistic + write_corpus(out / 'docs', documents) + (out / 'queries.json').write_text(json.dumps(records, indent=2), encoding='utf-8') + + # the Wikipedia extracts behind the FRAMES corpus are fetched live and change as + # articles are edited; recording a digest makes that drift visible when metrics + # move between runs + digest = hashlib.sha256() + for doc_id, content in sorted(documents.items()): + digest.update(doc_id.encode('utf-8')) + digest.update(content.encode('utf-8')) + corpus_hash = digest.hexdigest()[:16] + (out / 'corpus_info.json').write_text( + json.dumps({'corpus_documents': len(documents), 'corpus_sha256': corpus_hash}, indent=2), + encoding='utf-8', + ) + print(f'Corpus digest: {corpus_hash}') + + strata = defaultdict(int) + for record in records: + strata[record['stratum']] += 1 + print(f'Corpus documents: {len(documents)}') + print(f'Queries: {len(records)}') + for name in sorted(strata): + print(f' {name}: {strata[name]}') + + +def build_multihop(out: Path, limit: int) -> None: + print('Downloading MultiHop-RAG corpus and queries') + corpus = json.loads(http_get(MULTIHOP_CORPUS_URL).decode('utf-8')) + queries = json.loads(http_get(MULTIHOP_QUERIES_URL).decode('utf-8')) + + documents: dict[str, str] = {} + by_url: dict[str, str] = {} + by_title: dict[str, str] = {} + + for row in corpus: + url = (row.get('url') or '').strip() + title = (row.get('title') or '').strip() + if not url and not title: + continue + doc_id = doc_id_for(url or title) + documents[doc_id] = f'{title}\n\n{row.get("body") or ""}' + if url: + by_url.setdefault(url, doc_id) + if title: + by_title.setdefault(title, doc_id) + + records: list[dict] = [] + dropped = 0 + for row in queries: + question_type = row.get('question_type') or 'unknown' + # null_query rows carry no evidence by design; they test abstention, not retrieval + if question_type == 'null_query': + dropped += 1 + continue + + evidence = set() + for item in row.get('evidence_list') or []: + doc_id = by_url.get((item.get('url') or '').strip()) or by_title.get((item.get('title') or '').strip()) + if doc_id: + evidence.add(doc_id) + + if not evidence: + dropped += 1 + continue + + records.append({ + 'query': row['query'], + 'stratum': question_type, + 'evidence_list': sorted(evidence), + }) + + print(f'Dropped {dropped} queries without usable evidence') + finish(out, documents, records, limit) + + +def wiki_request(titles: list[str], intro_only: bool = True) -> dict: + params = { + 'action': 'query', + 'format': 'json', + 'formatversion': '2', + 'prop': 'extracts', + 'explaintext': '1', + # whole-article extracts are capped at one title per request, lead sections + # allow the full batch + 'exlimit': str(WIKI_BATCH) if intro_only else '1', + 'redirects': '1', + 'titles': '|'.join(titles), + } + if intro_only: + params['exintro'] = '1' + url = f'{WIKI_API}?{urllib.parse.urlencode(params)}' + request = urllib.request.Request(url, headers={'User-Agent': USER_AGENT}) # noqa: S310 - fixed https host + + last_error: Exception | None = None + for attempt in range(WIKI_RETRIES): + try: + with urllib.request.urlopen(request, timeout=60) as response: # noqa: S310 + return json.loads(response.read().decode('utf-8')).get('query', {}) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as error: + last_error = error + time.sleep(2 ** attempt) + + raise RuntimeError(f'Wikipedia API failed after {WIKI_RETRIES} attempts') from last_error + + +def resolve_alias(title: str, alias: dict[str, str]) -> str: + seen = set() + while title in alias and title not in seen: + seen.add(title) + title = alias[title] + return title + + +def fetch_wiki_pages(titles: list[str]) -> dict[str, tuple[str, str]]: + """Map requested title -> (canonical title, lead section text).""" + resolved: dict[str, tuple[str, str]] = {} + + for start in range(0, len(titles), WIKI_BATCH): + batch = titles[start:start + WIKI_BATCH] + query = wiki_request(batch) + + alias: dict[str, str] = {} + for entry in query.get('normalized', []): + alias[entry['from']] = entry['to'] + for entry in query.get('redirects', []): + alias[entry['from']] = entry['to'] + + pages = {} + for page in query.get('pages', []): + if page.get('missing'): + continue + extract = (page.get('extract') or '').strip() + if extract: + pages[page['title']] = extract + + for title in batch: + canonical = resolve_alias(title, alias) + if canonical in pages: + resolved[title] = (canonical, pages[canonical]) + + print(f' fetched {min(start + WIKI_BATCH, len(titles))}/{len(titles)} pages', flush=True) + time.sleep(0.2) + + return resolved + + +def fetch_wiki_articles(titles: list[str]) -> dict[str, str]: + """Fetch whole-article text, one request per title (the API caps exlimit at 1). + + Used for titles that FRAMES links with a section fragment, whose supporting text + sits below the lead section. + """ + articles: dict[str, str] = {} + + for index, title in enumerate(titles, start=1): + query = wiki_request([title], intro_only=False) + + alias: dict[str, str] = {} + for entry in query.get('normalized', []): + alias[entry['from']] = entry['to'] + for entry in query.get('redirects', []): + alias[entry['from']] = entry['to'] + + canonical = resolve_alias(title, alias) + for page in query.get('pages', []): + if page.get('missing'): + continue + extract = (page.get('extract') or '').strip() + # list-style articles are mostly tables, which explaintext strips, so an + # empty extract here is expected and the lead section is kept instead + if extract and page['title'] == canonical: + articles[title] = extract + + if index % 25 == 0 or index == len(titles): + print(f' fetched {index}/{len(titles)} full articles', flush=True) + time.sleep(0.2) + + return articles + + +def title_from_link(link: str) -> str: + """Strip the section fragment; the API normalises it away anyway.""" + return urllib.parse.unquote(link.rsplit('/', 1)[-1]).split('#')[0].replace('_', ' ').strip() + + +def has_fragment(link: str) -> bool: + return '#' in link.rsplit('/', 1)[-1] + + +def build_frames(out: Path, limit: int) -> None: + """Build the FRAMES corpus from Wikipedia. + + FRAMES links some articles with a section fragment (``#Final league tables``, + ``#Composite box``, ...), meaning the supporting text sits below the lead + section. Those articles are fetched in full; everything else uses the lead + section, which is what lets the bulk of the corpus come down in 20-title + batches. + + Residual limitation: ``explaintext`` drops tables and infoboxes, so purely + tabular articles (``List of Egyptian pharaohs`` and similar) still yield little + or no usable text. Questions whose answer lives only in a table therefore remain + unanswerable from this corpus, and the numbers are still not directly comparable + to published FRAMES results. + """ + print('Downloading FRAMES benchmark') + # csv module default field size cap is too small for the wiki_links column + csv.field_size_limit(10 * 1024 * 1024) + text = http_get(FRAMES_URL).decode('utf-8') + dataset = list(csv.DictReader(text.splitlines(), delimiter='\t')) + + # wiki_links is a *string* holding a Python list literal, not a list + rows = [] + for row in dataset: + try: + links = ast.literal_eval(row['wiki_links']) + except (ValueError, SyntaxError): + continue + if not isinstance(links, list): + continue + # reasoning_types is a " | "-joined combination; the leading type keeps the + # strata coarse enough to stay balanced at small sample sizes + reasoning = (row.get('reasoning_types') or 'unknown').split('|')[0].strip() or 'unknown' + rows.append((row['Prompt'], reasoning, links)) + + titles = sorted({title_from_link(link) for _, _, links in rows for link in links if link}) + # titles the dataset links to a specific section need the whole article + fragment_titles = sorted({ + title_from_link(link) + for _, _, links in rows for link in links + if link and has_fragment(link) + }) + + print(f'Resolving {len(titles)} Wikipedia pages') + # FRAMES ships only links, so the corpus has to be fetched; the full link set is + # indexed (not just the sampled queries') to keep a realistic distractor pool + pages = fetch_wiki_pages(titles) + print(f'Resolved {len(pages)}/{len(titles)} pages') + + print(f'Fetching {len(fragment_titles)} section-linked articles in full') + full_articles = fetch_wiki_articles(fragment_titles) + upgraded = 0 + for title, extract in full_articles.items(): + if title in pages: + canonical, lead = pages[title] + # only take the fuller text; a stripped-tables extract can be shorter + if len(extract) > len(lead): + pages[title] = (canonical, extract) + upgraded += 1 + print(f'Upgraded {upgraded}/{len(fragment_titles)} articles to full text') + + documents: dict[str, str] = {} + title_to_doc: dict[str, str] = {} + for requested, (canonical, extract) in pages.items(): + doc_id = doc_id_for(canonical) + documents[doc_id] = f'{canonical}\n\n{extract}' + title_to_doc[requested] = doc_id + + records: list[dict] = [] + dropped = 0 + for prompt, reasoning, links in rows: + evidence = {title_to_doc[title_from_link(link)] for link in links if title_from_link(link) in title_to_doc} + if not evidence: + dropped += 1 + continue + records.append({ + 'query': prompt, + 'stratum': reasoning, + 'evidence_list': sorted(evidence), + }) + + print(f'Dropped {dropped} queries without usable evidence') + finish(out, documents, records, limit) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('dataset', choices=['multihop-rag', 'frames']) + parser.add_argument('--out', type=Path, default=Path('benchmark_data')) + parser.add_argument('--limit', type=int, default=200, help='max queries to keep (0 = all)') + args = parser.parse_args() + + args.out.mkdir(parents=True, exist_ok=True) + if args.dataset == 'multihop-rag': + build_multihop(args.out, args.limit) + else: + build_frames(args.out, args.limit) + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/benchmark/run_benchmark.py b/benchmark/run_benchmark.py new file mode 100755 index 00000000..c2d4eb13 --- /dev/null +++ b/benchmark/run_benchmark.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +# +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# +"""Run retrieval queries against Context Chat and compute retrieval metrics. + +Uses ``occ context_chat:search``, which runs the vector search alone. The +``context_chat:prompt`` path would additionally invoke a text2text provider per query, +which costs minutes per query and measures the LLM rather than retrieval. + +Metrics follow the MultiHop-RAG paper (arXiv:2401.15391, section 2.3): + +hit_rate@k + Fraction of queries with *at least one* evidence document in the top-k. This is + the number that matters for Context Chat, which passes the whole retrieved set to + the LLM, so partial evidence is frequently enough to answer. +evidence_recall@k + Fraction of a query's evidence documents present in the top-k, averaged over + queries. This is the paper's "Hits@K" and is the value comparable to its Table 5. +map@k + Mean average precision at k; rewards ranking evidence documents highly. +mrr + Reciprocal rank of the first evidence document. + +k=30 matches the ``CONTEXT_LIMIT`` the backend uses when feeding the LLM, so it +reflects the production operating point. The MultiHop-RAG paper reports only +Hits@4 and Hits@10 (Table 5, page 7), so k=4 and k=10 are the cutoffs that can be +held against it; k=1, 5 and 30 are reported for this project's own tracking. + +Comparisons with that table remain indicative rather than exact: + +* The paper scores *chunks*; ``do_doc_search`` dedups chunks to source documents + before returning, so these numbers are document-level and read higher. +* Chunking differs (``embedding_chunk_size`` 2000 characters here versus 256 + tokens there). +* ``do_doc_search`` searches ``ctx_limit * 2`` chunks and then dedups, so with + ``--limit 30`` only the top 60 chunks are considered. A document whose best + chunk ranks 61st or lower cannot appear, which makes evidence_recall@30 a lower + bound on true document-level recall@30. + +Each query requests ``--limit`` documents, defaulting to the largest scored cutoff so +every k is measured against a full result set. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +import time +from pathlib import Path + +DEFAULT_K_VALUES = (1, 4, 5, 10, 30) +# matches " 0 => '...'," entries emitted by var_export(), honouring escaped quotes +VAR_EXPORT_ITEM = re.compile(r"^\s*\d+\s*=>\s*'((?:[^'\\]|\\.)*)',?\s*$", re.MULTILINE) +# presence of the 'sources' key proves occ returned a well-formed result, even when +# the array inside it is empty +VAR_EXPORT_SOURCES = re.compile(r"^\s*'sources'\s*=>", re.MULTILINE) + + +def parse_sources(stdout: str) -> list[str]: + """Extract ordered source labels from occ's var_export() output. + + occ prints ``var_export($task->getOutput(), true)`` - PHP literal syntax, not JSON. + Each ``sources`` entry is a JSON string with id/label/icon/url. ``label`` holds the + file name, and file sources keep retrieval rank order. + """ + labels = [] + for match in VAR_EXPORT_ITEM.finditer(stdout): + raw = match.group(1).replace("\\'", "'").replace('\\\\', '\\') + try: + source = json.loads(raw) + except json.JSONDecodeError: + continue + if isinstance(source, dict) and 'label' in source: + labels.append(source['label']) + return labels + + +def doc_ids_from_labels(labels: list[str]) -> list[str]: + """Map file names back to corpus document ids, preserving rank and dropping repeats.""" + seen = set() + ordered = [] + for label in labels: + doc_id = label.rsplit('.', 1)[0] if '.' in label else label + if doc_id not in seen: + seen.add(doc_id) + ordered.append(doc_id) + return ordered + + +def run_query(occ: str, user: str, query: str, timeout: int, limit: int) -> tuple[list[str], str | None]: + try: + result = subprocess.run( # noqa: S603 - fixed argv, no shell + [occ, 'context_chat:search', user, query, '--limit', str(limit)], + capture_output=True, text=True, timeout=timeout, check=False, + ) + except subprocess.TimeoutExpired: + return [], f'timeout after {timeout}s' + + if result.returncode != 0: + detail = (result.stderr or result.stdout or '').strip().replace('\n', ' ') + return [], f'exit {result.returncode}: {detail[:300]}' + + # an empty "sources" array is a legitimate retrieval result (the query matched + # nothing) and must score as a miss; only output that is not a well-formed + # var_export dump counts as a failure, otherwise misses are silently dropped + # from the denominator and every metric is inflated + if not VAR_EXPORT_SOURCES.search(result.stdout): + return [], f'unparseable output: {result.stdout.strip()[:300]}' + + return doc_ids_from_labels(parse_sources(result.stdout)), None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--queries', type=Path, default=Path('benchmark_data/queries.json')) + parser.add_argument('--results', type=Path, default=Path('benchmark_results.json')) + parser.add_argument('--occ', default='./occ') + parser.add_argument('--user', default='admin') + parser.add_argument('--query-timeout', type=int, default=120) + parser.add_argument('--budget-seconds', type=int, default=3600, help='wall-clock cap for the whole run') + parser.add_argument('--max-error-rate', type=float, default=0.05) + parser.add_argument( + '--k-values', + type=lambda v: tuple(sorted({int(x) for x in v.split(',') if x.strip()})), + default=DEFAULT_K_VALUES, + help='comma-separated cutoffs to score at (default: 1,4,5,10,30)', + ) + parser.add_argument( + '--limit', + type=int, + default=None, + help='documents to request per query (default: the largest --k-values entry)', + ) + args = parser.parse_args() + + if not args.k_values: + print('--k-values must contain at least one cutoff', file=sys.stderr) + return 1 + + queries = json.loads(args.queries.read_text(encoding='utf-8')) + if not queries: + print('No queries to evaluate', file=sys.stderr) + return 1 + + k_values = args.k_values + # retrieve deep enough to score the largest cutoff + retrieval_limit = args.limit if args.limit is not None else max(k_values) + if retrieval_limit < max(k_values): + print( + f'--limit {retrieval_limit} is below the largest k ({max(k_values)}); ' + 'deeper cutoffs will be truncated', + file=sys.stderr, + ) + + hits = dict.fromkeys(k_values, 0) + evidence_recall = dict.fromkeys(k_values, 0.0) + average_precision = dict.fromkeys(k_values, 0.0) + reciprocal_rank = 0.0 + evaluated = 0 + errors: list[str] = [] + started = time.monotonic() + truncated = False + aborted: str | None = None + + for index, item in enumerate(queries, start=1): + elapsed = time.monotonic() - started + if elapsed > args.budget_seconds: + print(f'Time budget of {args.budget_seconds}s exhausted after {index - 1} queries, stopping') + truncated = True + break + + returned, error = run_query(args.occ, args.user, item['query'], args.query_timeout, retrieval_limit) + if error is not None: + errors.append(f'[{index}] {error}') + # a handful of failures is tolerable; a broken contract is not + if len(errors) > max(5, int(len(queries) * args.max_error_rate)): + print('\n'.join(errors[-5:]), file=sys.stderr) + aborted = f'{len(errors)} query failures exceed the allowed error rate' + break + continue + + evaluated += 1 + evidence = set(item['evidence_list']) + + for k in k_values: + top_k = returned[:k] + found = evidence & set(top_k) + + # hit_rate@k: at least one evidence doc retrieved + if found: + hits[k] += 1 + + # evidence_recall@k: share of this query's evidence that was retrieved + evidence_recall[k] += len(found) / len(evidence) + + # average precision@k: precision at each rank holding an evidence doc, + # normalised by the number of evidence docs reachable within k + relevant_seen = 0 + precision_sum = 0.0 + for rank, doc_id in enumerate(top_k, start=1): + if doc_id in evidence: + relevant_seen += 1 + precision_sum += relevant_seen / rank + average_precision[k] += precision_sum / min(len(evidence), k) + + for rank, doc_id in enumerate(returned, start=1): + if doc_id in evidence: + reciprocal_rank += 1.0 / rank + break + + if index % 25 == 0: + rate = (time.monotonic() - started) / index + print(f'Progress: {index}/{len(queries)} ({rate:.1f}s/query)', flush=True) + + # always write what was measured, even on the abort paths: a failed run is + # exactly when the partial numbers and the failure list are most useful + results = { + 'queries_planned': len(queries), + 'queries_evaluated': evaluated, + 'query_failures': len(errors), + 'truncated': truncated, + 'aborted': aborted, + 'retrieval_limit': retrieval_limit, + 'elapsed_seconds': round(time.monotonic() - started, 1), + 'first_failures': errors[:5], + } + if evaluated: + results.update({ + 'mrr': round(reciprocal_rank / evaluated, 4), + **{f'hit_rate_at_{k}': round(hits[k] / evaluated, 4) for k in k_values}, + **{f'evidence_recall_at_{k}': round(evidence_recall[k] / evaluated, 4) for k in k_values}, + **{f'map_at_{k}': round(average_precision[k] / evaluated, 4) for k in k_values}, + }) + args.results.write_text(json.dumps(results, indent=2), encoding='utf-8') + + if evaluated == 0: + print('No query produced a usable result', file=sys.stderr) + print('\n'.join(errors[:5]), file=sys.stderr) + return 1 + + print('\n=== Benchmark results ===') + width = max(len(str(k)) for k in k_values) + print(f'{"k":>{width}} {"hit_rate":>9} {"evid_recall":>11} {"MAP":>7}') + for k in k_values: + print( + f'{k:>{width}} {results[f"hit_rate_at_{k}"]:>9.4f}' + f' {results[f"evidence_recall_at_{k}"]:>11.4f}' + f' {results[f"map_at_{k}"]:>7.4f}' + ) + print(f'MRR: {results["mrr"]:.4f}') + print(f'Evaluated {evaluated}/{len(queries)} queries with {len(errors)} failures') + if errors: + print('First failures:') + for line in errors[:5]: + print(f' {line}') + + if aborted is not None: + print(f'Aborting: {aborted}', file=sys.stderr) + return 1 + + # a run that evaluated almost nothing is not a result worth publishing + if evaluated < len(queries) * 0.5: + print(f'Aborting: only {evaluated}/{len(queries)} queries completed', file=sys.stderr) + return 1 + + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/context_chat_backend/task_fetcher.py b/context_chat_backend/task_fetcher.py index ca2ab52a..2b0a5a5b 100644 --- a/context_chat_backend/task_fetcher.py +++ b/context_chat_backend/task_fetcher.py @@ -59,7 +59,9 @@ # task processing or request processing TP_TRIGGER = Event() TP_CHECK_INTERVAL = 5 -TP_CHECK_INTERVAL_WITH_TRIGGER = 5 * 60 +# todo: trigger does not always fire for some reason, a 5 min poll is slow +# todo: in case it did fire at some point but did not at some other +TP_CHECK_INTERVAL_WITH_TRIGGER = 10 TP_CHECK_INTERVAL_ON_ERROR = 15 CONTEXT_LIMIT = 30