Skip to content

dspy.JSONAdapter

dspy.JSONAdapter()

Bases: Adapter

Source code in dspy/adapters/json_adapter.py 
31
32
def __init__(self):
    pass

Functions

__call__(lm, lm_kwargs, signature, demos, inputs)

Source code in dspy/adapters/json_adapter.py 
34
35
36
37
38
39
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
def __call__(self, lm, lm_kwargs, signature, demos, inputs):
    inputs = self.format(signature, demos, inputs)
    inputs = dict(prompt=inputs) if isinstance(inputs, str) else dict(messages=inputs)

    try:
        provider = lm.model.split("/", 1)[0] or "openai"
        if "response_format" in litellm.get_supported_openai_params(model=lm.model, custom_llm_provider=provider):
            try:
                response_format = _get_structured_outputs_response_format(signature)
                outputs = lm(**inputs, **lm_kwargs, response_format=response_format)
            except Exception:
                logger.debug(
                    "Failed to obtain response using signature-based structured outputs"
                    " response format: Falling back to default 'json_object' response format."
                    " Exception: {e}"
                )
                outputs = lm(**inputs, **lm_kwargs, response_format={"type": "json_object"})
        else:
            outputs = lm(**inputs, **lm_kwargs)

    except litellm.UnsupportedParamsError:
        outputs = lm(**inputs, **lm_kwargs)

    values = []

    for output in outputs:
        value = self.parse(signature, output)
        assert set(value.keys()) == set(
            signature.output_fields.keys()
        ), f"Expected {signature.output_fields.keys()} but got {value.keys()}"
        values.append(value)

    return values

format(signature, demos, inputs)

Source code in dspy/adapters/json_adapter.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def format(self, signature, demos, inputs):
    messages = []

    # Extract demos where some of the output_fields are not filled in.
    incomplete_demos = [demo for demo in demos if not all(k in demo for k in signature.fields)]
    complete_demos = [demo for demo in demos if demo not in incomplete_demos]
    incomplete_demos = [
        demo
        for demo in incomplete_demos
        if any(k in demo for k in signature.input_fields) and any(k in demo for k in signature.output_fields)
    ]

    demos = incomplete_demos + complete_demos

    messages.append({"role": "system", "content": prepare_instructions(signature)})

    for demo in demos:
        messages.append(format_turn(signature, demo, role="user", incomplete=demo in incomplete_demos))
        messages.append(format_turn(signature, demo, role="assistant", incomplete=demo in incomplete_demos))

    messages.append(format_turn(signature, inputs, role="user"))

    return messages

format_fields(signature, values, role)

Source code in dspy/adapters/json_adapter.py 
109
110
111
112
113
114
115
116
117
118
def format_fields(self, signature, values, role):
    fields_with_values = {
        FieldInfoWithName(name=field_name, info=field_info): values.get(
            field_name, "Not supplied for this particular example."
        )
        for field_name, field_info in signature.fields.items()
        if field_name in values
    }

    return format_fields(role=role, fields_with_values=fields_with_values)

format_turn(signature, values, role, incomplete=False)

Source code in dspy/adapters/json_adapter.py 
106
107
def format_turn(self, signature, values, role, incomplete=False):
    return format_turn(signature, values, role, incomplete)

parse(signature, completion)

Source code in dspy/adapters/json_adapter.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def parse(self, signature, completion):
    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