Repository: stanfordnlp/dspy

Issue:
[Bug] Token Tracking Issue - Bedrock with Streamified Predict
### What happened?

Token tracking is not working as expected when using a streamified module using Amazon Bedrock. `completion_tokens` is populated as expected, but all other fields have a value of 0. The one exception is `total_tokens`, but that just has the same value as `completion_tokens` since all other fields are 0.  

I have spent a lot of time trying to track down the issue on this. I've walked through the LiteLLM code in the debugger and I can see that all usage fields are being set, but at some point in the chunk iterator it seems that they are getting stripped. The LiteLLM code is way too dense and indirect for me to follow exactly where that is happening.  

The interesting thing is that while walking through LiteLLM code all of the usage fields looked right, total_tokens was the sum of prompt and completion tokens, which tells me that it's getting re-computed at some point.  

### To Recap
Token usage is getting set properly in LiteLLM, but at some point between LiteLLM and DSPy all usage is set to 0 except for `completion_tokens`

I have tried updating to the newest version of LiteLLM and that didn't make a difference.
LiteLLM version tested: `1.71.1`, `1.72.6.post1`

### Steps to reproduce

I have tried many different parameters for the streamify call. The example below uses `async_streaming=False` because the results are the same either way and the example is less cumbersome without async streaming.  

Here is a minimal example that will reproduce the error:
```python
import os
import dspy

# If no AWS profile is configured you can use access key and secret key instead
os.environ["AWS_PROFILE"] = "YOUR_AWS_PROFILE"
# os.environ["AWS_ACCESS_KEY_ID"] = "YOUR_AWS_ACCESS_KEY_ID"
# os.environ["AWS_SECRET_ACCESS_KEY"] = "YOUR_AWS_SECRET_ACCESS_KEY"
# os.environ["AWS_SESSION_TOKEN"] = "YOUR_AWS_SESSION_TOKEN" <-- Optional
os.environ["AWS_REGION"] = "us-east-1"


class QAPredictor(dspy.Signature):
    """Answer questions with helpful, accurate responses."""

    question = dspy.InputField()
    answer = dspy.OutputField()


predictor = dspy.Predict(QAPredictor)
lm = dspy.LM("bedrock/us.amazon.nova-lite-v1:0")
dspy.configure(lm=lm, track_usage=True)
streaming_predictor = dspy.streamify(
    predictor,
    async_streaming=False,  # <--- Doesn't matter if this is True or False fot token tracking
    include_final_prediction_in_output_stream=True,
)


def generate_response(question):
    for chunk in streaming_predictor(question=question):
        if isinstance(chunk, dspy.Prediction):
            usage = chunk.get_lm_usage()
            print(usage)


if __name__ == "__main__":
    generate_response(question="What is the capital of Maine?")
```

### DSPy version

3.0.0b1, also tested with 2.6.27

Discussion:
@chenmoneygithub: @BTripp1986 Thanks for reporting the issue! If you don't turn on streaming, i.e., directly call `predictor ` without `dspy.streamify`, are you seeing the same issue?  @BTripp1986: @chenmoneygithub Yes that is correct. My application has several DSPy modules, most of which don't use streaming(query reformulation, query routing, etc.) and they report the correct token usage. I should have included that in my original report.@chenmoneygithub: I did some debugging, but I cannot produce this issue with OpenAI, Anthropic, Gemini or Databricks, so I am suspecting it's an issue with how bedrock sends back the streaming chunks. 

For the providers I checked, they all have a last chunk containing the usage, if bedrock breaks this assumption then it will fail. I don't have an AWS account for bedrock testing, but the easiest way to debug is turning on MLflow tracing: https://dspy.ai/tutorials/observability/ so that you can view if the LM response contains the usage information.@BTripp1986: Ok, I have mlfow tracing enabled in my app, but I don't see any traces with usage information. I've spent a lot of time on this already and exhausted everything I could think of to look at. Attached are some screenshots of LM calls in mlflow, I'm assuming the first one is where we would see the uage info.
I have inspected the final chunk that comes from the stream(before the final Prediction return) and it is just an end_reason with no token tracking info.

![Image](https://github.com/user-attachments/assets/ef0502d5-977b-4e86-a97f-01bd59d1013c)

![Image](https://github.com/user-attachments/assets/058b2787-a736-495a-95f0-318ab42994aa)

![Image](https://github.com/user-attachments/assets/74e20918-4b72-4ece-8118-e38a5fb3369d)

Let me know if there is anything else I can do or anything else I can investigate.@chenmoneygithub: > I have inspected the final chunk that comes from the stream(before the final Prediction return) and it is just an end_reason with no token tracking info.

This pretty much explains the failure reason. DSPy usage tracking doesn't implement new logic like counting tokens, but just utilizes the usage information from LM response. With that said, if the response doesn't have usage information then DSPy won't be able to tell the usage. 

The easiest way to confirm that it's due to LM is calling your LM directly with litellm streaming: https://docs.litellm.ai/docs/completion/stream, then check if your response stream doesn't contain usage information. @lambda-science: Same issue over here, was working kind of fine in the past@BTripp1986: > > I have inspected the final chunk that comes from the stream(before the final Prediction return) and it is just an end_reason with no token tracking info.
> 
> This pretty much explains the failure reason. DSPy usage tracking doesn't implement new logic like counting tokens, but just utilizes the usage information from LM response. With that said, if the response doesn't have usage information then DSPy won't be able to tell the usage.
> 
> The easiest way to confirm that it's due to LM is calling your LM directly with litellm streaming: https://docs.litellm.ai/docs/completion/stream, then check if your response stream doesn't contain usage information.

I have tested with LiteLLM using async streaming and I can confirm that I am in-fact getting a full usage dict. Here is the code:
```python
from litellm import acompletion
import asyncio, os, traceback

os.environ["AWS_PROFILE"] = "my-profile"
os.environ["AWS_REGION"] = "us-east-1"


async def completion_call():
    try:
        response = await acompletion(
            model="bedrock/us.amazon.nova-lite-v1:0",
            messages=[
                {
                    "content": "Hello, how are you? Tell me in a single word.",
                    "role": "user",
                }
            ],
            stream=True,
            stream_options={"include_usage": True},
        )
        print(f"response: {response}")
        async for chunk in response:
            print(chunk)
    except:
        print(f"error occurred: {traceback.format_exc()}")
        pass


asyncio.run(completion_call())
```

When I run that I get the following output(added some newlines for readability between each print stmt):
```
response: <litellm.litellm_core_utils.streaming_handler.CustomStreamWrapper object at 0x78be8afcacf0>


ModelResponseStream(id='chatcmpl-8c92e264-4b03-4b06-bd00-cdb55223d2a6', created=1750431309, model='us.amazon.nova-lite-v1:0', object='chat.completion.chunk', system_fingerprint=None, choices=[StreamingChoices(finish_reason=None, index=0, delta=Delta(provider_specific_fields=None, content='Good', role='assistant', function_call=None, tool_calls=None, audio=None), logprobs=None)], provider_specific_fields={}, stream_options={'include_usage': True}, citations=None)


ModelResponseStream(id='chatcmpl-8c92e264-4b03-4b06-bd00-cdb55223d2a6', created=1750431309, model='us.amazon.nova-lite-v1:0', object='chat.completion.chunk', system_fingerprint=None, choices=[StreamingChoices(finish_reason=None, index=0, delta=Delta(provider_specific_fields=None, content='.', role=None, function_call=None, tool_calls=None, audio=None), logprobs=None)], provider_specific_fields={}, stream_options={'include_usage': True}, citations=None)


ModelResponseStream(id='chatcmpl-8c92e264-4b03-4b06-bd00-cdb55223d2a6', created=1750431309, model='us.amazon.nova-lite-v1:0', object='chat.completion.chunk', system_fingerprint=None, choices=[StreamingChoices(finish_reason='stop', index=0, delta=Delta(provider_specific_fields=None, content=None, role=None, function_call=None, tool_calls=None, audio=None), logprobs=None)], provider_specific_fields={}, stream_options={'include_usage': True})


ModelResponseStream(id='chatcmpl-8c92e264-4b03-4b06-bd00-cdb55223d2a6', created=1750431309, model='us.amazon.nova-lite-v1:0', object='chat.completion.chunk', system_fingerprint=None, choices=[StreamingChoices(finish_reason=None, index=0, delta=Delta(provider_specific_fields=None, content=None, role=None, function_call=None, tool_calls=None, audio=None), logprobs=None)], provider_specific_fields=None, stream_options={'include_usage': True}, usage=Usage(completion_tokens=2, prompt_tokens=13, total_tokens=15, completion_tokens_details=CompletionTokensDetailsWrapper(accepted_prediction_tokens=None, audio_tokens=None, reasoning_tokens=0, rejected_prediction_tokens=None, text_tokens=None), prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=None, cached_tokens=0, text_tokens=None, image_tokens=None), cache_creation_input_tokens=0, cache_read_input_tokens=0))
```

As you can see the final chunk includes `completion_tokens`, `prompt_tokens`, and `total_tokens`.

I have also tested without streaming, and with the syncronous `call` method and any way I run this is will produce the correct token output.  

If there is anything else I can do let me know. This is a pretty big issue for us as this is for a client project and correct token tracking is vital.

Thanks for you help.@BTripp1986: @chenmoneygithub I've found the root of the problem. I will submit a PR.

Solution patch:
diff --git a/docs/docs/tutorials/email_extraction/index.md b/docs/docs/tutorials/email_extraction/index.md
new file mode 100644
index 00000000..5664b76f
--- /dev/null
+++ b/docs/docs/tutorials/email_extraction/index.md
@@ -0,0 +1,324 @@
+# Extracting Information from Emails with DSPy
+
+This tutorial demonstrates how to build an intelligent email processing system using DSPy. We'll create a system that can automatically extract key information from various types of emails, classify their intent, and structure the data for further processing.
+
+## What You'll Build
+
+By the end of this tutorial, you'll have a DSPy-powered email processing system that can:
+
+- **Classify email types** (order confirmation, support request, meeting invitation, etc.)
+- **Extract key entities** (dates, amounts, product names, contact info)
+- **Determine urgency levels** and required actions
+- **Structure extracted data** into consistent formats
+- **Handle multiple email formats** robustly
+
+## Prerequisites
+
+- Basic understanding of DSPy modules and signatures
+- Python 3.9+ installed
+- OpenAI API key (or access to another supported LLM)
+
+## Installation and Setup
+
+```bash
+pip install dspy
+```
+
+## Step 1: Define Our Data Structures
+
+First, let's define the types of information we want to extract from emails:
+
+```python
+import dspy
+from typing import List, Optional, Literal
+from datetime import datetime
+from pydantic import BaseModel
+from enum import Enum
+
+class EmailType(str, Enum):
+    ORDER_CONFIRMATION = "order_confirmation"
+    SUPPORT_REQUEST = "support_request"
+    MEETING_INVITATION = "meeting_invitation"
+    NEWSLETTER = "newsletter"
+    PROMOTIONAL = "promotional"
+    INVOICE = "invoice"
+    SHIPPING_NOTIFICATION = "shipping_notification"
+    OTHER = "other"
+
+class UrgencyLevel(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+class ExtractedEntity(BaseModel):
+    entity_type: str
+    value: str
+    confidence: float
+
+class EmailInsight(BaseModel):
+    email_type: EmailType
+    urgency: UrgencyLevel
+    summary: str
+    key_entities: List[ExtractedEntity]
+    action_required: bool
+    deadline: Optional[str] = None
+    amount: Optional[float] = None
+    sender_info: Optional[str] = None
+```
+
+## Step 2: Create DSPy Signatures
+
+Now let's define the signatures for our email processing pipeline:
+
+```python
+class ClassifyEmail(dspy.Signature):
+    """Classify the type and urgency of an email based on its content."""
+
+    email_subject: str = dspy.InputField(desc="The subject line of the email")
+    email_body: str = dspy.InputField(desc="The main content of the email")
+    sender: str = dspy.InputField(desc="Email sender information")
+
+    email_type: EmailType = dspy.OutputField(desc="The classified type of email")
+    urgency: UrgencyLevel = dspy.OutputField(desc="The urgency level of the email")
+    reasoning: str = dspy.OutputField(desc="Brief explanation of the classification")
+
+class ExtractEntities(dspy.Signature):
+    """Extract key entities and information from email content."""
+
+    email_content: str = dspy.InputField(desc="The full email content including subject and body")
+    email_type: EmailType = dspy.InputField(desc="The classified type of email")
+
+    key_entities: List[ExtractedEntity] = dspy.OutputField(desc="List of extracted entities with type, value, and confidence")
+    financial_amount: Optional[float] = dspy.OutputField(desc="Any monetary amounts found (e.g., '$99.99')")
+    important_dates: list[str] = dspy.OutputField(desc="List of important dates found in the email")
+    contact_info: list[str] = dspy.OutputField(desc="Relevant contact information extracted")
+
+class GenerateActionItems(dspy.Signature):
+    """Determine what actions are needed based on the email content and extracted information."""
+
+    email_type: EmailType = dspy.InputField()
+    urgency: UrgencyLevel = dspy.InputField()
+    email_summary: str = dspy.InputField(desc="Brief summary of the email content")
+    extracted_entities: List[ExtractedEntity] = dspy.InputField(desc="Key entities found in the email")
+
+    action_required: bool = dspy.OutputField(desc="Whether any action is required")
+    action_items: list[str] = dspy.OutputField(desc="List of specific actions needed")
+    deadline: Optional[str] = dspy.OutputField(desc="Deadline for action if applicable")
+    priority_score: int = dspy.OutputField(desc="Priority score from 1-10")
+
+class SummarizeEmail(dspy.Signature):
+    """Create a concise summary of the email content."""
+
+    email_subject: str = dspy.InputField()
+    email_body: str = dspy.InputField()
+    key_entities: List[ExtractedEntity] = dspy.InputField()
+
+    summary: str = dspy.OutputField(desc="A 2-3 sentence summary of the email's main points")
+```
+
+## Step 3: Build the Email Processing Module
+
+Now let's create our main email processing module:
+
+```python
+class EmailProcessor(dspy.Module):
+    """A comprehensive email processing system using DSPy."""
+
+    def __init__(self):
+        super().__init__()
+
+        # Initialize our processing components
+        self.classifier = dspy.ChainOfThought(ClassifyEmail)
+        self.entity_extractor = dspy.ChainOfThought(ExtractEntities)
+        self.action_generator = dspy.ChainOfThought(GenerateActionItems)
+        self.summarizer = dspy.ChainOfThought(SummarizeEmail)
+
+    def forward(self, email_subject: str, email_body: str, sender: str = ""):
+        """Process an email and extract structured information."""
+
+        # Step 1: Classify the email
+        classification = self.classifier(
+            email_subject=email_subject,
+            email_body=email_body,
+            sender=sender
+        )
+
+        # Step 2: Extract entities
+        full_content = f"Subject: {email_subject}\n\nFrom: {sender}\n\n{email_body}"
+        entities = self.entity_extractor(
+            email_content=full_content,
+            email_type=classification.email_type
+        )
+
+        # Step 3: Generate summary
+        summary = self.summarizer(
+            email_subject=email_subject,
+            email_body=email_body,
+            key_entities=entities.key_entities
+        )
+
+        # Step 4: Determine actions
+        actions = self.action_generator(
+            email_type=classification.email_type,
+            urgency=classification.urgency,
+            email_summary=summary.summary,
+            extracted_entities=entities.key_entities
+        )
+
+        # Step 5: Structure the results
+        return dspy.Prediction(
+            email_type=classification.email_type,
+            urgency=classification.urgency,
+            summary=summary.summary,
+            key_entities=entities.key_entities,
+            financial_amount=entities.financial_amount,
+            important_dates=entities.important_dates,
+            action_required=actions.action_required,
+            action_items=actions.action_items,
+            deadline=actions.deadline,
+            priority_score=actions.priority_score,
+            reasoning=classification.reasoning,
+            contact_info=entities.contact_info
+        )
+```
+
+## Step 4: Running the Email Processing System
+
+Let's create a simple function to test our email processing system:
+
+```python
+import os
+def run_email_processing_demo():
+    """Demonstration of the email processing system."""
+    
+    # Configure DSPy
+    lm = dspy.LM(model='openai/gpt-4o-mini')
+    dspy.configure(lm=lm)
+    os.environ["OPENAI_API_KEY"] = "<YOUR OPENAI KEY>"
+    
+    # Create our email processor
+    processor = EmailProcessor()
+    
+    # Sample emails for testing
+    sample_emails = [
+        {
+            "subject": "Order Confirmation #12345 - Your MacBook Pro is on the way!",
+            "body": """Dear John Smith,
+
+Thank you for your order! We're excited to confirm that your order #12345 has been processed.
+
+Order Details:
+- MacBook Pro 14-inch (Space Gray)
+- Order Total: $2,399.00
+- Estimated Delivery: December 15, 2024
+- Tracking Number: 1Z999AA1234567890
+
+If you have any questions, please contact our support team at support@techstore.com.
+
+Best regards,
+TechStore Team""",
+            "sender": "orders@techstore.com"
+        },
+        {
+            "subject": "URGENT: Server Outage - Immediate Action Required",
+            "body": """Hi DevOps Team,
+
+We're experiencing a critical server outage affecting our production environment.
+
+Impact: All users unable to access the platform
+Started: 2:30 PM EST
+
+Please join the emergency call immediately: +1-555-123-4567
+
+This is our highest priority.
+
+Thanks,
+Site Reliability Team""",
+            "sender": "alerts@company.com"
+        },
+        {
+            "subject": "Meeting Invitation: Q4 Planning Session",
+            "body": """Hello team,
+
+You're invited to our Q4 planning session.
+
+When: Friday, December 20, 2024 at 2:00 PM - 4:00 PM EST
+Where: Conference Room A
+
+Please confirm your attendance by December 18th.
+
+Best,
+Sarah Johnson""",
+            "sender": "sarah.johnson@company.com"
+        }
+    ]
+    
+    # Process each email and display results
+    print("🚀 Email Processing Demo")
+    print("=" * 50)
+    
+    for i, email in enumerate(sample_emails):
+        print(f"\n📧 EMAIL {i+1}: {email['subject'][:50]}...")
+        
+        # Process the email
+        result = processor(
+            email_subject=email["subject"],
+            email_body=email["body"],
+            sender=email["sender"]
+        )
+        
+        # Display key results
+        print(f"   📊 Type: {result.email_type}")
+        print(f"   🚨 Urgency: {result.urgency}")
+        print(f"   📝 Summary: {result.summary}")
+        
+        if result.financial_amount:
+            print(f"   💰 Amount: ${result.financial_amount:,.2f}")
+        
+        if result.action_required:
+            print(f"   ✅ Action Required: Yes")
+            if result.deadline:
+                print(f"   ⏰ Deadline: {result.deadline}")
+        else:
+            print(f"   ✅ Action Required: No")
+
+# Run the demo
+if __name__ == "__main__":
+    run_email_processing_demo()
+```
+
+## Expected Output
+```
+🚀 Email Processing Demo
+==================================================
+
+📧 EMAIL 1: Order Confirmation #12345 - Your MacBook Pro is on...
+   📊 Type: order_confirmation
+   🚨 Urgency: low
+   📝 Summary: The email confirms John Smith's order #12345 for a MacBook Pro 14-inch in Space Gray, totaling $2,399.00, with an estimated delivery date of December 15, 2024. It includes a tracking number and contact information for customer support.
+   💰 Amount: $2,399.00
+   ✅ Action Required: No
+
+📧 EMAIL 2: URGENT: Server Outage - Immediate Action Required...
+   📊 Type: other
+   🚨 Urgency: critical
+   📝 Summary: The Site Reliability Team has reported a critical server outage that began at 2:30 PM EST, preventing all users from accessing the platform. They have requested the DevOps Team to join an emergency call immediately to address the issue.
+   ✅ Action Required: Yes
+   ⏰ Deadline: Immediately
+
+📧 EMAIL 3: Meeting Invitation: Q4 Planning Session...
+   📊 Type: meeting_invitation
+   🚨 Urgency: medium
+   📝 Summary: Sarah Johnson has invited the team to a Q4 planning session on December 20, 2024, from 2:00 PM to 4:00 PM EST in Conference Room A. Attendees are asked to confirm their participation by December 18th.
+   ✅ Action Required: Yes
+   ⏰ Deadline: December 18th
+```
+
+## Next Steps
+
+- **Add more email types** and refine classification (newsletter, promotional, etc.)
+- **Add integration** with email providers (Gmail API, Outlook, IMAP)
+- **Experiment with different LLMs** and optimization strategies
+- **Add multilingual support** for international email processing
+- **Optimization** for increasing the performance of your program
diff --git a/docs/docs/tutorials/mem0_react_agent/index.md b/docs/docs/tutorials/mem0_react_agent/index.md
new file mode 100644
index 00000000..f4d146e4
--- /dev/null
+++ b/docs/docs/tutorials/mem0_react_agent/index.md
@@ -0,0 +1,291 @@
+# Building Memory-Enabled Agents with DSPy ReAct and Mem0
+
+This tutorial demonstrates how to build intelligent conversational agents that can remember information across interactions using DSPy's ReAct framework combined with [Mem0](https://docs.mem0.ai/)'s memory capabilities. You'll learn to create agents that can store, retrieve, and use contextual information to provide personalized and coherent responses.
+
+## What You'll Build
+
+By the end of this tutorial, you'll have a memory-enabled agent that can:
+
+- **Remember user preferences** and past conversations
+- **Store and retrieve factual information** about users and topics
+- **Use memory to inform decisions** and provide personalized responses
+- **Handle complex multi-turn conversations** with context awareness
+- **Manage different types of memories** (facts, preferences, experiences)
+
+## Prerequisites
+
+- Basic understanding of DSPy and ReAct agents
+- Python 3.9+ installed
+- API keys for your preferred LLM provider
+
+## Installation and Setup
+
+```bash
+pip install dspy mem0ai
+```
+
+## Step 1: Understanding Mem0 Integration
+
+Mem0 provides a memory layer that can store, search, and retrieve memories for AI agents. Let's start by understanding how to integrate it with DSPy:
+
+```python
+import dspy
+from mem0 import Memory
+import os
+from typing import List, Dict, Any, Optional
+from datetime import datetime
+
+# Configure environment
+os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
+
+# Initialize Mem0 memory system
+config = {
+    "llm": {
+        "provider": "openai",
+        "config": {
+            "model": "gpt-4o-mini",
+            "temperature": 0.1
+        }
+    },
+    "embedder": {
+        "provider": "openai",
+        "config": {
+            "model": "text-embedding-3-small"
+        }
+    }
+}
+```
+
+## Step 2: Create Memory-Aware Tools
+
+Let's create tools that can interact with the memory system:
+
+```python
+import datetime
+
+class MemoryTools:
+    """Tools for interacting with the Mem0 memory system."""
+
+    def __init__(self, memory: Memory):
+        self.memory = memory
+
+    def store_memory(self, content: str, user_id: str = "default_user") -> str:
+        """Store information in memory."""
+        try:
+            self.memory.add(content, user_id=user_id)
+            return f"Stored memory: {content}"
+        except Exception as e:
+            return f"Error storing memory: {str(e)}"
+
+    def search_memories(self, query: str, user_id: str = "default_user", limit: int = 5) -> str:
+        """Search for relevant memories."""
+        try:
+            results = self.memory.search(query, user_id=user_id, limit=limit)
+            if not results:
+                return "No relevant memories found."
+
+            memory_text = "Relevant memories found:\n"
+            for i, result in enumerate(results["results"]):
+                memory_text += f"{i}. {result['memory']}\n"
+            return memory_text
+        except Exception as e:
+            return f"Error searching memories: {str(e)}"
+
+    def get_all_memories(self, user_id: str = "default_user") -> str:
+        """Get all memories for a user."""
+        try:
+            results = self.memory.get_all(user_id=user_id)
+            if not results:
+                return "No memories found for this user."
+
+            memory_text = "All memories for user:\n"
+            for i, result in enumerate(results["results"]):
+                memory_text += f"{i}. {result['memory']}\n"
+            return memory_text
+        except Exception as e:
+            return f"Error retrieving memories: {str(e)}"
+
+    def update_memory(self, memory_id: str, new_content: str) -> str:
+        """Update an existing memory."""
+        try:
+            self.memory.update(memory_id, new_content)
+            return f"Updated memory with new content: {new_content}"
+        except Exception as e:
+            return f"Error updating memory: {str(e)}"
+
+    def delete_memory(self, memory_id: str) -> str:
+        """Delete a specific memory."""
+        try:
+            self.memory.delete(memory_id)
+            return "Memory deleted successfully."
+        except Exception as e:
+            return f"Error deleting memory: {str(e)}"
+
+def get_current_time() -> str:
+    """Get the current date and time."""
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+```
+
+## Step 3: Build the Memory-Enhanced ReAct Agent
+
+Now let's create our main ReAct agent that can use memory:
+
+```python
+class MemoryQA(dspy.Signature):
+    """
+    You're a helpful assistant and have access to memory method.
+    Whenever you answer a user's input, remember to store the information in memory
+    so that you can use it later.
+    """
+    user_input: str = dspy.InputField()
+    response: str = dspy.OutputField()
+
+class MemoryReActAgent(dspy.Module):
+    """A ReAct agent enhanced with Mem0 memory capabilities."""
+
+    def __init__(self, memory: Memory):
+        super().__init__()
+        self.memory_tools = MemoryTools(memory)
+
+        # Create tools list for ReAct
+        self.tools = [
+            self.memory_tools.store_memory,
+            self.memory_tools.search_memories,
+            self.memory_tools.get_all_memories,
+            get_current_time,
+            self.set_reminder,
+            self.get_preferences,
+            self.update_preferences,
+        ]
+
+        # Initialize ReAct with our tools
+        self.react = dspy.ReAct(
+            signature=MemoryQA,
+            tools=self.tools,
+            max_iters=6
+        )
+
+    def forward(self, user_input: str):
+        """Process user input with memory-aware reasoning."""
+        
+        return self.react(user_input=user_input)
+
+    def set_reminder(self, reminder_text: str, date_time: str = None, user_id: str = "default_user") -> str:
+        """Set a reminder for the user."""
+        reminder = f"Reminder set for {date_time}: {reminder_text}"
+        return self.memory_tools.store_memory(
+            f"REMINDER: {reminder}", 
+            user_id=user_id
+        )
+
+    def get_preferences(self, category: str = "general", user_id: str = "default_user") -> str:
+        """Get user preferences for a specific category."""
+        query = f"user preferences {category}"
+        return self.memory_tools.search_memories(
+            query=query,
+            user_id=user_id
+        )
+
+    def update_preferences(self, category: str, preference: str, user_id: str = "default_user") -> str:
+        """Update user preferences."""
+        preference_text = f"User preference for {category}: {preference}"
+        return self.memory_tools.store_memory(
+            preference_text,
+            user_id=user_id
+        )
+```
+
+## Step 4: Running the Memory-Enhanced Agent
+
+Let's create a simple interface to interact with our memory-enabled agent:
+
+```python
+import time
+def run_memory_agent_demo():
+    """Demonstration of memory-enhanced ReAct agent."""
+
+    # Configure DSPy
+    lm = dspy.LM(model='openai/gpt-4o-mini')
+    dspy.configure(lm=lm)
+
+    # Initialize memory system
+    memory = Memory.from_config(config)
+
+    # Create our agent
+    agent = MemoryReActAgent(memory)
+
+    # Sample conversation demonstrating memory capabilities
+    print("🧠 Memory-Enhanced ReAct Agent Demo")
+    print("=" * 50)
+
+    conversations = [
+        "Hi, I'm Alice and I love Italian food, especially pasta carbonara.",
+        "I'm Alice. I prefer to exercise in the morning around 7 AM.",
+        "I'm Alice. What do you remember about my food preferences?",
+        "I'm Alice. Set a reminder for me to go grocery shopping tomorrow.",
+        "I'm Alice. What are my exercise preferences?",
+        "I'm Alice. I also enjoy hiking on weekends.",
+        "I'm Alice. What do you know about me so far?"
+    ]
+
+    for i, user_input in enumerate(conversations, 1):
+        print(f"\n📝 User: {user_input}")
+
+        try:
+            response = agent(user_input=user_input)
+            print(f"🤖 Agent: {response.response}")
+            time.sleep(1)
+
+        except Exception as e:
+            print(f"❌ Error: {e}")
+
+# Run the demonstration
+if __name__ == "__main__":
+    run_memory_agent_demo()
+```
+
+## Expected Output
+```
+🧠 Memory-Enhanced ReAct Agent Demo
+==================================================
+
+📝 User: Hi, I'm Alice and I love Italian food, especially pasta carbonara.
+
+🤖 Agent: Hi Alice! It's great to meet you! Pasta carbonara is a delicious choice. Do you have a favorite place to get it, or do you enjoy making it at home?
+
+📝 User: I'm Alice. I prefer to exercise in the morning around 7 AM.
+
+🤖 Agent: Nice to meet you, Alice! I've noted that you prefer to exercise in the morning around 7 AM. If you have any questions or need tips related to your morning workouts, feel free to ask!
+
+📝 User: I'm Alice. What do you remember about my food preferences?
+
+🤖 Agent: I remember that you love Italian food, and your favorite dish is pasta carbonara!
+
+📝 User: I'm Alice. Set a reminder for me to go grocery shopping tomorrow.
+
+🤖 Agent: I have set a reminder for you, Alice, to go grocery shopping tomorrow.
+
+📝 User: I'm Alice. What are my exercise preferences?
+
+🤖 Agent: Your exercise preference is to work out in the morning around 7 AM.
+
+📝 User: I'm Alice. I also enjoy hiking on weekends.
+
+🤖 Agent: Nice to meet you, Alice! It's great to hear that you enjoy hiking on weekends. If you need any recommendations for trails or tips for your hikes, feel free to ask!
+
+📝 User: I'm Alice. What do you know about me so far?
+
+🤖 Agent: So far, I know that you enjoy hiking on weekends, prefer to exercise in the morning around 7 AM, and love Italian food, especially pasta carbonara. Additionally, you have a reminder set for grocery shopping on October 7, 2023.
+```
+
+## Next Steps
+
+- **Implement memory persistence** with databases (PostgreSQL, MongoDB)
+- **Add memory categorization** and tagging for better organization
+- **Create memory expiration policies** for data management
+- **Build multi-user memory isolation** for production applications
+- **Add memory analytics** and insights
+- **Integrate with vector databases** for enhanced semantic search
+- **Implement memory compression** for long-term storage efficiency
+
+This tutorial demonstrates how DSPy's ReAct framework can be enhanced with Mem0's memory capabilities to create intelligent, context-aware agents that can learn and remember information across interactions, making them more useful for real-world applications.
\ No newline at end of file
diff --git a/docs/docs/tutorials/real_world_examples/index.md b/docs/docs/tutorials/real_world_examples/index.md
new file mode 100644
index 00000000..12057305
--- /dev/null
+++ b/docs/docs/tutorials/real_world_examples/index.md
@@ -0,0 +1,25 @@
+# Real-World Examples
+
+This section demonstrates practical applications of DSPy across different domains and use cases. Each tutorial shows how to build production-ready AI systems using DSPy's modular programming approach.
+
+## Featured Examples
+
+### 📄 [Generating llms.txt](../llms_txt_generation/index.md)
+Learn how to create AI-powered documentation generators that analyze codebases and produce structured, LLM-friendly documentation following the llms.txt standard.
+
+**Key Concepts:** Repository analysis, meta-programming, documentation generation
+
+### 📧 [Email Information Extraction](../email_extraction/index.md)
+Build intelligent email processing systems that classify messages, extract entities, and identify action items using DSPy's structured prediction capabilities.
+
+**Key Concepts:** Information extraction, classification, text processing
+
+### 🧠 [Memory-Enabled ReAct Agents with Mem0](../mem0_react_agent/index.md)
+Create conversational agents with persistent memory using DSPy ReAct and Mem0 integration for context-aware interactions across sessions.
+
+**Key Concepts:** Memory systems, conversational AI, agent persistence
+
+### 💰 [Financial Analysis with Yahoo Finance](../yahoo_finance_react/index.md)
+Develop financial analysis agents that fetch real-time market data, analyze news sentiment, and provide investment insights using LangChain tool integration.
+
+**Key Concepts:** Tool integration, financial data, real-time analysis
diff --git a/docs/docs/tutorials/yahoo_finance_react/index.md b/docs/docs/tutorials/yahoo_finance_react/index.md
new file mode 100644
index 00000000..de55f60a
--- /dev/null
+++ b/docs/docs/tutorials/yahoo_finance_react/index.md
@@ -0,0 +1,162 @@
+# Financial Analysis with DSPy ReAct and Yahoo Finance News
+
+This tutorial shows how to build a financial analysis agent using DSPy ReAct with [LangChain's Yahoo Finance News tool](https://python.langchain.com/docs/integrations/tools/yahoo_finance_news/) for real-time market analysis.
+
+## What You'll Build
+
+A financial agent that fetches news, analyzes sentiment, and provides investment insights.
+
+## Setup
+
+```bash
+pip install dspy langchain langchain-community yfinance
+```
+
+## Step 1: Convert LangChain Tool to DSPy
+
+```python
+import dspy
+from langchain_community.tools.yahoo_finance_news import YahooFinanceNewsTool
+from dspy.adapters.types.tool import Tool
+import json
+import yfinance as yf
+
+# Configure DSPy
+lm = dspy.LM(model='openai/gpt-4o-mini')
+dspy.configure(lm=lm)
+
+# Convert LangChain Yahoo Finance tool to DSPy
+yahoo_finance_tool = YahooFinanceNewsTool()
+finance_news_tool = Tool.from_langchain(yahoo_finance_tool)
+```
+
+## Step 2: Create Supporting Financial Tools
+
+```python
+def get_stock_price(ticker: str) -> str:
+    """Get current stock price and basic info."""
+    try:
+        stock = yf.Ticker(ticker)
+        info = stock.info
+        hist = stock.history(period="1d")
+        
+        if hist.empty:
+            return f"Could not retrieve data for {ticker}"
+        
+        current_price = hist['Close'].iloc[-1]
+        prev_close = info.get('previousClose', current_price)
+        change_pct = ((current_price - prev_close) / prev_close * 100) if prev_close else 0
+        
+        result = {
+            "ticker": ticker,
+            "price": round(current_price, 2),
+            "change_percent": round(change_pct, 2),
+            "company": info.get('longName', ticker)
+        }
+        
+        return json.dumps(result)
+    except Exception as e:
+        return f"Error: {str(e)}"
+
+def compare_stocks(tickers: str) -> str:
+    """Compare multiple stocks (comma-separated)."""
+    try:
+        ticker_list = [t.strip().upper() for t in tickers.split(',')]
+        comparison = []
+        
+        for ticker in ticker_list:
+            stock = yf.Ticker(ticker)
+            info = stock.info
+            hist = stock.history(period="1d")
+            
+            if not hist.empty:
+                current_price = hist['Close'].iloc[-1]
+                prev_close = info.get('previousClose', current_price)
+                change_pct = ((current_price - prev_close) / prev_close * 100) if prev_close else 0
+                
+                comparison.append({
+                    "ticker": ticker,
+                    "price": round(current_price, 2),
+                    "change_percent": round(change_pct, 2)
+                })
+        
+        return json.dumps(comparison)
+    except Exception as e:
+        return f"Error: {str(e)}"
+```
+
+## Step 3: Build the Financial ReAct Agent
+
+```python
+class FinancialAnalysisAgent(dspy.Module):
+    """ReAct agent for financial analysis using Yahoo Finance data."""
+    
+    def __init__(self):
+        super().__init__()
+        
+        # Combine all tools
+        self.tools = [
+            finance_news_tool,  # LangChain Yahoo Finance News
+            get_stock_price,
+            compare_stocks
+        ]
+        
+        # Initialize ReAct
+        self.react = dspy.ReAct(
+            signature="financial_query -> analysis_response",
+            tools=self.tools,
+            max_iters=6
+        )
+    
+    def forward(self, financial_query: str):
+        return self.react(financial_query=financial_query)
+```
+
+## Step 4: Run Financial Analysis
+
+```python
+def run_financial_demo():
+    """Demo of the financial analysis agent."""
+    
+    # Initialize agent
+    agent = FinancialAnalysisAgent()
+    
+    # Example queries
+    queries = [
+        "What's the latest news about Apple (AAPL) and how might it affect the stock price?",
+        "Compare AAPL, GOOGL, and MSFT performance",
+        "Find recent Tesla news and analyze sentiment"
+    ]
+    
+    for query in queries:
+        print(f"Query: {query}")
+        response = agent(financial_query=query)
+        print(f"Analysis: {response.analysis_response}")
+        print("-" * 50)
+
+# Run the demo
+if __name__ == "__main__":
+    run_financial_demo()
+```
+
+## Example Output
+
+When you run the agent with a query like "What's the latest news about Apple?", it will:
+
+1. Use the Yahoo Finance News tool to fetch recent Apple news
+2. Get current stock price data
+3. Analyze the information and provide insights
+
+**Sample Response:**
+```
+Analysis: Given the current price of Apple (AAPL) at $196.58 and the slight increase of 0.48%, it appears that the stock is performing steadily in the market. However, the inability to access the latest news means that any significant developments that could influence investor sentiment and stock price are unknown. Investors should keep an eye on upcoming announcements or market trends that could impact Apple's performance, especially in comparison to other tech stocks like Microsoft (MSFT), which is also showing a positive trend.
+```
+
+## Key Benefits
+
+- **Tool Integration**: Seamlessly combine LangChain tools with DSPy ReAct
+- **Real-time Data**: Access current market data and news
+- **Extensible**: Easy to add more financial analysis tools
+- **Intelligent Reasoning**: ReAct framework provides step-by-step analysis
+
+This tutorial shows how DSPy's ReAct framework works with LangChain's financial tools to create intelligent market analysis agents.
diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml
index fcb2128b..bdb94661 100644
--- a/docs/mkdocs.yml
+++ b/docs/mkdocs.yml
@@ -60,7 +60,11 @@ nav:
             - Streaming: tutorials/streaming/index.md
             - Async: tutorials/async/index.md
         - Real-World Examples:
+            - Overview: tutorials/real_world_examples/index.md
             - Generating llms.txt: tutorials/llms_txt_generation/index.md
+            - Memory-Enabled ReAct Agents: tutorials/mem0_react_agent/index.md
+            - Financial Analysis with Yahoo Finance: tutorials/yahoo_finance_react/index.md
+            - Email Information Extraction: tutorials/email_extraction/index.md
     - DSPy in Production: production/index.md
     - Community:
         - Community Resources: community/community-resources.md
@@ -186,15 +190,6 @@ plugins:
     - search:
         lang: en
         separator: '[\s\-\.]+'
-        min_search_length: 2
-        prebuild_index: true
-        indexing: 'full'
-        properties:
-            - title
-            - tags
-            - content
-            - sections
-        search_index_only: false
     - mkdocstrings:
         handlers:
             python:
diff --git a/dspy/adapters/chat_adapter.py b/dspy/adapters/chat_adapter.py
index 97bb173c..7e785997 100644
--- a/dspy/adapters/chat_adapter.py
+++ b/dspy/adapters/chat_adapter.py
@@ -50,6 +50,26 @@ class ChatAdapter(Adapter):
                 raise e
             return JSONAdapter()(lm, lm_kwargs, signature, demos, inputs)
 
+    async def acall(
+        self,
+        lm: LM,
+        lm_kwargs: dict[str, Any],
+        signature: Type[Signature],
+        demos: list[dict[str, Any]],
+        inputs: dict[str, Any],
+    ) -> list[dict[str, Any]]:
+        try:
+            return await super().acall(lm, lm_kwargs, signature, demos, inputs)
+        except Exception as e:
+            # fallback to JSONAdapter
+            from dspy.adapters.json_adapter import JSONAdapter
+
+            if isinstance(e, ContextWindowExceededError) or isinstance(self, JSONAdapter):
+                # On context window exceeded error or already using JSONAdapter, we don't want to retry with a different
+                # adapter.
+                raise e
+            return await JSONAdapter().acall(lm, lm_kwargs, signature, demos, inputs)
+
     def format_field_description(self, signature: Type[Signature]) -> str:
         return (
             f"Your input fields are:\n{get_field_description_string(signature.input_fields)}\n"
diff --git a/dspy/adapters/json_adapter.py b/dspy/adapters/json_adapter.py
index e43561b0..5e599d29 100644
--- a/dspy/adapters/json_adapter.py
+++ b/dspy/adapters/json_adapter.py
@@ -22,6 +22,11 @@ from dspy.utils.exceptions import AdapterParseError
 
 logger = logging.getLogger(__name__)
 
+ERROR_MESSAGE_ON_JSON_ADAPTER_FAILURE = (
+    "Both structured output format and JSON mode failed. Please choose a model that supports "
+    "`response_format` argument. Original error: {}"
+)
+
 
 def _has_open_ended_mapping(signature: SignatureMeta) -> bool:
     """
@@ -37,6 +42,18 @@ def _has_open_ended_mapping(signature: SignatureMeta) -> bool:
 
 
 class JSONAdapter(ChatAdapter):
+    def _json_adapter_call_common(self, lm, lm_kwargs, signature, demos, inputs, call_fn):
+        """Common call logic to be used for both sync and async calls."""
+        provider = lm.model.split("/", 1)[0] or "openai"
+        params = litellm.get_supported_openai_params(model=lm.model, custom_llm_provider=provider)
+
+        if not params or "response_format" not in params:
+            return call_fn(lm, lm_kwargs, signature, demos, inputs)
+
+        if _has_open_ended_mapping(signature):
+            lm_kwargs["response_format"] = {"type": "json_object"}
+            return call_fn(lm, lm_kwargs, signature, demos, inputs)
+
     def __call__(
         self,
         lm: LM,
@@ -45,37 +62,49 @@ class JSONAdapter(ChatAdapter):
         demos: list[dict[str, Any]],
         inputs: dict[str, Any],
     ) -> list[dict[str, Any]]:
-        provider = lm.model.split("/", 1)[0] or "openai"
-        params = litellm.get_supported_openai_params(model=lm.model, custom_llm_provider=provider)
+        result = self._json_adapter_call_common(lm, lm_kwargs, signature, demos, inputs, super().__call__)
+        if result:
+            return result
 
-        # If response_format is not supported, use basic call
-        if not params or "response_format" not in params:
+        try:
+            structured_output_model = _get_structured_outputs_response_format(signature)
+            lm_kwargs["response_format"] = structured_output_model
             return super().__call__(lm, lm_kwargs, signature, demos, inputs)
+        except Exception:
+            logger.warning("Failed to use structured output format, falling back to JSON mode.")
+            try:
+                lm_kwargs["response_format"] = {"type": "json_object"}
+                return super().__call__(lm, lm_kwargs, signature, demos, inputs)
+            except AdapterParseError as e:
+                raise e
+            except Exception as e:
+                raise RuntimeError(ERROR_MESSAGE_ON_JSON_ADAPTER_FAILURE.format(e)) from e
 
-        # Check early for open-ended mapping types before trying structured outputs.
-        if _has_open_ended_mapping(signature):
-            lm_kwargs["response_format"] = {"type": "json_object"}
-            return super().__call__(lm, lm_kwargs, signature, demos, inputs)
+    async def acall(
+        self,
+        lm: LM,
+        lm_kwargs: dict[str, Any],
+        signature: Type[Signature],
+        demos: list[dict[str, Any]],
+        inputs: dict[str, Any],
+    ) -> list[dict[str, Any]]:
+        result = self._json_adapter_call_common(lm, lm_kwargs, signature, demos, inputs, super().acall)
+        if result:
+            return await result
 
-        # Try structured output first, fall back to basic JSON if it fails.
         try:
             structured_output_model = _get_structured_outputs_response_format(signature)
             lm_kwargs["response_format"] = structured_output_model
-            return super().__call__(lm, lm_kwargs, signature, demos, inputs)
+            return await super().acall(lm, lm_kwargs, signature, demos, inputs)
         except Exception:
             logger.warning("Failed to use structured output format, falling back to JSON mode.")
             try:
                 lm_kwargs["response_format"] = {"type": "json_object"}
-                return super().__call__(lm, lm_kwargs, signature, demos, inputs)
+                return await super().acall(lm, lm_kwargs, signature, demos, inputs)
             except AdapterParseError as e:
-                # On AdapterParseError, we raise the original error.
                 raise e
             except Exception as e:
-                # On any other error, we raise a RuntimeError with the original error message.
-                raise RuntimeError(
-                    "Both structured output format and JSON mode failed. Please choose a model that supports "
-                    f"`response_format` argument. Original error: {e}"
-                ) from e
+                raise RuntimeError(ERROR_MESSAGE_ON_JSON_ADAPTER_FAILURE.format(e)) from e
 
     def _call_preprocess(
         self,
diff --git a/dspy/clients/lm.py b/dspy/clients/lm.py
index e9dfc901..228d78b6 100644
--- a/dspy/clients/lm.py
+++ b/dspy/clients/lm.py
@@ -269,6 +269,9 @@ def _get_stream_completion_fn(
     stream = cast(MemoryObjectSendStream, stream)
     caller_predict_id = id(caller_predict) if caller_predict else None
 
+    if dspy.settings.track_usage:
+        request["stream_options"] = {"include_usage": True}
+
     async def stream_completion(request: Dict[str, Any], cache_kwargs: Dict[str, Any]):
         response = await litellm.acompletion(
             cache=cache_kwargs,
diff --git a/dspy/dsp/utils/settings.py b/dspy/dsp/utils/settings.py
index 4dbe73aa..eea41bcb 100644
--- a/dspy/dsp/utils/settings.py
+++ b/dspy/dsp/utils/settings.py
@@ -31,6 +31,7 @@ DEFAULT_CONFIG = dotdict(
 # Global base configuration and owner tracking
 main_thread_config = copy.deepcopy(DEFAULT_CONFIG)
 config_owner_thread_id = None
+config_owner_async_task = None
 
 # Global lock for settings configuration
 global_lock = threading.Lock()
@@ -105,13 +106,12 @@ class Settings:
         return self.copy()
 
     def _ensure_configure_allowed(self):
-        global main_thread_config, config_owner_thread_id
+        global main_thread_config, config_owner_thread_id, config_owner_async_task
         current_thread_id = threading.get_ident()
 
         if config_owner_thread_id is None:
-            # First `configure` call is always allowed.
+            # First `configure` call assigns the owner thread id.
             config_owner_thread_id = current_thread_id
-            return
 
         if config_owner_thread_id != current_thread_id:
             # Disallow a second `configure` calls from other threads.
@@ -130,6 +130,11 @@ class Settings:
         if not is_async_task:
             return
 
+        if config_owner_async_task is None:
+            # First `configure` call assigns the owner async task.
+            config_owner_async_task = asyncio.current_task()
+            return
+
         # We are in an async task. Now check for IPython and allow calling `configure` from IPython.
         in_ipython = False
         try:
@@ -142,10 +147,10 @@ class Settings:
             # If `IPython` is not installed or `get_ipython` failed, we are not in an IPython environment.
             in_ipython = False
 
-        if not in_ipython:
+        if not in_ipython and config_owner_async_task != asyncio.current_task():
             raise RuntimeError(
-                "dspy.settings.configure(...) cannot be called a second time from an async task. Use "
-                "`dspy.context(...)` instead."
+                "dspy.settings.configure(...) can only be called from the same async task that called it first. Please "
+                "use `dspy.context(...)` in other async tasks instead."
             )
 
     def configure(self, **kwargs):
diff --git a/dspy/primitives/prediction.py b/dspy/primitives/prediction.py
index 670b816b..4f32fe9f 100644
--- a/dspy/primitives/prediction.py
+++ b/dspy/primitives/prediction.py
@@ -2,6 +2,19 @@ from dspy.primitives.example import Example
 
 
 class Prediction(Example):
+    """A prediction object that contains the output of a DSPy module.
+    
+    Prediction inherits from Example.
+    
+    To allow feedback-augmented scores, Prediction supports comparison operations
+    (<, >, <=, >=) for Predictions with a `score` field. The comparison operations
+    compare the 'score' values as floats. For equality comparison, Predictions are equal
+    if their underlying data stores are equal (inherited from Example).
+    
+    Arithmetic operations (+, /, etc.) are also supported for Predictions with a 'score'
+    field, operating on the score value.
+    """
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
diff --git a/dspy/streaming/messages.py b/dspy/streaming/messages.py
index 9a3cf74c..49023e8a 100644
--- a/dspy/streaming/messages.py
+++ b/dspy/streaming/messages.py
@@ -1,7 +1,10 @@
 import asyncio
+import concurrent.futures
 from dataclasses import dataclass
 from typing import Any, Dict, Optional
 
+from asyncer import syncify
+
 from dspy.dsp.utils.settings import settings
 from dspy.utils.callback import BaseCallback
 
@@ -21,36 +24,29 @@ class StatusMessage:
 
 
 def sync_send_to_stream(stream, message):
-    """Send message to stream in a sync context, regardless of whether the caller is async or not."""
-    # Try to get current event loop, create one if none exists
-    try:
-        loop = asyncio.get_event_loop()
-    except RuntimeError:
-        # "There is no current event loop in thread" error
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
+    """Send message to stream in a sync context, regardless of event loop state."""
 
-    # If we're in an async context
-    if loop.is_running():
-        # In an async context, we need to use an approach that doesn't block
-        # Create a new thread and run a new event loop there
-        import concurrent.futures
+    async def _send():
+        await stream.send(message)
 
-        def run_async_in_new_loop():
+    try:
+        asyncio.get_running_loop()
+
+        # If we're in an event loop, offload to a new thread with its own event loop
+        def run_in_new_loop():
             new_loop = asyncio.new_event_loop()
             asyncio.set_event_loop(new_loop)
             try:
-                return new_loop.run_until_complete(stream.send(message))
+                return new_loop.run_until_complete(_send())
             finally:
                 new_loop.close()
 
-        # Run the function in a separate thread
         with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
-            future = executor.submit(run_async_in_new_loop)
-            return future.result()  # This shouldn't hang now
-    else:
-        # We're in a sync context, use run_until_complete
-        return loop.run_until_complete(stream.send(message))
+            future = executor.submit(run_in_new_loop)
+            return future.result()
+    except RuntimeError:
+        # Not in an event loop, safe to use a new event loop in this thread
+        return syncify(_send)()
 
 
 class StatusMessageProvider:


Test patch:
diff --git a/tests/adapters/test_chat_adapter.py b/tests/adapters/test_chat_adapter.py
index 969a8f17..e599da58 100644
--- a/tests/adapters/test_chat_adapter.py
+++ b/tests/adapters/test_chat_adapter.py
@@ -3,6 +3,7 @@ from unittest import mock
 
 import pydantic
 import pytest
+from litellm.utils import Choices, Message, ModelResponse
 
 import dspy
 
@@ -376,3 +377,48 @@ def test_chat_adapter_formats_conversation_history():
     assert messages[2]["content"] == "[[ ## answer ## ]]\nParis\n\n[[ ## completed ## ]]\n"
     assert messages[3]["content"] == "[[ ## question ## ]]\nWhat is the capital of Germany?"
     assert messages[4]["content"] == "[[ ## answer ## ]]\nBerlin\n\n[[ ## completed ## ]]\n"
+
+
+def test_chat_adapter_fallback_to_json_adapter_on_exception():
+    signature = dspy.make_signature("question->answer")
+    adapter = dspy.ChatAdapter()
+
+    with mock.patch("litellm.completion") as mock_completion:
+        # Mock returning a response compatible with JSONAdapter but not ChatAdapter
+        mock_completion.return_value = ModelResponse(
+            choices=[Choices(message=Message(content="{'answer': 'Paris'}"))],
+            model="openai/gpt-4o-mini",
+        )
+
+        lm = dspy.LM("openai/gpt-4o-mini", cache=False)
+
+        with mock.patch("dspy.adapters.json_adapter.JSONAdapter.__call__") as mock_json_adapter_call:
+            adapter(lm, {}, signature, [], {"question": "What is the capital of France?"})
+            mock_json_adapter_call.assert_called_once()
+
+        # The parse should succeed
+        result = adapter(lm, {}, signature, [], {"question": "What is the capital of France?"})
+        assert result == [{"answer": "Paris"}]
+
+
+@pytest.mark.asyncio
+async def test_chat_adapter_fallback_to_json_adapter_on_exception_async():
+    signature = dspy.make_signature("question->answer")
+    adapter = dspy.ChatAdapter()
+
+    with mock.patch("litellm.acompletion") as mock_completion:
+        # Mock returning a response compatible with JSONAdapter but not ChatAdapter
+        mock_completion.return_value = ModelResponse(
+            choices=[Choices(message=Message(content="{'answer': 'Paris'}"))],
+            model="openai/gpt-4o-mini",
+        )
+
+        lm = dspy.LM("openai/gpt-4o-mini", cache=False)
+
+        with mock.patch("dspy.adapters.json_adapter.JSONAdapter.acall") as mock_json_adapter_acall:
+            await adapter.acall(lm, {}, signature, [], {"question": "What is the capital of France?"})
+            mock_json_adapter_acall.assert_called_once()
+
+        # The parse should succeed
+        result = await adapter.acall(lm, {}, signature, [], {"question": "What is the capital of France?"})
+        assert result == [{"answer": "Paris"}]
diff --git a/tests/adapters/test_json_adapter.py b/tests/adapters/test_json_adapter.py
index 7db4a320..f688af43 100644
--- a/tests/adapters/test_json_adapter.py
+++ b/tests/adapters/test_json_adapter.py
@@ -5,6 +5,7 @@ import pytest
 from litellm.utils import Choices, Message, ModelResponse
 
 import dspy
+from dspy.adapters.json_adapter import ERROR_MESSAGE_ON_JSON_ADAPTER_FAILURE
 
 
 def test_json_adapter_passes_structured_output_when_supported_by_model():
@@ -22,7 +23,7 @@ def test_json_adapter_passes_structured_output_when_supported_by_model():
     program = dspy.Predict(TestSignature)
 
     # Configure DSPy to use an OpenAI LM that supports structured outputs
-    dspy.configure(lm=dspy.LM(model="openai/gpt4o"), adapter=dspy.JSONAdapter())
+    dspy.configure(lm=dspy.LM(model="openai/gpt-4o"), adapter=dspy.JSONAdapter())
     with mock.patch("litellm.completion") as mock_completion:
         program(input1="Test input")
 
@@ -43,14 +44,28 @@ def test_json_adapter_passes_structured_output_when_supported_by_model():
     assert issubclass(response_format, pydantic.BaseModel)
     assert response_format.model_fields.keys() == {"output1", "output2", "output3", "output4_unannotated"}
 
+
+def test_json_adapter_not_using_structured_outputs_when_not_supported_by_model():
+    class TestSignature(dspy.Signature):
+        input1: str = dspy.InputField()
+        output1: str = dspy.OutputField()
+        output2: bool = dspy.OutputField()
+
+    program = dspy.Predict(TestSignature)
+
     # Configure DSPy to use a model from a fake provider that doesn't support structured outputs
-    dspy.configure(lm=dspy.LM(model="fakeprovider/fakemodel"), adapter=dspy.JSONAdapter())
+    dspy.configure(lm=dspy.LM(model="fakeprovider/fakemodel", cache=False), adapter=dspy.JSONAdapter())
     with mock.patch("litellm.completion") as mock_completion:
+        mock_completion.return_value = ModelResponse(
+            choices=[Choices(message=Message(content=("{'output1': 'Test output', 'output2': True}")))],
+            model="openai/gpt-4o",
+        )
+
         program(input1="Test input")
 
     mock_completion.assert_called_once()
     _, call_kwargs = mock_completion.call_args
-    assert response_format not in call_kwargs
+    assert "response_format" not in call_kwargs
 
 
 def test_json_adapter_falls_back_when_structured_outputs_fails():
@@ -442,3 +457,186 @@ def test_json_adapter_formats_conversation_history():
     assert messages[2]["content"] == '{\n  "answer": "Paris"\n}'
     assert messages[3]["content"] == "[[ ## question ## ]]\nWhat is the capital of Germany?"
     assert messages[4]["content"] == '{\n  "answer": "Berlin"\n}'
+
+
+@pytest.mark.asyncio
+async def test_json_adapter_on_pydantic_model_async():
+    from litellm.utils import Choices, Message, ModelResponse
+
+    class User(pydantic.BaseModel):
+        id: int
+        name: str
+        email: str
+
+    class Answer(pydantic.BaseModel):
+        analysis: str
+        result: str
+
+    class TestSignature(dspy.Signature):
+        user: User = dspy.InputField(desc="The user who asks the question")
+        question: str = dspy.InputField(desc="Question the user asks")
+        answer: Answer = dspy.OutputField(desc="Answer to this question")
+
+    program = dspy.Predict(TestSignature)
+
+    with mock.patch("litellm.acompletion") as mock_completion:
+        mock_completion.return_value = ModelResponse(
+            choices=[
+                Choices(
+                    message=Message(
+                        content="{'answer': {'analysis': 'Paris is the captial of France', 'result': 'Paris'}}"
+                    )
+                )
+            ],
+            model="openai/gpt4o",
+        )
+
+        with dspy.context(lm=dspy.LM(model="openai/gpt4o", cache=False), adapter=dspy.JSONAdapter()):
+            result = await program.acall(
+                user={"id": 5, "name": "name_test", "email": "email_test"}, question="What is the capital of France?"
+            )
+
+        # Check that litellm.acompletion was called exactly once
+        mock_completion.assert_called_once()
+
+        _, call_kwargs = mock_completion.call_args
+        # Assert that there are exactly 2 messages (system + user)
+        assert len(call_kwargs["messages"]) == 2
+
+        assert call_kwargs["messages"][0]["role"] == "system"
+        content = call_kwargs["messages"][0]["content"]
+        assert content is not None
+
+        # Assert that system prompt includes correct input field descriptions
+        expected_input_fields = (
+            "1. `user` (User): The user who asks the question\n2. `question` (str): Question the user asks\n"
+        )
+        assert expected_input_fields in content
+
+        # Assert that system prompt includes correct output field description
+        expected_output_fields = "1. `answer` (Answer): Answer to this question\n"
+        assert expected_output_fields in content
+
+        # Assert that system prompt includes input formatting structure
+        expected_input_structure = "[[ ## user ## ]]\n{user}\n\n[[ ## question ## ]]\n{question}\n\n"
+        assert expected_input_structure in content
+
+        # Assert that system prompt includes output formatting structure
+        expected_output_structure = (
+            "Outputs will be a JSON object with the following fields.\n\n{\n  "
+            '"answer": "{answer}        # note: the value you produce must adhere to the JSON schema: '
+            '{\\"type\\": \\"object\\", \\"properties\\": {\\"analysis\\": {\\"type\\": \\"string\\", \\"title\\": '
+            '\\"Analysis\\"}, \\"result\\": {\\"type\\": \\"string\\", \\"title\\": \\"Result\\"}}, \\"required\\": '
+            '[\\"analysis\\", \\"result\\"], \\"title\\": \\"Answer\\"}"\n}'
+        )
+        assert expected_output_structure in content
+
+        assert call_kwargs["messages"][1]["role"] == "user"
+        user_message_content = call_kwargs["messages"][1]["content"]
+        assert user_message_content is not None
+
+        # Assert that the user input data is formatted correctly
+        expected_input_data = (
+            '[[ ## user ## ]]\n{"id": 5, "name": "name_test", "email": "email_test"}\n\n[[ ## question ## ]]\n'
+            "What is the capital of France?\n\n"
+        )
+        assert expected_input_data in user_message_content
+
+        # Assert that the adapter output has expected fields and values
+        assert result.answer.analysis == "Paris is the captial of France"
+        assert result.answer.result == "Paris"
+
+
+def test_json_adapter_fallback_to_json_mode_on_structured_output_failure():
+    class TestSignature(dspy.Signature):
+        question: str = dspy.InputField()
+        answer: str = dspy.OutputField(desc="String output field")
+
+    dspy.configure(lm=dspy.LM(model="openai/gpt-4o-mini", cache=False), adapter=dspy.JSONAdapter())
+    program = dspy.Predict(TestSignature)
+
+    with mock.patch("litellm.completion") as mock_completion:
+        # First call raises error to simulate structured output failure, second call returns a valid response
+        mock_completion.side_effect = [
+            RuntimeError("Structured output failed!"),
+            ModelResponse(choices=[Choices(message=Message(content="{'answer': 'Test output'}"))]),
+        ]
+
+        result = program(question="Dummy question!")
+        # The parse should succeed on the second call
+        assert mock_completion.call_count == 2
+        assert result.answer == "Test output"
+
+        # The first call should have tried structured output
+        _, first_call_kwargs = mock_completion.call_args_list[0]
+        assert issubclass(first_call_kwargs.get("response_format"), pydantic.BaseModel)
+
+        # The second call should have used JSON mode
+        _, second_call_kwargs = mock_completion.call_args_list[1]
+        assert second_call_kwargs.get("response_format") == {"type": "json_object"}
+
+
+@pytest.mark.asyncio
+async def test_json_adapter_fallback_to_json_mode_on_structured_output_failure_async():
+    class TestSignature(dspy.Signature):
+        question: str = dspy.InputField()
+        answer: str = dspy.OutputField(desc="String output field")
+
+    program = dspy.Predict(TestSignature)
+
+    with mock.patch("litellm.acompletion") as mock_acompletion:
+        # First call raises error to simulate structured output failure, second call returns a valid response
+        mock_acompletion.side_effect = [
+            RuntimeError("Structured output failed!"),
+            ModelResponse(choices=[Choices(message=Message(content="{'answer': 'Test output'}"))]),
+        ]
+
+        with dspy.context(lm=dspy.LM(model="openai/gpt-4o-mini", cache=False), adapter=dspy.JSONAdapter()):
+            result = await program.acall(question="Dummy question!")
+        # The parse should succeed on the second call
+        assert mock_acompletion.call_count == 2
+        assert result.answer == "Test output"
+
+        # The first call should have tried structured output
+        _, first_call_kwargs = mock_acompletion.call_args_list[0]
+        assert issubclass(first_call_kwargs.get("response_format"), pydantic.BaseModel)
+
+        # The second call should have used JSON mode
+        _, second_call_kwargs = mock_acompletion.call_args_list[1]
+        assert second_call_kwargs.get("response_format") == {"type": "json_object"}
+
+
+def test_error_message_on_json_adapter_failure():
+    class TestSignature(dspy.Signature):
+        question: str = dspy.InputField()
+        answer: str = dspy.OutputField(desc="String output field")
+
+    program = dspy.Predict(TestSignature)
+
+    dspy.configure(lm=dspy.LM(model="openai/gpt-4o-mini", cache=False), adapter=dspy.JSONAdapter())
+
+    with mock.patch("litellm.completion") as mock_completion:
+        mock_completion.side_effect = RuntimeError("Failed!")
+
+        with pytest.raises(RuntimeError) as error:
+            program(question="Dummy question!")
+
+        assert ERROR_MESSAGE_ON_JSON_ADAPTER_FAILURE[:50] in str(error.value)
+
+
+@pytest.mark.asyncio
+async def test_error_message_on_json_adapter_failure_async():
+    class TestSignature(dspy.Signature):
+        question: str = dspy.InputField()
+        answer: str = dspy.OutputField(desc="String output field")
+
+    program = dspy.Predict(TestSignature)
+
+    with mock.patch("litellm.acompletion") as mock_acompletion:
+        mock_acompletion.side_effect = RuntimeError("Failed!")
+
+        with dspy.context(lm=dspy.LM(model="openai/gpt-4o-mini", cache=False), adapter=dspy.JSONAdapter()):
+            with pytest.raises(RuntimeError) as error:
+                await program.acall(question="Dummy question!")
+
+        assert ERROR_MESSAGE_ON_JSON_ADAPTER_FAILURE[:50] in str(error.value)
diff --git a/tests/streaming/test_streaming.py b/tests/streaming/test_streaming.py
index e3283765..e9befaa3 100644
--- a/tests/streaming/test_streaming.py
+++ b/tests/streaming/test_streaming.py
@@ -1,4 +1,6 @@
+import asyncio
 import os
+import time
 from unittest import mock
 from unittest.mock import AsyncMock
 
@@ -649,3 +651,59 @@ async def test_stream_listener_returns_correct_chunk_json_adapter_untokenized_st
         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."
+
+
+@pytest.mark.anyio
+async def test_status_message_non_blocking():
+    def dummy_tool():
+        time.sleep(1)
+        return "dummy_tool_output"
+
+    class MyProgram(dspy.Module):
+        def forward(self, question, **kwargs):
+            dspy.Tool(dummy_tool)()
+            return dspy.Prediction(answer="dummy_tool_output")
+
+    program = dspy.streamify(MyProgram(), status_message_provider=StatusMessageProvider())
+
+    with mock.patch("litellm.acompletion", new_callable=AsyncMock, side_effect=[dummy_tool]):
+        with dspy.context(lm=dspy.LM("openai/gpt-4o-mini", cache=False)):
+            output = program(question="why did a chicken cross the kitchen?")
+            timestamps = []
+            async for value in output:
+                if isinstance(value, dspy.streaming.StatusMessage):
+                    timestamps.append(time.time())
+
+    # timestamps[0]: tool start message
+    # timestamps[1]: tool end message
+    # There should be ~1 second delay between the tool start and end messages because we explicitly sleep for 1 second
+    # in the tool.
+    assert timestamps[1] - timestamps[0] >= 1
+
+
+@pytest.mark.anyio
+async def test_status_message_non_blocking_async_program():
+    async def dummy_tool():
+        await asyncio.sleep(1)
+        return "dummy_tool_output"
+
+    class MyProgram(dspy.Module):
+        async def aforward(self, question, **kwargs):
+            await dspy.Tool(dummy_tool).acall()
+            return dspy.Prediction(answer="dummy_tool_output")
+
+    program = dspy.streamify(MyProgram(), status_message_provider=StatusMessageProvider(), is_async_program=True)
+
+    with mock.patch("litellm.acompletion", new_callable=AsyncMock, side_effect=[dummy_tool]):
+        with dspy.context(lm=dspy.LM("openai/gpt-4o-mini", cache=False)):
+            output = program(question="why did a chicken cross the kitchen?")
+            timestamps = []
+            async for value in output:
+                if isinstance(value, dspy.streaming.StatusMessage):
+                    timestamps.append(time.time())
+
+    # timestamps[0]: tool start message
+    # timestamps[1]: tool end message
+    # There should be ~1 second delay between the tool start and end messages because we explicitly sleep for 1 second
+    # in the tool.
+    assert timestamps[1] - timestamps[0] >= 1
diff --git a/tests/utils/test_settings.py b/tests/utils/test_settings.py
index 590420a1..f2bd93ec 100644
--- a/tests/utils/test_settings.py
+++ b/tests/utils/test_settings.py
@@ -20,27 +20,13 @@ def test_forbid_configure_call_in_child_thread():
     dspy.configure(lm=dspy.LM("openai/gpt-4o"), adapter=dspy.JSONAdapter(), callbacks=[lambda x: x])
 
     def worker():
-        with pytest.raises(RuntimeError, match="Cannot call dspy.configure in a child thread"):
+        with pytest.raises(RuntimeError, match="Cannot call dspy.configure"):
             dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"), callbacks=[])
 
     with ThreadPoolExecutor(max_workers=1) as executor:
         executor.submit(worker)
 
 
-@pytest.mark.asyncio
-async def test_forbid_configure_call_in_async_function():
-    with pytest.raises(
-        RuntimeError,
-        match=r"dspy.settings.configure\(\.\.\.\) cannot be called a second time from*",
-    ):
-        dspy.configure(lm=dspy.LM("openai/gpt-4o"), adapter=dspy.JSONAdapter(), callbacks=[lambda x: x])
-        dspy.configure(lm=dspy.LM("openai/gpt-4o"), adapter=dspy.JSONAdapter(), callbacks=[lambda x: x])
-
-    with dspy.context(lm=dspy.LM("openai/gpt-4o-mini"), callbacks=[]):
-        # context is allowed
-        pass
-
-
 def test_dspy_context():
     dspy.configure(lm=dspy.LM("openai/gpt-4o"), adapter=dspy.JSONAdapter(), callbacks=[lambda x: x])
     with dspy.context(lm=dspy.LM("openai/gpt-4o-mini"), callbacks=[]):
@@ -165,3 +151,43 @@ async def test_dspy_context_with_async_task_group():
         # The main thread is not affected by the context
         assert dspy.settings.lm.model == "openai/gpt-4.1"
         assert dspy.settings.trace == []
+
+
+@pytest.mark.asyncio
+async def test_dspy_configure_allowance_async():
+    def bar1():
+        # `dspy.configure` is disallowed in different async tasks from the initial one.
+        # In this case, foo1 (async) calls bar1 (sync), and bar1 uses the async task from foo1.
+        with pytest.raises(RuntimeError) as e:
+            dspy.configure(lm=dspy.LM("openai/gpt-4o"))
+        assert "dspy.settings.configure(...) can only be called from the same async" in str(e.value)
+
+    async def foo1():
+        bar1()
+        await asyncio.sleep(0.1)
+
+    async def foo2():
+        # `dspy.configure` is disallowed in different async tasks from the initial one.
+        with pytest.raises(RuntimeError) as e:
+            dspy.configure(lm=dspy.LM("openai/gpt-4o"))
+        assert "dspy.settings.configure(...) can only be called from the same async" in str(e.value)
+        await asyncio.sleep(0.1)
+
+    async def foo3():
+        # `dspy.context` is allowed in different async tasks from the initial one.
+        with dspy.context(lm=dspy.LM("openai/gpt-4o")):
+            await asyncio.sleep(0.1)
+
+    async def foo4():
+        # foo4 is directly invoked by the entry task, so it has the same async task as the entry task.
+        dspy.configure(lm=dspy.LM("openai/gpt-4o"))
+        await asyncio.sleep(0.1)
+
+    # `dspy.configure` is allowed to be called multiple times in the same async task.
+    dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
+    dspy.configure(lm=dspy.LM("openai/gpt-4o"))
+    dspy.configure(adapter=dspy.JSONAdapter())
+
+    await asyncio.gather(foo1(), foo2(), foo3())
+
+    foo4()