Implementing Advanced RAG using LlamaIndex Workflow and Groq
Introduction
AI applications consist of various tasks executed by distinct components. To streamline the work of AI engineers, open-source frameworks offer user-friendly abstractions for essential elements such as data loaders, large language models (LLMs), vector databases, and rerankers, extending to external services. These frameworks are also exploring the most effective ways to coordinate these components, focusing on what is most intuitive and efficient for developers to create cohesive AI systems.
Among the orchestration methods being considered are chains and pipelines, both based on the Directed Acyclic Graph (DAG) model.
Earlier this year(2024), LlamaIndex introduced our Query Pipelines, a declarative API designed to facilitate the orchestration of various query workflows for applications like question answering, structured data extraction, and automated processes. However, as we attempted to enhance this by integrating cycles for more intricate workflows, we encountered several challenges. This prompted to reconsider the suitability of the DAG model for agentic environments and explore alternative solutions within our framework.
This lead to the exploration of Workflows. Workflows is a new way to implement Agents. It follows an Event Driven Architecture.
What is the Limitation of Graph Based Framework ?
Graph/DAG based approaches like LangGraph (and our previous query pipelines) have to do a ton of heavy lifting under the hood to figure out “what runs next, what’s the inputs, etc.”. All this logic comes with tons of edge cases, which was becoming very evident in our increasingly complicated query pipeline codebase.
In the context of Directed Acyclic Graphs (DAGs), the acyclic nature means there are no loops, which can be limiting in increasingly agentic AI applications. Developers need the ability to implement self-correction mechanisms when components yield poor results. Even without incorporating cycles, query pipelines faced several challenges:
- Debugging Difficulties: Identifying issues became cumbersome.
- Obscured Execution: The execution flow of components was not transparent.
- Complex Orchestration: The orchestrator grew overly complicated, needing to manage numerous edge cases.
- Readability Issues: Complex pipelines were hard to comprehend.
When cycles were added to the query pipelines, these developer experience issues intensified. Significant pain points were encountered , such as:
- Cumbersome Logic: Core orchestration logic, like if-else and while loops, cluttered the graph’s edges, making them verbose.
- Edge Case Handling: Managing optional and default values became problematic as it was unclear if parameters would be passed from upstream nodes.
- Natural Flow Disruption: Defining graphs with cycles felt unnatural for developers creating agent-based systems. The requirement for explicit incoming and outgoing edges for “agent” nodes led to verbose communication patterns with other nodes.
To overcome the above issues LlamaIndex proposed a new paradigm shift in design leading to even driven architecture called Workflows.
What is a Workflow ?
It is an event driven step based way to control the execution flow of an application. Here the application is divided into sections called steps which is triggered by events.
Steps are literally python functions. They could be either single line of code or multiple lines of complex code.
An event literally is a change of state or anything which we could notice, observe or record.
By combining steps and events we can create arbitrarily complex flows that encapsulate logic and make our application more maintainable and easier to understand.
In summary, Workflows (and event-driven programming in general) offer a more robust solution. The code is much more simple under the hood, because now it’s entirely in the user’s hands what steps run next, how a user debug things, and removes a lot of the “black box” issues that other approaches like our query pipelines have.
Value Proposition of LlamaIndex Workflows over LangGraph?
LlamaIndex Workflows offer several unique features and functionalities that differentiate them from LangGraph:
- Data Connectors: Tools to ingest data from various native sources and formats, such as APIs, PDFs, and SQL databases.
- Data Indexes: Structures data in intermediate representations that are easy and performant for LLMs to consume.
- Engines: Different types of engines for natural language access to data:
- Query Engines: For question-answering interfaces (e.g., RAG pipelines).
- Chat Engines: For conversational interfaces with multi-message interactions.
4. Agents: LLM-powered knowledge workers augmented by tools, including simple helper functions and API integrations. The ReActAgent implementation allows for defining a set of tools, which can be Python functions or LlamaIndex query engines, to perform automated reasoning over data.
5. Observability/Evaluation: Integrations for rigorous experimentation, evaluation, and monitoring of applications.
6. LlamaCloud: Managed services for data parsing, ingestion, indexing, and retrieval, including LlamaParse, a state-of-the-art document parsing solution.
7. Community and Ecosystem: Strong community presence and related projects like LlamaHub for custom data connectors and create-llama for quickly scaffolding projects.
8. Integration Flexibility: Allows for both starter and customized builds. Users can start with the llama-index
package for a quick setup or use llama-index-core
to add specific integrations from LlamaHub, which offers over 300 integration packages.
9. Advanced Retrieval/Query Interface: Provides an advanced interface for feeding LLM input prompts and retrieving context and knowledge-augmented outputs.
10. Ease of Use for Beginners and Advanced Users: High-level APIs for beginners to ingest and query data in just a few lines of code, while also providing lower-level APIs for advanced users to customize and extend modules.
11. Agentic Components: Core modules capable of automated reasoning for different use cases over your data, making them essentially agents. Examples include SubQuestionQueryEngine for multi-document analysis, query transformations, routing, and LLM reranking.
12. Native OpenAIAgent: Includes an OpenAIAgent
implementation built on the OpenAI API for function calling, allowing rapid agent development.
13. Integration with Other Frameworks: Can be used as a tool within other agent frameworks like LangChain and ChatGPT, providing deep integrations and additional functionalities.
These features collectively make LlamaIndex a comprehensive framework for building context-augmented generative AI applications with LLMs.
Designing the workflow
Here we will implement an Advanced RAG system using LlamaIndex workflows
- Indexing data, creating an index
- Using that index + a query to retrieve relevant text chunks
- Rerank the text retrieved text chunks using the original query
- Synthesizing a final response
Technology Stack used for implementation
Code Implementation
- install required dependencies
pip install -qU llama-index
pip install -qU llama-index-llms-groq
pip install -qU llama-index-embeddings-huggingface
pip install -qU lama-index-utils-workflow
pip install python-dotenv
pip install pyvis
2. Create .env file to save the api key
GROQ_API_KEY = <your api key>
3. Setup the Groq_API_Key
import os
from dotenv import load_dotenv
load_dotenv() # take environment variables from .env.
os.getenv("GROQ_API_KEY")
4. Import the required dependencies
from llama_index.core import VectorStoreIndex
from llama_index.core.schema import NodeWithScore
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.core import SimpleDirectoryReader
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.core.postprocessor.llm_rerank import LLMRerank
from llama_index.core.workflow import (
Context,
Workflow,
StartEvent,
StopEvent,
step,
Event
)
from llama_index.core.workflow.utils import get_steps_from_class, get_steps_from_instance
from llama_index.llms.groq import Groq
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
5. Setup the workflow events.
To handle these steps, we need to define a few events:
- An event to pass retrieved nodes to the reranker
- An event to pass reranked nodes to the synthesizer
The other steps will use the built-in StartEvent
and StopEvent
events.
class RetrieverEvent(Event):
"""Result of running retrieval"""
nodes: list[NodeWithScore]
class RerankEvent(Event):
"""Result of running reranking on retrieved nodes"""
nodes: list[NodeWithScore]
6. Setup the Workflow
class RAGWorkflow(Workflow):
@step
async def ingest(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
"""Entry point to ingest a document, triggered by a StartEvent with `dirname`."""
dirname = ev.get("dirname")
if not dirname:
return None
documents = SimpleDirectoryReader(dirname).load_data()
index = VectorStoreIndex.from_documents(
documents=documents,
embed_model=HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"),
)
return StopEvent(result=index)
@step
async def retrieve(
self, ctx: Context, ev: StartEvent
) -> RetrieverEvent | None:
"Entry point for RAG, triggered by a StartEvent with `query`."
query = ev.get("query")
index = ev.get("index")
if not query:
return None
print(f"Query the database with: {query}")
# store the query in the global context
await ctx.set("query", query)
# get the index from the global context
if index is None:
print("Index is empty, load some documents before querying!")
return None
retriever = index.as_retriever(similarity_top_k=2)
nodes = await retriever.aretrieve(query)
print(f"Retrieved {len(nodes)} nodes.")
return RetrieverEvent(nodes=nodes)
@step
async def rerank(self, ctx: Context, ev: RetrieverEvent) -> RerankEvent:
# Rerank the nodes
ranker = LLMRerank(
choice_batch_size=5, top_n=3,
llm=Groq(model="llama-3.1-70b-versatile")
)
print(await ctx.get("query", default=None), flush=True)
new_nodes = ranker.postprocess_nodes(
ev.nodes, query_str=await ctx.get("query", default=None)
)
print(f"Reranked nodes to {len(new_nodes)}")
print(new_nodes)
return RerankEvent(nodes=new_nodes)
@step
async def synthesize(self, ctx: Context, ev: RerankEvent) -> StopEvent:
"""Return a streaming response using reranked nodes."""
llm = Groq(model="llama-3.1-70b-versatile")
summarizer = CompactAndRefine(llm=llm, streaming=True, verbose=True)
query = await ctx.get("query", default=None)
response = await summarizer.asynthesize(query, nodes=ev.nodes)
return StopEvent(result=response)
7. Check if the steps are instantiated properly
# Check if steps have __step_config attribute
workflow = RAGWorkflow()
steps = get_steps_from_class(RAGWorkflow)
if not steps:
steps = get_steps_from_instance(workflow)
print(f"steps class :{steps}")
for step_name, step_func in steps.items():
step_config = getattr(step_func, "__step_config", None)
print(f"step config :{step_config}")
if step_config is None:
print(f"Step {step_name} is missing __step_config")
Response
steps class :{'_done': <function Workflow._done at 0x000001BD6E5F3880>, 'ingest': <function RAGWorkflow.ingest at 0x000001BD07DEAB60>, 'rerank': <function RAGWorkflow.rerank at 0x000001BD07DEA160>, 'retrieve': <function RAGWorkflow.retrieve at 0x000001BD07DEA5C0>, 'synthesize': <function RAGWorkflow.synthesize at 0x000001BD07DEA0C0>}
step config :accepted_events=[<class 'llama_index.core.workflow.events.StopEvent'>] event_name='ev' return_types=[<class 'NoneType'>] context_parameter='ctx' num_workers=1 requested_services=[]
step config :accepted_events=[<class 'llama_index.core.workflow.events.StartEvent'>] event_name='ev' return_types=[<class 'llama_index.core.workflow.events.StopEvent'>] context_parameter='ctx' num_workers=1 requested_services=[]
step config :accepted_events=[<class '__main__.RetrieverEvent'>] event_name='ev' return_types=[<class '__main__.RerankEvent'>] context_parameter='ctx' num_workers=1 requested_services=[]
step config :accepted_events=[<class 'llama_index.core.workflow.events.StartEvent'>] event_name='ev' return_types=[<class '__main__.RetrieverEvent'>] context_parameter='ctx' num_workers=1 requested_services=[]
step config :accepted_events=[<class '__main__.RerankEvent'>] event_name='ev' return_types=[<class 'llama_index.core.workflow.events.StopEvent'>] context_parameter='ctx' num_workers=1 requested_services=[]
8. Invoke the workflow and visualize
import nest_asyncio
nest_asyncio.apply()
# Visualization
from llama_index.utils.workflow import draw_all_possible_flows, draw_most_recent_execution
# Draw all possible flows
draw_all_possible_flows(RAGWorkflow, filename="multi_step_workflow.html")
# Draw the most recent execution
w = RAGWorkflow()
# Ingest the documents
index = await w.run(dirname="Data")
result = await w.run(query="What is Fibromyalgia?", index=index)
async for chunk in result.async_response_gen():
print(chunk, end="", flush=True)
draw_most_recent_execution(w, filename="rag_flow_recent.html")
Response
multi_step_workflow.html
Query the database with: What is Fibromyalgia?
Retrieved 2 nodes.
What is Fibromyalgia?
Reranked nodes to 2
[NodeWithScore(node=TextNode(id_='abde4b4c-b787-4003-acf5-2f5bd05d867c', embedding=None, metadata={'page_label': '137', 'file_name': 'fibromyalgia.pdf', 'file_path': 'c:\\Users\\PLNAYAK\\Documents\\workflow\\Data\\fibromyalgia.pdf', 'file_type': 'application/pdf', 'file_size': 632664, 'creation_date': '2024-09-09', 'last_modified_date': '2024-09-09'}, excluded_embed_metadata_keys=['file_name', 'file_type', 'file_size', 'creation_date', 'last_modified_date', 'last_accessed_date'], excluded_llm_metadata_keys=['file_name', 'file_type', 'file_size', 'creation_date', 'last_modified_date', 'last_accessed_date'], relationships={<NodeRelationship.SOURCE: '1'>: RelatedNodeInfo(node_id='7b100860-f0b3-445b-b5d6-21f021d8c3c0', node_type=<ObjectType.DOCUMENT: '4'>, metadata={'page_label': '137', 'file_name': 'fibromyalgia.pdf', 'file_path': 'c:\\Users\\PLNAYAK\\Documents\\workflow\\Data\\fibromyalgia.pdf', 'file_type': 'application/pdf', 'file_size': 632664, 'creation_date': '2024-09-09', 'last_modified_date': '2024-09-09'}, hash='65d90d8fae093e6a574c784e808701ce0596d595e3e6136f7f0b4a70be8d2b57'), <NodeRelationship.NEXT: '3'>: RelatedNodeInfo(node_id='7edbd742-55e9-4559-9e9f-55d8688c6e62', node_type=<ObjectType.TEXT: '1'>, metadata={}, hash='08366d434882c9ea475468bcff6e0fe183d4192c93617ec3cf3700cf03fd5a65')}, text='February 2023 ◆ Volume 107, Number 2 www.aafp.org/afp American Family Physician 137Fibromyalgia is characterized by diffuse mus -\nculoskeletal pain, fatigue, poor sleep, and other \nsomatic symptoms.1 Chronic diffuse pain affects \n10% to 15% of adults in the general population worldwide, many of whom have fibromyalgia.\n2,3 \nApproximately 2% of people in the United States have fibromyalgia, although the prevalence var -\nies across populations and with the diagnostic criteria used.\n3 Fibromyalgia can occur in chil -\ndren and adults and is found worldwide and across cultures. Women are diagnosed more \nfrequently than men; a Scot tish survey found \nthat women are diagnosed between two and 14 times as often as men depending on the crite -\nria used.\n3,4 Changes in the diagnostic criteria over the past decade, including the elimination of specific tender points, have resulted in more patients with chronic pain meeting the criteria for fibromyalgia.\n3-5\nPathophysiology\nFibromyalgia is likely caused by disordered cen -\ntral nociceptive signal processing that leads to sensitization expressed as hyperalgesia and allo -\ndynia, which is similar to chronic pain conditions such as irritable bowel syndrome, interstitial cys -\ntitis, chronic pelvic pain, and chronic low back pain.\n6,7 Functional brain imaging suggests that \nthis aberrant processing may be attributed to an imbalance between excitatory and inhibitory neu -\nrotransmitters, particularly within the insula.\n8 \nSuggested etiologies include dysfunction of the hypothalamic-pituitary-adrenal axis and the autonomic nervous system, diffuse inflammation, glial cell activation, small fiber neuropathy, and infections such as the Epstein-Barr virus, Lyme disease, and viral hepatitis.\n9 Twin studies suggest \na genetic component may also be a factor.10Fibromyalgia: Diagn osis and Management\nBradford T. Winslow, MD, University of Colorado School of Medicine, Aurora, \nColorado; Swedi sh Family Medicine Residency, Englewood, Colorado\nCarmen Vandal, MD, and Laurel Dang, MD, Swedish Family Medicine Residency, Englewood, Colorado\n CME This clinical content conforms to AAFP criteria for \nCME. See CME Quiz on page 127.\nAuthor disclosure: No relevant financial relationships.\nPatient information: A handout on this topic, written by the \nauthors of this article, is available with the online version of \nthis article.Fibromyalgia is a chronic, centralized pain syndrome characterized by disordered processing of painful stimuli. Fibromyal -\ngia is diagnosed more frequently in women and occurs globally, affecting 2% of people in the United States. Patients with \nfibromyalgia have diffuse chronic pain, poor sleep, fatigue, cognitive dysfunc -\ntion, and mood disturbances. Comorbid conditions, such as functional somatic syndromes, psychiatric diagnoses, and rheumatologic conditions may be pres -\nent. The Fibromyalgia Rapid Screening Tool is a helpful screening method for patients with diffuse chronic pain. The American College of Rheumatology criteria or the Analgesic, Anesthetic, and Addiction Clinical Trial Translations Innovations Opportunities and Networks–American Pain Society Pain Taxonomy diagnostic criteria can diagnose fibromyalgia. Establishing the diagnosis and providing education can reassure patients and decrease unnecessary testing. A multidisciplinary approach that incorporates nonpharmacologic therapies and medications to address problematic symptoms is most effective. Patient educa -\ntion, exercise, and cognitive behavior therapy can improve pain and function. Duloxetine, milnacipran, pregabalin, and amitriptyline are potentially effective medications for fibromyalgia. Nonsteroi -\ndal anti-inflammatory drugs and opioids have not demonstrated benefits for fibromyalgia and have significant limitations. \n(Am Fam Physician\n. 2023; 107(2): 137-1 44. Copyright © 2023 American Academy of Family Physicians.)\nIllustration by Jonathan Dimes\nDownloaded from the American Family Physician website at www.aafp.org/afp. Copyright © 2023 American Academy of Family Physicians. For the private, non -\ncommercial use of one individual user of the website. All other rights reserved. Contact copyrights@aafp.org for copyright questions and/or permission requests.Downloaded from the American Family Physician website at www.aafp.org/afp. Copyright © 2023 American Academy of Family Physicians.', mimetype='text/plain', start_char_idx=0, end_char_idx=4397, text_template='{metadata_str}\n\n{content}', metadata_template='{key}: {value}', metadata_seperator='\n'), score=10.0), NodeWithScore(node=TextNode(id_='0a680a09-1f8d-409e-bbdc-b562b4879f6f', embedding=None, metadata={'page_label': '138', 'file_name': 'fibromyalgia.pdf', 'file_path': 'c:\\Users\\PLNAYAK\\Documents\\workflow\\Data\\fibromyalgia.pdf', 'file_type': 'application/pdf', 'file_size': 632664, 'creation_date': '2024-09-09', 'last_modified_date': '2024-09-09'}, excluded_embed_metadata_keys=['file_name', 'file_type', 'file_size', 'creation_date', 'last_modified_date', 'last_accessed_date'], excluded_llm_metadata_keys=['file_name', 'file_type', 'file_size', 'creation_date', 'last_modified_date', 'last_accessed_date'], relationships={<NodeRelationship.SOURCE: '1'>: RelatedNodeInfo(node_id='f2591e71-4fd5-48ef-8c08-ab465fdabf88', node_type=<ObjectType.DOCUMENT: '4'>, metadata={'page_label': '138', 'file_name': 'fibromyalgia.pdf', 'file_path': 'c:\\Users\\PLNAYAK\\Documents\\workflow\\Data\\fibromyalgia.pdf', 'file_type': 'application/pdf', 'file_size': 632664, 'creation_date': '2024-09-09', 'last_modified_date': '2024-09-09'}, hash='e0f0653ead2af78ad773abacd139db824a7e40216476d9aa63077b83a7370686')}, text='138 American Family Physician www.aafp.org/afp Volume 107, Number 2 ◆ February 2023\nFIBROMYALGIAClinical Presentation\nChronic diffuse pain is the predominant \nsymptom in most patients with fibromyalgia. Patients may also experience muscle stiffness and tenderness. The physical examination in patients with fibromyalgia generally finds diffuse tenderness without other unusual findings. If joint swelling, inflammation, or deformities are present, an alternative or additional diagnosis should be investigated.\n5 \nFatigue and sleep disturbances are also com -\nmon.5,11 Sleep disturbances include difficulty \nfalling and staying asleep, frequent awaken -\nings, or feeling unrefreshed after sleeping. Comorbid mental health diagnoses are com -\nmon, as are cognitive symptoms such as poor concentration, forgetfulness, or altered think -\ning.\n5,6,12 This cognitive dysfunction has been \ntermed “fibrofog” and is described by patients as a mental slowing that adversely affects daily activities.\n13\nThe presence of another painful disorder \ndoes not exclude the diagnosis of fibromyal -\ngia. The Fibromyalgia Rapid Screening Tool can screen patients with diffuse chronic pain to help distinguish between fibromyalgia and other conditions (Table 1) .\n14 The tool may SORT: KEY RECOMMENDATIONS FOR PRACTICE\nClinical recommendationEvidence \nrating Comments\nThe diagnosis of fibromyalgia should be considered in patients \nwith diffuse pain, fatigue, and sleep disturbances that have been present for at least three months.\n5,11C Diagnosis of fibromyalgia can be made using AAPT 2019 diagnostic criteria or the American College of Radiology 2011/2016 criteria\nPatients with fibromyalgia should be offered a multidisci -\nplinary treatment approach that includes education, exercise, and nonpharmacologic and pharmacologic options.\n27,28C Consensus guidelines and systematic reviews\nCognitive behavior therapy leads to improvement in pain and disability in patients with fibromyalgia in the short and medium term.\n32,34,35A Systematic reviews demonstrate improvement\nAmitriptyline, cyclobenzaprine, duloxetine (Cymbalta), mil -\nnacipran (Savella), and pregabalin (Lyrica) are effective for pain in fibromyalgia.\n43,46-48,50,52,54A Systematic reviews demonstrate effectiveness of these medications\nAAPT = Analgesic, Anesthetic, and Addiction Clinical Trial Translations Innovations Opportunities and Networks–American Pain Society Pain \nTaxonomy.\nA = consistent, good-quality patient-oriented evidence; B = inconsistent or limited-quality patient-oriented evidence; C = consensus, disease -\noriented evidence, usual practice, expert opinion, or case series. For information about the SORT evidence rating system, go to https:// www.aafp.\norg/afpsort.\nTABLE 1\nFibromyalgia Rapid Screening Tool (FiRST)\n Yes\nI have pain all over my body. \nMy pain is accompanied by a continuous and very unpleas -\nant general fatigue. \nMy pain feels like burns, electric shocks, or cramps. \nMy pain is accompanied by other unusual sensations \nthroughout my body, such as pins and needles, tingling, or numbness. \nMy pain is accompanied by other health problems such as digestive problems, urinary problems, headaches, or restless legs. \nMy pain has a significant impact on my life, particularly on my sleep and my ability to concentrate, making me feel slower in general. \nTotal* \n*—One point for each yes answer. A score of 5 or greater suggests fibromyalgia.\nAdapted with permission from Perrot S, Bouhassira D, Fermanian J; CEDR (C ercle \nd’Etude de la Douleur en Rhumatologie). Development and validation of the Fibro -\nmyalgia Rapid Screening Tool (FiRST). Pain. 2010; 150(2): 255.\nDescargado para Boletin -BINASSS (bolet-binas@binasss.sa.cr) en National Library of Health and Social Security de ClinicalKey.es por Elsevier en marzo 24, \n2023. Para uso personal exclusivamente. No se permiten otros usos sin autorización. Copyright ©2023. Elsevier Inc. Todos los derechos reservados.', mimetype='text/plain', start_char_idx=0, end_char_idx=3975, text_template='{metadata_str}\n\n{content}', metadata_template='{key}: {value}', metadata_seperator='\n'), score=8.0)]
Fibromyalgia is a chronic, centralized pain syndrome characterized by disordered processing of painful stimuli. It is characterized by diffuse musculoskeletal pain, fatigue, poor sleep, and other somatic symptoms.rag_flow_recent.html
Visualize the Workflow
Workflow Observations:
- We have two entry points (the steps that accept
StartEvent
) - The steps themselves decide when they can run
- The workflow context is used to store the user query
- The nodes are passed around, and finally a streaming response is returned
RAGWorkflow :
- The workflow is triggered by the StartEvent.
- The StartEvent triggers the ingest step to index ,chunk and load documents into the VectorStore.
- The ingest step returns the index generated and this step is closed by the StopEvent
- The StartEvent then triggers the retrieve step with query and index as input
- The retrieve step triggers the RetrieverEvent which return the nodes matching the query.
- The RetrieverEvent triggers the rerank step with matched nodes as input.
- The rerank step re-ranks the matched nodes as per their proximity to the query and triggers the RerankEvent.
- The RerankEvent then triggers the synthesize step to generate the final response taking into consideration the reranked nodes.
Visualizing the Current Execution Workflow
Few Observations regarding the Workflow Execution:
Here are some detailed insights and recommendations based on my observations:
1. Python 3.11.0 Compatibility
Observation: The workflow and visualization work correctly with Python 3.11.0.
Recommendation: Python 3.11.0 supports the | operator for type annotations, which simplifies type hinting. If possible, use Python 3.11.0 or later for development to leverage these features and avoid compatibility issues.
2. Python 3.9.12 Compatibility
Observation: The | operator does not work in Python 3.9.12, and you had to modify num_of_events in utils.py.
Recommendation:
Type Hinting: Use Union from the typing module instead of the | operator for type annotations in Python 3.9.12.
Validation Fix: Modifying num_of_events directly in utils.py is a workaround but not ideal. Instead, ensure that each step function has exactly one parameter annotated with the Event type to pass the validation.
3. Visualization in Google Colab
Observation: Visualization is not working in Google Colab even with notebook=True.
Recommendation:
Ensure Proper Installation: Make sure all required packages are installed in Colab.
!pip install pyvis llama-index-core llama-index-llms-openai llama-index-embeddings-openai llama-index-readers-file llama-index-utils-workflow
#Check HTML Rendering: Use the following approach to render the HTML content in Colab:
from pyvis.network import Network
from IPython.core.display import display, HTML
def draw_all_possible_flows(
workflow: Workflow,
filename: str = "workflow_all_flows.html",
notebook: bool = True, # Set notebook to True
) -> None:
net = Network(directed=True, height="750px", width="100%")
# Add the nodes + edge for stop events
net.add_node(
StopEvent.__name__,
label=StopEvent.__name__,
color="#FFA07A",
shape="ellipse",
)
net.add_node("_done", label="_done", color="#ADD8E6", shape="box")
net.add_edge(StopEvent.__name__, "_done")
# Add nodes from all steps
steps = get_steps_from_class(workflow)
if not steps:
# If no steps are defined in the class, try to get them from the instance
steps = get_steps_from_instance(workflow)
step_config: Optional[StepConfig] = None
for step_name, step_func in steps.items():
step_config = getattr(step_func, "__step_config", None)
if step_config is None:
continue
net.add_node(
step_name, label=step_name, color="#ADD8E6", shape="box"
) # Light blue for steps
for event_type in step_config.accepted_events:
net.add_node(
event_type.__name__,
label=event_type.__name__,
color="#90EE90" if event_type != StartEvent else "#E27AFF",
shape="ellipse",
) # Light green for events
# Add edges from all steps
for step_name, step_func in steps.items():
step_config = getattr(step_func, "__step_config", None)
if step_config is None:
continue
for return_type in step_config.return_types:
if return_type != type(None):
net.add_edge(step_name, return_type.__name__)
for event_type in step_config.accepted_events:
net.add_edge(event_type.__name__, step_name)
if notebook:
net.show(filename, notebook=True)
with open(filename, "r") as file:
display(HTML(file.read()))
else:
net.show(filename)
# Example usage in Google Colab
draw_all_possible_flows(
RAGWorkflow, filename="multi_step_workflow.html", notebook=True
)
Summary
- Python 3.11.0: Works well with the | operator and the workflow visualization.
- Python 3.9.12: Use Union for type annotations and ensure proper event parameter annotations to avoid modifying utils.py.
- Google Colab: Ensure all dependencies are installed and use the provided method to render HTML content.
By following these recommendations, you should be able to achieve consistent functionality across different environments and Python version
Conclusion
As AI applications evolve toward greater complexity and agentic functionality, the limitations of traditional Directed Acyclic Graphs (DAGs) become increasingly apparent. The inability to incorporate loops restricts developers’ capacity to implement self-correction mechanisms, which are essential for maintaining robust and adaptive AI systems.
Moving forward, it is crucial to explore alternative frameworks that accommodate the dynamic and iterative nature of modern AI applications. By prioritizing intuitive design and efficient orchestration methods, we can empower developers to create more resilient and adaptable AI systems that can learn from their mistakes and thrive in an increasingly complex landscape.
Here we have explored and implemented a use case using LlamaIndex Workflow. We still need to experiment more to understand it’s full potential and easy in implementing and interpreting complex architecture.
References:
Note: In one means I claim that the above experimentation is of my own.I have referenced materials online for implementation.