Merge branch 'main' into feat/gpt5-support

.github/ISSUE_TEMPLATE/1-bug_report.yml

@@ -90,6 +90,8 @@ body:
       multiple: false
       options:
         - "Python dev (main branch)"
+        - "Python 0.7.4"
+        - "Python 0.7.3"
         - "Python 0.7.2"
         - "Python 0.7.1"
         - "Python 0.6.4"

.github/workflows/docs.yml

@@ -39,7 +39,7 @@ jobs:
               poe-dir: ".",
             },
             {
-              ref: "python-v0.7.2",
+              ref: "python-v0.7.4",
               dest-dir: stable,
               uv-version: "0.7.13",
               sphinx-release-override: "stable",
@@ -199,6 +199,20 @@ jobs:
               sphinx-release-override: "",
               poe-dir: ".",
             },
+            {
+              ref: "python-v0.7.3",
+              dest-dir: "0.7.3",
+              uv-version: "0.7.13",
+              sphinx-release-override: "",
+              poe-dir: ".",
+            },
+            {
+              ref: "python-v0.7.4",
+              dest-dir: "0.7.4",
+              uv-version: "0.7.13",
+              sphinx-release-override: "",
+              poe-dir: ".",
+            },
           ]
     steps:
       - name: Checkout
@@ -216,8 +230,6 @@ jobs:
       - run: |
           uv venv --python=3.11
           source .venv/bin/activate
-          # Only pin version to ensure compatibility
-          uv pip install -U 'setuptools-scm==8.3.1'
           uv sync --locked --all-extras
           poe --directory ${{ matrix.version.poe-dir }} docs-build
           mkdir -p docs-staging/${{ matrix.version.dest-dir }}/

README.md

@@ -43,7 +43,7 @@ from autogen_agentchat.agents import AssistantAgent
 from autogen_ext.models.openai import OpenAIChatCompletionClient
 
 async def main() -> None:
-    model_client = OpenAIChatCompletionClient(model="gpt-4o")
+    model_client = OpenAIChatCompletionClient(model="gpt-4.1")
     agent = AssistantAgent("assistant", model_client=model_client)
     print(await agent.run(task="Say 'Hello World!'"))
     await model_client.close()
@@ -90,6 +90,58 @@ asyncio.run(main())
 > **Warning**: Only connect to trusted MCP servers as they may execute commands
 > in your local environment or expose sensitive information.
 
+### Multi-Agent Orchestration
+
+You can use `AgentTool` to create a basic multi-agent orchestration setup.
+
+```python
+import asyncio
+
+from autogen_agentchat.agents import AssistantAgent
+from autogen_agentchat.tools import AgentTool
+from autogen_agentchat.ui import Console
+from autogen_ext.models.openai import OpenAIChatCompletionClient
+
+
+async def main() -> None:
+    model_client = OpenAIChatCompletionClient(model="gpt-4.1")
+
+    math_agent = AssistantAgent(
+        "math_expert",
+        model_client=model_client,
+        system_message="You are a math expert.",
+        description="A math expert assistant.",
+        model_client_stream=True,
+    )
+    math_agent_tool = AgentTool(math_agent, return_value_as_last_message=True)
+
+    chemistry_agent = AssistantAgent(
+        "chemistry_expert",
+        model_client=model_client,
+        system_message="You are a chemistry expert.",
+        description="A chemistry expert assistant.",
+        model_client_stream=True,
+    )
+    chemistry_agent_tool = AgentTool(chemistry_agent, return_value_as_last_message=True)
+
+    agent = AssistantAgent(
+        "assistant",
+        system_message="You are a general assistant. Use expert tools when needed.",
+        model_client=model_client,
+        model_client_stream=True,
+        tools=[math_agent_tool, chemistry_agent_tool],
+        max_tool_iterations=10,
+    )
+    await Console(agent.run_stream(task="What is the integral of x^2?"))
+    await Console(agent.run_stream(task="What is the molecular weight of water?"))
+
+
+asyncio.run(main())
+```
+
+For more advanced multi-agent orchestrations and workflows, read
+[AgentChat documentation](https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/index.html).
+
 ### AutoGen Studio
 
 Use AutoGen Studio to prototype and run multi-agent workflows without writing code.

docs/dotnet/core/index.md

@@ -81,7 +81,7 @@ You can run the backend on its own:
 dotnet run --project Microsoft.AutoGen.AgentHost
 ```
 
-or you can run iclude it inside your own application:
+or you can include it inside your own application:
 
 ```csharp
 using Microsoft.AutoGen.RuntimeGateway;

docs/switcher.json

@@ -5,11 +5,21 @@
         "url": "/autogen/dev/"
     },
     {
-        "name": "0.7.2 (stable)",
+        "name": "0.7.4 (stable)",
         "version": "stable",
         "url": "/autogen/stable/",
         "preferred": true
     },
+    {
+        "name": "0.7.3",
+        "version": "0.7.3",
+        "url": "/autogen/0.7.3/"
+    },
+    {
+        "name": "0.7.2",
+        "version": "0.7.2",
+        "url": "/autogen/0.7.2/"
+    },
     {
         "name": "0.7.1",
         "version": "0.7.1",

python/packages/autogen-agentchat/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "autogen-agentchat"
-version = "0.7.2"
+version = "0.7.4"
 license = {file = "LICENSE-CODE"}
 description = "AutoGen agents and teams library"
 readme = "README.md"
@@ -15,7 +15,7 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    "autogen-core==0.7.2",
+    "autogen-core==0.7.4",
 ]
 
 [tool.ruff]

python/packages/autogen-core/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "autogen-core"
-version = "0.7.2"
+version = "0.7.4"
 license = {file = "LICENSE-CODE"}
 description = "Foundational interfaces and agent runtime implementation for AutoGen"
 readme = "README.md"

python/packages/autogen-ext/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "autogen-ext"
-version = "0.7.2"
+version = "0.7.4"
 license = {file = "LICENSE-CODE"}
 description = "AutoGen extensions library"
 readme = "README.md"
@@ -15,7 +15,7 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    "autogen-core==0.7.2",
+    "autogen-core==0.7.4",
 ]
 
 [project.optional-dependencies]
@@ -32,7 +32,7 @@ docker = ["docker~=7.0", "asyncio_atexit>=1.0.1"]
 ollama = ["ollama>=0.4.7", "tiktoken>=0.8.0"]
 openai = ["openai>=1.93", "tiktoken>=0.8.0", "aiofiles"]
 file-surfer = [
-    "autogen-agentchat==0.7.2",
+    "autogen-agentchat==0.7.4",
     "magika>=0.6.1rc2",
     "markitdown[all]~=0.1.0a3",
 ]
@@ -50,21 +50,21 @@ mem0-local = [
     "chromadb>=1.0.0"
 ]
 web-surfer = [
-    "autogen-agentchat==0.7.2",
+    "autogen-agentchat==0.7.4",
     "playwright>=1.48.0",
     "pillow>=11.0.0",
     "magika>=0.6.1rc2",
     "markitdown[all]~=0.1.0a3",
 ]
 magentic-one = [
-    "autogen-agentchat==0.7.2",
+    "autogen-agentchat==0.7.4",
     "magika>=0.6.1rc2",
     "markitdown[all]~=0.1.0a3",
     "playwright>=1.48.0",
     "pillow>=11.0.0",
 ]
 video-surfer = [
-    "autogen-agentchat==0.7.2",
+    "autogen-agentchat==0.7.4",
     "opencv-python>=4.5",
     "ffmpeg-python",
     "openai-whisper",

python/packages/autogen-ext/src/autogen_ext/cache_store/redis.py

@@ -90,6 +90,7 @@ class RedisStore(CacheStore[T], Component[RedisStoreConfig]):
 
         This method handles both primitive values and complex objects:
         - Pydantic models are automatically serialized to JSON
+        - Lists containing Pydantic models are serialized to JSON
         - Primitive values (strings, numbers, etc.) are stored as-is
 
         Args:
@@ -101,10 +102,21 @@ class RedisStore(CacheStore[T], Component[RedisStoreConfig]):
                 # Serialize Pydantic models to JSON
                 serialized_value = value.model_dump_json().encode("utf-8")
                 self.cache.set(key, serialized_value)
+            elif isinstance(value, list):
+                # Serialize lists (which may contain Pydantic models) to JSON
+                serializable_list: list[Any] = []
+                item: Any
+                for item in value:
+                    if isinstance(item, BaseModel):
+                        serializable_list.append(item.model_dump())
+                    else:
+                        serializable_list.append(item)
+                serialized_value = json.dumps(serializable_list).encode("utf-8")
+                self.cache.set(key, serialized_value)
             else:
                 # Backward compatibility for primitives
                 self.cache.set(key, cast(Any, value))
-        except (redis.RedisError, ConnectionError, UnicodeEncodeError):
+        except (redis.RedisError, ConnectionError, UnicodeEncodeError, TypeError):
             # Log the error but don't re-raise to maintain robustness
             pass
 

python/packages/autogen-ext/src/autogen_ext/models/cache/_chat_completion_cache.py

@@ -1,7 +1,7 @@
 import hashlib
 import json
 import warnings
-from typing import Any, AsyncGenerator, List, Literal, Mapping, Optional, Sequence, Union, cast
+from typing import Any, AsyncGenerator, List, Literal, Mapping, Optional, Sequence, Union
 
 from autogen_core import CacheStore, CancellationToken, Component, ComponentModel, InMemoryStore
 from autogen_core.models import (
@@ -13,7 +13,7 @@ from autogen_core.models import (
     RequestUsage,
 )
 from autogen_core.tools import Tool, ToolSchema
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing_extensions import Self
 
 CHAT_CACHE_VALUE_TYPE = Union[CreateResult, List[Union[str, CreateResult]]]
@@ -77,6 +77,81 @@ class ChatCompletionCache(ChatCompletionClient, Component[ChatCompletionCacheCon
 
         asyncio.run(main())
 
+    For Redis caching:
+
+    .. code-block:: python
+
+        import asyncio
+
+        from autogen_core.models import UserMessage
+        from autogen_ext.models.openai import OpenAIChatCompletionClient
+        from autogen_ext.models.cache import ChatCompletionCache, CHAT_CACHE_VALUE_TYPE
+        from autogen_ext.cache_store.redis import RedisStore
+        import redis
+
+
+        async def main():
+            # Initialize the original client
+            openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")
+
+            # Initialize Redis cache store
+            redis_instance = redis.Redis()
+            cache_store = RedisStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
+            cache_client = ChatCompletionCache(openai_model_client, cache_store)
+
+            response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
+            print(response)  # Should print response from OpenAI
+            response = await cache_client.create([UserMessage(content="Hello, how are you?", source="user")])
+            print(response)  # Should print cached response
+
+
+        asyncio.run(main())
+
+    For streaming with Redis caching:
+
+    .. code-block:: python
+
+        import asyncio
+
+        from autogen_core.models import UserMessage, CreateResult
+        from autogen_ext.models.openai import OpenAIChatCompletionClient
+        from autogen_ext.models.cache import ChatCompletionCache, CHAT_CACHE_VALUE_TYPE
+        from autogen_ext.cache_store.redis import RedisStore
+        import redis
+
+
+        async def main():
+            # Initialize the original client
+            openai_model_client = OpenAIChatCompletionClient(model="gpt-4o")
+
+            # Initialize Redis cache store
+            redis_instance = redis.Redis()
+            cache_store = RedisStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
+            cache_client = ChatCompletionCache(openai_model_client, cache_store)
+
+            # First streaming call
+            async for chunk in cache_client.create_stream(
+                [UserMessage(content="List all countries in Africa", source="user")]
+            ):
+                if isinstance(chunk, CreateResult):
+                    print("\\n")
+                    print("Cached: ", chunk.cached)  # Should print False
+                else:
+                    print(chunk, end="")
+
+            # Second streaming call (cached)
+            async for chunk in cache_client.create_stream(
+                [UserMessage(content="List all countries in Africa", source="user")]
+            ):
+                if isinstance(chunk, CreateResult):
+                    print("\\n")
+                    print("Cached: ", chunk.cached)  # Should print True
+                else:
+                    print(chunk, end="")
+
+
+        asyncio.run(main())
+
     You can now use the `cached_client` as you would the original client, but with caching enabled.
 
     Args:
@@ -126,8 +201,29 @@ class ChatCompletionCache(ChatCompletionClient, Component[ChatCompletionCacheCon
         serialized_data = json.dumps(data, sort_keys=True)
         cache_key = hashlib.sha256(serialized_data.encode()).hexdigest()
 
-        cached_result = cast(Optional[CreateResult], self.store.get(cache_key))
+        cached_result = self.store.get(cache_key)
         if cached_result is not None:
+            # Handle case where cache store returns dict instead of CreateResult (e.g., Redis)
+            if isinstance(cached_result, dict):
+                try:
+                    cached_result = CreateResult.model_validate(cached_result)
+                except ValidationError:
+                    # If reconstruction fails, treat as cache miss
+                    return None, cache_key
+            elif isinstance(cached_result, list):
+                # Handle streaming results - reconstruct CreateResult instances from dicts
+                try:
+                    reconstructed_list: List[Union[str, CreateResult]] = []
+                    for item in cached_result:
+                        if isinstance(item, dict):
+                            reconstructed_list.append(CreateResult.model_validate(item))
+                        else:
+                            reconstructed_list.append(item)
+                    cached_result = reconstructed_list
+                except ValidationError:
+                    # If reconstruction fails, treat as cache miss
+                    return None, cache_key
+            # If it's already the right type (CreateResult or list), return as-is
             return cached_result, cache_key
 
         return None, cache_key
@@ -150,10 +246,18 @@ class ChatCompletionCache(ChatCompletionClient, Component[ChatCompletionCacheCon
         NOTE: cancellation_token is ignored for cached results.
         """
         cached_result, cache_key = self._check_cache(messages, tools, json_output, extra_create_args)
-        if cached_result:
-            assert isinstance(cached_result, CreateResult)
-            cached_result.cached = True
-            return cached_result
+        if cached_result is not None:
+            if isinstance(cached_result, CreateResult):
+                # Cache hit from previous non-streaming call
+                cached_result.cached = True
+                return cached_result
+            elif isinstance(cached_result, list):
+                # Cache hit from previous streaming call - extract the final CreateResult
+                for item in reversed(cached_result):
+                    if isinstance(item, CreateResult):
+                        item.cached = True
+                        return item
+                # If no CreateResult found in list, fall through to make actual call
 
         result = await self.client.create(
             messages,
@@ -191,13 +295,24 @@ class ChatCompletionCache(ChatCompletionClient, Component[ChatCompletionCacheCon
                 json_output,
                 extra_create_args,
             )
-            if cached_result:
-                assert isinstance(cached_result, list)
-                for result in cached_result:
-                    if isinstance(result, CreateResult):
-                        result.cached = True
-                    yield result
-                return
+            if cached_result is not None:
+                if isinstance(cached_result, list):
+                    # Cache hit from previous streaming call
+                    for result in cached_result:
+                        if isinstance(result, CreateResult):
+                            result.cached = True
+                        yield result
+                    return
+                elif isinstance(cached_result, CreateResult):
+                    # Cache hit from previous non-streaming call - convert to streaming format
+                    cached_result.cached = True
+
+                    # If content is a non-empty string, yield it as a streaming chunk first
+                    if isinstance(cached_result.content, str) and cached_result.content:
+                        yield cached_result.content
+
+                    yield cached_result
+                    return
 
             result_stream = self.client.create_stream(
                 messages,
@@ -209,12 +324,14 @@ class ChatCompletionCache(ChatCompletionClient, Component[ChatCompletionCacheCon
             )
 
             output_results: List[Union[str, CreateResult]] = []
-            self.store.set(cache_key, output_results)
 
             async for result in result_stream:
                 output_results.append(result)
                 yield result
 
+            # Store the complete results only after streaming is finished
+            self.store.set(cache_key, output_results)
+
         return _generator()
 
     async def close(self) -> None:

python/packages/autogen-ext/tests/cache_store/test_redis_store.py

@@ -1,8 +1,9 @@
 import json
-from typing import cast
+from typing import Dict, List, Union, cast
 from unittest.mock import MagicMock
 
 import pytest
+from autogen_core.models import CreateResult, RequestUsage
 from pydantic import BaseModel
 
 redis = pytest.importorskip("redis")
@@ -180,3 +181,220 @@ def test_redis_store_nested_model_serialization() -> None:
     assert retrieved_model["nested"]["id"] == 1  # type: ignore
     assert retrieved_model["nested"]["data"] == "nested_data"  # type: ignore
     assert retrieved_model["tags"] == ["tag1", "tag2", "tag3"]  # type: ignore
+
+
+def test_redis_store_list_with_strings_only() -> None:
+    """Test serialization of lists containing only strings (streaming scenario)."""
+    from autogen_ext.cache_store.redis import RedisStore
+
+    redis_instance = MagicMock()
+    store = RedisStore[List[Union[str, CreateResult]]](redis_instance)
+    test_key = "test_string_list_key"
+
+    # Create a list with only strings (partial streaming result)
+    string_list: List[Union[str, CreateResult]] = ["Hello", " world", "!", " How", " are", " you", "?"]
+
+    # Test setting the list
+    store.set(test_key, string_list)
+
+    # Verify Redis was called with JSON-serialized data
+    args, _ = redis_instance.set.call_args
+    assert args[0] == test_key
+    assert isinstance(args[1], bytes)
+
+    # Verify the serialized data is correct
+    serialized_json = args[1].decode("utf-8")
+    deserialized_data = json.loads(serialized_json)
+    assert deserialized_data == string_list
+
+    # Test retrieving the list
+    redis_instance.get.return_value = args[1]  # Return the serialized data
+    retrieved_list = store.get(test_key)
+
+    assert retrieved_list is not None
+    assert isinstance(retrieved_list, list)
+    assert retrieved_list == string_list
+
+
+def test_redis_store_list_with_create_results_only() -> None:
+    """Test serialization of lists containing only CreateResult objects."""
+    from autogen_ext.cache_store.redis import RedisStore
+
+    redis_instance = MagicMock()
+    store = RedisStore[List[Union[str, CreateResult]]](redis_instance)
+    test_key = "test_create_result_list_key"
+
+    # Create a list with only CreateResult objects
+    usage = RequestUsage(prompt_tokens=10, completion_tokens=20)
+    create_result_list: List[Union[str, CreateResult]] = [
+        CreateResult(content="First response", usage=usage, finish_reason="stop", cached=False),
+        CreateResult(content="Second response", usage=usage, finish_reason="stop", cached=False),
+    ]
+
+    # Test setting the list
+    store.set(test_key, create_result_list)
+
+    # Verify Redis was called with JSON-serialized data
+    args, _ = redis_instance.set.call_args
+    assert args[0] == test_key
+    assert isinstance(args[1], bytes)
+
+    # Verify the serialized data structure
+    serialized_json = args[1].decode("utf-8")
+    deserialized_data = json.loads(serialized_json)
+
+    assert isinstance(deserialized_data, list)
+    # Type narrowing: after isinstance check, deserialized_data is known to be a list
+    deserialized_list: List[Dict[str, Union[str, int]]] = deserialized_data  # Now properly typed as list
+    assert len(deserialized_list) == 2
+    assert deserialized_list[0]["content"] == "First response"
+    assert deserialized_list[1]["content"] == "Second response"
+    assert deserialized_list[0]["finish_reason"] == "stop"
+
+    # Test retrieving the list
+    redis_instance.get.return_value = args[1]  # Return the serialized data
+    retrieved_list = store.get(test_key)
+
+    assert retrieved_list is not None
+    assert isinstance(retrieved_list, list)
+    assert len(retrieved_list) == 2
+
+    # The retrieved items should be dicts (as Redis returns JSON-parsed objects)
+    assert isinstance(retrieved_list[0], dict)
+    assert isinstance(retrieved_list[1], dict)
+    assert retrieved_list[0]["content"] == "First response"  # type: ignore
+    assert retrieved_list[1]["content"] == "Second response"  # type: ignore
+
+
+def test_redis_store_mixed_list_streaming_scenario() -> None:
+    """Test serialization of mixed lists (strings + CreateResult) for streaming cache scenario."""
+    from autogen_ext.cache_store.redis import RedisStore
+
+    redis_instance = MagicMock()
+    store = RedisStore[List[Union[str, CreateResult]]](redis_instance)
+    test_key = "test_mixed_streaming_list_key"
+
+    # Create a mixed list simulating a streaming response
+    usage = RequestUsage(prompt_tokens=15, completion_tokens=30)
+    mixed_list: List[Union[str, CreateResult]] = [
+        "The",
+        " capital",
+        " of",
+        " France",
+        " is",
+        " Paris",
+        ".",
+        CreateResult(content="The capital of France is Paris.", usage=usage, finish_reason="stop", cached=False),
+    ]
+
+    # Test setting the mixed list
+    store.set(test_key, mixed_list)
+
+    # Verify Redis was called with JSON-serialized data
+    args, _ = redis_instance.set.call_args
+    assert args[0] == test_key
+    assert isinstance(args[1], bytes)
+
+    # Verify the serialized data structure
+    serialized_json = args[1].decode("utf-8")
+    deserialized_data = json.loads(serialized_json)
+
+    assert isinstance(deserialized_data, list)
+    # Type narrowing: after isinstance check, deserialized_data is known to be a list
+    deserialized_list: List[Union[str, Dict[str, Union[str, int]]]] = deserialized_data  # Now properly typed as list
+    assert len(deserialized_list) == 8  # 7 strings + 1 CreateResult
+
+    # First 7 items should be strings
+    for i in range(7):
+        assert isinstance(deserialized_list[i], str)
+
+    # Last item should be the serialized CreateResult (as dict)
+    assert isinstance(deserialized_list[7], dict)
+    assert deserialized_list[7]["content"] == "The capital of France is Paris."
+    assert deserialized_list[7]["finish_reason"] == "stop"
+    assert deserialized_data[7]["usage"]["prompt_tokens"] == 15
+    assert deserialized_data[7]["usage"]["completion_tokens"] == 30
+
+    # Test retrieving the mixed list
+    redis_instance.get.return_value = args[1]  # Return the serialized data
+    retrieved_list = store.get(test_key)
+
+    assert retrieved_list is not None
+    assert isinstance(retrieved_list, list)
+    assert len(retrieved_list) == 8
+
+    # First 7 items should still be strings
+    for i in range(7):
+        assert isinstance(retrieved_list[i], str)
+        assert retrieved_list[i] == mixed_list[i]
+
+    # Last item should be a dict (CreateResult deserialized from JSON)
+    assert isinstance(retrieved_list[7], dict)
+    assert retrieved_list[7]["content"] == "The capital of France is Paris."  # type: ignore
+    assert retrieved_list[7]["cached"] is False  # type: ignore
+
+
+def test_redis_store_empty_list() -> None:
+    """Test serialization of empty lists."""
+    from autogen_ext.cache_store.redis import RedisStore
+
+    redis_instance = MagicMock()
+    store = RedisStore[List[Union[str, CreateResult]]](redis_instance)
+    test_key = "test_empty_list_key"
+
+    # Test setting an empty list
+    empty_list: List[Union[str, CreateResult]] = []
+    store.set(test_key, empty_list)
+
+    # Verify Redis was called with JSON-serialized data
+    args, _ = redis_instance.set.call_args
+    assert args[0] == test_key
+    assert isinstance(args[1], bytes)
+
+    # Verify the serialized data is an empty JSON array
+    serialized_json = args[1].decode("utf-8")
+    deserialized_data = json.loads(serialized_json)
+    assert deserialized_data == []
+
+    # Test retrieving the empty list
+    redis_instance.get.return_value = args[1]
+    retrieved_list = store.get(test_key)
+
+    assert retrieved_list is not None
+    assert isinstance(retrieved_list, list)
+    assert len(retrieved_list) == 0
+
+
+def test_redis_store_list_serialization_error_handling() -> None:
+    """Test error handling during list serialization."""
+    from autogen_ext.cache_store.redis import RedisStore
+
+    redis_instance = MagicMock()
+    store = RedisStore[List[Union[str, CreateResult]]](redis_instance)
+
+    # Test Redis error during set
+    redis_instance.set.side_effect = redis.RedisError("Redis connection failed")
+
+    mixed_list: List[Union[str, CreateResult]] = [
+        "test",
+        CreateResult(
+            content="test content",
+            usage=RequestUsage(prompt_tokens=1, completion_tokens=1),
+            finish_reason="stop",
+            cached=False,
+        ),
+    ]
+
+    # This should not raise an exception due to our try/except block
+    try:
+        store.set("error_key", mixed_list)
+    except Exception:
+        pytest.fail("set() method didn't handle the Redis exception properly")
+
+    # Test get with corrupted JSON data for lists
+    redis_instance.get.side_effect = None  # Reset side effect
+    redis_instance.get.return_value = b'[{"invalid": json}]'  # Invalid JSON
+
+    retrieved_value = store.get("corrupted_key", default=[])
+    # Should return the decoded string when JSON parsing fails (backward compatibility)
+    assert retrieved_value == '[{"invalid": json}]'  # type: ignore[comparison-overlap]

python/packages/autogen-ext/tests/models/test_chat_completion_cache.py

@@ -1,15 +1,17 @@
 import copy
-from typing import List, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
 import pytest
+from autogen_core import CacheStore, FunctionCall
 from autogen_core.models import (
     ChatCompletionClient,
     CreateResult,
     LLMMessage,
+    RequestUsage,
     SystemMessage,
     UserMessage,
 )
-from autogen_ext.models.cache import ChatCompletionCache
+from autogen_ext.models.cache import CHAT_CACHE_VALUE_TYPE, ChatCompletionCache
 from autogen_ext.models.replay import ReplayChatCompletionClient
 from pydantic import BaseModel
 
@@ -184,3 +186,650 @@ async def test_cache_create_stream() -> None:
     # cached_client_config = cached_client.dump_component()
     # loaded_client = ChatCompletionCache.load_component(cached_client_config)
     # assert loaded_client.client == cached_client.client
+
+
+class MockCacheStore(CacheStore[CHAT_CACHE_VALUE_TYPE]):
+    """Mock cache store for testing deserialization scenarios."""
+
+    def __init__(self, return_value: Optional[CHAT_CACHE_VALUE_TYPE] = None) -> None:
+        self._return_value = return_value
+        self._storage: Dict[str, CHAT_CACHE_VALUE_TYPE] = {}
+
+    def get(self, key: str, default: Optional[CHAT_CACHE_VALUE_TYPE] = None) -> Optional[CHAT_CACHE_VALUE_TYPE]:
+        return self._return_value  # type: ignore
+
+    def set(self, key: str, value: CHAT_CACHE_VALUE_TYPE) -> None:
+        self._storage[key] = value
+
+    def _to_config(self) -> BaseModel:
+        """Dummy implementation for testing."""
+        return BaseModel()
+
+    @classmethod
+    def _from_config(cls, _config: BaseModel) -> "MockCacheStore":
+        """Dummy implementation for testing."""
+        return cls()
+
+
+def test_check_cache_redis_dict_deserialization_success() -> None:
+    """Test _check_cache when Redis cache returns a dict that can be deserialized to CreateResult.
+    This tests the core Redis deserialization fix where Redis returns serialized Pydantic
+    models as dictionaries instead of the original objects.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a CreateResult instance (simulating deserialized Redis data)
+    create_result = CreateResult(
+        content="test response from redis",
+        usage=RequestUsage(prompt_tokens=15, completion_tokens=8),
+        cached=False,
+        finish_reason="stop",
+    )
+
+    # Mock cache store that returns a CreateResult (simulating Redis behavior)
+    mock_store = MockCacheStore(return_value=create_result)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method directly using proper test data
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    assert cached_result is not None
+    assert isinstance(cached_result, CreateResult)
+    assert cached_result.content == "test response from redis"
+    assert cache_key is not None
+
+
+def test_check_cache_redis_dict_deserialization_failure() -> None:
+    """Test _check_cache gracefully handles corrupted Redis data.
+    This ensures the system degrades gracefully when Redis returns corrupted
+    or invalid data that cannot be deserialized back to CreateResult.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Mock cache store that returns None (simulating deserialization failure)
+    mock_store = MockCacheStore(return_value=None)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method directly using proper test data
+    messages = [system_prompt, UserMessage(content=prompts[1], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should return None (cache miss) when deserialization fails
+    assert cached_result is None
+    assert cache_key is not None
+
+
+def test_check_cache_redis_streaming_dict_deserialization() -> None:
+    """Test _check_cache with Redis streaming data containing dicts that need deserialization.
+    This tests the streaming scenario where Redis returns a list containing
+    serialized CreateResult objects as dictionaries mixed with string chunks.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a list with CreateResult objects mixed with strings (streaming scenario)
+    create_result = CreateResult(
+        content="final streaming response from redis",
+        usage=RequestUsage(prompt_tokens=12, completion_tokens=6),
+        cached=False,
+        finish_reason="stop",
+    )
+
+    cached_list: List[Union[str, CreateResult]] = [
+        "streaming chunk 1",
+        create_result,  # Proper CreateResult object
+        "streaming chunk 2",
+    ]
+
+    # Mock cache store that returns the list with CreateResults (simulating Redis streaming)
+    mock_store = MockCacheStore(return_value=cached_list)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method directly using proper test data
+    messages = [system_prompt, UserMessage(content=prompts[2], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    assert cached_result is not None
+    assert isinstance(cached_result, list)
+    assert len(cached_result) == 3
+    assert cached_result[0] == "streaming chunk 1"
+    assert isinstance(cached_result[1], CreateResult)
+    assert cached_result[1].content == "final streaming response from redis"
+    assert cached_result[2] == "streaming chunk 2"
+    assert cache_key is not None
+
+
+def test_check_cache_redis_streaming_deserialization_failure() -> None:
+    """Test _check_cache gracefully handles corrupted Redis streaming data.
+    This ensures the system degrades gracefully when Redis returns streaming
+    data with corrupted CreateResult dictionaries that cannot be deserialized.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data(num_messages=4)
+
+    # Mock cache store that returns None (simulating deserialization failure)
+    mock_store = MockCacheStore(return_value=None)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method directly using proper test data
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should return None (cache miss) when deserialization fails
+    assert cached_result is None
+    assert cache_key is not None
+
+
+def test_check_cache_dict_reconstruction_success() -> None:
+    """Test _check_cache successfully reconstructs CreateResult from a dict.
+    This tests the line: cached_result = CreateResult.model_validate(cached_result)
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a dict that can be successfully validated as CreateResult
+    valid_dict = {
+        "content": "reconstructed response",
+        "usage": {"prompt_tokens": 10, "completion_tokens": 5},
+        "cached": False,
+        "finish_reason": "stop",
+    }
+
+    # Create a MockCacheStore that returns the dict directly (simulating Redis)
+    mock_store = MockCacheStore(return_value=cast(Any, valid_dict))
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should successfully reconstruct the CreateResult from dict
+    assert cached_result is not None
+    assert isinstance(cached_result, CreateResult)
+    assert cached_result.content == "reconstructed response"
+    assert cache_key is not None
+
+
+def test_check_cache_dict_reconstruction_failure() -> None:
+    """Test _check_cache handles ValidationError when dict cannot be reconstructed.
+    This tests the except ValidationError block for single dicts.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create an invalid dict that will fail CreateResult validation
+    invalid_dict = {
+        "invalid_field": "value",
+        "missing_required_fields": True,
+    }
+
+    # Create a MockCacheStore that returns the invalid dict
+    mock_store = MockCacheStore(return_value=cast(Any, invalid_dict))
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should return None (cache miss) when reconstruction fails
+    assert cached_result is None
+    assert cache_key is not None
+
+
+def test_check_cache_list_reconstruction_success() -> None:
+    """Test _check_cache successfully reconstructs CreateResult objects from dicts in a list.
+    This tests the line: reconstructed_list.append(CreateResult.model_validate(item))
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a list with valid dicts that can be reconstructed
+    valid_dict1 = {
+        "content": "first result",
+        "usage": {"prompt_tokens": 8, "completion_tokens": 3},
+        "cached": False,
+        "finish_reason": "stop",
+    }
+    valid_dict2 = {
+        "content": "second result",
+        "usage": {"prompt_tokens": 12, "completion_tokens": 7},
+        "cached": False,
+        "finish_reason": "stop",
+    }
+
+    cached_list = [
+        "streaming chunk 1",
+        valid_dict1,
+        "streaming chunk 2",
+        valid_dict2,
+    ]
+
+    # Create a MockCacheStore that returns the list with dicts
+    mock_store = MockCacheStore(return_value=cast(Any, cached_list))
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should successfully reconstruct the list with CreateResult objects
+    assert cached_result is not None
+    assert isinstance(cached_result, list)
+    assert len(cached_result) == 4
+    assert cached_result[0] == "streaming chunk 1"
+    assert isinstance(cached_result[1], CreateResult)
+    assert cached_result[1].content == "first result"
+    assert cached_result[2] == "streaming chunk 2"
+    assert isinstance(cached_result[3], CreateResult)
+    assert cached_result[3].content == "second result"
+    assert cache_key is not None
+
+
+def test_check_cache_list_reconstruction_failure() -> None:
+    """Test _check_cache handles ValidationError when list contains invalid dicts.
+    This tests the except ValidationError block for lists.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a list with an invalid dict that will fail validation
+    invalid_dict = {
+        "invalid_field": "value",
+        "missing_required": True,
+    }
+
+    cached_list = [
+        "streaming chunk 1",
+        invalid_dict,  # This will cause ValidationError
+        "streaming chunk 2",
+    ]
+
+    # Create a MockCacheStore that returns the list with invalid dict
+    mock_store = MockCacheStore(return_value=cast(Any, cached_list))
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should return None (cache miss) when list reconstruction fails
+    assert cached_result is None
+    assert cache_key is not None
+
+
+def test_check_cache_already_correct_type() -> None:
+    """Test _check_cache returns data as-is when it's already the correct type.
+    This tests the final return path when no reconstruction is needed.
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a proper CreateResult object (already correct type)
+    create_result = CreateResult(
+        content="already correct type",
+        usage=RequestUsage(prompt_tokens=15, completion_tokens=8),
+        cached=False,
+        finish_reason="stop",
+    )
+
+    # Create a MockCacheStore that returns the proper type
+    mock_store = MockCacheStore(return_value=create_result)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Test _check_cache method
+    messages = [system_prompt, UserMessage(content=prompts[0], source="user")]
+    cached_result, cache_key = cached_client._check_cache(messages, [], None, {})  # type: ignore
+
+    # Should return the same object without reconstruction
+    assert cached_result is not None
+    assert cached_result is create_result  # Same object reference
+    assert isinstance(cached_result, CreateResult)
+    assert cached_result.content == "already correct type"
+    assert cache_key is not None
+
+
+@pytest.mark.asyncio
+async def test_redis_streaming_cache_integration() -> None:
+    """Integration test for Redis streaming cache scenario.
+    This test covers the original streaming cache issues:
+    1. Cache is stored after streaming completes (not before)
+    2. Redis cache properly handles lists containing CreateResult objects
+    3. ChatCompletionCache properly reconstructs CreateResult from Redis dicts
+    """
+    from unittest.mock import MagicMock
+
+    # Skip this test if redis is not available
+    pytest.importorskip("redis")
+
+    from autogen_ext.cache_store.redis import RedisStore
+
+    # Use standardized test data
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Mock Redis instance to control what gets stored/retrieved
+    redis_instance = MagicMock()
+    redis_store = RedisStore[CHAT_CACHE_VALUE_TYPE](redis_instance)
+
+    # Create the cached client with Redis store
+    cached_client = ChatCompletionCache(replay_client, redis_store)
+
+    # Simulate first streaming call (should cache after completion)
+    first_stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        first_stream_results.append(copy.copy(chunk))
+
+    # Verify Redis set was called with the complete streaming results
+    redis_instance.set.assert_called_once()
+    call_args = redis_instance.set.call_args
+    serialized_data = call_args[0][1]
+
+    # Verify the serialized data represents the complete stream
+    assert isinstance(serialized_data, bytes)
+    import json
+
+    deserialized = json.loads(serialized_data.decode("utf-8"))
+    assert isinstance(deserialized, list)
+    # Type narrowing: after isinstance check, deserialized is known to be a list
+    deserialized_list: List[Union[str, Dict[str, Union[str, int]]]] = deserialized  # Now properly typed as list
+    # Should contain both string chunks and final CreateResult (as dict)
+    assert len(deserialized_list) > 0
+
+    # Reset the mock for the second call
+    redis_instance.reset_mock()
+
+    # Configure Redis to return the serialized data (simulating cache hit)
+    redis_instance.get.return_value = serialized_data
+
+    # Second streaming call should hit the cache
+    second_stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        second_stream_results.append(copy.copy(chunk))
+
+    # Verify Redis get was called but set was not (cache hit)
+    redis_instance.get.assert_called_once()
+    redis_instance.set.assert_not_called()
+
+    # Verify both streams have the same content
+    assert len(first_stream_results) == len(second_stream_results)
+
+    # Verify cached results are marked as cached
+    for first, second in zip(first_stream_results, second_stream_results, strict=True):
+        if isinstance(first, CreateResult) and isinstance(second, CreateResult):
+            assert not first.cached  # First call should not be cached
+            assert second.cached  # Second call should be cached
+            assert first.content == second.content
+        elif isinstance(first, str) and isinstance(second, str):
+            assert first == second
+        else:
+            pytest.fail(f"Unexpected chunk types: {type(first)}, {type(second)}")
+
+
+@pytest.mark.asyncio
+async def test_cache_cross_compatibility_create_to_stream() -> None:
+    """Test that create() cache can be used by create_stream() call.
+    This tests the scenario where:
+    1. User calls create() - stores CreateResult
+    2. User calls create_stream() with same inputs - should get cache hit and yield the CreateResult
+    """
+    responses, prompts, system_prompt, _, cached_client = get_test_data()
+
+    # First call: create() - should cache a CreateResult
+    create_result = await cached_client.create([system_prompt, UserMessage(content=prompts[0], source="user")])
+    assert isinstance(create_result, CreateResult)
+    assert not create_result.cached
+    assert create_result.content == responses[0]
+
+    # Second call: create_stream() with same inputs - should hit the cache
+    stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        stream_results.append(copy.copy(chunk))
+
+    # Should yield exactly two items: the string content + the cached CreateResult
+    assert len(stream_results) == 2
+
+    # First item should be the string content
+    assert isinstance(stream_results[0], str)
+    assert stream_results[0] == responses[0]
+
+    # Second item should be the cached CreateResult
+    assert isinstance(stream_results[1], CreateResult)
+    assert stream_results[1].cached  # Should be marked as cached
+    assert stream_results[1].content == responses[0]
+
+    # Verify no additional API calls were made (cache hit)
+    initial_usage = cached_client.total_usage()
+
+    # Third call: create_stream() again - should still hit cache
+    stream_results_2: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        stream_results_2.append(copy.copy(chunk))
+
+    # Usage should be the same (no new API calls)
+    assert cached_client.total_usage().prompt_tokens == initial_usage.prompt_tokens
+    assert cached_client.total_usage().completion_tokens == initial_usage.completion_tokens
+
+
+@pytest.mark.asyncio
+async def test_cache_cross_compatibility_stream_to_create() -> None:
+    """Test that create_stream() cache can be used by create() call.
+    This tests the scenario where:
+    1. User calls create_stream() - stores List[Union[str, CreateResult]]
+    2. User calls create() with same inputs - should get cache hit and return the final CreateResult
+    """
+    _, prompts, system_prompt, _, cached_client = get_test_data()
+
+    # First call: create_stream() - should cache a List[Union[str, CreateResult]]
+    first_stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        first_stream_results.append(copy.copy(chunk))
+
+    # Verify we got streaming results
+    assert len(first_stream_results) > 0
+    final_create_result = None
+    for item in first_stream_results:
+        if isinstance(item, CreateResult):
+            final_create_result = item
+            break
+
+    assert final_create_result is not None
+    assert not final_create_result.cached  # First call should not be cached
+
+    # Second call: create() with same inputs - should hit the streaming cache
+    create_result = await cached_client.create([system_prompt, UserMessage(content=prompts[0], source="user")])
+
+    assert isinstance(create_result, CreateResult)
+    assert create_result.cached  # Should be marked as cached
+    assert create_result.content == final_create_result.content
+
+    # Verify no additional API calls were made (cache hit)
+    initial_usage = cached_client.total_usage()
+
+    # Third call: create() again - should still hit cache
+    create_result_2 = await cached_client.create([system_prompt, UserMessage(content=prompts[0], source="user")])
+
+    # Usage should be the same (no new API calls)
+    assert cached_client.total_usage().prompt_tokens == initial_usage.prompt_tokens
+    assert cached_client.total_usage().completion_tokens == initial_usage.completion_tokens
+    assert create_result_2.cached
+
+
+@pytest.mark.asyncio
+async def test_cache_cross_compatibility_mixed_sequence() -> None:
+    """Test mixed sequence of create() and create_stream() calls with caching.
+    This tests a realistic scenario with multiple interleaved calls:
+    create() → create_stream() → create() → create_stream()
+    """
+    responses, prompts, system_prompt, _, cached_client = get_test_data(num_messages=4)
+
+    # Call 1: create() with prompt[0] - should make API call
+    result1 = await cached_client.create([system_prompt, UserMessage(content=prompts[0], source="user")])
+    assert not result1.cached
+    assert result1.content == responses[0]
+    usage_after_1 = copy.copy(cached_client.total_usage())
+
+    # Call 2: create_stream() with prompt[0] - should hit cache from call 1
+    stream1_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        stream1_results.append(chunk)
+
+    assert len(stream1_results) == 2  # Should yield string content + cached CreateResult
+    assert isinstance(stream1_results[0], str)  # First item: string content
+    assert stream1_results[0] == responses[0]
+    assert isinstance(stream1_results[1], CreateResult)  # Second item: cached CreateResult
+    assert stream1_results[1].cached
+    usage_after_2 = copy.copy(cached_client.total_usage())
+    # No new API call should have been made
+    assert usage_after_2.prompt_tokens == usage_after_1.prompt_tokens
+
+    # Call 3: create_stream() with prompt[1] - should make new API call
+    stream2_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[1], source="user")]):
+        stream2_results.append(copy.copy(chunk))
+
+    # Should have made a new API call
+    usage_after_3 = copy.copy(cached_client.total_usage())
+    assert usage_after_3.prompt_tokens > usage_after_2.prompt_tokens
+
+    # Find the final CreateResult
+    final_result = None
+    for item in stream2_results:
+        if isinstance(item, CreateResult):
+            final_result = item
+            break
+    assert final_result is not None
+    assert not final_result.cached
+
+    # Call 4: create() with prompt[1] - should hit cache from call 3
+    result4 = await cached_client.create([system_prompt, UserMessage(content=prompts[1], source="user")])
+    assert result4.cached
+    assert result4.content == final_result.content
+    usage_after_4 = copy.copy(cached_client.total_usage())
+    # No new API call should have been made
+    assert usage_after_4.prompt_tokens == usage_after_3.prompt_tokens
+
+
+@pytest.mark.asyncio
+async def test_cache_streaming_list_without_create_result() -> None:
+    """Test edge case where streaming cache contains only strings (no CreateResult).
+    This could happen if streaming was interrupted or in unusual scenarios.
+    The create() method should handle this gracefully by falling through to make a real API call.
+    """
+    responses, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a mock cache store that returns a list with only strings (no CreateResult)
+    string_only_list: List[Union[str, CreateResult]] = ["Hello", " world", "!"]
+    mock_store = MockCacheStore(return_value=string_only_list)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Call create() - should fall through and make API call since no CreateResult in cached list
+    result = await cached_client.create([system_prompt, UserMessage(content=prompts[0], source="user")])
+
+    assert isinstance(result, CreateResult)
+    assert not result.cached  # Should be from real API call, not cache
+    assert result.content == responses[0]
+
+
+@pytest.mark.asyncio
+async def test_create_stream_with_cached_non_streaming_result_string_content() -> None:
+    """
+    Test that when create_stream() finds a cached non-streaming result with string content,
+    it yields both the content string as a streaming chunk and then the CreateResult.
+    """
+    responses, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a CreateResult with string content (simulating a cached non-streaming result)
+    cached_create_result = CreateResult(
+        content=responses[0],  # This is a string
+        finish_reason="stop",
+        usage=RequestUsage(prompt_tokens=10, completion_tokens=20),
+        cached=False,  # Will be set to True when retrieved from cache
+    )
+
+    # Mock cache store that returns the non-streaming CreateResult
+    mock_store = MockCacheStore(return_value=cached_create_result)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Call create_stream() - should yield string content first, then CreateResult
+    stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        stream_results.append(copy.copy(chunk))
+
+    # Should have exactly 2 items: the string content, then the CreateResult
+    assert len(stream_results) == 2
+
+    # First item should be the string content
+    assert isinstance(stream_results[0], str)
+    assert stream_results[0] == responses[0]
+
+    # Second item should be the CreateResult
+    assert isinstance(stream_results[1], CreateResult)
+    assert stream_results[1].content == responses[0]
+    assert stream_results[1].finish_reason == "stop"
+    assert stream_results[1].cached is True  # Should be marked as cached
+
+
+@pytest.mark.asyncio
+async def test_create_stream_with_cached_non_streaming_result_empty_content() -> None:
+    """
+    Test that when create_stream() finds a cached non-streaming result with empty string content,
+    it only yields the CreateResult (no separate string chunk).
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a CreateResult with empty string content
+    cached_create_result = CreateResult(
+        content="",  # Empty string
+        finish_reason="stop",
+        usage=RequestUsage(prompt_tokens=10, completion_tokens=0),
+        cached=False,
+    )
+
+    # Mock cache store that returns the non-streaming CreateResult
+    mock_store = MockCacheStore(return_value=cached_create_result)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Call create_stream() - should yield only the CreateResult (no string chunk)
+    stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        stream_results.append(copy.copy(chunk))
+
+    # Should have exactly 1 item: just the CreateResult
+    assert len(stream_results) == 1
+
+    # Only item should be the CreateResult
+    assert isinstance(stream_results[0], CreateResult)
+    assert stream_results[0].content == ""
+    assert stream_results[0].finish_reason == "stop"
+    assert stream_results[0].cached is True
+
+
+@pytest.mark.asyncio
+async def test_create_stream_with_cached_non_streaming_result_non_string_content() -> None:
+    """
+    Test that when create_stream() finds a cached non-streaming result with non-string content,
+    it only yields the CreateResult (no separate string chunk).
+    """
+    _, prompts, system_prompt, replay_client, _ = get_test_data()
+
+    # Create a CreateResult with non-string content (e.g., list of function calls)
+    cached_create_result = CreateResult(
+        content=[
+            FunctionCall(id="call_123", name="test_func", arguments='{"param": "value"}')
+        ],  # List of FunctionCall objects
+        finish_reason="function_calls",  # Valid finish reason for function calls
+        usage=RequestUsage(prompt_tokens=10, completion_tokens=15),
+        cached=False,
+    )
+
+    # Mock cache store that returns the non-streaming CreateResult
+    mock_store = MockCacheStore(return_value=cached_create_result)
+    cached_client = ChatCompletionCache(replay_client, mock_store)
+
+    # Call create_stream() - should yield only the CreateResult (no string chunk)
+    stream_results: List[Union[str, CreateResult]] = []
+    async for chunk in cached_client.create_stream([system_prompt, UserMessage(content=prompts[0], source="user")]):
+        stream_results.append(copy.copy(chunk))
+
+    # Should have exactly 1 item: just the CreateResult
+    assert len(stream_results) == 1
+
+    # Only item should be the CreateResult
+    assert isinstance(stream_results[0], CreateResult)
+    expected_function_call = FunctionCall(id="call_123", name="test_func", arguments='{"param": "value"}')
+    assert stream_results[0].content == [expected_function_call]
+    assert stream_results[0].finish_reason == "function_calls"
+    assert stream_results[0].cached is True

python/uv.lock

@@ -469,7 +469,7 @@ wheels = [
 
 [[package]]
 name = "autogen-agentchat"
-version = "0.7.2"
+version = "0.7.4"
 source = { editable = "packages/autogen-agentchat" }
 dependencies = [
     { name = "autogen-core" },
@@ -480,7 +480,7 @@ requires-dist = [{ name = "autogen-core", editable = "packages/autogen-core" }]
 
 [[package]]
 name = "autogen-core"
-version = "0.7.2"
+version = "0.7.4"
 source = { editable = "packages/autogen-core" }
 dependencies = [
     { name = "jsonref" },
@@ -573,7 +573,7 @@ dev = [
 
 [[package]]
 name = "autogen-ext"
-version = "0.7.2"
+version = "0.7.4"
 source = { editable = "packages/autogen-ext" }
 dependencies = [
     { name = "autogen-core" },