Observability for LangChain with Opik

1
from langchain_openai import ChatOpenAI
2
from langchain_core.prompts import ChatPromptTemplate
3
from opik.integrations.langchain import OpikTracer
4
5
# Initialize the tracer
6
opik_tracer = OpikTracer(project_name="langchain-examples")
7
8
llm = ChatOpenAI(model="gpt-4o", temperature=0)
9
prompt = ChatPromptTemplate.from_messages([
10
    ("human", "Translate the following text to French: {text}")
11
])
12
chain = prompt | llm
13
14
result = chain.invoke(
15
    {"text": "Hello, how are you?"},
16
    config={"callbacks": [opik_tracer]}
17
)
18
print(result.content)

1
import os
2
import getpass
3
import opik
4
from opik.integrations.openai import track_openai
5
from openai import OpenAI
6
7
# Configure Opik
8
opik.configure(use_local=False)
9
os.environ["OPIK_PROJECT_NAME"] = "langchain-integration-demo"
10
11
# Set up API keys
12
if "OPENAI_API_KEY" not in os.environ:
13
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

1
import json
2
from langchain_community.utilities import SQLDatabase
3
4
# Download and set up the Chinook database
5
import requests
6
7
url = "https://github.com/lerocha/chinook-database/raw/master/ChinookDatabase/DataSources/Chinook_Sqlite.sqlite"
8
filename = "./data/chinook/Chinook_Sqlite.sqlite"
9
10
folder = os.path.dirname(filename)
11
if not os.path.exists(folder):
12
    os.makedirs(folder)
13
14
if not os.path.exists(filename):
15
    response = requests.get(url)
16
    with open(filename, "wb") as file:
17
        file.write(response.content)
18
    print("Chinook database downloaded")
19
20
db = SQLDatabase.from_uri(f"sqlite:///{filename}")
21
22
# Create synthetic questions using OpenAI
23
client = OpenAI()
24
openai_client = track_openai(client)
25
26
prompt = """
27
Create 20 different example questions a user might ask based on the Chinook Database.
28
These questions should be complex and require the model to think. They should include complex joins and window functions to answer.
29
Return the response as a json object with a "result" key and an array of strings with the question.
30
"""
31
32
completion = openai_client.chat.completions.create(
33
    model="gpt-3.5-turbo",
34
    messages=[{"role": "user", "content": prompt}]
35
)
36
37
synthetic_questions = json.loads(completion.choices[0].message.content)["result"]
38
39
# Create dataset in Opik
40
opik_client = opik.Opik()
41
dataset = opik_client.get_or_create_dataset(name="synthetic_questions")
42
dataset.insert([{"question": question} for question in synthetic_questions])

1
from langchain.chains import create_sql_query_chain
2
from langchain_openai import ChatOpenAI
3
from opik.integrations.langchain import OpikTracer
4
5
# Create the LangChain chain with OpikTracer
6
opik_tracer = OpikTracer(tags=["sql_generation"])
7
8
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
9
chain = create_sql_query_chain(llm, db).with_config({"callbacks": [opik_tracer]})
10
11
# Test the chain
12
response = chain.invoke({"question": "How many employees are there?"})
13
print(response)

1
from opik import track
2
from opik.evaluation import evaluate
3
from opik.evaluation.metrics import base_metric, score_result
4
from typing import Any
5
6
class ValidSQLQuery(base_metric.BaseMetric):
7
    def __init__(self, name: str, db: Any):
8
        self.name = name
9
        self.db = db
10
11
    def score(self, output: str, **ignored_kwargs: Any):
12
        try:
13
            db.run(output)
14
            return score_result.ScoreResult(
15
                name=self.name, value=1, reason="Query ran successfully"
16
            )
17
        except Exception as e:
18
            return score_result.ScoreResult(name=self.name, value=0, reason=str(e))
19
20
# Set up evaluation
21
valid_sql_query = ValidSQLQuery(name="valid_sql_query", db=db)
22
dataset = opik_client.get_dataset("synthetic_questions")
23
24
@track()
25
def llm_chain(input: str) -> str:
26
    response = chain.invoke({"question": input})
27
    return response
28
29
def evaluation_task(item):
30
    response = llm_chain(item["question"])
31
    return {"output": response}
32
33
# Run evaluation
34
res = evaluate(
35
    experiment_name="SQL question answering",
36
    dataset=dataset,
37
    task=evaluation_task,
38
    scoring_metrics=[valid_sql_query],
39
    nb_samples=20,
40
)

1
from langchain_openai import ChatOpenAI
2
from opik.integrations.langchain import OpikTracer
3
4
llm = ChatOpenAI(
5
    model="gpt-4o",
6
    streaming=True,
7
    stream_usage=True,  # Required for cost tracking with streaming
8
)
9
10
opik_tracer = OpikTracer()
11
12
for chunk in llm.stream("Hello", config={"callbacks": [opik_tracer]}):
13
    print(chunk.content, end="")

1
from opik.integrations.langchain import OpikTracer
2
3
opik_tracer = OpikTracer(
4
    tags=["langchain", "production"],
5
    metadata={"use-case": "customer-support", "version": "1.0"},
6
    thread_id="conversation-123",  # For conversational applications
7
    project_name="my-langchain-project"
8
)

1
from opik.integrations.langchain import OpikTracer
2
3
opik_tracer = OpikTracer()
4
5
# Calling Langchain object
6
traces = opik_tracer.created_traces()
7
print([trace.id for trace in traces])

1
import opik
2
from opik.integrations.langchain import OpikTracer
3
opik_client = opik.Opik()
4
5
opik_tracer = OpikTracer()
6
7
8
# Calling Langchain object
9
10
# Getting the content of the logged traces
11
traces = opik_tracer.created_traces()
12
for trace in traces:
13
    content = opik_client.get_trace_content(trace.id)
14
    print(content)

1
from opik.integrations.langchain import OpikTracer
2
3
opik_tracer = OpikTracer(project_name="langchain-examples")
4
5
# ... calling Langchain object
6
7
traces = opik_tracer.created_traces()
8
9
for trace in traces:
10
    trace.update(tags=["my-tag"])
11
    trace.log_feedback_score(name="user-feedback", value=0.5)

1
import opik
2
from langchain_openai import ChatOpenAI
3
from opik.integrations.langchain import OpikTracer
4
5
@opik.track
6
def my_langchain_workflow(user_input: str) -> str:
7
    llm = ChatOpenAI(model="gpt-4o")
8
    opik_tracer = OpikTracer()
9
10
    # The LangChain call will create spans within the existing trace
11
    response = llm.invoke(user_input, config={"callbacks": [opik_tracer]})
12
    return response.content
13
14
result = my_langchain_workflow("What is machine learning?")

1
from opik.integrations.langchain import OpikTracer
2
3
# All traces with the same thread_id will be grouped together
4
opik_tracer = OpikTracer(thread_id="user-session-123")

1
from opik import opik_context
2
from opik.integrations.langchain import OpikTracer
3
from opik.types import DistributedTraceHeadersDict
4
5
# In your service that receives distributed trace headers.
6
# The distributed_headers dict can be obtained in the "parent" service via `opik_context.get_distributed_trace_headers()`
7
distributed_headers = DistributedTraceHeadersDict(
8
    opik_trace_id="trace-id-from-upstream",
9
    opik_parent_span_id="parent-span-id-from-upstream"
10
)
11
12
opik_tracer = OpikTracer(distributed_headers=distributed_headers)
13
14
# LangChain operations will be attached to the existing distributed trace
15
chain.invoke(input_data, config={"callbacks": [opik_tracer]})

1
from langgraph.graph import StateGraph
2
from opik.integrations.langchain import OpikTracer
3
4
# Your LangGraph setup
5
graph = StateGraph(...)
6
compiled_graph = graph.compile()
7
8
opik_tracer = OpikTracer()
9
result = compiled_graph.invoke(
10
    input_data,
11
    config={"callbacks": [opik_tracer]}
12
)

1
from opik.integrations.langchain import OpikTracer
2
3
opik_tracer = OpikTracer()
4
opik_tracer.flush()