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

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, ChatRole, 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 has_content = bool(text_contents or tool_calls or tool_call_results or images) 

73 # We convert an assistant message with no content part into a message with empty content, which the API accepts 

74 if not has_content and not message.is_from(ChatRole.ASSISTANT): 

75 raise ValueError( 

76 f"A `ChatMessage` from `{message._role.value}` must contain at least one `TextContent`, `ToolCall`, " 

77 "`ToolCallResult`, or `ImageContent`. Only assistant messages can be empty." 

78 ) 

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

80 raise ValueError( 

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

82 "cannot contain any other content." 

83 ) 

84 

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

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

87 

88 if tool_call_results: 

89 result = tool_call_results[0] 

90 hf_msg["content"] = result.result 

91 if tc_id := result.origin.id: 

92 hf_msg["tool_call_id"] = tc_id 

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

94 return hf_msg 

95 

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

97 if text_contents or images: 

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

99 for part in message._content: 

100 if isinstance(part, TextContent): 

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

102 elif isinstance(part, ImageContent): 

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

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

105 

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

107 # content is a string 

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

109 else: 

110 hf_msg["content"] = content_parts 

111 

112 if tool_calls: 

113 hf_tool_calls = [] 

114 for tc in tool_calls: 

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

116 if tc.id is not None: 

117 hf_tool_call["id"] = tc.id 

118 hf_tool_calls.append(hf_tool_call) 

119 hf_msg["tool_calls"] = hf_tool_calls 

120 

121 return hf_msg