wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(ai): 建立 PixelRAG VLM route readiness

Browse Source
This commit is contained in:
ogt
2026-07-10 00:02:22 +08:00
parent c82bfc1835
commit d1ca067054
15 changed files with 722 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
tests/test_ai_automation_smoke_service.py
View File

@@ -1434,12 +1434,13 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_ocr_vlm_replay_check", lambda: smoke._check("pixelrag ocr vlm", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check("pixelrag vlm route", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check("pixelrag vlm worker", "ok", "ok"))
result = smoke.collect_ai_automation_smoke(record_history=False)
assert result["status"] == "critical"
assert result["summary"] == {"ok": 33, "warning": 1, "critical": 1, "total": 35}
assert result["summary"] == {"ok": 34, "warning": 1, "critical": 1, "total": 36}
def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch):
@@ -3969,6 +3970,7 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat
monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_ocr_vlm_replay_check", lambda: smoke._check("pixelrag ocr vlm", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check("pixelrag vlm route", "ok", "ok"))
monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check("pixelrag vlm worker", "ok", "ok"))
first = smoke.collect_ai_automation_smoke(history_limit=5)
@@ -4025,7 +4027,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
json.dumps({
"generated_at": datetime.now().isoformat(timespec="seconds"),
"status": "ok",
"summary": {"ok": 35, "warning": 0, "critical": 0, "total": 35},
"summary": {"ok": 36, "warning": 0, "critical": 0, "total": 36},
"checks": [
{
"name": "PChome 受控落地 drift monitor",
@@ -4151,6 +4153,24 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
"next_machine_action": "run_ollama_first_vlm_replay_worker",
},
},
{
"name": "PixelRAG VLM route readiness",
"status": "ok",
"summary": "PixelRAG VLM route readiness hosts=1/5, configured=minicpm-v:latest, configured_available=1, candidate=minicpm-v:latest",
"details": {
"policy": "read_only_pixelrag_vlm_route_readiness_v1",
"host_count": 5,
"reachable_host_count": 1,
"configured_model": "minicpm-v:latest",
"configured_model_available_count": 1,
"candidate_model": "minicpm-v:latest",
"candidate_host": "http://34.21.145.224:11434",
"candidate_provider": "ollama_secondary",
"model_route_ready": True,
"model_call_performed": False,
"next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
},
},
{
"name": "PixelRAG VLM replay worker",
"status": "ok",
@@ -4165,6 +4185,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
"executed_ok_count": 0,
"executed_warning_count": 0,
"model_error_count": 0,
"model_route_not_ready_count": 0,
"parse_error_count": 0,
"receipt_written_count": 0,
"model_call_performed": False,
@@ -4189,7 +4210,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
)
assert summary["policy"] == "read_only_ai_automation_scheduled_health_summary"
assert summary["status"] == "ok"
assert summary["summary"]["total"] == 32
assert summary["summary"]["total"] == 33
assert summary["summary"]["primary_human_gate_count"] == 0
assert summary["summary"]["writes_database_count"] == 0
assert pchome_family["status"] == "ok"
@@ -5872,6 +5893,14 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects(
assert pixelrag_ocr_vlm_replay_family["status"] == "ok"
assert pixelrag_ocr_vlm_replay_family["details"]["replay_ready_count"] == 1
assert pixelrag_ocr_vlm_replay_family["details"]["extraction_execution_performed"] is False
pixelrag_vlm_route_readiness_family = next(
item for item in summary["families"]
if item["key"] == "pixelrag_vlm_route_readiness"
)
assert pixelrag_vlm_route_readiness_family["status"] == "ok"
assert pixelrag_vlm_route_readiness_family["details"]["model_route_ready"] is True
assert pixelrag_vlm_route_readiness_family["details"]["model_call_performed"] is False
assert pixelrag_vlm_route_readiness_family["details"]["writes_database_count"] == 0
pixelrag_vlm_replay_worker_family = next(
item for item in summary["families"]
if item["key"] == "pixelrag_vlm_replay_worker"
@@ -6502,6 +6531,11 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
"ok",
"pixelrag ocr vlm ok",
))
monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check(
"PixelRAG VLM route readiness",
"ok",
"pixelrag vlm route ok",
))
monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check(
"PixelRAG VLM replay worker",
"ok",
@@ -6522,7 +6556,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch):
item for item in result["checks"]
if item["name"] == "Sitewide visual QA readback"
)
assert result["summary"]["total"] == 35
assert result["summary"]["total"] == 36
assert surface_check["status"] == "ok"
assert surface_check["details"]["checked_surface_count"] == 10
assert sitewide_check["status"] == "ok"

82
tests/test_pixelrag_vlm_replay_worker_service.py
View File

@@ -106,6 +106,7 @@ def test_pixelrag_vlm_replay_worker_execute_writes_artifact_receipt(tmp_path, mo
execute=True,
write_receipt=True,
tile_limit=1,
auto_select_model=False,
)
assert payload["status"] == "ok"
@@ -160,6 +161,7 @@ def test_pixelrag_vlm_replay_worker_writes_model_error_receipt(tmp_path, monkeyp
execute=True,
write_receipt=True,
tile_limit=1,
auto_select_model=False,
)
assert payload["status"] == "critical"
@@ -175,6 +177,86 @@ def test_pixelrag_vlm_replay_worker_writes_model_error_receipt(tmp_path, monkeyp
assert receipt["next_machine_action"] == "repair_ollama_vlm_runtime_or_model_route"
def test_pixelrag_vlm_replay_worker_auto_selects_installed_candidate(tmp_path, monkeypatch):
from services import pixelrag_vlm_replay_worker_service as service
_write_receipt(
tmp_path,
platform="shopee_tw",
manifest_id="shopee-ok",
title="Shopee 防曬乳",
url="https://shopee.tw/search?keyword=sunscreen",
)
monkeypatch.setattr(
service,
"build_pixelrag_vlm_route_readiness",
lambda **kwargs: {
"policy": "read_only_pixelrag_vlm_route_readiness_v1",
"status": "warning",
"summary": {
"configured_model": "minicpm-v:latest",
"candidate_model": "gemma3:4b",
"candidate_host": "http://34.21.145.224:11434",
"model_route_ready": True,
},
"next_machine_action": "run_pixelrag_vlm_replay_worker_execute_with_selected_model",
},
)
class FakeOllama:
def __init__(self, model):
self.model = model
def generate(self, *args, **kwargs):
assert kwargs["model"] == "gemma3:4b"
return SimpleNamespace(
success=True,
content=json.dumps({
"blocked_page_detected": False,
"fields": {
"title": {
"value": "防曬乳 SPF50",
"confidence": 0.92,
"evidence_refs": ["tile:1"],
},
"price": {
"value": "399",
"confidence": 0.88,
"evidence_refs": ["tile:1"],
},
},
"quality": {
"overall_confidence": 0.90,
"missing_required_fields": [],
"requires_identity_matcher_replay": True,
"requires_promotion_gate": True,
},
}),
model="gemma3:4b",
error=None,
total_duration=1.0,
host="http://34.21.145.224:11434",
input_tokens=10,
output_tokens=50,
)
monkeypatch.setattr(service, "OllamaService", FakeOllama)
payload = service.run_pixelrag_ollama_vlm_replay_worker(
artifact_root=tmp_path,
platform="shopee_tw",
model="minicpm-v:latest",
execute=True,
tile_limit=1,
auto_select_model=True,
)
assert payload["model"] == "gemma3:4b"
assert payload["configured_model"] == "minicpm-v:latest"
assert payload["route_readiness"]["summary"]["candidate_model"] == "gemma3:4b"
assert payload["summary"]["executed_ok_count"] == 1
def test_pixelrag_vlm_replay_worker_cli_outputs_machine_readable_json(tmp_path):
_write_receipt(
tmp_path,

110
tests/test_pixelrag_vlm_route_readiness_service.py Normal file
View File

@@ -0,0 +1,110 @@
import json
class FakeResponse:
def __init__(self, status_code=200, payload=None, text=""):
self.status_code = status_code
self._payload = payload or {}
self.text = text
def json(self):
return self._payload
def test_pixelrag_vlm_route_readiness_selects_installed_candidate(monkeypatch):
from services import pixelrag_vlm_route_readiness_service as service
hosts = [
"http://primary:11434",
"http://secondary:11434",
]
monkeypatch.setattr(service, "_approved_hosts", lambda: hosts)
def fake_get(url, timeout):
if url.startswith("http://primary"):
raise TimeoutError("primary timeout")
return FakeResponse(
payload={
"models": [
{"name": "gemma3:4b"},
{"name": "bge-m3:latest"},
]
}
)
monkeypatch.setattr(service.requests, "get", fake_get)
payload = service.build_pixelrag_vlm_route_readiness(
model="minicpm-v:latest",
timeout_seconds=1,
)
assert payload["policy"] == "read_only_pixelrag_vlm_route_readiness_v1"
assert payload["status"] == "warning"
assert payload["summary"]["reachable_host_count"] == 1
assert payload["summary"]["configured_model_available_count"] == 0
assert payload["summary"]["candidate_model"] == "gemma3:4b"
assert payload["summary"]["model_route_ready"] is True
assert payload["controlled_apply"]["model_call"] is False
assert payload["controlled_apply"]["writes_database"] is False
assert payload["next_machine_action"] == "run_pixelrag_vlm_replay_worker_execute_with_selected_model"
def test_pixelrag_vlm_route_readiness_critical_when_no_candidate(monkeypatch):
from services import pixelrag_vlm_route_readiness_service as service
monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
monkeypatch.setattr(
service.requests,
"get",
lambda url, timeout: FakeResponse(payload={"models": [{"name": "bge-m3:latest"}]}),
)
payload = service.build_pixelrag_vlm_route_readiness(model="minicpm-v:latest")
assert payload["status"] == "critical"
assert payload["success"] is False
assert payload["summary"]["model_route_ready"] is False
assert payload["next_machine_action"] == "install_or_configure_pixelrag_vlm_model_on_approved_ollama_host"
def test_pixelrag_vlm_route_readiness_route_returns_readback(monkeypatch):
from flask import Flask
from routes import system_public_routes as routes
from services import pixelrag_vlm_route_readiness_service as service
monkeypatch.setattr(
service,
"build_pixelrag_vlm_route_readiness",
lambda **kwargs: {
"success": True,
"policy": "read_only_pixelrag_vlm_route_readiness_v1",
"status": "ok",
"summary": {"model_route_ready": True, "candidate_model": "gemma3:4b"},
"next_machine_action": "run_pixelrag_vlm_replay_worker_execute",
},
)
app = Flask(__name__)
with app.test_request_context(
"/api/ai-automation/pixelrag-vlm-route-readiness?model=minicpm-v:latest"
):
response = routes.ai_automation_pixelrag_vlm_route_readiness_api.__wrapped__()
payload = response.get_json()
assert payload["policy"] == "read_only_pixelrag_vlm_route_readiness_v1"
assert payload["summary"]["candidate_model"] == "gemma3:4b"
def test_pixelrag_vlm_route_readiness_cli_outputs_json(monkeypatch, tmp_path):
from services import pixelrag_vlm_route_readiness_service as service
monkeypatch.setattr(service, "_approved_hosts", lambda: ["http://secondary:11434"])
monkeypatch.setattr(
service.requests,
"get",
lambda url, timeout: FakeResponse(payload={"models": [{"name": "gemma3:4b"}]}),
)
payload = service.build_pixelrag_vlm_route_readiness(model="minicpm-v:latest")
assert json.dumps(payload, ensure_ascii=False)
assert payload["summary"]["candidate_model"] == "gemma3:4b"