[fix] Include response schema in model cache keys (#10621)

## Summary

`_get_model_cache_key` currently serializes a Pydantic response-format
class using its class name. Two different `Output` classes with
different fields therefore generate the same key for identical messages
and model settings. With `cache_response=True`, the second call can
reuse a response generated for the wrong output schema.

Serialize Pydantic response-format classes using `model_json_schema()`
instead. This includes nested field definitions in the cache key while
leaving other class serialization unchanged. This follows up on #7155,
which made class inputs serializable but used the class name as the
cache identity.

## Type of change

- [x] Bug fix

## Checklist

- [x] Code complies with style guidelines
- [x] Self-review completed
- [x] Documentation updated (regression test docstrings; no user-facing
docs change)
- [x] Tests added/updated

### Duplicate and AI-Generated PR Check

- [x] I have searched existing open pull requests and confirmed that no
other PR already addresses this issue
- [x] This PR was generated with AI assistance and reviewed with focused
tests.

## Additional Notes

The new same-name schema tests cover streaming and non-streaming keys;
equivalent schemas keep the same key. A mocked HTTP regression verifies
that sync and async requests do not replay another output schema's
cached response. `pytest -q
libs/agno/tests/unit/models/test_cache_key_serialization.py`: 10 passed.
`ruff check`, `ruff format --check`, and `git diff --check` pass. The
broader model suite was not run locally because optional provider
packages are missing; no provider call is needed for these focused
tests.

---------

Co-authored-by: Sannya Singal <[email protected]>
Co-authored-by: sannya-singal <[email protected]>

libs/agno/agno/models/base.py

@@ -29,7 +29,7 @@ if TYPE_CHECKING:
     from agno.offload.store import ResultStore
 from uuid import uuid4
 
-from pydantic import BaseModel
+from pydantic import BaseModel, PydanticUserError
 
 from agno.exceptions import (
     AgentRunException,
@@ -466,6 +466,12 @@ class Model(ABC):
 
         def _cache_default(obj: Any) -> Any:
             if isinstance(obj, type):
+                if issubclass(obj, BaseModel):
+                    try:
+                        return obj.model_json_schema()
+                    except PydanticUserError:
+                        # Some models (e.g. arbitrary types, Callable fields) have no JSON schema
+                        return f"{obj.__module__}.{obj.__qualname__}"
                 return obj.__name__
             if hasattr(obj, "model_dump"):
                 return obj.model_dump()

libs/agno/tests/unit/models/test_cache_key_serialization.py

@@ -6,9 +6,13 @@ and Pydantic model instances without raising TypeError.
 Regression test for: https://github.com/agno-agi/agno/issues/7126
 """
 
+import json
 import os
+from typing import Callable
 
-from pydantic import BaseModel
+import httpx
+import pytest
+from pydantic import BaseModel, ConfigDict, create_model
 
 os.environ.setdefault("OPENAI_API_KEY", "test-key-for-testing")
 
@@ -55,3 +59,93 @@ class TestCacheKeySerialization:
         )
         key_without = self.model._get_model_cache_key(self.messages, stream=False, response_format=None)
         assert key_with_class != key_without
+
+    @pytest.mark.parametrize("stream", [False, True])
+    def test_same_named_response_formats_have_distinct_cache_keys(self, stream):
+        first_format = create_model("Output", answer=(str, ...))
+        second_format = create_model("Output", count=(int, ...))
+        first_key = self.model._get_model_cache_key(self.messages, stream=stream, response_format=first_format)
+        second_key = self.model._get_model_cache_key(self.messages, stream=stream, response_format=second_format)
+
+        assert first_key != second_key
+
+    @pytest.mark.parametrize("stream", [False, True])
+    def test_equivalent_response_schemas_share_cache_keys(self, stream):
+        first_format = create_model("Output", answer=(str, ...))
+        equivalent_format = create_model("Output", answer=(str, ...))
+
+        assert self.model._get_model_cache_key(
+            self.messages, stream=stream, response_format=first_format
+        ) == self.model._get_model_cache_key(self.messages, stream=stream, response_format=equivalent_format)
+
+    def test_response_format_without_json_schema_falls_back_to_qualified_name(self):
+        class Custom:
+            pass
+
+        class CustomOutput(BaseModel):
+            model_config = ConfigDict(arbitrary_types_allowed=True)
+            value: Custom
+
+        class CallableOutput(BaseModel):
+            handler: Callable
+
+        custom_key = self.model._get_model_cache_key(self.messages, stream=False, response_format=CustomOutput)
+        callable_key = self.model._get_model_cache_key(self.messages, stream=False, response_format=CallableOutput)
+
+        assert custom_key != callable_key
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_response_cache_does_not_replay_another_same_named_schema(tmp_path, asynchronous):
+    first_format = create_model("Output", answer=(str, ...))
+    second_format = create_model("Output", count=(int, ...))
+    requested_fields = []
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        properties = body["response_format"]["json_schema"]["schema"]["properties"]
+        requested_fields.append(list(properties))
+        content = {"answer": "sample"} if "answer" in properties else {"count": 2}
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-test",
+                "object": "chat.completion",
+                "created": 0,
+                "model": "test-model",
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": json.dumps(content)},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+            },
+        )
+
+    transport = httpx.MockTransport(respond)
+    with httpx.Client(transport=transport) as sync_client:
+        async with httpx.AsyncClient(transport=transport) as async_client:
+            model = OpenAIChat(
+                id="test-model",
+                api_key="test-key",
+                cache_response=True,
+                cache_dir=str(tmp_path),
+                http_client=async_client if asynchronous else sync_client,
+            )
+            for response_format, expected in [
+                (first_format, {"answer": "sample"}),
+                (second_format, {"count": 2}),
+                (first_format, {"answer": "sample"}),
+            ]:
+                messages = [Message(role="user", content="Return the requested result")]
+                if asynchronous:
+                    result = await model.aresponse(messages, response_format=response_format)
+                else:
+                    result = model.response(messages, response_format=response_format)
+                assert json.loads(result.content) == expected
+
+    assert requested_fields == [["answer"], ["count"]]
+    assert len(list(tmp_path.glob("*.json"))) == 2