Async și streaming în DSPy: acall, asyncify, streamify în DSPy & Programare Algoritmică LLM: explicație și exemple

Modulele predefinite au acall(), cu aceeași interfață ca apelul sincron. Un modul propriu implementează aforward(); dacă lipsește, acall() rulează forward() într-un fir de lucru. Uneltele async se apelează cu Tool.acall(), iar ReAct.acall() execută uneltele asincron. Ghidul oficial recomandă codul sincron pentru prototipuri și cel async pentru servicii cu multe cereri simultane sau unelte care există doar în variantă async.

dspy.asyncify(program) transformă un program sincron într-unul care se poate aștepta cu await; rularea are loc într-un fir separat, cu cel mult async_max_workers fire (implicit 8), setat prin dspy.configure. Tutorialul de deployment îl folosește într-o aplicație FastAPI.

dspy.streamify(program, stream_listeners=[...]) întoarce o funcție care produce un generator async. Pentru fiecare câmp ascultat (doar câmpuri str) primești obiecte StreamResponse cu predict_name, signature_field_name și chunk, iar la final un dspy.Prediction. Un StatusMessageProvider adaugă mesaje StatusMessage la pornirea și oprirea modelelor, modulelor și uneltelor. Când răspunsul vine din cache, nu se trimit tokeni, ci doar predicția finală.

Concepte cheie și caracteristici

Concept / Parametru	Descriere practică
`Module.acall(**kwargs)`	Apelul async al unui modul.
`Module.aforward(**kwargs)`	Metoda de implementat într-un modul propriu async.
`dspy.asyncify(program)`	Program sincron rulat în fire separate, care poate fi așteptat cu `await`.
`async_max_workers`	Setare `dspy.configure`: numărul maxim de fire pentru asyncify; implicit 8.
`dspy.streamify(program, status_message_provider=None, stream_listeners=None, include_final_prediction_in_output_stream=True, ...)`	Întoarce o funcție care produce un generator async de `StreamResponse`, `StatusMessage` și `Prediction`.
`StreamListener(signature_field_name, predict=None, predict_name=None, allow_reuse=False)`	Ascultă un câmp. Implicit se închide după prima transmisie; `allow_reuse=True` pentru module apelate în buclă (ReAct).
`StatusMessageProvider`	Metode de suprascris: `lm_start_status_message`, `lm_end_status_message`, `module_start_status_message`, `module_end_status_message`, `tool_start_status_message`, `tool_end_status_message`.
`dspy.streaming.streaming_response`	Funcție ajutătoare care transformă fluxul în server-sent events.

Exemple practice și scenarii reale

Exemplul 1: Apel async și modul propriu async

import asyncio
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-6-luna"))

class RezumaSimplu(dspy.Module):
    def __init__(self):
        super().__init__()
        self.raspunde = dspy.ChainOfThought("question -> answer")
        self.simplifica = dspy.ChainOfThought("answer -> simplified_answer")

    async def aforward(self, question):
        rasp = await self.raspunde.acall(question=question)
        return await self.simplifica.acall(answer=rasp.answer)

async def main():
    rez = await RezumaSimplu().acall(question="Ce este un hash criptografic?")
    print(rez.simplified_answer)

asyncio.run(main())

Exemplul 2: Streaming pentru câmpul answer

import asyncio
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-6-luna", cache=False))
predict = dspy.Predict("question -> answer")

stream_predict = dspy.streamify(
    predict,
    stream_listeners=[dspy.streaming.StreamListener(signature_field_name="answer")],
)

async def citeste():
    final = None
    async for chunk in stream_predict(question="Explică pe scurt DNS."):
        if isinstance(chunk, dspy.streaming.StreamResponse):
            print(chunk.chunk, end="", flush=True)
        elif isinstance(chunk, dspy.Prediction):
            final = chunk
    return final

rez = asyncio.run(citeste())

Exemplul 3: Mesaje de stare pentru unelte

class Stare(dspy.streaming.StatusMessageProvider):
    def tool_start_status_message(self, instance, inputs):
        return f"Apelez unealta {instance.name}..."

    def tool_end_status_message(self, outputs):
        return "Unealta a terminat."

stream_agent = dspy.streamify(agent, status_message_provider=Stare())

Greșeli frecvente și capcane

Iterezi rezultatul lui streamify într-o funcție sincronă. Generatorul este async: folosește async for în asyncio.run(...) sau direct într-un notebook, care are deja o buclă de evenimente.
Testezi streaming-ul cu cache-ul activ. La un răspuns din cache nu primești tokeni, doar predicția finală; folosește cache=False pe LM în teste.
Asculți un câmp din ReAct fără allow_reuse=True. Listener-ul se închide după prima transmisie și nu mai vezi pașii următori.
Presupui că asyncify face apelurile non-blocante. Programul rulează tot sincron, într-un fir separat, limitat de async_max_workers.

`Async și streaming în DSPy: acall, asyncify, streamify` în DSPy & Programare Algoritmică LLM