将 DSPy 程序包装起来,使其可以异步调用。这对于与另一个任务(例如另一个 DSPy 程序)并行运行程序非常有用。
此实现将当前线程的配置上下文传播到工作线程。
参数
名称 |
类型 |
描述 |
默认值 |
program
|
Module
|
|
必需
|
返回值
类型 |
描述 |
Callable[[Any, Any], Awaitable[Any]]
|
一个异步函数:一个异步函数,当被等待时,它将在工作线程中运行程序。每次调用都会继承当前线程的配置上下文。
|
源代码位于 dspy/utils/asyncify.py
| 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).
This implementation propagates the current thread's configuration context to the worker thread.
Args:
program: The DSPy program to be wrapped for asynchronous execution.
Returns:
An async function: An async function that, when awaited, runs the program in a worker thread.
The current thread's configuration context is inherited for each call.
"""
async def async_program(*args, **kwargs) -> Any:
# Capture the current overrides at call-time.
from dspy.dsp.utils.settings import thread_local_overrides
parent_overrides = thread_local_overrides.overrides.copy()
def wrapped_program(*a, **kw):
from dspy.dsp.utils.settings import thread_local_overrides
original_overrides = thread_local_overrides.overrides
thread_local_overrides.overrides = parent_overrides.copy()
try:
return program(*a, **kw)
finally:
thread_local_overrides.overrides = original_overrides
# Create a fresh asyncified callable each time, ensuring the latest context is used.
call_async = asyncer.asyncify(wrapped_program, abandon_on_cancel=True, limiter=get_limiter())
return await call_async(*args, **kwargs)
return async_program
|