Coverage for haystack/utils/hf.py: 75%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from typing import Any 

6 

7from haystack import logging 

8from haystack.dataclasses import ChatMessage, ImageContent, ReasoningContent, TextContent 

9from haystack.lazy_imports import LazyImport 

10 

11with LazyImport(message="Run 'pip install \"transformers[torch]\"'") as torch_import: 

12 import torch 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17def serialize_hf_model_kwargs(kwargs: dict[str, Any]) -> None: 

18 """ 

19 Recursively serialize HuggingFace specific model keyword arguments in-place to make them JSON serializable. 

20 

21 :param kwargs: The keyword arguments to serialize 

22 """ 

23 torch_import.check() 

24 

25 for k, v in kwargs.items(): 

26 # torch.dtype 

27 if isinstance(v, torch.dtype): 

28 kwargs[k] = str(v) 

29 

30 if isinstance(v, dict): 

31 serialize_hf_model_kwargs(v) 

32 

33 

34def deserialize_hf_model_kwargs(kwargs: dict[str, Any]) -> None: 

35 """ 

36 Recursively deserialize HuggingFace specific model keyword arguments in-place to make them JSON serializable. 

37 

38 :param kwargs: The keyword arguments to deserialize 

39 """ 

40 torch_import.check() 

41 

42 for k, v in kwargs.items(): 

43 # torch.dtype 

44 if isinstance(v, str) and v.startswith("torch."): 

45 dtype_str = v.split(".")[1] 

46 dtype = getattr(torch, dtype_str, None) 

47 if dtype is not None and isinstance(dtype, torch.dtype): 

48 kwargs[k] = dtype 

49 

50 if isinstance(v, dict): 

51 deserialize_hf_model_kwargs(v) 

52 

53 

54def convert_message_to_hf_format(message: ChatMessage) -> dict[str, Any]: 

55 """ 

56 Convert a message to the format expected by Hugging Face. 

57 

58 Note: ReasoningContent is skipped during conversion because the HuggingFace Inference API 

59 (which follows the OpenAI-compatible chat completion format) does not support reasoning 

60 in input messages. Reasoning is captured from model outputs for transparency but is not 

61 sent back to the API in multi-turn conversations. 

62 """ 

63 text_contents = message.texts 

64 tool_calls = message.tool_calls 

65 tool_call_results = message.tool_call_results 

66 images = message.images 

67 

68 # Filter out ReasoningContent from the content list for validation 

69 # ReasoningContent is for human transparency only, not sent to the API 

70 non_reasoning_content = [c for c in message._content if not isinstance(c, ReasoningContent)] 

71 

72 if not text_contents and not tool_calls and not tool_call_results and not images: 

73 raise ValueError( 

74 "A `ChatMessage` must contain at least one `TextContent`, `ToolCall`, `ToolCallResult`, or `ImageContent`." 

75 ) 

76 if len(tool_call_results) > 0 and len(non_reasoning_content) > 1: 

77 raise ValueError( 

78 "For compatibility with the Hugging Face API, a `ChatMessage` with a `ToolCallResult` " 

79 "cannot contain any other content." 

80 ) 

81 

82 # HF always expects a content field, even if it is empty 

83 hf_msg: dict[str, Any] = {"role": message._role.value, "content": ""} 

84 

85 if tool_call_results: 

86 result = tool_call_results[0] 

87 hf_msg["content"] = result.result 

88 if tc_id := result.origin.id: 

89 hf_msg["tool_call_id"] = tc_id 

90 # HF does not provide a way to communicate errors in tool invocations, so we ignore the error field 

91 return hf_msg 

92 

93 # Handle multimodal content (text + images) preserving order 

94 if text_contents or images: 

95 content_parts: list[dict[str, Any]] = [] 

96 for part in message._content: 

97 if isinstance(part, TextContent): 

98 content_parts.append({"type": "text", "text": part.text}) 

99 elif isinstance(part, ImageContent): 

100 image_url = f"data:{part.mime_type or 'image/jpeg'};base64,{part.base64_image}" 

101 content_parts.append({"type": "image_url", "image_url": {"url": image_url}}) 

102 

103 if len(content_parts) == 1 and not images: 

104 # content is a string 

105 hf_msg["content"] = content_parts[0]["text"] 

106 else: 

107 hf_msg["content"] = content_parts 

108 

109 if tool_calls: 

110 hf_tool_calls = [] 

111 for tc in tool_calls: 

112 hf_tool_call = {"type": "function", "function": {"name": tc.tool_name, "arguments": tc.arguments}} 

113 if tc.id is not None: 

114 hf_tool_call["id"] = tc.id 

115 hf_tool_calls.append(hf_tool_call) 

116 hf_msg["tool_calls"] = hf_tool_calls 

117 

118 return hf_msg