ReActV2: Production-ready agent module with native FC, semantic history, and forced submit

dspy/adapters/base.py

@@ -1,9 +1,7 @@
 import logging
 from typing import Any, get_origin
 
-import json_repair
-
-from dspy.adapters.types import History, Type
+from dspy.adapters.types import ActionEvent, History, InputEvent, LegacyEvent, OutputEvent, Type
 from dspy.adapters.types.base_type import split_message_content_for_custom_types
 from dspy.adapters.types.reasoning import Reasoning
 from dspy.adapters.types.tool import Tool, ToolCalls
@@ -93,6 +91,7 @@ def _call_preprocess(
                 signature_for_native_function_calling = signature_for_native_function_calling.delete(
                     tool_call_input_field_name
                 )
+                signature_for_native_function_calling.__dspy_native_fc__ = True
 
                 return signature_for_native_function_calling
 
@@ -148,14 +147,7 @@ def _call_postprocess(
                 )
 
             if tool_calls and tool_call_output_field_name:
-                tool_calls = [
-                    {
-                        "name": v["function"]["name"],
-                        "args": json_repair.loads(v["function"]["arguments"]),
-                    }
-                    for v in tool_calls
-                ]
-                value[tool_call_output_field_name] = ToolCalls.from_dict_list(tool_calls)
+                value[tool_call_output_field_name] = ToolCalls.model_validate(tool_calls)
 
             # Parse custom types that does not rely on the `Adapter.parse()` method
             for name, field in original_signature.output_fields.items():
@@ -269,24 +261,39 @@ def format(
         # 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)
+        has_open_episode = False
         if history_field_name:
+            # Check if history has an open episode BEFORE format_conversation_history deletes it from inputs_copy.
+            history_obj = inputs_copy.get(history_field_name)
+            has_open_episode = hasattr(history_obj, "has_open_episode") and history_obj.has_open_episode()
+
             # 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)
+            if getattr(signature, "__dspy_native_fc__", False):
+                signature_without_history.__dspy_native_fc__ = True
             conversation_history = self.format_conversation_history(
                 signature_without_history,
                 history_field_name,
                 inputs_copy,
             )
+            inputs_copy.pop(history_field_name, None)
 
         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)
-            messages.append({"role": "user", "content": content})
+            if has_open_episode and hasattr(self, "user_message_output_requirements"):
+                # The InputEvent in the conversation history already contains the current query inputs.
+                # Only append the output requirements suffix to avoid sending the query twice.
+                output_req = self.user_message_output_requirements(signature_without_history)
+                if output_req:
+                    messages.append({"role": "user", "content": output_req})
+            else:
+                # No open episode — include the full input + output requirements as before.
+                content = self.format_user_message_content(signature_without_history, inputs_copy, main_request=True)
+                messages.append({"role": "user", "content": content})
         else:
             # Only current input
             content = self.format_user_message_content(signature, inputs_copy, main_request=True)
@@ -501,22 +508,70 @@ def format_conversation_history(
             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]
+        for event_idx, message in enumerate(conversation_history):
+            if isinstance(message, InputEvent):
+                content = self.format_user_message_content(signature, message.inputs)
+                if content.strip():
+                    messages.append({"role": "user", "content": content})
+            elif isinstance(message, ActionEvent):
+                is_native_fc = getattr(signature, "__dspy_native_fc__", False)
+                tc_obj = message.tool_calls
+                obs = message.observations
+
+                if is_native_fc and tc_obj and hasattr(tc_obj, "tool_calls"):
+                    import json as _json
+                    tc_list = []
+                    # Pre-generate stable IDs for tool calls that lack them so the
+                    # assistant message and the corresponding tool-response messages
+                    # reference the same ID (required by the OpenAI API).
+                    resolved_ids = []
+                    for call_idx, tc in enumerate(tc_obj.tool_calls):
+                        resolved_ids.append(tc.id or f"call_{event_idx}_{call_idx}")
+                    for tc, tid in zip(tc_obj.tool_calls, resolved_ids, strict=True):
+                        fmt = tc.format()
+                        # Ensure the id is always present
+                        fmt["id"] = tid
+                        if isinstance(fmt.get("function", {}).get("arguments"), dict):
+                            fmt["function"]["arguments"] = _json.dumps(fmt["function"]["arguments"])
+                        tc_list.append(fmt)
+                    asst_msg = {"role": "assistant", "tool_calls": tc_list}
+                    thought = message.thought
+                    if thought:
+                        asst_msg["content"] = str(thought)
+                    messages.append(asst_msg)
+                    for tid, obs_item in zip(resolved_ids, obs, strict=False):
+                        content = str(obs_item.value) if not isinstance(obs_item.value, list) else "\n".join(str(r) for r in obs_item.value)
+                        messages.append({"role": "tool", "content": content, "tool_call_id": tid})
+                else:
+                    asst_content = self.format_assistant_message_content(
+                        signature,
+                        {"next_thought": message.thought, "tool_calls": tc_obj},
+                        missing_field_message="Not supplied for this conversation history message. ",
+                    )
+                    messages.append({"role": "assistant", "content": asst_content})
+                    if obs:
+                        obs_parts = []
+                        for obs_item in obs:
+                            label = "Error" if obs_item.is_error else "Observation"
+                            if isinstance(obs_item.value, list):
+                                obs_parts.append(f"{label}:\n" + "\n".join(str(item) for item in obs_item.value))
+                            else:
+                                obs_parts.append(f"{label}: {obs_item.value}")
+                        messages.append({"role": "user", "content": "\n\n".join(obs_parts)})
+            elif isinstance(message, OutputEvent):
+                pass
+            elif isinstance(message, LegacyEvent):
+                # Backward compat fallback for plain dicts wrapped as LegacyEvent
+                messages.append({"role": "user", "content": self.format_user_message_content(signature, message.data)})
+                messages.append(
+                    {
+                        "role": "assistant",
+                        "content": self.format_assistant_message_content(
+                            signature, message.data,
+                            missing_field_message="Not supplied for this conversation history message. ",
+                        ),
+                    }
+                )
 
         return messages
 

dspy/adapters/chat_adapter.py

@@ -78,9 +78,10 @@ def __call__(
                 isinstance(e, ContextWindowExceededError)
                 or isinstance(self, JSONAdapter)
                 or not self.use_json_adapter_fallback
+                or "tools" in lm_kwargs  # Don't fall back to JSON mode when native FC is active
             ):
-                # On context window exceeded error, already using JSONAdapter, or use_json_adapter_fallback is False
-                # we don't want to retry with a different adapter. Raise the original error instead of the fallback error.
+                # On context window exceeded error, already using JSONAdapter, use_json_adapter_fallback is False,
+                # or native function calling is active — don't retry with a different adapter.
                 raise e
             return JSONAdapter()(lm, lm_kwargs, signature, demos, inputs)
 
@@ -102,9 +103,10 @@ async def acall(
                 isinstance(e, ContextWindowExceededError)
                 or isinstance(self, JSONAdapter)
                 or not self.use_json_adapter_fallback
+                or "tools" in lm_kwargs  # Don't fall back to JSON mode when native FC is active
             ):
-                # On context window exceeded error, already using JSONAdapter, or use_json_adapter_fallback is False
-                # we don't want to retry with a different adapter. Raise the original error instead of the fallback error.
+                # On context window exceeded error, already using JSONAdapter, use_json_adapter_fallback is False,
+                # or native function calling is active — don't retry with a different adapter.
                 raise e
             return await JSONAdapter().acall(lm, lm_kwargs, signature, demos, inputs)
 
@@ -120,6 +122,9 @@ def format_field_structure(self, signature: type[Signature]) -> str:
         `[[ ## field_name ## ]]`. An arbitrary field `completed` ([[ ## completed ## ]]) is added to the end of the
         output fields section to indicate the end of the output fields.
         """
+        if getattr(signature, "__dspy_native_fc__", False):
+            return self._format_native_fc_structure(signature)
+
         parts = []
         parts.append("All interactions will be structured in the following way, with the appropriate values filled in.")
 
@@ -136,6 +141,14 @@ def format_signature_fields_for_instructions(fields: dict[str, FieldInfo]):
         parts.append("[[ ## completed ## ]]\n")
         return "\n\n".join(parts).strip()
 
+    def _format_native_fc_structure(self, signature: type[Signature]) -> str:
+        parts = ["You will receive inputs and must respond with your reasoning in plain text, then call the appropriate tool."]
+        for name, field in signature.output_fields.items():
+            desc = get_field_description_string({name: field})
+            parts.append(f"Your response text should contain: {desc.strip()}")
+        parts.append("Do NOT use any special markers or delimiters. Think step-by-step, then call the appropriate tool via the API.")
+        return "\n".join(parts)
+
     def format_task_description(self, signature: type[Signature]) -> str:
         instructions = textwrap.dedent(signature.instructions)
         objective = ("\n" + " " * 8).join([""] + instructions.splitlines())
@@ -164,23 +177,9 @@ def format_user_message_content(
         messages.append(suffix)
         return "\n\n".join(messages).strip()
 
-    def user_message_output_requirements(self, signature: type[Signature]) -> str:
-        """Returns a simplified format reminder for the language model.
-
-        In chat-based interactions, language models may lose track of the required output format
-        as the conversation context grows longer. This method generates a concise reminder of
-        the expected output structure that can be included in user messages.
-
-        Args:
-            signature (Type[Signature]): The DSPy signature defining the expected input/output fields.
-
-        Returns:
-            str: A simplified description of the required output format.
-
-        Note:
-            This is a more lightweight version of `format_field_structure` specifically designed
-            for inline reminders within chat messages.
-        """
+    def user_message_output_requirements(self, signature: type[Signature]) -> str | None:
+        if getattr(signature, "__dspy_native_fc__", False):
+            return "Think step-by-step about what to do next, then call the appropriate tool."
 
         def type_info(v):
             if v.annotation is not str:
@@ -209,6 +208,9 @@ def format_assistant_message_content(
         return assistant_message_content
 
     def parse(self, signature: type[Signature], completion: str) -> dict[str, Any]:
+        if getattr(signature, "__dspy_native_fc__", False):
+            return self._parse_native_fc(signature, completion)
+
         sections = [(None, [])]
 
         for line in completion.splitlines():
@@ -245,11 +247,23 @@ def parse(self, signature: type[Signature], completion: str) -> dict[str, Any]:
 
         return fields
 
+    def _parse_native_fc(self, signature: type[Signature], completion: str) -> dict[str, Any]:
+        """Parse native FC response: assign free-form text to the single str output field."""
+        str_fields = [k for k, v in signature.output_fields.items() if v.annotation is str]
+        if len(str_fields) == 1:
+            return {str_fields[0]: completion.strip()}
+        raise AdapterParseError(
+            adapter_name="ChatAdapter",
+            signature=signature,
+            lm_response=completion,
+            message="Native FC response with multiple output fields cannot be parsed without markers.",
+        )
+
     def format_field_with_value(self, fields_with_values: dict[FieldInfoWithName, Any]) -> str:
         """
         Formats the values of the specified fields according to the field's DSPy type (input or output),
         annotation (e.g. str, int, etc.), and the type of the value itself. Joins the formatted values
-        into a single string, which is is a multiline string if there are multiple fields.
+        into a single string, which is a multiline string if there are multiple fields.
 
         Args:
             fields_with_values: A dictionary mapping information about a field to its corresponding

dspy/adapters/types/__init__.py

@@ -2,9 +2,20 @@
 from dspy.adapters.types.base_type import Type
 from dspy.adapters.types.code import Code
 from dspy.adapters.types.file import File
-from dspy.adapters.types.history import History
+from dspy.adapters.types.history import (
+    ActionEvent,
+    History,
+    HistoryEvent,
+    InputEvent,
+    LegacyEvent,
+    Observation,
+    OutputEvent,
+)
 from dspy.adapters.types.image import Image
 from dspy.adapters.types.reasoning import Reasoning
 from dspy.adapters.types.tool import Tool, ToolCalls
 
-__all__ = ["History", "Image", "Audio", "File", "Type", "Tool", "ToolCalls", "Code", "Reasoning"]
+__all__ = [
+    "ActionEvent", "History", "HistoryEvent", "InputEvent", "LegacyEvent", "Observation", "OutputEvent",
+    "Image", "Audio", "File", "Type", "Tool", "ToolCalls", "Code", "Reasoning",
+]