Comprehensive guide for building LLM applications with LlamaIndex, including data loaders, indexes, query engines, chat engines, vector stores, retrievers, agents, evaluation, streaming, and observability.
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Comprehensive guide for building LLM applications with LlamaIndex, including data loaders, indexes, query engines, chat engines, vector stores, retrievers, agents, evaluation, streaming, and observability.
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.selectors import LLMSingleSelector
# Create multiple query engines
summary_engine = summary_index.as_query_engine()
vector_engine = vector_index.as_query_engine()
# Create tools
query_engine_tools = [
QueryEngineTool(
query_engine=summary_engine,
metadata=ToolMetadata(
name="summary_tool",
description="Useful for summarizing documents",
),
),
QueryEngineTool(
query_engine=vector_engine,
metadata=ToolMetadata(
name="vector_tool",
description="Useful for specific questions about documents",
),
),
]
# Create router
router_engine = RouterQueryEngine(
selector=LLMSingleSelector.from_defaults(),
query_engine_tools=query_engine_tools,
verbose=True,
)
response = router_engine.query("What is the main topic?")
Sub-Question Query Engine
from llama_index.core.query_engine import SubQuestionQueryEngine
# Create sub-question engine
sub_question_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=query_engine_tools,
use_async=True,
verbose=True,
)
# Automatically decomposes into sub-questions
response = sub_question_engine.query(
"Compare the revenue growth of Company A and Company B"
)
Multi-Step Query Engine
from llama_index.core.query_engine import MultiStepQueryEngine
# Create multi-step engine
multi_step_engine = MultiStepQueryEngine(
query_engine=base_query_engine,
llm=OpenAI(model="gpt-4o"),
max_iterations=5,
verbose=True,
)
# Breaks down complex questions
response = multi_step_engine.query(
"What factors contributed to the market cap change?"
)
Agents
ReAct Agent
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
# Define toolsdefadd(x: int, y: int) -> int:
"""Add two numbers."""return x + y
defmultiply(x: int, y: int) -> int:
"""Multiply two numbers."""return x * y
defsearch_knowledge(query: str) -> str:
"""Search the knowledge base."""
response = query_engine.query(query)
returnstr(response)
tools = [
FunctionTool.from_defaults(add),
FunctionTool.from_defaults(multiply),
FunctionTool.from_defaults(search_knowledge),
]
# Create agent
agent = ReActAgent(
llm=OpenAI(model="gpt-4o", temperature=0.0),
tools=tools,
max_iterations=10,
verbose=True,
)
# Run
response = agent.chat("What is 20 + (5 * 3)?")
print(response)
Function Calling Agent
from llama_index.core.agent import FunctionCallingAgent
# Use for models with native function calling
agent = FunctionCallingAgent(
llm=OpenAI(model="gpt-4o"),
tools=tools,
max_iterations=10,
verbose=True,
)
response = agent.chat("Calculate and search")
from llama_index.core.tools import QueryEngineTool, ToolMetadata
# Wrap query engine as tool
query_tool = QueryEngineTool(
query_engine=index.as_query_engine(),
metadata=ToolMetadata(
name="knowledge_base",
description="Search the knowledge base for information",
),
)
# Use in agent
agent = ReActAgent(
llm=OpenAI(model="gpt-4o"),
tools=[query_tool, other_tool],
)
Streaming
Streaming Responses
from llama_index.core import VectorStoreIndex
# Enable streaming
query_engine = index.as_query_engine(streaming=True)
# Stream response
response = query_engine.query("What is the main topic?")
# Print tokens as they arrivefor token in response.response_gen:
print(token, end="", flush=True)
from llama_index.core.memory import ChatMemoryBuffer
# Create chat engine with memory
memory = ChatMemoryBuffer.from_defaults(token_limit=1500)
chat_engine = index.as_chat_engine(
chat_mode="context",
memory=memory,
streaming=True,
)
# Stream chat
response = chat_engine.stream_chat("Tell me about topic X")
for token in response.response_gen:
print(token, end="", flush=True)
# Continue conversation
response = chat_engine.stream_chat("Can you elaborate?")
for token in response.response_gen:
print(token, end="", flush=True)
FastAPI Streaming
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/stream")asyncdefstream_response(query: str):
asyncdefgenerate():
streaming_response = await query_engine.aquery(query)
asyncfor token in streaming_response.async_response_gen():
yieldf"data: {token}\n\n"yield"data: [DONE]\n\n"return StreamingResponse(
generate(),
media_type="text/event-stream",
)
Evaluation
Faithfulness Evaluation
from llama_index.core.evaluation import FaithfulnessEvaluator
from llama_index.llms.openai import OpenAI
# Create evaluator
llm = OpenAI(model="gpt-4o", temperature=0.0)
evaluator = FaithfulnessEvaluator(llm=llm)
# Evaluate response
query_engine = index.as_query_engine()
response = query_engine.query("What are the key points?")
eval_result = evaluator.evaluate_response(response=response)
print(f"Passing: {eval_result.passing}")
print(f"Feedback: {eval_result.feedback}")
Relevancy Evaluation
from llama_index.core.evaluation import RelevancyEvaluator
evaluator = RelevancyEvaluator(llm=llm)
eval_result = evaluator.evaluate_response(
query="What is the main topic?",
response=response,
)
print(f"Passing: {eval_result.passing}")
print(f"Score: {eval_result.score}")
Ragas Integration
from llama_index.core.evaluation import RagasEvaluator
from llama_index.core.evaluation.ragas import RagasMetric
# Create evaluator
evaluator = RagasEvaluator(
metric=RagasMetric.FAITHFULNESS,
)
# Evaluate
result = evaluator.evaluate_response(
query=query,
response=response,
contexts=[node.text for node in response.source_nodes],
)
print(f"Score: {result.score}")
Batch Evaluation
from tqdm import tqdm
defbatch_evaluate(queries, responses, evaluator):
results = []
for query, response in tqdm(zip(queries, responses)):
result = evaluator.evaluate_response(
query=query,
response=response,
)
results.append(result)
passing_rate = sum(1for r in results if r.passing) / len(results)
avg_score = sum(r.score for r in results if r.score) / len(results)
return {
"passing_rate": passing_rate,
"average_score": avg_score,
"results": results,
}
Observability
Callbacks and Token Tracking
from llama_index.core import Settings
from llama_index.core.callbacks import (
CallbackManager,
LlamaDebugHandler,
TokenCountingHandler,
)
import tiktoken
# Create handlers
token_counter = TokenCountingHandler(
tokenizer=tiktoken.encoding_for_model("gpt-4o").encode,
)
debug_handler = LlamaDebugHandler(print_trace_on_end=True)
# Set callback manager
Settings.callback_manager = CallbackManager([token_counter, debug_handler])
# Run query
response = query_engine.query("Your query")
# Get token countsprint(f"Total LLM tokens: {token_counter.total_llm_token_count}")
print(f"Embedding tokens: {token_counter.total_embedding_token_count}")
LlamaIndex Debugging
from llama_index.core.callbacks import LlamaDebugHandler
debug_handler = LlamaDebugHandler()
Settings.callback_manager = CallbackManager([debug_handler])
# Run query
response = query_engine.query("Your query")
# Get events
events = debug_handler.get_event_pairs()
for event in events:
print(f"Event: {event[0].type}")
print(f"Duration: {event[1].time - event[0].time}")
LangSmith Integration
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"# LlamaIndex automatically logs to LangSmith
response = query_engine.query("Your query")
Workflow Engine (v0.14+)
The workflow engine provides a structured way to build complex, multi-step LLM applications.
# ❌ BAD: Load all documents at once
documents = SimpleDirectoryReader("./huge_folder").load_data()
# ✅ GOOD: Process in batchesfrom llama_index.core import StorageContext
for batch in document_batches:
nodes = splitter.get_nodes_from_documents(batch)
index.insert_nodes(nodes)
Embedding Dimension Mismatch
# ❌ BAD: Index created with different embedding# Pinecone index: 1536 dimensions# Using: 768 dimension embeddings# ✅ GOOD: Match dimensions
embed_model = OpenAIEmbedding(model="text-embedding-3-small") # 1536 dims# Create Pinecone index with same dimensions
Best Practices
Use appropriate chunk sizes (512-1024 for most use cases)
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Document,
Settings,
StorageContext,
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
