forked from odysseus-dev/odysseus
-
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathanswer_checks.py
More file actions
583 lines (508 loc) · 26.4 KB
/
Copy pathanswer_checks.py
File metadata and controls
583 lines (508 loc) · 26.4 KB
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
"""Deterministic checks on a final answer before it reaches the user (weekdays,
visible working, suggested slots that the listed calendar says are busy).
Two things a local model gets wrong in plain sight, seen live on a three-line
quiz ("si hoy es jueves 25 de septiembre de 2026, ¿qué día será el 25 de
diciembre?"):
* it names the wrong weekday for a full date ("el 25 de diciembre de 2026
cae en domingo" — it is a Friday), which the calendar settles exactly;
* it thinks aloud in the reply ("…espera, recalculo… no: … Déjame ser
riguroso…"), leaving its working and self-corrections as the answer.
Both are found here without a model; `agent_loop` asks for one rewrite of
the answer when either shows up. Only full dates (day, month and year) are
checked, and only when the weekday is written right next to the date, so a
sentence that merely mentions both is not taken for a claim.
"""
from __future__ import annotations
import datetime as _dt
import re
from decimal import Decimal
from typing import Dict, List, Optional
_ES_MONTHS = {
"enero": 1, "febrero": 2, "marzo": 3, "abril": 4, "mayo": 5, "junio": 6, "julio": 7,
"agosto": 8, "septiembre": 9, "setiembre": 9, "octubre": 10, "noviembre": 11, "diciembre": 12,
}
_EN_MONTHS = {
"january": 1, "february": 2, "march": 3, "april": 4, "may": 5, "june": 6, "july": 7,
"august": 8, "september": 9, "october": 10, "november": 11, "december": 12,
}
_ES_MONTH_ABBR = {
"ene": 1, "feb": 2, "mar": 3, "abr": 4, "may": 5, "jun": 6, "jul": 7, "ago": 8,
"sep": 9, "sept": 9, "set": 9, "oct": 10, "nov": 11, "dic": 12,
}
_ES_DAYS = ["lunes", "martes", "miércoles", "jueves", "viernes", "sábado", "domingo"]
_EN_DAYS = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
_ES_DAY_RE = r"(lunes|martes|mi[eé]rcoles|jueves|viernes|s[aá]bado|domingo)"
_EN_DAY_RE = r"(monday|tuesday|wednesday|thursday|friday|saturday|sunday)"
_ES_MONTH_RE = "(" + "|".join(_ES_MONTHS) + ")"
_EN_MONTH_RE = "(" + "|".join(_EN_MONTHS) + ")"
# An aside in brackets between a date and its weekday: "(Todos los Santos)".
_ASIDE = r"(?:\s*\((?!\s*(?:lunes|martes|mi[eé]rcoles|jueves|viernes|s[aá]bado|domingo)\s*\))[^()\n]{1,60}\))?"
# "25 de diciembre de 2026 cae en domingo", "el 25 de diciembre de 2026 será domingo",
# "25 de diciembre de 2026 (domingo)"
_ES_DATE_THEN_DAY = re.compile(
r"\b(\d{1,2})\s+de\s+" + _ES_MONTH_RE + r"\s+de(?:l)?\s+(\d{4})\**" + _ASIDE + r"\s*"
r"(?:\(|,|:|—|-|\bes\b|\bser[aá]\b|\bfue\b|\bcae(?:r[aá])?\s+en\b|\bca[yí]o\s+en\b|\bera\b)\s*(?:un\s+|en\s+)?\**"
+ _ES_DAY_RE + r"\b",
re.IGNORECASE,
)
# "domingo 25 de diciembre de 2026", "domingo, 25 de diciembre de 2026"
_ES_DAY_THEN_DATE = re.compile(
r"\b" + _ES_DAY_RE + r"\**\s*[,(]?\s*(?:el\s+)?(\d{1,2})\s+de\s+" + _ES_MONTH_RE + r"\s+de(?:l)?\s+(\d{4})\b",
re.IGNORECASE,
)
# "December 25, 2026 is a Sunday", "December 25, 2026 (Sunday)"
_EN_DATE_THEN_DAY = re.compile(
r"\b" + _EN_MONTH_RE + r"\s+(\d{1,2})(?:st|nd|rd|th)?,?\s+(\d{4})\**\s*"
r"(?:\(|,|:|—|-|\bis\b|\bwill\s+be\b|\bwas\b|\bfalls\s+on\b|\bfell\s+on\b)\s*(?:a\s+|on\s+)?\**"
+ _EN_DAY_RE + r"\b",
re.IGNORECASE,
)
# "Sunday, December 25, 2026", "Sunday 25 December 2026"
_EN_DAY_THEN_DATE = re.compile(
r"\b" + _EN_DAY_RE + r"\**\s*[,(]?\s*(?:" + _EN_MONTH_RE + r"\s+(\d{1,2})(?:st|nd|rd|th)?|(\d{1,2})(?:st|nd|rd|th)?\s+"
+ _EN_MONTH_RE + r"),?\s+(\d{4})\b",
re.IGNORECASE,
)
# A date written without its year, when the answer (or the question) has
# already said which year it is talking about: "el 2 de noviembre (Todos los
# Santos) cae en sábado", "el 6 de diciembre (Día de la Constitución) en
# viernes" (seen live, both wrong for 2026, a few lines after "12 de octubre
# de 2026"). An aside in brackets may sit between the date and the weekday.
_ES_NOYEAR_THEN_DAY = re.compile(
r"\b(\d{1,2})\s+de\s+" + _ES_MONTH_RE + r"(?!\s+de(?:l)?\s+\d{4})\**" + _ASIDE + r"\s*"
r"(?:\(|,|:|—|-|\bes\b|\bser[aá]\b|\bfue\b|\bcae(?:r[aá])?(?:\s+en)?\b|\bca[yí]o(?:\s+en)?\b|\bera\b|\ben\b)"
r"\s*(?:un\s+|en\s+)?\**" + _ES_DAY_RE + r"\b",
re.IGNORECASE,
)
_ES_DAY_THEN_NOYEAR = re.compile(
r"\b" + _ES_DAY_RE + r"\**\s*,?\s*(?:el\s+)?(\d{1,2})\s+(?:de\s+)?"
r"(" + "|".join(sorted(list(_ES_MONTHS) + list(_ES_MONTH_ABBR), key=len, reverse=True)) + r")\.?"
r"(?!\s+de(?:l)?\s+\d{4})(?![a-záéíóú])",
re.IGNORECASE,
)
_YEAR = re.compile(r"\b(20\d{2}|19\d{2})\b")
# The same without a year in English: "Monday, September 29", "Mon Sep 29"
# is not matched (weekday abbreviations are too common as words), "Monday
# 29 September", "December 25 is a Friday", "Dec 25 (Friday)".
_EN_MONTH_ABBR = {
"jan": 1, "feb": 2, "mar": 3, "apr": 4, "jun": 6, "jul": 7, "aug": 8,
"sep": 9, "sept": 9, "oct": 10, "nov": 11, "dec": 12,
}
_EN_ANY_MONTH_RE = "(" + "|".join(sorted(list(_EN_MONTHS) + list(_EN_MONTH_ABBR), key=len, reverse=True)) + r")\.?"
_EN_NO_YEAR_AFTER = r"(?!,?\s+\d{4})(?![a-z0-9])"
_EN_DAY_THEN_NOYEAR = re.compile(
r"\b" + _EN_DAY_RE + r"\**\s*,?\s*(?:the\s+)?(?:" + _EN_ANY_MONTH_RE + r"\s+(\d{1,2})(?:st|nd|rd|th)?"
r"|(\d{1,2})(?:st|nd|rd|th)?\s+(?:of\s+)?" + _EN_ANY_MONTH_RE + r")" + _EN_NO_YEAR_AFTER,
re.IGNORECASE,
)
_EN_NOYEAR_THEN_DAY = re.compile(
r"\b" + _EN_ANY_MONTH_RE + r"\s+(\d{1,2})(?:st|nd|rd|th)?(?!,?\s+\d{4})\**" + _ASIDE + r"\s*"
r"(?:\(|,|:|—|-|\bis\b|\bwill\s+be\b|\bwas\b|\bfalls\s+on\b|\bfell\s+on\b)\s*(?:a\s+|on\s+)?\**"
+ _EN_DAY_RE + r"\b",
re.IGNORECASE,
)
def _en_month_number(name: str) -> int:
key = name.lower().rstrip(".")
return _EN_MONTHS.get(key) or _EN_MONTH_ABBR[key]
def _year_before(text: str, pos: int, fallback: Optional[str]) -> Optional[str]:
"""The last year the text names before ``pos``, else ``fallback``."""
years = [m.group(1) for m in _YEAR.finditer(text, 0, pos)]
return years[-1] if years else fallback
def _near_year(month: int, day: int, today: Optional[_dt.date]) -> Optional[str]:
"""The year that puts day/month within half a year of today ("el lunes
29 sep" in a list of next week's events), or None when none does."""
if today is None:
return None
best = None
for year in (today.year - 1, today.year, today.year + 1):
try:
gap = abs((_dt.date(year, month, day) - today).days)
except ValueError:
continue
if gap <= 183 and (best is None or gap < best[0]):
best = (gap, year)
return str(best[1]) if best else None
def _month_number(name: str) -> int:
key = name.lower().rstrip(".")
return _ES_MONTHS.get(key) or _ES_MONTH_ABBR[key]
def _fold_day(name: str) -> str:
return name.lower().replace("é", "e").replace("á", "a")
def _check(out: List[Dict[str, str]], day: str, month: int, dnum: str, year: str, lang: str, span: str) -> None:
try:
date = _dt.date(int(year), month, int(dnum))
except (TypeError, ValueError):
return
names = _ES_DAYS if lang == "es" else _EN_DAYS
real = names[date.weekday()]
if _fold_day(real) != _fold_day(day):
item = {"date": date.isoformat(), "said": day.lower(), "real": real, "lang": lang, "text": span.strip()}
if item not in out:
out.append(item)
def weekday_mismatches(text: str, context: str = "",
today: Optional[_dt.date] = None) -> List[Dict[str, str]]:
"""Dates the text pairs with a weekday the calendar contradicts.
Full dates always; a date without its year only when a year can be read
from the text before it or, failing that, from ``context`` (the user's
question) — never guessed from today's date."""
body = str(text or "")
out: List[Dict[str, str]] = []
ctx_years = _YEAR.findall(str(context or ""))
ctx_year = ctx_years[-1] if len(set(ctx_years)) == 1 else None
for m in _ES_NOYEAR_THEN_DAY.finditer(body):
month = _ES_MONTHS[m.group(2).lower()]
year = _year_before(body, m.start(), ctx_year) or _near_year(month, int(m.group(1)), today)
if year:
_check(out, m.group(3), month, m.group(1), year, "es", m.group(0))
for m in _ES_DAY_THEN_NOYEAR.finditer(body):
month = _month_number(m.group(3))
year = _year_before(body, m.start(), ctx_year) or _near_year(month, int(m.group(2)), today)
if year:
_check(out, m.group(1), month, m.group(2), year, "es", m.group(0))
for m in _ES_DATE_THEN_DAY.finditer(body):
_check(out, m.group(4), _ES_MONTHS[m.group(2).lower()], m.group(1), m.group(3), "es", m.group(0))
for m in _ES_DAY_THEN_DATE.finditer(body):
_check(out, m.group(1), _ES_MONTHS[m.group(3).lower()], m.group(2), m.group(4), "es", m.group(0))
for m in _EN_DAY_THEN_NOYEAR.finditer(body):
month = _en_month_number(m.group(2) or m.group(5))
dnum = m.group(3) or m.group(4)
year = _year_before(body, m.start(), ctx_year) or _near_year(month, int(dnum), today)
if year:
_check(out, m.group(1), month, dnum, year, "en", m.group(0))
for m in _EN_NOYEAR_THEN_DAY.finditer(body):
month = _en_month_number(m.group(1))
year = _year_before(body, m.start(), ctx_year) or _near_year(month, int(m.group(2)), today)
if year:
_check(out, m.group(3), month, m.group(2), year, "en", m.group(0))
for m in _EN_DATE_THEN_DAY.finditer(body):
_check(out, m.group(4), _EN_MONTHS[m.group(1).lower()], m.group(2), m.group(3), "en", m.group(0))
for m in _EN_DAY_THEN_DATE.finditer(body):
month = m.group(2) or m.group(5)
dnum = m.group(3) or m.group(4)
_check(out, m.group(1), _EN_MONTHS[month.lower()], dnum, m.group(6), "en", m.group(0))
return out
# Self-corrections and working left in the visible answer.
_THINKING_ALOUD = re.compile(
r"(?:\b(?:espera|wait)\s*[,.:!…]"
r"|\bno,?\s+espera\b"
r"|\brecalcul(?:o|emos|ando)\b"
r"|\brevisemos\b"
r"|\bd[eé]jame\s+(?:ser\s+riguros[oa]|recalcular|calcularlo|revisarlo|volver\s+a\s+calcular|pensar)"
r"|\blet\s+me\s+(?:re-?check|recalculate|recount|redo|double[- ]check|think\s+again)"
r"|\bactually,?\s+no\b"
r"|(?:\.\.\.|…)\s*no\s*[:,]"
# "Martes 29 de octubre… o sea, 29 de septiembre" (seen live)
r"|(?:\.\.\.|…)\s*(?:o\s+sea|mejor\s+dicho|quiero\s+decir|perd[oó]n|I\s+mean|sorry)\b)",
re.IGNORECASE,
)
_CODE_BLOCK = re.compile(r"```.*?```", re.DOTALL)
def thinking_aloud(text: str) -> List[str]:
"""The phrases of visible working found outside code blocks."""
body = _CODE_BLOCK.sub(" ", str(text or ""))
found: List[str] = []
for m in _THINKING_ALOUD.finditer(body):
phrase = m.group(0).strip()
if phrase.lower() not in (f.lower() for f in found):
found.append(phrase)
return found
def rewrite_note(mismatches: List[Dict[str, str]], aloud: List[str],
slots: Optional[List[Dict[str, str]]] = None,
calculation: Optional[Dict[str, str]] = None,
grounding: Optional[str] = None) -> str:
"""The runtime's request for one clean rewrite of the answer."""
parts = ["[Harness check — automatic runtime message, not a new user request] "
"Your last message is not shown to the user yet. Write the complete answer again, "
"from the start, as the final answer only."]
for m in mismatches:
parts.append(
f"The calendar says {m['date']} is a {m['real']}, but your answer says {m['said']} "
f"(\"{m['text'][:80]}\"). Use the calendar's weekday and fix anything that depended on it."
)
if aloud:
quoted = ", ".join(f'"{a}"' for a in aloud[:4])
parts.append(
f"It also shows your working and self-corrections ({quoted}). Give only the results: "
"no working, no second thoughts. Check any arithmetic before stating it (use the python "
"tool if you have it)."
)
if slots:
parts.append(slot_note(slots))
if calculation:
parts.append("A deterministic calculation found: " + calculation["explanation"] +
" Include the correct requested result in the final answer.")
if grounding:
parts.append(grounding)
return " ".join(parts)
def grounding_review(answer: str, question: str = "", tool_outputs: Optional[List[str]] = None, *,
retry_used: bool = False, enabled: Optional[bool] = None,
lang: Optional[str] = None) -> Dict[str, object]:
"""The figure-grounding decision for a final answer (see src/grounding_ledger.py).
Returns {"action": "none" | "retry" | "mark", "note", "answer", "ledger", "trace"}. "retry"
means: pass `note` as `rewrite_note(..., grounding=note)`; "mark" means: replace the answer
with `answer` (unsupported figures struck through). Always "none" while the setting
`agent_answer_grounding_ledger` is off."""
from src import grounding_ledger
return grounding_ledger.grounding_review(
answer, question, list(tool_outputs or []), retry_used=retry_used, enabled=enabled, lang=lang)
_ASKS_WEEKDAY = re.compile(
r"(?:qu[eé]\s+d[ií]a\s+(?:de\s+la\s+semana\s+)?(?:es|ser[aá]|fue|cae|caer[aá]|cay[oó]|era)"
r"|what\s+day\s+(?:of\s+the\s+week\s+)?(?:is|was|will)|which\s+weekday)",
re.IGNORECASE,
)
_ES_FULL_DATE = re.compile(r"\b(\d{1,2})\s+de\s+" + _ES_MONTH_RE + r"\s+de(?:l)?\s+(\d{4})\b", re.IGNORECASE)
_EN_FULL_DATE = re.compile(
r"\b(?:" + _EN_MONTH_RE + r"\s+(\d{1,2})(?:st|nd|rd|th)?|(\d{1,2})(?:st|nd|rd|th)?\s+" + _EN_MONTH_RE
+ r"),?\s+(\d{4})\b", re.IGNORECASE)
def _full_dates(text: str) -> List[_dt.date]:
out: List[_dt.date] = []
for m in _ES_FULL_DATE.finditer(text):
try:
out.append(_dt.date(int(m.group(3)), _ES_MONTHS[m.group(2).lower()], int(m.group(1))))
except ValueError:
pass
for m in _EN_FULL_DATE.finditer(text):
month = (m.group(1) or m.group(4)).lower()
try:
out.append(_dt.date(int(m.group(5)), _EN_MONTHS[month], int(m.group(2) or m.group(3))))
except ValueError:
pass
return out
def asked_weekday_mismatch(question: str, answer: str) -> List[Dict[str, str]]:
"""The user asked which weekday ONE full date falls on and the answer
names weekdays but never the right one ("2. Domingo." for a Friday, seen
live: the date was only in the question, so `weekday_mismatches` had
nothing to pair)."""
q = str(question or "")
a = str(answer or "")
asked = _ASKS_WEEKDAY.search(q)
if not asked:
return []
dates = list(dict.fromkeys(_full_dates(q)))
if len(dates) > 1:
# "si hoy es jueves 25 de septiembre de 2026, ¿qué día será el 25 de
# diciembre de 2026?": the date asked about follows the question.
dates = list(dict.fromkeys(_full_dates(q[asked.start():])))
if len(dates) != 1:
return []
date = dates[0]
folded = _fold_day(a)
for names, lang in ((_ES_DAYS, "es"), (_EN_DAYS, "en")):
said = [n for n in names if re.search(r"\b" + _fold_day(n) + r"\b", folded)]
if not said:
continue
real = names[date.weekday()]
if _fold_day(real) in (_fold_day(n) for n in said):
return []
return [{"date": date.isoformat(), "said": said[0], "real": real, "lang": lang,
"text": f"{date.isoformat()} → {said[0]}"}]
return []
# ── A suggested free slot that is not free ──────────────────────────────────
# Live: after listing "2026-09-29T17:00 -> 18:00: Cita con el dentista", the
# answer suggested "el martes 29 de septiembre a las 17:00" as a free slot.
_EVENT_SPAN = re.compile(
r"(\d{4}-\d{2}-\d{2})T(\d{2}):(\d{2})(?::\d{2})?[^\n]{0,12}?->\s*(\d{4}-\d{2}-\d{2})T(\d{2}):(\d{2})(?::\d{2})?"
r":?\s*(?:\[([^\]\n]{1,120})\]|([^\n#(]{1,120}))?"
)
_SUGGESTS = re.compile(
r"\b(?:sugier\w*|sugerencia|propon\w*|propuesta|hueco\w*|libre\w*|podr[ií]as|qu[eé]\s+tal"
r"|suggest\w*|propos\w*|free|slot|how\s+about|you\s+could)\b",
re.IGNORECASE,
)
_SLOT = re.compile(
r"\b(\d{1,2})(?:\s+de\s+" + _ES_MONTH_RE + r"|\s+" + _EN_MONTH_RE + r")?\b[^\n.;]{0,40}?"
r"(?:\ba\s+las\s+|\bat\s+|\bdesde\s+las\s+|\bfrom\s+)\**(\d{1,2})(?:[:.h](\d{2}))?",
re.IGNORECASE,
)
def _events_from(outputs: List[str]) -> List[Dict[str, object]]:
events: List[Dict[str, object]] = []
for text in outputs:
for m in _EVENT_SPAN.finditer(str(text or "")):
try:
start = _dt.datetime.fromisoformat(f"{m.group(1)}T{m.group(2)}:{m.group(3)}")
end = _dt.datetime.fromisoformat(f"{m.group(4)}T{m.group(5)}:{m.group(6)}")
except ValueError:
continue
title = (m.group(7) or m.group(8) or "").strip()
events.append({"start": start, "end": end, "title": title})
return events
def slot_conflicts(answer: str, tool_outputs: List[str]) -> List[Dict[str, str]]:
"""Slots the answer suggests that overlap an event a calendar tool listed
in this turn. Only sentences that suggest (and do not name the event
itself) are read; the day number is matched against the listed events'
days, so no month or year has to be guessed."""
events = _events_from(tool_outputs)
if not events:
return []
out: List[Dict[str, str]] = []
for sentence in re.split(r"(?<=[.!?])\s+|\n+", str(answer or "")):
if not _SUGGESTS.search(sentence):
continue
low = sentence.lower()
if any(e["title"] and str(e["title"]).lower() in low for e in events):
continue
for m in _SLOT.finditer(sentence):
day, hour, minute = int(m.group(1)), int(m.group(4)), int(m.group(5) or 0)
if not (1 <= day <= 31 and 0 <= hour <= 23 and 0 <= minute <= 59):
continue
for e in events:
start, end = e["start"], e["end"]
if start.day != day:
continue
slot = start.replace(hour=hour, minute=minute)
if start <= slot < end:
item = {"slot": slot.strftime("%Y-%m-%d %H:%M"), "event": str(e["title"] or "an event"),
"from": start.strftime("%H:%M"), "to": end.strftime("%H:%M"),
"text": m.group(0).strip()}
if item not in out:
out.append(item)
return out
def slot_note(conflicts: List[Dict[str, str]]) -> str:
parts = [f"You suggest {c['slot']} as a free slot, but the calendar you listed has "
f"\"{c['event']}\" from {c['from']} to {c['to']} that day." for c in conflicts]
return " ".join(parts) + " Suggest only times that are free in the listed calendar."
# Quantities for a number of people ("lista de la compra para 8", "receta para
# 6 personas", "for 4 servings"). Seen live (25-09): a shopping list for 8 with
# "6 muslos (2,5-3 kg)" — too few pieces, and a weight that does not match the
# count. Nothing here can check a recipe the model recalls from memory, so the
# runtime asks for the arithmetic up front instead of rewriting afterwards.
_SERVINGS = re.compile(
r"\bpara\s+(\d{1,3})\s+(?:personas?|comensales|raciones|invitados|adultos|ni[ñn]os)\b"
r"|\b(?:for|serves?|feeds?)\s+(\d{1,3})\s+(?:people|persons|servings|guests|adults|kids)\b",
re.IGNORECASE,
)
_QUANTITY_ASK = re.compile(
r"\b(?:lista\s+de\s+(?:la\s+)?compra|receta|ingredientes|cantidades|cu[aá]nt[oa]s?|men[uú]|"
r"shopping\s+list|recipe|ingredients|quantities|how\s+(?:much|many)|menu)\b",
re.IGNORECASE,
)
def servings_requested(text: str) -> Optional[int]:
"""The number of people a request asks quantities for, or None."""
text = str(text or "")
if not _QUANTITY_ASK.search(text):
return None
found = [int(a or b) for a, b in _SERVINGS.findall(text)]
found = [n for n in found if 1 < n <= 500]
return found[-1] if found else None
def servings_note(people: int) -> str:
return (f"The user needs quantities for {people} people. Work each one out as the portion per "
f"person times {people} (use the python tool for the arithmetic when you have it), and "
"keep counts and weights consistent with each other (pieces x weight per piece).")
# Small calculations with an unambiguous answer are worth checking outside the
# model. Keep this deliberately narrow: guessing which number is a price or a
# headcount in a general finance question would create false corrections.
_LETTER_COUNT = re.compile(
r"(?:cu[aá]ntas?\s+veces|how\s+many\s+times).*?"
r"(?:letra|letter)\s+([^\W\d_])\b.*?"
r"(?:palabra|word)\s+([^\W\d_]+)\b", re.IGNORECASE | re.DOTALL,
)
_TIP_PERCENT = re.compile(r"\bpropina\s+del?\s+(\d{1,2}(?:[,.]\d+)?)\s*%", re.IGNORECASE)
_TIP_BILL = re.compile(r"\bcuenta\s+de\s+(\d{1,6}(?:[,.]\d{1,2})?)\s*€?", re.IGNORECASE)
_TIP_PEOPLE = re.compile(r"\bsomos\s+(\d{1,3})\b", re.IGNORECASE)
_TIP_PER_PERSON = re.compile(r"\b(?:cada\s+uno|por\s+persona)\b", re.IGNORECASE)
def verified_calculation(question: str) -> Optional[Dict[str, str]]:
"""A tiny, exact result for a narrowly phrased count or bill split."""
question = str(question or "")
if len(question) > 250 or "\n" in question or question.count("?") > 1:
return None
match = _LETTER_COUNT.search(question)
if match:
letter, word = match.group(1), match.group(2)
return {"kind": "letter_count", "value": str(word.casefold().count(letter.casefold())),
"explanation": f"'{word}' contiene {word.casefold().count(letter.casefold())} veces la letra '{letter}'."}
pct, bill, people = (_TIP_PERCENT.search(question), _TIP_BILL.search(question),
_TIP_PEOPLE.search(question))
if pct and bill and people and _TIP_PER_PERSON.search(question):
n = int(people.group(1))
if not 1 < n <= 100:
return None
rate = Decimal(pct.group(1).replace(",", "."))
amount = Decimal(bill.group(1).replace(",", "."))
value = (amount * (1 + rate / 100) / n).quantize(Decimal("0.01"))
return {"kind": "tip_split", "value": str(value),
"explanation": f"Cuenta {str(amount).replace('.', ',')} € + {rate} % de propina, entre {n} personas: {str(value).replace('.', ',')} € por persona."}
return None
def calculation_mismatch(question: str, answer: str) -> Optional[Dict[str, str]]:
"""Return the verified result when the requested final number is absent."""
fact = verified_calculation(question)
if not fact or not str(answer or "").strip():
return None
value = fact["value"]
if fact["kind"] == "letter_count":
if re.search(rf"(?<!\d){re.escape(value)}(?!\d)", answer):
return None
else:
whole, cents = value.split(".")
if re.search(rf"(?<!\d){re.escape(whole)}[,.]{re.escape(cents)}(?!\d)", answer):
return None
return fact
def saved_links_count_requested(question: str) -> bool:
"""An explicit count of the user's Links Hoard library (not a web fact)."""
body = str(question or "").casefold()
return ("links hoard" in body and "enlace" in body
and re.search(r"\b(?:cu[aá]ntos?|n[uú]mero|total)\b", body) is not None)
def saved_links_total(tool_events: List[Dict[str, object]]) -> Optional[int]:
"""The total from a successful all-state list_links call in this turn."""
for event in reversed(tool_events):
if not str(event.get("tool") or "").endswith("__list_links") or event.get("exit_code") not in (None, 0):
continue
try:
import json
args = json.loads(str(event.get("command") or "{}"))
if args.get("state") != "all":
continue
output = json.loads(str(event.get("output") or "{}"))
total = output.get("total")
if isinstance(total, int) and total >= 0:
return total
except (TypeError, ValueError, AttributeError):
continue
return None
_MARKDOWN_TARGET = re.compile(r'(!?\[[^\]\n]*\]\()[ \t]*<?([^()\s<>]+)>?([ \t]+"[^"\n]*")?[ \t]*\)')
_BARE_IMAGE_URL = re.compile(r'(?<![(<"\w])((?:https?|sandbox|file):/{1,3}[^\s()<>"\]]+?\.(?:png|jpe?g|webp))(?![\w/])',
re.IGNORECASE)
def _produced_generated_images(tool_events: List[Dict[str, object]]) -> Dict[str, str]:
"""image_id -> the gallery URL a tool of THIS turn actually returned."""
produced: Dict[str, str] = {}
for event in tool_events or []:
if not isinstance(event, dict) or event.get("exit_code") not in (None, 0):
continue
url = event.get("image_url")
if not isinstance(url, str) or not url.startswith("/api/generated-image/"):
continue
stem = url.rsplit("/", 1)[-1].split(".", 1)[0]
image_id = event.get("image_id")
# An approved call's saved event carries the URL but not the ID; the
# gallery file is named after its ID, so the stem identifies it.
if stem and re.fullmatch(r"[A-Za-z0-9_-]{1,128}", stem) and image_id in (None, "", stem):
produced[stem] = url
return produced
def _produced_url_for(target: str, produced: Dict[str, str]) -> Optional[str]:
if target in produced.values():
return None
from urllib.parse import urlsplit
try:
stem = urlsplit(target).path.rsplit("/", 1)[-1].split(".", 1)[0]
except ValueError:
return None
return produced.get(stem)
def canonical_generated_image_links(answer: str, tool_events: List[Dict[str, object]]) -> str:
"""Point links to a just-produced gallery image at the URL its tool returned.
Small models copy the right image ID into an invented host
(``https://api.gallery.example.com/generated-image/<id>.png``), which the
chat then renders as a broken image. Only targets whose file stem is an
image ID a successful tool of this turn produced are rewritten, to that
tool's own relative URL; every other link is left exactly as written.
"""
produced = _produced_generated_images(tool_events)
text = str(answer or "")
if not produced or not text:
return text
def markdown(match: "re.Match[str]") -> str:
url = _produced_url_for(match.group(2), produced)
if url is None:
return match.group(0)
return f"{match.group(1)}{url}{match.group(3) or ''})"
def bare(match: "re.Match[str]") -> str:
url = _produced_url_for(match.group(1), produced)
return match.group(0) if url is None else url
text = _MARKDOWN_TARGET.sub(markdown, text)
return _BARE_IMAGE_URL.sub(bare, text)