diff --git a/dspy/adapters/base.py b/dspy/adapters/base.py index d69ae43ee1..c122206a78 100644 --- a/dspy/adapters/base.py +++ b/dspy/adapters/base.py @@ -1,8 +1,5 @@ import json -import logging -from typing import Any, get_origin - -import json_repair +from typing import Any, get_args, get_origin from dspy.adapters._legacy_type_markers import ( _expand_legacy_custom_type_markers_in_chat_message, @@ -13,22 +10,16 @@ from dspy.adapters.types.tool import Tool, ToolCallResults, ToolCalls from dspy.adapters.utils import serialize_for_json from dspy.clients.base_lm import BaseLM -from dspy.clients.openai_format import ( - legacy_outputs_from_lm_response, - lm_response_from_legacy_outputs, - to_openai_chat_request, -) -from dspy.core.types import LMMessage, LMRequest, LMResponse +from dspy.clients.openai_format import lm_response_from_legacy_outputs, to_openai_chat_request +from dspy.core.types import LMRequest, LMResponse from dspy.experimental import Citations from dspy.signatures.field import InputField from dspy.signatures.signature import Signature from dspy.utils.callback import BaseCallback, with_callbacks from dspy.utils.exceptions import AdapterParseError -logger = logging.getLogger(__name__) - -_DEFAULT_NATIVE_RESPONSE_TYPES = [Citations, Reasoning] _TOOL_CALL_RESULTS_SIGNATURE = Signature({"tool_call_results": (ToolCallResults, InputField())}) +_SUPPORTED_CONTENT_BLOCKS = {"text", "image_url", "input_audio", "file", "document", "video"} class Adapter: @@ -71,103 +62,99 @@ def __init__( self.callbacks = callbacks or [] self.use_native_function_calling = use_native_function_calling self.parallel_tool_calls = parallel_tool_calls - self.native_response_types = native_response_types or _DEFAULT_NATIVE_RESPONSE_TYPES + self.native_response_types = native_response_types or [Citations, Reasoning] def __init_subclass__(cls, **kwargs) -> None: super().__init_subclass__(**kwargs) - - # Decorate format() and parse() method with with_callbacks cls.format = with_callbacks(cls.format) cls.parse = with_callbacks(cls.parse) + def __call__( + self, + lm: BaseLM, + lm_kwargs: dict[str, Any], + signature, + demos, + inputs, + ) -> list[dict[str, Any]]: + processed, request = self._prepare_call(lm, lm_kwargs, signature, demos, inputs) + return self._call_postprocess(processed, signature, _call_lm(lm, request), lm, lm_kwargs) + + async def acall( + self, + lm: BaseLM, + lm_kwargs: dict[str, Any], + signature, + demos, + inputs, + ) -> list[dict[str, Any]]: + processed, request = self._prepare_call(lm, lm_kwargs, signature, demos, inputs) + return self._call_postprocess(processed, signature, await _acall_lm(lm, request), lm, lm_kwargs) + + def _prepare_call(self, lm, lm_kwargs, signature, demos, inputs) -> tuple[type[Signature], LMRequest]: + signature = self._call_preprocess(lm, lm_kwargs, signature, inputs) + legacy_messages = self.format(signature, demos, inputs) + messages = [_preserve_unknown_content_blocks(message) for message in legacy_messages] + request = LMRequest.from_call(model=lm.model, messages=messages, **lm_kwargs) + messages = [_expand_legacy_custom_type_markers_in_lm_message(message) for message in request.messages] + return signature, request.model_copy(update={"messages": messages}) + def _call_preprocess( self, lm: BaseLM, lm_kwargs: dict[str, Any], signature: type[Signature], - inputs: dict[str, Any], + inputs, ) -> type[Signature]: - # TODO(adapters-plan): This remains the pre-normalized planning hook. It - # mutates `lm_kwargs` and returns only the render signature, which loses - # information we will need for plan-driven rendering/parsing. The next - # stacked PR should replace this with an explicit `_AdapterPlan` that - # records deleted fields, native tools, native output fields, inserted - # messages/parts, and LM config patches. + tool_input = _tool_input_field_name(signature) + tool_output = self._get_tool_call_output_field_name(signature) + if not self.use_native_function_calling: for key in ("tools", "tool_choice", "parallel_tool_calls"): lm_kwargs.pop(key, None) - else: - tool_call_input_field_name = self._get_tool_call_input_field_name(signature) - tool_call_output_field_name = self._get_tool_call_output_field_name(signature) - - if tool_call_output_field_name and tool_call_input_field_name is None: + elif tool_output: + if tool_input is None: raise ValueError( - f"You provided an output field {tool_call_output_field_name} to receive the tool calls information, " - "but did not provide any tools as the input. Please provide a list of tools as the input by adding an " - "input field with type `list[dspy.Tool]`." + f"You provided an output field {tool_output} to receive the tool calls information, but did not " + "provide any tools as the input. Please provide a list of tools as the input by adding an input " + "field with type `list[dspy.Tool]`." ) - - if tool_call_output_field_name and lm.supports_function_calling: - tools = inputs[tool_call_input_field_name] + if lm.supports_function_calling: + tools = inputs[tool_input] tools = tools if isinstance(tools, list) else [tools] - - lm_tools = [tool.format_as_litellm_function_call() for tool in tools] - - lm_kwargs["tools"] = lm_tools - if self.parallel_tool_calls is not None and lm_kwargs.get("parallel_tool_calls") is None: - lm_kwargs["parallel_tool_calls"] = self.parallel_tool_calls - - signature = signature.delete(tool_call_output_field_name) - signature = signature.delete(tool_call_input_field_name) - - # TODO(adapters-plan): Built-in/native response planning should move out - # of `Type.adapt_to_native_lm_feature()` and into adapter-owned planning - # renderers. Keep this compatibility hook for this boundary-only PR. - # Handle custom types that use native LM features, e.g., reasoning, citations, etc. + lm_kwargs["tools"] = [tool.format_as_litellm_function_call() for tool in tools] + if self.parallel_tool_calls is not None: + lm_kwargs.setdefault("tool_choice", "auto") + if lm_kwargs.get("parallel_tool_calls") is None: + lm_kwargs["parallel_tool_calls"] = self.parallel_tool_calls + signature = signature.delete(tool_output).delete(tool_input) for name, field in signature.output_fields.items(): - if ( - isinstance(field.annotation, type) - and field.annotation in self.native_response_types - and issubclass(field.annotation, Type) - ): + if self._is_native_response_type(field.annotation): signature = field.annotation.adapt_to_native_lm_feature(signature, name, lm, lm_kwargs) - return signature def _call_postprocess( self, - processed_signature: type[Signature], - original_signature: type[Signature], - outputs: list[dict[str, Any] | str], - lm: BaseLM, - lm_kwargs: dict[str, Any], + processed_signature, + original_signature, + response: LMResponse, + lm, + lm_kwargs, ) -> list[dict[str, Any]]: - # TODO(adapters-plan): This still parses legacy adapter output objects. - # PR1 normalizes the LM boundary, then immediately converts back to this - # shape to avoid changing parser semantics. A later PR should parse - # `LMResponse` directly and merge text-parsed fields with explicit - # native fields from `_AdapterPlan`. - values = [] - - tool_call_output_field_name = self._get_tool_call_output_field_name(original_signature) - - for output in outputs: - output_logprobs = None - tool_calls = None - text = output - - if isinstance(output, dict): - text = output["text"] - output_logprobs = output.get("logprobs") - tool_calls = output.get("tool_calls") - - if text and not (tool_calls and tool_call_output_field_name): + if not isinstance(response, LMResponse): + response = lm_response_from_legacy_outputs(response, LMRequest(model=getattr(lm, "model", ""), messages=[])) + + values, tool_output = [], self._get_tool_call_output_field_name(original_signature) + + for output in response.outputs: + if output.metadata.get("empty_provider_outputs") or output.metadata.get("empty_legacy_outputs"): + continue + + text, logprobs, tool_calls = output.text, output.logprobs, output.tool_calls + if tool_calls and tool_output: + value = self._parse_or_empty(processed_signature, text) + elif text: value = self.parse(processed_signature, text) - elif tool_calls and tool_call_output_field_name: - try: - value = self.parse(processed_signature, text) if text and processed_signature.output_fields else {} - except AdapterParseError: - value = {} else: raise AdapterParseError( adapter_name=type(self).__name__, @@ -176,192 +163,30 @@ def _call_postprocess( message="The LM returned an empty or null response.", ) - # Fields removed for native features are absent from the processed parse. for field_name in original_signature.output_fields: value.setdefault(field_name, None) - if tool_calls and tool_call_output_field_name: - tool_calls = [_provider_tool_call_to_tool_call_dict(tool_call) for tool_call in tool_calls] - value[tool_call_output_field_name] = ToolCalls.from_dict_list(tool_calls) + if tool_calls and tool_output: + value[tool_output] = _provider_tool_calls(tool_calls) - # TODO(adapter-types): Once `Type.parse_lm_output(context, output)` is - # the real normalized hook, this should not call the legacy - # provider-shaped `parse_lm_response()` directly. - # Parse custom types that does not rely on the `Adapter.parse()` method for name, field in original_signature.output_fields.items(): if ( - isinstance(field.annotation, type) - and field.annotation in self.native_response_types - and issubclass(field.annotation, Type) + self._is_native_response_type(field.annotation) + and (parsed := field.annotation.parse_lm_output(output)) is not None ): - parsed_value = field.annotation.parse_lm_response(output) - if parsed_value is not None: - value[name] = parsed_value + value[name] = parsed - if output_logprobs: - value["logprobs"] = output_logprobs + if logprobs: + value["logprobs"] = logprobs values.append(value) - return values - def _render_request( - self, - lm: BaseLM, - lm_kwargs: dict[str, Any], - messages: list[LMMessage | dict[str, Any]], - ) -> LMRequest: - """Build the normalized LM request for the current adapter call path. - - TODO(adapters-plan): This currently receives already-rendered messages. - Once planning lands, this should render from `_AdapterPlan` and apply - planned message/part insertions before creating `LMRequest`. - """ - return LMRequest.from_call( - model=lm.model, - messages=self._coerce_lm_messages(messages), - **lm_kwargs, - ) - - def _call_lm(self, lm: BaseLM, request: LMRequest) -> LMResponse: - """Call current `BaseLM` through the normalized request/response boundary. - - TODO(language-models): When `BaseLM` is replaced by/updated to the - normalized `BaseLM.forward(request: LMRequest) -> LMResponse` contract, - remove this compatibility shim and let adapters call the normalized LM - entry point directly. The OpenAI-shaped compatibility kwargs should live - only inside concrete LM backends. - """ - data = self._legacy_call_kwargs(request) - outputs = lm(messages=data.pop("messages"), **data) - return self._normalize_legacy_outputs(outputs, request) - - async def _acall_lm(self, lm: BaseLM, request: LMRequest) -> LMResponse: - """Async variant of `_call_lm`. - - TODO(language-models): Same transitional boundary as `_call_lm()`; this - should eventually call a normalized async LM method directly. - """ - data = self._legacy_call_kwargs(request) - outputs = await lm.acall(messages=data.pop("messages"), **data) - return self._normalize_legacy_outputs(outputs, request) - - def _legacy_call_kwargs(self, request: LMRequest) -> dict[str, Any]: - # TODO(language-models): Current `BaseLM` expects OpenAI/LiteLLM-shaped - # chat kwargs. We intentionally use `dspy.clients.openai_format` here so - # the conversion code lives in the future LM/client layer, not in - # adapters. Remove this adapter helper once `BaseLM` accepts `LMRequest`. - data = to_openai_chat_request(request) - data.pop("model", None) - # TODO(language-models): `cache` and `rollout_id` are DSPy BaseLM - # execution controls, not provider request fields. The future - # normalized LM base should own them before provider-format conversion. - if request.config.cache is not None: - if request.config.cache.enabled is not None: - data["cache"] = request.config.cache.enabled - if request.config.cache.rollout_id is not None: - data["rollout_id"] = request.config.cache.rollout_id - return data - - def _coerce_lm_messages(self, messages: list[LMMessage | dict[str, Any]]) -> list[LMMessage]: - """Normalize subclass `format()` output before the LM boundary. - - TODO(adapters-normalized-rendering): Adapter `format()` methods still - return OpenAI-chat-shaped dictionaries. This coercion is the bridge until - adapters render `LMMessage` / `LMPart` directly. - """ - return [ - _expand_legacy_custom_type_markers_in_lm_message( - message if isinstance(message, LMMessage) else self._chat_dict_to_lm_message(message) - ) - for message in messages - ] - - def _chat_dict_to_lm_message(self, message: dict[str, Any]) -> LMMessage: + def _parse_or_empty(self, signature, text): try: - return LMMessage(**message) - except Exception: - # TODO(legacy-custom-types): Unknown OpenAI content blocks are - # temporarily preserved as `legacy_content_block` metadata on an - # empty text part so `openai_format` can round-trip them back to - # current BaseLM calls. Replace this with either explicit opaque - # provider parts or remove it when marker-based custom type - # serialization is retired. - message = dict(message) - content = message.get("content") - if isinstance(content, list): - sanitized = [] - supported = {"text", "image_url", "input_audio", "file", "document", "video"} - for block in content: - if isinstance(block, dict) and block.get("type") in supported: - sanitized.append(block) - elif isinstance(block, dict): - sanitized.append({"type": "text", "text": "", "metadata": {"legacy_content_block": block}}) - else: - sanitized.append({"type": "text", "text": json.dumps(block, ensure_ascii=False)}) - message["content"] = sanitized - return LMMessage(**message) - - def _normalize_legacy_outputs(self, outputs: list[dict[str, Any] | str | None], request: LMRequest) -> LMResponse: - """Convert current `BaseLM` outputs into a normalized `LMResponse`. - - TODO(language-models): Current `BaseLM` returns `list[str | dict | None]`. - Future LMs should return `LMResponse` directly, making this method a - compatibility-only path for old/custom LMs. - """ - return lm_response_from_legacy_outputs(outputs, request) - - def __call__( - self, - lm: BaseLM, - lm_kwargs: dict[str, Any], - signature: type[Signature], - demos: list[dict[str, Any]], - inputs: dict[str, Any], - ) -> list[dict[str, Any]]: - """ - Execute the adapter pipeline: format inputs, call LM, and parse outputs. - - Args: - lm: The Language Model instance to use for generation. Must be an instance of `dspy.BaseLM`. - lm_kwargs: Additional keyword arguments to pass to the LM call (e.g., temperature, max_tokens). These are - passed directly to the LM. - signature: The DSPy signature associated with this LM call. - demos: List of few-shot examples to include in the prompt. Each dictionary should contain keys matching the - signature's input and output field names. Examples are formatted as user/assistant message pairs. - inputs: The current input values for this call. Keys must match the signature's input field names. - - Returns: - List of dictionaries representing parsed LM responses. Each dictionary contains keys matching the - signature's output field names. For multiple generations (n > 1), returns multiple dictionaries. - """ - processed_signature = self._call_preprocess(lm, lm_kwargs, signature, inputs) - messages = self.format(processed_signature, demos, inputs) - request = self._render_request(lm, lm_kwargs, messages) - response = self._call_lm(lm, request) - # TODO(adapters-response): We normalize at the LM boundary, but still - # convert back to legacy postprocess dictionaries here to keep this PR - # behavior-preserving. Replace with direct `LMResponse` parsing once the - # explicit adapter plan exists. - outputs = legacy_outputs_from_lm_response(response) - return self._call_postprocess(processed_signature, signature, outputs, lm, lm_kwargs) - - async def acall( - self, - lm: BaseLM, - lm_kwargs: dict[str, Any], - signature: type[Signature], - demos: list[dict[str, Any]], - inputs: dict[str, Any], - ) -> list[dict[str, Any]]: - processed_signature = self._call_preprocess(lm, lm_kwargs, signature, inputs) - messages = self.format(processed_signature, demos, inputs) - request = self._render_request(lm, lm_kwargs, messages) - response = await self._acall_lm(lm, request) - # TODO(adapters-response): Keep in sync with `__call__()` until both use - # direct `LMResponse` parsing. - outputs = legacy_outputs_from_lm_response(response) - return self._call_postprocess(processed_signature, signature, outputs, lm, lm_kwargs) + return self.parse(signature, text) if text and signature.output_fields else {} + except AdapterParseError: + return {} def format( self, @@ -371,395 +196,217 @@ def format( ) -> 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}, + This method converts DSPy structured input, few-shot examples, and conversation history into multiturn messages + as expected by the LM. Custom adapters can override it to customize message formatting. + """ + inputs = dict(inputs) + history_name = next( + (name for name, field in signature.input_fields.items() if field.annotation == History), + None, + ) + prompt_signature = signature.delete(history_name) if history_name else signature + messages = [ + {"role": "system", "content": self.format_system_message(signature)}, + *self.format_demos(signature, demos), ] - 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, - ) + if history_name: + messages.extend(self.format_conversation_history(prompt_signature, history_name, inputs)) - messages = [] - system_message = self.format_system_message(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) - if content: - messages.append({"role": "user", "content": content}) - else: - # Only current input - content = self.format_user_message_content(signature, inputs_copy, main_request=True) - if content: - messages.append({"role": "user", "content": content}) + if content := self.format_user_message_content(prompt_signature, inputs, main_request=True): + messages.append({"role": "user", "content": content}) return [_expand_legacy_custom_type_markers_in_chat_message(message) for message in messages] def format_system_message(self, signature: type[Signature]) -> str: - """Format the system message for the LM call. - - - Args: - signature: The DSPy signature for which to format the system message. - """ + """Format the system message for the LM call.""" return ( f"{self.format_field_description(signature)}\n" f"{self.format_field_structure(signature)}\n" f"{self.format_task_description(signature)}" ) - 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 - - 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 + def format_demos(self, signature: type[Signature], demos: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Format the few-shot examples.""" + messages = [] + incomplete_prefix = "This is an example of the task, though some input or output fields are not supplied." - def format_task_description(self, signature: type[Signature]) -> str: - """Format the task description for the system message. + for demo in demos: + has_input = any(k in demo for k in signature.input_fields) + has_output = any(k in demo for k in signature.output_fields) - This method formats the task description for the system message. In most cases this is just a thin wrapper - over `signature.instructions`. + if all(k in demo and demo[k] is not None for k in signature.fields): + prefix, missing = "", "Not supplied for this conversation history message. " + elif has_input and has_output: + prefix, missing = incomplete_prefix, "Not supplied for this particular example. " + else: + continue - Args: - signature: The DSPy signature of the DSpy module. + messages.extend( + [ + {"role": "user", "content": self.format_user_message_content(signature, demo, prefix=prefix)}, + { + "role": "assistant", + "content": self.format_assistant_message_content( + signature, + demo, + missing_field_message=missing, + ), + }, + ] + ) - Returns: - A string that describes the task. - """ - raise NotImplementedError + return messages - def format_user_message_content( + def format_conversation_history( self, signature: type[Signature], + history_field_name: str, inputs: dict[str, Any], - prefix: str = "", - suffix: str = "", - main_request: bool = False, - ) -> str: - """Format the user message content. - - This method formats the user message content, which can be used in formatting few-shot examples, conversation - history, and the current input. - - Args: - signature: The DSPy signature for which to format the user message content. - inputs: The input arguments to the DSPy module. - prefix: A prefix to the user message content. - suffix: A suffix to the user message content. - - Returns: - A string that contains the user message content. - """ - raise NotImplementedError - - def format_assistant_message_content( - self, - signature: type[Signature], - outputs: dict[str, Any], - missing_field_message: str | None = None, - ) -> str: - """Format the assistant message content. - - This method formats the assistant message content, which can be used in formatting few-shot examples, - conversation history. - - Args: - signature: The DSPy signature for which to format the assistant message content. - outputs: The output fields to be formatted. - missing_field_message: A message to be used when a field is missing. - - Returns: - A string that contains the assistant message content. - """ - raise NotImplementedError - - 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. + ) -> list[dict[str, Any]]: + """Format the conversation history.""" + history = inputs[history_field_name].messages if history_field_name in inputs else None + if history is None: + return [] + messages = [message for item in history for message in self._history_messages(signature, item)] + del inputs[history_field_name] + return messages - Returns: - A list of multiturn messages. - """ - complete_demos = [] - incomplete_demos = [] + def _history_messages(self, signature, message) -> list[dict[str, Any]]: + tool_field, tool_calls = _tool_calls_from_message(message) + tool_results = ( + ToolCallResults.model_validate(tool_calls.tool_call_results) + if tool_calls and tool_calls.tool_call_results is not None + else None + ) - 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) + messages = [] + if content := self.format_user_message_content(signature, message): + messages.append({"role": "user", "content": content}) - # 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 self.use_native_function_calling and tool_calls: + return messages + self._native_tool_history_messages(signature, message, tool_calls, tool_results) - 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) + assistant_values = message + if tool_field and tool_results: + assistant_values = dict(message, **{tool_field: tool_calls.model_copy(update={"tool_call_results": None})}) - messages = [] + if content := self.format_assistant_message_content(signature, assistant_values): + messages.append({"role": "assistant", "content": content}) - incomplete_demo_prefix = "This is an example of the task, though some input or output fields are not supplied." - for demo in incomplete_demos: + if tool_results: 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. " + "content": self.format_user_message_content( + _TOOL_CALL_RESULTS_SIGNATURE, + {"tool_call_results": tool_results}, ), } ) return messages - def _get_history_field_name(self, signature: type[Signature]) -> bool: - for name, field in signature.input_fields.items(): - if field.annotation == History: - return name - return None - - def _get_tool_call_input_field_name(self, signature: type[Signature]) -> bool: - for name, field in signature.input_fields.items(): - # Look for annotation `list[dspy.Tool]` or `dspy.Tool` - origin = get_origin(field.annotation) - if origin is list and field.annotation.__args__[0] == Tool: - return name - if field.annotation == Tool: - return name - return None - - def _get_tool_call_output_field_name(self, signature: type[Signature]) -> bool: + def _native_tool_history_messages(self, signature, message, tool_calls, tool_results) -> list[dict[str, Any]]: + content_signature = signature for name, field in signature.output_fields.items(): - if field.annotation == ToolCalls: - return name - return None - - 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. + if field.annotation == ToolCalls or message.get(name) is None: + content_signature = content_signature.delete(name) + content = "" + if content_signature.output_fields: + content = self.format_assistant_message_content(content_signature, message) - Returns: - A list of multiturn messages as expected by the LM. - """ - conversation_history = inputs[history_field_name].messages if history_field_name in inputs else None - - if conversation_history is None: - return [] + ids = [tool_call.id for tool_call in tool_calls.tool_calls] + result_ids = [result.call_id for result in tool_results.tool_call_results] if tool_results else [] + results = tool_results if tool_results and ids == result_ids and all(ids) else None messages = [] - for message in conversation_history: - tool_call_field_name, tool_calls = _tool_calls_from_message(message) - tool_call_results = ( - ToolCallResults.model_validate(tool_calls.tool_call_results) - if tool_calls is not None and tool_calls.tool_call_results is not None - else None - ) - - user_content = self.format_user_message_content(signature, message) - if user_content: - messages.append({"role": "user", "content": user_content}) + if content or results: + assistant_message = {"role": "assistant", "content": content or None} + if results: + assistant_message["tool_calls"] = [ + _tool_call_as_openai_message_tool_call(tool_call) for tool_call in tool_calls.tool_calls + ] + messages.append(assistant_message) - if self.use_native_function_calling and tool_calls is not None: - content_signature = signature - for name, field in signature.output_fields.items(): - if field.annotation == ToolCalls or message.get(name) is None: - content_signature = content_signature.delete(name) + if results: + messages.extend(_tool_result_as_openai_message(result) for result in results.tool_call_results) - content = ( - self.format_assistant_message_content(content_signature, message) - if content_signature.output_fields - else "" - ) + return messages - if tool_call_results is not None: - tool_call_ids = [tool_call.id for tool_call in tool_calls.tool_calls] - result_ids = [result.call_id for result in tool_call_results.tool_call_results] - if tool_call_ids != result_ids or not all(tool_call_ids): - tool_call_results = None - - if content or tool_call_results is not None: - assistant_message: dict[str, Any] = {"role": "assistant", "content": content or None} - if tool_call_results is not None: - assistant_message["tool_calls"] = [ - _tool_call_as_openai_message_tool_call(tool_call) for tool_call in tool_calls.tool_calls - ] - messages.append(assistant_message) - - if tool_call_results is not None: - for result in tool_call_results.tool_call_results: - content = _tool_result_content(result.value) - messages.append( - {"role": "tool", "tool_call_id": result.call_id, "name": result.name, "content": content} - ) - continue + def _get_tool_call_output_field_name(self, signature: type[Signature]) -> str | None: + return next((name for name, field in signature.output_fields.items() if field.annotation == ToolCalls), None) - assistant_values = message - if tool_call_field_name is not None and tool_call_results is not None: - assistant_values = dict(message) - assistant_values[tool_call_field_name] = tool_calls.model_copy(update={"tool_call_results": None}) + def _is_native_response_type(self, annotation) -> bool: + return ( + isinstance(annotation, type) + and annotation in self.native_response_types + and issubclass(annotation, Type) + ) - assistant_content = self.format_assistant_message_content(signature, assistant_values) - if assistant_content: - messages.append({"role": "assistant", "content": assistant_content}) - if tool_call_results is not None: - result_input = {"tool_call_results": tool_call_results} - content = self.format_user_message_content(_TOOL_CALL_RESULTS_SIGNATURE, result_input) - messages.append({"role": "user", "content": content}) - # Remove the history field from the inputs - del inputs[history_field_name] +def _preserve_unknown_content_blocks(message): + if not isinstance(message, dict) or not isinstance(content := message.get("content"), list): + return message - return messages + return { + **message, + "content": [ + block + if isinstance(block, dict) and block.get("type") in _SUPPORTED_CONTENT_BLOCKS + else _unknown_content_block(block) + for block in content + ], + } - def parse(self, signature: type[Signature], completion: str) -> dict[str, Any]: - """Parse the LM output into a dictionary of the output fields. - This method parses the LM output into a dictionary of the output fields. +def _provider_tool_calls(tool_calls) -> ToolCalls: + return ToolCalls.model_validate([_provider_tool_call_data(call) for call in tool_calls]) - Args: - signature: The DSPy signature for which to parse the LM output. - completion: The LM output to be parsed. - Returns: - A dictionary of the output fields. - """ - raise NotImplementedError +def _call_lm(lm, request: LMRequest) -> LMResponse: + data = _legacy_call_kwargs(request) + outputs = lm(messages=data.pop("messages"), **data) + return lm_response_from_legacy_outputs(outputs, request) -def _provider_value(value: Any, key: str, default: Any = None) -> Any: - if isinstance(value, dict): - return value.get(key, default) - return getattr(value, key, default) +async def _acall_lm(lm, request: LMRequest) -> LMResponse: + data = _legacy_call_kwargs(request) + outputs = await lm.acall(messages=data.pop("messages"), **data) + return lm_response_from_legacy_outputs(outputs, request) -def _provider_tool_call_to_tool_call_dict(tool_call: Any) -> dict[str, Any]: - function = _provider_value(tool_call, "function", {}) or {} - arguments = _provider_value(function, "arguments", {}) - if isinstance(arguments, str): - parsed_arguments = json_repair.loads(arguments) - elif isinstance(arguments, dict): - parsed_arguments = arguments - else: - parsed_arguments = {} - - return { - "id": _provider_value(tool_call, "id") or _provider_value(tool_call, "call_id"), - "name": _provider_value(function, "name") or _provider_value(tool_call, "name"), - "args": parsed_arguments, - } +def _legacy_call_kwargs(request: LMRequest) -> dict[str, Any]: + data = to_openai_chat_request(request) + data.pop("model", None) + if request.config.cache is not None: + if request.config.cache.enabled is not None: + data["cache"] = request.config.cache.enabled + if request.config.cache.rollout_id is not None: + data["rollout_id"] = request.config.cache.rollout_id + return data def _tool_calls_from_message(message: dict[str, Any]) -> tuple[str | None, ToolCalls | None]: - for name, value in message.items(): - if isinstance(value, ToolCalls) or (isinstance(value, dict) and "tool_calls" in value): - return name, ToolCalls.model_validate(value) - return None, None + return next( + ( + (name, ToolCalls.model_validate(value)) + for name, value in message.items() + if isinstance(value, ToolCalls) or (isinstance(value, dict) and "tool_calls" in value) + ), + (None, None), + ) + +def _tool_input_field_name(signature: type[Signature]) -> str | None: + return next( + (name for name, field in signature.input_fields.items() if _is_tool_input_annotation(field.annotation)), + None, + ) -def _tool_result_content(value: Any) -> str: - if isinstance(value, str): - return value - return json.dumps(serialize_for_json(value), ensure_ascii=False) +def _is_tool_input_annotation(annotation) -> bool: + return annotation == Tool or (get_origin(annotation) is list and get_args(annotation) == (Tool,)) def _tool_call_as_openai_message_tool_call(tool_call: ToolCalls.ToolCall) -> dict[str, Any]: @@ -771,3 +418,24 @@ def _tool_call_as_openai_message_tool_call(tool_call: ToolCalls.ToolCall) -> dic "arguments": json.dumps(serialize_for_json(tool_call.args), ensure_ascii=False), }, } + + +def _tool_result_as_openai_message(result: ToolCallResults.ToolCallResult) -> dict[str, Any]: + if isinstance(result.value, str): + content = result.value + else: + content = json.dumps(serialize_for_json(result.value), ensure_ascii=False) + + return {"role": "tool", "tool_call_id": result.call_id, "name": result.name, "content": content} + + +def _unknown_content_block(block) -> dict[str, Any]: + if isinstance(block, dict): + return {"type": "text", "text": "", "metadata": {"legacy_content_block": block}} + return {"type": "text", "text": json.dumps(block, ensure_ascii=False), "metadata": {}} + + +def _provider_tool_call_data(call): + if hasattr(call, "model_dump"): + return call.provider_data or call.model_dump() + return call