dspy.InterpreterPool¶

InterpreterPool is an interpreter_factory that keeps interpreters started ahead of need. Each call still returns a fresh interpreter that has run no code, but usually one that has already started, so a dspy.Flex forward or a dspy.RLM call doesn’t wait for a sandbox to boot. With the default PythonInterpreter, a pooled interpreter has also loaded dspy, which is the first thing a Flex forward imports.

import dspy

pool = dspy.InterpreterPool(size=2)
program = dspy.Flex("question -> answer", interpreter_factory=pool)

The pool starts size interpreters in the background and starts a replacement whenever it hands one out. When none is ready, a call makes a new interpreter on the spot, as the plain factory would.

Pick size by how calls arrive. One interpreter is enough when each call spends longer in LM calls than a sandbox takes to start (well under a second). Calls that run back to back with little LM time, or in parallel threads such as dspy.Evaluate(num_threads=...), need one for each call in flight.

factory must return interchangeable interpreters, because the pool calls it before the interpreter is needed. A factory whose interpreters depend on when it is called, for example on the current input, should not be pooled.

pool = dspy.InterpreterPool(size=2)
program = dspy.Flex(MySignature, interpreter_factory=pool)

40
41
42
43
44
45
46
47
48
49
50
51

def __init__(self, factory: Callable[[], CodeInterpreter] = PythonInterpreter, size: int = 1) -> None:
    if size < 1:
        raise ValueError("size must be at least 1")
    self._factory = factory
    self._size = size
    self._starting = 0
    self._ready: queue.Queue[CodeInterpreter] = queue.Queue()
    self._lock = threading.Lock()
    self._closed = False
    # Shuts the idle interpreters down when the pool is closed, garbage-collected, or the process exits.
    self._finalizer = weakref.finalize(self, _shut_down_idle, self._ready)
    self._refill()

57
58
59
60
61
62
63

def __call__(self) -> CodeInterpreter:
    try:
        interpreter = self._ready.get_nowait()
    except queue.Empty:
        interpreter = None
    self._refill()
    return interpreter if interpreter is not None else self._factory()

65
66
67
68
69

def close(self) -> None:
    """Shut down the idle interpreters and stop starting new ones."""
    with self._lock:
        self._closed = True
    self._finalizer()