zodiac.toga.signatures

View Source

  1# SPDX-License-Identifier: MPL-2.0 AND LicenseRef-Commons-Clause-License-Condition-1.0
  2# <!-- // /*  d a r k s h a p e s */ -->
  3
  4import dspy
  5from zodiac.providers.registry_entry import RegistryEntry
  6
  7from zodiac.providers.constants import CueType, PkgType, MIR_DB
  8
  9dspy.configure_cache(enable_disk_cache=False)
 10
 11
 12class StreamActivity(dspy.streaming.StatusMessageProvider):
 13    def lm_start_status_message(self, instance, inputs):
 14        return "Processing..."
 15
 16    def module_start_status_message(self, instance, inputs):
 17        return "Module started."
 18
 19    def lm_end_status_message(self, outputs):
 20        return "Done."
 21
 22    def tool_start_status_message(self, instance, inputs):
 23        return "Tool started..."
 24
 25    def tool_end_status_message(self, outputs):
 26        return "Tool finished."
 27
 28
 29class QATask(dspy.Signature):
 30    """Reply with short responses within 60-90 word/10k character code limits"""
 31
 32    question: str = dspy.InputField(desc="The question to respond to")
 33    answer = dspy.OutputField(desc="Often between 60 and 90 words and limited to 10000 character code blocks")
 34
 35
 36TARGET_LANGUAGE = "English"
 37
 38
 39class TranslateTask(dspy.Signature):
 40    f"""Translate from a language to {TARGET_LANGUAGE}"""
 41
 42    message: dspy.Image | dspy.Audio | str = dspy.InputField(desc="The input to translate.")
 43    translation: dspy.Image | dspy.Audio | str = dspy.OutputField(desc="A translation of the input")
 44
 45
 46class VisionTask(dspy.Signature):
 47    """Describe the image in detail."""
 48
 49    image: dspy.Image = dspy.InputField(desc="An image")
 50    description: str = dspy.OutputField(desc="A detailed description of the image.")
 51
 52
 53class TranscribeTask(dspy.Signature):
 54    """Transcribe spoken words into text"""
 55
 56    message: dspy.Audio = dspy.InputField(desc="The speech to transcribe.")
 57    answer: str = dspy.OutputField(desc="A transcript of the recorded speech")
 58
 59
 60class GenerativeImageTask(dspy.Signature):
 61    message: str = dspy.InputField(desc="Description of the image to generate")
 62    image: dspy.Image = dspy.OutputField(desc="An image matching the description")
 63
 64
 65class GenerativeAudioTask(dspy.Signature):
 66    message: str = dspy.InputField(desc="Description of the audio to generate")
 67    audio: dspy.Audio = dspy.OutputField(desc="An audio file matching the description")
 68
 69
 70# cherry pick examples
 71# dspy.Adapter.format(demos=[{:,:}],signatures:,inputs:)
 72
 73
 74class QuestionAnswer(dspy.Module):
 75    def __init__(self):
 76        super().__init__()
 77        self.predict = dspy.Predict(QATask)
 78
 79    def forward(self, question, **kwargs):
 80        self.predict(question=question, **kwargs)
 81        return self.predict(question=question, **kwargs)
 82
 83
 84class Predictor(dspy.Module):
 85    def __init__(self):
 86        super().__init__
 87        self.program = dspy.Predict(signature=QATask)
 88
 89    def __call__(self, question: str):
 90        from litellm.exceptions import APIConnectionError
 91        from litellm.llms.ollama.common_utils import OllamaError
 92        from httpx import ConnectError
 93        from dspy.utils.exceptions import AdapterParseError
 94        from aiohttp.client_exceptions import ClientConnectorError
 95
 96        try:
 97            return self.program(question=question)
 98        except (ClientConnectorError, ConnectError, AdapterParseError, APIConnectionError, OllamaError, OSError):
 99            pass
100
101    # from nnll.tensor_pipe import segments
102    # from nnll.configure.init_gpu import seed_planter
103    # mir_db = MIR_DB.database
104    # mir_arch = registry_entries.mir
105    # arch_data = mir_db[mir_arch[0][mir_arch[1]]]
106    # pkg_data = mir_db[mir_arch[0]]["pkg"][0]
107    # init_modules = find_package
108    # self.pipe, model, self.import_pkg, self.pipe_kwargs = self.factory.create_pipeline(arch_data=registry_entry.mir, init_modules=init_modules)
109
110    #     if lora is not None:
111    #         lora_arch = self.mir_db.database[series].get(lora[1])
112    #         lora_repo = next(iter(lora_arch["repo"]))  # <- user location here OR this
113    #         scheduler = self.mir_db.database[series]["[init]"].get("scheduler")
114    #         kwargs = {}
115    #         if scheduler:
116    #             sched = self.mir_db.database[scheduler]["[init]"]
117    #             scheduler_kwargs = self.mir_db.database[series]["[init]"].get("scheduler_kwargs")
118    #             kwargs = {sched: sched, scheduler_kwargs: scheduler_kwargs}
119    #         init_kwargs = lora_arch.get("init_kwargs")
120    #         if lora:
121    #             self.pipe = self.factory.add_lora(self.pipe, lora_repo=lora_repo, init_kwargs=init_kwargs, **kwargs)
122
123    #     noise_seed = seed_planter(device=self.device)
124    #     user_set = {
125    #         "output_type": "pil",
126    #     }
127    #     self.pipe_kwargs.update(user_set)
128
129    #     if ChipType.MPS[0]:
130    #         self.pipe.enable_attention_slicing()
131    #     nfo(f"Pre-generator Model {model}  Pipe {self.pipe} Arguments {self.pipe_kwargs}")  # Lora {lora_opt}
132    #     if "diffusers" in self.import_pkg:
133    #         self.pipe.to(self.device)
134    #         self.pipe = segments.add_generator(pipe=self.pipe, noise_seed=noise_seed)
135    #     else:
136    #         self.pipe = self.pipe[0]
137    #         self.pipe.to(self.device)
138    #     if "audiogen" in self.import_pkg:
139    #         self.pipe_kwargs.update({"sample_rate": self.pipe.config.sampling_rate})
140    #     elif "parler_tts" in self.import_pkg:
141    #         self.pipe_kwargs.update({"sampling_rate": self.pipe.config.sampling_rate})
142
143
144async def ready_predictor(
145    registry_entry: RegistryEntry,
146    async_stream: bool = True,
147    dspy_stream: bool = True,
148    max_workers: int = 8,
149    cache: bool = True,
150):
151    from zodiac.providers.constants import ChipType
152
153    if registry_entry.cuetype == CueType.HUB:
154        device = getattr(ChipType, next(iter(ChipType._show_ready())), "CPU")
155
156    else:
157        lm_kwargs = {"async_max_workers": max_workers, "cache": cache}
158        lm_model = dspy.LM(
159            registry_entry.model,
160            **registry_entry.api_kwargs,
161            **lm_kwargs,
162        )
163    stream_listeners = [dspy.streaming.StreamListener(signature_field_name="answer")]
164    context_kwargs = {"lm": lm_model}  # , "adapter": dspy.ChatAdapter()}
165    predictor_kwargs = {
166        "status_message_provider": StreamActivity(),
167        "async_streaming": async_stream,
168        "include_final_prediction_in_output_stream": False,
169    }
170    if dspy_stream:
171        predictor_kwargs["stream_listeners"] = stream_listeners
172
173    return context_kwargs, predictor_kwargs
174
175
176# import sounddevice as sd

class StreamActivity(dspy.streaming.messages.StatusMessageProvider): View Source

13class StreamActivity(dspy.streaming.StatusMessageProvider):
14    def lm_start_status_message(self, instance, inputs):
15        return "Processing..."
16
17    def module_start_status_message(self, instance, inputs):
18        return "Module started."
19
20    def lm_end_status_message(self, outputs):
21        return "Done."
22
23    def tool_start_status_message(self, instance, inputs):
24        return "Tool started..."
25
26    def tool_end_status_message(self, outputs):
27        return "Tool finished."

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())

def lm_start_status_message(self, instance, inputs): View Source

14    def lm_start_status_message(self, instance, inputs):
15        return "Processing..."

Status message before a dspy.LM is called.

def module_start_status_message(self, instance, inputs): View Source

17    def module_start_status_message(self, instance, inputs):
18        return "Module started."

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

def lm_end_status_message(self, outputs): View Source

20    def lm_end_status_message(self, outputs):
21        return "Done."

Status message after a dspy.LM is called.

def tool_start_status_message(self, instance, inputs): View Source

23    def tool_start_status_message(self, instance, inputs):
24        return "Tool started..."

Status message before a dspy.Tool is called.

def tool_end_status_message(self, outputs): View Source

26    def tool_end_status_message(self, outputs):
27        return "Tool finished."

Status message after a dspy.Tool is called.

class QATask(dspy.signatures.signature.Signature): View Source

30class QATask(dspy.Signature):
31    """Reply with short responses within 60-90 word/10k character code limits"""
32
33    question: str = dspy.InputField(desc="The question to respond to")
34    answer = dspy.OutputField(desc="Often between 60 and 90 words and limited to 10000 character code blocks")

Reply with short responses within 60-90 word/10k character code limits

question: str = PydanticUndefined

answer: str = PydanticUndefined

TARGET_LANGUAGE = 'English'

class TranslateTask(dspy.signatures.signature.Signature): View Source

40class TranslateTask(dspy.Signature):
41    f"""Translate from a language to {TARGET_LANGUAGE}"""
42
43    message: dspy.Image | dspy.Audio | str = dspy.InputField(desc="The input to translate.")
44    translation: dspy.Image | dspy.Audio | str = dspy.OutputField(desc="A translation of the input")

Given the fields message, produce the fields translation.

message: dspy.adapters.types.image.Image | dspy.adapters.types.audio.Audio | str = PydanticUndefined

translation: dspy.adapters.types.image.Image | dspy.adapters.types.audio.Audio | str = PydanticUndefined

class VisionTask(dspy.signatures.signature.Signature): View Source

47class VisionTask(dspy.Signature):
48    """Describe the image in detail."""
49
50    image: dspy.Image = dspy.InputField(desc="An image")
51    description: str = dspy.OutputField(desc="A detailed description of the image.")

Describe the image in detail.

image: dspy.adapters.types.image.Image = PydanticUndefined

description: str = PydanticUndefined

class TranscribeTask(dspy.signatures.signature.Signature): View Source

54class TranscribeTask(dspy.Signature):
55    """Transcribe spoken words into text"""
56
57    message: dspy.Audio = dspy.InputField(desc="The speech to transcribe.")
58    answer: str = dspy.OutputField(desc="A transcript of the recorded speech")

Transcribe spoken words into text

message: dspy.adapters.types.audio.Audio = PydanticUndefined

answer: str = PydanticUndefined

class GenerativeImageTask(dspy.signatures.signature.Signature): View Source

61class GenerativeImageTask(dspy.Signature):
62    message: str = dspy.InputField(desc="Description of the image to generate")
63    image: dspy.Image = dspy.OutputField(desc="An image matching the description")

Given the fields message, produce the fields image.

message: str = PydanticUndefined

image: dspy.adapters.types.image.Image = PydanticUndefined

class GenerativeAudioTask(dspy.signatures.signature.Signature): View Source

66class GenerativeAudioTask(dspy.Signature):
67    message: str = dspy.InputField(desc="Description of the audio to generate")
68    audio: dspy.Audio = dspy.OutputField(desc="An audio file matching the description")

Given the fields message, produce the fields audio.

message: str = PydanticUndefined

audio: dspy.adapters.types.audio.Audio = PydanticUndefined

class QuestionAnswer(dspy.primitives.module.Module): View Source

75class QuestionAnswer(dspy.Module):
76    def __init__(self):
77        super().__init__()
78        self.predict = dspy.Predict(QATask)
79
80    def forward(self, question, **kwargs):
81        self.predict(question=question, **kwargs)
82        return self.predict(question=question, **kwargs)

predict

def forward(self, question, **kwargs): View Source

80    def forward(self, question, **kwargs):
81        self.predict(question=question, **kwargs)
82        return self.predict(question=question, **kwargs)

class Predictor(dspy.primitives.module.Module): View Source

 85class Predictor(dspy.Module):
 86    def __init__(self):
 87        super().__init__
 88        self.program = dspy.Predict(signature=QATask)
 89
 90    def __call__(self, question: str):
 91        from litellm.exceptions import APIConnectionError
 92        from litellm.llms.ollama.common_utils import OllamaError
 93        from httpx import ConnectError
 94        from dspy.utils.exceptions import AdapterParseError
 95        from aiohttp.client_exceptions import ClientConnectorError
 96
 97        try:
 98            return self.program(question=question)
 99        except (ClientConnectorError, ConnectError, AdapterParseError, APIConnectionError, OllamaError, OSError):
100            pass
101
102    # from nnll.tensor_pipe import segments
103    # from nnll.configure.init_gpu import seed_planter
104    # mir_db = MIR_DB.database
105    # mir_arch = registry_entries.mir
106    # arch_data = mir_db[mir_arch[0][mir_arch[1]]]
107    # pkg_data = mir_db[mir_arch[0]]["pkg"][0]
108    # init_modules = find_package
109    # self.pipe, model, self.import_pkg, self.pipe_kwargs = self.factory.create_pipeline(arch_data=registry_entry.mir, init_modules=init_modules)
110
111    #     if lora is not None:
112    #         lora_arch = self.mir_db.database[series].get(lora[1])
113    #         lora_repo = next(iter(lora_arch["repo"]))  # <- user location here OR this
114    #         scheduler = self.mir_db.database[series]["[init]"].get("scheduler")
115    #         kwargs = {}
116    #         if scheduler:
117    #             sched = self.mir_db.database[scheduler]["[init]"]
118    #             scheduler_kwargs = self.mir_db.database[series]["[init]"].get("scheduler_kwargs")
119    #             kwargs = {sched: sched, scheduler_kwargs: scheduler_kwargs}
120    #         init_kwargs = lora_arch.get("init_kwargs")
121    #         if lora:
122    #             self.pipe = self.factory.add_lora(self.pipe, lora_repo=lora_repo, init_kwargs=init_kwargs, **kwargs)
123
124    #     noise_seed = seed_planter(device=self.device)
125    #     user_set = {
126    #         "output_type": "pil",
127    #     }
128    #     self.pipe_kwargs.update(user_set)
129
130    #     if ChipType.MPS[0]:
131    #         self.pipe.enable_attention_slicing()
132    #     nfo(f"Pre-generator Model {model}  Pipe {self.pipe} Arguments {self.pipe_kwargs}")  # Lora {lora_opt}
133    #     if "diffusers" in self.import_pkg:
134    #         self.pipe.to(self.device)
135    #         self.pipe = segments.add_generator(pipe=self.pipe, noise_seed=noise_seed)
136    #     else:
137    #         self.pipe = self.pipe[0]
138    #         self.pipe.to(self.device)
139    #     if "audiogen" in self.import_pkg:
140    #         self.pipe_kwargs.update({"sample_rate": self.pipe.config.sampling_rate})
141    #     elif "parler_tts" in self.import_pkg:
142    #         self.pipe_kwargs.update({"sampling_rate": self.pipe.config.sampling_rate})

program

async def ready_predictor( registry_entry: zodiac.providers.registry_entry.RegistryEntry, async_stream: bool = True, dspy_stream: bool = True, max_workers: int = 8, cache: bool = True): View Source

145async def ready_predictor(
146    registry_entry: RegistryEntry,
147    async_stream: bool = True,
148    dspy_stream: bool = True,
149    max_workers: int = 8,
150    cache: bool = True,
151):
152    from zodiac.providers.constants import ChipType
153
154    if registry_entry.cuetype == CueType.HUB:
155        device = getattr(ChipType, next(iter(ChipType._show_ready())), "CPU")
156
157    else:
158        lm_kwargs = {"async_max_workers": max_workers, "cache": cache}
159        lm_model = dspy.LM(
160            registry_entry.model,
161            **registry_entry.api_kwargs,
162            **lm_kwargs,
163        )
164    stream_listeners = [dspy.streaming.StreamListener(signature_field_name="answer")]
165    context_kwargs = {"lm": lm_model}  # , "adapter": dspy.ChatAdapter()}
166    predictor_kwargs = {
167        "status_message_provider": StreamActivity(),
168        "async_streaming": async_stream,
169        "include_final_prediction_in_output_stream": False,
170    }
171    if dspy_stream:
172        predictor_kwargs["stream_listeners"] = stream_listeners
173
174    return context_kwargs, predictor_kwargs