Skip to content

dspy.Adapter

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

Bases: ABC

Source code in dspy/adapters/base.py 
12
13
def __init__(self, callbacks: Optional[list[BaseCallback]] = None):
    self.callbacks = callbacks or []

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/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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]]:
    inputs_ = self.format(signature, demos, inputs)
    inputs_ = dict(prompt=inputs_) if isinstance(inputs_, str) else dict(messages=inputs_)

    outputs = lm(**inputs_, **lm_kwargs)
    values = []

    for output in outputs:
        output_logprobs = None

        if isinstance(output, dict):
            output, output_logprobs = output["text"], output["logprobs"]

        value = self.parse(signature, output)

        if set(value.keys()) != set(signature.output_fields.keys()):
            raise ValueError(
                "Parsed output fields do not match signature output fields. "
                f"Expected: {set(signature.output_fields.keys())}, Got: {set(value.keys())}"
            )

        if output_logprobs is not None:
            value["logprobs"] = output_logprobs

        values.append(value)

    return values

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

Source code in dspy/adapters/base.py 
51
52
53
@abstractmethod
def format(self, signature: Type[Signature], demos: list[dict[str, Any]], inputs: dict[str, Any]) -> list[dict[str, Any]]:
    raise NotImplementedError

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

Source code in dspy/adapters/base.py 
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
def format_conversation_history(self, signature: Type[Signature], inputs: dict[str, Any]) -> list[dict[str, Any]]:
    history_field_name = None
    for name, field in signature.input_fields.items():
        if field.annotation == History:
            history_field_name = name
            break

    if history_field_name is None:
        return []

    # 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 = 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(
            self.format_turn(signature_without_history, message, role="user", is_conversation_history=True)
        )
        messages.append(
            self.format_turn(signature_without_history, message, role="assistant", is_conversation_history=True)
        )

    inputs_copy = dict(inputs)
    del inputs_copy[history_field_name]

    messages.append(self.format_turn(signature_without_history, inputs_copy, role="user"))
    return messages

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/base.py 
62
63
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]]:
    raise NotImplementedError

format_turn(signature: Type[Signature], values, role: str, incomplete: bool = False, is_conversation_history: bool = False) -> dict[str, Any]

Source code in dspy/adapters/base.py 
65
66
def format_turn(self, signature: Type[Signature], values, role: str, incomplete: bool = False, is_conversation_history: bool = False) -> dict[str, Any]:
    raise NotImplementedError

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

Source code in dspy/adapters/base.py 
55
56
57
@abstractmethod
def parse(self, signature: Type[Signature], completion: str) -> dict[str, Any]:
    raise NotImplementedError