Coverage for app \ knowledge_graph \ autopilot.py: 100%

1import json 

2from app.llm.ollama_client import call_ollama 

3from app.knowledge_graph.patient_graph_reader import _get_driver as get_driver 

4from app.utils.logger import get_logger 

5 

6logger = get_logger(__name__) 

7 

8def analyze_health_intent(user_text: str): 

9 """ 

10 Analyzes user text to find NEW medical facts (Diagnosis, Medication, Allergy). 

11 Returns a LIST of facts found. 

12 """ 

13 prompt = f""" 

14 Analyze the user text. Identify ALL NEW medical facts about the user. 

15  

16 Rules: 

17 1. Ignore questions, hypotheticals, or third-party statements. 

18 2. NORMALIZE terms (e.g., "sugar disease" -> "Diabetes Mellitus"). 

19 3. If multiple facts exist (e.g., "I have fever and take aspirin"), extract BOTH. 

20  

21 User Text: "{user_text}" 

22  

23 Return ONLY a valid JSON ARRAY of objects. Example: 

24 [ 

25 {{ "category": "Condition", "original_term": "high fever", "normalized_term": "Fever" }}, 

26 {{ "category": "Medication", "original_term": "aspirin", "normalized_term": "Aspirin" }} 

27 ] 

28  

29 If no facts found, return: [] 

30 """ 

31 

32 try: 

33 response = call_ollama(prompt) 

34 # Clean potential markdown 

35 clean_json = response.replace("```json", "").replace("```", "").strip() 

36 data = json.loads(clean_json) 

37 

38 # Ensure it's a list 

39 if isinstance(data, dict): 

40 data = [data] 

41 

42 # Filter for valid entries 

43 valid_facts = [item for item in data if item.get("category") and item.get("normalized_term")] 

44 return valid_facts 

45 

46 except Exception as e: 

47 logger.error(f"Autopilot analysis failed: {e}") 

48 return [] 

49 

50def apply_graph_update(user_id: str, category: str, entity_name: str): 

51 """ 

52 Writes to Neo4j ONLY when confirmed by user. 

53 """ 

54 driver = get_driver() 

55 query = "" 

56 

57 if category == "Condition": 

58 query = """ 

59 MATCH (u:Patient {id: $uid}) 

60 MERGE (c:Condition {name: $name}) 

61 MERGE (u)-[:HAS_CONDITION]->(c) 

62 RETURN u.id 

63 """ 

64 elif category == "Medication": 

65 query = """ 

66 MATCH (u:Patient {id: $uid}) 

67 MERGE (d:Drug {name: $name}) 

68 MERGE (u)-[:TAKES_DRUG]->(d) 

69 RETURN u.id 

70 """ 

71 elif category == "Allergy": 

72 query = """ 

73 MATCH (u:Patient {id: $uid}) 

74 MERGE (a:Allergy {name: $name}) 

75 MERGE (u)-[:HAS_ALLERGY]->(a) 

76 RETURN u.id 

77 """ 

78 

79 if not query: 

80 return False, "Invalid category" 

81 

82 try: 

83 with driver.session() as session: 

84 result = session.run(query, uid=user_id, name=entity_name) 

85 if result.single(): 

86 return True, f"Successfully added {category}: {entity_name}" 

87 else: 

88 return False, f"Patient {user_id} not found in Neo4j." 

89 except Exception as e: 

90 logger.error(f"Graph update failed: {e}") 

91 return False, str(e)