From 9aee9464a49b52464c22a7ce6abe62470ac1f7b7 Mon Sep 17 00:00:00 2001 From: shipitdev Date: Wed, 11 Nov 2026 12:24:59 +0530 Subject: [PATCH 1/4] Run DeepSeek with bounded spending and accurate provider traces --- .env.example | 6 ++ .gitignore | 5 ++ rag_app/api.py | 41 ++++++------ rag_app/generation.py | 121 ++++++++++++++++++++++++----------- scripts/eval_budget.py | 63 ++++++++++++++++++ tests/test_app_api.py | 2 +- tests/test_app_generation.py | 41 ++++++++++++ 7 files changed, 221 insertions(+), 58 deletions(-) create mode 100644 scripts/eval_budget.py diff --git a/.env.example b/.env.example index cc12a10..601cd00 100644 --- a/.env.example +++ b/.env.example @@ -17,3 +17,9 @@ ENABLE_GENUI=1 LANGSMITH_API_KEY= LANGSMITH_PROJECT=techqa-rag LANGSMITH_TRACING=true + +# Select groq or tokenharbor for generation. +RAG_LLM_PROVIDER=groq +TOKENHARBOR_BASE_URL=https://tokenharbor.ai/v1 +TOKENHARBOR_MODEL=deepseek-v4.1-flash:free +TOKENHARBOR_API_KEY= diff --git a/.gitignore b/.gitignore index 4fd9487..a011d50 100644 --- a/.gitignore +++ b/.gitignore @@ -144,3 +144,8 @@ !/evaluations/results/techqa_wide_v5.json !/evaluations/results/generation_v6.json + +!/scripts/eval_budget.py +!/scripts/eval_layers.py + +!/evaluations/results/layered_audit_10.json diff --git a/rag_app/api.py b/rag_app/api.py index 0c782b3..3bb20fa 100644 --- a/rag_app/api.py +++ b/rag_app/api.py @@ -18,7 +18,7 @@ from rag_app.state import SessionStore, AnswerCache from rag_app.observability import persist_run from rag_app.tracing import trace_span -from rag_app.generation import MODEL, PROMPT_VERSION, GroqREST, AnswerGenerator, GenerationError +from rag_app.generation import MODEL, PROMPT_VERSION, configured_provider, AnswerGenerator, GenerationError ROOT = Path(__file__).resolve().parents[1] load_dotenv(ROOT / '.env') @@ -35,7 +35,7 @@ class Feedback(BaseModel): comment: str = Field(default='', max_length=2000) -def create_app(retriever=None, generator=None, resolver=None): +def create_app(retriever=None, generator=None, resolver=None, provider=None): app = FastAPI(title='TechQA evidence workspace') serial = threading.Lock() admission = threading.BoundedSemaphore(5) @@ -45,7 +45,7 @@ def create_app(retriever=None, generator=None, resolver=None): limits = defaultdict(deque) limiter_lock = threading.Lock() feedback_lock = threading.Lock() - runtime = {'retriever': retriever, 'generator': generator, 'resolver': resolver} + runtime = {'retriever': retriever, 'generator': generator, 'resolver': resolver, 'provider': provider} def pipeline_version(): """Invalidate exact answers when data, runtime code, or UI library changes.""" @@ -79,7 +79,7 @@ def dependencies(): from rag_app.retriever import Retriever runtime['retriever'] = Retriever() if runtime['generator'] is None: - provider = GroqREST() + provider = configured_provider() runtime['provider'] = provider runtime['generator'] = AnswerGenerator(provider.json_call) from rag_app.generation import ConversationResolver, history_summarizer @@ -97,9 +97,11 @@ def live(): @app.get('/health/ready') def ready(): if retriever is not None: - return {'status': 'ready', 'presentation': 'direct_groq'} - if not (os.getenv('GROQ_API_KEY') or os.getenv('GROK_API_KEY')): - raise HTTPException(503, 'Groq API key is not configured') + return {'status': 'ready', 'presentation': 'direct_' + os.getenv('RAG_LLM_PROVIDER', 'groq')} + try: + configured_provider() + except GenerationError: + raise HTTPException(503, 'Model provider is not configured') from None try: from rag_app.retriever import Retriever check = Retriever() @@ -110,7 +112,7 @@ def ready(): check.close() except Exception: raise HTTPException(503, 'Qdrant/source import is not ready') from None - return {'status': 'ready', 'presentation': 'direct_groq', 'models': 'lazy-loaded on first request', 'index': identity, 'tracing': bool(os.getenv('LANGSMITH_API_KEY')) and os.getenv('LANGSMITH_TRACING', 'true').lower() == 'true'} + return {'status': 'ready', 'presentation': 'direct_' + os.getenv('RAG_LLM_PROVIDER', 'groq'), 'models': 'lazy-loaded on first request', 'index': identity, 'tracing': bool(os.getenv('LANGSMITH_API_KEY')) and os.getenv('LANGSMITH_TRACING', 'true').lower() == 'true'} @app.post('/session/end') def end(request: Request): @@ -223,8 +225,9 @@ def run_worker(): if dependent else None), 'pipeline': pipeline_version(), 'index': retrieve.index_identity() if hasattr(retrieve, 'index_identity') else 'injected_test_index', - 'model': MODEL, - 'prompt': PROMPT_VERSION, 'generation': {'provider': 'groq', 'reasoning_effort': 'low', 'max_output_tokens': 1500}} + 'model': provider.model if provider else MODEL, + 'prompt': PROMPT_VERSION, 'generation': {'provider': provider.provider if provider else 'injected', + 'max_output_tokens': provider.max_tokens if provider else 1500}} cached = None if regen or result else answers.get(session.id, identity, session_expires_at=session.expires_at) if cached: result = dict(cached) @@ -266,7 +269,7 @@ def run_worker(): 'answer_cache_hit': False, 'regenerated': regen, }, } - if result['status'] == 'answered' and (os.getenv('GROQ_API_KEY') or os.getenv('GROK_API_KEY')) and os.getenv('ENABLE_GENUI', '1') == '1' and runtime.get('provider'): + if result['status'] == 'answered' and os.getenv('ENABLE_GENUI', '1') == '1' and runtime.get('provider'): if stopped(): return emit('stage', {'stage': 'validating_presentation'}) from rag_app.presentation import generate_presentation @@ -278,7 +281,7 @@ def run_worker(): answers.put(session.id, identity, result, session_expires_at=session.expires_at) result['request_id'] = rid result.setdefault('metrics', {})['total_ms'] = round((time.monotonic() - started) * 1000, 2) - result['presentation_status'] = 'direct_groq' if result.get('openui') else 'verified_text_fallback' + result['presentation_status'] = ('direct_' + os.getenv('RAG_LLM_PROVIDER', 'groq')) if result.get('openui') else 'verified_text_fallback' provider = runtime.get('provider') if provider: provider_calls = list(provider.usage) @@ -297,18 +300,18 @@ def run_worker(): feedback_runs.put(session.id, rid, { 'question': body.question, 'resolved_question': question, 'result': result, 'pipeline_version': identity['pipeline'], 'prompt_version': PROMPT_VERSION, - 'model': MODEL, 'index': identity['index'], 'private_trace': private_trace, + 'model': identity['model'], 'index': identity['index'], 'private_trace': private_trace, }, session_expires_at=session.expires_at) emit('final', result) write_metrics({'request_id': rid, 'status': result['status'], 'metrics': result['metrics']},private_trace) completed = True except Exception as exc: provider_messages = { - 'http_402': 'Groq billing or credits need attention. Check the Groq Console before retrying.', - 'http_429': 'Groq quota was reached. Retry after your provider quota resets.', - 'quota_deadline': 'Groq quota will not reset within this request. Try again after it resets.', - 'transport_TimeoutError': 'Groq did not respond before the time limit. No answer was released.', - 'transport_ReadTimeout': 'Groq did not respond before the time limit. No answer was released.', + 'http_402': 'Model provider credits need attention. Check your provider account before retrying.', + 'http_429': 'Model provider quota was reached. Retry after your provider quota resets.', + 'quota_deadline': 'Model provider quota will not reset within this request. Try again after it resets.', + 'transport_TimeoutError': 'The model provider did not respond before the time limit. No answer was released.', + 'transport_ReadTimeout': 'The model provider did not respond before the time limit. No answer was released.', } if isinstance(exc, GenerationError): message = provider_messages.get(exc.category, 'Answer generation failed; please retry.') @@ -353,7 +356,7 @@ def run_worker(): emit('done', {}) def worker(): - with trace_span('TechQA question', inputs={'question': body.question, 'regenerate': body.regenerate}, run_id=uuid.UUID(rid), metadata={'model': MODEL, 'prompt_version': PROMPT_VERSION, 'pipeline_sha256': pipeline_version()}) as run: + with trace_span('Pipeline / Question', inputs={'question': body.question, 'regenerate': body.regenerate}, run_id=uuid.UUID(rid), metadata={'model': os.getenv('TOKENHARBOR_MODEL', MODEL) if os.getenv('RAG_LLM_PROVIDER') == 'tokenharbor' else MODEL, 'prompt_version': PROMPT_VERSION, 'pipeline_sha256': pipeline_version()}) as run: run_worker() if run and not run.outputs: run.end(outputs={'status': 'cancelled_or_timed_out'}) diff --git a/rag_app/generation.py b/rag_app/generation.py index 5b185da..bab679f 100644 --- a/rag_app/generation.py +++ b/rag_app/generation.py @@ -55,11 +55,14 @@ def _groq_schema(schema: dict[str, Any]) -> dict[str, Any]: class GroqREST: def __init__(self, api_key: str | None = None, *, model: str = MODEL, - timeout: float = 75, post: Callable[..., Any] = httpx.post): + timeout: float = 75, post: Callable[..., Any] = httpx.post, + provider: str = "groq", base_url: str = "https://api.groq.com/openai/v1", + max_tokens: int = 1500): self.api_key = api_key if api_key is not None else os.getenv("GROQ_API_KEY") or os.getenv("GROK_API_KEY") if not self.api_key: - raise GenerationError("Groq API key is not configured") - self.model, self.timeout, self.post = model, min(timeout, 75), post + raise GenerationError("Model provider API key is not configured") + self.provider, self.base_url, self.max_tokens = provider, base_url.rstrip("/"), max_tokens + self.model, self.timeout, self.post = model, min(timeout, 120 if provider == "tokenharbor" else 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 @@ -79,24 +82,32 @@ def _remember_quota(self, headers): def json_call(self, prompt: ChatPrompt, schema: dict[str, Any]) -> dict[str, Any]: messages = ([{"role": "system", "content": SYSTEM_RULES}, {"role": "user", "content": prompt}] if isinstance(prompt, str) else prompt) - with trace_span("Groq structured completion", inputs={"messages": messages}, run_type="llm", - metadata={"ls_provider": "groq", "ls_model_name": self.model}) as span: - result = self._json_call(messages, schema) + with trace_span(f"Pipeline / LLM / {self.provider}", inputs={"messages": messages}, run_type="llm", + metadata={"ls_provider": self.provider, "ls_model_name": self.model}) as span: + before = len(self.usage) + try: + result = self._json_call(messages, schema) + finally: + if span and len(self.usage) > before: + usage = self.usage[-1] + recorded = {"input_tokens": usage.get("prompt_tokens"), + "output_tokens": usage.get("output_tokens"), + "total_tokens": usage.get("total_tokens")} + span.end(outputs={"usage_metadata": recorded}, metadata={"usage_metadata": recorded}) if span: - usage = self.usage[-1] - span.end(outputs=result, metadata={"usage_metadata": { - "input_tokens": usage.get("prompt_tokens"), - "output_tokens": usage.get("output_tokens"), - "total_tokens": usage.get("total_tokens"), - }}) + span.end(outputs=result) return result def _json_call(self, messages: list[dict[str, str]], schema: dict[str, Any]) -> dict[str, Any]: - url = "https://api.groq.com/openai/v1/chat/completions" + url = self.base_url + "/chat/completions" body = {"model": self.model, "messages": messages, - "reasoning_effort": "low", "max_completion_tokens": 1500, "stream": False, + "stream": False, "response_format": {"type": "json_schema", "json_schema": { "name": "rag_response", "strict": True, "schema": _groq_schema(schema)}}} + if self.provider == "groq": + body.update(reasoning_effort="low", max_completion_tokens=self.max_tokens) + else: + body.update(max_tokens=self.max_tokens, thinking={"type": "disabled"}) headers = {"Authorization": f"Bearer {self.api_key}"} prompt_chars = sum(len(message["content"]) for message in messages) started = time.monotonic() @@ -111,7 +122,7 @@ def _json_call(self, messages: list[dict[str, str]], schema: dict[str, Any]) -> time.sleep(wait) remaining = deadline - time.monotonic() if remaining <= 0: - raise GenerationError("Groq request deadline exceeded", category="request_deadline") from None + raise GenerationError("Model provider 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}) @@ -133,43 +144,77 @@ def _json_call(self, messages: list[dict[str, str]], schema: dict[str, Any]) -> if retry and delay < deadline - time.monotonic(): time.sleep(delay) continue - raise GenerationError(f"Groq HTTP {status}", category=f"http_{status}") from None + raise GenerationError(f"Model provider 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 + raise GenerationError("Model provider 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 + raise GenerationError("Model provider returned an invalid HTTP response", category="invalid_response") from None + if not isinstance(payload, dict): + raise GenerationError("Model provider returned an invalid response", category="invalid_response") + metadata = payload.get("usage") or {} + if not isinstance(metadata, dict): + metadata = {} + details = metadata.get("completion_tokens_details") or {} + prompt_details = metadata.get("prompt_tokens_details") or {} + self.usage.append({"provider": self.provider, "model": self.model, "latency_seconds": time.monotonic() - started, + "prompt_tokens": metadata.get("prompt_tokens"), + "output_tokens": metadata.get("completion_tokens"), + "reasoning_tokens": details.get("reasoning_tokens"), + "cached_tokens": prompt_details.get("cached_tokens"), + "total_tokens": metadata.get("total_tokens")}) try: choice = payload["choices"][0] message = choice["message"] if choice.get("finish_reason") == "length": - raise GenerationError("Groq response exceeded the output limit", category="output_truncated") + raise GenerationError("Model provider response exceeded the output limit", category="output_truncated") if choice.get("finish_reason") not in {"stop", None}: - raise GenerationError("Groq returned an incomplete response", category="incomplete_response") + raise GenerationError("Model provider returned an incomplete response", category="incomplete_response") if message.get("refusal"): - raise GenerationError("Groq declined the request", category="refusal") + raise GenerationError("Model provider declined the request", category="refusal") text = message.get("content") if not isinstance(text, str) or not text.strip(): - raise GenerationError("Groq returned an empty structured response", category="invalid_response") + raise GenerationError("Model provider returned an empty structured response", category="invalid_response") result = json.loads(text) except GenerationError: raise except (KeyError, IndexError, TypeError, ValueError) as exc: - raise GenerationError("Groq returned an invalid structured response", category="invalid_response") from None + raise GenerationError("Model provider returned an invalid structured response", category="invalid_response") from None if not isinstance(result, dict): - raise GenerationError("Groq returned an invalid structured response", category="invalid_response") - metadata = payload.get("usage", {}) - details = metadata.get("completion_tokens_details") or {} - prompt_details = metadata.get("prompt_tokens_details") or {} - self.usage.append({"provider": "groq", "model": self.model, "latency_seconds": time.monotonic() - started, - "prompt_tokens": metadata.get("prompt_tokens"), - "output_tokens": metadata.get("completion_tokens"), - "reasoning_tokens": details.get("reasoning_tokens"), - "cached_tokens": prompt_details.get("cached_tokens"), - "total_tokens": metadata.get("total_tokens")}) + raise GenerationError("Model provider returned an invalid structured response", category="invalid_response") return result +def configured_provider(*, post=httpx.post): + """Use the provider selected in server settings.""" + provider = os.getenv("RAG_LLM_PROVIDER", "groq") + if provider == "groq": + return GroqREST(post=post) + if provider != "tokenharbor": + raise GenerationError("Unknown RAG_LLM_PROVIDER") + transport = post + if os.getenv("EVAL_BUDGET_FILE"): + from scripts.eval_budget import EvaluationBudget, BudgetLimitError + ledger = EvaluationBudget(os.environ["EVAL_BUDGET_FILE"]) + def transport(url, **kwargs): + try: + ticket = ledger.begin(kwargs["json"]) + except BudgetLimitError: + raise GenerationError("Evaluation budget reached", category="budget_exhausted") from None + response = post(url, **kwargs) + try: + data = response.json() + usage = data.get("usage", {}) if isinstance(data, dict) else {} + except ValueError: + usage = {} + ledger.finish(ticket, usage, response.status_code) + return response + return GroqREST(api_key=os.getenv("TOKENHARBOR_API_KEY", ""), + model=os.getenv("TOKENHARBOR_MODEL", "deepseek-v4.1-flash:free"), + provider=provider, base_url=os.getenv("TOKENHARBOR_BASE_URL", "https://tokenharbor.ai/v1"), + max_tokens=6000, timeout=90, post=transport) + + ANSWER_SCHEMA = {"type": "OBJECT", "properties": {"answer": {"type": "STRING"}, "status": {"type": "STRING", "enum": ["answered", "needs_clarification", "insufficient_evidence"]}, "citations": {"type": "ARRAY", "items": {"type": "STRING"}}}, @@ -210,7 +255,7 @@ def resolve(self, question: str, context: dict[str, Any]) -> dict[str, Any]: if (not isinstance(result.get("dependent"), bool) or not isinstance(result.get("question"), str) or not isinstance(result.get("needs_clarification"), bool)): - raise GenerationError("Groq returned an invalid follow-up resolution") + raise GenerationError("Model provider returned an invalid follow-up resolution") if not result["dependent"]: return {"dependent": False, "question": question, "needs_clarification": False} if result["needs_clarification"] or not result["question"].strip(): @@ -228,7 +273,7 @@ def summarize(previous: str, exchanges: list[Any]) -> str: "instead of merging them. Do not invent facts; this summary is context, never evidence.", {"previous_summary": previous, "older_exchanges": history}), SUMMARY_SCHEMA) if not isinstance(result.get("summary"), str): - raise GenerationError("Groq returned an invalid history summary") + raise GenerationError("Model provider returned an invalid history summary") return result["summary"] return summarize @@ -248,7 +293,7 @@ def _call(self, stage, prompt, schema): if getattr(self, "on_progress", None): self.on_progress(stage) try: - with trace_span(stage, inputs={"messages": prompt}, metadata={"stage": stage}) as span: + with trace_span(f"Pipeline / {stage.title()}", inputs={"messages": prompt}, metadata={"stage": stage}) as span: result = self.json_call(prompt, schema) if span: span.end(outputs=result) @@ -284,7 +329,7 @@ def _write(self, question: str, evidence: list[dict[str, Any]]) -> dict[str, Any (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", category="invalid_answer_shape") + raise GenerationError("Model provider returned an invalid answer shape", category="invalid_answer_shape") return result @staticmethod @@ -331,7 +376,7 @@ def _verify(self, question: str, evidence: list[dict[str, Any]], draft: dict[str not isinstance(result.get("findings"), list) or not isinstance(result.get("unsupported_claims"), list) or not all(isinstance(x, str) for key in ("missing_steps", "version_conflicts", "findings", "unsupported_claims") for x in result[key])): - raise GenerationError("Groq returned an invalid verification shape", category="invalid_verification_shape") + raise GenerationError("Model provider returned an invalid verification shape", category="invalid_verification_shape") if result["unsupported_claims"]: result.update(verdict="evidence_gap", supported=False, complete=False) return result diff --git a/scripts/eval_budget.py b/scripts/eval_budget.py new file mode 100644 index 0000000..e2d1d85 --- /dev/null +++ b/scripts/eval_budget.py @@ -0,0 +1,63 @@ +"""Shared local evaluation budget, valued at published standard-route prices.""" +import fcntl +import json +from pathlib import Path +from uuid import uuid4 + +from rag_app.indexes import atomic_json + + +class BudgetLimitError(RuntimeError): + pass + + +class EvaluationBudget: + def __init__(self, path, limit=0.50): + self.path = Path(path) + self.limit = limit + + def _change(self, update): + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.with_suffix('.lock').open('a') as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + data = json.loads(self.path.read_text()) if self.path.exists() else { + 'limit_usd': self.limit, 'input_usd_per_million': 0.30, + 'output_usd_per_million': 1.20, 'cache_read_usd_per_million': 0.0015, + 'basis': 'standard-route token value; selected free route bills zero', + 'requests': {}} + result = update(data) + atomic_json(self.path, data) + return result + + def begin(self, body): + if body.get('model', 'deepseek-v4.1-flash:free') not in {'deepseek-v4.1-flash', 'deepseek-v4.1-flash:free'}: + raise BudgetLimitError('No configured price for this judge model') + # UTF-8 byte count bounds text tokens conservatively before a request. + size = len(json.dumps(body, ensure_ascii=False).encode()) + 256 + output = body.get('max_tokens', body.get('max_completion_tokens', 3000)) + reserve = (size * 0.30 + output * 1.20) / 1_000_000 + + def reserve_request(data): + if sum(r['value_usd'] for r in data['requests'].values()) + reserve > data['limit_usd']: + raise BudgetLimitError('Evaluation token-value budget reached') + ticket = str(uuid4()) + data['requests'][ticket] = {'state': 'reserved', 'value_usd': reserve} + return ticket + + return self._change(reserve_request) + + def finish(self, ticket, usage, status): + def settle(data): + if status == 200 and not usage: + data['requests'][ticket].update(state='usage_missing', status=status) + return + prompt = usage.get('prompt_tokens') or 0 + completion = usage.get('completion_tokens') or 0 + cached = min(prompt, usage.get('prompt_cache_hit_tokens') or + (usage.get('prompt_tokens_details') or {}).get('cached_tokens') or 0) + value = ((prompt - cached) * 0.30 + cached * 0.0015 + completion * 1.20) / 1_000_000 + data['requests'][ticket] = {'state': 'reported', 'status': status, + 'input_tokens': prompt, 'output_tokens': completion, 'cached_input_tokens': cached, + 'value_usd': value} + + self._change(settle) diff --git a/tests/test_app_api.py b/tests/test_app_api.py index 6afe730..f376760 100644 --- a/tests/test_app_api.py +++ b/tests/test_app_api.py @@ -123,7 +123,7 @@ def generate(self, question, sources): raise api.GenerationError('provider billing details',category='http_402') client=TestClient(api.create_app(Search(),DepletedWriter())) response=client.post('/ask',json={'question':'How do I restart?'}) - assert 'Groq billing or credits need attention' in response.text + assert 'Model provider credits need attention' in response.text assert 'event: final' not in response.text assert 'provider billing details' not in response.text diff --git a/tests/test_app_generation.py b/tests/test_app_generation.py index e82631c..1d5f666 100644 --- a/tests/test_app_generation.py +++ b/tests/test_app_generation.py @@ -367,3 +367,44 @@ def fake(prompt, schema): assert len(calls) == 2 assert draft["answer"] == "Use X." assert json.loads(calls[1][1]["content"])["draft"] == result + + +def test_tokenharbor_transport_preserves_schema_and_records_actual_provider(monkeypatch): + from rag_app.generation import configured_provider + monkeypatch.setenv("RAG_LLM_PROVIDER", "tokenharbor") + monkeypatch.setenv("TOKENHARBOR_API_KEY", "test") + monkeypatch.setenv("TOKENHARBOR_MODEL", "deepseek-v4.1-flash:free") + sent = [] + def post(url, **kwargs): + sent.append((url, kwargs["json"])) + return httpx.Response(200, request=httpx.Request("POST", url), json={ + "choices": [{"finish_reason": "stop", "message": {"content": '{"answer":"ok"}'}}], + "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5}}) + provider = configured_provider(post=post) + assert provider.json_call("Return JSON", {}) == {"answer": "ok"} + assert sent[0][0] == "https://tokenharbor.ai/v1/chat/completions" + assert sent[0][1]["max_tokens"] == 6000 + assert "reasoning_effort" not in sent[0][1] + assert provider.usage[0]["provider"] == "tokenharbor" + + +def test_truncated_response_keeps_usage_and_provider_neutral_error(): + def post(url, **kwargs): + return httpx.Response(200, request=httpx.Request("POST", url), json={ + "choices": [{"finish_reason": "length", "message": {"content": ""}}], + "usage": {"prompt_tokens": 10, "completion_tokens": 3000, "total_tokens": 3010}}) + provider = GroqREST("test", post=post, provider="tokenharbor") + with unittest.TestCase().assertRaises(GenerationError) as error: + provider.json_call("test", {}) + assert error.exception.category == "output_truncated" + assert "Groq" not in str(error.exception) + assert provider.usage[-1]["total_tokens"] == 3010 + + +def test_malformed_provider_payload_is_a_categorized_error(): + def post(url, **kwargs): + return httpx.Response(200, request=httpx.Request("POST", url), json=[]) + provider = GroqREST("test", post=post) + with unittest.TestCase().assertRaises(GenerationError) as error: + provider.json_call("test", {}) + assert error.exception.category == "invalid_response" From ff14a51bfaac09051265a26d2626103ededc3826 Mon Sep 17 00:00:00 2001 From: shipitdev Date: Wed, 11 Nov 2026 12:24:59 +0530 Subject: [PATCH 2/4] Evaluate components and workflows and import recorded run metrics --- requirements-ragas.txt | 3 + scripts/eval_layers.py | 125 +++++++++++++++++++ scripts/eval_ragas.py | 62 +++++++--- scripts/eval_suite.py | 3 +- scripts/publish_langsmith.py | 212 +++++++++++++++------------------ tests/test_langsmith_export.py | 116 +++++++++--------- tests/test_ragas_eval.py | 88 ++++++++++++++ 7 files changed, 424 insertions(+), 185 deletions(-) create mode 100644 scripts/eval_layers.py diff --git a/requirements-ragas.txt b/requirements-ragas.txt index 65e3f92..dddb817 100644 --- a/requirements-ragas.txt +++ b/requirements-ragas.txt @@ -14,3 +14,6 @@ qdrant-client==1.16.2 pytest==9.2.0 langsmith==0.14.17 + +# Real API workflow collection in the evaluation environment. +fastapi==0.162.0 diff --git a/scripts/eval_layers.py b/scripts/eval_layers.py new file mode 100644 index 0000000..ea33651 --- /dev/null +++ b/scripts/eval_layers.py @@ -0,0 +1,125 @@ +"""Collect supplied-evidence component cases and real API workflow cases for RAGAS.""" +import argparse +import json +import os +from pathlib import Path +import time + +from dotenv import load_dotenv +from fastapi.testclient import TestClient + +from rag_app.api import create_app +from rag_app.generation import AnswerGenerator, ConversationResolver, GenerationError, configured_provider +from rag_app.retriever import Retriever +from rag_app.tracing import trace_span +from scripts.audit_techqa import RAW, ROOT +from scripts.eval_ragas import save +from scripts.eval_suite import load_cases + + +def stream_result(response): + for block in response.text.split('\n\n'): + if block.startswith('event: final\n'): + return json.loads(block.split('data: ', 1)[1]) + if block.startswith('event: error\n'): + error = json.loads(block.split('data: ', 1)[1]) + return {'status': 'generation_error', 'answer': '', 'citations': [], + 'sources': [], 'metrics': error.get('metrics', {})} + raise ValueError('API response has no terminal event') + + +def collect_components(pipeline, output, question_ids): + manifest = json.loads((pipeline / 'manifest.json').read_text()) + cases = {c['question_id']: c for c in load_cases(pipeline)} + if len(set(question_ids)) != len(question_ids) or not set(question_ids) <= set(manifest['question_ids']): + raise ValueError('Component IDs must belong to the pipeline cohort') + documents = json.loads((RAW / 'TechQA/training_and_dev/training_dev_technotes.json').read_text()) + provider = configured_provider() + for qid in question_ids: + case = dict(cases[qid]) + doc = case['reference_provenance']['document_id'] + evidence = [{'id': 'S1', 'document_id': doc, 'text': documents[doc]['text']}] + provider.usage.clear() + provider.attempts.clear() + generator = AnswerGenerator(provider.json_call) + started = time.perf_counter() + with trace_span('Component / Generator / ' + qid, inputs={'question': case['user_input']}, + metadata={'evidence_mode': 'annotated_original_article'}) as run: + try: + answer = generator.generate(case['user_input'], evidence) + except GenerationError as exc: + answer = {'status': 'generation_error', 'answer': '', 'citations': [], 'error_type': exc.category} + case.update(status=answer['status'], response=answer['answer'], citations=answer['citations'], + sources=evidence, retrieved_contexts=[json.dumps(evidence[0], ensure_ascii=False)], + evaluation_layer='component', langsmith_run_id=str(run.id) if run else None, + generation={'seconds': time.perf_counter() - started, 'trace': generator.trace, + 'provider_calls': list(provider.usage), 'http_attempts': list(provider.attempts), + 'error_type': answer.get('error_type')}) + if run: + run.end(outputs={'status': case['status'], 'response': case['response']}) + save(output / (qid + '.json'), case) + save(output / 'manifest.json', {**manifest, 'question_ids': question_ids, + 'evaluation_layer': 'component', 'scope': 'generator with supplied original annotated article'}) + + +def collect_workflow(pipeline, output, root_id, followup): + manifest = json.loads((pipeline / 'manifest.json').read_text()) + if root_id not in manifest['question_ids']: + raise ValueError('Workflow root must belong to the pipeline cohort') + base = {c['question_id']: c for c in load_cases(pipeline)}[root_id] + provider = configured_provider() + generator = AnswerGenerator(provider.json_call) + search = Retriever() + old_ui = os.environ.get('ENABLE_GENUI') + os.environ['ENABLE_GENUI'] = '0' + try: + with TestClient(create_app(search, generator, ConversationResolver(provider.json_call), provider)) as client: + turns = [(root_id, base['user_input']), (root_id + '_FOLLOWUP', followup)] + root_status = None + for qid, question in turns: + started = time.perf_counter() + with trace_span('Workflow / ' + qid, inputs={'question': question}) as run: + result = stream_result(client.post('/ask', json={'question': question})) + if root_status is None: + root_status = result['status'] + sources = result.get('sources', []) + case = {**base, 'question_id': qid, 'user_input': question, + 'status': result['status'], 'response': result['answer'], 'citations': result['citations'], + 'sources': sources, 'retrieved_contexts': [json.dumps(s, ensure_ascii=False) for s in sources], + 'evaluation_layer': 'workflow', 'langsmith_run_id': str(run.id) if run else None, + 'generation': {'seconds': time.perf_counter() - started, 'trace': list(generator.trace), + 'provider_calls': result['metrics'].get('provider_calls', []), + 'http_attempts': result['metrics'].get('http_attempts', []), + 'error_type': result['metrics'].get('error_category')}} + if qid.endswith('_FOLLOWUP'): + case['user_input'] += '\nConversation question: ' + base['user_input'] + if run: + run.end(outputs={'status': case['status'], 'response': case['response']}) + save(output / (qid + '.json'), case) + repeated = stream_result(client.post('/ask', json={'question': base['user_input']})) + client.post('/session/end') + save(output / 'contracts.json', {'turns': 2, 'same_session': True, + 'cache_check_applicable': root_status == 'answered', + 'repeat_answer_cache_hit': repeated['metrics'].get('answer_cache_hit', False), + 'session_cleared': 'techqa_session' not in client.cookies}) + finally: + search.close() + if old_ui is None: + os.environ.pop('ENABLE_GENUI', None) + else: + os.environ['ENABLE_GENUI'] = old_ui + save(output / 'manifest.json', {**manifest, 'question_ids': [q for q, _ in turns], + 'evaluation_layer': 'workflow', 'scope': 'API follow-up turns and session/cache contracts; canonical text'}) + + +if __name__ == '__main__': + load_dotenv(ROOT / '.env') + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--pipeline', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--component-ids', nargs='+', required=True) + parser.add_argument('--workflow-root', required=True) + parser.add_argument('--followup', required=True) + args = parser.parse_args() + collect_components(args.pipeline, args.output / 'component', args.component_ids) + collect_workflow(args.pipeline, args.output / 'workflow', args.workflow_root, args.followup) diff --git a/scripts/eval_ragas.py b/scripts/eval_ragas.py index 60e2e7f..6ec6492 100644 --- a/scripts/eval_ragas.py +++ b/scripts/eval_ragas.py @@ -104,7 +104,15 @@ async def cached_judgment(generate, settings, prompt, response_model, directory, 'seconds': time.perf_counter() - started, 'http_calls': []}) return result try: - result = await generate(prompt, response_model) + from rag_app.tracing import trace_span + with trace_span('RAGAS / Judge', inputs={'prompt': prompt}, run_type='llm', + metadata={'ls_provider': settings.get('provider'), 'ls_model_name': settings.get('judge')} ) as span: + result = await generate(prompt, response_model) + if span: + calls = http_calls[offset:] + span.end(outputs={**result.model_dump(mode='json'), 'usage_metadata': { + 'input_tokens': sum((c.get('usage') or {}).get('prompt_tokens') or 0 for c in calls), + 'output_tokens': sum((c.get('usage') or {}).get('completion_tokens') or 0 for c in calls)}}) save(path, {'output': result.model_dump(mode='json')}) return result finally: @@ -123,9 +131,14 @@ async def evaluate_case(case, scorers, *, timeout=300, journal=None, budget=None results[name] = {'state': 'skipped', 'score': None, 'reason': 'no_context_or_no_released_answer'} continue try: - async with asyncio.timeout(timeout): - result = await scorer.ascore(**metric_inputs(name, case)) - value = float(result.value) + from rag_app.tracing import trace_span + with trace_span('RAGAS / ' + name, inputs={'question_id': case['question_id']}, + metadata={'layer': case.get('evaluation_layer', 'pipeline')}) as metric_run: + async with asyncio.timeout(timeout): + result = await scorer.ascore(**metric_inputs(name, case)) + value = float(result.value) + if metric_run: + metric_run.end(outputs={'score': value if math.isfinite(value) else None}) results[name] = {'state': 'success' if math.isfinite(value) else 'undefined', 'score': value if math.isfinite(value) else None} except Exception as exc: @@ -154,7 +167,7 @@ def collect(args): from dotenv import load_dotenv load_dotenv(ROOT / ".env") from rag_app.retriever import Retriever - from rag_app.generation import AnswerGenerator, GroqREST, GenerationError + from rag_app.generation import AnswerGenerator, configured_provider, GenerationError split_path = ROOT / 'data/splits/techqa_v1.json' split = json.loads(split_path.read_text()) base = RAW / 'TechQA/training_and_dev' @@ -177,7 +190,7 @@ def collect(args): def bounded_post(*args, **kwargs): budget.consume() return httpx.post(*args, **kwargs) - provider = GroqREST(post=bounded_post) + provider = configured_provider(post=bounded_post) generator = AnswerGenerator(provider.json_call) identity = {'split_sha256': sha256(split_path), 'source_hashes': split['source_hashes'], 'index': retriever.index_identity(), 'model': provider.model, @@ -207,7 +220,7 @@ def bounded_post(*args, **kwargs): generator.trace.clear() question = q['QUESTION_TITLE'] + '\n' + q['QUESTION_TEXT'] print(q['QUESTION_ID'] + ': retrieval → writer → verifier', flush=True) - with trace_span('TechQA evaluation', inputs={'question': question}, metadata={'question_id': q['QUESTION_ID'], 'evaluation': True, 'split': 'training_tuning', 'model': provider.model, 'index': identity['index']}) as run: + with trace_span('Pipeline / Evaluation / ' + q['QUESTION_ID'], inputs={'question': question}, metadata={'question_id': q['QUESTION_ID'], 'evaluation': True, 'split': 'training_tuning', 'model': provider.model, 'index': identity['index']}) as run: started = time.perf_counter() with trace_span('Retrieve and select evidence', inputs={'question': question}) as retrieval_run: retrieved = retriever.retrieve(question) @@ -270,14 +283,22 @@ async def run(args): not c.get('reference', '').strip() or not c.get('reference_provenance', {}).get('review_notes', '').strip() for c in cases): raise ValueError('Feedback evaluation requires approved references and review notes') - key = os.getenv('GROQ_API_KEY') or os.getenv('GROK_API_KEY') + provider_name = getattr(args, 'provider', 'groq') + key = os.getenv('TOKENHARBOR_API_KEY') if provider_name == 'tokenharbor' else os.getenv('GROQ_API_KEY') or os.getenv('GROK_API_KEY') + base_url = os.getenv('TOKENHARBOR_BASE_URL', 'https://tokenharbor.ai/v1') if provider_name == 'tokenharbor' else 'https://api.groq.com/openai/v1' + judge_model = args.judge or (os.getenv('TOKENHARBOR_MODEL') if provider_name == 'tokenharbor' else 'openai/gpt-oss-120b') + output_limit = 3000 if provider_name == 'tokenharbor' else 2048 + ledger = None + if os.getenv('EVAL_BUDGET_FILE') and provider_name == 'tokenharbor': + from scripts.eval_budget import EvaluationBudget + ledger = EvaluationBudget(os.environ['EVAL_BUDGET_FILE']) if not key and args.max_requests: - raise ValueError('Groq key missing') + raise ValueError('Judge provider key missing') retriever = Retriever() if 'answer_relevancy' in args.metrics else None budget = RequestBudget(args.max_requests) - settings = {'provider': 'groq', 'judge': args.judge, 'temperature': 0, 'reasoning_effort': 'low', + settings = {'thinking': 'disabled' if provider_name == 'tokenharbor' else None, 'provider': provider_name, 'judge': judge_model, 'temperature': 0, 'reasoning_effort': 'low' if provider_name == 'groq' else None, 'transport': {'sdk_max_retries': 3, 'retry_after_buffer_seconds': 2, 'minimum_retry_seconds': 5}, - 'max_tokens': 2048, 'strictness': 3, 'timeout_per_metric': args.timeout, + 'max_tokens': output_limit, 'strictness': 3, 'timeout_per_metric': args.timeout, 'packages': {p: importlib.metadata.version(p) for p in (('ragas', 'openai', 'instructor', 'langchain-community') + (('sentence-transformers', 'torch', 'transformers', 'torchvision') if retriever else ()))}, @@ -290,6 +311,13 @@ async def run(args): async def request_hook(request): budget.consume() # Includes SDK retries; blocks before any request is sent. + if ledger: + from scripts.eval_budget import BudgetLimitError + try: + request.extensions['budget_ticket'] = ledger.begin(json.loads(request.content)) + except BudgetLimitError: + budget.exhausted = True + raise BudgetExceeded('Evaluation dollar budget reached') from None request.extensions['eval_started'] = time.perf_counter() request.extensions['eval_call_index'] = len(http_calls) http_calls.append({'status': None, 'usage': {}, 'seconds': None}) @@ -303,19 +331,22 @@ async def response_hook(response): http_calls[response.request.extensions['eval_call_index']].update( {'status': response.status_code, 'usage': usage, 'seconds': time.perf_counter() - response.request.extensions['eval_started']}) + if ledger: + ledger.finish(response.request.extensions['budget_ticket'], usage, response.status_code) if response.status_code == 429: delay = retry_delay(response.headers) http_calls[response.request.extensions['eval_call_index']]['retry_after_seconds'] = delay if delay >= args.timeout: budget.exhausted = True raise BudgetExceeded('Provider quota reset exceeds metric deadline') - print(f' Groq quota: waiting about {delay:g}s before retry', flush=True) + print(f' Judge quota: waiting about {delay:g}s before retry', flush=True) await asyncio.sleep(delay) response.headers['retry-after'] = '0' - client = AsyncOpenAI(api_key=key or 'cache-only-no-key', base_url='https://api.groq.com/openai/v1', timeout=60, max_retries=3, + client = AsyncOpenAI(api_key=key or 'cache-only-no-key', base_url=base_url, timeout=60, max_retries=3, http_client=httpx.AsyncClient(event_hooks={'request': [request_hook], 'response': [response_hook]})) - backend = llm_factory(args.judge, client=client, temperature=0, max_tokens=2048, reasoning_effort='low') + backend_options = {'reasoning_effort': 'low'} if provider_name == 'groq' else {'extra_body': {'thinking': {'type': 'disabled'}}} + backend = llm_factory(judge_model, client=client, temperature=0, max_tokens=output_limit, **backend_options) class CachedJudge(InstructorBaseRagasLLM): def __init__(self, draws_per_prompt=1): @@ -391,7 +422,8 @@ def main(): r.add_argument('--limit', type=int, default=3) r.add_argument('--max-requests', type=int, default=12, help='Maximum HTTP attempts including retries; 0 is cache-only') r.add_argument('--cache', type=Path, default=ROOT / 'data/ragas/judge_cache') - r.add_argument('--judge', default='openai/gpt-oss-120b') + r.add_argument('--provider', choices=['groq', 'tokenharbor'], default=os.getenv('RAG_LLM_PROVIDER', 'groq')) + r.add_argument('--judge') r.add_argument('--timeout', type=float, default=300) args = parser.parse_args() if args.command == 'collect': diff --git a/scripts/eval_suite.py b/scripts/eval_suite.py index ff86f53..a892479 100644 --- a/scripts/eval_suite.py +++ b/scripts/eval_suite.py @@ -85,7 +85,8 @@ def build_report(cases, ragas=None): if m['state'] == 'success' and (not isinstance(value, (int, float)) or not math.isfinite(value)): raise ValueError('Successful judge score must be finite') metrics[name] = {'state': m['state'], 'score': value, - 'needs_review': any(f['metric'] == name for f in flags)} + 'needs_review': any(f['metric'] == name for f in flags), + 'reason': m.get('reason') or m.get('error_type')} rows.append({'question_id': c['question_id'], 'outcome': c['status'], 'selected_gold_document_hit': hit, 'selected_passage_count': len(c['sources']), 'citation_contract': citation, 'ragas': metrics, 'judge_integrity_flags': flags}) diff --git a/scripts/publish_langsmith.py b/scripts/publish_langsmith.py index b253455..d8c61ed 100644 --- a/scripts/publish_langsmith.py +++ b/scripts/publish_langsmith.py @@ -1,136 +1,120 @@ -"""Publish saved evaluation outcomes to LangSmith without new model calls.""" - +"""Import saved evaluation measurements into LangSmith without new model calls.""" import argparse +from datetime import datetime, timedelta, timezone import hashlib import json +import math from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4, uuid5 from dotenv import load_dotenv +from langsmith.run_trees import RunTree from langsmith.utils import LangSmithNotFoundError from rag_app.tracing import get_client from scripts.eval_suite import build_report, load_cases -def publish(directory, client, ragas=None): +def publish(directory, client, ragas=None, *, dataset_name=None, experiment_prefix='techqa-saved-pipeline'): cases = load_cases(directory) report = build_report(cases, ragas) - identity_file = directory / "manifest.json" if directory.is_dir() else directory - identity = report["cohort_id"] - artifact_hash = hashlib.sha256(identity_file.read_bytes()).hexdigest() - feedback_cohort = all(c["question_id"].startswith("FEEDBACK_") for c in cases) - name = ("techqa-feedback-" if feedback_cohort else "techqa-training-") + identity[ - :12 - ] + identity_file = directory / 'manifest.json' if directory.is_dir() else directory + identity = report['cohort_id'] + feedback_cohort = all(c['question_id'].startswith('FEEDBACK_') for c in cases) + name = dataset_name or (('techqa-feedback-' if feedback_cohort else 'techqa-training-') + identity[:12]) try: dataset = client.read_dataset(dataset_name=name) except LangSmithNotFoundError: - dataset = client.create_dataset( - name, - description=( - "Reviewed feedback regression cases" - if feedback_cohort - else "Fixed training tuning sample" - ) - + ". References are used only for scoring.", - ) - existing = { - example.inputs["question_id"] - for example in client.list_examples(dataset_id=dataset.id) - } - examples = [ - { - "inputs": {"question_id": c["question_id"], "question": c["user_input"]}, - "outputs": {"reference": c["reference"]}, - } - for c in cases - if c["question_id"] not in existing - ] - if examples: - client.create_examples(dataset_id=dataset.id, examples=examples) - lookup = {c["question_id"]: c for c in cases} - rows = {r["question_id"]: r for r in report["cases"]} - - def saved_answer(inputs): - case = lookup[inputs["question_id"]] - return { - "question_id": case["question_id"], - "response": case["response"], - "status": case["status"], - "sources": case["sources"], - "original_trace_id": case.get("langsmith_run_id"), - "model_calls": [ - {key: call.get(key) for key in ("provider", "model")} - for call in case.get("generation", {}).get("provider_calls", []) - ], - } - - def outcome(run, example): - return {"key": "response_status", "value": run.outputs["status"]} - - def selected_document(run, example): - return { - "key": "selected_gold_document_hit", - "score": rows[example.inputs["question_id"]]["selected_gold_document_hit"], - } - - def citations(run, example): - check = rows[example.inputs["question_id"]]["citation_contract"] - return { - "key": "citation_contract", - "score": int(check["pass"]) if check["state"] == "checked" else None, - "comment": "Citation format and source IDs only. Semantic support is evaluated separately.", - } - - evaluators = [outcome, selected_document, citations] - if ragas is not None: - - def semantic_scores(run, example): - metrics = rows[example.inputs["question_id"]]["ragas"] - return { - "results": [ - { - "key": "ragas_" + key, - "score": value["score"] - if value["state"] == "success" and not value["needs_review"] - else None, - "comment": "Judge review required" - if value["needs_review"] - else value["state"], - } - for key, value in metrics.items() - ] - } - - evaluators.append(semantic_scores) - - results = client.evaluate( - saved_answer, - data=name, - evaluators=evaluators, - experiment_prefix="techqa-saved-pipeline", - max_concurrency=1, - description="Saved canonical pipeline results replayed for inspection. No new inference or semantic judge calls.", - metadata={ - "cohort_id": identity, - "collection_artifact_sha256": artifact_hash, - "mode": "saved_result_replay", - "split": "reviewed_user_feedback" if feedback_cohort else "training_tuning", - }, - ) + dataset = client.create_dataset(name, description='TechQA evaluation. References are used for scoring.') + existing = {e.inputs['question_id']: e for e in client.list_examples(dataset_id=dataset.id)} + for case in cases: + example = existing.get(case['question_id']) + if example and (example.inputs.get('question') != case['user_input'] or + (example.outputs or {}).get('reference') != case['reference']): + raise ValueError('Existing dataset question/reference differs; use a versioned dataset') + missing = [{'inputs': {'question_id': c['question_id'], 'question': c['user_input']}, + 'outputs': {'reference': c['reference']}} for c in cases if c['question_id'] not in existing] + if missing: + client.create_examples(dataset_id=dataset.id, examples=missing) + existing = {e.inputs['question_id']: e for e in client.list_examples(dataset_id=dataset.id)} + if not all(c['question_id'] in existing for c in cases): + raise ValueError('Dataset example creation is incomplete') + experiment_name = experiment_prefix + '-' + str(uuid4())[:8] + metadata = {'cohort_id': identity, 'mode': 'recorded_result_import', + 'evaluation_layer': cases[0].get('evaluation_layer', 'pipeline'), + 'collection_artifact_sha256': hashlib.sha256(identity_file.read_bytes()).hexdigest(), + 'split': 'reviewed_user_feedback' if feedback_cohort else 'training_tuning'} + project = client.create_project(experiment_name, reference_dataset_id=dataset.id, + num_examples=len(cases), num_repetitions=1, metadata=metadata, + description='Saved measurements imported with original duration, usage and failure status.') + rows = {r['question_id']: r for r in report['cases']} + for case in cases: + qid = case['question_id'] + row = rows[qid] + gen = case.get('generation', {}) + end = datetime.now(timezone.utc) + seconds = gen.get('seconds') + measured = isinstance(seconds, (int, float)) and math.isfinite(seconds) and seconds >= 0 + calls = gen.get('provider_calls', []) + outputs = {'question_id': qid, 'response': case['response'], 'status': case['status'], + 'sources': case['sources'], 'original_trace_id': case.get('langsmith_run_id'), + 'evaluation_layer': case.get('evaluation_layer', 'pipeline'), + 'model_calls': [{k: call.get(k) for k in ('provider', 'model')} for call in calls]} + run = RunTree(name=case.get('evaluation_layer', 'pipeline').title() + ' / ' + qid, run_type='chain', client=client, + project_name=experiment_name, reference_example_id=existing[qid].id, + inputs={'question_id': qid, 'question': case['user_input']}, outputs=outputs, + start_time=end - timedelta(seconds=seconds) if measured else end, end_time=end, + error=(gen.get('error_type') or 'generation_error') if case['status'] == 'generation_error' else None, + extra={'metadata': {**metadata, 'timestamp_basis': 'import_time', + 'duration_source': 'saved_generation_seconds' if measured else 'unavailable'}}) + run.post() + if calls: + usage = {'input_tokens': sum(c.get('prompt_tokens') or 0 for c in calls), + 'output_tokens': sum(c.get('output_tokens', c.get('completion_tokens')) or 0 for c in calls), + 'total_tokens': sum(c.get('total_tokens') or 0 for c in calls)} + child = run.create_child('Pipeline / Recorded model usage', run_type='llm', + inputs={'messages': []}, outputs={'usage_metadata': usage}, start_time=end, end_time=end, + extra={'metadata': {'ls_provider': calls[0].get('provider'), + 'ls_model_name': calls[0].get('model'), 'aggregate_usage': True}}) + child.post() + feedback = [ + {'key': 'response_status', 'value': case['status']}, + {'key': 'selected_gold_document_hit', 'score': row['selected_gold_document_hit']}, + {'key': 'pipeline_error', 'score': int(case['status'] == 'generation_error')}, + {'key': 'citation_contract', 'score': int(row['citation_contract']['pass']) + if row['citation_contract']['state'] == 'checked' else None, + 'comment': 'Citation format and source IDs.'}, + {'key': 'recorded_latency_seconds', 'score': seconds if measured else None}, + {'key': 'recorded_tokens', 'score': sum(c.get('total_tokens') or 0 for c in calls)}, + {'key': 'cost_status', 'value': 'free_route' if calls and all(str(c.get('model', '')).endswith(':free') for c in calls) else 'not_recorded'}, + ] + for metric, value in row['ragas'].items(): + scored = value['state'] == 'success' and not value['needs_review'] + reason = 'Judge review required' if value['needs_review'] else value.get('reason') or value['state'] + feedback.extend([ + {'key': 'ragas_' + metric, 'score': value['score'] if scored else None, 'comment': reason}, + {'key': 'ragas_state_' + metric, 'value': 'needs_review' if value['needs_review'] else value['state'], 'comment': reason}, + {'key': 'ragas_scored_' + metric, 'score': int(scored)}, + {'key': 'ragas_eligible_' + metric, 'score': int(metric.startswith('context_') or case['status'] == 'answered')}, + ]) + for item in feedback: + client.create_feedback(run.id, **item, session_id=project.id, trace_id=run.trace_id, + feedback_id=uuid5(run.id, item['key']), source_info={'imported_measurement': True}) client.flush() - return results + client.update_project(project.id, end_time=datetime.now(timezone.utc)) + url = dataset.url + '/compare?selectedSessions=' + str(project.id) if dataset.url else str(project.id) + print('Published experiment', experiment_name, '\n' + url, flush=True) + return SimpleNamespace(experiment_id=str(project.id), experiment_name=experiment_name, url=url) -if __name__ == "__main__": - load_dotenv(Path(__file__).resolve().parents[1] / ".env") +if __name__ == '__main__': + load_dotenv(Path(__file__).resolve().parents[1] / '.env') parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--cases", type=Path, required=True) - parser.add_argument( - "--ragas", - type=Path, - help="Optional matching saved RAGAS report; no new judge calls", - ) + parser.add_argument('--cases', type=Path, required=True) + parser.add_argument('--ragas', type=Path) + parser.add_argument('--dataset-name') + parser.add_argument('--experiment-prefix', default='techqa-saved-pipeline') args = parser.parse_args() - scores = json.loads(args.ragas.read_text()) if args.ragas else None - publish(args.cases, get_client(), scores) + publish(args.cases, get_client(), json.loads(args.ragas.read_text()) if args.ragas else None, + dataset_name=args.dataset_name, experiment_prefix=args.experiment_prefix) diff --git a/tests/test_langsmith_export.py b/tests/test_langsmith_export.py index ebbed3a..3db5e72 100644 --- a/tests/test_langsmith_export.py +++ b/tests/test_langsmith_export.py @@ -1,72 +1,78 @@ +import json from pathlib import Path from types import SimpleNamespace -from unittest.mock import Mock +from unittest.mock import Mock, patch +from uuid import uuid4 from scripts.publish_langsmith import publish -def test_saved_experiment_exports_contracts_without_model_calls(): +def client_for(cases): client = Mock() - client.read_dataset.return_value = SimpleNamespace(id="dataset") - client.list_examples.return_value = [] + client.read_dataset.return_value = SimpleNamespace(id=uuid4(), url='https://example.com/dataset') + client.list_examples.return_value = [SimpleNamespace(id=uuid4(), inputs={'question_id': c['question_id'], 'question': c['user_input']}, outputs={'reference': c['reference']}) for c in cases] + client.create_project.return_value = SimpleNamespace(id=uuid4()) + return client - def evaluate(target, **kwargs): - example = client.create_examples.call_args.kwargs["examples"][0] - inputs = example["inputs"] - output = target(inputs) - run = SimpleNamespace(outputs=output) - reference = SimpleNamespace(inputs=inputs) - feedback = [scorer(run, reference) for scorer in kwargs["evaluators"]] - assert output["question_id"] == inputs["question_id"] - assert [f["key"] for f in feedback] == [ - "response_status", - "selected_gold_document_hit", - "citation_contract", - ] - assert kwargs["max_concurrency"] == 1 - assert kwargs["metadata"]["mode"] == "saved_result_replay" - return "experiment" - client.evaluate.side_effect = evaluate - assert ( - publish(Path("evaluations/fixtures/offline_demo.json"), client) == "experiment" - ) - client.flush.assert_called_once() +def test_saved_experiment_imports_measurements_before_run_creation(tmp_path): + payload = json.loads(Path('evaluations/fixtures/offline_demo.json').read_text()) + case = payload['cases'][0] + case['status'] = 'generation_error' + case['generation'] = {'seconds': 12, 'error_type': 'http_429', 'provider_calls': [ + {'provider': 'tokenharbor', 'model': 'deepseek-v4.1-flash:free', + 'prompt_tokens': 10, 'completion_tokens': 4, 'total_tokens': 14, 'private': 'omit'}]} + path = tmp_path / 'cases.json' + path.write_text(json.dumps(payload)) + client = client_for(payload['cases']) + with patch('scripts.publish_langsmith.RunTree') as tree: + tree.return_value.id = uuid4() + publish(path, client) + root = tree.call_args_list[0].kwargs + assert (root['end_time'] - root['start_time']).total_seconds() == 12 + assert root['error'] == 'http_429' + assert root['outputs']['model_calls'] == [{'provider': 'tokenharbor', 'model': 'deepseek-v4.1-flash:free'}] + assert client.evaluate.call_count == 0 + assert client.update_run.call_count == 0 + assert any(c.kwargs.get('key') == 'pipeline_error' and c.kwargs.get('score') == 1 + for c in client.create_feedback.call_args_list) def test_dataset_identity_survives_changed_answers(tmp_path): - import json - - source = Path("evaluations/fixtures/offline_demo.json") - payload = json.loads(source.read_text()) - client = Mock() - client.read_dataset.return_value = SimpleNamespace(id="dataset") - client.list_examples.return_value = [] - publish(source, client) - original = client.read_dataset.call_args.kwargs["dataset_name"] - payload["cases"][0]["response"] = "A changed output. [S1]" - changed = tmp_path / "changed.json" - changed.write_text(json.dumps(payload)) - publish(changed, client) - assert client.read_dataset.call_args.kwargs["dataset_name"] == original + payload = json.loads(Path('evaluations/fixtures/offline_demo.json').read_text()) + client = client_for(payload['cases']) + with patch('scripts.publish_langsmith.RunTree') as tree: + tree.return_value.id = uuid4() + publish(Path('evaluations/fixtures/offline_demo.json'), client) + name = client.read_dataset.call_args.kwargs['dataset_name'] + payload['cases'][0]['response'] = 'Changed. [S1]' + path = tmp_path / 'changed.json' + path.write_text(json.dumps(payload)) + publish(path, client) + assert client.read_dataset.call_args.kwargs['dataset_name'] == name -def test_model_identity_is_visible_without_exporting_extra_call_fields(tmp_path): - import json - - payload = json.loads(Path("evaluations/fixtures/offline_demo.json").read_text()) - payload["cases"][0]["generation"] = {"provider_calls": [{ - "provider": "tokenharbor", "model": "deepseek-v4.1-flash:free", "private": "omit-me"}]} - path = tmp_path / "cases.json" +def test_unscored_metric_has_explicit_state_and_coverage(tmp_path): + payload = json.loads(Path('evaluations/fixtures/offline_demo.json').read_text()) + cases = payload['cases'] + path = tmp_path / 'cases.json' path.write_text(json.dumps(payload)) - client = Mock() - client.read_dataset.return_value = SimpleNamespace(id="dataset") - client.list_examples.return_value = [] + ragas = {'results': [{'question_id': c['question_id'], 'status': c['status'], + 'metrics': {'answer_relevancy': {'state': 'skipped', 'score': None, 'reason': 'quota'}}} for c in cases], + 'settings': {}, 'http_calls': []} + client = client_for(cases) + with patch('scripts.publish_langsmith.RunTree') as tree: + tree.return_value.id = uuid4() + publish(path, client, ragas) + fields = [c.kwargs for c in client.create_feedback.call_args_list] + assert any(x.get('key') == 'ragas_state_answer_relevancy' and x.get('value') == 'skipped' for x in fields) + assert any(x.get('key') == 'ragas_scored_answer_relevancy' and x.get('score') == 0 for x in fields) - def evaluate(target, **kwargs): - example = client.create_examples.call_args.kwargs["examples"][0] - assert target(example["inputs"])["model_calls"] == [{ - "provider": "tokenharbor", "model": "deepseek-v4.1-flash:free"}] - client.evaluate.side_effect = evaluate - publish(path, client) +def test_custom_dataset_rejects_changed_reference(): + import pytest + payload = json.loads(Path('evaluations/fixtures/offline_demo.json').read_text()) + client = client_for(payload['cases']) + client.list_examples.return_value[0].outputs['reference'] = 'Different reference' + with pytest.raises(ValueError, match='question/reference differs'): + publish(Path('evaluations/fixtures/offline_demo.json'), client, dataset_name='existing') diff --git a/tests/test_ragas_eval.py b/tests/test_ragas_eval.py index b46ceac..6cf0890 100644 --- a/tests/test_ragas_eval.py +++ b/tests/test_ragas_eval.py @@ -211,3 +211,91 @@ def test_unreviewed_feedback_is_rejected_before_judging(tmp_path): (tmp_path / (qid + '.json')).write_text(json.dumps({'question_id': qid, 'review_status': 'pending', 'reference': 'Unreviewed claim'})) with pytest.raises(ValueError, match='approved references'): asyncio.run(run(SimpleNamespace(cases=tmp_path, limit=1))) + + +def test_shared_dollar_budget_reserves_then_settles_and_reuses_cache_discount(tmp_path): + import json + from scripts.eval_budget import EvaluationBudget, BudgetLimitError + ledger = EvaluationBudget(tmp_path / "budget.json", limit=0.02) + body = {"messages": [{"role": "user", "content": "test"}], "max_tokens": 100} + ticket = ledger.begin(body) + ledger.finish(ticket, {"prompt_tokens": 100, "completion_tokens": 10, + "prompt_tokens_details": {"cached_tokens": 90}}, 200) + data = json.loads((tmp_path / "budget.json").read_text()) + assert data["requests"][ticket]["cached_input_tokens"] == 90 + assert data["requests"][ticket]["value_usd"] < 0.00002 + with pytest.raises(BudgetLimitError): + ledger.begin({"messages": [], "max_tokens": 100000}) + + +def test_workflow_stream_parser_keeps_generation_errors(): + from scripts.eval_layers import stream_result + response = SimpleNamespace(text='event: error\ndata: {"metrics":{"error_category":"http_429"}}\n\n') + result = stream_result(response) + assert result["status"] == "generation_error" + assert result["metrics"]["error_category"] == "http_429" + + +def test_public_workflow_collector_uses_api_memory_and_cache(tmp_path, monkeypatch): + import json + import rag_app.api as api + import scripts.eval_layers as layers + payload = json.loads(Path('evaluations/fixtures/offline_demo.json').read_text())['cases'][0] + pipeline = tmp_path / 'pipeline' + pipeline.mkdir() + (pipeline / 'manifest.json').write_text(json.dumps({'question_ids': [payload['question_id']]})) + (pipeline / (payload['question_id'] + '.json')).write_text(json.dumps(payload)) + monkeypatch.setattr(api, 'ROOT', tmp_path) + + class Search: + def index_identity(self): return 'controlled' + def retrieve(self, question): return {'sources': payload['sources'], 'metrics': {}} + def close(self): pass + + def reply(prompt, schema): + if 'dependent' in schema['properties']: + data = json.loads(prompt[1]['content']) + dependent = data['original_question'] == 'What follows that edit?' + return {'dependent': dependent, 'question': 'Beacon restart?' if dependent else data['original_question'], + 'needs_clarification': False} + if 'verdict' in schema['properties']: + return {'verdict': 'complete', 'supported': True, 'complete': True, + 'missing_steps': [], 'version_conflicts': [], 'findings': [], 'unsupported_claims': []} + return {'status': 'answered', 'answer': 'Restart the connector. [S1]', 'citations': ['S1']} + + provider = SimpleNamespace(json_call=reply, usage=[], attempts=[], model='controlled', provider='controlled', max_tokens=100) + monkeypatch.setattr(layers, 'configured_provider', lambda: provider) + monkeypatch.setattr(layers, 'Retriever', Search) + output = tmp_path / 'workflow' + layers.collect_workflow(pipeline, output, payload['question_id'], 'What follows that edit?') + checks = json.loads((output / 'contracts.json').read_text()) + assert checks['repeat_answer_cache_hit'] and checks['session_cleared'] + assert json.loads((output / (payload['question_id'] + '_FOLLOWUP.json')).read_text())['status'] == 'answered' + + +def test_component_collector_supplies_gold_article_instead_of_search_context(tmp_path, monkeypatch): + import json + import scripts.eval_layers as layers + c = json.loads(Path('evaluations/fixtures/offline_demo.json').read_text())['cases'][0] + pipeline = tmp_path / 'pipeline' + pipeline.mkdir() + (pipeline / 'manifest.json').write_text(json.dumps({'question_ids': [c['question_id']]})) + (pipeline / (c['question_id'] + '.json')).write_text(json.dumps(c)) + raw = tmp_path / 'raw' + folder = raw / 'TechQA/training_and_dev' + folder.mkdir(parents=True) + (folder / 'training_dev_technotes.json').write_text(json.dumps({'beacon': {'text': 'GOLD_ARTICLE'}})) + seen = [] + class Generator: + trace = [] + def __init__(self, callback): pass + def generate(self, question, sources): + seen.append(sources) + return {'status':'answered','answer':'Approved [S1]','citations':['S1']} + monkeypatch.setattr(layers, 'RAW', raw) + monkeypatch.setattr(layers, 'AnswerGenerator', Generator) + monkeypatch.setattr(layers, 'configured_provider', lambda: SimpleNamespace(usage=[], attempts=[], json_call=None)) + output = tmp_path / 'component' + layers.collect_components(pipeline, output, [c['question_id']]) + assert seen[0][0]['text'] == 'GOLD_ARTICLE' + assert json.loads((output / (c['question_id'] + '.json')).read_text())['evaluation_layer'] == 'component' From 6c5a58bcd7e56d45d611a4bf1aeb74a04ba87898 Mon Sep 17 00:00:00 2001 From: shipitdev Date: Wed, 11 Nov 2026 12:24:59 +0530 Subject: [PATCH 3/4] Record ten-question coverage and reviewed failure findings --- README.md | 56 ++- evaluations/overview.svg | 18 +- evaluations/results/layered_audit_10.json | 565 ++++++++++++++++++++++ 3 files changed, 623 insertions(+), 16 deletions(-) create mode 100644 evaluations/results/layered_audit_10.json diff --git a/README.md b/README.md index ba9dba6..af2a475 100644 --- a/README.md +++ b/README.md @@ -62,15 +62,57 @@ Approval and review notes are required for feedback-based RAGAS evaluation. Use ![Evaluation layers: components, pipeline, application, regression and planned monitoring](evaluations/overview.svg) -**Evaluate the parts, then their handoff, then the delivered answer.** RAGAS supplies semantic metrics; our Python checks cover source/citation contracts, recorded usage, judge integrity and paired regressions. Unit tests exercise failure handling with controlled fixtures. No single score is called “RAG accuracy.” +The suite records retrieval quality, generated-answer quality and API workflow behavior. RAGAS supplies semantic judgments. Python checks source/citation contracts, session/cache behavior, recorded usage and paired regressions. -| Layer | What we check | Implementation and status | +| Layer | What we check | Implemented measurements | |---|---|---| -| Components | Retrieval Recall@K, MRR and nDCG against known article labels; writer/verifier behavior with supplied evidence | [Rank metrics](scripts/baseline.py), [retrieval tests](tests/test_app_retriever.py), [generator contract fixtures](tests/test_app_generation.py). Controlled fixtures are not a generator quality benchmark. | -| Pipeline | Selected context precision/recall, answer faithfulness and relevancy | [RAGAS runner](scripts/eval_ragas.py), measured on the pilot below. Reference text stays outside inference. | -| Application | Reference-based correctness and completeness; citation validity; withholding and layout contracts; recorded timing/tokens/cache use | Correctness = RAGAS `FactualCorrectness(mode="f1")`; completeness = its `mode="recall"` proxy. Both are opt-in and have sampled results below. [Free saved-case checks](scripts/eval_suite.py), [API contracts](tests/test_app_api.py), [layout checks](frontend/tests/genui-contract.test.mjs) are implemented. | -| Regression | Re-run a fixed case set, then inspect per-case differences | [Comparison CLI](scripts/eval_suite.py) rejects changed cohorts/references. Semantic comparisons also require matching judge settings and exclude scores flagged for review. | -| Online | User feedback, reviewed regression cases, sampled judging and drift | Helpful/unhelpful votes and optional comments save the run for review and sync to LangSmith. Reviewed references can be exported for RAGAS. Sampled live judging and drift monitoring remain planned. | +| Component | Retriever ranking and generator with supplied original article evidence | Recall@K, MRR and nDCG in the retrieval benchmark. Two supplied-evidence generator cases scored with RAGAS faithfulness, relevancy and reference metrics. | +| Pipeline | Actual retrieved context through writing and verification | Ten fixed training questions. All six metric types scored for eligible cases, with explicit skipped/error states and denominators. | +| Workflow | API conversation turns, follow-up resolution and exact answer caching | Two real API turns scored with RAGAS. Same-session follow-up and repeated-question cache checks also run. [Layer collector](scripts/eval_layers.py). | +| Regression | Matched cases and references | [Comparison CLI](scripts/eval_suite.py) checks cohort and judge identity. The saved v5/v6 outputs were rescored with the same DeepSeek judge. | +| Online | Feedback and run inspection | LangSmith traces, user ratings and reviewed regression exports. Live sampled judging and drift monitoring are planned. | + +### Ten-question evaluation audit + +The [redacted audit](evaluations/results/layered_audit_10.json) uses the first ten IDs from the fixed training sample. Today's writer, verifier and judge use TokenHarbor `deepseek-v4.1-flash:free`. Historical v5/v6 outputs retain their original Groq model attribution. The judge passed supported/unsupported faithfulness sanity checks at 1.0/0.0. + +| Metric | DeepSeek pipeline mean | Scored / eligible | +|---|---:|---:| +| Context recall | 100.0% | 10/10 | +| Context precision | 83.9% | 10/10 | +| Faithfulness | 96.6% | 7/7 | +| Answer relevancy | 88.5% | 7/7 | +| Reference factual correctness | 40.9% | 7/7 | +| Reference-fact recall | 78.6% | 7/7 | + +The pipeline released seven answers, withheld two and requested clarification once. It had zero generation errors after the DeepSeek adapter correction. Supplied-evidence component faithfulness was 100% over two cases. Workflow faithfulness was 100% and answer relevancy 94.2% over two turns. Repeated-question caching passed. These small training diagnostics require human review before a release decision. + +On the four shared released historical cases, the same DeepSeek judge scored reference correctness **50.0% → 37.8%** from v5 to v6. Q117's broader version assertions contribute to the drop. Q548 also exposes bundled-claim and overly literal error-string judgments. Q376 mentions details and asks where to find them; its reference expects a vulnerability explanation. A reviewed task rubric should resolve that scope difference. The new verifier also incorrectly rejects qualified version differences and demands unrequested remediation. These findings guide the next change. + +The legacy `answer_completeness` field uses RAGAS `FactualCorrectness(mode="recall")`. It measures reference agreement. Required-task completeness needs explicit review of the question and requested steps. Claim decomposition and coverage settings affect reference scores. [RAGAS metric definitions](https://docs.ragas.io/en/stable/concepts/metrics/available_metrics/factual_correctness/) + +Each dashboard metric has `ragas_state_*`, `ragas_scored_*` and `ragas_eligible_*` feedback. Skipped answers and failed judgments have reasons. Historical imports include saved duration, model usage and original generation errors. Imported wall clocks anchor those durations at import time. Aggregate usage spans carry recorded token totals. [LangSmith usage instrumentation](https://docs.langchain.com/langsmith/log-llm-trace) + +The shared [budget guard](scripts/eval_budget.py) tracks cached inputs, failed preflights and unresolved reservations across generation and judging. The audit stayed at **$0.295/$0.50** of standard-route token value. The selected free route bills zero and consumes its free allowance. Trace names use `Pipeline / ...`, `Component / ...`, `Workflow / ...` and `RAGAS / ...`. + +Run the layer collector after collecting the ten pipeline cases. Use the RAGAS runner on each output directory. + +```bash +python -m scripts.eval_layers \ + --pipeline data/ragas/ten_questions \ + --output data/ragas/layers \ + --component-ids TRAIN_Q117 TRAIN_Q548 \ + --workflow-root TRAIN_Q013 \ + --followup "Which version must I upgrade it to first?" + +python -m scripts.eval_ragas run \ + --provider tokenharbor --judge deepseek-v4.1-flash:free \ + --cases data/ragas/layers/component --limit 2 --max-requests 40 \ + --metrics faithfulness answer_relevancy factual_correctness answer_completeness +``` + +Set `EVAL_BUDGET_FILE` to a shared ledger path to enforce the $0.50 token-value budget across processes. Set `RAG_LLM_PROVIDER=tokenharbor` for DeepSeek generation. The serving environment and evaluation environment use their pinned requirements. + Style rubrics and broad toxicity/PII/jailbreak evaluation are also planned. Existing origin/session, withholding and generated-component checks cover specific application contracts; they are not a comprehensive safety benchmark. Optional semantic metrics compare against available references, which may omit valid alternative answers. diff --git a/evaluations/overview.svg b/evaluations/overview.svg index 41a2423..54e372f 100644 --- a/evaluations/overview.svg +++ b/evaluations/overview.svg @@ -5,7 +5,7 @@ TECHQA RAG / EVALUATION SUITE -Measure each handoff. +TechQA evaluation coverage RAGAS for semantic judgments. Python for contracts, operations and comparisons. SHIPPED @@ -21,8 +21,8 @@ Labeled rankings · programmatic checks Generator with supplied evidence -Writer / verifier / repair fixtures -Controlled contracts; quality benchmark pending +Writer / verifier / citation checks +RAGAS · 2 supplied-evidence cases 02 Pipeline — evaluate the real retrieved context @@ -33,7 +33,7 @@ Selected metrics · case limits · request caps · cached marking decisions -03 Application — answer quality, contracts and operating cost +03 Workflow — API turns, quality and operations @@ -41,16 +41,16 @@ OPT-IN Factual correctness (F1) -Completeness (recall mode) -Style / broad safety: planned +Reference-fact recall +Style and broad safety planned Contracts -Citations · withholding · layouts -Judge version-preservation flags +Follow-ups · cache · citations +Sessions · layouts · withholding Human review for flagged scores Operations Time · tokens · cache use HTTP attempts and quota errors -Dollar pricing not configured +Shared token-value budget guard Regression is a comparison, not another metric diff --git a/evaluations/results/layered_audit_10.json b/evaluations/results/layered_audit_10.json new file mode 100644 index 0000000..60f2159 --- /dev/null +++ b/evaluations/results/layered_audit_10.json @@ -0,0 +1,565 @@ +{ + "scope": "Fixed ten-question training diagnostic sample plus supplied-evidence and real API workflow checks.", + "models": { + "today_pipeline": "deepseek-v4.1-flash:free", + "judge": "deepseek-v4.1-flash:free", + "historical_v5_v6_writer": "openai/gpt-oss-120b" + }, + "judge_sanity_checks": { + "supported_faithfulness": 1.0, + "unsupported_faithfulness": 0.0 + }, + "layers": { + "pipeline": { + "cases": 10, + "outcomes": { + "answered": 7, + "insufficient_evidence": 2, + "needs_clarification": 1 + }, + "metrics": { + "context_recall": { + "success": 10, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + }, + "context_precision": { + "success": 10, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.8385714285112444, + "flagged_for_review": 0 + }, + "faithfulness": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.9659863945578231, + "flagged_for_review": 0 + }, + "answer_relevancy": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.8849442137920638, + "flagged_for_review": 0 + }, + "factual_correctness": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.40857142857142853, + "flagged_for_review": 0 + }, + "answer_completeness": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.7857142857142857, + "flagged_for_review": 0 + } + }, + "coverage": { + "context_recall": { + "scored": 10, + "eligible": 10, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "context_precision": { + "scored": 10, + "eligible": 10, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "faithfulness": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_relevancy": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + }, + "factual_correctness": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_completeness": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + } + }, + "citation_format": 1.0 + }, + "component": { + "cases": 2, + "outcomes": { + "answered": 2 + }, + "metrics": { + "faithfulness": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + }, + "answer_relevancy": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.8169765529503135, + "flagged_for_review": 0 + }, + "factual_correctness": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.25, + "flagged_for_review": 0 + }, + "answer_completeness": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + } + }, + "coverage": { + "faithfulness": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_relevancy": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "factual_correctness": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_completeness": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + } + }, + "citation_format": 1.0 + }, + "workflow": { + "cases": 2, + "outcomes": { + "answered": 2 + }, + "metrics": { + "faithfulness": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + }, + "answer_relevancy": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.9418714459143624, + "flagged_for_review": 0 + }, + "factual_correctness": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.7350000000000001, + "flagged_for_review": 0 + }, + "answer_completeness": { + "success": 2, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + } + }, + "coverage": { + "faithfulness": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_relevancy": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "factual_correctness": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_completeness": { + "scored": 2, + "eligible": 2, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + } + }, + "citation_format": 1.0 + }, + "v5": { + "cases": 10, + "outcomes": { + "generation_error": 3, + "insufficient_evidence": 2, + "answered": 5 + }, + "metrics": { + "context_recall": { + "success": 10, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + }, + "context_precision": { + "success": 10, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.8385714285112444, + "flagged_for_review": 0 + }, + "faithfulness": { + "success": 5, + "error": 0, + "undefined": 0, + "skipped": 5, + "budget_exhausted": 0, + "mean": 0.8777777777777779, + "flagged_for_review": 0 + }, + "answer_relevancy": { + "success": 5, + "error": 0, + "undefined": 0, + "skipped": 5, + "budget_exhausted": 0, + "mean": 0.9281414053744047, + "flagged_for_review": 0 + }, + "factual_correctness": { + "success": 5, + "error": 0, + "undefined": 0, + "skipped": 5, + "budget_exhausted": 0, + "mean": 0.48, + "flagged_for_review": 0 + }, + "answer_completeness": { + "success": 5, + "error": 0, + "undefined": 0, + "skipped": 5, + "budget_exhausted": 0, + "mean": 0.6, + "flagged_for_review": 0 + } + }, + "coverage": { + "context_recall": { + "scored": 10, + "eligible": 10, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "context_precision": { + "scored": 10, + "eligible": 10, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "faithfulness": { + "scored": 5, + "eligible": 5, + "skipped": 5, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_relevancy": { + "scored": 5, + "eligible": 5, + "skipped": 5, + "errors": 0, + "budget_exhausted": 0 + }, + "factual_correctness": { + "scored": 5, + "eligible": 5, + "skipped": 5, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_completeness": { + "scored": 5, + "eligible": 5, + "skipped": 5, + "errors": 0, + "budget_exhausted": 0 + } + }, + "citation_format": 1.0 + }, + "v6": { + "cases": 10, + "outcomes": { + "answered": 7, + "insufficient_evidence": 3 + }, + "metrics": { + "context_recall": { + "success": 10, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 1.0, + "flagged_for_review": 0 + }, + "context_precision": { + "success": 10, + "error": 0, + "undefined": 0, + "skipped": 0, + "budget_exhausted": 0, + "mean": 0.8385714285112444, + "flagged_for_review": 0 + }, + "faithfulness": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.8898046398046399, + "flagged_for_review": 0 + }, + "answer_relevancy": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.8986788023844029, + "flagged_for_review": 0 + }, + "factual_correctness": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.4857142857142857, + "flagged_for_review": 0 + }, + "answer_completeness": { + "success": 7, + "error": 0, + "undefined": 0, + "skipped": 3, + "budget_exhausted": 0, + "mean": 0.6628571428571429, + "flagged_for_review": 0 + } + }, + "coverage": { + "context_recall": { + "scored": 10, + "eligible": 10, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "context_precision": { + "scored": 10, + "eligible": 10, + "skipped": 0, + "errors": 0, + "budget_exhausted": 0 + }, + "faithfulness": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_relevancy": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + }, + "factual_correctness": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + }, + "answer_completeness": { + "scored": 7, + "eligible": 7, + "skipped": 3, + "errors": 0, + "budget_exhausted": 0 + } + }, + "citation_format": 1.0 + } + }, + "paired_historical_v5_v6": { + "faithfulness": { + "count": 4, + "v5": 0.875, + "v6": 0.8557692307692308 + }, + "answer_relevancy": { + "count": 4, + "v5": 0.9170897981710456, + "v6": 0.9419035384467649 + }, + "factual_correctness": { + "count": 4, + "v5": 0.5, + "v6": 0.3775 + }, + "answer_completeness": { + "count": 4, + "v5": 0.5, + "v6": 0.625 + } + }, + "workflow_contracts": { + "turns": 2, + "same_session": true, + "repeat_answer_cache_hit": true, + "followup_result": "answered" + }, + "budget": { + "limit_usd": 0.5, + "tracked_standard_route_token_value_usd": 0.29459167199999997, + "reported_value_usd": 0.28468837199999997, + "unsettled_reservations": 1, + "http_attempts": 352, + "pricing_input_per_million": 0.3, + "pricing_output_per_million": 1.2, + "pricing_cache_read_per_million": 0.0015, + "route": "TokenHarbor free allowance", + "basis": "Conservative standard-route token value including preflight attempts and reservations; actual free-route billing is zero." + }, + "failure_review": [ + { + "case": "TRAIN_Q376", + "finding": "Question mentions details and asks where to find them. Link-only responses have zero reference coverage; task scope and reference alignment require review. Earlier automatic reference-only diagnosis was premature." + }, + { + "case": "TRAIN_Q117", + "finding": "Historical answer extrapolates version scope. New verifier also treats an explicitly resolved scope difference as a version conflict." + }, + { + "case": "TRAIN_Q548", + "finding": "Unrequested remediation, bundled claims and overly literal error-string judging. New verifier demands remediation beyond a cause question." + }, + { + "case": "TRAIN_Q310", + "finding": "Historical withholding rejects an unsupported reboot step. New answer gives the documented cache timeout guidance." + }, + { + "case": "TRAIN_Q353", + "finding": "Historical answer adds commands for a different procedure; one verifier finding about missing citations is also mistaken." + } + ], + "limitations": [ + "Training diagnostic sample, no held-out development evaluation.", + "Today writer and judge share the DeepSeek model; two sanity canaries do not establish human alignment.", + "RAGAS answer_completeness is reference-fact recall mode. Required-task completeness is a separate criterion.", + "Workflow semantic scores are per turn with conversation context, plus code checks for cache/session behavior. Canonical text only.", + "Imported wall-clock timestamps are anchored at import time using saved durations. Aggregate usage spans carry token counts without per-call timing.", + "Source references remain unchanged. Supplied-evidence component cases are distinct from production retrieval.", + "Historical native clocks could not be patched after finalization. New measured imports preserve the original experiments." + ], + "provenance": { + "scorer_sha256": "4024b7d7422acb2e166f5cf366476e29df778d6a0845425510fe4cc18f9ae00d", + "pipeline_manifest_sha256": "4c69828bb542728f0915e4ead1a88319bc8b445ce9b5e05199c8782a7ba63883" + } +} From 8676e3db3450db4bd539def25394321767faa728 Mon Sep 17 00:00:00 2001 From: shipitdev Date: Wed, 11 Nov 2026 12:34:13 +0530 Subject: [PATCH 4/4] Expose cached reference claim coverage alongside RAGAS scores --- README.md | 2 +- evaluations/results/layered_audit_10.json | 63 ++++++++++++++++++++--- scripts/eval_suite.py | 27 +++++++++- scripts/publish_langsmith.py | 2 + tests/test_eval_suite.py | 12 +++++ 5 files changed, 98 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index af2a475..8d7641f 100644 --- a/README.md +++ b/README.md @@ -89,7 +89,7 @@ The pipeline released seven answers, withheld two and requested clarification on On the four shared released historical cases, the same DeepSeek judge scored reference correctness **50.0% → 37.8%** from v5 to v6. Q117's broader version assertions contribute to the drop. Q548 also exposes bundled-claim and overly literal error-string judgments. Q376 mentions details and asks where to find them; its reference expects a vulnerability explanation. A reviewed task rubric should resolve that scope difference. The new verifier also incorrectly rejects qualified version differences and demands unrequested remediation. These findings guide the next change. -The legacy `answer_completeness` field uses RAGAS `FactualCorrectness(mode="recall")`. It measures reference agreement. Required-task completeness needs explicit review of the question and requested steps. Claim decomposition and coverage settings affect reference scores. [RAGAS metric definitions](https://docs.ragas.io/en/stable/concepts/metrics/available_metrics/factual_correctness/) +The legacy `answer_completeness` field uses RAGAS `FactualCorrectness(mode="recall")`. It measures reference agreement. Required-task completeness needs explicit review of the question and requested steps. A separate `reference_claim_coverage` diagnostic reuses cached reference-to-answer judgments. Q425 covers both reference commands even though the original recall-mode score is zero because it rejected bundled answer claims. Both values stay visible and require review. Claim decomposition and coverage settings affect reference scores. [RAGAS metric definitions](https://docs.ragas.io/en/stable/concepts/metrics/available_metrics/factual_correctness/) Each dashboard metric has `ragas_state_*`, `ragas_scored_*` and `ragas_eligible_*` feedback. Skipped answers and failed judgments have reasons. Historical imports include saved duration, model usage and original generation errors. Imported wall clocks anchor those durations at import time. Aggregate usage spans carry recorded token totals. [LangSmith usage instrumentation](https://docs.langchain.com/langsmith/log-llm-trace) diff --git a/evaluations/results/layered_audit_10.json b/evaluations/results/layered_audit_10.json index 60f2159..f7039a0 100644 --- a/evaluations/results/layered_audit_10.json +++ b/evaluations/results/layered_audit_10.json @@ -117,7 +117,13 @@ "budget_exhausted": 0 } }, - "citation_format": 1.0 + "citation_format": 1.0, + "reference_claim_coverage": { + "mean": 0.8571428571428571, + "scored": 7, + "contract": "reference_claim_coverage_v1", + "source": "custom aggregation of cached reverse NLI judgments" + } }, "component": { "cases": 2, @@ -192,7 +198,13 @@ "budget_exhausted": 0 } }, - "citation_format": 1.0 + "citation_format": 1.0, + "reference_claim_coverage": { + "mean": 1.0, + "scored": 2, + "contract": "reference_claim_coverage_v1", + "source": "custom aggregation of cached reverse NLI judgments" + } }, "workflow": { "cases": 2, @@ -267,7 +279,13 @@ "budget_exhausted": 0 } }, - "citation_format": 1.0 + "citation_format": 1.0, + "reference_claim_coverage": { + "mean": 1.0, + "scored": 2, + "contract": "reference_claim_coverage_v1", + "source": "custom aggregation of cached reverse NLI judgments" + } }, "v5": { "cases": 10, @@ -376,7 +394,13 @@ "budget_exhausted": 0 } }, - "citation_format": 1.0 + "citation_format": 1.0, + "reference_claim_coverage": { + "mean": 0.74, + "scored": 5, + "contract": "reference_claim_coverage_v1", + "source": "custom aggregation of cached reverse NLI judgments" + } }, "v6": { "cases": 10, @@ -484,7 +508,13 @@ "budget_exhausted": 0 } }, - "citation_format": 1.0 + "citation_format": 1.0, + "reference_claim_coverage": { + "mean": 0.6357142857142857, + "scored": 7, + "contract": "reference_claim_coverage_v1", + "source": "custom aggregation of cached reverse NLI judgments" + } } }, "paired_historical_v5_v6": { @@ -547,6 +577,26 @@ { "case": "TRAIN_Q353", "finding": "Historical answer adds commands for a different procedure; one verifier finding about missing citations is also mistaken." + }, + { + "case": "TRAIN_Q252", + "finding": "Five faithfulness flags include framing, a user-attributed condition and clarification requests. Review claim selection before interpreting every flagged statement as hallucination." + }, + { + "case": "TRAIN_Q425", + "finding": "Both reference commands are marked covered. RAGAS recall mode is zero after bundled response claims are rejected. Added reference-claim coverage diagnostic scores 2/2." + }, + { + "case": "TRAIN_Q414", + "finding": "Support-withdrawal answer matches the reference; supplementary lifecycle link contributes extra reference precision requirements." + }, + { + "case": "TRAIN_Q361", + "finding": "Answer differentiates distributed and z/OS support dates while the reference gives one platform row. Review platform scope." + }, + { + "case": "TRAIN_Q488", + "finding": "Evidence establishes certified encryption modules but does not establish digital-signature applicability. Withholding is justified." } ], "limitations": [ @@ -556,7 +606,8 @@ "Workflow semantic scores are per turn with conversation context, plus code checks for cache/session behavior. Canonical text only.", "Imported wall-clock timestamps are anchored at import time using saved durations. Aggregate usage spans carry token counts without per-call timing.", "Source references remain unchanged. Supplied-evidence component cases are distinct from production retrieval.", - "Historical native clocks could not be patched after finalization. New measured imports preserve the original experiments." + "Historical native clocks could not be patched after finalization. New measured imports preserve the original experiments.", + "Custom reference-claim coverage reuses fallible cached judge decisions and is distinct from the legacy RAGAS recall-mode score." ], "provenance": { "scorer_sha256": "4024b7d7422acb2e166f5cf366476e29df778d6a0845425510fe4cc18f9ae00d", diff --git a/scripts/eval_suite.py b/scripts/eval_suite.py index a892479..d2b1816 100644 --- a/scripts/eval_suite.py +++ b/scripts/eval_suite.py @@ -49,6 +49,30 @@ def judge_version_flags(case, result): return list(flags.values()) +def reference_claim_coverage(metric): + """Aggregate cached reference-to-response NLI from the pinned factual metric.""" + groups = [] + claims = None + for call in metric.get('judge_calls', []): + decision = call.get('decision') or {} + if 'claims' in decision: + claims = decision['claims'] + if 'statements' in decision and claims is not None: + statements = decision['statements'] + if (statements and len(statements) == len(claims) + and all(isinstance(s, dict) and s.get('verdict') in (0, 1) for s in statements) + and [' '.join(s.get('statement', '').split()) for s in statements] + == [' '.join(c.split()) for c in claims]): + groups.append(statements) + claims = None + if metric.get('state') != 'success' or len(groups) != 2: + return {'state': 'unavailable', 'score': None} + reference_verdicts = groups[-1] + return {'state': 'derived_from_cached_judge', + 'score': sum(s['verdict'] for s in reference_verdicts) / len(reference_verdicts), + 'reference_claims': len(reference_verdicts), 'contract': 'reference_claim_coverage_v1'} + + def build_report(cases, ragas=None): if not cases or len({c['question_id'] for c in cases}) != len(cases): raise ValueError('Expected nonempty unique cases') @@ -89,7 +113,8 @@ def build_report(cases, ragas=None): 'reason': m.get('reason') or m.get('error_type')} rows.append({'question_id': c['question_id'], 'outcome': c['status'], 'selected_gold_document_hit': hit, 'selected_passage_count': len(c['sources']), - 'citation_contract': citation, 'ragas': metrics, 'judge_integrity_flags': flags}) + 'citation_contract': citation, 'ragas': metrics, 'judge_integrity_flags': flags, + 'reference_claim_coverage': reference_claim_coverage(judged.get(c['question_id'], {}).get('metrics', {}).get('answer_completeness', {}))}) gen = c.get('generation', {}) elapsed = gen.get('seconds') if isinstance(elapsed, (int, float)) and math.isfinite(elapsed): diff --git a/scripts/publish_langsmith.py b/scripts/publish_langsmith.py index d8c61ed..cf29421 100644 --- a/scripts/publish_langsmith.py +++ b/scripts/publish_langsmith.py @@ -87,6 +87,8 @@ def publish(directory, client, ragas=None, *, dataset_name=None, experiment_pref 'comment': 'Citation format and source IDs.'}, {'key': 'recorded_latency_seconds', 'score': seconds if measured else None}, {'key': 'recorded_tokens', 'score': sum(c.get('total_tokens') or 0 for c in calls)}, + {'key': 'reference_claim_coverage', 'score': row['reference_claim_coverage']['score'], + 'comment': 'Custom reference-claim coverage from cached reverse NLI judgments, version 1.'}, {'key': 'cost_status', 'value': 'free_route' if calls and all(str(c.get('model', '')).endswith(':free') for c in calls) else 'not_recorded'}, ] for metric, value in row['ragas'].items(): diff --git a/tests/test_eval_suite.py b/tests/test_eval_suite.py index 15a0488..5154cb8 100644 --- a/tests/test_eval_suite.py +++ b/tests/test_eval_suite.py @@ -78,3 +78,15 @@ def test_changed_embedding_identity_prevents_semantic_comparison_and_flagged_sco def test_case_loader_rejects_manifest_path_traversal(tmp_path): (tmp_path / 'manifest.json').write_text(json.dumps({'question_ids': ['../outside']})) with pytest.raises(ValueError): load_cases(tmp_path) + + +def test_reference_claim_coverage_is_independent_of_bundled_response_claims(): + from scripts.eval_suite import reference_claim_coverage + metric = {'state': 'success', 'score': 0, 'judge_calls': [ + {'decision': {'claims': ['commands plus an extra condition']}}, + {'decision': {'statements': [{'statement': 'commands plus an extra condition', 'verdict': 0}]}}, + {'decision': {'claims': ['command one', 'command two']}}, + {'decision': {'statements': [{'statement': 'command one', 'verdict': 1}, + {'statement': 'command two', 'verdict': 1}]}}]} + assert reference_claim_coverage(metric)['score'] == 1.0 + assert reference_claim_coverage({'state': 'error', 'judge_calls': []})['score'] is None