Spaces:
Sleeping
Sleeping
File size: 6,512 Bytes
8691ce5 | 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 | import json
from app.tasks import order_comms as oc
from app.tasks.order_comms import (
TASK, TAXONOMY, WORLD_FRIENDLY_IDS, WORLD_UUIDS,
execute_tool, generate, lookup_truth, parse_output, score,
)
def _answer(truth: dict, entities: list[dict], classification: dict | None = None) -> str:
obj = {
"classification": classification or truth["classification"],
"entities": [{**e, "confidence": 0.9, "reasoning": "because"} for e in entities],
"search_summary": "looked it up",
}
return "```json\n" + json.dumps(obj) + "\n```"
def test_generation_deterministic_and_diverse():
texts = set()
for seed in range(1, 80):
truth, text = generate(seed)
assert generate(seed) == (truth, text) # deterministic
assert lookup_truth(f"comm-{seed}") == truth # truth re-derivable from id
texts.add(text)
assert len(texts) > 75 # near-unique renderings
def test_gold_is_valid_by_construction():
"""Every gold entity must actually exist in the world, and intent labels are valid."""
for seed in range(1, 200):
truth, _ = generate(seed)
cat = truth["classification"]["category"]
assert truth["classification"]["subcategory"] in TAXONOMY[cat]
for e in [truth["primary"], *truth["required"], *truth["optional"]]:
if e["friendly_id"]:
assert e["friendly_id"].upper() in WORLD_FRIENDLY_IDS, (seed, e)
assert e["entity_id"] in WORLD_UUIDS or e["entity_id"].upper() in WORLD_FRIENDLY_IDS
def test_search_entities_exact_and_fuzzy_and_errors():
truth, _ = generate(4)
onum = next(e["friendly_id"] for e in truth["required"] if e["entity_type"] == "Order")
r = json.loads(execute_tool("search_entities",
{"search_terms": onum, "reason": "x", "entity_types": ["Order"]}))
assert r["results"][0]["fields"]["order_number"] == onum
assert r["results"][0]["score"] >= 100 # exact-match boost
short = json.loads(execute_tool("search_entities", {"search_terms": "ab", "reason": "x"}))
assert "INVALID_ARGUMENT" in short["error"]
miss = json.loads(execute_tool("search_entities",
{"search_terms": "zzqq nonexistent thing", "reason": "x"}))
assert "NOT_FOUND" in miss["error"]
# phone is deliberately NOT searchable -> must use SQL
phone_search = json.loads(execute_tool("search_entities",
{"search_terms": "+12029729975", "reason": "x"}))
assert "error" in phone_search
def test_execute_sql_chain_and_guardrails():
rows = json.loads(execute_tool(
"execute_sql", {"sql": "SELECT customer_id, full_name FROM shop.customers LIMIT 1", "reason": "x"}))
assert rows["row_count"] == 1 and "customer_id" in rows["columns"]
# canonical phone -> orders join runs
cid = rows["rows"][0]["customer_id"]
joined = json.loads(execute_tool("execute_sql", {"sql": (
"SELECT o.order_number, p.product_name FROM shop.orders o "
"JOIN shop.order_items oi ON oi.order_id=o.order_id "
"JOIN shop.products p ON p.sku=oi.sku WHERE o.customer_id=?".replace("?", f"'{cid}'")),
"reason": "x"}))
assert "order_number" in joined["columns"]
assert "error" in json.loads(execute_tool("execute_sql", {"sql": "DELETE FROM shop.orders", "reason": "x"}))
assert "error" in json.loads(execute_tool("execute_sql",
{"sql": "SELECT 1; DROP TABLE shop.orders", "reason": "x"}))
assert "error" in json.loads(execute_tool("execute_sql",
{"sql": "SELECT missing_col FROM shop.orders", "reason": "x"}))
def test_auto_limit_applied():
res = json.loads(execute_tool("execute_sql", {"sql": "SELECT * FROM shop.order_items", "reason": "x"}))
assert res["row_count"] <= 50 # auto LIMIT 50 even though the table is larger
def test_perfect_answer_scores_full_marks():
for seed in (4, 11, 23, 42):
truth, _ = generate(seed)
gold = truth["required"] + truth["optional"]
s = score(truth, parse_output(_answer(truth, gold)))
assert s["intent_exact"] == 1.0
assert s["primary_found"] == 1.0
assert s["entity_id_correct"] == 1.0
assert s["both_correct"] == 1.0
assert s["no_hallucinated_id"] == 1.0
assert s["recall"] == 1.0
def test_wrong_answer_scores_below_correct():
truth, _ = generate(4)
gold = truth["required"] + truth["optional"]
good = score(truth, parse_output(_answer(truth, gold)))
# wrong intent + a real-but-irrelevant order, no uuid
other = next(o for o in oc.ORDERS
if o["order_number"] != truth["required"][-1]["friendly_id"])
bad = score(truth, parse_output(_answer(
truth,
[{"entity_type": "Order", "friendly_id": other["order_number"], "entity_id": None}],
classification={"category": "Account", "subcategory": "login_issue"})))
assert bad["both_correct"] < good["both_correct"]
assert bad["primary_found"] == 0.0
assert bad["intent_exact"] == 0.0
assert bad["f1"] < good["f1"]
assert bad["no_hallucinated_id"] == 1.0 # the distractor is a real order, not hallucinated
def test_hallucinated_id_penalized():
truth, _ = generate(4)
s = score(truth, parse_output(_answer(
truth, [{"entity_type": "Order", "friendly_id": "ORD-999999-0000", "entity_id": "not-a-real-uuid"}])))
assert s["no_hallucinated_id"] == 0.0
def test_parse_output_tolerates_garbage():
s = score(*(generate(4)[0],) * 0 or (generate(4)[0], parse_output("no json here, sorry")))
assert s["intent_format_ok"] == 0.0
assert s["tag_format_ok"] == 0.0
assert s["both_correct"] == 0.0
def test_parse_output_picks_last_json_block():
text = "draft:\n```json\n{\"classification\": {\"category\": \"X\"}}\n```\n" \
"final:\n```json\n{\"classification\": {\"category\": \"Shipping\", " \
"\"subcategory\": \"late_delivery\"}, \"entities\": [], \"search_summary\": \"s\"}\n```"
parsed = parse_output(text)
assert parsed["category"] == "Shipping" and parsed["subcategory"] == "late_delivery"
def test_task_registered_with_tools():
from app.tasks import REGISTRY
assert REGISTRY.get("order-comms") is TASK
assert TASK.tools and TASK.execute_tool is not None
assert TASK.ui["output"] == "comms"
|