[Projeto] Contornando o problema da depreciação do query_pipeline

Oi pessoal, eu vi que o Yuri postou aqui no fórum uma solução para o fato do query_pipeline usado em classe estar depreciado. Queria contribuir também. Utilizei o Claude para me ajudar em uma solução e o teste foi satisfatório e praticamente idêntico ao usado pela professora na aula:

# Importa as classes principais da API de Workflows do llama_index:
# Workflow: classe base que você deve herdar para criar seu próprio pipeline
# step: decorador que transforma um método comum em uma etapa do workflow
# Event: classe base para criar eventos customizados que trafegam entre os steps
# StartEvent: evento especial que carrega os dados iniciais (o que você passa em w.run(...))
# StopEvent: evento especial que, ao ser retornado, encerra o workflow e devolve o resultado final
from llama_index.core.workflow import Workflow, step, Event, StartEvent, StopEvent

# Define um evento customizado que vai carregar os dados entre o 1º e o 2º step. Ele é um Pydantic model — cada atributo declarado abaixo é um campo obrigatório que precisa ser preenchido quando o evento for criado.
class PandasEvent(Event):
    query_str: str            # a pergunta original do usuário, em linguagem natural
    pandas_instructions: str  # o código Pandas gerado pelo LLM (saída do llm1)
    pandas_output: str        # o resultado da execução desse código sobre o df (saída do pandas_output_parser)

# Cria a classe do Workflow, herdando de Workflow. Cada método decorado com @step é uma "etapa" — o llama_index infere automaticamente as ligações entre os steps olhando o tipo do parâmetro de entrada e o tipo de retorno de cada um (não precisa mais de add_link!).
class PandasRAGWorkflow(Workflow):

    # Primeiro step: recebe o StartEvent (equivalente ao antigo "input"/InputComponent) e retorna um PandasEvent — por isso o llama_index sabe que esse step  deve rodar assim que o workflow começa.
    @step
    async def gerar_instrucoes_pandas(self, ev: StartEvent) -> PandasEvent:
        # Extrai o texto da pergunta que foi passado em w.run(query_str="...")
        query_str = ev.query_str

        # Preenche o template pandas_prompt com a pergunta do usuário (equivalente ao antigo Link "input" -> "pandas_prompt")
        prompt_formatado = pandas_prompt.format(query_str=query_str)

        # Envia o prompt formatado para o LLM (equivalente ao antigo módulo "llm1") "await" porque acomplete é assíncrono — o workflow inteiro roda em modo async
        resposta_llm1 = await llm.acomplete(prompt_formatado)

        # Converte a resposta do LLM (objeto CompletionResponse) para string simples, que é o código Pandas gerado
        pandas_instructions = str(resposta_llm1)

        # Executa esse código Pandas sobre o dataframe real usando o parser (equivalente ao antigo módulo "pandas_output_parser")
        pandas_output = pandas_output_parser.parse(pandas_instructions)

        # Empacota os três dados que o próximo step vai precisar e os retorna como um PandasEvent — isso "dispara" automaticamente o próximo step que aceita PandasEvent como entrada
        return PandasEvent(
            query_str=query_str,
            pandas_instructions=pandas_instructions,
            pandas_output=pandas_output,
        )

    # Segundo step: recebe o PandasEvent retornado pelo step anterior e retorna um StopEvent — por isso o llama_index sabe que esse é o último step
    @step
    async def sintetizar_resposta(self, ev: PandasEvent) -> StopEvent:
        # Preenche o template response_synthesis_prompt com os três dados recebidos do step anterior (equivalente aos três antigos Links que apontavam para "response_synthesis_prompt")
        prompt_formatado = response_synthesis_prompt.format(
            query_str=ev.query_str,
            pandas_instructions=ev.pandas_instructions,
            pandas_output=ev.pandas_output,
        )

        # Envia o prompt final para o LLM (equivalente ao antigo módulo "llm2")
        resposta_llm2 = await llm.acomplete(prompt_formatado)

        # Retorna um StopEvent com o resultado final — isso encerra o workflow e o valor passado em "result" é o que você recebe de volta em w.run(...)
        return StopEvent(result=str(resposta_llm2))

# Instancia o workflow.
# timeout=120 -> tempo máximo (em segundos) que o workflow pode rodar antes de dar erro
# verbose=True -> imprime no console qual step está rodando, útil para debug
w = PandasRAGWorkflow(verbose=True, timeout=120)

Testando:

response = await w.run(query_str = 'Qual é média de gasto por cada tipo de cliente?')
texto_formatado = textwrap.fill(response, width = 100)
print(texto_formatado)

A média de gasto por cada tipo de cliente é de aproximadamente 327,79 para os clientes Membros e
318,12 para os clientes Normais. Isso sugere que os clientes Membros tendem a gastar um pouco mais
em comparação com os clientes Normais.  O código utilizado foi
df.groupby('tipo_cliente')['total'].mean()