fix(anthropic): preserve const in structured output schemas

libs/partners/anthropic/langchain_anthropic/chat_models.py

@@ -2392,6 +2392,68 @@ def _lc_tool_calls_to_anthropic_tool_use_blocks(
     ]
 
 
+def _json_schema_values_equal(left: Any, right: Any) -> bool:
+    """Compare JSON values using JSON Schema equality rules."""
+    if isinstance(left, bool) or isinstance(right, bool):
+        return isinstance(left, bool) and isinstance(right, bool) and left == right
+    if isinstance(left, (int, float)) and isinstance(right, (int, float)):
+        return left == right
+    if isinstance(left, dict) and isinstance(right, dict):
+        return left.keys() == right.keys() and all(
+            _json_schema_values_equal(left[key], right[key]) for key in left
+        )
+    if isinstance(left, list) and isinstance(right, list):
+        return len(left) == len(right) and all(
+            _json_schema_values_equal(left_item, right_item)
+            for left_item, right_item in zip(left, right, strict=True)
+        )
+    return type(left) is type(right) and left == right
+
+
+def _replace_consts_with_singleton_enums(
+    schema: dict[str, Any],
+) -> dict[str, Any]:
+    """Copy `schema` and replace `const` before Anthropic moves it to descriptions."""
+    rewritten_schema = copy.deepcopy(schema)
+
+    def rewrite(schema_node: dict[str, Any]) -> None:
+        if "const" in schema_node:
+            const_value = schema_node.pop("const")
+            if "enum" in schema_node:
+                enum_values = schema_node["enum"]
+                if not isinstance(enum_values, list) or not any(
+                    _json_schema_values_equal(const_value, enum_value)
+                    for enum_value in enum_values
+                ):
+                    msg = (
+                        "JSON Schema `const` must match one of the values in `enum` "
+                        "when both constraints are present."
+                    )
+                    raise ValueError(msg)
+            schema_node["enum"] = [const_value]
+
+        for keyword in ("$defs", "properties"):
+            nested_schemas = schema_node.get(keyword)
+            if isinstance(nested_schemas, dict):
+                for nested_schema in nested_schemas.values():
+                    if isinstance(nested_schema, dict):
+                        rewrite(nested_schema)
+
+        for keyword in ("allOf", "anyOf", "oneOf"):
+            nested_schemas = schema_node.get(keyword)
+            if isinstance(nested_schemas, list):
+                for nested_schema in nested_schemas:
+                    if isinstance(nested_schema, dict):
+                        rewrite(nested_schema)
+
+        items_schema = schema_node.get("items")
+        if isinstance(items_schema, dict):
+            rewrite(items_schema)
+
+    rewrite(rewritten_schema)
+    return rewritten_schema
+
+
 def _convert_to_anthropic_output_config_format(schema: dict | type) -> dict[str, Any]:
     """Convert JSON schema, Pydantic model, or `TypedDict` into `output_config.format`.
 
@@ -2406,11 +2468,14 @@ def _convert_to_anthropic_output_config_format(schema: dict | type) -> dict[str,
     from anthropic import transform_schema
 
     is_pydantic_class = isinstance(schema, type) and is_basemodel_subclass(schema)
-    if is_pydantic_class or isinstance(schema, dict):
-        json_schema = transform_schema(schema)
+    if is_pydantic_class:
+        json_schema = cast("type[BaseModel]", schema).model_json_schema()
+    elif isinstance(schema, dict):
+        json_schema = schema
     else:
         # TypedDict
-        json_schema = transform_schema(convert_to_json_schema(schema))
+        json_schema = convert_to_json_schema(schema)
+    json_schema = transform_schema(_replace_consts_with_singleton_enums(json_schema))
     return {"type": "json_schema", "schema": json_schema}
 
 

libs/partners/anthropic/tests/unit_tests/test_chat_models.py

@@ -6,13 +6,14 @@ import copy
 import os
 import warnings
 from collections.abc import Callable
-from typing import Any, Literal, cast
+from typing import Annotated, Any, Literal, cast
 from unittest.mock import MagicMock, patch
 
 import anthropic
 import pytest
 from anthropic.types import Message, TextBlock, Usage
 from blockbuster import blockbuster_ctx
+from langchain.agents.structured_output import ProviderStrategy
 from langchain_core.exceptions import ContextOverflowError
 from langchain_core.messages import (
     AIMessage,
@@ -2438,10 +2439,10 @@ def test_response_format_with_output_config() -> None:
         "Test query",
         response_format=Person.model_json_schema(),
     )
-    assert "output_config" in payload
-    assert "format" in payload["output_config"]
-    assert payload["output_config"]["format"]["type"] == "json_schema"
-    assert "schema" in payload["output_config"]["format"]
+    assert payload["output_config"]["format"] == {
+        "type": "json_schema",
+        "schema": anthropic.transform_schema(Person.model_json_schema()),
+    }
 
     # No response_format - output_config should not have format
     model = ChatAnthropic(model=MODEL_NAME)
@@ -2450,6 +2451,127 @@ def test_response_format_with_output_config() -> None:
         assert "format" not in payload["output_config"]
 
 
+def test_with_structured_output_preserves_discriminated_union_literals() -> None:
+    class CriterionPass(BaseModel):
+        name: str
+        passed: Literal[True]
+
+    class CriterionFail(BaseModel):
+        name: str
+        passed: Literal[False]
+        gap: str
+
+    class GraderResponse(BaseModel):
+        criteria: list[
+            Annotated[
+                CriterionPass | CriterionFail,
+                Field(discriminator="passed"),
+            ]
+        ]
+
+    model = ChatAnthropic(model=MODEL_NAME)
+    structured_model = model.with_structured_output(
+        GraderResponse,
+        method="json_schema",
+    )
+    bound_model = cast("RunnableBinding", cast("Any", structured_model).first)
+    output_schema = bound_model.kwargs["output_config"]["format"]["schema"]
+
+    criterion_schema = output_schema["properties"]["criteria"]["items"]
+    assert "oneOf" not in criterion_schema
+    assert criterion_schema["anyOf"] == [
+        {"$ref": "#/$defs/CriterionPass"},
+        {"$ref": "#/$defs/CriterionFail"},
+    ]
+
+    pass_schema = output_schema["$defs"]["CriterionPass"]
+    fail_schema = output_schema["$defs"]["CriterionFail"]
+    assert pass_schema["properties"]["passed"] == {
+        "type": "boolean",
+        "enum": [True],
+        "title": "Passed",
+    }
+    assert fail_schema["properties"]["passed"] == {
+        "type": "boolean",
+        "enum": [False],
+        "title": "Passed",
+    }
+    assert fail_schema["required"] == ["name", "passed", "gap"]
+
+
+def test_response_format_preserves_nested_consts_without_mutating_schema() -> None:
+    raw_schema: dict[str, Any] = {
+        "$defs": {
+            "Options": {
+                "type": "object",
+                "properties": {
+                    "enabled": {"type": "boolean", "const": True},
+                },
+                "required": ["enabled"],
+            },
+        },
+        "type": "object",
+        "properties": {
+            "const": {"type": "string"},
+            "status": {
+                "type": "string",
+                "const": "ready",
+                "enum": ["ready", "waiting"],
+            },
+            "values": {
+                "type": "array",
+                "items": {
+                    "oneOf": [
+                        {"type": "string", "const": "alpha"},
+                        {"type": "integer", "const": 1},
+                    ],
+                },
+            },
+        },
+        "required": ["const", "status", "values"],
+    }
+    original_schema = copy.deepcopy(raw_schema)
+    model = ChatAnthropic(model=MODEL_NAME)
+
+    payload = model._get_request_payload(
+        "Test query",
+        **ProviderStrategy(raw_schema).to_model_kwargs(),
+    )
+
+    assert raw_schema == original_schema
+    output_schema = payload["output_config"]["format"]["schema"]
+    assert output_schema["$defs"]["Options"]["properties"]["enabled"] == {
+        "type": "boolean",
+        "enum": [True],
+    }
+    assert output_schema["properties"]["const"] == {"type": "string"}
+    assert output_schema["properties"]["status"] == {
+        "type": "string",
+        "enum": ["ready"],
+    }
+    item_schema = output_schema["properties"]["values"]["items"]
+    assert "oneOf" not in item_schema
+    assert item_schema["anyOf"] == [
+        {"type": "string", "enum": ["alpha"]},
+        {"type": "integer", "enum": [1]},
+    ]
+
+
+def test_response_format_rejects_conflicting_const_and_enum() -> None:
+    model = ChatAnthropic(model=MODEL_NAME)
+    schema = {
+        "type": "string",
+        "const": "ready",
+        "enum": ["waiting"],
+    }
+
+    with pytest.raises(ValueError, match="must match one of the values"):
+        model._get_request_payload(
+            "Test query",
+            **ProviderStrategy(schema).to_model_kwargs(),
+        )
+
+
 def test_strict_tool_use_payload() -> None:
     """Test that strict tool use property is correctly passed through to payload."""