dspy.TwoStepAdapter

dspy.TwoStepAdapter(extraction_model: LM)

基类: Adapter

一个两阶段适配器，它

1. 为主语言模型 (main LM) 使用更简单、更自然的提示
2. 使用带有聊天适配器 (chat adapter) 的较小语言模型从主语言模型的响应中提取结构化数据

此适配器使用基类 Adapter 中定义的通用调用逻辑。此类在与推理模型作为主语言模型交互时特别有用，因为已知推理模型难以处理结构化输出。

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)

源代码位于 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

函数

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

源代码位于 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)
    return self._call_post_process(outputs, signature)

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

源代码位于 dspy/adapters/two_step_adapter.py

async def acall(
    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 = await lm.acall(messages=inputs, **lm_kwargs)
    # The signature is supposed to be "text -> {original output fields}"
    extractor_signature = self._create_extractor_signature(signature)

    values = []

    for output in outputs:
        output_logprobs = None

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

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

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

        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 方法而不是 format_field_description 或 format_field_structure。

参数

名称	类型	描述	默认值
signature	Type[Signature]	原始任务的签名	必需
demos	list[dict[str, Any]]	演示示例列表	必需
inputs	dict[str, Any]	当前输入	必需

返回值

类型	描述
list[dict[str, Any]]	要传递给主语言模型的消息列表。

源代码位于 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: Optional[str] = None) -> str

源代码位于 dspy/adapters/two_step_adapter.py

def format_assistant_message_content(
    self,
    signature: Type[Signature],
    outputs: dict[str, Any],
    missing_field_message: Optional[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]]

格式化对话历史。

此方法将对话历史和当前输入格式化为多轮消息。

参数

名称	类型	描述	默认值
signature	Type[Signature]	需要格式化对话历史的 DSPy 签名。	必需
history_field_name	str	签名中历史字段的名称。	必需
inputs	dict[str, Any]	DSPy 模块的输入参数。	必需

返回值

类型	描述
list[dict[str, Any]]	多轮消息列表。

源代码位于 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]]

格式化少样本示例。

此方法将少样本示例格式化为多轮消息。

参数

名称	类型	描述	默认值
signature	Type[Signature]	需要格式化少样本示例的 DSPy 签名。	必需
demos	list[dict[str, Any]]	少样本示例列表，每个元素是一个字典，其键为签名的输入和输出字段。	必需

返回值

类型	描述
list[dict[str, Any]]	多轮消息列表。

源代码位于 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

为系统消息格式化字段描述。

此方法为系统消息格式化字段描述。它应该返回一个包含输入字段和输出字段描述的字符串。

参数

名称	类型	描述	默认值
signature	Type[Signature]	需要格式化字段描述的 DSPy 签名。	必需

返回值

类型	描述
str	包含输入字段和输出字段描述的字符串。

源代码位于 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

为系统消息格式化字段结构。

此方法为系统消息格式化字段结构。它应该返回一个字符串，指定输入字段提供给语言模型的格式，以及响应中输出字段的格式。请参阅 ChatAdapter 和 JsonAdapter 获取示例。

参数

名称	类型	描述	默认值
signature	Type[Signature]	需要格式化字段结构的 DSPy 签名。	必需

源代码位于 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

格式化任务描述(signature: Signature) -> str

format_task_description(signature: Signature) -> str

源代码位于 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)

根据签名创建任务描述

源代码位于 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()

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

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

参数

名称	类型	描述	默认值
signature	Signature	原始任务的签名	必需
使用带有聊天适配器 (chat adapter) 的较小语言模型 (extraction_model) 从主语言模型的原始补全文本中提取结构化数据。	str	completion	必需

返回值

类型	描述
dict[str, Any]	主语言模型的补全文本

源代码位于 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