Coverage for backend / app / job_rating / claude.py: 42%

1"""AI scoring using Anthropic Claude Messages API.""" 

2 

3import json 

4import logging 

5import re 

6from typing import Any 

7 

8from anthropic import Anthropic 

9 

10from app.config import settings 

11 

12logger = logging.getLogger(__name__) 

13 

14client = Anthropic(api_key=settings.anthropic_api_key) 

15 

16MODEL = "claude-haiku-4-5-20251001" 

17 

18 

19class ClaudeError(Exception): 

20 """Raised when AI query fails.""" 

21 

22 

23def claude_query(system_prompt: str, llm_prompt: str, max_tokens: int = 1024) -> dict[str, Any]: 

24 """Query Anthropic Claude Messages API with constructed prompt. 

25 :param system_prompt: System prompt defining the AI's role. 

26 :param llm_prompt: Fully constructed prompt describing the job. 

27 :param max_tokens: Maximum tokens in the response. 

28 return: parsed JSON output.""" 

29 

30 try: 

31 # noinspection PyTypeChecker 

32 response = client.messages.create( 

33 model=MODEL, 

34 system=[{"type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}}], 

35 messages=[ 

36 {"role": "user", "content": llm_prompt}, 

37 {"role": "assistant", "content": "{"}, # forces JSON-only output 

38 ], 

39 max_tokens=max_tokens, 

40 temperature=0.2, 

41 ) 

42 

43 content = "{" + response.content[0].text 

44 

45 if not content: 

46 raise ClaudeError("Empty response from Claude") 

47 

48 # Strip markdown code fences if present (e.g. ```json ... ```) 

49 stripped = re.sub(r"^```(?:json)?\s*\n?", "", content.strip()) 

50 stripped = re.sub(r"\n?```\s*$", "", stripped).strip() 

51 

52 try: 

53 return json.loads(stripped) 

54 

55 except json.JSONDecodeError as exc: 

56 logger.error("Invalid JSON returned by Claude: %s", content) 

57 raise ClaudeError("Invalid JSON from Claude") from exc 

58 

59 except Exception as exc: 

60 logger.exception("Claude query failed") 

61 raise ClaudeError from exc