Using Opik with LangGraph

This notebook showcases how to use Opik with LangGraph. LangGraph is a library for building stateful, multi-actor applications with LLMs, used to create agent and multi-agent workflows

In this notebook, we will create a simple LangGraph workflow and focus on how to track it’s execution with Opik. To learn more about LangGraph, check out the official documentation.

Creating an account on Opik Cloud

Comet provides a hosted version of the Opik platform, simply create an account and grab you API Key.

You can also run the Opik platform locally, see the installation guide for more information.

1 %pip install --quiet -U langchain langgraph opik

1 import opik
2 
3 opik.configure(use_local=False)

Create the LangGraph graph

The LangGraph graph we will be created in made up of 3 nodes:

classify_input: Classify the input question
handle_greeting: Handle the greeting question
handle_search: Handle the search question

Note: We will not be using any LLM calls or tools in this example to keep things simple. However in most cases, you will want to use tools to interact with external systems.

1 # We will start by creating simple functions to classify the input question and handle the greeting and search questions.
2 def classify(question: str) -> str:
3     return "greeting" if question.startswith("Hello") else "search"
4 
5 
6 def classify_input_node(state):
7     question = state.get("question", "").strip()
8     classification = classify(question)  # Assume a function that classifies the input
9     return {"classification": classification}
10 
11 
12 def handle_greeting_node(state):
13     return {"response": "Hello! How can I help you today?"}
14 
15 
16 def handle_search_node(state):
17     question = state.get("question", "").strip()
18     search_result = f"Search result for '{question}'"
19     return {"response": search_result}

1 from langgraph.graph import StateGraph, END
2 
3 from typing import TypedDict, Optional
4 
5 
6 class GraphState(TypedDict):
7     question: Optional[str] = None
8     classification: Optional[str] = None
9     response: Optional[str] = None
10 
11 
12 workflow = StateGraph(GraphState)
13 workflow.add_node("classify_input", classify_input_node)
14 workflow.add_node("handle_greeting", handle_greeting_node)
15 workflow.add_node("handle_search", handle_search_node)
16 
17 
18 def decide_next_node(state):
19     return (
20         "handle_greeting"
21         if state.get("classification") == "greeting"
22         else "handle_search"
23     )
24 
25 
26 workflow.add_conditional_edges(
27     "classify_input",
28     decide_next_node,
29     {"handle_greeting": "handle_greeting", "handle_search": "handle_search"},
30 )
31 
32 workflow.set_entry_point("classify_input")
33 workflow.add_edge("handle_greeting", END)
34 workflow.add_edge("handle_search", END)
35 
36 app = workflow.compile()
37 
38 # Display the graph
39 try:
40     from IPython.display import Image, display
41 
42     display(Image(app.get_graph().draw_mermaid_png()))
43 except Exception:
44     # This requires some extra dependencies and is optional
45     pass

Calling the graph with Opik tracing enabled

In order to log the execution of the graph, we need to define the OpikTracer callback:

1 from opik.integrations.langchain import OpikTracer
2 
3 tracer = OpikTracer(graph=app.get_graph(xray=True))
4 inputs = {"question": "Hello, how are you?"}
5 result = app.invoke(inputs, config={"callbacks": [tracer]})
6 print(result)

The graph execution is now logged on the Opik platform and can be viewed in the UI:

LangGraph screenshot

1	# We will start by creating simple functions to classify the input question and handle the greeting and search questions.
2	def classify(question: str) -> str:
3	return "greeting" if question.startswith("Hello") else "search"
4
5
6	def classify_input_node(state):
7	question = state.get("question", "").strip()
8	classification = classify(question) # Assume a function that classifies the input
9	return {"classification": classification}
10
11
12	def handle_greeting_node(state):
13	return {"response": "Hello! How can I help you today?"}
14
15
16	def handle_search_node(state):
17	question = state.get("question", "").strip()
18	search_result = f"Search result for '{question}'"
19	return {"response": search_result}

1	from langgraph.graph import StateGraph, END
2
3	from typing import TypedDict, Optional
4
5
6	class GraphState(TypedDict):
7	question: Optional[str] = None
8	classification: Optional[str] = None
9	response: Optional[str] = None
10
11
12	workflow = StateGraph(GraphState)
13	workflow.add_node("classify_input", classify_input_node)
14	workflow.add_node("handle_greeting", handle_greeting_node)
15	workflow.add_node("handle_search", handle_search_node)
16
17
18	def decide_next_node(state):
19	return (
20	"handle_greeting"
21	if state.get("classification") == "greeting"
22	else "handle_search"
23	)
24
25
26	workflow.add_conditional_edges(
27	"classify_input",
28	decide_next_node,
29	{"handle_greeting": "handle_greeting", "handle_search": "handle_search"},
30	)
31
32	workflow.set_entry_point("classify_input")
33	workflow.add_edge("handle_greeting", END)
34	workflow.add_edge("handle_search", END)
35
36	app = workflow.compile()
37
38	# Display the graph
39	try:
40	from IPython.display import Image, display
41
42	display(Image(app.get_graph().draw_mermaid_png()))
43	except Exception:
44	# This requires some extra dependencies and is optional
45	pass

1	from opik.integrations.langchain import OpikTracer
2
3	tracer = OpikTracer(graph=app.get_graph(xray=True))
4	inputs = {"question": "Hello, how are you?"}
5	result = app.invoke(inputs, config={"callbacks": [tracer]})
6	print(result)