Repository: stanfordnlp/dspy
Pull request #7760 (2025-02-05 02:03:22)

Add 'DSPy in Production' page
As discussed offline, adding a new page describes the production related topics about DSPy.

![Screenshot 2025-02-05 at 11 02 08](https://github.com/user-attachments/assets/b90639b0-400d-49a1-819a-ae30849e4aba)



Patch:
diff --git a/docs/docs/production/index.md b/docs/docs/production/index.md
new file mode 100644
index 00000000..b952efb1
--- /dev/null
+++ b/docs/docs/production/index.md
@@ -0,0 +1,53 @@
+# Using DSPy in Production
+
+<div class="grid cards" style="text-align: left;" markdown>
+
+- :material-earth:{ .lg .middle } __Real-World Use Cases__
+
+    ---
+
+    DSPy is deployed in production by many enterprises and startups. Explore real-world case studies.
+
+    [:octicons-arrow-right-24: Use Cases](../dspy-usecases.md)
+
+- :material-magnify-expand:{ .lg .middle } __Monitoring & Observability__
+
+    ---
+
+    Monitor your DSPy programs using **MLflow Tracing**, based on OpenTelemetry.
+
+    [:octicons-arrow-right-24: Set Up Observability](../tutorials/observability/#tracing)
+
+- :material-ab-testing: __Reproducibility__
+
+    ---
+
+    Log programs, metrics, configs, and environments for full reproducibility with DSPy's native MLflow integration.
+
+    [:octicons-arrow-right-24: MLflow Integration](https://mlflow.org/docs/latest/llms/dspy/index.html)
+
+- :material-rocket-launch: __Deployment__
+
+    ---
+
+    When it's time to productionize, deploy your application easily with DSPy's integration with MLflow Model Serving.
+
+    [:octicons-arrow-right-24: Deployment Guide](../tutorials/deployment)
+
+- :material-arrow-up-right-bold: __Scalability__
+
+    ---
+
+    DSPy is designed with thread-safety in mind and offers native asynchronous execution support for high-throughput environments.
+
+    [:octicons-arrow-right-24: Async Program](../api/utils/asyncify.md)
+
+- :material-alert-rhombus: __Guardrails & Controllability__
+
+    ---
+
+    DSPy's **Signatures**, **Modules**, and **Optimizers** help you control and guide LM outputs.
+
+    [:octicons-arrow-right-24: Learn Signature](../learn/programming/signatures.md)
+
+</div>
diff --git a/docs/docs/stylesheets/extra.css b/docs/docs/stylesheets/extra.css
index 9a745886..d0bef54c 100644
--- a/docs/docs/stylesheets/extra.css
+++ b/docs/docs/stylesheets/extra.css
@@ -51,6 +51,10 @@
     text-align: justify;
 }
 
+/* Left-aligned text for grid cards */
+.md-content__inner .grid.cards p {
+    text-align: left;
+}
 
 /* Base styling for the output area */
 .jp-Cell-outputWrapper .jp-OutputArea-output pre {
diff --git a/docs/docs/tutorials/observability/index.md b/docs/docs/tutorials/observability/index.md
index e7568107..5b82796e 100644
--- a/docs/docs/tutorials/observability/index.md
+++ b/docs/docs/tutorials/observability/index.md
@@ -4,7 +4,7 @@ This guide demonstrates how to debug problems and improve observability in DSPy.
 
 However, as systems grow more sophisticated, the ability to **understand what your system is doing** becomes critical. Without transparency, the prediction process can easily become a black box, making failures or quality issues difficult to diagnose and production maintenance challenging.
 
-By the end of this tutorial, you'll understand how to debug an issue and improve observability using [MLflow Tracing](#mlflow-tracing). You'll also explore how to build a custom logging solution using callbacks.
+By the end of this tutorial, you'll understand how to debug an issue and improve observability using [MLflow Tracing](#tracing). You'll also explore how to build a custom logging solution using callbacks.
 
 
 
diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml
index 9c195c47..bdb64321 100644
--- a/docs/mkdocs.yml
+++ b/docs/mkdocs.yml
@@ -127,6 +127,8 @@ nav:
             - load: api/utils/load.md
             - streamify: api/utils/streamify.md
 
+    - DSPy in Production: production/index.md
+
 theme:
     name: material
     custom_dir: overrides
@@ -231,6 +233,7 @@ markdown_extensions:
     - pymdownx.superfences
     - pymdownx.mark
     - attr_list
+    - md_in_html
     - pymdownx.emoji:
         emoji_index: !!python/name:material.extensions.emoji.twemoji
         emoji_generator: !!python/name:materialx.emoji.to_svg
diff --git a/dspy/adapters/json_adapter.py b/dspy/adapters/json_adapter.py
index 7cfc2d36..eda982ea 100644
--- a/dspy/adapters/json_adapter.py
+++ b/dspy/adapters/json_adapter.py
@@ -129,7 +129,7 @@ def parse_value(value, annotation):
         parsed_value = find_enum_member(annotation, value)
     elif isinstance(value, str):
         try:
-            parsed_value = json.loads(value)
+            parsed_value = json_repair.loads(value)
         except json.JSONDecodeError:
             try:
                 parsed_value = ast.literal_eval(value)
diff --git a/dspy/utils/__init__.py b/dspy/utils/__init__.py
index b2f2d053..b6e95337 100644
--- a/dspy/utils/__init__.py
+++ b/dspy/utils/__init__.py
@@ -1,5 +1,6 @@
 from dspy.utils.callback import BaseCallback, with_callbacks
 from dspy.utils.dummies import DummyLM, DummyVectorizer, dummy_rm
+from dspy.utils.streaming import StatusMessage, StatusMessageProvider, streamify
 
 import os
 import requests
@@ -24,4 +25,7 @@ __all__ = [
     "DummyLM",
     "DummyVectorizer",
     "dummy_rm",
+    "StatusMessage",
+    "StatusMessageProvider",
+    "streamify",
 ]
diff --git a/dspy/utils/asyncify.py b/dspy/utils/asyncify.py
index 0680bc39..7596a03f 100644
--- a/dspy/utils/asyncify.py
+++ b/dspy/utils/asyncify.py
@@ -1,9 +1,10 @@
-from typing import Any, Awaitable, Callable
+from typing import TYPE_CHECKING, Any, Awaitable, Callable
 
 import asyncer
 from anyio import CapacityLimiter
 
-from dspy.primitives.program import Module
+if TYPE_CHECKING:
+    from dspy.primitives.program import Module
 
 _limiter = None
 
@@ -26,7 +27,7 @@ def get_limiter():
     return _limiter
 
 
-def asyncify(program: Module) -> Callable[[Any, Any], Awaitable[Any]]:
+def asyncify(program: "Module") -> Callable[[Any, Any], Awaitable[Any]]:
     """
     Wraps a DSPy program so that it can be called asynchronously. This is useful for running a
     program in parallel with another task (e.g., another DSPy program).
diff --git a/dspy/utils/streaming.py b/dspy/utils/streaming.py
index f51ac11b..d885591c 100644
--- a/dspy/utils/streaming.py
+++ b/dspy/utils/streaming.py
@@ -1,23 +1,178 @@
+import logging
 from asyncio import iscoroutinefunction
-from typing import Any, AsyncGenerator, Awaitable, Callable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Awaitable, Callable, Dict, Optional
 
 import litellm
 import ujson
 from anyio import create_memory_object_stream, create_task_group
 from anyio.streams.memory import MemoryObjectSendStream
+from asyncer import syncify
 
+from dspy.dsp.utils.settings import settings
 from dspy.primitives.prediction import Prediction
-from dspy.primitives.program import Module
 from dspy.utils.asyncify import asyncify
+from dspy.utils.callback import BaseCallback
 
+logger = logging.getLogger(__name__)
 
-def streamify(program: Module) -> Callable[[Any, Any], Awaitable[Any]]:
+if TYPE_CHECKING:
+    from dspy.primitives.program import Module
+
+
+@dataclass
+class StatusMessage:
+    message: str
+
+
+class StatusMessageProvider:
+    def tool_start_status_message(self, instance: Any, inputs: Dict[str, Any]):
+        return f"Calling tool {instance.name}..."
+
+    def tool_end_status_message(self, outputs: Any):
+        return "Tool calling finished! Querying the LLM with tool calling results..."
+
+    def module_start_status_message(self, instance: Any, inputs: Dict[str, Any]):
+        pass
+
+    def module_end_status_message(self, outputs: Any):
+        pass
+
+    def lm_start_status_message(self, instance: Any, inputs: Dict[str, Any]):
+        pass
+
+    def lm_end_status_message(self, outputs: Any):
+        pass
+
+
+class StatusStreamingCallback(BaseCallback):
+    def __init__(self, status_message_provider: Optional[StatusMessageProvider] = None):
+        self.status_message_provider = status_message_provider or StatusMessageProvider()
+
+    def on_tool_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: Dict[str, Any],
+    ):
+        stream = settings.send_stream
+        if stream is None or instance.name == "finish":
+            return
+
+        @syncify
+        async def send_status():
+            status_message = self.status_message_provider.tool_start_status_message(instance, inputs)
+            if status_message:
+                await stream.send(StatusMessage(status_message))
+
+        send_status()
+
+    def on_tool_end(
+        self,
+        call_id: str,
+        outputs: Optional[Dict[str, Any]],
+        exception: Optional[Exception] = None,
+    ):
+        stream = settings.send_stream
+        if stream is None or outputs == "Completed.":
+            return
+
+        @syncify
+        async def send_status():
+            status_message = self.status_message_provider.tool_end_status_message(outputs)
+            if status_message:
+                await stream.send(StatusMessage(status_message))
+
+        send_status()
+
+    def on_lm_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: Dict[str, Any],
+    ):
+        stream = settings.send_stream
+        if stream is None:
+            return
+
+        @syncify
+        async def send_status():
+            status_message = self.status_message_provider.lm_start_status_message(instance, inputs)
+            if status_message:
+                await stream.send(StatusMessage(status_message))
+
+        send_status()
+
+    def on_lm_end(
+        self,
+        call_id: str,
+        outputs: Optional[Dict[str, Any]],
+        exception: Optional[Exception] = None,
+    ):
+        stream = settings.send_stream
+        if stream is None:
+            return
+
+        @syncify
+        async def send_status():
+            status_message = self.status_message_provider.lm_end_status_message(outputs)
+            if status_message:
+                await stream.send(StatusMessage(status_message))
+
+        send_status()
+
+    def on_module_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: Dict[str, Any],
+    ):
+        stream = settings.send_stream
+        if stream is None:
+            return
+
+        @syncify
+        async def send_status():
+            status_message = self.status_message_provider.module_start_status_message(instance, inputs)
+            if status_message:
+                await stream.send(StatusMessage(status_message))
+
+        send_status()
+
+    def on_module_end(
+        self,
+        call_id: str,
+        outputs: Optional[Dict[str, Any]],
+        exception: Optional[Exception] = None,
+    ):
+        stream = settings.send_stream
+        if stream is None:
+            return
+
+        @syncify
+        async def send_status():
+            status_message = self.status_message_provider.module_end_status_message(outputs)
+            if status_message:
+                await stream.send(StatusMessage(status_message))
+
+        send_status()
+
+
+def streamify(
+    program: "Module",
+    status_message_provider: Optional[StatusMessageProvider] = None,
+) -> Callable[[Any, Any], Awaitable[Any]]:
     """
     Wrap a DSPy program so that it streams its outputs incrementally, rather than returning them
-    all at once.
+    all at once. It also provides status messages to the user to indicate the progress of the program, and users
+    can implement their own status message provider to customize the status messages and what module to generate
+    status messages for.
 
     Args:
         program: The DSPy program to wrap with streaming functionality.
+        status_message_provider: A custom status message generator to use instead of the default one. Users can
+            implement their own status message generator to customize the status messages and what module to generate
+            status messages for.
 
     Returns:
         A function that takes the same arguments as the original program, but returns an async
@@ -26,27 +181,68 @@ def streamify(program: Module) -> Callable[[Any, Any], Awaitable[Any]]:
     Example:
 
     ```python
-    class TestSignature(dspy.Signature):
-        input_text: str = dspy.InputField()
-        output_text: str = dspy.OutputField()
+    import asyncio
+    import dspy
 
     # Create the program and wrap it with streaming functionality
-    program = dspy.streamify(dspy.Predict(TestSignature))
+    program = dspy.streamify(dspy.Predict("q->a"))
 
     # Use the program with streaming output
     async def use_streaming():
-        output_stream = program(input_text="Test")
-        async for value in output_stream:
-            print(value)  # Print each streamed value incrementally
+        output = program(q="Why did a chicken cross the kitchen?")
+        return_value = None
+        async for value in output:
+            if isinstance(value, dspy.Prediction):
+                return_value = value
+            else:
+                print(value)
+        return return_value
+
+    output = asyncio.run(use_streaming())
+    print(output)
     ```
-    """
+
+    Example with custom status message provider:
+    ```python
+    import asyncio
     import dspy
 
+    class MyStatusMessageProvider(StatusMessageProvider):
+        def module_start_status_message(self, instance, inputs):
+            return f"Predicting..."
+
+        def tool_end_status_message(self, outputs):
+            return f"Tool calling finished with output: {outputs}!"
+
+    # Create the program and wrap it with streaming functionality
+    program = dspy.streamify(dspy.Predict("q->a"), status_message_provider=MyStatusMessageProvider())
+
+    # Use the program with streaming output
+    async def use_streaming():
+        output = program(q="Why did a chicken cross the kitchen?")
+        return_value = None
+        async for value in output:
+            if isinstance(value, dspy.Prediction):
+                return_value = value
+            else:
+                print(value)
+        return return_value
+
+    output = asyncio.run(use_streaming())
+    print(output)
+    ```
+    """
     if not iscoroutinefunction(program):
         program = asyncify(program)
 
+    callbacks = settings.callbacks
+    status_streaming_callback = StatusStreamingCallback(status_message_provider)
+    if not any(isinstance(c, StatusStreamingCallback) for c in callbacks):
+        callbacks.append(status_streaming_callback)
+    settings.configure(callbacks=callbacks)
+
     async def generator(args, kwargs, stream: MemoryObjectSendStream):
-        with dspy.settings.context(send_stream=stream):
+        with settings.context(send_stream=stream):
             prediction = await program(*args, **kwargs)
 
         await stream.send(prediction)


Test patch:
diff --git a/tests/utils/test_streaming.py b/tests/utils/test_streaming.py
index 2b9c9fd5..3e890780 100644
--- a/tests/utils/test_streaming.py
+++ b/tests/utils/test_streaming.py
@@ -1,7 +1,7 @@
 import pytest
 
 import dspy
-from dspy.utils.streaming import streaming_response
+from dspy.utils.streaming import StatusMessage, StatusMessageProvider, streaming_response
 from ..test_utils.server import litellm_test_server
 
 
@@ -58,3 +58,67 @@ async def test_streaming_response_yields_expected_response_chunks(litellm_test_s
         assert all(chunk.startswith("data: ") for chunk in output_chunks)
         assert 'data: {"prediction":{"output_text":"Hello!"}}\n\n' in output_chunks
         assert output_chunks[-1] == "data: [DONE]\n\n"
+
+
+@pytest.mark.anyio
+async def test_default_status_streaming():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.generate_question = dspy.Tool(lambda x: f"What color is the {x}?", name="generate_question")
+            self.predict = dspy.Predict("question->answer")
+
+        def __call__(self, x: str):
+            question = self.generate_question(x=x)
+            return self.predict(question=question)
+
+    lm = dspy.utils.DummyLM([{"answer": "red"}, {"answer": "blue"}])
+    with dspy.context(lm=lm):
+        program = dspy.streamify(MyProgram())
+        output = program("sky")
+
+        status_messages = []
+        async for value in output:
+            if isinstance(value, StatusMessage):
+                status_messages.append(value)
+
+    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_custom_status_streaming():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.generate_question = dspy.Tool(lambda x: f"What color is the {x}?", name="generate_question")
+            self.predict = dspy.Predict("question->answer")
+
+        def __call__(self, x: str):
+            question = self.generate_question(x=x)
+            return self.predict(question=question)
+
+    class MyStatusMessageProvider(StatusMessageProvider):
+        def tool_start_status_message(self, instance, inputs):
+            return f"Tool starting!"
+
+        def tool_end_status_message(self, outputs):
+            return "Tool finished!"
+
+        def module_start_status_message(self, instance, inputs):
+            if isinstance(instance, dspy.Predict):
+                return "Predict starting!"
+
+    lm = dspy.utils.DummyLM([{"answer": "red"}, {"answer": "blue"}])
+    with dspy.context(lm=lm):
+        program = dspy.streamify(MyProgram(), status_message_provider=MyStatusMessageProvider())
+        output = program("sky")
+
+        status_messages = []
+        async for value in output:
+            if isinstance(value, StatusMessage):
+                status_messages.append(value)
+
+        assert len(status_messages) == 3
+        assert status_messages[0].message == "Tool starting!"
+        assert status_messages[1].message == "Tool finished!"
+        assert status_messages[2].message == "Predict starting!"