From 7fb59ca52fa6796e63191851a65317bba2ea422a Mon Sep 17 00:00:00 2001 From: shipitdev Date: Sat, 7 Nov 2026 20:56:15 +0530 Subject: [PATCH] Fix quota pacing and safe withholding diagnostics --- README.md | 2 + rag_app/api.py | 4 +- rag_app/generation.py | 71 +++++++++++++++++++++++++++++------- scripts/eval_ragas.py | 4 +- tests/test_app_generation.py | 53 +++++++++++++++++++++++++++ 5 files changed, 119 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 35863ec..b707721 100644 --- a/README.md +++ b/README.md @@ -121,6 +121,8 @@ Set `GROQ_API_KEY` in the root `.env`. Existing installations can use the `GROK_ The model namespace starts with `openai/`, but requests go to **Groq's API**. The model is configured centrally in `rag_app/generation.py`. +Generation uses remaining-token/reset headers to pace calls and keeps quota waits within request deadlines. The token estimate is approximate and applies to one provider instance, not a shared multi-worker quota. Offline case collection saves private writer/repair/verifier results and content-free HTTP statuses for diagnosis; these artifacts stay outside Git and are never returned as released answers. + ### Prepare or restore artifacts This is a code-only checkout, not a bundled dataset or one-command model download. Restore the matching prepared artifacts before serving: diff --git a/rag_app/api.py b/rag_app/api.py index b01d18f..3d4e1d4 100644 --- a/rag_app/api.py +++ b/rag_app/api.py @@ -166,6 +166,7 @@ def worker(): provider = runtime.get('provider') if provider: provider.usage.clear() + provider.attempts.clear() provider.deadline = started + 180 context = sessions.context(session.id) question = body.question @@ -191,7 +192,7 @@ def worker(): 'pipeline': pipeline_version(), 'index': retrieve.index_identity() if hasattr(retrieve, 'index_identity') else 'injected_test_index', 'model': MODEL, - 'prompt': 'strict_v1', 'generation': {'provider': 'groq', 'reasoning_effort': 'low', 'max_output_tokens': 1500}} + 'prompt': 'strict_v2', 'generation': {'provider': 'groq', 'reasoning_effort': 'low', 'max_output_tokens': 1500}} cached = None if regen or result else answers.get(session.id, identity, session_expires_at=session.expires_at) if cached: result = dict(cached) @@ -246,6 +247,7 @@ def worker(): if not completed: provider_calls = list(provider.usage) provider.usage.clear() + provider.attempts.clear() if not completed: write_metrics({'request_id': rid, 'status': 'cancelled_or_timed_out', 'metrics': {'provider_calls': provider_calls}}) diff --git a/rag_app/generation.py b/rag_app/generation.py index fe0ecf6..a36749d 100644 --- a/rag_app/generation.py +++ b/rag_app/generation.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import math import os import random import re @@ -44,6 +45,20 @@ def __init__(self, api_key: str | None = None, *, model: str = MODEL, raise GenerationError("Groq API key is not configured") self.model, self.timeout, self.post = model, min(timeout, 75), post self.usage: list[dict[str, Any]] = [] + self.attempts: list[dict[str, Any]] = [] + self._remaining_tokens, self._quota_ready_at = float("inf"), 0.0 + + def _remember_quota(self, headers): + try: + remaining = float(headers["x-ratelimit-remaining-tokens"]) + reset = re.fullmatch(r"(?:(\d+(?:\.\d+)?)m)?(?:(\d+(?:\.\d+)?)s)?", headers["x-ratelimit-reset-tokens"]) + if reset and any(reset.groups()) and math.isfinite(remaining) and remaining >= 0: + seconds = float(reset[1] or 0) * 60 + float(reset[2] or 0) + if math.isfinite(seconds): + self._remaining_tokens = remaining + self._quota_ready_at = time.monotonic() + seconds + 2 + except (KeyError, TypeError, ValueError): + pass def json_call(self, prompt: str, schema: dict[str, Any]) -> dict[str, Any]: url = "https://api.groq.com/openai/v1/chat/completions" @@ -57,30 +72,40 @@ def json_call(self, prompt: str, schema: dict[str, Any]) -> dict[str, Any]: started = time.monotonic() deadline = min(started + self.timeout, getattr(self, "deadline", float("inf"))) payload = None - for attempt in range(2): + for attempt in range(3): + # shortcut: estimate English prompt tokens; quota is per client, not shared across processes. + wait = max(0, self._quota_ready_at - time.monotonic()) if self._remaining_tokens < len(prompt) // 4 + 1600 else 0 + if wait: + if wait >= deadline - time.monotonic(): + raise GenerationError("Quota reset exceeds request deadline", category="quota_deadline") + time.sleep(wait) remaining = deadline - time.monotonic() if remaining <= 0: - raise GenerationError("Groq request failed") from None + raise GenerationError("Groq request deadline exceeded", category="request_deadline") from None try: response = self.post(url, headers=headers, json=body, timeout=remaining) + self.attempts.append({"status": response.status_code}) + self._remember_quota(response.headers) response.raise_for_status() payload = response.json() break except httpx.HTTPStatusError as exc: status = exc.response.status_code - retry = attempt == 0 and (status == 429 or 500 <= status <= 599) + retry = attempt < 2 and (status == 429 or 500 <= status <= 599) retry_after = 0.0 if retry: try: - retry_after = max(0.0, float(exc.response.headers.get("Retry-After", 0))) + value = float(exc.response.headers.get("Retry-After", 0)) + retry_after = max(0.0, value) if math.isfinite(value) else 0 except (TypeError, ValueError): pass - delay = retry_after + random.uniform(0.05, 0.2) + delay = max(5 if status == 429 else 0, retry_after + 2) + random.uniform(0.05, 0.2) if retry and delay < deadline - time.monotonic(): time.sleep(delay) continue raise GenerationError(f"Groq HTTP {status}", category=f"http_{status}") from None except httpx.TransportError as exc: + self.attempts.append({"status": None, "error_type": type(exc).__name__}) raise GenerationError("Groq transport " + type(exc).__name__, category="transport_" + type(exc).__name__) from None except (ValueError, TypeError) as exc: raise GenerationError("Groq returned an invalid HTTP response", category="invalid_response") from None @@ -184,6 +209,20 @@ class AnswerGenerator: def __init__(self, json_call: Callable[[str, dict[str, Any]], dict[str, Any]]): self.json_call = json_call + self.trace: list[dict[str, Any]] = [] + + def _call(self, stage, prompt, schema): + # Private diagnostics only; never included in the released answer. + event = {"stage": stage} + self.trace.append(event) + try: + result = self.json_call(prompt, schema) + except GenerationError as exc: + event["error_type"] = exc.category + raise + event["result"] = deepcopy(result) + return result + @staticmethod def _json(value: Any) -> str: @@ -201,12 +240,13 @@ def _write(self, question: str, evidence: list[dict[str, Any]], repair: dict[str schema = deepcopy(ANSWER_SCHEMA) schema["properties"]["citations"]["items"]["enum"] = list(dict.fromkeys( str(item["id"]) for item in evidence if "id" in item)) - result = self.json_call(prompt, schema) + result = self._call("repair" if repair else "write", prompt, schema) if (result.get("status") not in {"answered", "needs_clarification", "insufficient_evidence"} or - not isinstance(result.get("answer"), str) or not result["answer"].strip() or + not isinstance(result.get("answer"), str) or + (result.get("status") == "answered" and not result["answer"].strip()) or not isinstance(result.get("citations"), list) or not all(isinstance(c, str) for c in result["citations"])): - raise GenerationError("Groq returned an invalid answer shape") + raise GenerationError("Groq returned an invalid answer shape", category="invalid_answer_shape") return result @staticmethod @@ -230,17 +270,22 @@ def _non_answer(status: str) -> dict[str, Any]: return {"status": "insufficient_evidence", "answer": "The available sources do not support a complete answer.", "citations": []} def _verify(self, question: str, evidence: list[dict[str, Any]], draft: dict[str, Any]) -> dict[str, Any]: - result = self.json_call("Verify support and completeness using only evidence. Check missing required steps and version conflicts. " + result = self._call("verify", "Verify support and completeness using only evidence. " + "The supported field means the answer's claims are backed by evidence, not that a product supports a feature. " + "An evidence-backed answer saying a feature is NOT supported must have supported=true. " + "Judge completeness against the question asked; do not require unrequested setup or remediation steps. " + "Check missing required steps and version conflicts for the same product, component and conditions. " + "A complete verdict requires supported=true and complete=true; otherwise use evidence_gap or clarify. " "If the draft contains approved_answer, also check that the presented answer preserves its meaning and qualifications. " f"Question: {question}\nEvidence: {self._json(evidence)}\nDraft: {self._json(draft)}", VERIFY_SCHEMA) if (result.get("verdict") not in {"complete", "evidence_gap", "clarify"} or not isinstance(result.get("supported"), bool) or not isinstance(result.get("complete"), bool) or not isinstance(result.get("missing_steps"), list) or not isinstance(result.get("version_conflicts"), list) or not isinstance(result.get("findings"), list)): - raise GenerationError("Groq returned an invalid verification shape") + raise GenerationError("Groq returned an invalid verification shape", category="invalid_verification_shape") return result - def generate(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, Any]: + self.trace.clear() source_ids = {str(item["id"]) for item in evidence if "id" in item} if not evidence or not source_ids: return self._non_answer("insufficient_evidence") @@ -256,7 +301,7 @@ def generate(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, A if draft["status"] != "answered": return self._non_answer(draft["status"]) if self._citation_error(draft, source_ids): - raise GenerationError("Answer contains invalid or missing citations") + raise GenerationError("Answer contains invalid or missing citations", category="invalid_citations") verdict = self._verify(question, evidence, draft) if (verdict["verdict"] == "complete" and verdict["supported"] and verdict["complete"] and not verdict["missing_steps"] and not verdict["version_conflicts"]): @@ -275,7 +320,7 @@ def generate(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, A if repaired["status"] != "answered": return self._non_answer(repaired["status"]) if self._citation_error(repaired, source_ids): - raise GenerationError("Answer contains invalid or missing citations") + raise GenerationError("Answer contains invalid or missing citations", category="invalid_citations") final_verdict = self._verify(question, evidence, repaired) if (final_verdict["verdict"] == "complete" and final_verdict["supported"] and final_verdict["complete"] and not final_verdict["missing_steps"] and not final_verdict["version_conflicts"]): diff --git a/scripts/eval_ragas.py b/scripts/eval_ragas.py index c0d76e6..ea2e992 100644 --- a/scripts/eval_ragas.py +++ b/scripts/eval_ragas.py @@ -189,6 +189,7 @@ def collect(args): started = time.perf_counter() retrieved = retriever.retrieve(question) provider.usage.clear() + provider.attempts.clear() provider.deadline = time.monotonic() + 180 try: answer = generator.generate(question, retrieved['sources']) @@ -196,7 +197,8 @@ def collect(args): answer = {'status': 'generation_error', 'answer': '', 'error_type': exc.category} case = make_case(q, original_docs[q['DOCUMENT']]['text'], retrieved, answer) case['generation'] = {'seconds': time.perf_counter() - started, - 'provider_calls': list(provider.usage), 'error_type': answer.get('error_type')} + 'provider_calls': list(provider.usage), 'error_type': answer.get('error_type'), + 'http_attempts': list(provider.attempts), 'trace': list(generator.trace)} save(path, case) print(f" saved {case['status']}; {len(case['sources'])} passages", flush=True) finally: diff --git a/tests/test_app_generation.py b/tests/test_app_generation.py index c8179d6..3dd7a3c 100644 --- a/tests/test_app_generation.py +++ b/tests/test_app_generation.py @@ -108,6 +108,59 @@ def test_missing_key_is_bounded_error(self): with self.assertRaisesRegex(GenerationError, "not configured"): GroqREST(api_key="") + def test_empty_nonanswers_are_safe_but_empty_answers_are_errors(self): + for status in ("insufficient_evidence", "needs_clarification"): + result, calls = self.run_with([{"status": status, "answer": "", "citations": []}]) + self.assertEqual(result["status"], status) + self.assertTrue(result["answer"]) + self.assertEqual(len(calls), 1) + with self.assertRaises(GenerationError) as error: + self.run_with([{"status": "answered", "answer": "", "citations": []}]) + self.assertEqual(error.exception.category, "invalid_answer_shape") + + def test_private_trace_records_verifier_rejection_without_changing_output(self): + replies = [self.answer.copy(), {**self.complete, "verdict": "evidence_gap", "complete": False}] + generator = AnswerGenerator(lambda *args: replies.pop(0)) + result = generator.generate("How?", self.evidence) + self.assertEqual(result["status"], "insufficient_evidence") + self.assertEqual([x["stage"] for x in generator.trace], ["write", "verify"]) + self.assertEqual(generator.trace[-1]["result"]["verdict"], "evidence_gap") + self.assertNotIn("trace", result) + + def test_quota_headers_pace_next_call_and_respect_deadline(self): + clock = [0.0] + calls = [] + def sleep(seconds): clock[0] += seconds + def post(url, **kwargs): + calls.append(clock[0]) + return httpx.Response(200, request=httpx.Request("POST", url), + headers={"x-ratelimit-remaining-tokens": "10", "x-ratelimit-reset-tokens": "1m2s"}, + json={"choices": [{"message": {"content": "{}"}, "finish_reason": "stop"}]}) + with patch("rag_app.generation.time.monotonic", lambda: clock[0]), patch("rag_app.generation.time.sleep", sleep): + client = GroqREST("secret", post=post) + client.json_call("test", {}) + client.json_call("test", {}) + self.assertGreaterEqual(calls[1]-calls[0], 64) + client.deadline = clock[0] + 5 + with self.assertRaises(GenerationError) as error: client.json_call("test", {}) + self.assertEqual(error.exception.category, "quota_deadline") + self.assertEqual(len(calls), 2) + + def test_two_rate_limits_get_buffered_bounded_retries(self): + clock, calls = [0.0], [] + def post(url, **kwargs): + calls.append(clock[0]) + return httpx.Response(429 if len(calls) < 3 else 200, + request=httpx.Request("POST", url), headers={"retry-after": "nan"}, + json={"choices": [{"message": {"content": "{}"}, "finish_reason": "stop"}]}) + def sleep(seconds): clock[0] += seconds + with patch("rag_app.generation.time.monotonic", lambda: clock[0]), patch("rag_app.generation.time.sleep", sleep): + client = GroqREST("secret", post=post) + self.assertEqual(client.json_call("test", {}), {}) + self.assertEqual(len(calls), 3) + self.assertGreaterEqual(calls[1]-calls[0], 5) + self.assertGreaterEqual(calls[2]-calls[1], 5) + def test_unreferenced_citation_array_does_not_pass(self): uncited = {**self.answer, "answer": "Use X without an inline citation."} calls = []