Is that possible to bring query pipe line 'qp' in to workflow?

[mraguth]

Here my objective is bring my pandas query pipe line 'qp' in to work flow. Is that possible? All I see in workflow examples are 'engine'
How I replace engine with query pipe line?

    @step
    async def naive_rag(
        self, ctx: Context, ev: NaiveRAGEvent
    ) -> ResponseEvent:
        index = ctx.data["index"]
        engine = index.as_query_engine(similarity_top_k=5)
        response = engine.query(ev.query)
        print("Naive response:", response)
        return ResponseEvent(
            query=ev.query, source="Naive", response=str(response)
        )

My query pipe line example:

qp = QP(
    modules={
        "input": InputComponent(),
        "pandas_prompt": pandas_prompt,
        "llm1": llm,
        "pandas_output_parser": pandas_output_parser,
        "response_synthesis_prompt": response_synthesis_prompt,
        "llm2": llm,
    },
    verbose=True,
)
qp.add_chain(["input", "pandas_prompt", "llm1", "pandas_output_parser"])
qp.add_links(
    [
        Link("input", "response_synthesis_prompt", dest_key="query_str"),
        Link(
            "llm1", "response_synthesis_prompt", dest_key="pandas_instructions"
        ),
        Link(
            "pandas_output_parser",
            "response_synthesis_prompt",
            dest_key="pandas_output",
        ),
    ]
)
# add link from response synthesis prompt to llm2
qp.add_link("response_synthesis_prompt", "llm2")

[dosubot[bot]]

Yes, it is possible to replace the 'engine' with your pandas query pipeline 'qp' in the workflow. Here is an example of how you can integrate your pandas query pipeline into the workflow:

# First, install the necessary dependencies
!pip install llama-index-core llama-index-llms-openai llama-index-embeddings-openai llama-index-readers-file llama-index-utils-workflow

# Import the required modules
import os, json
from llama_index.core import (
    SimpleDirectoryReader,
    VectorStoreIndex,
    StorageContext,
    load_index_from_storage,
)
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.workflow import (
    step,
    Context,
    Workflow,
    Event,
    StartEvent,
    StopEvent,
)
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI
from llama_index.utils.workflow import draw_all_possible_flows

# Define your pandas query pipeline 'qp'
qp = QP(
    modules={
        "input": InputComponent(),
        "pandas_prompt": pandas_prompt,
        "llm1": llm,
        "pandas_output_parser": pandas_output_parser,
        "response_synthesis_prompt": response_synthesis_prompt,
        "llm2": llm,
    },
    verbose=True,
)
qp.add_chain(["input", "pandas_prompt", "llm1", "pandas_output_parser"])
qp.add_links(
    [
        Link("input", "response_synthesis_prompt", dest_key="query_str"),
        Link(
            "llm1", "response_synthesis_prompt", dest_key="pandas_instructions"
        ),
        Link(
            "pandas_output_parser",
            "response_synthesis_prompt",
            dest_key="pandas_output",
        ),
    ]
)
# add link from response synthesis prompt to llm2
qp.add_link("response_synthesis_prompt", "llm2")

# Example workflow step using 'qp'
@step
async def naive_rag(ctx: Context, ev: NaiveRAGEvent) -> ResponseEvent:
    query = ev.query
    result = qp.execute(query)  # Assuming 'qp' has an 'execute' method
    print("Naive response:", result)
    return ResponseEvent(
        query=ev.query, source="Naive", response=str(result)
    )

# Define the workflow
workflow = Workflow(
    steps=[
        StartEvent(),
        naive_rag,
        StopEvent()
    ]
)

# Visualize the workflow
draw_all_possible_flows(workflow, filename="pandas_query_pipeline_workflow.html")

This example demonstrates how to integrate your pandas query pipeline into a workflow, replacing the default query engine with your custom pipeline ^[1][2].

To continue talking to Dosu, mention @dosu.

[mraguth]

@dosu....I am getting this error at the end. AttributeError: 'QueryPipeline' object has no attribute 'execute' Please help

[dosubot[bot]]

The error you're encountering, AttributeError: 'QueryPipeline' object has no attribute 'execute', indicates that the QueryPipeline class does not have an execute method. Instead, you should use the run method to execute the pipeline.

Here's how you can modify your workflow step to use the run method:

# Example workflow step using 'qp'
@step
async def naive_rag(ctx: Context, ev: NaiveRAGEvent) -> ResponseEvent:
    query = ev.query
    result = qp.run(input=query)  # Use the 'run' method instead of 'execute'
    print("Naive response:", result)
    return ResponseEvent(
        query=ev.query, source="Naive", response=str(result)
    )

# Define the workflow
workflow = Workflow(
    steps=[
        StartEvent(),
        naive_rag,
        StopEvent()
    ]
)

# Visualize the workflow
draw_all_possible_flows(workflow, filename="pandas_query_pipeline_workflow.html")

This should resolve the AttributeError and allow your pandas query pipeline to run correctly within the workflow ^[1].

To continue talking to Dosu, mention @dosu.

[mraguth]

@dosu... really great answer. Thank you