IDKHowToCodeFr commited on
Commit
86a56f9
·
1 Parent(s): c86035c

fix(core): resolve caching, lazy load, and thread leak defects

Browse files

- Prevent generic fallback response from caching and poisoning state.
- Read LOCAL_7B_P95 from environment variables instead of a magic hardcoded number.
- Refactor local tier to stream generation and abort executor thread if deadline is exceeded.
- Remove over-broad words 'JSON' and 'system' from default taboo list.
- Wrap local model loading in a lazy-load function to prevent HF space boot failure.

Files changed (2) hide show
  1. scripts/composer.py +4 -3
  2. scripts/llm.py +46 -21
scripts/composer.py CHANGED
@@ -64,7 +64,7 @@ async def compose_with_fallback(prompt: str, full_system: str, local_system: str
64
  return await func(p, system_to_use, timeout)
65
  else:
66
  try:
67
- return await asyncio.wait_for(func(p, system_to_use), timeout=deadline - time.time())
68
  except asyncio.TimeoutError:
69
  return None
70
 
@@ -140,7 +140,7 @@ async def compose(category: dict, merchant: dict, trigger: dict, customer: Optio
140
  # 2. Dynamic Tone & Taboos
141
  voice = category.get("voice", {})
142
  tone = voice.get("tone", "professional peer")
143
- taboos_list = voice.get("taboos", ["internal metrics", "JSON", "system", "guaranteed"])
144
  taboos = ", ".join(taboos_list)
145
 
146
  prefix = "Dr. " if category.get('slug') == 'dentists' else ""
@@ -226,5 +226,6 @@ Example Output format for a good message:
226
 
227
  # Add sender attribution required by the challenge
228
  res["send_as"] = "merchant_on_behalf" if customer else "vera"
229
- _cache[key] = res
 
230
  return res
 
64
  return await func(p, system_to_use, timeout)
65
  else:
66
  try:
67
+ return await asyncio.wait_for(func(p, system_to_use, deadline), timeout=deadline - time.time())
68
  except asyncio.TimeoutError:
69
  return None
70
 
 
140
  # 2. Dynamic Tone & Taboos
141
  voice = category.get("voice", {})
142
  tone = voice.get("tone", "professional peer")
143
+ taboos_list = voice.get("taboos", ["guaranteed", "100%", "click here", "promise"])
144
  taboos = ", ".join(taboos_list)
145
 
146
  prefix = "Dr. " if category.get('slug') == 'dentists' else ""
 
226
 
227
  # Add sender attribution required by the challenge
228
  res["send_as"] = "merchant_on_behalf" if customer else "vera"
229
+ if res.get("rationale") != "fallback_default":
230
+ _cache[key] = res
231
  return res
scripts/llm.py CHANGED
@@ -13,7 +13,8 @@ load_dotenv()
13
  logger = logging.getLogger("VeraArchitect")
14
 
15
  LOCAL_MAX_TOKENS = 250
16
- local_model = None
 
17
 
18
  try:
19
  _schema = {"type":"object","properties":{"body":{"type":"string"},"cta":{"type":"string"},"rationale":{"type":"string"}},"required":["body","cta","rationale"]}
@@ -21,23 +22,31 @@ try:
21
  except Exception:
22
  JSON_GRAMMAR = None
23
 
24
- try:
25
- LOCAL_7B_P95 = 37.47
26
- if LOCAL_7B_P95 <= 14.0:
27
- local_repo = "bartowski/Qwen2.5-7B-Instruct-GGUF"
28
- local_file = "Qwen2.5-7B-Instruct-Q4_K_M.gguf"
29
- LOCAL_MAX_TOKENS = 150
30
- logger.info(f"Using 7b for local tier since p95 latency ({LOCAL_7B_P95}s) is within 14s budget.")
31
- else:
32
- local_repo = "Qwen/Qwen2.5-3B-Instruct-GGUF"
33
- local_file = "qwen2.5-3b-instruct-q4_k_m.gguf"
34
- LOCAL_MAX_TOKENS = 250
35
- logger.info(f"7b p95 {LOCAL_7B_P95}s exceeds 14s budget, using 3b instead.")
 
 
 
 
 
 
 
 
 
 
 
36
 
37
- local_model = Llama.from_pretrained(repo_id=local_repo, filename=local_file, n_ctx=2048, n_threads=os.cpu_count(), verbose=False)
38
- except Exception as e:
39
- logger.warning(f"Failed to load local model: {e}")
40
- local_model = None
41
 
42
  def clean_json(text: str) -> Optional[dict]:
43
  try:
@@ -88,21 +97,37 @@ async def call_gemini(prompt: str, system: str, timeout: float) -> Optional[dict
88
  res.raise_for_status()
89
  return clean_json(res.json()["candidates"][0]["content"]["parts"][0]["text"])
90
 
91
- async def call_local(prompt: str, system: str) -> Optional[dict]:
92
- if not local_model: return None
 
 
93
  def run_inference():
94
  kwargs = {
95
  "messages": [{"role": "system", "content": system + "\n\nRETURN JSON ONLY."}, {"role": "user", "content": prompt}],
96
  "temperature": 0.2,
97
  "max_tokens": LOCAL_MAX_TOKENS,
 
98
  }
99
  if JSON_GRAMMAR:
100
  kwargs["grammar"] = JSON_GRAMMAR
101
  else:
102
  kwargs["response_format"] = {"type": "json_object"}
103
 
104
- res = local_model.create_chat_completion(**kwargs)
105
- return res["choices"][0]["message"]["content"]
 
 
 
 
 
 
 
 
 
 
 
 
106
  loop = asyncio.get_event_loop()
107
  text = await loop.run_in_executor(None, run_inference)
 
108
  return clean_json(text)
 
13
  logger = logging.getLogger("VeraArchitect")
14
 
15
  LOCAL_MAX_TOKENS = 250
16
+ _local_model_instance = None
17
+ _local_model_loaded = False
18
 
19
  try:
20
  _schema = {"type":"object","properties":{"body":{"type":"string"},"cta":{"type":"string"},"rationale":{"type":"string"}},"required":["body","cta","rationale"]}
 
22
  except Exception:
23
  JSON_GRAMMAR = None
24
 
25
+ def get_local_model():
26
+ global _local_model_instance, _local_model_loaded, LOCAL_MAX_TOKENS
27
+ if _local_model_loaded:
28
+ return _local_model_instance
29
+
30
+ _local_model_loaded = True
31
+ try:
32
+ LOCAL_7B_P95 = float(os.getenv("LOCAL_7B_P95", "999.0"))
33
+ if LOCAL_7B_P95 <= 14.0:
34
+ local_repo = "bartowski/Qwen2.5-7B-Instruct-GGUF"
35
+ local_file = "Qwen2.5-7B-Instruct-Q4_K_M.gguf"
36
+ LOCAL_MAX_TOKENS = 150
37
+ logger.info(f"Using 7b for local tier since p95 latency ({LOCAL_7B_P95}s) is within 14s budget.")
38
+ else:
39
+ local_repo = "Qwen/Qwen2.5-3B-Instruct-GGUF"
40
+ local_file = "qwen2.5-3b-instruct-q4_k_m.gguf"
41
+ LOCAL_MAX_TOKENS = 250
42
+ logger.info(f"7b p95 {LOCAL_7B_P95}s exceeds 14s budget, using 3b instead.")
43
+
44
+ _local_model_instance = Llama.from_pretrained(repo_id=local_repo, filename=local_file, n_ctx=2048, n_threads=os.cpu_count(), verbose=False)
45
+ except Exception as e:
46
+ logger.warning(f"Failed to load local model: {e}")
47
+ _local_model_instance = None
48
 
49
+ return _local_model_instance
 
 
 
50
 
51
  def clean_json(text: str) -> Optional[dict]:
52
  try:
 
97
  res.raise_for_status()
98
  return clean_json(res.json()["candidates"][0]["content"]["parts"][0]["text"])
99
 
100
+ async def call_local(prompt: str, system: str, deadline: float = None) -> Optional[dict]:
101
+ model = get_local_model()
102
+ if not model: return None
103
+
104
  def run_inference():
105
  kwargs = {
106
  "messages": [{"role": "system", "content": system + "\n\nRETURN JSON ONLY."}, {"role": "user", "content": prompt}],
107
  "temperature": 0.2,
108
  "max_tokens": LOCAL_MAX_TOKENS,
109
+ "stream": True,
110
  }
111
  if JSON_GRAMMAR:
112
  kwargs["grammar"] = JSON_GRAMMAR
113
  else:
114
  kwargs["response_format"] = {"type": "json_object"}
115
 
116
+ res_text = ""
117
+ try:
118
+ for chunk in model.create_chat_completion(**kwargs):
119
+ if deadline and time.time() > deadline:
120
+ logger.warning("Local tier exceeded deadline mid-generation. Aborting thread.")
121
+ return None
122
+ delta = chunk["choices"][0].get("delta", {})
123
+ if "content" in delta and delta["content"]:
124
+ res_text += delta["content"]
125
+ except Exception as e:
126
+ logger.warning(f"Local stream aborted: {e}")
127
+ return None
128
+ return res_text
129
+
130
  loop = asyncio.get_event_loop()
131
  text = await loop.run_in_executor(None, run_inference)
132
+ if not text: return None
133
  return clean_json(text)