Spaces:
Configuration error
Configuration error
File size: 5,528 Bytes
ad0932c 525124a ad0932c 525124a | 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 157 158 159 160 | """Relationship context compaction tests."""
import json
from pycatan.ai.agent_state import AgentState
from pycatan.ai.config import AIConfig
from pycatan.ai.llm_client import LLMResponse
from pycatan.ai.memory_compactor import MemoryCompactor
from pycatan.ai.schemas import (
ResponseType,
SchemaVersion,
get_schema_description,
get_schema_for_response_type,
)
class _FakeLLMClient:
def generate(self, *args, **kwargs):
return LLMResponse(
success=True,
content=json.dumps({
"compacted_memory": "Prefer ore expansion; Shon may be racing city upgrades.",
"recent_notes_to_keep": ["Recent note A", "Recent note B"],
"relationship_updates": [
"Shon refused a fair trade, making future promises less credible.",
],
"discarded_as_irrelevant": [],
}),
)
class _BrokenJSONLLMClient:
def generate(self, *args, **kwargs):
return LLMResponse(
success=True,
content="I cannot provide that as JSON.",
model="fake-model",
)
class _FailingLLMClient:
def generate(self, *args, **kwargs):
return LLMResponse(
success=False,
error="provider rejected response_format",
model="fake-model",
)
def test_player_response_schemas_do_not_ask_for_relationship_update():
for version in [SchemaVersion.V1, SchemaVersion.V2]:
for response_type in [ResponseType.ACTIVE_TURN, ResponseType.OBSERVING]:
schema = get_schema_for_response_type(response_type, version)
assert "relationship_update" not in schema["properties"]
assert "relationship_update" not in schema["propertyOrdering"]
assert "relationship_update" not in get_schema_description(response_type, version)
def test_memory_compactor_extracts_relationship_updates():
agent = AgentState(player_name="Hadar", player_id=0, player_color="Red")
agent.memory_history = [
{"note": "Shon refused my fair trade after promising to help."},
{"note": "Recent note A"},
{"note": "Recent note B"},
]
result = MemoryCompactor(AIConfig()).compact(
agent=agent,
game_state={
"meta": {"curr": "Hadar", "phase": "NORMAL_PLAY"},
"H": [],
"N": [],
"state": {"bld": [], "rds": []},
"players": {"Hadar": {"vp": 0, "res": {}}, "Shon": {"vp": 0, "res": {}}},
},
chat_history=[
{"from": "Shon", "message": "I'll help next time, promise."},
],
llm_client=_FakeLLMClient(),
)
assert result is not None
assert result["relationship_updates"] == [
"Shon refused a fair trade, making future promises less credible.",
]
assert "relationship_updates" in result["prompt"]["output_requirements"]["schema"]
assert "existing_relationship_updates" in result["prompt"]["memory_input"]
def test_memory_compactor_filters_repeated_relationship_updates():
compactor = MemoryCompactor(AIConfig())
updates = compactor._clean_relationship_updates(
[
"Shon refused a fair trade.",
"Shon refused a fair trade.",
"Hadar backed my warning.",
],
[{"note": "Shon refused a fair trade."}],
)
assert updates == ["Hadar backed my warning."]
def test_memory_compactor_falls_back_when_model_returns_unparseable_json():
agent = AgentState(player_name="Hadar", player_id=0, player_color="Red")
agent.memory_history = [
{"note": "Need brick and sheep to build the winning settlement at node 23."},
{"note": "Ziv is blocking trades that help Hadar win."},
{"note": "Recent note A"},
{"note": "Recent note B"},
]
result = MemoryCompactor(AIConfig()).compact(
agent=agent,
game_state={
"meta": {"curr": "Hadar", "phase": "NORMAL_PLAY"},
"H": [],
"N": [],
"state": {"bld": [], "rds": []},
"players": {"Hadar": {"vp": 4, "res": {}}},
},
chat_history=[],
llm_client=_BrokenJSONLLMClient(),
)
assert result is not None
assert result["fallback_used"] is True
assert result["fallback_reason"] == "unparseable_response"
assert "winning settlement" in result["compacted_memory"]
assert result["recent_entries"] == agent.memory_history[-2:]
def test_memory_compactor_falls_back_when_llm_call_fails():
agent = AgentState(player_name="Ziv", player_id=1, player_color="Blue")
agent.memory_history = [
{"note": "Hadar is at 4 VP and must not receive brick."},
{"note": "Need wood and brick for my own road."},
{"note": "Recent note A"},
{"note": "Recent note B"},
]
result = MemoryCompactor(AIConfig()).compact(
agent=agent,
game_state={
"meta": {"curr": "Ziv", "phase": "NORMAL_PLAY"},
"H": [],
"N": [],
"state": {"bld": [], "rds": []},
"players": {"Ziv": {"vp": 3, "res": {}}},
},
chat_history=[{"from": "Hadar", "message": "Can anyone trade brick?"}],
llm_client=_FailingLLMClient(),
)
assert result is not None
assert result["fallback_used"] is True
assert result["fallback_reason"] == "llm_error: provider rejected response_format"
assert "must not receive brick" in result["compacted_memory"]
|