Skip to content

dspy.utils.StatusMessageProvider

dspy.utils.StatusMessageProvider

Provides customizable status message streaming for DSPy programs.

This class serves as a base for creating custom status message providers. Users can subclass and override its methods to define specific status messages for different stages of program execution, each method must return a string.

Example: 

class MyStatusMessageProvider(StatusMessageProvider):
    def lm_start_status_message(self, instance, inputs):
        return f"Calling LM with inputs {inputs}..."

    def module_end_status_message(self, outputs):
        return f"Module finished with output: {outputs}!"

program = dspy.streamify(dspy.Predict("q->a"), status_message_provider=MyStatusMessageProvider())

Functions

lm_end_status_message(outputs: Any)

Status message after a dspy.LM is called.

Source code in dspy/utils/streaming.py 
70
71
72
def lm_end_status_message(self, outputs: Any):
    """Status message after a `dspy.LM` is called."""
    pass

lm_start_status_message(instance: Any, inputs: Dict[str, Any])

Status message before a dspy.LM is called.

Source code in dspy/utils/streaming.py 
66
67
68
def lm_start_status_message(self, instance: Any, inputs: Dict[str, Any]):
    """Status message before a `dspy.LM` is called."""
    pass

module_end_status_message(outputs: Any)

Status message after a dspy.Module or dspy.Predict is called.

Source code in dspy/utils/streaming.py 
62
63
64
def module_end_status_message(self, outputs: Any):
    """Status message after a `dspy.Module` or `dspy.Predict` is called."""
    pass

module_start_status_message(instance: Any, inputs: Dict[str, Any])

Status message before a dspy.Module or dspy.Predict is called.

Source code in dspy/utils/streaming.py 
58
59
60
def module_start_status_message(self, instance: Any, inputs: Dict[str, Any]):
    """Status message before a `dspy.Module` or `dspy.Predict` is called."""
    pass

tool_end_status_message(outputs: Any)

Status message after a dspy.Tool is called.

Source code in dspy/utils/streaming.py 
54
55
56
def tool_end_status_message(self, outputs: Any):
    """Status message after a `dspy.Tool` is called."""
    return "Tool calling finished! Querying the LLM with tool calling results..."

tool_start_status_message(instance: Any, inputs: Dict[str, Any])

Status message before a dspy.Tool is called.

Source code in dspy/utils/streaming.py 
50
51
52
def tool_start_status_message(self, instance: Any, inputs: Dict[str, Any]):
    """Status message before a `dspy.Tool` is called."""
    return f"Calling tool {instance.name}..."