dspy.TwoStepAdapter

dspy.TwoStepAdapter(extraction_model: LM)

Bases: Adapter

A two-stage adapter that

1. Uses a simpler, more natural prompt for the main LM
2. Uses a smaller LM with chat adapter to extract structured data from the response of main LM

This adapter uses a common call logic defined in base Adapter class. This class is particularly useful when interacting with reasoning models as the main LM since reasoning models are known to struggle with structured outputs.

Example:

import dspy
lm = dspy.LM(model="openai/o3-mini", max_tokens=10000, temperature = 1.0)
adapter = dspy.TwoStepAdapter(dspy.LM("openai/gpt-4o-mini"))
dspy.configure(lm=lm, adapter=adapter)
program = dspy.ChainOfThought("question->answer")
result = program("What is the capital of France?")
print(result)

Source code in dspy/adapters/two_step_adapter.py

def __init__(self, extraction_model: LM):
    if not isinstance(extraction_model, LM):
        raise ValueError("extraction_model must be an instance of LM")
    self.extraction_model = extraction_model

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

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)

    outputs = lm(messages=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 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]]

Format a prompt for the first stage with the main LM. This no specific structure is required for the main LM, we customize the format method instead of format_field_description or format_field_structure.

Parameters:

Name	Type	Description	Default
signature	Type[Signature]	The signature of the original task	required
demos	list[dict[str, Any]]	A list of demo examples	required
inputs	dict[str, Any]	The current input	required

Returns:

Type	Description
list[dict[str, Any]]	A list of messages to be passed to the main LM.

Source code in dspy/adapters/two_step_adapter.py

def format(
    self, signature: Type[Signature], demos: list[dict[str, Any]], inputs: dict[str, Any]
) -> list[dict[str, Any]]:
    """
    Format a prompt for the first stage with the main LM.
    This no specific structure is required for the main LM, we customize the format method
    instead of format_field_description or format_field_structure.

    Args:
        signature: The signature of the original task
        demos: A list of demo examples
        inputs: The current input

    Returns:
        A list of messages to be passed to the main LM.
    """
    messages = []

    # Create a task description for the main LM
    task_description = self.format_task_description(signature)
    messages.append({"role": "system", "content": task_description})

    messages.extend(self.format_demos(signature, demos))

    # Format the current input
    messages.append({"role": "user", "content": self.format_user_message_content(signature, inputs)})

    return messages

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

Source code in dspy/adapters/two_step_adapter.py

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

    for name in signature.output_fields.keys():
        if name in outputs:
            parts.append(f"{name}: {outputs.get(name, missing_field_message)}")

    return "\n\n".join(parts).strip()

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

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

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

Format the field description for the system message.

This method formats the field description for the system message. It should return a string that contains the field description for the input fields and the output fields.

Parameters:

Name	Type	Description	Default
signature	Type[Signature]	The DSPy signature for which to format the field description.	required

Returns:

Type	Description
str	A string that contains the field description for the input fields and the output fields.

Source code in dspy/adapters/base.py

def format_field_description(self, signature: Type[Signature]) -> str:
    """Format the field description for the system message.

    This method formats the field description for the system message. It should return a string that contains
    the field description for the input fields and the output fields.

    Args:
        signature: The DSPy signature for which to format the field description.

    Returns:
        A string that contains the field description for the input fields and the output fields.
    """
    raise NotImplementedError

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

Format the field structure for the system message.

This method formats the field structure for the system message. It should return a string that dictates the format the input fields should be provided to the LM, and the format the output fields will be in the response. Refer to the ChatAdapter and JsonAdapter for an example.

Parameters:

Name	Type	Description	Default
signature	Type[Signature]	The DSPy signature for which to format the field structure.	required

Source code in dspy/adapters/base.py

def format_field_structure(self, signature: Type[Signature]) -> str:
    """Format the field structure for the system message.

    This method formats the field structure for the system message. It should return a string that dictates the
    format the input fields should be provided to the LM, and the format the output fields will be in the response.
    Refer to the ChatAdapter and JsonAdapter for an example.

    Args:
        signature: The DSPy signature for which to format the field structure.
    """
    raise NotImplementedError

format_task_description(signature: Signature) -> str

Create a description of the task based on the signature

Source code in dspy/adapters/two_step_adapter.py

def format_task_description(self, signature: Signature) -> str:
    """Create a description of the task based on the signature"""
    parts = []

    parts.append("You are a helpful assistant that can solve tasks based on user input.")
    parts.append("As input, you will be provided with:\n" + get_field_description_string(signature.input_fields))
    parts.append("Your outputs must contain:\n" + get_field_description_string(signature.output_fields))
    parts.append("You should lay out your outputs in detail so that your answer can be understood by another agent")

    if signature.instructions:
        parts.append(f"Specific instructions: {signature.instructions}")

    return "\n".join(parts)

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

Source code in dspy/adapters/two_step_adapter.py

def format_user_message_content(
    self,
    signature: Type[Signature],
    inputs: dict[str, Any],
    prefix: str = "",
    suffix: str = "",
) -> str:
    parts = [prefix]

    for name in signature.input_fields.keys():
        if name in inputs:
            parts.append(f"{name}: {inputs.get(name, '')}")

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

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

Use a smaller LM (extraction_model) with chat adapter to extract structured data from the raw completion text of the main LM.

Parameters:

Name	Type	Description	Default
signature	Signature	The signature of the original task	required
completion	str	The completion from the main LM	required

Returns:

Type	Description
dict[str, Any]	A dictionary containing the extracted structured data.

Source code in dspy/adapters/two_step_adapter.py

def parse(self, signature: Signature, completion: str) -> dict[str, Any]:
    """
    Use a smaller LM (extraction_model) with chat adapter to extract structured data
    from the raw completion text of the main LM.

    Args:
        signature: The signature of the original task
        completion: The completion from the main LM

    Returns:
        A dictionary containing the extracted structured data.
    """
    # The signature is supposed to be "text -> {original output fields}"
    extractor_signature = self._create_extractor_signature(signature)

    try:
        # Call the smaller LM to extract structured data from the raw completion text with ChatAdapter
        parsed_result = ChatAdapter()(
            lm=self.extraction_model,
            lm_kwargs={},
            signature=extractor_signature,
            demos=[],
            inputs={"text": completion},
        )
        return parsed_result[0]

    except Exception as e:
        raise ValueError(f"Failed to parse response from the original completion: {completion}") from e