Provide a dspy.syncify so that users can run optimizer on async dspy programs (#8509)

* Add sync wrapper so that async program can be used with optimizer

* fix tests

Author: chenmoneygithub

dspy/__init__.py

@@ -9,6 +9,7 @@
 from dspy.adapters import Adapter, ChatAdapter, JSONAdapter, XMLAdapter, TwoStepAdapter, Image, Audio, History, Type, Tool, ToolCalls  # isort: skip
 from dspy.utils.logging_utils import configure_dspy_loggers, disable_logging, enable_logging
 from dspy.utils.asyncify import asyncify
+from dspy.utils.syncify import syncify
 from dspy.utils.saving import load
 from dspy.streaming.streamify import streamify
 from dspy.utils.usage_tracker import track_usage

dspy/teleprompt/bootstrap.py

@@ -245,11 +245,12 @@ def _bootstrap_one_example(self, example, round_idx=0):
 
             # Update the traces
             for name, demos in name2traces.items():
-                from datasets.fingerprint import Hasher
 
                 # If there are multiple traces for the same predictor in the sample example,
                 # sample 50/50 from the first N-1 traces or the last trace.
                 if len(demos) > 1:
+                    from datasets.fingerprint import Hasher
+
                     rng = random.Random(Hasher.hash(tuple(demos)))
                     demos = [rng.choice(demos[:-1]) if rng.random() < 0.5 else demos[-1]]
                 self.name2traces[name].extend(demos)

dspy/utils/__init__.py

@@ -7,6 +7,7 @@
 from dspy.utils.callback import BaseCallback, with_callbacks
 from dspy.utils.dummies import DummyLM, DummyVectorizer, dummy_rm
 from dspy.utils.inspect_history import pretty_print_history
+from dspy.utils.syncify import syncify
 
 
 def download(url):

dspy/utils/syncify.py

@@ -0,0 +1,60 @@
+import asyncio
+from types import MethodType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from dspy.primitives.module import Module
+
+
+def run_async(coro):
+    """Run an async coroutine from a synchronous context."""
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = None
+
+    if loop and loop.is_running():
+        # If we're in a running event loop (e.g., Jupyter), use asyncio.create_task and run until done
+        import nest_asyncio
+
+        nest_asyncio.apply()
+        return asyncio.get_event_loop().run_until_complete(coro)
+    else:
+        return asyncio.run(coro)
+
+
+def syncify(program: "Module", in_place: bool = True) -> "Module":
+    """Convert an async DSPy module to a sync program.
+
+    There are two modes of this function:
+
+    - `in_place=True` (recommended): Modify the module in place. But this may not work if you already have a `forward`
+        method which does different things from `aforward`.
+    - `in_place=False`: Return a wrapper module. This changes the module's architecture, but it's more robust.
+
+    Args:
+        program: The async program to convert, must have an `aforward` method implemented.
+        in_place: If True, modify the module in place. Otherwise, return a wrapper module.
+
+    Returns:
+        The sync program, which has a `forward` method that can be called from a synchronous context.
+    """
+    if in_place:
+
+        def forward(self, *args, **kwargs):
+            return run_async(self.aforward(*args, **kwargs))
+
+        # Create the `forward` method in place.
+        program.forward = MethodType(forward, program)
+        return program
+    else:
+        from dspy.primitives.module import Module
+
+        class SyncWrapper(Module):
+            def __init__(self, program: "Module"):
+                self.program = program
+
+            def forward(self, *args, **kwargs):
+                return run_async(self.program.aforward(*args, **kwargs))
+
+        return SyncWrapper(program)

tests/utils/test_syncify.py

@@ -0,0 +1,57 @@
+import asyncio
+
+import dspy
+
+
+def test_syncify_in_place():
+    class MyProgram(dspy.Module):
+        async def aforward(self, x: int) -> int:
+            await asyncio.sleep(0.01)
+            return x + 1
+
+    sync_program = dspy.syncify(MyProgram())
+    assert sync_program(1) == 2
+    assert sync_program(2) == 3
+
+
+def test_syncify_with_wrapper():
+    class MyProgram(dspy.Module):
+        async def aforward(self, x: int) -> int:
+            await asyncio.sleep(0.01)
+            return x + 1
+
+    sync_program = dspy.syncify(MyProgram(), in_place=False)
+    assert sync_program(1) == 2
+    assert sync_program(2) == 3
+
+
+def test_syncify_works_with_optimizers():
+    class MyProgram(dspy.Module):
+        def __init__(self):
+            self.predict = dspy.Predict("question->answer")
+
+        async def aforward(self, question: str):
+            return await self.predict.acall(question=question)
+
+    async_program = MyProgram()
+
+    def dummy_metric(gold, pred, traces=None):
+        return True
+
+    # We only test the optimizer completes without errors, so the LM response doesn't matter.
+    lm = dspy.utils.DummyLM([{"answer": "dummy"} for _ in range(100)])
+    dspy.configure(lm=lm)
+
+    dataset = [dspy.Example(question="question", answer="answer").with_inputs("question") for _ in range(10)]
+
+    optimizer = dspy.BootstrapFewShot(metric=dummy_metric, max_bootstrapped_demos=2, max_labeled_demos=0)
+
+    # Test syncify in place
+    sync_program = dspy.syncify(async_program, in_place=True)
+    optimized_program = optimizer.compile(sync_program, trainset=dataset)
+    assert len(optimized_program.predictors()[0].demos) == 2
+
+    # Test syncify with wrapper
+    sync_program = dspy.syncify(async_program, in_place=False)
+    optimized_program = optimizer.compile(sync_program, trainset=dataset)
+    assert len(optimized_program.predictors()[0].demos) == 2