Observability for LangGraph with Opik

1
from typing import List, Annotated
2
from pydantic import BaseModel
3
from opik.integrations.langchain import OpikTracer, track_langgraph
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 OpikTracer and track the graph once - no need to pass callbacks anymore!
23
# The graph visualization is automatically extracted by track_langgraph
24
opik_tracer = OpikTracer(
25
    tags=["production"],
26
    metadata={"version": "1.0"}
27
)
28
app = track_langgraph(app, opik_tracer)
29
30
# Now all invocations are automatically tracked
31
for s in app.stream({"messages": [HumanMessage(content = "How to use LangGraph ?")]}):
32
    print(s)
33
34
# No callbacks needed here either!
35
result = app.invoke({"messages": [HumanMessage(content = "How to use LangGraph ?")]})

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()
24
25
# Pass the OpikTracer callback to each invocation
26
for s in app.stream({"messages": [HumanMessage(content = "How to use LangGraph ?")]},
27
                      config={"callbacks": [opik_tracer]}):
28
    print(s)
29
30
result = app.invoke({"messages": [HumanMessage(content = "How to use LangGraph ?")]},
31
                      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, track_langgraph
2
3
# Create OpikTracer and track the graph once
4
# The graph visualization is automatically extracted by track_langgraph
5
opik_tracer = OpikTracer(
6
    project_name="classification-workflow"
7
)
8
app = track_langgraph(app, opik_tracer)
9
10
# Execute the workflow - no callbacks needed!
11
inputs = {"question": "Hello, how are you?"}
12
result = app.invoke(inputs)
13
print(result)
14
15
# Test with a different input - still tracked automatically
16
inputs = {"question": "What is machine learning?"}
17
result = app.invoke(inputs)
18
print(result)

1
import opik_context
2
from opik import track
3
from opik.integrations.langchain import OpikTracer, track_langgraph
4
from langgraph.graph import StateGraph, START, END
5
6
@track
7
def process_data(value: int) -> int:
8
    """Custom tracked function that will be attached to the trace tree."""
9
    return value * 2
10
11
def my_node(state):
12
    current_trace_data = opik_context.get_current_trace_data()
13
    current_span_data = opik_context.get_current_span_data()  # will return the span for `my_node`, created by OpikTracer
14
15
    # This tracked function call will automatically be part of the trace tree
16
    result = process_data(state["value"])
17
    return {"value": result}
18
19
# Build and execute graph
20
graph = StateGraph(dict)
21
graph.add_node("processor", my_node)
22
graph.add_edge(START, "processor")
23
graph.add_edge("processor", END)
24
25
app = graph.compile()
26
opik_tracer = OpikTracer()
27
app = track_langgraph(app, opik_tracer)
28
29
# Synchronous execution - tracked functions work automatically
30
result = app.invoke({"value": 21})

1
from opik import track
2
from opik.integrations.langchain import OpikTracer, track_langgraph, extract_current_langgraph_span_data
3
from langgraph.graph import StateGraph, START, END
4
5
@track
6
def process_data(value: int) -> int:
7
    """Custom tracked function that needs distributed trace headers in async context."""
8
    return value * 2
9
10
async def my_async_node(state, config):
11
    # Extract current span data from LangGraph config. `opik_context` doesn't work here due to langgraph platform limitations related to context propagation.
12
    span_data = extract_current_langgraph_span_data(config)
13
    
14
    # Pass distributed trace headers to attach the tracked function to the trace tree
15
    result = process_data(
16
        state["value"],
17
        opik_distributed_trace_headers=span_data.get_distributed_trace_headers()  # all tracked functions implicitly support this parameter
18
    )
19
    return {"value": result}
20
21
# Build and execute graph
22
graph = StateGraph(dict)
23
graph.add_node("processor", my_async_node)
24
graph.add_edge(START, "processor")
25
graph.add_edge("processor", END)
26
27
app = graph.compile()
28
opik_tracer = OpikTracer()
29
app = track_langgraph(app, opik_tracer)
30
31
# Asynchronous execution - requires explicit trace context propagation
32
result = await app.ainvoke({"value": 21})

1
import opik
2
from opik.integrations.langchain import OpikTracer, track_langgraph, extract_current_langgraph_span_data
3
from langgraph.graph import StateGraph, START, END
4
5
async def my_async_node(state, config):
6
    span_data = extract_current_langgraph_span_data(config)
7
    
8
    # Use context manager with distributed headers
9
    with opik.start_as_current_span(
10
        name="custom_operation",
11
        input={"input": state["value"]},
12
        opik_distributed_trace_headers=span_data.get_distributed_trace_headers()
13
    ) as span_data:
14
        # Your custom logic here
15
        result = state["value"] * 2
16
        span_data.output = {"output": result}
17
    
18
    return {"value": result}
19
20
# Build and execute graph
21
graph = StateGraph(dict)
22
graph.add_node("processor", my_async_node)
23
graph.add_edge(START, "processor")
24
graph.add_edge("processor", END)
25
26
app = graph.compile()
27
opik_tracer = OpikTracer()
28
app = track_langgraph(app, opik_tracer)
29
30
result = await app.ainvoke({"value": 21})

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, track_langgraph
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 and track the graph
38
opik_tracer = OpikTracer()
39
app = track_langgraph(app, opik_tracer)
40
41
thread_id = "e424a45e-7763-443a-94ae-434b39b67b72"
42
config = {"configurable": {"thread_id": thread_id}}
43
44
# Initialize the state
45
state = State(**app.get_state(config).values) or State(messages=[])
46
print("STATE", state)
47
48
# Add the user message
49
state.messages.append(HumanMessage(content="Hello, my name is Bob, how are you doing ?"))
50
# state.messages.append(HumanMessage(content="What is my name ?"))
51
52
result = app.invoke(state, config=config)
53
54
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()