Observability for LangGraph with Opik

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
import opik
2
3
# Configure Opik
4
opik.configure(use_local=False)

1
from langgraph.graph import StateGraph, END
2
from typing import TypedDict, Optional
3
4
# Define the graph state
5
class GraphState(TypedDict):
6
    question: Optional[str] = None
7
    classification: Optional[str] = None
8
    response: Optional[str] = None
9
10
# Create the node functions
11
def classify(question: str) -> str:
12
    return "greeting" if question.startswith("Hello") else "search"
13
14
def classify_input_node(state):
15
    question = state.get("question", "").strip()
16
    classification = classify(question)
17
    return {"classification": classification}
18
19
def handle_greeting_node(state):
20
    return {"response": "Hello! How can I help you today?"}
21
22
def handle_search_node(state):
23
    question = state.get("question", "").strip()
24
    search_result = f"Search result for '{question}'"
25
    return {"response": search_result}
26
27
# Create the workflow
28
workflow = StateGraph(GraphState)
29
workflow.add_node("classify_input", classify_input_node)
30
workflow.add_node("handle_greeting", handle_greeting_node)
31
workflow.add_node("handle_search", handle_search_node)
32
33
# Add conditional routing
34
def decide_next_node(state):
35
    return (
36
        "handle_greeting"
37
        if state.get("classification") == "greeting"
38
        else "handle_search"
39
    )
40
41
workflow.add_conditional_edges(
42
    "classify_input",
43
    decide_next_node,
44
    {"handle_greeting": "handle_greeting", "handle_search": "handle_search"},
45
)
46
47
workflow.set_entry_point("classify_input")
48
workflow.add_edge("handle_greeting", END)
49
workflow.add_edge("handle_search", END)
50
51
app = workflow.compile()

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

1
import sqlite3
2
from langgraph.checkpoint.sqlite import SqliteSaver
3
from typing import Annotated
4
from pydantic import BaseModel
5
from opik.integrations.langchain import OpikTracer
6
from langchain_core.messages import HumanMessage
7
from langgraph.graph import StateGraph, START, END
8
from langgraph.graph.message import add_messages
9
from langchain.chat_models import init_chat_model
10
11
llm = init_chat_model("openai:gpt-4.1")
12
13
14
# create your LangGraph graph
15
class State(BaseModel):
16
    messages: Annotated[list, add_messages]
17
18
19
def chatbot(state):
20
    # Typically your LLM calls would be done here
21
    return {"messages": [llm.invoke(state.messages)]}
22
23
24
graph = StateGraph(State)
25
graph.add_node("chatbot", chatbot)
26
graph.add_edge(START, "chatbot")
27
graph.add_edge("chatbot", END)
28
29
# Create a new SqliteSaver instance
30
# Note: check_same_thread=False is OK as the implementation uses a lock
31
# to ensure thread safety.
32
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
33
memory = SqliteSaver(conn)
34
35
app = graph.compile(checkpointer=memory)
36
37
# Create the OpikTracer
38
opik_tracer = OpikTracer(graph=app.get_graph(xray=True))
39
40
thread_id = "e424a45e-7763-443a-94ae-434b39b67b72"
41
config = {"callbacks": [opik_tracer], "configurable": {"thread_id": thread_id}}
42
43
# Initialize the state
44
state = State(**app.get_state(config).values) or State(messages=[])
45
print("STATE", state)
46
47
# Add the user message
48
state.messages.append(HumanMessage(content="Hello, my name is Bob, how are you doing ?"))
49
# state.messages.append(HumanMessage(content="What is my name ?"))
50
51
result = app.invoke(state, config=config)
52
53
print("Result", result)

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