RefineFlow 0.0.3

Create GenAI flows to handle dynamic prompt creation.

RefineFlow

This is the first draft of the open source project RefineFlow, a passion project defined by my delirious hate of agents and their unpredictability. Whenever a GenAI use case is proposed to me, other people get bored and always assume that there are only 3 possible ways to develop a GenAI solution without fine-tuning: you either use simple prompting, you upgrade to a Rag system if you require specific knowledge, or, if the task is extremely nuanced, you use Agents to solve all your problems. This approach, while completely ignoring the fact that these 3 methods could all be needed with a specific order to solve your use case, completely stumps development on new ways to utilize LLMs. RefineFlow proposes a fourth way, another approach to the development of GenAI applications.

General Usage

RefineFlow is a new Agent-like LLM workflow to improve your RAG systems. RefineFlow provides a flexible structure to extract information in different scopes while maintaining low answering latency. RefineFlow is composed of two different but harmonious parts: Refiners... and Flows! The Refiner is the smallest component in this framework, and it's the object that actually satisfies a user's request. The Flow is conceptually related to the idea of an orchestrator, the Flow handles the refiners output, rewrites a prompt depending on the activated refiners, and finally answers to the user using all the information retrived. To ensure low latency, a Flow parallelizes all refiners in its scope.

Refiners

General characteristics

The concept of Refiner in RefineFlow is similar to the concept of an agent, but with a different activation. The refiner is composed of 3 layers: the input layer, the execution layer, and the output layer. While an Agent chooses a set of action depending on the user's request, a layer of the Refiner is always initially activated by a user's request (input layer), this layer is crucial to the correct Refiner functioning, since this layer decides whether the Refiner should effectively handle the user's request, based on the specific instructions of the user. The main objective of the input layer in specific is to output a query, a small piece of information that the user wants to utilize in the later layers. If the refiner is triggered by the input layer, this layer will pass on information to the execution layer, which, in its standard form, just executes a custom RAG pipeline defined by the user, where the input is the keyword extracted by the input layer, and the output is a list of most relevant chunks. Finally, the execution layer passes the output to the output layer, where an instruction written in natural language is combined with the user's request. This is the final output of the refiners, a prompt ready to be injected into the more general request. Each Refiner should handle just a small part of each request, in order to be as specific as possible.

Strenghts of Refiners

The Refiner is the real protagonist of this framework, and it's similar to the concept of Agent tools, but with some notable key differences.

• A Refiner is always indipendently activated, and it indipendetly decides whether to stay activate or terminate the execution early
• A Refiner does not immediately sole a user's requests, but prepares a custom prompt that will later be injected.

Flows

General characteristics

The concept of Flows is heavily inspired by the common agent orchestrator seen throughout the internet. A Flow is defined by a list of refiners in scope for an activity, an initial prompt, and a final prompt. Whenever the flow recieved a requests it activates the refiners in scope and (differently from an agent) parallelizes all the refiner's computations in order to mantain the lowest possible latency, Whenever all refiners have terminated (either because they were stopped at the input layer or they correctly finish their execution), the Flow takes the initial prompt, and adds all the outputs of the refiners, and finishes the job by appending the final prompt recieved in input. The unified and complete prompt is then sent to an LLM in order to recieve an answer.

A point on the parallelization

The innovative idea that Flows bring is the built-in parallelization of tasks, which for simple agent structures needs to be coded. This approach ensures low latency, but is of course a limitation of the workflow itself: each refiner needs to be an entirely indipendent and self-contained unit, and not all use cases may benefit from such an architecture

Ask

Ask are the custom functions you sholuld create to empower your refiners with answering and rag capabilities. RefineFlow is completely separated by your LLM implementation of choice and lets you define freely the code to interact with your APIs or local LLMs. There are, of course, some built-in asks method to interact with the OpenAI API (ask_openai) and to make a very simple RAG-like search (ask_chunks_w_simple_dict). The usage of these asks in combination with refiners will be further discussed in the [Code example](#Code example).

Usage

RefineFlow, like any other workflow, has its strenght and weaknesses, it is best utilized when:

• You need a low latency solutions that has a lot of different knowledge bases
• You want to decrease final prompt size, decreasing the possibility of hallucinations in your final answer, by passing to the LLM only the necessaries informations found by the individial Refiners.

I suggest to not use RefineFlow if:

• You need a large amount of knowledge base to solve each specific task, since each prompt gets appendend, the final prompt may be too large to be perfectly understood by a single LLM call.
• Parallelization is a problem and you need a more sequential solution.

Code example

Firstly i will need to define a specific function for a Refiner, an 'ask' function, which is the function that actually implements the LLM of choice API implementation, and needs to take in input at least a list of strings as a chat history, and another string as the system prompt. You can find all examples defined as a python test in the test folder of the package. Here below we report an example of a built-in ask for the OpenAI API

from typing import List, Optional
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()

def ask_openai(chat_history:List[str], prompt:str, model_name:Optional[str] = None, dotenv_api_key_name: Optional[str]=None):
    """
    Pre-built ask function implementing a GPT request.
    """
    if model_name is None:
        model_name = "gpt-3.5-turbo"
    if dotenv_api_key_name is None:
        dotenv_api_key_name = "OPENAI_API_KEY"
    client = OpenAI(api_key=os.environ.get(dotenv_api_key_name))
    messages: list[dict[str, str]] = []
    messages.append({"role": "system", "content": prompt})
    for i, message in enumerate(chat_history):
        role = "user" if i % 2 == 0 else "model"
        messages.append({"role": role, "content": message})
    request_params: dict[str, Any] = {
        "model": model_name,
        "messages": messages,
    }
    response = client.chat.completions.create(**request_params)

    return response.choices[0].message.content.strip()

This ask function follows the OpenAI API and is already implemented in RefineFlow as an ask, a function that should handle the api choice and implementation. Note that the ask function, while needing a list of strings and a string, allows the possibility to define other inputs, as we will see shortly. This structure has been chosen to de-couple LLM definition to this python package, but of course it's required a set of minimal standardization to ensure a smooth communication between your custom definitions and the structure of the workflow.

Now that the concept of 'ask' is clear, we will now define theoretically a RagRefiner, we will look at a more realistic version later:

rrefiner = RagRefiner(
ask_llm = ask_openai,
ask_chunks = ...,
)

The RagRefiner takes in input only Callable objects wich correspond to two ask functions, one seen previously as the ask_openai function and another new ask function which tells the system how to extract the chunk. In a normal rag use case you will implement here you custom function that takes as input a string, and in output gives a list of strings corresponding to the top_k most similar chunks. For this example, and to show the flexibility of the framework, i will deifne a completely custom function that serializes informations:

from typing import List, Dict

def ask_chunks_w_simple_dict(query: str, dict_w_chunks:Dict[str, List[str]]):
    chunks = []
    lower_keys= [key.lower() for key in dict_w_chunks.keys()]
    if query.lower() in lower_keys:
        chunks = dict_w_chunks[query.lower()]
    else:
        chunks = ["No information found regarding this topic"]
    return chunks

Again, this example is already defined in the ask components of this package. This ask hard-codes the informations depending on the query extracted by the input layer, previously explained in the Dedicated section.

So, a more realistic version of the RagRefiner definition may look a bit like this:

from functools import partial 

knowledge_base={
    "Sport":["Sport is very tiring", "You can get injured while playing", "The best sport is Martial Arts"],
    "Videogames":["Videogames are cool!", "Videogames are a unique form of art", "My favourite videogame is Mirror's Edge"]
}
topic_refiner = RagRefiner(
ask_llm = partial(ask_openai, model_name="gpt-3.5-turbo"),
ask_chunks = partial(ask_chunks_w_simple_dict, dict_w_chunks=knowledge_base)
)

As you can see, the custom Callables can be partially defined to ensure the most customizable Refiners possible.

Now we have succsesfully created a refiner, but we still need to tell the refiner how to behave! We can tell the Refiner like this:

topic_input_prompt = """
    You are a topic classifier, you classify a sentence among these possible topics: 
    "sport", "videogames", "literature", "art"
    Whenever the user asks you something you answer only with one of the previous topics, considering which one is most fitting.
    Always answer with one of the possible topics and never add any notes or comments.
    If the user is not talking about anything remotely related to the previous topics, just answer with a "" string. 
"""
topic_refiner.set_input_prompt(topic_input_prompt)
topic_output_prompt ="""
Consider this to be your knowledge base:
{top_k_chunks}
Answer the user's question utilizing your knowledge base.
"""
topic_refiner.set_output_prompt(topic_output_prompt)

Just to have a specific example, we will define another special refiner, which will not utilize a knowledge base, in fact we can define it as:

angry_refiner = RagRefiner(
    ask_llm = partial(ask_openai, model_name="gpt-3.5-turbo"),
    ask_chunks = None,
    skip_execution_layer = True
)
angry_input_prompt = """
    You are a sport detector, you detect wheter a user is talking about anything related to sports.
    Whenever the user asks you something you answer with only the word sport if the user's request is related to sports.
    Answer with the word sport also if the subject of the sentence is lightly related to sports.
    If the user is not talking about anything remotely related to sports, just answer with a "" string. 
"""
angry_refiner.set_input_prompt(angry_input_prompt)
angry_output_prompt ="""
You really hate sports. Show this in your answer.
"""
angry_refiner.set_output_prompt(angry_output_prompt)

As you can see this refiners just adds instruction regarding how to behave in a specific case.

Now let's define a Flow, now that ve have correctly built everything we cna easily define a Flow as:

flow = Flow(ask_openai, [topic_refiner, angry_refiner])

start_prompt = """
Your name is Marco, you are a helpful assistant.
"""

end_prompt = ""
answer_sport = flow.run(["Hello! what do you know about rugby?"], start_prompt, end_prompt)
answer_videogames = flow.run(["Hello! what do you know about Mirror's Edge?"], start_prompt, end_prompt)
answer_both = flow.run(["Hello! Do you play fifa?"], start_prompt, end_prompt)
print(f"\n[Experiment]:\nanswer_sport: {answer_sport}\nanswer_videogames:{answer_videogames}\nanswer_both:{answer_both}\n")

You are now ready to run the script and see the results for yourself.

What now?

After having run the previous example you may tilt your head a bit. The topic_refiner probably outputted one among "sports" and "videogames", which would both be correct considering the user's request, and you may want to define them both as your input layer output.

While this could be a great feature to implement as part of the package, this functionality is currently still in TODO, but this doesn't stop you from coding it yourself! You can define any custom refiner you'd like starting from the basic refiners, an example is provided below:

import ast

class CoolRagRefiner(RagRefiner):

    def input_layer(self):
        self.input_prompt="""
            You are a topic classifier, you classify a sentence among these possible topics: 
            "sport", "videogames", "literature", "art"
            Whenever the user asks you something you answer only with a python list containing only the previous topics, considering which one is most fitting.
            Always answer with a python list of possible topics and never add any notes or comments.
            If the user is not talking about anything remotely related to the previous topics, just answer with a [] list. 
        """
        key = self.ask_llm([self.query], self.input_prompt)
        key = key.replace("''", "").replace('""', "")
        key_list = ast.literal_eval(key)
        self.input_key= key_list
        return key_list

    def output_layer(self):
        answer_dict = {
            "sport": "Lebron James",
            "videogames": "Faker", 
            "literature": "Herman Hesse",
            "art": "Leonardo Da Vinci"
        }
        answer = ""
        for key in self.input_key:
            answer += answer_dict[key] + ", "
        answer = answer[:-2]
        output = f"Here is a list of important VIP(s) in the topic(s) {self.input_key}: {answer}. Always say this in you answer."
        return output

In this very simple example i have redefined a rag refiner with a very similar structure to the ask_chunks_w_simple_dict normal rag refiner. This freedom allows you to still use the refiner structure but customize all layers however you want, for example by allowing multiple keys to be retrieved from the input layer and then injected in the output_layer. I can easily redefine a flow following the previous instructions:

coolrefiner = CoolRagRefiner(        
    ask_llm = partial(ask_openai, model_name="gpt-3.5-turbo"),
    ask_chunks = None,
    skip_execution_layer = True
)

flow = Flow(ask_openai, [coolrefiner])


answer_vip = flow.run(["Do you like videogames or sports??"])

print(f"\n[Experiment]:\nanswer_vip: {answer_vip}\n")