fix(anthropic): route strict structured output through forced tool_use instead of prompt-injection (#1002) (#2339)

Fixes #1002

Author: eldar702

hindsight-api-slim/hindsight_api/engine/providers/anthropic_llm.py

@@ -136,7 +136,9 @@ async def call(
             initial_backoff: Initial backoff time in seconds.
             max_backoff: Maximum backoff time in seconds.
             skip_validation: Return raw JSON without Pydantic validation.
-            strict_schema: Use strict JSON schema enforcement (not supported by Anthropic).
+            strict_schema: Route structured output through a forced tool_use tool for
+                native constrained decoding (issue #1002). When False, falls back to
+                schema-in-prompt + JSON parse.
             return_usage: If True, return tuple (result, TokenUsage) instead of just result.
 
         Returns:
@@ -167,14 +169,21 @@ async def call(
             else:
                 anthropic_messages.append({"role": role, "content": content})
 
-        # Add JSON schema instruction if response_format is provided
+        # Structured output: prefer Anthropic-native constrained decoding via a single
+        # forced tool_use tool (strict_schema) over text-injecting the schema and
+        # parsing the reply. Native constrained decoding guarantees schema-valid JSON,
+        # eliminating the invalid-JSON retry storm (issue #1002). When strict_schema is
+        # off we keep the text-inject + json.loads fallback for backward compatibility.
+        schema = None
+        use_forced_tool = False
+        _tool_name = "structured_response"
         if response_format is not None and hasattr(response_format, "model_json_schema"):
             schema = response_format.model_json_schema()
-            schema_msg = f"\n\nYou must respond with valid JSON matching this schema:\n{json.dumps(schema, indent=2, ensure_ascii=False)}"
-            if system_prompt:
-                system_prompt += schema_msg
+            if strict_schema:
+                use_forced_tool = True
             else:
-                system_prompt = schema_msg
+                schema_msg = f"\n\nYou must respond with valid JSON matching this schema:\n{json.dumps(schema, indent=2, ensure_ascii=False)}"
+                system_prompt = (system_prompt + schema_msg) if system_prompt else schema_msg
 
         # Prepare parameters
         call_params: dict[str, Any] = {
@@ -186,6 +195,14 @@ async def call(
         if system_prompt:
             call_params["system"] = system_prompt
 
+        if use_forced_tool:
+            # Single tool whose input_schema IS the response schema; force the model to
+            # emit it via tool_choice so the SDK does constrained decoding for us.
+            call_params["tools"] = [
+                {"name": _tool_name, "description": "Return the structured response.", "input_schema": schema}
+            ]
+            call_params["tool_choice"] = {"type": "tool", "name": _tool_name}
+
         if self._extra_body:
             call_params["extra_body"] = self._extra_body
 
@@ -195,32 +212,49 @@ async def call(
             try:
                 response = await self._client.messages.create(**call_params)
 
-                # Anthropic response content is a list of blocks
-                content = ""
-                for block in response.content:
-                    if block.type == "text":
-                        content += block.text
-
-                if response_format is not None:
-                    # Models may wrap JSON in markdown code blocks
-                    clean_content = content
-                    if "```json" in content:
-                        clean_content = content.split("```json")[1].split("```")[0].strip()
-                    elif "```" in content:
-                        clean_content = content.split("```")[1].split("```")[0].strip()
-
-                    try:
-                        json_data = json.loads(clean_content)
-                    except json.JSONDecodeError:
-                        # Fallback to parsing raw content if markdown stripping failed
-                        json_data = json.loads(content)
-
-                    if skip_validation:
-                        result = json_data
-                    else:
-                        result = response_format.model_validate(json_data)
+                if use_forced_tool:
+                    # Forced tool_use → the validated args are already a dict; no parsing,
+                    # no markdown-strip, no JSON-decode retry possible.
+                    tool_input = None
+                    for block in response.content:
+                        if block.type == "tool_use" and block.name == _tool_name:
+                            tool_input = block.input or {}
+                            break
+                    if tool_input is None:
+                        # Model ignored the forced tool (rare, e.g. a gateway that drops
+                        # tool_choice). Fall back to text parse so we don't hard-fail; the
+                        # existing retry loop still covers genuine errors.
+                        content = "".join(b.text for b in response.content if b.type == "text")
+                        tool_input = json.loads(content)
+                    content = json.dumps(tool_input)
+                    result = tool_input if skip_validation else response_format.model_validate(tool_input)
                 else:
-                    result = content
+                    # Anthropic response content is a list of blocks
+                    content = ""
+                    for block in response.content:
+                        if block.type == "text":
+                            content += block.text
+
+                    if response_format is not None:
+                        # Models may wrap JSON in markdown code blocks
+                        clean_content = content
+                        if "```json" in content:
+                            clean_content = content.split("```json")[1].split("```")[0].strip()
+                        elif "```" in content:
+                            clean_content = content.split("```")[1].split("```")[0].strip()
+
+                        try:
+                            json_data = json.loads(clean_content)
+                        except json.JSONDecodeError:
+                            # Fallback to parsing raw content if markdown stripping failed
+                            json_data = json.loads(content)
+
+                        if skip_validation:
+                            result = json_data
+                        else:
+                            result = response_format.model_validate(json_data)
+                    else:
+                        result = content
 
                 # Record metrics and log slow calls
                 duration = time.time() - start_time

hindsight-api-slim/tests/test_anthropic_structured_tool_use.py

@@ -0,0 +1,111 @@
+"""Regression tests for issue #1002 — Anthropic structured output via forced tool_use.
+
+When strict_schema=True, AnthropicLLM.call() must request the schema through a single
+forced tool_use tool (tool_choice={"type":"tool",...}) and read the validated args from
+the tool_use block, NOT inject the schema as text and json.loads() the reply (which caused
+a ~1:1 invalid-JSON retry storm / OOM in production).
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from pydantic import BaseModel
+
+
+class _Decision(BaseModel):
+    action: str
+    reason: str
+
+
+def _make_anthropic_provider():
+    with patch("anthropic.AsyncAnthropic") as mock_client_cls:
+        mock_client_cls.return_value = MagicMock()
+        from hindsight_api.engine.providers.anthropic_llm import AnthropicLLM
+
+        provider = AnthropicLLM(
+            provider="anthropic",
+            api_key="fake-key",
+            base_url="",
+            model="claude-sonnet-4-20250514",
+        )
+    provider._client = MagicMock()
+    return provider
+
+
+def _tool_use_response(args: dict):
+    block = MagicMock()
+    block.type = "tool_use"
+    block.name = "structured_response"
+    block.input = args
+    resp = MagicMock()
+    resp.content = [block]
+    resp.usage = MagicMock(input_tokens=5, output_tokens=2, cache_read_input_tokens=0)
+    resp.stop_reason = "tool_use"
+    return resp
+
+
+@pytest.mark.asyncio
+async def test_strict_schema_uses_forced_tool_choice():
+    """strict_schema=True ⇒ a single tool is defined and tool_choice forces it (no schema text-injection)."""
+    provider = _make_anthropic_provider()
+    provider._client.messages.create = AsyncMock(return_value=_tool_use_response({"action": "skip", "reason": "dup"}))
+    with patch("hindsight_api.engine.providers.anthropic_llm.get_metrics_collector"):
+        result = await provider.call(
+            messages=[{"role": "user", "content": "decide"}],
+            response_format=_Decision,
+            strict_schema=True,
+            scope="test",
+            max_retries=0,
+        )
+    kwargs = provider._client.messages.create.call_args.kwargs
+    # forced tool_use requested
+    assert "tools" in kwargs and len(kwargs["tools"]) == 1
+    assert kwargs["tool_choice"] == {"type": "tool", "name": "structured_response"}
+    # schema NOT injected as text into the system prompt
+    assert "valid JSON matching this schema" not in (kwargs.get("system") or "")
+    # validated model returned straight from tool_use.input
+    assert isinstance(result, _Decision)
+    assert result.action == "skip"
+
+
+@pytest.mark.asyncio
+async def test_strict_schema_tool_use_never_hits_json_retry_loop():
+    """A tool_use response is structurally valid → no second messages.create call (no retry storm)."""
+    provider = _make_anthropic_provider()
+    create = AsyncMock(return_value=_tool_use_response({"action": "keep", "reason": "novel"}))
+    provider._client.messages.create = create
+    with patch("hindsight_api.engine.providers.anthropic_llm.get_metrics_collector"):
+        await provider.call(
+            messages=[{"role": "user", "content": "x"}],
+            response_format=_Decision,
+            strict_schema=True,
+            scope="test",
+            max_retries=10,  # would allow 11 attempts on the old text-parse path
+        )
+    assert create.await_count == 1  # exactly one call — the bug was N retries on malformed text
+
+
+@pytest.mark.asyncio
+async def test_non_strict_keeps_text_injection_fallback():
+    """strict_schema=False (default) preserves the legacy schema-in-prompt behavior."""
+    provider = _make_anthropic_provider()
+    block = MagicMock()
+    block.type = "text"
+    block.text = '{"action":"skip","reason":"d"}'
+    resp = MagicMock()
+    resp.content = [block]
+    resp.usage = MagicMock(input_tokens=5, output_tokens=2, cache_read_input_tokens=0)
+    resp.stop_reason = "end_turn"
+    provider._client.messages.create = AsyncMock(return_value=resp)
+    with patch("hindsight_api.engine.providers.anthropic_llm.get_metrics_collector"):
+        result = await provider.call(
+            messages=[{"role": "user", "content": "decide"}],
+            response_format=_Decision,
+            strict_schema=False,
+            scope="test",
+            max_retries=0,
+        )
+    kwargs = provider._client.messages.create.call_args.kwargs
+    assert "tools" not in kwargs  # no forced tool when not strict
+    assert "valid JSON matching this schema" in (kwargs.get("system") or "")
+    assert isinstance(result, _Decision)