133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Hermes 分析師必須透過 OllamaService 三主機級聯。"""
|
|
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import services.ai_call_logger as logger_mod
|
|
import services.hermes_analyst_service as hermes_mod
|
|
from services.ai_call_logger import _reset_kill_switch
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_ai_logger(monkeypatch):
|
|
_reset_kill_switch()
|
|
captured = []
|
|
|
|
def fake_write(state):
|
|
captured.append({
|
|
'caller': state.caller,
|
|
'provider': state.provider,
|
|
'model': state.model,
|
|
'status': state.status,
|
|
'fallback_to': state.fallback_to,
|
|
'error': state.error,
|
|
'meta': dict(state.meta),
|
|
})
|
|
|
|
monkeypatch.setattr(logger_mod, '_write_to_db', fake_write)
|
|
monkeypatch.setenv('AI_CALL_LOGGING_ENABLED', 'true')
|
|
yield captured
|
|
|
|
|
|
def _wait_for(captured, n=1, timeout=2.0):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if len(captured) >= n:
|
|
return True
|
|
time.sleep(0.01)
|
|
return False
|
|
|
|
|
|
def _stub_ollama(monkeypatch, *, content: str, host: str):
|
|
fake_resp = SimpleNamespace(
|
|
success=True,
|
|
content=content,
|
|
model=hermes_mod.HERMES_MODEL,
|
|
error=None,
|
|
total_duration=1.2,
|
|
host=host,
|
|
input_tokens=33,
|
|
output_tokens=22,
|
|
)
|
|
|
|
class FakeOllamaService:
|
|
instances = []
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.init_args = args
|
|
self.init_kwargs = kwargs
|
|
self.generate_calls = []
|
|
FakeOllamaService.instances.append(self)
|
|
|
|
def generate(self, **kwargs):
|
|
self.generate_calls.append(kwargs)
|
|
return fake_resp
|
|
|
|
monkeypatch.setattr(hermes_mod, 'OllamaService', FakeOllamaService)
|
|
return FakeOllamaService
|
|
|
|
|
|
def test_hermes_intent_uses_ollama_service_and_logs_actual_host(monkeypatch, reset_ai_logger):
|
|
fake_service = _stub_ollama(
|
|
monkeypatch,
|
|
content='{"intent":"query_sales","confidence":0.9,"complexity_score":0.8,'
|
|
'"requires_data_fetch":true,"preliminary_answer":""}',
|
|
host='http://192.168.0.111:11434',
|
|
)
|
|
|
|
svc = hermes_mod.HermesAnalystService()
|
|
result = svc._call_hermes_intent("本週業績如何？")
|
|
|
|
assert result['intent'] == 'query_sales'
|
|
assert result['metadata']['source'] == 'hermes_llm'
|
|
call_kwargs = fake_service.instances[0].generate_calls[0]
|
|
assert call_kwargs['model'] == hermes_mod.HERMES_MODEL
|
|
assert call_kwargs['keep_alive'] == hermes_mod.HERMES_KEEP_ALIVE
|
|
|
|
assert _wait_for(reset_ai_logger, 1)
|
|
rec = reset_ai_logger[0]
|
|
assert rec['caller'] == 'hermes_intent'
|
|
assert rec['provider'] == 'ollama_111'
|
|
assert rec['meta']['host_label'] == '111 備援'
|
|
|
|
|
|
def test_hermes_batch_analyze_uses_ollama_service_and_logs_secondary(monkeypatch, reset_ai_logger):
|
|
fake_service = _stub_ollama(
|
|
monkeypatch,
|
|
content='[{"sku":"A1","name":"測試商品","category":"家電","momo_price":120,'
|
|
'"pchome_price":100,"gap_pct":20,"sales_7d_delta_pct":-30,'
|
|
'"risk":"HIGH","recommended_action":"建議人工評估","confidence":0.8}]',
|
|
host='http://34.21.145.224:11434',
|
|
)
|
|
monkeypatch.setattr(hermes_mod, 'build_mcp_context', lambda *args, **kwargs: 'MCP context')
|
|
|
|
candidates = [{
|
|
'sku': 'A1',
|
|
'name': '測試商品',
|
|
'category': '家電',
|
|
'momo_price': 120,
|
|
'pchome_price': 100,
|
|
'sales_7d_prev': 1000,
|
|
'sales_7d_curr': 700,
|
|
'competitor_tags': [],
|
|
}]
|
|
|
|
svc = hermes_mod.HermesAnalystService()
|
|
raw_threats, items = svc._batch_analyze(candidates)
|
|
|
|
assert raw_threats[0]['sku'] == 'A1'
|
|
assert items[0]['gap_pct'] == 20.0
|
|
call_kwargs = fake_service.instances[0].generate_calls[0]
|
|
assert call_kwargs['system_prompt'] == svc.SYSTEM_PROMPT
|
|
assert call_kwargs['keep_alive'] == hermes_mod.HERMES_KEEP_ALIVE
|
|
|
|
assert _wait_for(reset_ai_logger, 1)
|
|
rec = reset_ai_logger[0]
|
|
assert rec['caller'] == 'hermes_analyst'
|
|
assert rec['provider'] == 'ollama_secondary'
|
|
assert rec['meta']['host_label'] == 'GCP-SSD-2'
|