Coverage for haystack/components/builders/chat_prompt_builder.py: 98%
114 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 09:42 +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 OpenAIResponsesChatGenerator
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 = OpenAIResponsesChatGenerator(model="gpt-5.6-luna")
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["llm"]["replies"][0].text)
92 # >> Berlin is the capital city of Germany and one of the most vibrant and diverse cities in Europe. Here are some
93 # >> key things to know...Enjoy your time exploring the vibrant and dynamic capital of Germany!
95 messages = [system_message, ChatMessage.from_user("What's the weather forecast for {{location}} in the next {{day_count}} days?")]
97 res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "day_count": "5"},
98 "template": messages}})
100 print(res["llm"]["replies"][0].text)
101 # >> Here is the weather forecast for Berlin in the next 5 days: Day 1: Mostly cloudy with a high of 22°C (72°F)
102 # >> and...so it's always a good idea to check for updates closer to your visit.
103 ```
105 #### String prompt template
106 ```python
107 from haystack.components.builders import ChatPromptBuilder
108 from haystack.dataclasses.image_content import ImageContent
110 template = \"\"\"
111 {% message role="system" %}
112 You are a helpful assistant.
113 {% endmessage %}
115 {% message role="user" %}
116 Hello! I am {{user_name}}. What's the difference between the following images?
117 {% for image in images %}
118 {{ image | templatize_part }}
119 {% endfor %}
120 {% endmessage %}
121 \"\"\"
123 images = [ImageContent.from_file_path("test/test_files/images/apple.jpg"),
124 ImageContent.from_file_path("test/test_files/images/haystack-logo.png")]
126 builder = ChatPromptBuilder(template=template)
127 builder.run(user_name="John", images=images)
128 ```
129 """ # noqa: E501
131 def __init__(
132 self,
133 template: list[ChatMessage] | str | None = None,
134 required_variables: list[str] | Literal["*"] | None = "*",
135 variables: list[str] | None = None,
136 ) -> None:
137 """
138 Constructs a ChatPromptBuilder component.
140 :param template:
141 A list of `ChatMessage` objects or a string template. The component looks for Jinja2 template syntax and
142 renders the prompt with the provided variables. Provide the template in either
143 the `init` method` or the `run` method.
144 :param required_variables:
145 List variables that must be provided as input to ChatPromptBuilder.
146 Defaults to `"*"`, which marks every variable found in the prompt as required.
147 Pass an explicit list to only require a subset of the variables; any variable not listed becomes
148 optional and is replaced with an empty string in the rendered prompt when missing.
149 Set to `None` to mark every variable as optional.
150 :param variables:
151 List input variables to use in prompt templates instead of the ones inferred from the
152 `template` parameter. For example, to use more variables during prompt engineering than the ones present
153 in the default template, you can provide them here.
154 """
155 self._variables = variables
156 self._required_variables = required_variables
157 self.template = template
159 self._env = HaystackSandboxedEnvironment(extensions=[ChatMessageExtension])
160 if arrow_import.is_successful():
161 self._env.add_extension(Jinja2TimeExtension)
163 extracted_variables = []
164 if template and not variables:
165 if isinstance(template, list):
166 for message in template:
167 if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
168 # infer variables from template
169 if message.text is None:
170 raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
171 for text in message.texts:
172 if "templatize_part" in text:
173 raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
174 assigned_variables, template_variables = _extract_template_variables_and_assignments(
175 env=self._env, template=text
176 )
177 extracted_variables += list(template_variables - assigned_variables)
178 elif isinstance(template, str):
179 assigned_variables, template_variables = _extract_template_variables_and_assignments(
180 env=self._env, template=template
181 )
182 extracted_variables = list(template_variables - assigned_variables)
184 extracted_variables = extracted_variables or []
185 self.variables = variables or extracted_variables
186 self.required_variables = required_variables or []
188 if len(self.variables) > 0 and required_variables is None:
189 logger.warning(
190 "ChatPromptBuilder has {length} prompt variables and `required_variables` is explicitly set to "
191 "`None`. This treats all prompt variables as optional, which may lead to unintended behavior in "
192 "multi-branch pipelines. Only set `required_variables` to `None` if you intentionally want all "
193 "variables to be optional.",
194 length=len(self.variables),
195 )
197 # setup inputs
198 for var in self.variables:
199 if self.required_variables == "*" or var in self.required_variables:
200 component.set_input_type(self, var, Any)
201 else:
202 component.set_input_type(self, var, Any, "")
204 @component.output_types(prompt=list[ChatMessage])
205 def run(
206 self,
207 template: list[ChatMessage] | str | None = None,
208 template_variables: dict[str, Any] | None = None,
209 **kwargs: Any,
210 ) -> dict[str, list[ChatMessage]]:
211 """
212 Renders the prompt template with the provided variables.
214 It applies the template variables to render the final prompt. You can provide variables with pipeline kwargs.
215 To overwrite the default template, you can set the `template` parameter.
216 To overwrite pipeline kwargs, you can set the `template_variables` parameter.
218 :param template:
219 An optional list of `ChatMessage` objects or string template to overwrite ChatPromptBuilder's default
220 template.
221 If `None`, the default template provided at initialization is used.
222 :param template_variables:
223 An optional dictionary of template variables to overwrite the pipeline variables.
224 :param kwargs:
225 Pipeline variables used for rendering the prompt.
227 :returns: A dictionary with the following keys:
228 - `prompt`: The updated list of `ChatMessage` objects after rendering the templates.
229 :raises ValueError:
230 If `template` is empty or contains elements that are not instances of `ChatMessage`.
231 """
232 kwargs = kwargs or {}
233 template_variables = template_variables or {}
234 template_variables_combined = {**kwargs, **template_variables}
236 if template is None:
237 template = self.template
239 if not template:
240 raise ValueError(
241 f"The {self.__class__.__name__} requires a non-empty list of ChatMessage instances. "
242 f"Please provide a valid list of ChatMessage instances to render the prompt."
243 )
245 if isinstance(template, list) and not all(isinstance(message, ChatMessage) for message in template):
246 raise ValueError(
247 f"The {self.__class__.__name__} expects a list containing only ChatMessage instances. "
248 f"The provided list contains other types. Please ensure that all elements in the list "
249 f"are ChatMessage instances."
250 )
252 processed_messages = []
253 if isinstance(template, list):
254 for message in template:
255 if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
256 self._validate_variables(set(template_variables_combined.keys()))
257 if message.text is None:
258 raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
259 rendered_content = list(message._content)
260 for index, part in enumerate(rendered_content):
261 if isinstance(part, TextContent):
262 if "templatize_part" in part.text:
263 raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
264 rendered_text = self._env.from_string(part.text).render(template_variables_combined)
265 rendered_content[index] = TextContent(text=rendered_text)
266 # use dataclasses.replace to avoid in-place mutation of the original message
267 rendered_message: ChatMessage = replace(message, _content=rendered_content)
268 processed_messages.append(rendered_message)
269 else:
270 processed_messages.append(message)
271 elif isinstance(template, str):
272 self._validate_variables(set(template_variables_combined.keys()))
273 processed_messages = self._render_chat_messages_from_str_template(template, template_variables_combined)
275 return {"prompt": processed_messages}
277 def _render_chat_messages_from_str_template(
278 self, template: str, template_variables: dict[str, Any]
279 ) -> list[ChatMessage]:
280 """
281 Renders a chat message from a string template.
283 This must be used in conjunction with the `ChatMessageExtension` Jinja2 extension
284 and the `templatize_part` filter.
285 """
286 compiled_template = self._env.from_string(template)
287 rendered = compiled_template.render(template_variables)
289 messages = []
290 for line in rendered.strip().split("\n"):
291 line = line.strip()
292 if line:
293 messages.append(ChatMessage.from_dict(json.loads(line)))
295 return messages
297 def _validate_variables(self, provided_variables: set[str]) -> None:
298 """
299 Checks if all the required template variables are provided.
301 :param provided_variables:
302 A set of provided template variables.
303 :raises ValueError:
304 If no template is provided or if all the required template variables are not provided.
305 """
306 if self.required_variables == "*":
307 required_variables = sorted(self.variables)
308 else:
309 required_variables = self.required_variables
310 missing_variables = [var for var in required_variables if var not in provided_variables]
311 if missing_variables:
312 missing_vars_str = ", ".join(missing_variables)
313 raise ValueError(
314 f"Missing required input variables in ChatPromptBuilder: {missing_vars_str}. "
315 f"Required variables: {required_variables}. Provided variables: {provided_variables}."
316 )
318 def to_dict(self) -> dict[str, Any]:
319 """
320 Returns a dictionary representation of the component.
322 :returns:
323 Serialized dictionary representation of the component.
324 """
325 template: list[dict[str, Any]] | str | None = None
326 if isinstance(self.template, list):
327 template = [m.to_dict() for m in self.template]
328 elif isinstance(self.template, str):
329 template = self.template
331 return default_to_dict(
332 self, template=template, variables=self._variables, required_variables=self._required_variables
333 )
335 @classmethod
336 def from_dict(cls, data: dict[str, Any]) -> "ChatPromptBuilder":
337 """
338 Deserialize this component from a dictionary.
340 :param data:
341 The dictionary to deserialize and create the component.
343 :returns:
344 The deserialized component.
345 """
346 init_parameters = data["init_parameters"]
347 template = init_parameters.get("template")
348 if template:
349 if isinstance(template, list):
350 init_parameters["template"] = [ChatMessage.from_dict(d) for d in template]
351 elif isinstance(template, str):
352 init_parameters["template"] = template
354 return default_from_dict(cls, data)