Spaces:
Running
Running
File size: 4,829 Bytes
79df050 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | import logging
import json
import pytest
from core import memory_system
@pytest.mark.asyncio
async def test_memory_manager_skips_ai_analysis_for_transient_market_query(monkeypatch):
class FailingAnalyzer:
async def analyze_conversation(self, *args, **kwargs):
raise AssertionError("AI memory analysis should be skipped for transient queries")
manager = memory_system.MemoryManager()
manager.analyzer = FailingAnalyzer()
monkeypatch.setattr(memory_system, "_get_memory_client", lambda: object())
monkeypatch.setattr(memory_system, "db_available", False)
result = await manager.process_conversation(
user_id="u1",
user_message="今天台積電股價收盤價多少？",
assistant_response="查詢中。",
conversation_history=[],
)
assert result["extracted_memories"] == 0
assert result["saved_memories"] == 0
assert result["errors"] == []
@pytest.mark.asyncio
async def test_memory_analyzer_quietly_degrades_on_transient_upstream_error(monkeypatch, caplog):
class Responses:
def create(self, **kwargs):
raise Exception("Error code: 502 - {'error': {'message': 'Upstream request failed'}}")
class Client:
responses = Responses()
class Settings:
OPENAI_USE_RESPONSES = True
GPT_INTENT_MODEL = "gpt-5.4"
OPENAI_MODEL = "gpt-5.4"
OPENAI_TIMEOUT = 30
monkeypatch.setattr(memory_system, "_get_memory_client", lambda: Client())
monkeypatch.setattr(memory_system, "settings", Settings)
analyzer = memory_system.MemoryAnalyzer()
with caplog.at_level(logging.ERROR):
memories = await analyzer.analyze_conversation(
user_message="我喜歡安靜的咖啡店",
assistant_response="我記住了。",
conversation_history=[],
)
assert memories == []
assert "AI記憶分析時發生錯誤" not in caplog.text
def test_memory_analyzer_uses_structured_responses_payload(monkeypatch):
captured = {}
class Responses:
def create(self, **kwargs):
captured.update(kwargs)
class Response:
output_text = '{"memories":[]}'
output = []
return Response()
class Client:
responses = Responses()
class Settings:
OPENAI_USE_RESPONSES = True
GPT_INTENT_MODEL = "gpt-5.4-mini"
OPENAI_MODEL = "gpt-5.4-mini"
monkeypatch.setattr(memory_system, "settings", Settings)
analyzer = memory_system.MemoryAnalyzer()
response = analyzer._create_analysis_response(
Client(),
[{"role": "system", "content": "JSON"}, {"role": "user", "content": "JSON"}],
500,
)
assert response.output_text == '{"memories":[]}'
assert captured["model"] == "gpt-5.4-mini"
assert captured["max_output_tokens"] == 500
assert captured["store"] is False
assert captured["text"]["format"]["type"] == "json_schema"
assert captured["text"]["format"]["strict"] is True
assert captured["text"]["format"]["schema"]["required"] == ["memories"]
assert "reasoning" not in captured
@pytest.mark.asyncio
async def test_memory_analyzer_retries_transient_upstream_error_then_succeeds(monkeypatch):
calls = {"count": 0}
class Responses:
def create(self, **kwargs):
calls["count"] += 1
if calls["count"] == 1:
raise Exception("502 upstream request failed")
class Response:
output_text = json.dumps(
{
"memories": [
{
"type": "preferences",
"content": "使用者喜歡安靜的咖啡店",
"importance": 0.8,
}
]
},
ensure_ascii=False,
)
output = []
return Response()
class Client:
responses = Responses()
class Settings:
OPENAI_USE_RESPONSES = True
GPT_INTENT_MODEL = "gpt-5.4-mini"
OPENAI_MODEL = "gpt-5.4-mini"
OPENAI_TIMEOUT = 30
monkeypatch.setattr(memory_system, "_get_memory_client", lambda: Client())
monkeypatch.setattr(memory_system, "settings", Settings)
monkeypatch.setattr(memory_system.MemoryAnalyzer, "_transient_backoff", staticmethod(lambda attempt: _noop()))
analyzer = memory_system.MemoryAnalyzer()
memories = await analyzer.analyze_conversation("我喜歡安靜的咖啡店", "我記住了。", [])
assert calls["count"] == 2
assert memories[0]["type"] == "preferences"
assert memories[0]["source"] == "ai_analysis"
async def _noop():
return None
|