Repository: stanfordnlp/dspy (python)

Issue:
[Bug] Google Models do not stream chunks using StreamListener
### What happened?

Streaming individual chunks doesn't work for several if not all Google models including Flash 2.0, 2.5, and Gemma 3. I also couldn't get Grok 3 mini to do it either, but that may be because it was a thinking model? Not sure.

I debugged the problem and the root cause is because of the way chunks are tokenized and transmitted.

Here's how they look like for Google models. Assume each line is a chunk:

```
[
[ ## sum
mary ## ]]\nOnce upon a
```

So you'll see that it doesn't cleanly start with `[[` nor does it end cleanly with `]]\n`.

I tried a variety of other models and most of them work with the current implementation: Mistral Small 3.1 24b, Llama 4 Scout, and Llama 3.3.

### Steps to reproduce

Use any Google model in openrouter

```python
dspy.LM(model="openrouter/google/gemma-3-12b-it", cache=False)
```

Set up streaming with a `StreamListener`.

```python
streaming_summary = streamify(
        Summary,
        stream_listeners=[StreamListener(signature_field_name="summary")]
    )
```

```python
stream = streaming_summary(article=article)
        async for chunk in stream:
            if isinstance(chunk, StreamResponse):
                output += chunk.chunk
                print(chunk.chunk, end="", flush=True)
```

#### Full Example

```python
import os
import dspy
from dotenv import load_dotenv
import asyncio
from dspy.streaming import streamify, StreamListener, StreamResponse

# Load environment variables from .env file
load_dotenv()

# Configure DSPy to use OpenRouter
openrouter_api_key = os.getenv("OPENROUTER_API_KEY")
if openrouter_api_key:
    # Use OpenRouter if API key is available
    lm = dspy.LM(model="openrouter/google/gemma-3-12b-it", cache=False) # doesnt work
else:
    # Fallback to local model
    print("OpenRouter API key not found. Using local model.")
    lm = dspy.LM('ollama_chat/gemma3:12b-it-qat', api_base='http://localhost:11434', api_key='')

# Set the language model
dspy.settings.configure(lm=lm, adapter=dspy.JSONAdapter())

# --- Article Summarizer Implementation ---

class ArticleSummarizerSignature(dspy.Signature):
    """
    You are an expert summarizer. Your task is to create a concise, informative summary of the provided article.
    
    The summary should:
    1. Capture the main points and key information
    2. Maintain the original meaning and intent
    3. Be significantly shorter than the original text
    4. Be written in clear, professional language
    5. Include the most important facts, findings, or arguments
    
    Article to summarize:
    {article_text}
    
    ---
    Summary:
    """
    article_text: str = dspy.InputField(desc="The full text of the article to be summarized.")
    summary: str = dspy.OutputField(desc="A one sentence summary of the article that captures the main points.")

class ArticleSummarizer(dspy.Module):
    def __init__(self):
        super().__init__()
        self.summarizer = dspy.Predict(ArticleSummarizerSignature)
    
    def forward(self, article: str):
        prediction = self.summarizer(
            article_text=article
        )
        return prediction


if __name__ == "__main__":
    summarizer = ArticleSummarizer()
    # Stream only the 'summary' output field
    streaming_summarizer = streamify(
        summarizer,
        stream_listeners=[StreamListener(signature_field_name="summary")],
        include_final_prediction_in_output_stream=False
    )

    async def main():
        print("\nArticle Summarizer (Streaming)\n----------------------")
        
        article = """
        Jack Clark, cofounder of Anthropic, predicts the rise of “manager nerds” as AI agents increasingly take over tasks traditionally done by humans. Speaking on the “Conversations With Tyler” podcast, Clark explained that future managers will orchestrate fleets of AI agents rather than direct people, potentially making small teams extremely powerful. He noted startups are already operating effectively with fewer employees due to AI coding agents. Industry leaders, including Meta CEO Mark Zuckerberg and Y Combinator CEO Garry Tan, echoed this sentiment, highlighting how AI enables lean, highly skilled teams to reach significant revenue milestones. Krieger, Instagram cofounder and Anthropic’s chief product officer, suggested developers will shift from writing code to managing and reviewing AI-generated content. However, experts caution that overreliance on AI carries risks, such as hallucinations from large language models and difficulties scaling or debugging AI-generated code. Anthropic positions itself as a testbed for adapting to AI-driven workplace changes, emphasizing AI’s role in collaboration rather than replacement.
        """
        
        if not article.strip():
            print("No article provided. Exiting.")
            return
            
        print("\nStreaming generated summary...\n")
        output = ""
        stream = streaming_summarizer(article=article)
        async for chunk in stream:
            if isinstance(chunk, StreamResponse):
                output += chunk.chunk
                print(chunk.chunk, end="", flush=True)
            elif isinstance(chunk, dspy.Prediction):
                print(chunk.summary, end="", flush=True)
        print("\n\n[Streaming complete]")

    asyncio.run(main())

```

### DSPy version

2.6.23

Pull request description:
Support models that don't split stream chunks in tokens
Resolve #8211 

Gemini models don't split stream chunks into 1-2 tokens per chunk, but instead have lots of tokens per chunk. We are reworking the listeners' logic to support this use case.

Side note - I do understand the rationale of buffering tokens for safety checks. but why not streaming back the original tokens for consistency with other models? This also creates a burden for app developer who wants to render the stream on UI. 

Patch:
diff --git a/dspy/streaming/streaming_listener.py b/dspy/streaming/streaming_listener.py
index c95086b0..8d9ad068 100644
--- a/dspy/streaming/streaming_listener.py
+++ b/dspy/streaming/streaming_listener.py
@@ -36,12 +36,18 @@ class StreamListener:
         self.stream_end = False
         self.cache_hit = False
 
-        self.json_adapter_start_identifier = f'{{"{self.signature_field_name}":"'
-        self.json_adapter_end_identifier = re.compile(r"\w*\",\w*")
+        self.json_adapter_start_identifier = f'"{self.signature_field_name}":'
+        self.json_adapter_end_identifier = re.compile(r"\w*\"(,|\s*})")
 
         self.chat_adapter_start_identifier = f"[[ ## {self.signature_field_name} ## ]]"
         self.chat_adapter_end_identifier = re.compile(r"\[\[ ## (\w+) ## \]\]")
 
+    def _buffered_message_end_with_start_identifier(self, concat_message: str, start_identifier: str) -> str:
+        for i in range(len(concat_message)):
+            if start_identifier.startswith(concat_message[len(concat_message) - i - 1 :]):
+                return True
+        return False
+
     def receive(self, chunk: ModelResponseStream):
         if isinstance(settings.adapter, JSONAdapter):
             start_identifier = self.json_adapter_start_identifier
@@ -52,7 +58,7 @@ class StreamListener:
             start_identifier = self.chat_adapter_start_identifier
             end_identifier = self.chat_adapter_end_identifier
 
-            start_indicator = "[["
+            start_indicator = "["
         else:
             raise ValueError(
                 f"Unsupported adapter for streaming: {settings.adapter}, please use either ChatAdapter or "
@@ -71,13 +77,18 @@ class StreamListener:
 
         if chunk_message and start_identifier in chunk_message:
             # If the cache is hit, the chunk_message could be the full response. When it happens we can
-            # directly end the stream listening.
-            self.cache_hit = True
-            self.stream_start = True
-            self.stream_end = True
-            return
+            # directly end the stream listening. In some models like gemini, each stream chunk can be multiple
+            # tokens, so it's posible that response only has one chunk, we also fall back to this logic.
+            message_after_start_identifier = chunk_message[
+                chunk_message.find(start_identifier) + len(start_identifier) :
+            ]
+            if re.search(end_identifier, message_after_start_identifier):
+                self.cache_hit = True
+                self.stream_start = True
+                self.stream_end = True
+                return
 
-        if len(self.field_start_queue) == 0 and start_indicator in chunk_message:
+        if len(self.field_start_queue) == 0 and not self.stream_start and start_indicator in chunk_message:
             # We look for the pattern of start_identifier, i.e., "[[ ## {self.signature_field_name} ## ]]" for
             # ChatAdapter to identify the start of the stream of our target field. Once the start_indicator, i.e., "[["
             # for ChatAdapter, is found, we start checking the next tokens
@@ -89,18 +100,28 @@ class StreamListener:
             # tokens no longer match our expected identifier.
             self.field_start_queue.append(chunk_message)
             concat_message = "".join(self.field_start_queue).strip()
-            start_token_index = concat_message.find(start_indicator)
-            concat_message = concat_message[start_token_index:]
-            if start_identifier == concat_message:
+
+            if start_identifier in concat_message:
                 # We have a full identifier, we can start the stream.
                 self.stream_start = True
-            elif start_identifier.startswith(concat_message):
-                # The concanated tokens still match our expected identifier, we keep listening.
+                self.field_start_queue = []
+                # Keep the part after the start_identifier from the concat_message, we need to write it to the buffer.
+                value_start_index = concat_message.find(start_identifier) + len(start_identifier)
+                chunk_message = concat_message[value_start_index:].lstrip()
+                if isinstance(settings.adapter, JSONAdapter) and chunk_message.startswith('"'):
+                    # For JSONAdapter, we need to remove the leading ". We cannot do this with the start_identifier
+                    # because there could be a few splitters between ':' and '"', e.g., '"name": "value"'.
+                    chunk_message = chunk_message[1:]
+
+            elif self._buffered_message_end_with_start_identifier(concat_message, start_identifier):
+                # If the buffered message ends with part of the start_identifier, we can start the stream.
                 return
             else:
                 # Doesn't match the expected identifier, reset the queue.
                 self.field_start_queue = []
-        elif self.stream_start:
+                return
+
+        if self.stream_start:
             # The stream is started, we keep returning the token until we see the start of the next field.
             token = None
             self.field_end_queue.put(chunk_message)
@@ -130,7 +151,11 @@ class StreamListener:
         last_tokens = "".join(self.field_end_queue.queue)
         self.field_end_queue = Queue()
         if isinstance(settings.adapter, JSONAdapter):
-            boundary_index = last_tokens.find('",')
+            match = re.search(r'",|"\s*}', last_tokens)
+            if match:
+                boundary_index = match.start()
+            else:
+                boundary_index = len(last_tokens)
             return last_tokens[:boundary_index]
         elif isinstance(settings.adapter, ChatAdapter) or settings.adapter is None:
             boundary_index = last_tokens.find("[[")


Test patch:
diff --git a/tests/streaming/test_streaming.py b/tests/streaming/test_streaming.py
index 9363f23c..23dead5e 100644
--- a/tests/streaming/test_streaming.py
+++ b/tests/streaming/test_streaming.py
@@ -4,6 +4,9 @@ import pytest
 
 import dspy
 from dspy.streaming import StatusMessage, StatusMessageProvider, streaming_response
+from unittest import mock
+from unittest.mock import AsyncMock
+from litellm.types.utils import ModelResponseStream, StreamingChoices, Delta
 
 
 @pytest.mark.anyio
@@ -287,3 +290,333 @@ def test_sync_status_streaming():
     assert len(status_messages) == 2
     assert status_messages[0].message == "Calling tool generate_question..."
     assert status_messages[1].message == "Tool calling finished! Querying the LLM with tool calling results..."
+
+
+@pytest.mark.anyio
+async def test_stream_listener_returns_correct_chunk_chat_adapter():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.predict1 = dspy.Predict("question->answer")
+            self.predict2 = dspy.Predict("question, answer->judgement")
+
+        def forward(self, question, **kwargs):
+            answer = self.predict1(question=question, **kwargs).answer
+            judgement = self.predict2(question=question, answer=answer, **kwargs)
+            return judgement
+
+    async def gpt_4o_mini_stream_1(*args, **kwargs):
+        # Recorded streaming from openai/gpt-4o-mini
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="[["))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" answer"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ]]\n\n"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="To"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" get"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" to"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" the"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" other"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" side"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" of"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" the"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" dinner"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" plate"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="!"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="\n\n"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="[[ ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" completed"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ]]"))])
+
+    async def gpt_4o_mini_stream_2():
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="[[ ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" judgement"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ]]\n\n"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="The"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" answer"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" is"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" humorous"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" and"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" plays"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" on"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" the"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" classic"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" joke"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" format"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="."))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="\n\n"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="[[ ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" completed"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ##"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" ]]"))])
+
+    dspy.settings.configure(lm=dspy.LM("openai/gpt-4o-mini", cache=False))
+
+    stream_generators = [gpt_4o_mini_stream_1, gpt_4o_mini_stream_2]
+
+    async def completion_side_effect(*args, **kwargs):
+        return stream_generators.pop(0)()  # return new async generator instance
+
+    with mock.patch("litellm.acompletion", side_effect=completion_side_effect) as mock_completion:
+        program = dspy.streamify(
+            MyProgram(),
+            stream_listeners=[
+                dspy.streaming.StreamListener(signature_field_name="answer"),
+                dspy.streaming.StreamListener(signature_field_name="judgement"),
+            ],
+        )
+        output = program(question="why did a chicken cross the kitchen?")
+        all_chunks = []
+        async for value in output:
+            if isinstance(value, dspy.streaming.StreamResponse):
+                all_chunks.append(value)
+
+        assert all_chunks[0].predict_name == "predict1"
+        assert all_chunks[0].signature_field_name == "answer"
+
+        assert all_chunks[0].chunk == "To"
+        assert all_chunks[1].chunk == " get"
+        assert all_chunks[2].chunk == " to"
+        assert all_chunks[3].chunk == " the"
+        assert all_chunks[4].chunk == " other"
+        assert all_chunks[5].chunk == " side of the dinner plate!"
+
+        # Start processing the second listened field.
+        assert all_chunks[6].predict_name == "predict2"
+        assert all_chunks[6].signature_field_name == "judgement"
+        assert all_chunks[6].chunk == "The"
+        assert all_chunks[7].chunk == " answer"
+        assert all_chunks[8].chunk == " is"
+        assert all_chunks[9].chunk == " humorous"
+        assert all_chunks[10].chunk == " and"
+        assert all_chunks[11].chunk == " plays"
+
+
+@pytest.mark.anyio
+async def test_stream_listener_returns_correct_chunk_json_adapter():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.predict1 = dspy.Predict("question->answer")
+            self.predict2 = dspy.Predict("question,answer->judgement")
+
+        def forward(self, question, **kwargs):
+            answer = self.predict1(question=question, **kwargs).answer
+            judgement = self.predict2(question=question, answer=answer, **kwargs)
+            return judgement
+
+    dspy.settings.configure(lm=dspy.LM("openai/gpt-4o-mini", cache=False), adapter=dspy.JSONAdapter())
+
+    async def gpt_4o_mini_stream_1(*args, **kwargs):
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content='{"'))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="answer"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content='":'))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="To"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" get"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" to"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" the"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" other"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" side"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" of"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" the"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" frying"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" pan"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content='!"'))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="}\n"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="None"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="None"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="None"))])
+
+    async def gpt_4o_mini_stream_2(*args, **kwargs):
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content='{"'))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="jud"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="gement"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content='":"'))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="The"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" answer"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" is"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" humorous"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" and"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" plays"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" on"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" the"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" very"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" funny"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" and"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" classic"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" joke"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content=" format"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content='."'))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="}"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="None"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="None"))])
+        yield ModelResponseStream(model="gpt-4o-mini", choices=[StreamingChoices(delta=Delta(content="None"))])
+
+    with mock.patch(
+        "litellm.acompletion", new_callable=AsyncMock, side_effect=[gpt_4o_mini_stream_1(), gpt_4o_mini_stream_2()]
+    ) as mock_completion:
+        program = dspy.streamify(
+            MyProgram(),
+            stream_listeners=[
+                dspy.streaming.StreamListener(signature_field_name="answer"),
+                dspy.streaming.StreamListener(signature_field_name="judgement"),
+            ],
+        )
+        output = program(question="why did a chicken cross the kitchen?")
+        all_chunks = []
+        async for value in output:
+            if isinstance(value, dspy.streaming.StreamResponse):
+                all_chunks.append(value)
+
+        assert all_chunks[0].predict_name == "predict1"
+        assert all_chunks[0].signature_field_name == "answer"
+
+        assert all_chunks[0].chunk == "To"
+        assert all_chunks[1].chunk == " get to the other side of the frying pan!"
+
+        # Start processing the second listened field.
+        assert all_chunks[2].predict_name == "predict2"
+        assert all_chunks[2].signature_field_name == "judgement"
+        assert all_chunks[2].chunk == "The"
+        assert all_chunks[3].chunk == " answer"
+        assert all_chunks[4].chunk == " is"
+        assert all_chunks[5].chunk == " humorous"
+        assert all_chunks[6].chunk == " and"
+        assert all_chunks[7].chunk == " plays on the very funny and classic joke format."
+
+
+@pytest.mark.anyio
+async def test_stream_listener_returns_correct_chunk_chat_adapter_untokenized_stream():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.predict1 = dspy.Predict("question->answer")
+            self.predict2 = dspy.Predict("question,answer->judgement")
+
+        def forward(self, question, **kwargs):
+            answer = self.predict1(question=question, **kwargs).answer
+            judgement = self.predict2(question=question, answer=answer, **kwargs)
+            return judgement
+
+    dspy.settings.configure(lm=dspy.LM("openai/gpt-4o-mini", cache=False), adapter=dspy.ChatAdapter())
+
+    async def gemini_stream_1(*args, **kwargs):
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content="[[ ##"))])
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content=" answer ## ]]"))])
+        yield ModelResponseStream(
+            model="gemini", choices=[StreamingChoices(delta=Delta(content="To get to the other side."))]
+        )
+        yield ModelResponseStream(
+            model="gemini", choices=[StreamingChoices(delta=Delta(content="\n\n[[ ## completed ## ]]"))]
+        )
+
+    async def gemini_stream_2(*args, **kwargs):
+        yield ModelResponseStream(
+            model="gemini", choices=[StreamingChoices(delta=Delta(content="[[ ## judgement ## ]]\n\n"))]
+        )
+        yield ModelResponseStream(
+            model="gemini",
+            choices=[
+                StreamingChoices(
+                    delta=Delta(
+                        content=(
+                            "The answer provides the standard punchline for this classic joke format, adapted to the "
+                            "specific location mentioned in the question. It is the expected and appropriate response."
+                        )
+                    )
+                )
+            ],
+        )
+        yield ModelResponseStream(
+            model="gemini",
+            choices=[StreamingChoices(delta=Delta(content="\n\n[[ ## completed ## ]]"))],
+        )
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content="}\n"))])
+
+    with mock.patch("litellm.acompletion", new_callable=AsyncMock, side_effect=[gemini_stream_1(), gemini_stream_2()]):
+        program = dspy.streamify(
+            MyProgram(),
+            stream_listeners=[
+                dspy.streaming.StreamListener(signature_field_name="answer"),
+                dspy.streaming.StreamListener(signature_field_name="judgement"),
+            ],
+        )
+        output = program(question="why did a chicken cross the kitchen?")
+        all_chunks = []
+        async for value in output:
+            if isinstance(value, dspy.streaming.StreamResponse):
+                all_chunks.append(value)
+
+        assert all_chunks[0].predict_name == "predict1"
+        assert all_chunks[0].signature_field_name == "answer"
+        assert all_chunks[0].chunk == "To get to the other side."
+
+        assert all_chunks[1].predict_name == "predict2"
+        assert all_chunks[1].signature_field_name == "judgement"
+        assert all_chunks[1].chunk == (
+            "The answer provides the standard punchline for this classic joke format, adapted to the specific location "
+            "mentioned in the question. It is the expected and appropriate response."
+        )
+
+
+@pytest.mark.anyio
+async def test_stream_listener_returns_correct_chunk_json_adapter_untokenized_stream():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.predict1 = dspy.Predict("question->answer")
+            self.predict2 = dspy.Predict("question,answer->judgement")
+
+        def forward(self, question, **kwargs):
+            answer = self.predict1(question=question, **kwargs).answer
+            judgement = self.predict2(question=question, answer=answer, **kwargs)
+            return judgement
+
+    dspy.settings.configure(lm=dspy.LM("openai/gpt-4o-mini", cache=False), adapter=dspy.JSONAdapter())
+
+    async def gemini_stream_1(*args, **kwargs):
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content="{\n"))])
+        yield ModelResponseStream(
+            model="gemini", choices=[StreamingChoices(delta=Delta(content='  "answer": "To get to'))]
+        )
+        yield ModelResponseStream(
+            model="gemini", choices=[StreamingChoices(delta=Delta(content=' the other side... of the cutting board!"'))]
+        )
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content="}\n"))])
+
+    async def gemini_stream_2(*args, **kwargs):
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content="{\n"))])
+        yield ModelResponseStream(
+            model="gemini", choices=[StreamingChoices(delta=Delta(content='  "judgement": "The'))]
+        )
+        yield ModelResponseStream(
+            model="gemini",
+            choices=[
+                StreamingChoices(
+                    delta=Delta(
+                        content=' answer provides a humorous and relevant punchline to the classic joke setup."'
+                    )
+                )
+            ],
+        )
+        yield ModelResponseStream(model="gemini", choices=[StreamingChoices(delta=Delta(content="}\n"))])
+
+    with mock.patch("litellm.acompletion", new_callable=AsyncMock, side_effect=[gemini_stream_1(), gemini_stream_2()]):
+        program = dspy.streamify(
+            MyProgram(),
+            stream_listeners=[
+                dspy.streaming.StreamListener(signature_field_name="answer"),
+                dspy.streaming.StreamListener(signature_field_name="judgement"),
+            ],
+        )
+        output = program(question="why did a chicken cross the kitchen?")
+        all_chunks = []
+        async for value in output:
+            if isinstance(value, dspy.streaming.StreamResponse):
+                all_chunks.append(value)
+
+        assert all_chunks[0].predict_name == "predict1"
+        assert all_chunks[0].signature_field_name == "answer"
+        assert all_chunks[0].chunk == "To get to the other side... of the cutting board!"
+
+        assert all_chunks[1].predict_name == "predict2"
+        assert all_chunks[1].signature_field_name == "judgement"
+        assert all_chunks[1].chunk == "The answer provides a humorous and relevant punchline to the classic joke setup."