intermediateView Python source

Route support tickets with a model in SQL (Jev)

Sail has SQL functions that call TypeSafe's Jev models; Apache Spark has none of them. This example routes 2,000 support tickets to one of three teams with jev_choice, calling the model 24 times, once per distinct ticket text.

Run it

Run all commands below from the repository root (sail-examples/):

uv run ai/jev-sql-routing/main.py

By default the example runs against the local TypeSafe stand-in that ships in pysail for Sail's own tests (pysail.testing.jev.JevMock). It needs no account, and nothing leaves the machine. The stand-in picks a team with a keyword rule.

To call the real service, set JEV_LIVE=1 and TYPESAFE_API_KEY in the environment of the process that runs Sail. Ticket texts are then sent to TypeSafe, and the answers come from the Jev model.

Output

The newest capture appears first. Older output stays here when a result changes.

Current output · Sail 0.7.3

Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.

+-------+-------+
|   team|tickets|
+-------+-------+
|billing|    667|
|    bug|    667|
| how-to|    666|
+-------+-------+

+-------+--------------+--------+
|tickets|distinct_texts|requests|
+-------+--------------+--------+
|   2000|            24|      24|
+-------+--------------+--------+

Python code

main.py175 lines

View main.py on GitHub

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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
# ]
# ///
"""Route support tickets with a model from plain SQL, using Sail's built-in Jev functions.

`jev_choice` asks a TypeSafe Jev model to pick one of several labels for each row. The query
runs on distinct texts only, saves the answers, and joins them back. By default the example
runs against the local TypeSafe stand-in that ships in pysail for Sail's own tests, so it needs
no account and sends nothing off the machine. Set JEV_LIVE=1 and TYPESAFE_API_KEY to call the
real service instead (row values are then sent to TypeSafe).
"""

import atexit
import contextlib
import os
import shutil
import tempfile

import pyarrow as pa
from pyspark.sql import DataFrame, SparkSession

TEMPLATES = [
    "I was charged twice for {p}, please refund me",
    "{p} crashes with an error when I open a report",
    "How do I export data from {p}?",
    "The {p} invoice amount is wrong",
    "{p} shows error 500 after the update",
    "How to invite a teammate to {p}",
]
PRODUCTS = ["Starter plan", "Team plan", "Business plan", "Enterprise plan"]
ROWS = 2_000
TEAMS = {
    "billing": "Payments, charges, refunds and invoices",
    "bug": "Crashes, errors and broken features",
    "how-to": "Questions about how to use the product",
}


def sql_string(value: str) -> str:
    """A SQL string literal, with single quotes escaped."""
    return "'" + value.replace("'", "''") + "'"


def ticket_text(i: int) -> str:
    return TEMPLATES[i % len(TEMPLATES)].replace("{p}", PRODUCTS[(i // len(TEMPLATES)) % len(PRODUCTS)])


def stand_in_route(text: str) -> str:
    """The stand-in model's rule. The real Jev model reads the descriptions in TEAMS instead."""
    lowered = text.lower()
    if "refund" in lowered or "charged" in lowered or "invoice" in lowered:
        return "billing"
    if "crash" in lowered or "error" in lowered:
        return "bug"
    return "how-to"


def answer_with_rule(response: dict, body: dict) -> dict:
    """Shape the stand-in's answer: choose a team by rule, with a fixed confidence."""
    route = stand_in_route(str(body["state"]))
    for answer in response["answers"].values():
        if answer["type"] == "choice":
            names = list(answer["probabilities"])
            answer["choice"] = route
            answer["confidence"] = 0.9
            answer["probabilities"] = {n: 0.9 if n == route else 0.1 / (len(names) - 1) for n in names}
    return response


def temp_dir(prefix: str) -> str:
    """A temporary directory that is removed when the process exits."""
    path = tempfile.mkdtemp(prefix=prefix)
    atexit.register(shutil.rmtree, path, ignore_errors=True)
    return path


@contextlib.contextmanager
def jev_service():
    """Yield a request counter: the local stand-in by default, the real service with JEV_LIVE=1."""
    if os.environ.get("JEV_LIVE") == "1":
        yield None
        return
    # JevMock is the TypeSafe stand-in that pysail ships for Sail's own tests, not a stable API.
    from pysail.testing.jev import JevMock

    saved = {k: os.environ.get(k) for k in ("TYPESAFE_BASE_URL", "TYPESAFE_API_KEY")}
    with JevMock() as mock:
        mock.transform = answer_with_rule
        # Set before the first Jev query, so the in-process Sail server sends requests to the stand-in.
        os.environ["TYPESAFE_BASE_URL"] = mock.url
        os.environ["TYPESAFE_API_KEY"] = "local-stand-in"
        try:
            yield mock
        finally:
            for key, value in saved.items():
                if value is None:
                    os.environ.pop(key, None)
                else:
                    os.environ[key] = value


def run(spark: SparkSession) -> dict[str, DataFrame]:
    spark.createDataFrame(
        [(i, ticket_text(i)) for i in range(ROWS)], "ticket_id long, text string"
    ).createOrReplaceTempView("tickets")
    criteria = ", ".join(
        f"{sql_string(name)}, {sql_string(description)}" for name, description in TEAMS.items()
    )

    with jev_service() as service:
        answers = spark.sql(
            f"""
            SELECT text, answer.choice AS team, answer.confidence AS confidence
            FROM (
              SELECT text, jev_choice(text, 'Which team should handle this support ticket?', map({criteria})) AS answer
              FROM (SELECT DISTINCT text FROM tickets)
            )
            """
        )
        # Save the answers once: every later query reads them instead of calling the model again.
        answers_path = temp_dir("routes-")
        answers.write.mode("overwrite").parquet(answers_path)
        requests = service.request_count if service is not None else None

    spark.read.parquet(answers_path).createOrReplaceTempView("routes")
    summary = spark.sql(
        "SELECT team, count(*) AS tickets FROM tickets JOIN routes USING (text) GROUP BY team ORDER BY team"
    )
    stats = spark.createDataFrame(
        [(ROWS, spark.table("routes").count(), requests)], "tickets long, distinct_texts long, requests long"
    )
    return {"routes": spark.sql("SELECT * FROM routes ORDER BY text"), "summary": summary, "stats": stats}


def reference() -> dict[str, pa.Table]:
    """The expected results, computed without Sail from the same data and the stand-in's rule."""
    texts = sorted({ticket_text(i) for i in range(ROWS)})
    routes = [{"text": t, "team": stand_in_route(t), "confidence": 0.9} for t in texts]
    counts: dict[str, int] = {}
    for i in range(ROWS):
        team = stand_in_route(ticket_text(i))
        counts[team] = counts.get(team, 0) + 1
    return {
        "routes": pa.Table.from_pylist(
            routes, pa.schema([("text", pa.string()), ("team", pa.string()), ("confidence", pa.float64())])
        ),
        "summary": pa.Table.from_pylist(
            [{"team": k, "tickets": v} for k, v in sorted(counts.items())],
            pa.schema([("team", pa.string()), ("tickets", pa.int64())]),
        ),
        "stats": pa.Table.from_pylist(
            [{"tickets": ROWS, "distinct_texts": len(texts), "requests": len(texts)}],
            pa.schema([("tickets", pa.int64()), ("distinct_texts", pa.int64()), ("requests", pa.int64())]),
        ),
    }


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    results = run(spark)
    results["summary"].show()
    results["stats"].show()
    spark.stop()
    server.stop()

How it works

• One SQL query asks the model. jev_choice(state, instructions, criteria) returns a struct with the chosen choice, its confidence and per-choice probabilities. Here, criteria is a map from each team name to a description of the team.
• Distinct first. The query runs jev_choice over SELECT DISTINCT text, so the model answers 24 questions, not 2,000. Without DISTINCT, Sail asks one question per row: adjacent rows with the same text share a request, but each row is still its own question, and repeats that are not adjacent are sent again.
• Save, then join. The answers are written to Parquet once, and the summary joins them back onto every ticket.

On Sail

• Jev functions need TYPESAFE_API_KEY, and read the service address from TYPESAFE_BASE_URL, in the environment of the Sail server. On a cluster, set them on every worker that runs Jev calls (per the Sail Jev docs).
• With an in-process server, as here, setting the variables before the first Jev query works. For sail spark server, set them before you start the server.
• cache() does not keep results on Sail, so the answers are saved to Parquet: in one run, each distinct text is asked once.

Related Sail behavior

• Call Jev models from SQL
• Deduplicate inputs before calling Jev
• Caching does not save the result

Keep exploring

AI data preparationEnrich a table with a model →Run a model once per distinct value, then join its answers back. AnalyticsTurn spoken requests into intent analytics →Turn spoken support requests into a report of customer needs.