@@ -65,7 +65,7 @@ The default. Builds a chat-style prompt with field markers, parses the response
 Outputs structured JSON. Internally extends ChatAdapter — formatting is similar, but the output instruction asks for JSON and parsing uses `json_repair`. The constructor’s `use_native_function_calling=True` default flips when tool calling is wired in.
 
 **`dspy.XMLAdapter(callbacks=None)`**  
-`<field_name>value</field_name>` tags. The parser is a regex (`r"<(\w+)>(.*?)</\1>"` with `DOTALL`); it’s robust to whitespace but doesn’t tolerate nested tags of the same name.
+`<field_name>value</field_name>` tags. Lists use repeated `<item>` tags.
 
 **`dspy.TwoStepAdapter(extraction_model: BaseLM, **kwargs)`**  
 Two LM calls per inference. Use it when the main LM is a reasoning model that’s bad at formatting — the extractor is usually a cheap general-purpose LM with ChatAdapter. Doesn’t support finetuning yet.
 
@@ -1,42 +1,60 @@
-import re
-from typing import Any
+import types
+import xml.etree.ElementTree as ET
+from collections import defaultdict
+from typing import Any, Union, get_args, get_origin
+from xml.sax.saxutils import quoteattr
 
-from pydantic.fields import FieldInfo
+import pydantic
+from pydantic import TypeAdapter
+from typing_extensions import is_typeddict
 
 from dspy.adapters.chat_adapter import ChatAdapter, FieldInfoWithName
-from dspy.adapters.utils import apply_output_field_defaults, format_field_value, translate_field_type
+from dspy.adapters.utils import (
+    apply_output_field_defaults,
+    format_field_value,
+    parse_value,
+    serialize_for_json,
+    translate_field_type,
+)
 from dspy.signatures.signature import Signature
+from dspy.utils.exceptions import AdapterParseError
 
 
 class XMLAdapter(ChatAdapter):
-    field_pattern = re.compile(r"<(?P<name>\w+)>((?P<content>.*?))</\1>", re.DOTALL)
-
     def format_field_with_value(self, fields_with_values: dict[FieldInfoWithName, Any]) -> str:
         output = []
-        for field, field_value in fields_with_values.items():
-            formatted = format_field_value(field_info=field.info, value=field_value)
+        for field, value in fields_with_values.items():
+            serialized = serialize_for_json(value)
+            is_output = (field.info.json_schema_extra or {}).get("__dspy_field_type") == "output"
+            if is_output and self._uses_nested_xml(field.info.annotation) and isinstance(serialized, (dict, list)):
+                output.append(self._value_to_xml(serialized, field.name))
+                continue
+            formatted = format_field_value(field_info=field.info, value=value)
+            if is_output and field.info.annotation is str:
+                formatted = formatted.replace("&", "&amp;").replace("<", "&lt;")
             output.append(f"<{field.name}>\n{formatted}\n</{field.name}>")
         return "\n\n".join(output).strip()
 
     def format_field_structure(self, signature: type[Signature]) -> str:
-        """
-        XMLAdapter requires input and output fields to be wrapped in XML tags like `<field_name>`.
-        """
-
-        parts = []
-        parts.append("All interactions will be structured in the following way, with the appropriate values filled in.")
-
-        def format_signature_fields_for_instructions(fields: dict[str, FieldInfo]):
+        def format_field(name, field):
+            if (field.json_schema_extra or {}).get("__dspy_field_type") == "output" and self._uses_nested_xml(
+                field.annotation
+            ):
+                return self._xml_schema(name, field.annotation)
             return self.format_field_with_value(
-                fields_with_values={
-                    FieldInfoWithName(name=field_name, info=field_info): translate_field_type(field_name, field_info)
-                    for field_name, field_info in fields.items()
-                },
+                {FieldInfoWithName(name=name, info=field): translate_field_type(name, field)}
             )
 
-        parts.append(format_signature_fields_for_instructions(signature.input_fields))
-        parts.append(format_signature_fields_for_instructions(signature.output_fields))
-        return "\n\n".join(parts).strip()
+        fields = (
+            "\n\n".join(format_field(name, field) for name, field in group.items())
+            for group in (signature.input_fields, signature.output_fields)
+        )
+        return "\n\n".join(
+            (
+                "All interactions will be structured in the following way, with the appropriate values filled in.",
+                *fields,
+            )
+        )
 
     def format_user_message_content(
         self,
@@ -46,75 +64,177 @@ def format_user_message_content(
         suffix: str = "",
         main_request: bool = False,
     ) -> str:
-        messages = [prefix]
-
-        messages.append(self.format_field_with_value(
-            {
-                FieldInfoWithName(name=k, info=v): inputs.get(k)
-                for k, v in signature.input_fields.items() if k in inputs
-            },
-        ))
-
+        fields = {
+            FieldInfoWithName(name=k, info=v): inputs[k] for k, v in signature.input_fields.items() if k in inputs
+        }
+        messages = [prefix, self.format_field_with_value(fields)]
         if main_request:
-            output_requirements = self.user_message_output_requirements(signature)
-            if output_requirements is not None:
-                messages.append(output_requirements)
-
-        messages.append(suffix)
-        return "\n\n".join(messages).strip()
+            messages.append(self.user_message_output_requirements(signature))
+        return "\n\n".join((*messages, suffix)).strip()
 
     def format_assistant_message_content(
-        self,
-        signature: type[Signature],
-        outputs: dict[str, Any],
-        missing_field_message=None,
+        self, signature: type[Signature], outputs: dict[str, Any], missing_field_message=None
     ) -> str:
-        return self.format_field_with_value(
-            {
-                FieldInfoWithName(name=k, info=v): outputs.get(k, missing_field_message)
-                for k, v in signature.output_fields.items()
-            },
-        )
+        fields = {
+            FieldInfoWithName(name=k, info=v): outputs.get(k, missing_field_message)
+            for k, v in signature.output_fields.items()
+        }
+        return self.format_field_with_value(fields)
 
     def user_message_output_requirements(self, signature: type[Signature]) -> str:
-        message = "Respond with the corresponding output fields wrapped in XML tags "
-        message += ", then ".join(f"`<{f}>`" for f in signature.output_fields)
-        message += "."
-        return message
+        fields = ", then ".join(f"`<{name}>`" for name in signature.output_fields)
+        schemas = [
+            self._xml_schema(name, field.annotation)
+            for name, field in signature.output_fields.items()
+            if self._uses_nested_xml(field.annotation)
+        ]
+        return f"Respond with the corresponding output fields wrapped in XML tags {fields}." + (
+            f" Use this nested XML structure: {' '.join(schemas)}" if schemas else ""
+        )
 
     def parse(self, signature: type[Signature], completion: str) -> dict[str, Any]:
-        fields = {}
-        for match in self.field_pattern.finditer(completion):
-            name = match.group("name")
-            content = match.group("content").strip()
-            if name in signature.output_fields and name not in fields:
-                fields[name] = content
-        # Cast values using base class parse_value helper
-        for k, v in fields.items():
-            fields[k] = self._parse_field_value(signature.output_fields[k], v, completion, signature)
-        fields = apply_output_field_defaults(signature, fields)
-        if fields.keys() != signature.output_fields.keys():
-            from dspy.utils.exceptions import AdapterParseError
-
+        try:
+            root = ET.fromstring(f"<dspy_root>{completion}</dspy_root>")
+        except ET.ParseError as e:
             raise AdapterParseError(
                 adapter_name="XMLAdapter",
                 signature=signature,
                 lm_response=completion,
-                parsed_result=fields,
+                message=f"Failed to parse XML: {e}",
+            ) from e
+        elements = self._group_children(root)
+        fields = {}
+        for name, field in signature.output_fields.items():
+            if name not in elements:
+                continue
+            adapter = TypeAdapter(field.annotation)
+            schema = adapter.json_schema(by_alias=False)
+            value = None
+            for candidate in [schema, *schema.get("anyOf", [])]:
+                try:
+                    value = self._elements_to_value(elements[name], candidate, schema.get("$defs", {}))
+                    try:
+                        fields[name] = parse_value(value, field.annotation)
+                    except pydantic.ValidationError:
+                        fields[name] = adapter.validate_python(value, by_name=True)
+                    break
+                except Exception as e:
+                    error = e
+            else:
+                raise AdapterParseError(
+                    adapter_name="XMLAdapter",
+                    signature=signature,
+                    lm_response=completion,
+                    message=f"Failed to parse field {field} with value {value}: {error}",
+                ) from error
+        fields = apply_output_field_defaults(signature, fields)
+        if fields.keys() != signature.output_fields.keys():
+            raise AdapterParseError(
+                adapter_name="XMLAdapter", signature=signature, lm_response=completion, parsed_result=fields
             )
         return fields
 
-    def _parse_field_value(self, field_info, raw, completion, signature):
-        from dspy.adapters.utils import parse_value
-
-        try:
-            return parse_value(raw, field_info.annotation)
-        except Exception as e:
-            from dspy.utils.exceptions import AdapterParseError
+    @classmethod
+    def _value_to_xml(cls, value: Any, tag: str, key: str | None = None) -> str:
+        attrs = f" key={quoteattr(key)}" if key is not None else ""
+        if isinstance(value, list):
+            children = "".join(cls._value_to_xml(item, "item") for item in value)
+            return f"<{tag}{attrs}>{children}</{tag}>" if children else f"<{tag}{attrs} />"
+        if isinstance(value, dict):
+            children = []
+            for name, child in value.items():
+                name = str(name)
+                valid_name = (name[:1] + name[1:].replace("-", "_").replace(".", "_")).isidentifier()
+                children.append(
+                    cls._value_to_xml(child, name) if valid_name else cls._value_to_xml(child, "entry", name)
+                )
+            return f"<{tag}{attrs}>{''.join(children)}</{tag}>"
+        return (
+            f"<{tag}{attrs}>{str(value).replace('&', '&amp;').replace('<', '&lt;')}</{tag}>"
+            if value is not None
+            else f"<{tag}{attrs} />"
+        )
 
-            raise AdapterParseError(
-                adapter_name="XMLAdapter",
-                signature=signature,
-                lm_response=completion,
-                message=f"Failed to parse field {field_info} with value {raw}: {e}",
+    @classmethod
+    def _xml_schema(cls, tag: str, annotation: Any) -> str:
+        schema = TypeAdapter(annotation).json_schema(by_alias=False)
+        return cls._schema_to_xml(tag, schema, schema.get("$defs", {}), frozenset())
+
+    @classmethod
+    def _schema_to_xml(cls, tag: str, schema: dict, definitions: dict, seen: frozenset[str]) -> str:
+        if ref := schema.get("$ref"):
+            name = ref.rsplit("/", 1)[-1]
+            return (
+                f"<{tag}>...</{tag}>"
+                if name in seen
+                else cls._schema_to_xml(tag, definitions[name], definitions, seen | {name})
             )
+        if choices := schema.get("anyOf"):
+            return cls._schema_to_xml(
+                tag, next((s for s in choices if s.get("type") != "null"), choices[0]), definitions, seen
+            )
+        if schema.get("type") == "array":
+            item = cls._schema_to_xml("item", schema.get("items", {}), definitions, seen)
+            return f"<{tag}>{item}</{tag}>"
+        children = "".join(
+            cls._schema_to_xml(name, child, definitions, seen) for name, child in schema.get("properties", {}).items()
+        )
+        return f"<{tag}>{children}</{tag}>" if children else f"<{tag}>...</{tag}>"
+
+    @classmethod
+    def _elements_to_value(cls, elements: list[ET.Element], schema: dict, definitions: dict) -> Any:
+        schema = definitions.get(schema.get("$ref", "").rsplit("/", 1)[-1], schema)
+        if choices := schema.get("anyOf"):
+            if {"type": "null"} in choices and not list(elements[0]) and not (elements[0].text or "").strip():
+                return None
+            choices = [choice for choice in choices if choice.get("type") != "null"]
+            schema = choices[0]
+            if list(elements[0]):
+                schema = next((choice for choice in choices if choice.get("type") != "string"), schema)
+        if schema.get("type") == "array":
+            if len(elements) == 1 and not list(elements[0]):
+                text = (elements[0].text or "").strip()
+                if not text or text.startswith("["):
+                    return [] if not text else text
+            if len(elements) == 1 and (items := cls._group_children(elements[0]).get("item")):
+                elements = items
+            return [cls._elements_to_value([element], schema.get("items", {}), definitions) for element in elements]
+        element = elements[0]
+        if schema.get("type") == "string" and list(element):
+            return (element.text or "") + "".join(ET.tostring(child, encoding="unicode") for child in element)
+        children = cls._group_children(element)
+        if not children:
+            if schema.get("type") == "object" and not (element.text or "").strip():
+                return {}
+            values = [(element.text or "").strip() for element in elements]
+            return values[0] if len(values) == 1 else values
+        properties = schema.get("properties", {})
+        child_schema = schema.get("additionalProperties", {})
+        if not isinstance(child_schema, dict):
+            child_schema = {}
+        return {
+            name: cls._elements_to_value(items, properties.get(name, child_schema), definitions)
+            for name, items in children.items()
+        }
+
+    @staticmethod
+    def _group_children(element: ET.Element) -> dict[str, list[ET.Element]]:
+        children = defaultdict(list)
+        for child in element:
+            children[child.attrib.get("key", child.tag) if child.tag == "entry" else child.tag].append(child)
+        return children
+
+    @staticmethod
+    def _uses_nested_xml(annotation: Any) -> bool:
+        args = [arg for arg in get_args(annotation) if arg is not type(None)]
+        annotation = args[0] if len(args) == 1 and get_origin(annotation) in (Union, types.UnionType) else annotation
+        origin = get_origin(annotation)
+        item = get_args(annotation)[0] if origin is list and get_args(annotation) else annotation
+        is_dspy_model = (
+            isinstance(item, type) and issubclass(item, pydantic.BaseModel) and item.__module__.startswith("dspy.")
+        )
+        return not is_dspy_model and (
+            origin in (list, dict)
+            or is_typeddict(annotation)
+            or (isinstance(annotation, type) and issubclass(annotation, pydantic.BaseModel))
+        )