Add TensorX models for Versuch 2
This commit is contained in:
Christoph Schwörer
@@ -0,0 +1,244 @@
|
||||
import importlib.util
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
ADAPTER_PATH = Path(__file__).with_name("glm-kimi-adapter.py")
|
||||
SPEC = importlib.util.spec_from_file_location("glm_kimi_adapter", ADAPTER_PATH)
|
||||
ADAPTER = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(ADAPTER)
|
||||
|
||||
|
||||
def response(message, finish_reason="tool_calls", tokens=10):
|
||||
return {
|
||||
"model": "test/model",
|
||||
"usage": {
|
||||
"prompt_tokens": tokens,
|
||||
"completion_tokens": 1,
|
||||
"total_tokens": tokens + 1,
|
||||
"prompt_tokens_details": {"cached_tokens": 0},
|
||||
"completion_tokens_details": {"reasoning_tokens": 0},
|
||||
},
|
||||
"choices": [{"finish_reason": finish_reason, "message": message}],
|
||||
}
|
||||
|
||||
|
||||
def subagent_result(agent_id):
|
||||
return {
|
||||
"result": f"Ergebnis {agent_id}",
|
||||
"usage": {
|
||||
"prompt_tokens": 2,
|
||||
"completion_tokens": 1,
|
||||
"total_tokens": 3,
|
||||
"cached_tokens": 0,
|
||||
"reasoning_tokens": 0,
|
||||
},
|
||||
"turns": 1,
|
||||
"tool_calls": 0,
|
||||
"errors": [],
|
||||
"status": "completed",
|
||||
}
|
||||
|
||||
|
||||
class AdapterTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.provider = {
|
||||
"base_url": "https://example.invalid/v1",
|
||||
"__id": "test",
|
||||
}
|
||||
self.liveness = ADAPTER.LivenessMonitor(interval_seconds=0)
|
||||
|
||||
def test_timeout_zero_is_forwarded_as_no_requests_timeout(self):
|
||||
captured = {}
|
||||
|
||||
class FakeResponse:
|
||||
status_code = 200
|
||||
text = ""
|
||||
|
||||
@staticmethod
|
||||
def json():
|
||||
return {"ok": True}
|
||||
|
||||
def fake_post(*args, **kwargs):
|
||||
captured["timeout"] = kwargs["timeout"]
|
||||
return FakeResponse()
|
||||
|
||||
with patch.object(ADAPTER.requests, "post", side_effect=fake_post):
|
||||
result = ADAPTER.post_chat_completion(
|
||||
"https://example.invalid", {}, {}, 0, self.liveness,
|
||||
"test", "Testaufruf",
|
||||
)
|
||||
|
||||
self.assertEqual({"ok": True}, result)
|
||||
self.assertIsNone(captured["timeout"])
|
||||
|
||||
def test_requested_tensorx_models_have_explicit_effort_mapping(self):
|
||||
self.assertEqual("thinking", ADAPTER.MODEL_EFFORT_TYPE["qwen"])
|
||||
self.assertEqual("thinking", ADAPTER.MODEL_EFFORT_TYPE["z-ai"])
|
||||
|
||||
def test_custom_mode_exposes_spawn_subagent(self):
|
||||
captured = {}
|
||||
|
||||
def fake_post(url, headers, body, timeout, liveness,
|
||||
activity_id, activity_label):
|
||||
captured["tools"] = body["tools"]
|
||||
return response(
|
||||
{"role": "assistant", "content": "Fertig."},
|
||||
finish_reason="stop",
|
||||
)
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir, \
|
||||
patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post):
|
||||
result = ADAPTER.run_agent_loop(
|
||||
provider=self.provider,
|
||||
model="qwen/qwen3.8-flash-next",
|
||||
system_prompt="System",
|
||||
user_prompt="Aufgabe",
|
||||
api_key="key",
|
||||
effort="high",
|
||||
root=temp_dir,
|
||||
output_dir=temp_dir,
|
||||
max_turns=0,
|
||||
temperature=1.0,
|
||||
timeout=0,
|
||||
liveness=self.liveness,
|
||||
mode="custom",
|
||||
)
|
||||
|
||||
tool_names = {
|
||||
tool["function"]["name"] for tool in captured["tools"]
|
||||
}
|
||||
self.assertIn("spawn_subagent", tool_names)
|
||||
self.assertFalse(result["is_error"])
|
||||
|
||||
def test_liveness_monitor_reports_active_main_and_subagent_operations(self):
|
||||
lines = []
|
||||
monitor = ADAPTER.LivenessMonitor(interval_seconds=0.02)
|
||||
with patch.object(ADAPTER, "log_stderr", side_effect=lines.append):
|
||||
monitor.set("main", "Hauptagent wartet auf Subagenten")
|
||||
monitor.set("subagent-1", "Subagent 1 wartet auf API-Antwort")
|
||||
monitor.start()
|
||||
time.sleep(0.06)
|
||||
monitor.stop()
|
||||
|
||||
output = "\n".join(lines)
|
||||
self.assertIn("LIFESIGN: Prozess lebt", output)
|
||||
self.assertIn("Hauptagent wartet auf Subagenten", output)
|
||||
self.assertIn("Subagent 1 wartet auf API-Antwort", output)
|
||||
self.assertIn("keinen serverseitigen Fortschritt", output)
|
||||
|
||||
def test_subagents_from_one_turn_run_in_parallel_without_count_limit(self):
|
||||
first_message = {
|
||||
"role": "assistant",
|
||||
"content": "Ich delegiere.",
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": f"spawn-{index}",
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "spawn_subagent",
|
||||
"arguments": (
|
||||
'{"description":"Aufgabe %d",'
|
||||
'"subagent_type":"explore"}' % index
|
||||
),
|
||||
},
|
||||
}
|
||||
for index in range(12)
|
||||
],
|
||||
}
|
||||
responses = [
|
||||
response(first_message),
|
||||
response(
|
||||
{"role": "assistant", "content": "Fertig."},
|
||||
finish_reason="stop",
|
||||
),
|
||||
]
|
||||
starts = []
|
||||
|
||||
def fake_post(*args, **kwargs):
|
||||
return responses.pop(0)
|
||||
|
||||
def fake_subagent(*args, **kwargs):
|
||||
agent_id = args[10]
|
||||
starts.append(time.monotonic())
|
||||
time.sleep(0.15)
|
||||
return subagent_result(agent_id)
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
started = time.monotonic()
|
||||
with patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post), \
|
||||
patch.object(ADAPTER, "run_subagent", side_effect=fake_subagent):
|
||||
result = ADAPTER.run_agent_loop(
|
||||
provider=self.provider,
|
||||
model="test/model",
|
||||
system_prompt="System",
|
||||
user_prompt="Aufgabe",
|
||||
api_key="key",
|
||||
effort="high",
|
||||
root=temp_dir,
|
||||
output_dir=temp_dir,
|
||||
max_turns=0,
|
||||
temperature=1.0,
|
||||
timeout=0,
|
||||
liveness=self.liveness,
|
||||
mode="builtin",
|
||||
subagent_max_turns=0,
|
||||
)
|
||||
elapsed = time.monotonic() - started
|
||||
|
||||
self.assertEqual(12, result["subagent_stats"]["spawned"])
|
||||
self.assertEqual(12, result["subagent_stats"]["completed"])
|
||||
self.assertEqual(0, result["subagent_stats"]["failed"])
|
||||
self.assertLess(max(starts) - min(starts), 0.12)
|
||||
self.assertLess(elapsed, 0.6)
|
||||
|
||||
def test_api_error_is_not_masked_by_earlier_content(self):
|
||||
first_message = {
|
||||
"role": "assistant",
|
||||
"content": "Zwischenstand",
|
||||
"tool_calls": [{
|
||||
"id": "list-1",
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "list_directory",
|
||||
"arguments": '{"path":""}',
|
||||
},
|
||||
}],
|
||||
}
|
||||
calls = [response(first_message), RuntimeError("Transportfehler")]
|
||||
|
||||
def fake_post(*args, **kwargs):
|
||||
item = calls.pop(0)
|
||||
if isinstance(item, Exception):
|
||||
raise item
|
||||
return item
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir, \
|
||||
patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post):
|
||||
result = ADAPTER.run_agent_loop(
|
||||
provider=self.provider,
|
||||
model="test/model",
|
||||
system_prompt="System",
|
||||
user_prompt="Aufgabe",
|
||||
api_key="key",
|
||||
effort="high",
|
||||
root=temp_dir,
|
||||
output_dir=temp_dir,
|
||||
max_turns=0,
|
||||
temperature=1.0,
|
||||
timeout=0,
|
||||
liveness=self.liveness,
|
||||
mode="solo",
|
||||
)
|
||||
|
||||
self.assertEqual("Zwischenstand", result["result"])
|
||||
self.assertTrue(result["is_error"])
|
||||
self.assertEqual("error", result["subtype"])
|
||||
self.assertIn("Transportfehler", result["errors"][0])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user