Skip to content

dspy.JSONAdapter

dspy.JSONAdapter(callbacks: Optional[list[BaseCallback]] = None)

Bases: ChatAdapter

Source code in dspy/adapters/chat_adapter.py 
29
30
def __init__(self, callbacks: Optional[list[BaseCallback]] = None):
    super().__init__(callbacks)

Functions

__call__(lm: LM, lm_kwargs: dict[str, Any], signature: Type[Signature], demos: list[dict[str, Any]], inputs: dict[str, Any]) -> list[dict[str, Any]]

Source code in dspy/adapters/json_adapter.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __call__(
    self,
    lm: LM,
    lm_kwargs: dict[str, Any],
    signature: Type[Signature],
    demos: list[dict[str, Any]],
    inputs: dict[str, Any],
) -> list[dict[str, Any]]:
    provider = lm.model.split("/", 1)[0] or "openai"
    params = litellm.get_supported_openai_params(model=lm.model, custom_llm_provider=provider)

    stream_listeners = settings.stream_listeners or []
    if len(stream_listeners) > 0:
        raise ValueError("Stream listener is not yet supported for JsonAdapter, please use ChatAdapter instead.")

    # If response_format is not supported, use basic call
    if not params or "response_format" not in params:
        return super().__call__(lm, lm_kwargs, signature, demos, inputs)

    # Check early for open-ended mapping types before trying structured outputs.
    if _has_open_ended_mapping(signature):
        lm_kwargs["response_format"] = {"type": "json_object"}
        return super().__call__(lm, lm_kwargs, signature, demos, inputs)

    # Try structured output first, fall back to basic JSON if it fails.
    try:
        structured_output_model = _get_structured_outputs_response_format(signature)
        lm_kwargs["response_format"] = structured_output_model
        return super().__call__(lm, lm_kwargs, signature, demos, inputs)
    except Exception as e:
        logger.warning(f"Failed to use structured output format. Falling back to JSON mode. Error: {e}")
        try:
            lm_kwargs["response_format"] = {"type": "json_object"}
            return super().__call__(lm, lm_kwargs, signature, demos, inputs)
        except Exception as e:
            raise RuntimeError(
                "Both structured output format and JSON mode failed. Please choose a model that supports "
                f"`response_format` argument. Original error: {e}"
            ) from e

format(signature: Type[Signature], demos: list[dict[str, Any]], inputs: dict[str, Any]) -> list[dict[str, Any]]

Format the input messages for the LM call.

This method converts the DSPy structured input along with few-shot examples and conversation history into multiturn messages as expected by the LM. For custom adapters, this method can be overridden to customize the formatting of the input messages.

In general we recommend the messages to have the following structure: 

[
    {"role": "system", "content": system_message},
    # Begin few-shot examples
    {"role": "user", "content": few_shot_example_1_input},
    {"role": "assistant", "content": few_shot_example_1_output},
    {"role": "user", "content": few_shot_example_2_input},
    {"role": "assistant", "content": few_shot_example_2_output},
    ...
    # End few-shot examples
    # Begin conversation history
    {"role": "user", "content": conversation_history_1_input},
    {"role": "assistant", "content": conversation_history_1_output},
    {"role": "user", "content": conversation_history_2_input},
    {"role": "assistant", "content": conversation_history_2_output},
    ...
    # End conversation history
    {"role": "user", "content": current_input},
]

And system message should contain the field description, field structure, and task description.

Parameters:

Name Type Description Default
signature Type[Signature]

The DSPy signature for which to format the input messages.

 required
demos list[dict[str, Any]]

A list of few-shot examples.

 required
inputs dict[str, Any]

The input arguments to the DSPy module.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of multiturn messages as expected by the LM.
Source code in dspy/adapters/base.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def format(
    self,
    signature: Type[Signature],
    demos: list[dict[str, Any]],
    inputs: dict[str, Any],
) -> list[dict[str, Any]]:
    """Format the input messages for the LM call.

    This method converts the DSPy structured input along with few-shot examples and conversation history into
    multiturn messages as expected by the LM. For custom adapters, this method can be overridden to customize
    the formatting of the input messages.

    In general we recommend the messages to have the following structure:
    ```
    [
        {"role": "system", "content": system_message},
        # Begin few-shot examples
        {"role": "user", "content": few_shot_example_1_input},
        {"role": "assistant", "content": few_shot_example_1_output},
        {"role": "user", "content": few_shot_example_2_input},
        {"role": "assistant", "content": few_shot_example_2_output},
        ...
        # End few-shot examples
        # Begin conversation history
        {"role": "user", "content": conversation_history_1_input},
        {"role": "assistant", "content": conversation_history_1_output},
        {"role": "user", "content": conversation_history_2_input},
        {"role": "assistant", "content": conversation_history_2_output},
        ...
        # End conversation history
        {"role": "user", "content": current_input},
    ]

    And system message should contain the field description, field structure, and task description.
    ```


    Args:
        signature: The DSPy signature for which to format the input messages.
        demos: A list of few-shot examples.
        inputs: The input arguments to the DSPy module.

    Returns:
        A list of multiturn messages as expected by the LM.
    """
    inputs_copy = dict(inputs)

    # If the signature and inputs have conversation history, we need to format the conversation history and
    # remove the history field from the signature.
    history_field_name = self._get_history_field_name(signature)
    if history_field_name:
        # In order to format the conversation history, we need to remove the history field from the signature.
        signature_without_history = signature.delete(history_field_name)
        conversation_history = self.format_conversation_history(
            signature_without_history,
            history_field_name,
            inputs_copy,
        )

    messages = []
    system_message = (
        f"{self.format_field_description(signature)}\n"
        f"{self.format_field_structure(signature)}\n"
        f"{self.format_task_description(signature)}"
    )
    messages.append({"role": "system", "content": system_message})
    messages.extend(self.format_demos(signature, demos))
    if history_field_name:
        # Conversation history and current input
        content = self.format_user_message_content(signature_without_history, inputs_copy, main_request=True)
        messages.extend(conversation_history)
        messages.append({"role": "user", "content": content})
    else:
        # Only current input
        content = self.format_user_message_content(signature, inputs_copy, main_request=True)
        messages.append({"role": "user", "content": content})

    messages = try_expand_image_tags(messages)
    return messages

format_assistant_message_content(signature: Type[Signature], outputs: dict[str, Any], missing_field_message=None) -> str

Source code in dspy/adapters/json_adapter.py 
111
112
113
114
115
116
117
118
119
120
121
def format_assistant_message_content(
    self,
    signature: Type[Signature],
    outputs: dict[str, Any],
    missing_field_message=None,
) -> str:
    fields_with_values = {
        FieldInfoWithName(name=k, info=v): outputs.get(k, missing_field_message)
        for k, v in signature.output_fields.items()
    }
    return self.format_field_with_value(fields_with_values, role="assistant")

format_conversation_history(signature: Type[Signature], history_field_name: str, inputs: dict[str, Any]) -> list[dict[str, Any]]

Format the conversation history.

This method formats the conversation history and the current input as multiturn messages.

Parameters:

Name Type Description Default
signature Type[Signature]

The DSPy signature for which to format the conversation history.

 required
history_field_name str

The name of the history field in the signature.

 required
inputs dict[str, Any]

The input arguments to the DSPy module.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of multiturn messages.
Source code in dspy/adapters/base.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def format_conversation_history(
    self,
    signature: Type[Signature],
    history_field_name: str,
    inputs: dict[str, Any],
) -> list[dict[str, Any]]:
    """Format the conversation history.

    This method formats the conversation history and the current input as multiturn messages.

    Args:
        signature: The DSPy signature for which to format the conversation history.
        history_field_name: The name of the history field in the signature.
        inputs: The input arguments to the DSPy module.

    Returns:
        A list of multiturn messages.
    """
    conversation_history = inputs[history_field_name].messages if history_field_name in inputs else None

    if conversation_history is None:
        return []

    messages = []
    for message in conversation_history:
        messages.append(
            {
                "role": "user",
                "content": self.format_user_message_content(signature, message),
            }
        )
        messages.append(
            {
                "role": "assistant",
                "content": self.format_assistant_message_content(signature, message),
            }
        )

    # Remove the history field from the inputs
    del inputs[history_field_name]

    return messages

format_demos(signature: Type[Signature], demos: list[dict[str, Any]]) -> list[dict[str, Any]]

Format the few-shot examples.

This method formats the few-shot examples as multiturn messages.

Parameters:

Name Type Description Default
signature Type[Signature]

The DSPy signature for which to format the few-shot examples.

 required
demos list[dict[str, Any]]

A list of few-shot examples, each element is a dictionary with keys of the input and output fields of the signature.

 required

Returns:

Type Description
list[dict[str, Any]]

A list of multiturn messages.
Source code in dspy/adapters/base.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def format_demos(self, signature: Type[Signature], demos: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Format the few-shot examples.

    This method formats the few-shot examples as multiturn messages.

    Args:
        signature: The DSPy signature for which to format the few-shot examples.
        demos: A list of few-shot examples, each element is a dictionary with keys of the input and output fields of
            the signature.

    Returns:
        A list of multiturn messages.
    """
    complete_demos = []
    incomplete_demos = []

    for demo in demos:
        # Check if all fields are present and not None
        is_complete = all(k in demo and demo[k] is not None for k in signature.fields)

        # Check if demo has at least one input and one output field
        has_input = any(k in demo for k in signature.input_fields)
        has_output = any(k in demo for k in signature.output_fields)

        if is_complete:
            complete_demos.append(demo)
        elif has_input and has_output:
            # We only keep incomplete demos that have at least one input and one output field
            incomplete_demos.append(demo)

    messages = []

    incomplete_demo_prefix = "This is an example of the task, though some input or output fields are not supplied."
    for demo in incomplete_demos:
        messages.append(
            {
                "role": "user",
                "content": self.format_user_message_content(signature, demo, prefix=incomplete_demo_prefix),
            }
        )
        messages.append(
            {
                "role": "assistant",
                "content": self.format_assistant_message_content(
                    signature, demo, missing_field_message="Not supplied for this particular example. "
                ),
            }
        )

    for demo in complete_demos:
        messages.append({"role": "user", "content": self.format_user_message_content(signature, demo)})
        messages.append(
            {
                "role": "assistant",
                "content": self.format_assistant_message_content(
                    signature, demo, missing_field_message="Not supplied for this conversation history message. "
                ),
            }
        )

    return messages

format_field_description(signature: Type[Signature]) -> str

Source code in dspy/adapters/chat_adapter.py 
52
53
54
55
56
def format_field_description(self, signature: Type[Signature]) -> str:
    return (
        f"Your input fields are:\n{get_field_description_string(signature.input_fields)}\n"
        f"Your output fields are:\n{get_field_description_string(signature.output_fields)}"
    )

format_field_structure(signature: Type[Signature]) -> str

Source code in dspy/adapters/json_adapter.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def format_field_structure(self, signature: Type[Signature]) -> str:
    parts = []
    parts.append("All interactions will be structured in the following way, with the appropriate values filled in.")

    def format_signature_fields_for_instructions(fields: Dict[str, FieldInfo]):
        return self.format_field_with_value(
            fields_with_values={
                FieldInfoWithName(name=field_name, info=field_info): translate_field_type(field_name, field_info)
                for field_name, field_info in fields.items()
            },
        )

    parts.append("Inputs will have the following structure:")
    parts.append(format_signature_fields_for_instructions(signature.input_fields))
    parts.append("Outputs will be a JSON object with the following fields.")
    parts.append(format_signature_fields_for_instructions(signature.output_fields))
    return "\n\n".join(parts).strip()

format_field_with_value(fields_with_values: Dict[FieldInfoWithName, Any], role: str = 'user') -> str

Formats the values of the specified fields according to the field's DSPy type (input or output), annotation (e.g. str, int, etc.), and the type of the value itself. Joins the formatted values into a single string, which is a multiline string if there are multiple fields.

Parameters:

Name Type Description Default
fields_with_values Dict[FieldInfoWithName, Any]

A dictionary mapping information about a field to its corresponding value.

 required

Returns: The joined formatted values of the fields, represented as a string.

Source code in dspy/adapters/json_adapter.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def format_field_with_value(self, fields_with_values: Dict[FieldInfoWithName, Any], role: str = "user") -> str:
    """
    Formats the values of the specified fields according to the field's DSPy type (input or output),
    annotation (e.g. str, int, etc.), and the type of the value itself. Joins the formatted values
    into a single string, which is a multiline string if there are multiple fields.

    Args:
        fields_with_values: A dictionary mapping information about a field to its corresponding value.
    Returns:
        The joined formatted values of the fields, represented as a string.
    """
    if role == "user":
        output = []
        for field, field_value in fields_with_values.items():
            formatted_field_value = format_field_value(field_info=field.info, value=field_value)
            output.append(f"[[ ## {field.name} ## ]]\n{formatted_field_value}")
        return "\n\n".join(output).strip()
    else:
        d = fields_with_values.items()
        d = {k.name: v for k, v in d}
        return json.dumps(serialize_for_json(d), indent=2)

format_finetune_data(signature: Type[Signature], demos: list[dict[str, Any]], inputs: dict[str, Any], outputs: dict[str, Any]) -> dict[str, list[Any]]

Source code in dspy/adapters/json_adapter.py 
163
164
165
166
167
def format_finetune_data(
    self, signature: Type[Signature], demos: list[dict[str, Any]], inputs: dict[str, Any], outputs: dict[str, Any]
) -> dict[str, list[Any]]:
    # TODO: implement format_finetune_data method in JSONAdapter
    raise NotImplementedError

format_task_description(signature: Type[Signature]) -> str

Source code in dspy/adapters/chat_adapter.py 
80
81
82
83
def format_task_description(self, signature: Type[Signature]) -> str:
    instructions = textwrap.dedent(signature.instructions)
    objective = ("\n" + " " * 8).join([""] + instructions.splitlines())
    return f"In adhering to this structure, your objective is: {objective}"

format_user_message_content(signature: Type[Signature], inputs: dict[str, Any], prefix: str = '', suffix: str = '', main_request: bool = False) -> str

Source code in dspy/adapters/chat_adapter.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def format_user_message_content(
    self,
    signature: Type[Signature],
    inputs: dict[str, Any],
    prefix: str = "",
    suffix: str = "",
    main_request: bool = False,
) -> str:
    messages = [prefix]
    for k, v in signature.input_fields.items():
        if k in inputs:
            value = inputs.get(k)
            formatted_field_value = format_field_value(field_info=v, value=value)
            messages.append(f"[[ ## {k} ## ]]\n{formatted_field_value}")

    if main_request:
        output_requirements = self.user_message_output_requirements(signature)
        if output_requirements is not None:
            messages.append(output_requirements)

    messages.append(suffix)
    return "\n\n".join(messages).strip()

parse(signature: Type[Signature], completion: str) -> dict[str, Any]

Source code in dspy/adapters/json_adapter.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def parse(self, signature: Type[Signature], completion: str) -> dict[str, Any]:
    pattern = r'\{(?:[^{}]|(?R))*\}'
    match = regex.search(pattern, completion, regex.DOTALL)  
    if match:  
        completion = match.group(0)
    fields = json_repair.loads(completion)
    fields = {k: v for k, v in fields.items() if k in signature.output_fields}

    # Attempt to cast each value to type signature.output_fields[k].annotation.
    for k, v in fields.items():
        if k in signature.output_fields:
            fields[k] = parse_value(v, signature.output_fields[k].annotation)

    if fields.keys() != signature.output_fields.keys():
        raise ValueError(f"Expected {signature.output_fields.keys()} but got {fields.keys()}")

    return fields

user_message_output_requirements(signature: Type[Signature]) -> str

Source code in dspy/adapters/json_adapter.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
def user_message_output_requirements(self, signature: Type[Signature]) -> str:
    def type_info(v):
        return (
            f" (must be formatted as a valid Python {get_annotation_name(v.annotation)})"
            if v.annotation is not str
            else ""
        )

    message = "Respond with a JSON object in the following order of fields: "
    message += ", then ".join(f"`{f}`{type_info(v)}" for f, v in signature.output_fields.items())
    message += "."
    return message