LangGraph

Opik provides a seamless integration with LangGraph, allowing you to easily log and trace your LangGraph-based applications. By using the OpikTracer callback, you can automatically capture detailed information about your LangGraph graph executions during both development and production.

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

Getting Started

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

$ pip install opik langgraph langchain

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 the OpikTracer

You can use the OpikTracer callback with any LangGraph graph by passing it in as an argument to the stream or invoke functions:

1 from typing import List, Annotated
2 from pydantic import BaseModel
3 from opik.integrations.langchain import OpikTracer
4 from langchain_core.messages import HumanMessage
5 from langgraph.graph import StateGraph, START, END
6 from langgraph.graph.message import add_messages
7 
8 # create your LangGraph graph
9 class State(BaseModel):
10     messages: Annotated[list, add_messages]
11 
12 def chatbot(state):
13     # Typically your LLM calls would be done here
14     return {"messages": "Hello, how can I help you today?"}
15 
16 graph = StateGraph(State)
17 graph.add_node("chatbot", chatbot)
18 graph.add_edge(START, "chatbot")
19 graph.add_edge("chatbot", END)
20 app = graph.compile()
21 
22 # Create the OpikTracer
23 opik_tracer = OpikTracer(graph=app.get_graph(xray=True))
24 
25 # Pass the OpikTracer callback to the Graph.stream function
26 for s in app.stream({"messages": [HumanMessage(content = "How to use LangGraph ?")]},
27                       config={"callbacks": [opik_tracer]}):
28     print(s)
29 
30 # Pass the OpikTracer callback to the Graph.invoke function
31 result = app.invoke({"messages": [HumanMessage(content = "How to use LangGraph ?")]},
32                       config={"callbacks": [opik_tracer]})

Once the OpikTracer is configured, you will start to see the traces in the Opik UI:

Updating logged traces

You can use the OpikTracer.created_traces method to access the trace IDs collected by the OpikTracer callback:

1 from opik.integrations.langchain import OpikTracer
2 
3 opik_tracer = OpikTracer()
4 
5 # Calling LangGraph stream or invoke functions
6 
7 traces = opik_tracer.created_traces()
8 print([trace.id for trace in traces])

These can then be used with the Opik.log_traces_feedback_scores method to update the logged traces.

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

1	from typing import List, Annotated
2	from pydantic import BaseModel
3	from opik.integrations.langchain import OpikTracer
4	from langchain_core.messages import HumanMessage
5	from langgraph.graph import StateGraph, START, END
6	from langgraph.graph.message import add_messages
7
8	# create your LangGraph graph
9	class State(BaseModel):
10	messages: Annotated[list, add_messages]
11
12	def chatbot(state):
13	# Typically your LLM calls would be done here
14	return {"messages": "Hello, how can I help you today?"}
15
16	graph = StateGraph(State)
17	graph.add_node("chatbot", chatbot)
18	graph.add_edge(START, "chatbot")
19	graph.add_edge("chatbot", END)
20	app = graph.compile()
21
22	# Create the OpikTracer
23	opik_tracer = OpikTracer(graph=app.get_graph(xray=True))
24
25	# Pass the OpikTracer callback to the Graph.stream function
26	for s in app.stream({"messages": [HumanMessage(content = "How to use LangGraph ?")]},
27	config={"callbacks": [opik_tracer]}):
28	print(s)
29
30	# Pass the OpikTracer callback to the Graph.invoke function
31	result = app.invoke({"messages": [HumanMessage(content = "How to use LangGraph ?")]},
32	config={"callbacks": [opik_tracer]})

1	from opik.integrations.langchain import OpikTracer
2
3	opik_tracer = OpikTracer()
4
5	# Calling LangGraph stream or invoke functions
6
7	traces = opik_tracer.created_traces()
8	print([trace.id for trace in traces])

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