Coverage for haystack/components/builders/chat_prompt_builder.py: 98%
110 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import json
6from dataclasses import replace
7from typing import Any, Literal
9from haystack import component, default_from_dict, default_to_dict, logging
10from haystack.dataclasses.chat_message import ChatMessage, ChatRole, TextContent
11from haystack.lazy_imports import LazyImport
12from haystack.utils import Jinja2TimeExtension
13from haystack.utils.jinja2_chat_extension import ChatMessageExtension
14from haystack.utils.jinja2_extensions import _extract_template_variables_and_assignments
15from haystack.utils.jinja2_sandbox import HaystackSandboxedEnvironment
17logger = logging.getLogger(__name__)
19with LazyImport("Run 'pip install \"arrow>=1.3.0\"'") as arrow_import:
20 import arrow # noqa: F401
22NO_TEXT_ERROR_MESSAGE = "ChatMessages from {role} role must contain text. Received ChatMessage with no text: {message}"
24FILTER_NOT_ALLOWED_ERROR_MESSAGE = (
25 "The templatize_part filter cannot be used with a template containing a list of"
26 "ChatMessage objects. Use a string template or remove the templatize_part filter "
27 "from the template."
28)
31@component
32class ChatPromptBuilder:
33 """
34 Renders a chat prompt from a template using Jinja2 syntax.
36 A template can be a list of `ChatMessage` objects, or a special string, as shown in the usage examples.
38 It constructs prompts using static or dynamic templates, which you can update for each pipeline run.
40 Template variables in the template are required by default. To make any subset of variables optional,
41 set `required_variables` to an explicit list of the variables that should remain required; any variable
42 not listed becomes optional and defaults to an empty string when missing.
43 Set `required_variables` to `None` to mark every variable as optional.
45 ### Usage examples
47 #### Static ChatMessage prompt template
49 ```python
50 template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")]
51 builder = ChatPromptBuilder(template=template)
52 builder.run(target_language="spanish", snippet="I can't speak spanish.")
53 ```
55 #### Overriding static ChatMessage template at runtime
57 ```python
58 template = [ChatMessage.from_user("Translate to {{ target_language }}. Context: {{ snippet }}; Translation:")]
59 builder = ChatPromptBuilder(template=template)
60 builder.run(target_language="spanish", snippet="I can't speak spanish.")
62 msg = "Translate to {{ target_language }} and summarize. Context: {{ snippet }}; Summary:"
63 summary_template = [ChatMessage.from_user(msg)]
64 builder.run(target_language="spanish", snippet="I can't speak spanish.", template=summary_template)
65 ```
67 #### Dynamic ChatMessage prompt template
69 ```python
70 from haystack.components.builders import ChatPromptBuilder
71 from haystack.components.generators.chat import OpenAIChatGenerator
72 from haystack.dataclasses import ChatMessage
73 from haystack import Pipeline
75 # no parameter init, we don't use any runtime template variables
76 prompt_builder = ChatPromptBuilder()
77 llm = OpenAIChatGenerator(model="gpt-5-mini")
79 pipe = Pipeline()
80 pipe.add_component("prompt_builder", prompt_builder)
81 pipe.add_component("llm", llm)
82 pipe.connect("prompt_builder.prompt", "llm.messages")
84 location = "Berlin"
85 language = "English"
86 system_message = ChatMessage.from_system("You are an assistant giving information to tourists in {{language}}")
87 messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")]
89 res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "language": language},
90 "template": messages}})
91 print(res)
92 # >> {'llm': {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text=
93 # "Berlin is the capital city of Germany and one of the most vibrant
94 # and diverse cities in Europe. Here are some key things to know...Enjoy your time exploring the vibrant and dynamic
95 # capital of Germany!")], _name=None, _meta={'model': 'gpt-5-mini',
96 # 'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 27, 'completion_tokens': 681, 'total_tokens':
97 # 708}})]}}
99 messages = [system_message, ChatMessage.from_user("What's the weather forecast for {{location}} in the next {{day_count}} days?")]
101 res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "day_count": "5"},
102 "template": messages}})
104 print(res)
105 # >> {'llm': {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text=
106 # "Here is the weather forecast for Berlin in the next 5
107 # days:\\n\\nDay 1: Mostly cloudy with a high of 22°C (72°F) and...so it's always a good idea to check for updates
108 # closer to your visit.")], _name=None, _meta={'model': 'gpt-5-mini',
109 # 'index': 0, 'finish_reason': 'stop', 'usage': {'prompt_tokens': 37, 'completion_tokens': 201,
110 # 'total_tokens': 238}})]}}
111 ```
113 #### String prompt template
114 ```python
115 from haystack.components.builders import ChatPromptBuilder
116 from haystack.dataclasses.image_content import ImageContent
118 template = \"\"\"
119 {% message role="system" %}
120 You are a helpful assistant.
121 {% endmessage %}
123 {% message role="user" %}
124 Hello! I am {{user_name}}. What's the difference between the following images?
125 {% for image in images %}
126 {{ image | templatize_part }}
127 {% endfor %}
128 {% endmessage %}
129 \"\"\"
131 images = [ImageContent.from_file_path("test/test_files/images/apple.jpg"),
132 ImageContent.from_file_path("test/test_files/images/haystack-logo.png")]
134 builder = ChatPromptBuilder(template=template)
135 builder.run(user_name="John", images=images)
136 ```
137 """ # noqa: E501
139 def __init__(
140 self,
141 template: list[ChatMessage] | str | None = None,
142 required_variables: list[str] | Literal["*"] | None = "*",
143 variables: list[str] | None = None,
144 ) -> None:
145 """
146 Constructs a ChatPromptBuilder component.
148 :param template:
149 A list of `ChatMessage` objects or a string template. The component looks for Jinja2 template syntax and
150 renders the prompt with the provided variables. Provide the template in either
151 the `init` method` or the `run` method.
152 :param required_variables:
153 List variables that must be provided as input to ChatPromptBuilder.
154 Defaults to `"*"`, which marks every variable found in the prompt as required.
155 Pass an explicit list to only require a subset of the variables; any variable not listed becomes
156 optional and is replaced with an empty string in the rendered prompt when missing.
157 Set to `None` to mark every variable as optional.
158 :param variables:
159 List input variables to use in prompt templates instead of the ones inferred from the
160 `template` parameter. For example, to use more variables during prompt engineering than the ones present
161 in the default template, you can provide them here.
162 """
163 self._variables = variables
164 self._required_variables = required_variables
165 self.template = template
167 self._env = HaystackSandboxedEnvironment(extensions=[ChatMessageExtension])
168 if arrow_import.is_successful():
169 self._env.add_extension(Jinja2TimeExtension)
171 extracted_variables = []
172 if template and not variables:
173 if isinstance(template, list):
174 for message in template:
175 if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
176 # infer variables from template
177 if message.text is None:
178 raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
179 if message.text and "templatize_part" in message.text:
180 raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
181 assigned_variables, template_variables = _extract_template_variables_and_assignments(
182 env=self._env, template=message.text
183 )
184 extracted_variables += list(template_variables - assigned_variables)
185 elif isinstance(template, str):
186 assigned_variables, template_variables = _extract_template_variables_and_assignments(
187 env=self._env, template=template
188 )
189 extracted_variables = list(template_variables - assigned_variables)
191 extracted_variables = extracted_variables or []
192 self.variables = variables or extracted_variables
193 self.required_variables = required_variables or []
195 if len(self.variables) > 0 and required_variables is None:
196 logger.warning(
197 "ChatPromptBuilder has {length} prompt variables and `required_variables` is explicitly set to "
198 "`None`. This treats all prompt variables as optional, which may lead to unintended behavior in "
199 "multi-branch pipelines. Only set `required_variables` to `None` if you intentionally want all "
200 "variables to be optional.",
201 length=len(self.variables),
202 )
204 # setup inputs
205 for var in self.variables:
206 if self.required_variables == "*" or var in self.required_variables:
207 component.set_input_type(self, var, Any)
208 else:
209 component.set_input_type(self, var, Any, "")
211 @component.output_types(prompt=list[ChatMessage])
212 def run(
213 self,
214 template: list[ChatMessage] | str | None = None,
215 template_variables: dict[str, Any] | None = None,
216 **kwargs: Any,
217 ) -> dict[str, list[ChatMessage]]:
218 """
219 Renders the prompt template with the provided variables.
221 It applies the template variables to render the final prompt. You can provide variables with pipeline kwargs.
222 To overwrite the default template, you can set the `template` parameter.
223 To overwrite pipeline kwargs, you can set the `template_variables` parameter.
225 :param template:
226 An optional list of `ChatMessage` objects or string template to overwrite ChatPromptBuilder's default
227 template.
228 If `None`, the default template provided at initialization is used.
229 :param template_variables:
230 An optional dictionary of template variables to overwrite the pipeline variables.
231 :param kwargs:
232 Pipeline variables used for rendering the prompt.
234 :returns: A dictionary with the following keys:
235 - `prompt`: The updated list of `ChatMessage` objects after rendering the templates.
236 :raises ValueError:
237 If `chat_messages` is empty or contains elements that are not instances of `ChatMessage`.
238 """
239 kwargs = kwargs or {}
240 template_variables = template_variables or {}
241 template_variables_combined = {**kwargs, **template_variables}
243 if template is None:
244 template = self.template
246 if not template:
247 raise ValueError(
248 f"The {self.__class__.__name__} requires a non-empty list of ChatMessage instances. "
249 f"Please provide a valid list of ChatMessage instances to render the prompt."
250 )
252 if isinstance(template, list) and not all(isinstance(message, ChatMessage) for message in template):
253 raise ValueError(
254 f"The {self.__class__.__name__} expects a list containing only ChatMessage instances. "
255 f"The provided list contains other types. Please ensure that all elements in the list "
256 f"are ChatMessage instances."
257 )
259 processed_messages = []
260 if isinstance(template, list):
261 for message in template:
262 if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
263 self._validate_variables(set(template_variables_combined.keys()))
264 if message.text is None:
265 raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
266 if message.text and "templatize_part" in message.text:
267 raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
268 compiled_template = self._env.from_string(message.text)
269 rendered_text = compiled_template.render(template_variables_combined)
270 # use dataclasses.replace to avoid in-place mutation of the original message
271 rendered_message: ChatMessage = replace(message, _content=[TextContent(text=rendered_text)])
272 processed_messages.append(rendered_message)
273 else:
274 processed_messages.append(message)
275 elif isinstance(template, str):
276 self._validate_variables(set(template_variables_combined.keys()))
277 processed_messages = self._render_chat_messages_from_str_template(template, template_variables_combined)
279 return {"prompt": processed_messages}
281 def _render_chat_messages_from_str_template(
282 self, template: str, template_variables: dict[str, Any]
283 ) -> list[ChatMessage]:
284 """
285 Renders a chat message from a string template.
287 This must be used in conjunction with the `ChatMessageExtension` Jinja2 extension
288 and the `templatize_part` filter.
289 """
290 compiled_template = self._env.from_string(template)
291 rendered = compiled_template.render(template_variables)
293 messages = []
294 for line in rendered.strip().split("\n"):
295 line = line.strip()
296 if line:
297 messages.append(ChatMessage.from_dict(json.loads(line)))
299 return messages
301 def _validate_variables(self, provided_variables: set[str]) -> None:
302 """
303 Checks if all the required template variables are provided.
305 :param provided_variables:
306 A set of provided template variables.
307 :raises ValueError:
308 If no template is provided or if all the required template variables are not provided.
309 """
310 if self.required_variables == "*":
311 required_variables = sorted(self.variables)
312 else:
313 required_variables = self.required_variables
314 missing_variables = [var for var in required_variables if var not in provided_variables]
315 if missing_variables:
316 missing_vars_str = ", ".join(missing_variables)
317 raise ValueError(
318 f"Missing required input variables in ChatPromptBuilder: {missing_vars_str}. "
319 f"Required variables: {required_variables}. Provided variables: {provided_variables}."
320 )
322 def to_dict(self) -> dict[str, Any]:
323 """
324 Returns a dictionary representation of the component.
326 :returns:
327 Serialized dictionary representation of the component.
328 """
329 template: list[dict[str, Any]] | str | None = None
330 if isinstance(self.template, list):
331 template = [m.to_dict() for m in self.template]
332 elif isinstance(self.template, str):
333 template = self.template
335 return default_to_dict(
336 self, template=template, variables=self._variables, required_variables=self._required_variables
337 )
339 @classmethod
340 def from_dict(cls, data: dict[str, Any]) -> "ChatPromptBuilder":
341 """
342 Deserialize this component from a dictionary.
344 :param data:
345 The dictionary to deserialize and create the component.
347 :returns:
348 The deserialized component.
349 """
350 init_parameters = data["init_parameters"]
351 template = init_parameters.get("template")
352 if template:
353 if isinstance(template, list):
354 init_parameters["template"] = [ChatMessage.from_dict(d) for d in template]
355 elif isinstance(template, str):
356 init_parameters["template"] = template
358 return default_from_dict(cls, data)