Coverage for graphqler / compiler / resolvers / llm / llm_mutation_object_resolver.py: 28%

1"""LLM-backed mutation resolver. 

2 

3Resolves mutationType + hardDependsOn + softDependsOn for every mutation by 

4asking an LLM for its interpretation of the schema, then merging that result 

5with the classic ID-based resolver output. 

6 

7The `.comparison` attribute (set after `resolve()`) holds the side-by-side diff 

8between the LLM result and the classic result for systematic analysis. 

9""" 

10 

11import json 

12import copy 

13import logging 

14 

15from graphqler import config 

16from graphqler.compiler.resolvers.mutation_object_resolver import MutationObjectResolver 

17from .llm_resolver import LLMResolver 

18from .prompt_templates import MUTATION_SYSTEM_PROMPT, MUTATION_USER_PROMPT_TEMPLATE 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23class LLMMutationObjectResolver(LLMResolver): 

24 """Resolves mutations using an LLM, with optional fallback to the classic resolver.""" 

25 

26 def __init__(self): 

27 super().__init__() 

28 self.comparison: dict = {} # populated by resolve(); keyed by mutation name 

29 

30 def resolve(self, objects: dict, mutations: dict, input_objects: dict) -> dict: 

31 """Resolve mutations via LLM, falling back to classic resolver on failure. 

32 

33 Args: 

34 objects (dict): Compiled objects. 

35 mutations (dict): Raw parsed mutations. 

36 input_objects (dict): Raw parsed input objects. 

37 

38 Returns: 

39 dict: Mutations enriched with mutationType, hardDependsOn, softDependsOn. 

40 """ 

41 # Always run classic resolver — used as fallback and for comparison 

42 classic_mutations = MutationObjectResolver().resolve(objects, copy.deepcopy(mutations), input_objects) 

43 

44 try: 

45 llm_raw = self._call_llm_for_mutations(objects, mutations) 

46 llm_validated = self.validate_llm_mutation_result(llm_raw, list(mutations.keys()), objects) 

47 merged = self.merge_with_classic(llm_validated, classic_mutations, list(mutations.keys())) 

48 self.comparison = self._build_comparison(classic_mutations, merged, mutations.keys()) 

49 logger.info(f"LLM mutation resolver: resolved {len(llm_validated)}/{len(mutations)} mutations") 

50 return merged 

51 except Exception as exc: 

52 if config.LLM_RESOLVER_FALLBACK_TO_ID: 

53 logger.warning(f"LLM mutation resolver failed ({exc}), falling back to classic resolver") 

54 self.comparison = {} 

55 return classic_mutations 

56 raise 

57 

58 def _call_llm_for_mutations(self, objects: dict, mutations: dict) -> dict: 

59 """Build the prompt and call the LLM. 

60 

61 Args: 

62 objects (dict): Compiled objects (for schema context). 

63 mutations (dict): Raw parsed mutations. 

64 

65 Returns: 

66 dict: Raw parsed JSON response from the LLM. 

67 """ 

68 schema_context = self.build_schema_context(objects) 

69 simplified = self.simplify_endpoints(mutations) 

70 mutations_json = json.dumps(simplified, indent=2) 

71 

72 user_prompt = MUTATION_USER_PROMPT_TEMPLATE.format( 

73 schema_context=schema_context, 

74 mutations_json=mutations_json, 

75 ) 

76 return self.call_llm(MUTATION_SYSTEM_PROMPT, user_prompt) 

77 

78 def _build_comparison(self, classic: dict, llm_merged: dict, mutation_names) -> dict: 

79 """Build a per-mutation comparison dict. 

80 

81 Args: 

82 classic (dict): Classic resolver output. 

83 llm_merged (dict): LLM resolver output (after merge). 

84 mutation_names: Iterable of mutation names. 

85 

86 Returns: 

87 dict: Keyed by mutation name; each entry has 'classic', 'llm', 'differs', 'diff'. 

88 """ 

89 comparison = {} 

90 for name in mutation_names: 

91 c = classic.get(name, {}) 

92 llm_entry = llm_merged.get(name, {}) 

93 

94 c_summary = { 

95 "mutationType": c.get("mutationType", "UNKNOWN"), 

96 "hardDependsOn": c.get("hardDependsOn", {}), 

97 "softDependsOn": c.get("softDependsOn", {}), 

98 } 

99 l_summary = { 

100 "mutationType": llm_entry.get("mutationType", "UNKNOWN"), 

101 "hardDependsOn": llm_entry.get("hardDependsOn", {}), 

102 "softDependsOn": llm_entry.get("softDependsOn", {}), 

103 } 

104 

105 diff = {} 

106 for key in ("mutationType", "hardDependsOn", "softDependsOn"): 

107 if c_summary[key] != l_summary[key]: 

108 diff[key] = {"classic": c_summary[key], "llm": l_summary[key]} 

109 

110 comparison[name] = { 

111 "classic": c_summary, 

112 "llm": l_summary, 

113 "differs": bool(diff), 

114 "diff": diff, 

115 } 

116 return comparison