From f04ad0c1a34878e6dc75ef05ad19c17c2e964d63 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Fri, 16 Oct 2026 21:56:41 +0800 Subject: [PATCH] test(rag): verify bounded Hybrid runtime provider smoke --- README.md | 14 ++ .../hybrid_runtime_smoke_2026-10-16.json | 46 ++++ scripts/smoke_hybrid_runtime.py | 216 ++++++++++++++++++ tests/test_hybrid_search.py | 76 ++++++ 4 files changed, 352 insertions(+) create mode 100644 docs/reports/hybrid_runtime_smoke_2026-10-16.json create mode 100644 scripts/smoke_hybrid_runtime.py diff --git a/README.md b/README.md index 30ddc436..c35e6370 100644 --- a/README.md +++ b/README.md @@ -251,6 +251,20 @@ API 默认保持 **Dense Chroma 检索**；`/rag/search` 另提供默认关闭 本阶段不调用答案生成、不改变拒答逻辑、不创建工单；完整问答与审批演示仍是后续独立验收范围。 +小范围接线验收脚本（默认只展示计划，零模型调用且不读取 `.env`）： + +```bash +python -m scripts.smoke_hybrid_runtime +# 仅在明确批准本次数据外发和费用后执行；此参数本身不构成授权： +python -m scripts.smoke_hybrid_runtime --execute --env-file .env +``` + +真实模式只发送脚本内固定的四篇虚构说明及两个查询，使用独立临时 SQLite、Chroma 与上传目录，经过 ASGI 路由及 Demo JWT 认证，不连接已运行的服务、不改现有数据库和开关。检查默认 Dense、Hybrid 简单排序、租户/分类隔离、删除生效、空语料零调用及检索审计；不是 HTTP 网络部署验收或准确率评测。 +固定使用北京 `text-embedding-v4`（最多 8 次）及新加坡 `qwen3-rerank`（最多 3 次），禁用 SDK/传输层重试与重定向。发送前校验端点、模型及文本白名单；单次请求体最多 4 KiB、累计 16 KiB。控制台只记录尝试次数、字节、状态、耗时和服务端 usage，不打印密钥或提供方错误正文。任何失败即停止，不自动重新运行。 +预算按 [官方价格](https://www.alibabacloud.com/help/en/model-studio/model-pricing) 保守预留 ¥1，短文本预计不足 ¥0.1；次数/字节限制不是提供方账单硬限额。中断后的请求可能已计费，重新执行须重新确认，不能把脚本重跑当成免费续跑。 + +2026-10-16 已获授权完成一次真实提供方冒烟：8 次 embedding + 3 次重排全部 HTTP 200，服务端 usage 分别为 126 / 162 tokens，9 项接线检查通过；[逐次结果与边界](docs/reports/hybrid_runtime_smoke_2026-10-16.json)。这是临时 ASGI 环境的小范围验证，不证明 Hybrid 优于 Dense，也不代表已开启现有服务或验收问答、拒答、工单整链。 + --- # 4. RAG 评测与迭代：从 Dense 基线到当前状态 diff --git a/docs/reports/hybrid_runtime_smoke_2026-10-16.json b/docs/reports/hybrid_runtime_smoke_2026-10-16.json new file mode 100644 index 00000000..6245104f --- /dev/null +++ b/docs/reports/hybrid_runtime_smoke_2026-10-16.json @@ -0,0 +1,46 @@ +{ + "date": "2026-10-16", + "status": "PASS", + "runtime_base_commit": "1557d670c8ce4a5aea8f6b436c0ed0502009e3e9", + "script": "scripts/smoke_hybrid_runtime.py", + "script_sha256": "392fee1dd5386e044a34aea238cce63b13f4a05cbbb22a32a57fe1b90638351f", + "execution_count": 1, + "scope": "ASGI TestClient, real provider HTTP, temporary SQLite/Chroma/storage, four fixed synthetic documents", + "models": { + "embedding": {"name": "text-embedding-v4", "region": "Beijing", "attempts": 8, "total_tokens": 126}, + "rerank": {"name": "qwen3-rerank", "region": "Singapore", "attempts": 3, "total_tokens": 162} + }, + "total_request_body_bytes": 1925, + "retries": 0, + "checks": [ + "upload/index", + "JWT tenant scope", + "body tenant ignored", + "default Dense", + "Hybrid VPN ranking", + "category", + "delete", + "empty corpus", + "typed audit" + ], + "provider_attempts": [ + {"kind": "embedding", "bytes": 155, "status": 200, "total_tokens": 21, "latency_ms": 335}, + {"kind": "embedding", "bytes": 163, "status": 200, "total_tokens": 21, "latency_ms": 146}, + {"kind": "embedding", "bytes": 159, "status": 200, "total_tokens": 19, "latency_ms": 136}, + {"kind": "embedding", "bytes": 154, "status": 200, "total_tokens": 21, "latency_ms": 146}, + {"kind": "embedding", "bytes": 99, "status": 200, "total_tokens": 12, "latency_ms": 160}, + {"kind": "embedding", "bytes": 99, "status": 200, "total_tokens": 12, "latency_ms": 151}, + {"kind": "rerank", "bytes": 356, "status": 200, "total_tokens": 84, "latency_ms": 394}, + {"kind": "embedding", "bytes": 98, "status": 200, "total_tokens": 8, "latency_ms": 125}, + {"kind": "rerank", "bytes": 270, "status": 200, "total_tokens": 36, "latency_ms": 357}, + {"kind": "embedding", "bytes": 99, "status": 200, "total_tokens": 12, "latency_ms": 130}, + {"kind": "rerank", "bytes": 273, "status": 200, "total_tokens": 42, "latency_ms": 334} + ], + "limitations": [ + "Synthetic wiring smoke, not independent retrieval quality or refusal/answer acceptance", + "No running-server HTTP, Docker, public deployment or production-load acceptance", + "Latency is per provider HTTP attempt, not end-to-end request latency", + "Provider usage is not a billing receipt; no actual billed amount is asserted", + "Default Hybrid switch, existing data, answer/refusal and ticket paths were not changed" + ] +} diff --git a/scripts/smoke_hybrid_runtime.py b/scripts/smoke_hybrid_runtime.py new file mode 100644 index 00000000..45b5b864 --- /dev/null +++ b/scripts/smoke_hybrid_runtime.py @@ -0,0 +1,216 @@ +"""Isolated ASGI/SQL/Chroma smoke; default is a zero-call plan, not a live server.""" + +import argparse +from contextlib import ExitStack +import json +import logging +import os +from pathlib import Path +import secrets +import sys +import tempfile +import time +from unittest.mock import patch +from urllib.parse import urlsplit +from uuid import uuid4 + +import httpx + + +DOCS = ( + ("vpn.md", "it", "support", "VPN error 809: reconnect the VPN client and check the network firewall settings."), + ("printer.md", "it", "support", "Printer paper jam: switch off the printer and remove the jammed paper from the tray."), + ("leave.md", "hr", "support", "Annual leave: submit a leave request in the HR portal and obtain manager approval."), + ("foreign.md", "it", "support_other", "FOREIGN_ONLY VPN error 809: this synthetic guide belongs to another tenant."), +) +VPN_QUERY = "How do I resolve VPN error 809?" +HR_QUERY = "How do I request annual leave?" +QUERIES = {VPN_QUERY, HR_QUERY} +LIMITS = {"embedding": 8, "rerank": 3} +MAX_BODY_BYTES = 4096 +MAX_TOTAL_BYTES = 16384 + + +class BoundedTransport(httpx.BaseTransport): + """Bound actual wire attempts, including failures; never log credentials or bodies.""" + + def __init__(self, kind, base_url, records, delegate): + self.kind, self.base_url, self.records, self.delegate = kind, base_url, records, delegate + + def handle_request(self, request): + expected = self.base_url.rstrip("/") + ("/embeddings" if self.kind == "embedding" else "/reranks") + body = json.loads(request.content) + if request.method != "POST" or str(request.url) != expected: + raise RuntimeError("Unapproved model endpoint") + if body.get("model") != ("text-embedding-v4" if self.kind == "embedding" else "qwen3-rerank"): + raise RuntimeError("Unapproved model") + if self.kind == "embedding": + allowed = QUERIES | {f"{Path(name).stem}\n\n{text}" for name, _, _, text in DOCS} + if not body.get("input") or not set(body["input"]) <= allowed: + raise RuntimeError("Unapproved embedding text") + else: + allowed = {text for _, _, user, text in DOCS if user == "support"} + if body.get("query") not in QUERIES or not body.get("documents") or not set(body["documents"]) <= allowed: + raise RuntimeError("Unapproved rerank text") + size = len(request.content) + if (sum(r["kind"] == self.kind for r in self.records) >= LIMITS[self.kind] + or size > MAX_BODY_BYTES or sum(r["bytes"] for r in self.records) + size > MAX_TOTAL_BYTES): + raise RuntimeError("Smoke call/size budget exhausted") + row = {"kind": self.kind, "bytes": size, "status": None, "total_tokens": None} + self.records.append(row) # Count an attempted send even if its response is lost. + started = time.perf_counter() + try: + response = self.delegate.handle_request(request) + response.read() + row["status"] = response.status_code + try: + row["total_tokens"] = response.json().get("usage", {}).get("total_tokens") + except (ValueError, AttributeError): + pass + return response + finally: + row["latency_ms"] = round((time.perf_counter() - started) * 1000) + + def close(self): + self.delegate.close() + + +def run_flow(client): + def request(method, path, expected=200, **kwargs): + response = client.request(method, path, **kwargs) + if response.status_code != expected: + raise RuntimeError(f"{method} {path}: HTTP {response.status_code}") + return response.json() + + headers = {} + for username in ("support", "support_other"): + token = request("POST", "/auth/token", json={"username": username, "password": "support"})["access_token"] + headers[username] = {"Authorization": f"Bearer {token}"} + ids = {} + for name, category, user, text in DOCS: + doc = request("POST", "/documents/upload", 201, headers=headers[user], + data={"category": category}, files={"file": (name, text.encode(), "text/markdown")}) + ids[name] = doc["id"] + indexed = request("POST", f"/documents/{doc['id']}/index", headers=headers[user]) + if indexed["status"] != "indexed" or indexed["chunk_count"] != 1: + raise RuntimeError("Synthetic document did not index as one chunk") + own = headers["support"] + if client.get(f"/documents/{ids['foreign.md']}", headers=own).status_code != 404: + raise RuntimeError("Cross-tenant document access allowed") + + def search(query, category, mode=None): + body = {"query": query, "category": category, "tenant_id": "tenant_other"} + if mode: + body["retrieval_mode"] = mode + data = request("POST", "/rag/search", headers=own, json=body) + if data["retrieval_mode"] != (mode or "dense"): + raise RuntimeError("Wrong retrieval mode") + if any(r["tenant_id"] != "tenant_demo" or r["category"] != category for r in data["results"]): + raise RuntimeError("Search scope violation") + return data + + dense = search(VPN_QUERY, "it") + hybrid = search(VPN_QUERY, "it", "hybrid") + if dense["top_k"] != 3 or {r["document_id"] for r in dense["results"]} != {ids['vpn.md'], ids['printer.md']}: + raise RuntimeError("Default Dense regression") + if hybrid["top_k"] != 14 or {r["document_id"] for r in hybrid["results"]} != {ids['vpn.md'], ids['printer.md']}: + raise RuntimeError("Hybrid candidate scope mismatch") + if hybrid["results"][0]["document_id"] != ids["vpn.md"]: + raise RuntimeError("VPN synthetic relevance check failed") + hr = search(HR_QUERY, "hr", "hybrid") + if [r["document_id"] for r in hr["results"]] != [ids["leave.md"]]: + raise RuntimeError("Category filtering failed") + request("DELETE", f"/documents/{ids['vpn.md']}", headers=own) + after_delete = search(VPN_QUERY, "it", "hybrid") + if [r["document_id"] for r in after_delete["results"]] != [ids["printer.md"]]: + raise RuntimeError("Deleted document still searchable") + request("DELETE", f"/documents/{ids['printer.md']}", headers=own) + if search(VPN_QUERY, "it", "hybrid")["results"]: + raise RuntimeError("Empty tenant/category scope was not empty") + logs = request("GET", "/agent-ops/retrieval-logs", headers=own) + hybrid_logs = [r for r in logs if isinstance(json.loads(r["scores_json"]), dict)] + if len(logs) != 5 or len(hybrid_logs) != 4 or any(r["top_distance"] is not None for r in hybrid_logs): + raise RuntimeError("Missing or incorrectly typed retrieval audit") + return {"status": "PASS", "checks": ["upload/index", "JWT tenant scope", "body tenant ignored", + "default Dense", "Hybrid VPN ranking", "category", "delete", "empty corpus", "typed audit"]} + + +def execute(env_file): + from dotenv import dotenv_values + from openai import OpenAI + import chromadb + from fastapi.testclient import TestClient + + if "database" in sys.modules: + raise RuntimeError("Execute in a fresh process to isolate the database") + config = dotenv_values(env_file) + embed_url = config.get("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1") + rerank_url = config.get("DASHSCOPE_RERANK_BASE_URL", "") + rerank_host = urlsplit(rerank_url).hostname or "" + if (embed_url.rstrip("/") != "https://dashscope.aliyuncs.com/compatible-mode/v1" + or not rerank_host.startswith("ws-") or not rerank_host.endswith(".ap-southeast-1.maas.aliyuncs.com") + or urlsplit(rerank_url).scheme != "https" or urlsplit(rerank_url).path.rstrip("/") != "/compatible-api/v1" + or not config.get("DASHSCOPE_API_KEY") or not config.get("DASHSCOPE_RERANK_API_KEY")): + raise RuntimeError("Missing credentials or endpoints outside the approved Beijing/Singapore scope") + records = [] + with tempfile.TemporaryDirectory(prefix="hybrid_smoke_") as temp, ExitStack() as stack: + os.environ.update(DATABASE_URL="sqlite:///" + str(Path(temp) / "smoke.db"), SQL_ECHO="false", + DOCUMENT_STORAGE_ROOT=str(Path(temp) / "documents"), AUTH_DEMO_USERS_JSON="", + JWT_SECRET=secrets.token_hex(32), RAG_HYBRID_SEARCH_ENABLED="true", ANONYMIZED_TELEMETRY="False", + EMBEDDING_MODEL="text-embedding-v4", PYTHON_DOTENV_DISABLED="1", + DASHSCOPE_API_KEY=config["DASHSCOPE_API_KEY"], DASHSCOPE_BASE_URL=embed_url) + import database + import main + import rag_runtime.build_rag_index as embeddings + import rag_runtime.hybrid_search as hybrid + import rag_runtime.query_chroma as dense + import services.document_service as documents + database.create_db_and_tables() + stack.callback(database.engine.dispose) + stack.callback(main.llm_client.close) + for name in ("httpx", "httpx2", "openai", "bm25s"): + logging.getLogger(name).setLevel(logging.WARNING) + chroma = chromadb.EphemeralClient(settings=chromadb.Settings(anonymized_telemetry=False)) + collection = chroma.create_collection("smoke_" + uuid4().hex, embedding_function=None) + stack.callback(chroma.delete_collection, collection.name) + + def model_client(kind, url, key): + transport = BoundedTransport(kind, url, records, httpx.HTTPTransport(retries=0)) + return OpenAI(api_key=key, base_url=url, max_retries=0, timeout=30, + http_client=httpx.Client(transport=transport, trust_env=False, follow_redirects=False)) + + embedding_client = stack.enter_context(model_client("embedding", embed_url, config["DASHSCOPE_API_KEY"])) + stack.enter_context(patch.object(embeddings, "get_client", lambda: embedding_client)) + stack.enter_context(patch.object(hybrid, "get_rerank_client", lambda: model_client( + "rerank", rerank_url, config["DASHSCOPE_RERANK_API_KEY"]))) + for module, attr in ((dense, "get_collection"), (hybrid, "get_collection"), (documents, "get_document_chroma_collection")): + stack.enter_context(patch.object(module, attr, lambda **_: collection)) + try: + with TestClient(main.app) as client: + result = run_flow(client) + result["calls"] = records + return result + finally: + print(json.dumps({"provider_attempts": records}, ensure_ascii=False)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--execute", action="store_true", help="Real paid calls; requires prior human authorization, not granted by this flag") + parser.add_argument("--env-file", type=Path) + args = parser.parse_args() + if not args.execute: + print(json.dumps({"mode": "PLAN_ONLY", "limits": LIMITS, "max_total_request_bytes": MAX_TOTAL_BYTES, + "documents": DOCS, "queries": sorted(QUERIES)}, ensure_ascii=False, indent=2)) + elif args.env_file is None: + parser.error("--execute requires --env-file") + else: + try: + print(json.dumps(execute(args.env_file), ensure_ascii=False, indent=2)) + except Exception as exc: + print(f"Smoke failed ({type(exc).__name__}); no automatic rerun. Provider bodies are not printed.", file=sys.stderr) + raise SystemExit(1) from None + + +if __name__ == "__main__": + main() diff --git a/tests/test_hybrid_search.py b/tests/test_hybrid_search.py index 8de5edf6..1c646c9f 100644 --- a/tests/test_hybrid_search.py +++ b/tests/test_hybrid_search.py @@ -268,3 +268,79 @@ def malformed(*args, **kwargs): assert "secret-key" not in response.text + logs[-1].error_message if failure == "disabled": assert calls == [] + + +def test_full_synthetic_smoke_flow_offline(runtime, tmp_path, monkeypatch): + from auth import get_current_user + from scripts.smoke_hybrid_runtime import run_flow + import services.agent_ops_service as agent_ops + + engine, _ = runtime + monkeypatch.setattr(agent_ops, "engine", engine) + monkeypatch.setenv("DOCUMENT_STORAGE_ROOT", str(tmp_path / "uploads")) + monkeypatch.setenv("AUTH_DEMO_USERS_JSON", "") + main.app.dependency_overrides.pop(get_current_user, None) + client = chromadb.EphemeralClient(settings=chromadb.Settings(anonymized_telemetry=False)) + collection = client.create_collection("smoke_" + uuid4().hex, embedding_function=None) + import rag_runtime.query_chroma as dense + monkeypatch.setattr(dense, "get_collection", lambda: collection) + monkeypatch.setattr(hybrid, "get_collection", lambda: collection) + monkeypatch.setattr(documents, "get_document_chroma_collection", lambda **_: collection) + monkeypatch.setattr(dense, "embed_texts", hybrid.embed_texts) + # index_document binds its embedding default at definition time. + original_index = documents.index_document + import routers.documents as document_router + monkeypatch.setattr(document_router, "index_document_service", lambda **kwargs: original_index( + **kwargs, embedding_function=hybrid.embed_texts)) + + def response(request): + body = json.loads(request.content) + order = sorted(range(len(body["documents"])), key=lambda i: "VPN error 809" not in body["documents"][i]) + return httpx.Response(200, json={"results": [ + {"index": i, "relevance_score": 1 / (rank + 1)} for rank, i in enumerate(order)]}) + monkeypatch.setattr(hybrid, "get_rerank_client", lambda: OpenAI(api_key="offline", + base_url="https://rerank.invalid/v1", max_retries=0, + http_client=httpx.Client(transport=httpx.MockTransport(response)))) + try: + with TestClient(main.app) as api: + assert run_flow(api)["status"] == "PASS" + finally: + client.delete_collection(collection.name) + + +def test_smoke_transport_bounds_real_attempts_and_synthetic_egress(): + from scripts.smoke_hybrid_runtime import BoundedTransport, LIMITS, VPN_QUERY + + records, sent = [], [] + + def respond(request): + sent.append(request) + return httpx.Response(200, json={"usage": {"total_tokens": 5}}) + url = "https://dashscope.aliyuncs.com/compatible-mode/v1" + with httpx.Client(transport=BoundedTransport("embedding", url, records, httpx.MockTransport(respond))) as client: + for _ in range(LIMITS["embedding"]): + assert client.post(url + "/embeddings", json={"model": "text-embedding-v4", "input": [VPN_QUERY]}).status_code == 200 + with pytest.raises(RuntimeError, match="budget"): + client.post(url + "/embeddings", json={"model": "text-embedding-v4", "input": [VPN_QUERY]}) + with pytest.raises(RuntimeError, match="Unapproved embedding text"): + client.post(url + "/embeddings", json={"model": "text-embedding-v4", "input": ["private data"]}) + with pytest.raises(RuntimeError, match="endpoint"): + client.post("https://foreign.invalid/embeddings", json={}) + assert len(sent) == LIMITS["embedding"] and all(r["total_tokens"] == 5 for r in records) + + +def test_smoke_transport_counts_failed_sends_and_rejects_foreign_rerank(): + from scripts.smoke_hybrid_runtime import BoundedTransport, DOCS, VPN_QUERY + + records = [] + url = "https://ws-test.ap-southeast-1.maas.aliyuncs.com/compatible-api/v1" + + def fail(request): + raise httpx.ConnectError("simulated transport failure") + with httpx.Client(transport=BoundedTransport("rerank", url, records, httpx.MockTransport(fail))) as client: + with pytest.raises(RuntimeError, match="Unapproved rerank text"): + client.post(url + "/reranks", json={"model": "qwen3-rerank", "query": VPN_QUERY, "documents": [DOCS[-1][-1]]}) + assert records == [] + with pytest.raises(httpx.ConnectError): + client.post(url + "/reranks", json={"model": "qwen3-rerank", "query": VPN_QUERY, "documents": [DOCS[0][-1]]}) + assert len(records) == 1 and records[0]["status"] is None