• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

deepset-ai / haystack / 19763653080

28 Nov 2025 12:20PM UTC coverage: 91.571% (+0.02%) from 91.552%
19763653080

Pull #10156

github

web-flow
Merge 3e20eec3d into 108204c07
Pull Request #10156: chore: Update code snippets in docs (audio and builders components)

13939 of 15222 relevant lines covered (91.57%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.21
haystack/components/builders/chat_prompt_builder.py
1
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2
#
3
# SPDX-License-Identifier: Apache-2.0
4

5
import json
1✔
6
from copy import deepcopy
1✔
7
from typing import Any, Literal, Optional, Union
1✔
8

9
from jinja2 import meta
1✔
10
from jinja2.sandbox import SandboxedEnvironment
1✔
11

12
from haystack import component, default_from_dict, default_to_dict, logging
1✔
13
from haystack.dataclasses.chat_message import ChatMessage, ChatRole, TextContent
1✔
14
from haystack.lazy_imports import LazyImport
1✔
15
from haystack.utils import Jinja2TimeExtension
1✔
16
from haystack.utils.jinja2_chat_extension import ChatMessageExtension, templatize_part
1✔
17

18
logger = logging.getLogger(__name__)
1✔
19

20
with LazyImport("Run 'pip install \"arrow>=1.3.0\"'") as arrow_import:
1✔
21
    import arrow  # pylint: disable=unused-import
1✔
22

23
NO_TEXT_ERROR_MESSAGE = "ChatMessages from {role} role must contain text. Received ChatMessage with no text: {message}"
1✔
24

25
FILTER_NOT_ALLOWED_ERROR_MESSAGE = (
1✔
26
    "The templatize_part filter cannot be used with a template containing a list of"
27
    "ChatMessage objects. Use a string template or remove the templatize_part filter "
28
    "from the template."
29
)
30

31

32
@component
1✔
33
class ChatPromptBuilder:
1✔
34
    """
35
    Renders a chat prompt from a template using Jinja2 syntax.
36

37
    A template can be a list of `ChatMessage` objects, or a special string, as shown in the usage examples.
38

39
    It constructs prompts using static or dynamic templates, which you can update for each pipeline run.
40

41
    Template variables in the template are optional unless specified otherwise.
42
    If an optional variable isn't provided, it defaults to an empty string. Use `variable` and `required_variables`
43
    to define input types and required variables.
44

45
    ### Usage examples
46

47
    #### Static ChatMessage prompt template
48

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
    ```
54

55
    #### Overriding static ChatMessage template at runtime
56

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.")
61

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
    ```
66

67
    #### Dynamic ChatMessage prompt template
68

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
74
    from haystack.utils import Secret
75

76
    # no parameter init, we don't use any runtime template variables
77
    prompt_builder = ChatPromptBuilder()
78
    llm = OpenAIChatGenerator(api_key=Secret.from_env_var("OPENAI_API_KEY"), model="gpt-5-mini")
79

80
    pipe = Pipeline()
81
    pipe.add_component("prompt_builder", prompt_builder)
82
    pipe.add_component("llm", llm)
83
    pipe.connect("prompt_builder.prompt", "llm.messages")
84

85
    location = "Berlin"
86
    language = "English"
87
    system_message = ChatMessage.from_system("You are an assistant giving information to tourists in {{language}}")
88
    messages = [system_message, ChatMessage.from_user("Tell me about {{location}}")]
89

90
    res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "language": language},
91
                                        "template": messages}})
92
    print(res)
93
    # Output example (truncated):
94
    # {'llm': {'replies': [ChatMessage(...)]}}
95

96
    messages = [system_message, ChatMessage.from_user("What's the forecast for {{location}}, next {{day_count}} days?")]
97

98
    res = pipe.run(data={"prompt_builder": {"template_variables": {"location": location, "day_count": "5"},
99
                                        "template": messages}})
100

101
    print(res)
102
    # Output example (truncated):
103
    # {'llm': {'replies': [ChatMessage(...)]}}
104
    ```
105

106
    #### String prompt template
107
    ```python
108
    from haystack.components.builders import ChatPromptBuilder
109
    from haystack.dataclasses.image_content import ImageContent
110

111
    template = \"\"\"
112
    {% message role="system" %}
113
    You are a helpful assistant.
114
    {% endmessage %}
115

116
    {% message role="user" %}
117
    Hello! I am {{user_name}}. What's the difference between the following images?
118
    {% for image in images %}
119
    {{ image | templatize_part }}
120
    {% endfor %}
121
    {% endmessage %}
122
    \"\"\"
123

124
    images = [ImageContent.from_file_path("test/test_files/images/apple.jpg"),
125
              ImageContent.from_file_path("test/test_files/images/haystack-logo.png")]
126

127
    builder = ChatPromptBuilder(template=template)
128
    builder.run(user_name="John", images=images)
129
    ```
130
    """
131

132
    def __init__(
1✔
133
        self,
134
        template: Optional[Union[list[ChatMessage], str]] = None,
135
        required_variables: Optional[Union[list[str], Literal["*"]]] = None,
136
        variables: Optional[list[str]] = None,
137
    ):
138
        """
139
        Constructs a ChatPromptBuilder component.
140

141
        :param template:
142
            A list of `ChatMessage` objects or a string template. The component looks for Jinja2 template syntax and
143
            renders the prompt with the provided variables. Provide the template in either
144
            the `init` method` or the `run` method.
145
        :param required_variables:
146
            List variables that must be provided as input to ChatPromptBuilder.
147
            If a variable listed as required is not provided, an exception is raised.
148
            If set to "*", all variables found in the prompt are required. Optional.
149
        :param variables:
150
            List input variables to use in prompt templates instead of the ones inferred from the
151
            `template` parameter. For example, to use more variables during prompt engineering than the ones present
152
            in the default template, you can provide them here.
153
        """
154
        self._variables = variables
1✔
155
        self._required_variables = required_variables
1✔
156
        self.template = template
1✔
157

158
        self._env = SandboxedEnvironment(extensions=[ChatMessageExtension])
1✔
159
        self._env.filters["templatize_part"] = templatize_part
1✔
160
        if arrow_import.is_successful():
1✔
161
            self._env.add_extension(Jinja2TimeExtension)
1✔
162

163
        extracted_variables = []
1✔
164
        if template and not variables:
1✔
165
            if isinstance(template, list):
1✔
166
                for message in template:
1✔
167
                    if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
1✔
168
                        # infer variables from template
169
                        if message.text is None:
1✔
170
                            raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
×
171
                        if message.text and "templatize_part" in message.text:
1✔
172
                            raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
1✔
173
                        ast = self._env.parse(message.text)
1✔
174
                        template_variables = meta.find_undeclared_variables(ast)
1✔
175
                        extracted_variables += list(template_variables)
1✔
176
            elif isinstance(template, str):
1✔
177
                ast = self._env.parse(template)
1✔
178
                extracted_variables = list(meta.find_undeclared_variables(ast))
1✔
179

180
        self.variables = variables or extracted_variables
1✔
181
        self.required_variables = required_variables or []
1✔
182

183
        if len(self.variables) > 0 and required_variables is None:
1✔
184
            logger.warning(
1✔
185
                "ChatPromptBuilder has {length} prompt variables, but `required_variables` is not set. "
186
                "By default, all prompt variables are treated as optional, which may lead to unintended behavior in "
187
                "multi-branch pipelines. To avoid unexpected execution, ensure that variables intended to be required "
188
                "are explicitly set in `required_variables`.",
189
                length=len(self.variables),
190
            )
191

192
        # setup inputs
193
        for var in self.variables:
1✔
194
            if self.required_variables == "*" or var in self.required_variables:
1✔
195
                component.set_input_type(self, var, Any)
1✔
196
            else:
197
                component.set_input_type(self, var, Any, "")
1✔
198

199
    @component.output_types(prompt=list[ChatMessage])
1✔
200
    def run(
1✔
201
        self,
202
        template: Optional[Union[list[ChatMessage], str]] = None,
203
        template_variables: Optional[dict[str, Any]] = None,
204
        **kwargs,
205
    ):
206
        """
207
        Renders the prompt template with the provided variables.
208

209
        It applies the template variables to render the final prompt. You can provide variables with pipeline kwargs.
210
        To overwrite the default template, you can set the `template` parameter.
211
        To overwrite pipeline kwargs, you can set the `template_variables` parameter.
212

213
        :param template:
214
            An optional list of `ChatMessage` objects or string template to overwrite ChatPromptBuilder's default
215
            template.
216
            If `None`, the default template provided at initialization is used.
217
        :param template_variables:
218
            An optional dictionary of template variables to overwrite the pipeline variables.
219
        :param kwargs:
220
            Pipeline variables used for rendering the prompt.
221

222
        :returns: A dictionary with the following keys:
223
            - `prompt`: The updated list of `ChatMessage` objects after rendering the templates.
224
        :raises ValueError:
225
            If `chat_messages` is empty or contains elements that are not instances of `ChatMessage`.
226
        """
227
        kwargs = kwargs or {}
1✔
228
        template_variables = template_variables or {}
1✔
229
        template_variables_combined = {**kwargs, **template_variables}
1✔
230

231
        if template is None:
1✔
232
            template = self.template
1✔
233

234
        if not template:
1✔
235
            raise ValueError(
1✔
236
                f"The {self.__class__.__name__} requires a non-empty list of ChatMessage instances. "
237
                f"Please provide a valid list of ChatMessage instances to render the prompt."
238
            )
239

240
        if isinstance(template, list) and not all(isinstance(message, ChatMessage) for message in template):
1✔
241
            raise ValueError(
1✔
242
                f"The {self.__class__.__name__} expects a list containing only ChatMessage instances. "
243
                f"The provided list contains other types. Please ensure that all elements in the list "
244
                f"are ChatMessage instances."
245
            )
246

247
        processed_messages = []
1✔
248
        if isinstance(template, list):
1✔
249
            for message in template:
1✔
250
                if message.is_from(ChatRole.USER) or message.is_from(ChatRole.SYSTEM):
1✔
251
                    self._validate_variables(set(template_variables_combined.keys()))
1✔
252
                    if message.text is None:
1✔
253
                        raise ValueError(NO_TEXT_ERROR_MESSAGE.format(role=message.role.value, message=message))
×
254
                    if message.text and "templatize_part" in message.text:
1✔
255
                        raise ValueError(FILTER_NOT_ALLOWED_ERROR_MESSAGE)
1✔
256
                    compiled_template = self._env.from_string(message.text)
1✔
257
                    rendered_text = compiled_template.render(template_variables_combined)
1✔
258
                    # deep copy the message to avoid modifying the original message
259
                    rendered_message: ChatMessage = deepcopy(message)
1✔
260
                    rendered_message._content = [TextContent(text=rendered_text)]
1✔
261
                    processed_messages.append(rendered_message)
1✔
262
                else:
263
                    processed_messages.append(message)
1✔
264
        elif isinstance(template, str):
1✔
265
            self._validate_variables(set(template_variables_combined.keys()))
1✔
266
            processed_messages = self._render_chat_messages_from_str_template(template, template_variables_combined)
1✔
267

268
        return {"prompt": processed_messages}
1✔
269

270
    def _render_chat_messages_from_str_template(
1✔
271
        self, template: str, template_variables: dict[str, Any]
272
    ) -> list[ChatMessage]:
273
        """
274
        Renders a chat message from a string template.
275

276
        This must be used in conjunction with the `ChatMessageExtension` Jinja2 extension
277
        and the `templatize_part` filter.
278
        """
279
        compiled_template = self._env.from_string(template)
1✔
280
        rendered = compiled_template.render(template_variables)
1✔
281

282
        messages = []
1✔
283
        for line in rendered.strip().split("\n"):
1✔
284
            line = line.strip()
1✔
285
            if line:
1✔
286
                messages.append(ChatMessage.from_dict(json.loads(line)))
1✔
287

288
        return messages
1✔
289

290
    def _validate_variables(self, provided_variables: set[str]):
1✔
291
        """
292
        Checks if all the required template variables are provided.
293

294
        :param provided_variables:
295
            A set of provided template variables.
296
        :raises ValueError:
297
            If no template is provided or if all the required template variables are not provided.
298
        """
299
        if self.required_variables == "*":
1✔
300
            required_variables = sorted(self.variables)
1✔
301
        else:
302
            required_variables = self.required_variables
1✔
303
        missing_variables = [var for var in required_variables if var not in provided_variables]
1✔
304
        if missing_variables:
1✔
305
            missing_vars_str = ", ".join(missing_variables)
1✔
306
            raise ValueError(
1✔
307
                f"Missing required input variables in ChatPromptBuilder: {missing_vars_str}. "
308
                f"Required variables: {required_variables}. Provided variables: {provided_variables}."
309
            )
310

311
    def to_dict(self) -> dict[str, Any]:
1✔
312
        """
313
        Returns a dictionary representation of the component.
314

315
        :returns:
316
            Serialized dictionary representation of the component.
317
        """
318
        template: Optional[Union[list[dict[str, Any]], str]] = None
1✔
319
        if isinstance(self.template, list):
1✔
320
            template = [m.to_dict() for m in self.template]
1✔
321
        elif isinstance(self.template, str):
1✔
322
            template = self.template
1✔
323

324
        return default_to_dict(
1✔
325
            self, template=template, variables=self._variables, required_variables=self._required_variables
326
        )
327

328
    @classmethod
1✔
329
    def from_dict(cls, data: dict[str, Any]) -> "ChatPromptBuilder":
1✔
330
        """
331
        Deserialize this component from a dictionary.
332

333
        :param data:
334
            The dictionary to deserialize and create the component.
335

336
        :returns:
337
            The deserialized component.
338
        """
339
        init_parameters = data["init_parameters"]
1✔
340
        template = init_parameters.get("template")
1✔
341
        if template:
1✔
342
            if isinstance(template, list):
1✔
343
                init_parameters["template"] = [ChatMessage.from_dict(d) for d in template]
1✔
344
            elif isinstance(template, str):
1✔
345
                init_parameters["template"] = template
1✔
346

347
        return default_from_dict(cls, data)
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc