LangChain (Python)

Opik provides seamless integration with LangChain, allowing you to easily log and trace your LangChain-based applications. By using the OpikTracer callback, you can automatically capture detailed information about your LangChain runs, including inputs, outputs, metadata, and cost tracking for each step in your chain.

Key Features

• Automatic cost tracking for supported LLM providers (OpenAI, Anthropic, Google AI, AWS Bedrock, and more)
• Full compatibility with the @opik.track decorator for hybrid tracing approaches
• Thread support for conversational applications with thread_id parameter
• Distributed tracing support for multi-service applications
• LangGraph compatibility for complex graph-based workflows

You can check out the Colab Notebook if you’d like to jump straight to the code:

Getting Started

To use the OpikTracer with LangChain, you’ll need to have both the opik and langchain packages installed. You can install them using pip:

$
pip install opik langchain langchain_openai

In addition, you can configure Opik using the opik configure command which will prompt you for the correct local server address or if you are using the Cloud platform your API key:

$
opik configure

Using OpikTracer

Here’s a basic example of how to use the OpikTracer callback with a LangChain chain:

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)

The OpikTracer will automatically log the run and its details to Opik, including the input prompt, the output, and metadata for each step in the chain.

For detailed parameter information, see the OpikTracer SDK reference.

Cost Tracking

The OpikTracer automatically tracks token usage and cost for all supported LLM models used within LangChain applications.

Cost information is automatically captured and displayed in the Opik UI, including:

• Token usage details
• Cost per request based on model pricing
• Total trace cost

For streaming with cost tracking, ensure stream_usage=True is set:

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="")

Settings tags and metadata

You can customize the OpikTracer callback to include additional metadata, logging options, and conversation threading:

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
)

Accessing logged traces

You can use the created_traces method to access the traces collected by the OpikTracer callback:

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])

The traces returned by the created_traces method are instances of the Trace class, which you can use to update the metadata, feedback scores and tags for the traces.

Accessing the content of logged traces

In order to access the content of logged traces you will need to use the Opik.get_trace_content method:

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)

Updating and scoring logged traces

You can update the metadata, feedback scores and tags for traces after they are created. For this you can use the created_traces method to access the traces and then update them using the update method and the log_feedback_score method:

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)

Compatibility with @track Decorator

The OpikTracer is fully compatible with the @track decorator, allowing you to create hybrid tracing approaches:

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?")

Thread Support

Use the thread_id parameter to group related conversations or interactions:

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")

Distributed Tracing

For multi-service/thread/process applications, you can use distributed tracing headers to connect traces across services:

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]})

LangGraph Integration

For LangGraph applications, Opik provides specialized support. The OpikTracer works seamlessly with LangGraph, and you can also visualize graph definitions:

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
)

Advanced usage

The OpikTracer object has a flush method that can be used to make sure that all traces are logged to the Opik platform before you exit a script. This method will return once all traces have been logged or if the timeout is reach, whichever comes first.

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

Important notes

1. Asynchronous streaming: If you are using asynchronous streaming mode (calling .astream() method), the input field in the trace UI may be empty due to a LangChain limitation for this mode. However, you can find the input data inside the nested spans of this chain.

2. Streaming with cost tracking: If you are planning to use streaming with LLM calls and want to calculate LLM call tokens/cost, you need to explicitly set stream_usage=True: