# LlamaIndex API Example

### Reader and Query Engine

documents = SimpleDirectoryReader('files').load\_data()

response = query\_engine.query("summarize each document in a few sentences")

### doc and textnode and index

doc = Document( text=text, metadata={'author': 'John Doe','category': 'others'}, id\_='1' )

nodes = \[ TextNode( text="Lionel Messi is a football player from Argentina." ), TextNode( text="He has won the Ballon d'Or trophy 7 times." ), TextNode(text="Lionel Messi's hometown is Rosario."), TextNode(text="He was born on June 24, 1987.") ] index = SummaryIndex(nodes)

### replace llm

from llama\_index.core.settings import Settings Settings.llm = OpenAI(temperature=0.8, model="gpt-4")

### Import Pipeline 4 cache

from llama\_index.core import SimpleDirectoryReader from llama\_index.core.extractors import SummaryExtractor,QuestionsAnsweredExtractor from llama\_index.core.node\_parser import TokenTextSplitter from llama\_index.core.ingestion import IngestionPipeline, IngestionCache from llama\_index.core.schema import TransformComponent

class CustomTransformation(TransformComponent): def **call**(self, nodes, \*\*kwargs): # run any node transformation logic here return nodes

reader = SimpleDirectoryReader('files') documents = reader.load\_data() try: cached\_hashes = IngestionCache.from\_persist\_path( "./ingestion\_cache.json" ) print("Cache file found. Running using cache...") except: cached\_hashes = "" print("No cache file found. Running without cache...")

pipeline = IngestionPipeline( transformations = \[ CustomTransformation(), TokenTextSplitter( separator=" ", chunk\_size=512, chunk\_overlap=128), SummaryExtractor(), QuestionsAnsweredExtractor( questions=3 ) ], cache=cached\_hashes )

nodes = pipeline.run(documents=documents, show\_progress=True) pipeline.cache.persist("./ingestion\_cache.json")

print("All documents loaded")

***

\#ingest uploaded documents from global\_settings import STORAGE\_PATH, INDEX\_STORAGE, CACHE\_FILE from logging\_functions import log\_action from llama\_index.core import SimpleDirectoryReader, VectorStoreIndex from llama\_index.core.ingestion import IngestionPipeline, IngestionCache from llama\_index.core.node\_parser import TokenTextSplitter from llama\_index.core.extractors import SummaryExtractor from llama\_index.embeddings.openai import OpenAIEmbedding

def ingest\_documents(): documents = SimpleDirectoryReader( STORAGE\_PATH, filename\_as\_id = True ).load\_data() for doc in documents: print(doc.id\_) log\_action( f"File '{doc.id\_}' uploaded user", action\_type="UPLOAD" )

```
try: 
    cached_hashes = IngestionCache.from_persist_path(
        CACHE_FILE
        )
    print("Cache file found. Running using cache...")
except:
    cached_hashes = ""
    print("No cache file found. Running without cache...")
pipeline = IngestionPipeline(
    transformations=[
        TokenTextSplitter(
            chunk_size=1024, 
            chunk_overlap=20
        ),
        SummaryExtractor(summaries=['self']),
        OpenAIEmbedding()
    ],
    cache=cached_hashes
)

nodes = pipeline.run(documents=documents)
pipeline.cache.persist(CACHE_FILE)

return nodes
```

if **name** == "**main**": embedded\_nodes = ingest\_documents()

### VectorStoreIndex

from llama\_index.core import VectorStoreIndex, SimpleDirectoryReader documents = SimpleDirectoryReader("files").load\_data() index = VectorStoreIndex.from\_documents(documents) print("Index created successfully!")

### use local embedding

或者集成LangChain,使用其提供的嵌入模型

from llama\_index.embeddings.huggingface import HuggingFaceEmbedding embedding\_model = HuggingFaceEmbedding( model\_name="WhereIsAI/UAE-Large-V1" ) embeddings = embedding\_model.get\_text\_embedding( "The quick brown fox jumps over the lazy cat!" ) print(embeddings\[:15])

### Index持久化 from 内存 to 磁盘

from llama\_index.core import VectorStoreIndex, SimpleDirectoryReader documents = SimpleDirectoryReader("data").load\_data() index = VectorStoreIndex.from\_documents(documents)

**index.storage\_context.persist(persist\_dir="index\_cache")** print("Index persisted to disk.")

#### reload index

from llama\_index.core import StorageContext, load\_index\_from\_storage **storage\_context = StorageContext.from\_defaults(persist\_dir="index\_cache")** index = load\_index\_from\_storage(storage\_context)

print("Index loaded successfully!")

### use chromadb

**import chromadb** **from llama\_index.vector\_stores.chroma import ChromaVectorStore** from llama\_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext

**db = chromadb.PersistentClient(path="chroma\_database")** chroma\_collection = db.get\_or\_create\_collection( "my\_chroma\_store" )

vector\_store = ChromaVectorStore( chroma\_collection=chroma\_collection ) storage\_context = StorageContext.from\_defaults( vector\_store=vector\_store )

documents = SimpleDirectoryReader("files").load\_data() index = VectorStoreIndex.from\_documents( documents=documents, storage\_context=storage\_context )

\#the following part displays the entire contents of the ChromaDB collection results = chroma\_collection.get() print(results)

''' We can use the next part to rebuild the Index from the ChromaDB in future sessions index = VectorStoreIndex.from\_vector\_store( vector\_store=vector\_store, storage\_context=storage\_context ) '''

### Index class

from llama\_index.core import SummaryIndex, SimpleDirectoryReader documents = SimpleDirectoryReader("files").load\_data() index = SummaryIndex.from\_documents(documents) query\_engine = index.as\_query\_engine() response = query\_engine.query("How many documents have you loaded?") print(response)

from llama\_index.core import TreeIndex, SimpleDirectoryReader documents = SimpleDirectoryReader("files").load\_data() index = TreeIndex.from\_documents(documents) query\_engine = index.as\_query\_engine() response = query\_engine.query("Tell me about dogs") print(response)

from llama\_index.core import ComposableGraph, SimpleDirectoryReader, TreeIndex, SummaryIndex

documents = SimpleDirectoryReader("files").load\_data() index1 = TreeIndex.from\_documents(\[documents\[0]]) index2 = TreeIndex.from\_documents(\[documents\[1]]) summary1 = "A short introduction to ancient Rome" summary2 = "Some facts about dogs"

graph = ComposableGraph.from\_indices( SummaryIndex, \[index1, index2], index\_summaries=\[summary1, summary2] ) query\_engine = graph.as\_query\_engine()

response = query\_engine.query("What can you tell me?") print(response)

from llama\_index.core import DocumentSummaryIndex, SimpleDirectoryReader

documents = SimpleDirectoryReader("files").load\_data() index = DocumentSummaryIndex.from\_documents( documents, show\_progress=True )

summary1 = index.get\_document\_summary(documents\[0].doc\_id) summary2 = index.get\_document\_summary(documents\[1].doc\_id) print("\nSummary of the first document:" + summary1) print("\nSummary of the second document:" + summary2)

from llama\_index.core import KeywordTableIndex, SimpleDirectoryReader documents = SimpleDirectoryReader("files").load\_data() index = KeywordTableIndex.from\_documents(documents) query\_engine = index.as\_query\_engine() response = query\_engine.query("What famous buildings were in ancient Rome?") print(response)

from llama\_index.core import KnowledgeGraphIndex, SimpleDirectoryReader documents = SimpleDirectoryReader("files").load\_data() index = KnowledgeGraphIndex.from\_documents(documents, max\_triplets\_per\_chunk=2, use\_async=True) query\_engine = index.as\_query\_engine() response = query\_engine.query("Tell me about dogs.") print(response)

### token mock test

import tiktoken from llama\_index.core import TreeIndex, SimpleDirectoryReader, Settings from llama\_index.core.llms.mock import MockLLM from llama\_index.core.callbacks import CallbackManager, TokenCountingHandler

llm = MockLLM(max\_tokens=256) token\_counter = TokenCountingHandler( tokenizer=tiktoken.encoding\_for\_model("gpt-3.5-turbo").encode ) callback\_manager = CallbackManager(\[token\_counter])

Settings.callback\_manager=callback\_manager Settings.llm=llm

documents = SimpleDirectoryReader("cost\_prediction\_samples").load\_data()

index = TreeIndex.from\_documents( documents=documents, num\_children=2, show\_progress=True)

print("Total LLM Token Count:", token\_counter.total\_llm\_token\_count)

import tiktoken from llama\_index.core import MockEmbedding, VectorStoreIndex, SimpleDirectoryReader, Settings from llama\_index.core.callbacks import CallbackManager, TokenCountingHandler from llama\_index.core.llms.mock import MockLLM

embed\_model = MockEmbedding(embed\_dim=1536) llm = MockLLM(max\_tokens=256) token\_counter = TokenCountingHandler( tokenizer=tiktoken.encoding\_for\_model("gpt-3.5-turbo").encode ) callback\_manager = CallbackManager(\[token\_counter])

Settings.embed\_model=embed\_model Settings.llm=llm Settings.callback\_manager=callback\_manager

documents = SimpleDirectoryReader("cost\_prediction\_samples").load\_data() index = VectorStoreIndex.from\_documents( documents=documents, show\_progress=True) print("Embedding Token Count:", token\_counter.total\_embedding\_token\_count)

query\_engine = index.as\_query\_engine()\
response = query\_engine.query("What's the cat's name?")\
print("Query LLM Token Count:", token\_counter.total\_llm\_token\_count) print("Query Embedding Token Count:",token\_counter.total\_embedding\_token\_count)

### PITS - index\_builder.py

from llama\_index.core import VectorStoreIndex, TreeIndex, load\_index\_from\_storage from llama\_index.core import StorageContext from global\_settings import INDEX\_STORAGE from document\_uploader import ingest\_documents

def build\_indexes(nodes): try: storage\_context = StorageContext.from\_defaults( persist\_dir=INDEX\_STORAGE ) vector\_index = load\_index\_from\_storage( storage\_context, index\_id="vector" ) tree\_index = load\_index\_from\_storage( storage\_context, index\_id="tree" ) print("All indices loaded from storage.") except Exception as e: print(f"Error occurred while loading indices: {e}") storage\_context = StorageContext.from\_defaults() vector\_index = VectorStoreIndex( nodes, storage\_context=storage\_context ) vector\_index.set\_index\_id("vector") tree\_index = TreeIndex( nodes, storage\_context=storage\_context ) tree\_index.set\_index\_id("tree") storage\_context.persist( persist\_dir=INDEX\_STORAGE ) print("New indexes created and persisted.") return vector\_index, tree\_index