Observability for LlamaIndex with Opik

1
import os
2
import getpass
3
4
if "OPENAI_API_KEY" not in os.environ:
5
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

1
from llama_index.core import global_handler, set_global_handler
2
3
set_global_handler("opik")
4
opik_callback_handler = global_handler

1
from llama_index.core import Settings
2
from llama_index.core.callbacks import CallbackManager
3
from opik.integrations.llama_index import LlamaIndexCallbackHandler
4
5
# Basic setup
6
opik_callback = LlamaIndexCallbackHandler()
7
8
# Or with optional parameters
9
opik_callback = LlamaIndexCallbackHandler(
10
    project_name="my-llamaindex-project",  # Set custom project name
11
    skip_index_construction_trace=True     # Skip tracking index construction
12
)
13
14
Settings.callback_manager = CallbackManager([opik_callback])

1
import os
2
from llama_index.core import global_handler, set_global_handler
3
4
# Set project name for better organization
5
os.environ["OPIK_PROJECT_NAME"] = "llamaindex-integration-demo"
6
7
set_global_handler("opik")
8
opik_callback_handler = global_handler

1
import os
2
import requests
3
4
# Create directory if it doesn't exist
5
os.makedirs('./data/paul_graham/', exist_ok=True)
6
7
# Download the file using requests
8
url = 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt'
9
response = requests.get(url)
10
with open('./data/paul_graham/paul_graham_essay.txt', 'wb') as f:
11
    f.write(response.content)

1
import os
2
import getpass
3
4
if "OPENAI_API_KEY" not in os.environ:
5
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

1
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
2
3
documents = SimpleDirectoryReader("./data/paul_graham").load_data()
4
index = VectorStoreIndex.from_documents(documents)
5
query_engine = index.as_query_engine()
6
7
response = query_engine.query("What did the author do growing up?")
8
print(response)

1
import opik
2
from llama_index.core import global_handler, set_global_handler
3
from llama_index.llms.openai import OpenAI
4
from llama_index.core.llms import ChatMessage
5
6
# Configure Opik integration
7
set_global_handler("opik")
8
opik_callback_handler = global_handler
9
10
@opik.track()
11
def my_llm_application(user_query: str):
12
    """Process user query with LlamaIndex"""
13
    llm = OpenAI(model="gpt-3.5-turbo")
14
    messages = [
15
        ChatMessage(role="system", content="You are a helpful assistant."),
16
        ChatMessage(role="user", content=user_query),
17
    ]
18
    
19
    response = llm.chat(messages)
20
    return response.message.content
21
22
# Call the tracked function
23
result = my_llm_application("What is the capital of France?")
24
print(result)

1
import opik
2
from llama_index.core import global_handler, set_global_handler
3
from llama_index.llms.openai import OpenAI
4
from llama_index.core.llms import ChatMessage
5
6
# Configure Opik integration
7
set_global_handler("opik")
8
opik_callback_handler = global_handler
9
10
# Create a manual trace
11
with opik.start_as_current_trace(name="user_query_processing"):
12
    llm = OpenAI(model="gpt-3.5-turbo")
13
    messages = [
14
        ChatMessage(role="user", content="Explain quantum computing in simple terms"),
15
    ]
16
    
17
    response = llm.chat(messages)
18
    print(response.message.content)

1
import opik
2
from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step, Event
3
from llama_index.core import Settings
4
from llama_index.core.callbacks import CallbackManager
5
from llama_index.core import global_handler, set_global_handler
6
7
# Configure Opik integration for LLM calls within steps
8
set_global_handler("opik")
9
10
class QueryEvent(Event):
11
    """Event for passing query through workflow."""
12
    query: str
13
14
class MyRAGWorkflow(Workflow):
15
    """Simple RAG workflow with tracked steps."""
16
    
17
    @step
18
    @opik.track()
19
    async def retrieve_context(self, ev: StartEvent) -> QueryEvent:
20
        """Retrieve relevant context for the query."""
21
        query = ev.get("query", "")
22
        # Your retrieval logic here
23
        context = f"Context for: {query}"
24
        return QueryEvent(query=f"{context} | {query}")
25
    
26
    @step
27
    @opik.track()
28
    async def generate_response(self, ev: QueryEvent) -> StopEvent:
29
        """Generate final response using the context."""
30
        # Your generation logic here
31
        result = f"Response based on: {ev.query}"
32
        return StopEvent(result=result)
33
34
# Create workflow instance
35
workflow = MyRAGWorkflow()
36
37
# Use start_as_current_span to track workflow execution
38
with opik.start_as_current_span(
39
    name="rag_workflow_execution",
40
    input={"query": "What are the key features?"},
41
    project_name="llama-index-workflows"
42
) as span:
43
    result = await workflow.run(query="What are the key features?")
44
    span.update(output={"result": result})
45
    
46
print(result)
47
opik.flush_tracker()  # Ensure all traces are sent

1
from llama_index.llms.openai import OpenAI
2
from llama_index.core.llms import ChatMessage
3
from llama_index.core import global_handler, set_global_handler
4
5
# Configure Opik integration
6
set_global_handler("opik")
7
8
# Configure OpenAI LLM with stream_options to include usage information
9
llm = OpenAI(
10
    model="gpt-3.5-turbo",
11
    additional_kwargs={
12
        "stream_options": {"include_usage": True}
13
    }
14
)
15
16
messages = [
17
    ChatMessage(role="user", content="Tell me a short joke")
18
]
19
20
# Token usage will now be tracked in streaming responses
21
response = llm.stream_chat(messages)
22
for chunk in response:
23
    print(chunk.delta, end="", flush=True)