Spaces:
Running
Running
File size: 8,596 Bytes
f4dcb85 | 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 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 | """TariffWise AI layer. The model does the language work.
The model has three jobs. It extracts facts from the user's description.
It phrases the engine's named gap as a plain question. It reads each
answer back into schedule terms. The engine makes every decision. No
model output assigns a code.
Provenance is recorded per fact. A fact is user_stated when extracted
from the user's own words, and answered when it came from an interview
reply. The audit record keeps both.
"""
import json
import os
from engine import walk, rationale, FACT_KEYS
MODEL_ID = "llama-3.3-70b-versatile"
BOOL_KEYS = {k for k, v in FACT_KEYS.items() if v.startswith("true")}
ENUM_KEYS = {
"sole_material": ["rubber_plastics", "leather", "composition_leather", "other"],
"upper_material": ["rubber_plastics", "leather", "composition_leather", "textile", "other"],
"upper_textile_kind": ["vegetable_fibers", "wool_felt", "other_textile"],
"gender_age": ["men", "women", "youths_boys", "misses", "children", "infants", "unisex"],
}
EXTRACT_SYSTEM = (
"You extract footwear facts for US tariff classification under HTS Chapter 64. "
"Apply the chapter notes. The upper material is the constituent material with the "
"greatest external surface area, disregarding accessories such as laces, eyelet stays, "
"and ornamentation. Textile with a visible external layer of rubber or plastics counts "
"as rubber_plastics. Return only a JSON object. Include a key only when the text "
"clearly states or strongly implies it. Omit anything not determinable.\n"
"Keys and values:\n"
+ "\n".join(
f" {k}: {'true or false, ' + v[5:] if k in BOOL_KEYS else ('one of ' + ', '.join(ENUM_KEYS[k]) if k in ENUM_KEYS else 'number, ' + v)}"
for k, v in FACT_KEYS.items()
)
)
def get_client():
from groq import Groq
key = os.environ.get("GROQ_API_KEY", "").strip()
if not key:
raise RuntimeError("GROQ_API_KEY is not set")
return Groq(api_key=key)
def _chat(client, system, user, json_mode=True, max_tokens=400):
kwargs = dict(
model=MODEL_ID,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
temperature=0,
max_tokens=max_tokens,
)
if json_mode:
kwargs["response_format"] = {"type": "json_object"}
out = client.chat.completions.create(**kwargs)
return out.choices[0].message.content
def normalize(raw):
"""Keep only known keys with valid values."""
clean = {}
for k, v in raw.items():
if k in BOOL_KEYS:
if isinstance(v, bool):
clean[k] = v
elif str(v).lower() in ("true", "yes"):
clean[k] = True
elif str(v).lower() in ("false", "no"):
clean[k] = False
elif k in ENUM_KEYS:
if str(v).lower() in ENUM_KEYS[k]:
clean[k] = str(v).lower()
elif k == "value_per_pair":
try:
clean[k] = float(v)
except (TypeError, ValueError):
pass
return clean
def extract_facts(client, text):
raw = _chat(client, EXTRACT_SYSTEM, text)
try:
return normalize(json.loads(raw))
except json.JSONDecodeError:
return {}
def ask_gap(client, fact_key):
"""Phrase the engine's named gap as one plain question."""
system = ("You interview a small importer about one footwear product. "
"Ask one short plain question that determines the stated fact. "
"One sentence. No preamble.")
user = f"The fact to determine: {FACT_KEYS[fact_key]}."
return _chat(client, system, user, json_mode=False, max_tokens=60).strip()
def ask_branch(client, branch_text):
"""Phrase an unparsed schedule branch as a yes or no question."""
system = ("You interview a small importer about one footwear product. "
"Restate the tariff condition below as one plain yes or no question "
"about their product. One sentence. No preamble.")
return _chat(client, system, f"Condition: {branch_text}", json_mode=False, max_tokens=80).strip()
def read_answer(client, fact_key, answer):
"""Read an interview answer back into the one fact it addresses."""
user = f"The question was about: {FACT_KEYS[fact_key]}.\nThe importer answered: {answer}"
raw = _chat(client, EXTRACT_SYSTEM, user)
try:
got = normalize(json.loads(raw))
except json.JSONDecodeError:
return {}
return {fact_key: got[fact_key]} if fact_key in got else {}
def read_branch_answer(client, branch_text, answer):
"""Read a yes or no reply to a branch condition. Returns True, False, or None."""
system = ("Decide whether the importer's answer means the condition holds. "
'Return only JSON like {"holds": true} or {"holds": false}. '
'Return {"holds": null} when the answer does not settle it.')
raw = _chat(client, system, f"Condition: {branch_text}\nAnswer: {answer}")
try:
v = json.loads(raw).get("holds")
return v if isinstance(v, bool) else None
except json.JSONDecodeError:
return None
# ---------------------------------------------------------------------
# The interview loop. ask_user is any callable that shows a question and
# returns the reply, which keeps the loop usable in a notebook or a web app.
# ---------------------------------------------------------------------
MAX_TURNS = 12
def classify_interactive(client, description, ask_user):
facts = extract_facts(client, description)
provenance = {k: "user_stated" for k in facts}
transcript = []
for _ in range(MAX_TURNS):
r = walk(facts)
if r["status"] == "code":
return {"status": "code", "code": r["code"], "rate": r["general"],
"facts": facts, "provenance": provenance,
"transcript": transcript, "path": r["path"],
"rationale": rationale(r, facts)}
if r["status"] == "need_fact":
q = ask_gap(client, r["fact"])
a = ask_user(q)
transcript.append({"q": q, "a": a, "about": r["fact"]})
got = read_answer(client, r["fact"], a)
if got:
facts.update(got)
provenance[r["fact"]] = "answered"
continue
q2 = f"{q} Please answer directly."
a2 = ask_user(q2)
transcript.append({"q": q2, "a": a2, "about": r["fact"]})
got = read_answer(client, r["fact"], a2)
if got:
facts.update(got)
provenance[r["fact"]] = "answered"
continue
return {"status": "review", "reason": f"the fact {r['fact']} could not be established",
"facts": facts, "provenance": provenance, "transcript": transcript}
if r["status"] == "review" and "at" in r:
q = ask_branch(client, r["at"])
a = ask_user(q)
transcript.append({"q": q, "a": a, "about": r["at"][:60]})
holds = read_branch_answer(client, r["at"], a)
if holds is None:
return {"status": "review",
"reason": "a schedule condition needs expert judgment",
"at": r["at"], "facts": facts,
"provenance": provenance, "transcript": transcript}
facts.setdefault("_overrides", {})[r["at"]] = holds
provenance[f"branch: {r['at'][:50]}"] = "answered"
continue
return {"status": "review", "reason": r.get("reason", "unresolved"),
"facts": facts, "provenance": provenance, "transcript": transcript}
return {"status": "review", "reason": "the interview exceeded its turn limit",
"facts": facts, "provenance": provenance, "transcript": transcript}
def audit_record(description, result):
rec = {
"input": description,
"status": result["status"],
"code": result.get("code"),
"rate": result.get("rate"),
"facts": {k: v for k, v in result["facts"].items() if k != "_overrides"},
"provenance": result["provenance"],
"interview": result["transcript"],
"path": [f"{s['htsno'] or 'grouping'} {s['desc']}" for s in result.get("path", [])],
"basis": "HTS Chapter 64, 2026 revision, walked per the chapter notes",
"disclaimer": "Decision support only. The importer or a licensed broker makes the final decision.",
}
return rec
|