Buckets:
| import pytest | |
| from utils import * | |
| from test_vision_api import get_img_url | |
| server = ServerPreset.tinylaya() | |
| def create_server(): | |
| global server | |
| server = ServerPreset.tinylaya() | |
| TEST_STATE = "I was charged twice for my order last week and nobody has replied." | |
| TEST_QUESTIONS = { | |
| "route": { | |
| "type": "choice", | |
| "instructions": "Which team should handle this?", | |
| "criteria": {"billing": "payments and refunds", "shipping": None, "technical": None}, | |
| }, | |
| "urgency": { | |
| "type": "score", | |
| "instructions": "How urgent is this?", | |
| "criteria": ["can wait", "this week", "today", "right now"], | |
| }, | |
| "angry": { | |
| "type": "noul", | |
| "instructions": "Is the customer angry?", | |
| }, | |
| } | |
| def get_prompt_metrics(server: ServerProcess) -> tuple[int, int]: | |
| """returns the number of prompt tokens (processed, cached) since the server started""" | |
| res = server.make_request("GET", "/metrics") | |
| assert res.status_code == 200 | |
| values = {} | |
| for line in res.body.splitlines(): | |
| if line.startswith("llamacpp:"): | |
| name, value = line.split(" ") | |
| values[name] = int(float(value)) | |
| return values["llamacpp:prompt_tokens_total"], values["llamacpp:prompt_tokens_cached_total"] | |
| def test_systemone(preset: str): | |
| global server | |
| server = getattr(ServerPreset, preset)() | |
| server.start() | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| }) | |
| assert res.status_code == 200 | |
| assert res.body["usage"]["input_tokens"] > 0 | |
| assert res.body["usage"]["output_tokens"] == 0 | |
| answers = res.body["answers"] | |
| assert list(answers.keys()) == ["route", "urgency", "angry"] | |
| route = answers["route"] | |
| assert route["type"] == "choice" | |
| assert list(route["probabilities"].keys()) == ["billing", "shipping", "technical"] | |
| assert abs(sum(route["probabilities"].values()) - 1.0) < 1e-4 | |
| assert route["choice"] == max(route["probabilities"], key=route["probabilities"].get) | |
| assert 0.0 <= route["confidence"] <= 1.0 | |
| urgency = answers["urgency"] | |
| assert urgency["type"] == "score" | |
| assert urgency["legend"] == {"0": "can wait", "1": "this week", "2": "today", "3": "right now"} | |
| assert list(urgency["probabilities"].keys()) == ["0", "1", "2", "3"] | |
| assert abs(sum(urgency["probabilities"].values()) - 1.0) < 1e-4 | |
| assert abs(urgency["score"] - sum(i * p for i, p in enumerate(urgency["probabilities"].values()))) < 1e-4 | |
| assert 0.0 <= urgency["confidence"] <= 1.0 | |
| angry = answers["angry"] | |
| assert angry["type"] == "noul" | |
| assert 0.0 <= angry["noul"] <= 1.0 | |
| def test_systemone_json_state(): | |
| global server | |
| server.start() | |
| questions = { | |
| "refund": { | |
| "type": "noul", | |
| "instructions": "Is a refund requested?", | |
| "criteria": {"false": "no refund is asked", "true": "a refund is asked"}, | |
| }, | |
| } | |
| res_obj = server.make_request("POST", "/v1/systemone", data={ | |
| "state": {"ticket": TEST_STATE, "plan": "pro"}, | |
| "questions": questions, | |
| }) | |
| assert res_obj.status_code == 200 | |
| # an object is given to the model as JSON text | |
| res_str = server.make_request("POST", "/v1/systemone", data={ | |
| "state": '{"ticket": "' + TEST_STATE + '", "plan": "pro"}', | |
| "questions": questions, | |
| }) | |
| assert res_str.status_code == 200 | |
| assert res_obj.body["usage"] == res_str.body["usage"] | |
| assert abs(res_obj.body["answers"]["refund"]["noul"] - res_str.body["answers"]["refund"]["noul"]) < 1e-4 | |
| def test_systemone_invalid_request(data: dict): | |
| global server | |
| server.start() | |
| res = server.make_request("POST", "/v1/systemone", data=data) | |
| assert res.status_code == 400 | |
| assert "error" in res.body | |
| def test_systemone_shared_prompt(): | |
| global server | |
| server = ServerPreset.tinyopenjev() | |
| server.server_metrics = True | |
| server.start() | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| }) | |
| assert res.status_code == 200 | |
| # the first question evaluates the shared prefix, the 2 others start from it | |
| n_processed, n_cached = get_prompt_metrics(server) | |
| assert n_cached > 0 | |
| assert n_cached % 2 == 0 | |
| assert n_processed + n_cached == res.body["usage"]["input_tokens"] | |
| # with one slot the prompt cannot be shared, the answers must be the same | |
| server.stop() | |
| server = ServerPreset.tinyopenjev() | |
| server.n_slots = 1 | |
| server.start() | |
| res_single = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| }) | |
| assert res_single.status_code == 200 | |
| assert res_single.body["usage"] == res.body["usage"] | |
| for qid in ["route", "urgency"]: | |
| probs_shared = res.body["answers"][qid]["probabilities"] | |
| probs_single = res_single.body["answers"][qid]["probabilities"] | |
| for key in probs_shared: | |
| assert abs(probs_shared[key] - probs_single[key]) < 0.01 | |
| assert abs(res.body["answers"]["angry"]["noul"] - res_single.body["answers"]["angry"]["noul"]) < 0.01 | |
| def test_systemone_prompt_cache(cache_ram, tmp_path): | |
| global server | |
| server.n_slots = 1 | |
| server.cache_ram = cache_ram | |
| server.debug = True | |
| server.log_path = str(tmp_path / "server.log") | |
| server.start() | |
| # the third request comes back to the state of the first one | |
| for state in ["Ticket A: " + TEST_STATE, "Ticket B: " + TEST_STATE * 2, "Ticket A: " + TEST_STATE]: | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": state, | |
| "questions": TEST_QUESTIONS, | |
| }) | |
| assert res.status_code == 200 | |
| server.stop() | |
| with open(server.log_path) as f: | |
| log = f.read() | |
| # the prompt cache is opt-in for decision models, and restores their states once enabled | |
| assert ("__TEST_TAG_CACHE_IDLE_SLOTS_ENABLED__" in log) == (cache_ram is not None) | |
| assert ("found better prompt" in log) == (cache_ram is not None) | |
| def test_systemone_prompt_cache_child_slots(tmp_path): | |
| global server | |
| server = ServerPreset.tinyopenjev() | |
| server.n_slots = len(TEST_QUESTIONS) | |
| server.cache_ram = 100 | |
| server.debug = True | |
| server.log_path = str(tmp_path / "server.log") | |
| server.start() | |
| states = [ | |
| "My parcel shows delivered but it never arrived at my door.", | |
| "The app crashes every time I open the settings page.", | |
| "Please close my account and delete all my personal data.", | |
| ] | |
| angry = {"angry": TEST_QUESTIONS["angry"]} | |
| def decide(state, questions): | |
| res = server.make_request("POST", "/v1/systemone", data={"state": state, "questions": questions}) | |
| assert res.status_code == 200 | |
| # one state per slot, then a request whose child tasks take every slot | |
| for state in states: | |
| decide(state, angry) | |
| decide(TEST_STATE, TEST_QUESTIONS) | |
| with open(server.log_path) as f: | |
| n_restored = f.read().count("found better prompt with f_keep = 1.000") | |
| # every overwritten state comes back whole from the prompt cache | |
| for state in states: | |
| decide(state, angry) | |
| server.stop() | |
| with open(server.log_path) as f: | |
| assert f.read().count("found better prompt with f_keep = 1.000") == n_restored + len(states) | |
| def test_systemone_images(): | |
| global server | |
| server = ServerPreset.tinyopenjev() | |
| server.start() | |
| image = get_img_url("IMG_BASE64_URI_0") | |
| res_text = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| }) | |
| assert res_text.status_code == 200 | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| "images": [image], | |
| }) | |
| assert res.status_code == 200 | |
| assert list(res.body["answers"].keys()) == ["route", "urgency", "angry"] | |
| assert res.body["usage"]["input_tokens"] > res_text.body["usage"]["input_tokens"] | |
| # same image, given as a part of a chat message | |
| res_part = server.make_request("POST", "/v1/systemone", data={ | |
| "state": [{"role": "user", "content": [ | |
| {"type": "image_url", "image_url": {"url": image}}, | |
| {"type": "text", "text": TEST_STATE}, | |
| ]}], | |
| "questions": TEST_QUESTIONS, | |
| }) | |
| assert res_part.status_code == 200 | |
| assert res_part.body["usage"]["input_tokens"] > res_text.body["usage"]["input_tokens"] | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| "images": [image] * 9, | |
| }) | |
| assert res.status_code == 400 | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| "images": ["https://example.com/image.png"], | |
| }) | |
| assert res.status_code == 400 | |
| def test_systemone_images_not_supported(): | |
| global server | |
| server.start() | |
| res = server.make_request("POST", "/v1/systemone", data={ | |
| "state": TEST_STATE, | |
| "questions": TEST_QUESTIONS, | |
| "images": [get_img_url("IMG_BASE64_URI_0")], | |
| }) | |
| assert res.status_code == 501 | |
Xet Storage Details
- Size:
- 10 kB
- Xet hash:
- b588bb2d2a9bf36eb2b2e6dfd23f1130b6305a55daad76a04441941cb5707abd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.