Coverage for mindsdb / api / a2a / utils.py: 0%

1from typing import Dict, List 

2from mindsdb.utilities.log import getLogger 

3 

4logger = getLogger(__name__) 

5 

6 

7def to_serializable(obj): 

8 # Primitives 

9 if isinstance(obj, (str, int, float, bool, type(None))): 

10 return obj 

11 # Pydantic v2 

12 if hasattr(obj, "model_dump"): 

13 return to_serializable(obj.model_dump(exclude_none=True)) 

14 # Pydantic v1 

15 if hasattr(obj, "dict"): 

16 return to_serializable(obj.dict(exclude_none=True)) 

17 # Custom classes with __dict__ 

18 if hasattr(obj, "__dict__"): 

19 return {k: to_serializable(v) for k, v in vars(obj).items() if not k.startswith("_")} 

20 # Dicts 

21 if isinstance(obj, dict): 

22 return {k: to_serializable(v) for k, v in obj.items()} 

23 # Lists, Tuples, Sets 

24 if isinstance(obj, (list, tuple, set)): 

25 return [to_serializable(v) for v in obj] 

26 # Fallback: string 

27 return str(obj) 

28 

29 

30def convert_a2a_message_to_qa_format(a2a_message: Dict) -> List[Dict[str, str]]: 

31 """ 

32 Convert A2A message format to question/answer format. 

33 

34 This is the format that the langchain agent expects and ensure effective multi-turn conversation 

35 

36 Args: 

37 a2a_message: A2A message containing history and current message parts 

38 

39 Returns: 

40 List of messages in question/answer format 

41 """ 

42 converted_messages = [] 

43 

44 # Process conversation history first 

45 if "history" in a2a_message and a2a_message["history"] is not None: 

46 for hist_msg in a2a_message["history"]: 

47 if hist_msg.get("role") == "user": 

48 # Extract text from parts 

49 text = "" 

50 for part in hist_msg.get("parts", []): 

51 if part.get("type") == "text": 

52 text = part.get("text", "") 

53 break 

54 # Create question with empty answer initially 

55 converted_messages.append({"question": text, "answer": ""}) 

56 elif hist_msg.get("role") in ["agent", "assistant"]: 

57 # Extract text from parts 

58 text = "" 

59 for part in hist_msg.get("parts", []): 

60 if part.get("type") == "text": 

61 text = part.get("text", "") 

62 break 

63 # Pair with the most recent question that has empty answer 

64 paired = False 

65 for i in range(len(converted_messages) - 1, -1, -1): 

66 if converted_messages[i].get("answer") == "": 

67 converted_messages[i]["answer"] = text 

68 paired = True 

69 break 

70 

71 if not paired: 

72 logger.warning("Could not pair agent response with question (no empty answer found)") 

73 

74 logger.debug(f"Converted {len(a2a_message['history'])} A2A history messages to Q&A format") 

75 

76 # Add current message as final question with empty answer 

77 current_text = "" 

78 for part in a2a_message.get("parts", []): 

79 if part.get("type") == "text": 

80 current_text = part.get("text", "") 

81 break 

82 converted_messages.append({"question": current_text, "answer": ""}) 

83 

84 return converted_messages