diff --git a/.env.example b/.env.example index d73f69f..79ed87e 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,17 @@ S3_BUCKET=aios-submissions GEMINI_API_KEY=your-gemini-api-key-here GEMINI_MODEL=gemini-2.5-pro +# ── Anthropic Claude AI (primary provider) ── +CLAUDE_API_KEY=your-anthropic-api-key-here +CLAUDE_MODEL=claude-opus-4-5 + +# ── LLM Provider Strategy ── +# claude_primary → Claude first, Gemini fallback (default, recommended) +# gemini_primary → Gemini first, Claude fallback +# claude_only → Claude only, no fallback +# gemini_only → Gemini only, no fallback +LLM_PROVIDER_STRATEGY=claude_primary + # ── Grading Defaults ── GRADING_TEMPERATURE=0.0 GRADING_MAX_RETRIES=3 diff --git a/backend/app/config.py b/backend/app/config.py index ce370ea..d24c3af 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -36,6 +36,17 @@ class Settings(BaseSettings): GEMINI_API_KEY: str = "" GEMINI_MODEL: str = "gemini-2.5-pro" + # ── Anthropic Claude AI (fallback provider) ── + CLAUDE_API_KEY: str = "" + CLAUDE_MODEL: str = "claude-opus-4-5" + + # ── LLM Provider Strategy ── + # "claude_primary" → Claude first, Gemini fallback (default) + # "gemini_primary" → Gemini first, Claude fallback + # "gemini_only" → Gemini only, no fallback + # "claude_only" → Claude only, no fallback + LLM_PROVIDER_STRATEGY: str = "claude_primary" + # ── Grading ── GRADING_TEMPERATURE: float = 0.0 GRADING_MAX_RETRIES: int = 3 diff --git a/backend/app/services/llm_client.py b/backend/app/services/llm_client.py index 1347bcd..bed2d9f 100644 --- a/backend/app/services/llm_client.py +++ b/backend/app/services/llm_client.py @@ -1,9 +1,15 @@ """ -LLM Client — Google Gemini API wrapper with retry logic. -Uses Gemini 2.5 Pro for grading physics derivations. +LLM Client — Multi-provider AI wrapper with Claude primary, Gemini fallback. + +Provider strategy is controlled by LLM_PROVIDER_STRATEGY in settings: + "claude_primary" → Claude first, Gemini fallback (default) + "gemini_primary" → Gemini first, Claude fallback + "gemini_only" → Gemini only, no fallback + "claude_only" → Claude only, no fallback """ import json import time +import base64 import logging import os from typing import Any @@ -15,8 +21,10 @@ logger = logging.getLogger(__name__) +# ─── Gemini client cache ───────────────────────────────────────────────────── + _client: genai.Client | None = None -_client_key: str | None = None # Track which key the cached client uses +_client_key: str | None = None def get_client(api_key: str | None = None) -> genai.Client: @@ -33,30 +41,228 @@ def get_client(api_key: str | None = None) -> genai.Client: if key: _client = genai.Client(api_key=key) else: - _client = genai.Client() # Fall back to SDK's automatic environment variable checking + _client = genai.Client() # Fall back to SDK's automatic env var checking _client_key = key return _client -def get_grading_system_prompt(subject: str = "General", board: str = "Generic", grade_level: str = "Unknown") -> str: - """Load the base system prompt dynamically based on the subject and append all markdown KB files.""" - - base_prompt = f"""You are an expert {board} examiner ({grade_level} {subject}) with 15 years of experience. -You grade subjective answers and derivations step by step, awarding partial marks according to the {board} marking scheme. -You output ONLY valid JSON — no preamble, no markdown, no code fences. +# ─── Claude helpers ────────────────────────────────────────────────────────── + +def _get_anthropic_client(api_key: str | None = None): + """Lazy-import anthropic and return a client.""" + try: + import anthropic # type: ignore + except ImportError: + raise RuntimeError( + "anthropic package is not installed. Run: pip install anthropic" + ) + key = api_key or settings.CLAUDE_API_KEY + if not key: + raise RuntimeError("CLAUDE_API_KEY is not configured.") + return anthropic.Anthropic(api_key=key) + + +def _call_claude_raw( + prompt: str, + system_prompt: str = "", + temperature: float = 0.0, + max_tokens: int = 2048, + image_bytes: bytes | None = None, + image_mime: str = "image/png", + api_key: str | None = None, +) -> dict[str, Any]: + """Single call to Claude. Returns same shape as call_gemini().""" + client = _get_anthropic_client(api_key) + model = settings.CLAUDE_MODEL + + # Build content list + content: list = [] + if image_bytes: + content.append({ + "type": "image", + "source": { + "type": "base64", + "media_type": image_mime, + "data": base64.standard_b64encode(image_bytes).decode("utf-8"), + }, + }) + content.append({"type": "text", "text": prompt}) + + try: + import anthropic # type: ignore + start = time.time() + response = client.messages.create( + model=model, + max_tokens=max_tokens, + system=system_prompt if system_prompt else anthropic.NOT_GIVEN, + messages=[{"role": "user", "content": content}], + temperature=temperature, + ) + latency_ms = int((time.time() - start) * 1000) + response_text = response.content[0].text if response.content else "" + tokens_in = response.usage.input_tokens if response.usage else 0 + tokens_out = response.usage.output_tokens if response.usage else 0 + logger.info(f"Claude {latency_ms}ms ({tokens_in}/{tokens_out} tokens)") + return { + "response_text": response_text, + "tokens_in": tokens_in, + "tokens_out": tokens_out, + "latency_ms": latency_ms, + "model": model, + "success": True, + } + except Exception as e: + return { + "response_text": "", "tokens_in": 0, "tokens_out": 0, + "latency_ms": 0, "model": model, "success": False, "error": str(e), + } -Grading philosophy: -- Award marks for correct method even if the final answer is wrong -- Penalize errors in intermediate steps that cascade -- Accept equivalent mathematical expressions (e.g., F=ma and a=F/m are equivalent) -- If a step is partially correct, award proportional partial credit""" + +def _call_gemini_raw( + prompt: str, + system_prompt: str = "", + temperature: float = 0.0, + max_tokens: int = 2048, + image_bytes: bytes | None = None, + image_mime: str = "image/png", + call_type: str = "general", + api_key: str | None = None, +) -> dict[str, Any]: + """Single-attempt Gemini call (no retry). Returns same shape as call_gemini().""" + client = get_client(api_key=api_key) + model = settings.GEMINI_MODEL + + contents: list = [] + if image_bytes: + contents.append(types.Part.from_bytes(data=image_bytes, mime_type=image_mime)) + contents.append(prompt) + + try: + start = time.time() + response = client.models.generate_content( + model=model, + contents=contents, + config=types.GenerateContentConfig( + system_instruction=system_prompt or None, + temperature=temperature, + max_output_tokens=max_tokens, + ), + ) + latency_ms = int((time.time() - start) * 1000) + response_text = response.text or "" + tokens_in = tokens_out = 0 + if response.usage_metadata: + tokens_in = response.usage_metadata.prompt_token_count or 0 + tokens_out = response.usage_metadata.candidates_token_count or 0 + logger.info(f"Gemini [{call_type}] {latency_ms}ms ({tokens_in}/{tokens_out} tokens)") + return { + "response_text": response_text, + "tokens_in": tokens_in, + "tokens_out": tokens_out, + "latency_ms": latency_ms, + "model": model, + "success": True, + } + except Exception as e: + return { + "response_text": "", "tokens_in": 0, "tokens_out": 0, + "latency_ms": 0, "model": model, "success": False, "error": str(e), + } + + +# ─── Unified call_llm ──────────────────────────────────────────────────────── + +def call_llm( + prompt: str, + system_prompt: str = "", + temperature: float | None = None, + max_tokens: int = 2048, + call_type: str = "general", + image_bytes: bytes | None = None, + image_mime: str = "image/png", + api_key: str | None = None, +) -> dict[str, Any]: + """ + Unified LLM call respecting LLM_PROVIDER_STRATEGY. + + Strategy routing: + claude_primary → Claude → Gemini fallback (default) + gemini_primary → Gemini → Claude fallback + claude_only → Claude only + gemini_only → Gemini only (with retries via call_gemini) + + image_bytes/image_mime are forwarded for vision tasks (OCR, diagram detection). + """ + temp = temperature if temperature is not None else settings.GRADING_TEMPERATURE + strategy = settings.LLM_PROVIDER_STRATEGY.lower() + + def _claude() -> dict[str, Any]: + try: + return _call_claude_raw(prompt, system_prompt, temp, max_tokens, + image_bytes, image_mime, api_key) + except RuntimeError as e: + # Key not configured or package missing — treat as provider failure + return {"response_text": "", "tokens_in": 0, "tokens_out": 0, + "latency_ms": 0, "model": settings.CLAUDE_MODEL, + "success": False, "error": str(e)} + + def _gemini() -> dict[str, Any]: + return _call_gemini_raw(prompt, system_prompt, temp, max_tokens, + image_bytes, image_mime, call_type, api_key) + + if strategy == "claude_only": + return _claude() + + if strategy == "gemini_only": + # Keep original retry behaviour for gemini-only mode + return call_gemini(prompt, system_prompt, temp, max_tokens, call_type, api_key) + + if strategy == "gemini_primary": + primary, fallback, primary_name, fallback_name = _gemini, _claude, "Gemini", "Claude" + else: + # Default: claude_primary + primary, fallback, primary_name, fallback_name = _claude, _gemini, "Claude", "Gemini" + + result = primary() + if result["success"]: + return result + + logger.warning( + f"[{call_type}] {primary_name} failed ({result.get('error', 'unknown')}). " + f"Falling back to {fallback_name}." + ) + fallback_result = fallback() + if not fallback_result["success"]: + logger.error( + f"[{call_type}] {fallback_name} fallback also failed: " + f"{fallback_result.get('error', 'unknown')}" + ) + return fallback_result + + +# ─── Legacy call_gemini (kept for backward compat & gemini_only strategy) ──── + +def get_grading_system_prompt( + subject: str = "General", + board: str = "Generic", + grade_level: str = "Unknown", +) -> str: + """Load the base system prompt dynamically and append all markdown KB files.""" + base_prompt = ( + f"You are an expert {board} examiner ({grade_level} {subject}) with 15 years of experience.\n" + f"You grade subjective answers and derivations step by step, awarding partial marks according to the {board} marking scheme.\n" + "You output ONLY valid JSON — no preamble, no markdown, no code fences.\n\n" + "Grading philosophy:\n" + "- Award marks for correct method even if the final answer is wrong\n" + "- Penalize errors in intermediate steps that cascade\n" + "- Accept equivalent mathematical expressions (e.g., F=ma and a=F/m are equivalent)\n" + "- If a step is partially correct, award proportional partial credit" + ) prompt = base_prompt - - # Path to the kb directory (backend/kb) current_dir = os.path.dirname(os.path.abspath(__file__)) kb_dir = os.path.join(os.path.dirname(os.path.dirname(current_dir)), "kb") - + if os.path.exists(kb_dir): prompt += "\n\n--- ADDITIONAL GRADING GUIDELINES (KNOWLEDGE BASE) ---" for filename in sorted(os.listdir(kb_dir)): @@ -68,23 +274,28 @@ def get_grading_system_prompt(subject: str = "General", board: str = "Generic", prompt += f.read() except Exception as e: logger.error(f"Failed to load KB file {filename}: {e}") - + return prompt -ALIGNMENT_SYSTEM_PROMPT = """You are an expert at analyzing student answers and mapping them to rubric steps. -Given rubric steps and student answer steps, map each rubric step to the most relevant student step(s). -Output ONLY valid JSON — no preamble, no markdown, no code fences.""" + +ALIGNMENT_SYSTEM_PROMPT = ( + "You are an expert at analyzing student answers and mapping them to rubric steps.\n" + "Given rubric steps and student answer steps, map each rubric step to the most relevant student step(s).\n" + "Output ONLY valid JSON — no preamble, no markdown, no code fences." +) + def extract_text_from_image_gemini(image_bytes: bytes) -> str: - """Use Gemini 2.5 Pro Vision to transcribe fuzzy handwritten text and math.""" + """Use Gemini Vision to transcribe fuzzy handwritten text and math.""" client = get_client() try: response = client.models.generate_content( model=settings.GEMINI_MODEL, contents=[ types.Part.from_bytes(data=image_bytes, mime_type="image/jpeg"), - "Extract all the handwritten text, mathematical equations, and derivations from this image exactly as written. Do not summarize. Just output the transcribed text." - ] + "Extract all the handwritten text, mathematical equations, and derivations " + "from this image exactly as written. Do not summarize. Just output the transcribed text.", + ], ) return response.text or "" except Exception as e: @@ -92,7 +303,6 @@ def extract_text_from_image_gemini(image_bytes: bytes) -> str: raise - def call_gemini( prompt: str, system_prompt: str = "", @@ -102,7 +312,7 @@ def call_gemini( api_key: str | None = None, ) -> dict[str, Any]: """Make a call to Gemini API with retry logic and exponential backoff. - + Args: api_key: Optional BYOK Gemini API key. If provided, uses this key instead of the system-wide key from settings. @@ -141,7 +351,10 @@ def call_gemini( except Exception as e: last_error = e wait_time = (2 ** attempt) + 0.5 - logger.warning(f"Gemini [{call_type}] attempt {attempt+1} failed: {e}. Retry in {wait_time:.1f}s") + logger.warning( + f"Gemini [{call_type}] attempt {attempt + 1} failed: {e}. " + f"Retry in {wait_time:.1f}s" + ) time.sleep(wait_time) logger.error(f"Gemini [{call_type}] failed after {settings.GRADING_MAX_RETRIES} attempts") @@ -176,8 +389,12 @@ def parse_json_response(response_text: str) -> dict | list | None: return None -def build_step_grading_prompt(rubric_step: dict, student_step: dict, - sympy_result: dict | None = None, board_notes: str = "") -> str: +def build_step_grading_prompt( + rubric_step: dict, + student_step: dict, + sympy_result: dict | None = None, + board_notes: str = "", +) -> str: """Build the grading prompt for a single rubric step.""" sympy_ctx = "" if sympy_result: @@ -187,28 +404,37 @@ def build_step_grading_prompt(rubric_step: dict, student_step: dict, sympy_ctx = f"[SYMBOLIC VALIDATION: ERROR — {sympy_result.get('error', 'Unknown')}]" mm = rubric_step.get("marks", 1) - return f"""RUBRIC STEP {rubric_step.get('step_num', '?')} (max {mm} marks): -Description: {rubric_step.get('description', '')} -Marking notes: {rubric_step.get('marking_notes', '')} -Partial credit: {rubric_step.get('partial_credit', True)} -{f'Board guidance: {board_notes}' if board_notes else ''} - -STUDENT ANSWER FOR THIS STEP: -{student_step.get('text', '')} - -{sympy_ctx} - -Grade this step. Return JSON: -{{"marks_awarded": , "grade_distribution": , "justification": "", "error_type": ""}}""" + return ( + f"RUBRIC STEP {rubric_step.get('step_num', '?')} (max {mm} marks):\n" + f"Description: {rubric_step.get('description', '')}\n" + f"Marking notes: {rubric_step.get('marking_notes', '')}\n" + f"Partial credit: {rubric_step.get('partial_credit', True)}\n" + f"{f'Board guidance: {board_notes}' if board_notes else ''}\n\n" + f"STUDENT ANSWER FOR THIS STEP:\n" + f"{student_step.get('text', '')}\n\n" + f"{sympy_ctx}\n\n" + f"Grade this step. Return JSON:\n" + f'{{"marks_awarded": , "grade_distribution": , ' + f'"justification": "", "error_type": ""}}' + ) def build_alignment_prompt(rubric_steps: list[dict], student_steps: list[dict]) -> str: """Build prompt to align student steps to rubric steps.""" - r_desc = "\n".join(f" Rubric Step {s.get('step_num', i+1)}: {s.get('description', '')}" for i, s in enumerate(rubric_steps)) - s_desc = "\n".join(f" Student Step {s.get('step_num', i+1)}: {s.get('text', '')[:200]}" for i, s in enumerate(student_steps)) - return f"""RUBRIC STEPS:\n{r_desc}\n\nSTUDENT STEPS:\n{s_desc}\n -Map each rubric step to the most relevant student step(s). Use only the integer step numbers for the keys and values. Return a JSON array: -[ - {{"rubric_step": 1, "student_steps": [2, 3], "confidence": 0.95}} -] -If no match, use "student_steps": [] with confidence 0.0.""" + r_desc = "\n".join( + f" Rubric Step {s.get('step_num', i + 1)}: {s.get('description', '')}" + for i, s in enumerate(rubric_steps) + ) + s_desc = "\n".join( + f" Student Step {s.get('step_num', i + 1)}: {s.get('text', '')[:200]}" + for i, s in enumerate(student_steps) + ) + return ( + f"RUBRIC STEPS:\n{r_desc}\n\nSTUDENT STEPS:\n{s_desc}\n\n" + "Map each rubric step to the most relevant student step(s). " + "Use only the integer step numbers for the keys and values. Return a JSON array:\n" + "[\n" + ' {"rubric_step": 1, "student_steps": [2, 3], "confidence": 0.95}\n' + "]\n" + 'If no match, use "student_steps": [] with confidence 0.0.' + ) diff --git a/backend/app/services/parsing.py b/backend/app/services/parsing.py index 98ba4b6..7101391 100644 --- a/backend/app/services/parsing.py +++ b/backend/app/services/parsing.py @@ -82,38 +82,29 @@ def _render_page_to_png(page: fitz.Page, scale: float = 2.0) -> bytes: def _ocr_page_single(image_bytes: bytes, page_num: int) -> str: - """Run OCR on a single page image using Gemini Client.""" - from app.services.llm_client import get_client - from app.config import settings - from google.genai import types as genai_types - - client = get_client() - try: - response = client.models.generate_content( - model=settings.GEMINI_MODEL, - contents=[ - genai_types.Part.from_bytes(data=image_bytes, mime_type="image/png"), - OCR_PROMPT, - ], - config=genai_types.GenerateContentConfig( - temperature=0.0, - max_output_tokens=4096, - ), - ) - return response.text or "" - except Exception as e: - logger.error(f"[OCR] Page {page_num} Gemini call failed: {e}") + """Run OCR on a single page image using the configured LLM provider (Claude primary, Gemini fallback).""" + from app.services.llm_client import call_llm + + result = call_llm( + prompt=OCR_PROMPT, + temperature=0.0, + max_tokens=4096, + call_type=f"ocr_page_{page_num}", + image_bytes=image_bytes, + image_mime="image/png", + ) + if not result["success"]: + logger.error(f"[OCR] Page {page_num} LLM call failed: {result.get('error')}") return "" + return result["response_text"] or "" def detect_diagram_boxes(image_bytes: bytes, mime_type: str) -> list[dict]: """ - Call Gemini to detect diagrams, tables, flowcharts and return their normalized + Call the configured LLM to detect diagrams, tables, flowcharts and return their normalized bounding boxes [ymin, xmin, ymax, xmax] in the range 0 to 1000. """ - from app.services.llm_client import get_client, parse_json_response - from app.config import settings - from google.genai import types as genai_types + from app.services.llm_client import call_llm, parse_json_response prompt = """ Identify all diagrams, hand-drawn figures, drawings, graphs, tables, maps, circuits, or flowcharts on this page. @@ -123,20 +114,19 @@ def detect_diagram_boxes(image_bytes: bytes, mime_type: str) -> list[dict]: {"type": "biological_diagram", "box_2d": [100, 200, 500, 800]} ] """ - client = get_client() try: - response = client.models.generate_content( - model=settings.GEMINI_MODEL, - contents=[ - genai_types.Part.from_bytes(data=image_bytes, mime_type=mime_type), - prompt, - ], - config=genai_types.GenerateContentConfig( - temperature=0.0, - max_output_tokens=1024, - ), + result = call_llm( + prompt=prompt, + temperature=0.0, + max_tokens=1024, + call_type="diagram_detection", + image_bytes=image_bytes, + image_mime=mime_type, ) - raw_text = response.text or "" + if not result["success"]: + logger.error(f"[Parse] Diagram detection failed: {result.get('error')}") + return [] + raw_text = result["response_text"] or "" parsed = parse_json_response(raw_text) if isinstance(parsed, list): return parsed @@ -312,12 +302,10 @@ def align_answers_to_questions( questions: list[dict], ) -> dict[str, str]: """ - Call Gemini to align student answer blocks from the transcript to the + Call the configured LLM to align student answer blocks from the transcript to the list of questions from the task rubric. """ - from app.services.llm_client import get_client, parse_json_response - from app.config import settings - from google.genai import types as genai_types + from app.services.llm_client import call_llm, parse_json_response if not questions: return {} @@ -350,17 +338,17 @@ def align_answers_to_questions( }} """ - client = get_client() try: - response = client.models.generate_content( - model=settings.GEMINI_MODEL, - contents=[prompt], - config=genai_types.GenerateContentConfig( - temperature=0.0, - max_output_tokens=4096, - ), + result = call_llm( + prompt=prompt, + temperature=0.0, + max_tokens=4096, + call_type="answer_alignment", ) - raw_text = response.text or "" + if not result["success"]: + logger.error(f"[Parse] LLM alignment failed: {result.get('error')}") + return parse_answers_fallback(full_transcript) + raw_text = result["response_text"] or "" parsed = parse_json_response(raw_text) if isinstance(parsed, dict): return {str(k): str(v) for k, v in parsed.items()} diff --git a/backend/app/tests/test_api_keys.py b/backend/app/tests/test_api_keys.py new file mode 100644 index 0000000..5e0b565 --- /dev/null +++ b/backend/app/tests/test_api_keys.py @@ -0,0 +1,161 @@ +""" +API Key Connectivity Tests — verifies that configured LLM provider keys +are valid and can reach their respective APIs. + +Run with: + pytest backend/app/tests/test_api_keys.py -v + +These are LIVE tests — they make real network calls. +They are skipped automatically when keys are not configured. +""" +import pytest +from app.config import settings + + +# ─── Gemini ────────────────────────────────────────────────────────────────── + +@pytest.mark.skipif( + not settings.GEMINI_API_KEY, + reason="GEMINI_API_KEY is not configured — skipping", +) +def test_gemini_api_key_is_valid(): + """ + Sends a minimal prompt to Gemini and asserts a non-empty response. + A 429 (quota exhausted) is treated as a key-is-valid result — + the key exists but the account is out of credits. + """ + from app.services.llm_client import _call_gemini_raw + + result = _call_gemini_raw( + prompt="Reply with the single word: PONG", + temperature=0.0, + max_tokens=16, + call_type="key_test", + ) + + # 429 quota error → key is valid, account is depleted + error_str = str(result.get("error", "")).lower() + is_quota_error = ( + "429" in error_str + or "resource_exhausted" in error_str + or "quota" in error_str + or "prepayment" in error_str + ) + + if is_quota_error: + pytest.skip( + f"Gemini key is valid but account quota is exhausted: {result.get('error')}" + ) + + assert result["success"], ( + f"Gemini API key check failed.\n" + f"Error: {result.get('error')}\n" + f"Tip: Check GEMINI_API_KEY in your .env" + ) + assert result["response_text"].strip(), "Gemini returned an empty response" + print(f"\n✅ Gemini OK | model={result['model']} | " + f"{result['latency_ms']}ms | response='{result['response_text'].strip()}'") + + +# ─── Claude ────────────────────────────────────────────────────────────────── + +@pytest.mark.skipif( + not settings.CLAUDE_API_KEY, + reason="CLAUDE_API_KEY is not configured — skipping", +) +def test_claude_api_key_is_valid(): + """ + Sends a minimal prompt to Claude and asserts a non-empty response. + Distinguishes between an invalid key (401) and a quota/billing issue (429). + """ + from app.services.llm_client import _call_claude_raw + + result = _call_claude_raw( + prompt="Reply with the single word: PONG", + temperature=0.0, + max_tokens=16, + ) + + error_str = str(result.get("error", "")).lower() + + # Billing / rate limit → key is valid but account needs attention + is_billing_error = ( + "credit" in error_str + or "billing" in error_str + or "429" in error_str + or "overloaded" in error_str + or "rate_limit" in error_str + ) + # Authentication failure → key is wrong + is_auth_error = ( + "401" in error_str + or "authentication" in error_str + or "invalid x-api-key" in error_str + or "permission" in error_str + ) + + if is_billing_error: + pytest.skip( + f"Claude key is valid but account has billing/rate issues: {result.get('error')}" + ) + + assert not is_auth_error, ( + f"Claude API key is invalid or unauthorized.\n" + f"Error: {result.get('error')}\n" + f"Tip: Check CLAUDE_API_KEY in your .env" + ) + assert result["success"], ( + f"Claude API key check failed.\n" + f"Error: {result.get('error')}\n" + f"Tip: Check CLAUDE_API_KEY in your .env" + ) + assert result["response_text"].strip(), "Claude returned an empty response" + print(f"\n✅ Claude OK | model={result['model']} | " + f"{result['latency_ms']}ms | response='{result['response_text'].strip()}'") + + +# ─── Strategy / Fallback ───────────────────────────────────────────────────── + +@pytest.mark.skipif( + not settings.GEMINI_API_KEY and not settings.CLAUDE_API_KEY, + reason="No LLM keys configured at all — skipping", +) +def test_call_llm_strategy_produces_response(): + """ + Calls the unified call_llm() with the current LLM_PROVIDER_STRATEGY + and verifies at least one provider returns a response. + """ + try: + from app.services.llm_client import call_llm + except ImportError as e: + pytest.skip(f"Missing dependency: {e}") + + result = call_llm( + prompt="Reply with the single word: PONG", + temperature=0.0, + max_tokens=16, + call_type="strategy_test", + ) + + error_str = str(result.get("error", "")).lower() + is_quota_or_billing = any( + k in error_str + for k in ("429", "quota", "resource_exhausted", "prepayment", "credit", "billing") + ) + + if is_quota_or_billing: + pytest.skip( + f"Strategy={settings.LLM_PROVIDER_STRATEGY!r}: key valid but account has quota/billing issues" + ) + + assert result["success"], ( + f"call_llm() failed with strategy={settings.LLM_PROVIDER_STRATEGY!r}.\n" + f"Error: {result.get('error')}\n" + f"Make sure at least one of GEMINI_API_KEY / CLAUDE_API_KEY is set in .env" + ) + assert result["response_text"].strip(), "call_llm() returned an empty response" + print( + f"\n✅ call_llm OK | strategy={settings.LLM_PROVIDER_STRATEGY!r} | " + f"model={result['model']} | {result['latency_ms']}ms | " + f"response='{result['response_text'].strip()}'" + ) diff --git a/backend/requirements.txt b/backend/requirements.txt index 3a0a457..ca12820 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -17,6 +17,9 @@ redis==5.3.0 # ── Google Gemini AI ── google-genai==1.16.1 +# ── Anthropic Claude AI ── +anthropic==0.40.0 + # ── Document Parsing & Math ── PyMuPDF==1.24.4 sympy==1.12.1