Sample code for the article on LlamaIndex

Author: lpozo

python-llamaindex/README.md

@@ -0,0 +1,3 @@
+# Python LlamaIndex: Step by Step RAG With Examples
+
+This folder provides the code examples for the Real Python tutorial [Python LlamaIndex: Step by Step RAG With Examples](https://realpython.com/python-llamaindex-examples/).

python-llamaindex/agent.py

@@ -0,0 +1,33 @@
+import asyncio
+
+import requests
+from llama_index.core.agent.workflow import AgentWorkflow
+from llama_index.llms.openai import OpenAI
+
+
+def get_breed_info(breed_name):
+    response = requests.get("https://api.thecatapi.com/v1/breeds")
+    response.raise_for_status()
+    for breed in response.json():
+        if breed["name"] == breed_name:
+            return breed
+    return None
+
+
+workflow = AgentWorkflow.from_tools_or_functions(
+    tools_or_functions=[get_breed_info],
+    llm=OpenAI(model="gpt-4o-mini"),
+    system_prompt="You are an agent that can provide information about cat breeds.",
+)
+
+
+async def main():
+    response = await workflow.run(
+        user_msg="Can you describe the temperament of Siamese cats?"
+    )
+    print(response)
+
+
+if __name__ == "__main__":
+
+    asyncio.run(main())

python-llamaindex/async_query.py

@@ -0,0 +1,56 @@
+import asyncio
+import logging
+from pathlib import Path
+
+from llama_index.core import (
+    SimpleDirectoryReader,
+    StorageContext,
+    VectorStoreIndex,
+    load_index_from_storage,
+)
+
+# Disable logging messages
+logging.getLogger("llama_index").setLevel(logging.WARNING)
+logging.getLogger("httpx").setLevel(logging.WARNING)
+
+# Define the storage directory
+PERSIST_DIR = "./storage"
+
+
+def get_index(persist_dir=PERSIST_DIR):
+    if Path(persist_dir).exists():
+        storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
+        index = load_index_from_storage(storage_context)
+        print("Index loaded from storage...")
+    else:
+        reader = SimpleDirectoryReader(input_files=["./data/pep8.rst"])
+        documents = reader.load_data()
+        index = VectorStoreIndex.from_documents(documents)
+        index.storage_context.persist(persist_dir=persist_dir)
+        print("Index created and persisted to storage...")
+
+    return index
+
+
+async def main():
+    index = get_index()
+    query_engine = index.as_query_engine()
+
+    queries = [
+        "What is this document about?",
+        "Summarize the naming conventions in Python?",
+    ]
+
+    # Run queries asynchronously
+    tasks = [query_engine.aquery(query) for query in queries]
+    responses = await asyncio.gather(*tasks)
+
+    # Print responses
+    for i, (query, response) in enumerate(zip(queries, responses), 1):
+        print(f"\nQuery {i}: {query}")
+        print(f"Response: {response}\n")
+        print("-" * 80)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())