Add LangChain/LangGraph integration example

README.md

@@ -23,6 +23,7 @@ Examples for teams using a specific framework who want to add Opik.
 | | Description |
 |---|---|
 | [integrations/google_adk/agentic_rag](integrations/google_adk/agentic_rag/) | Trace a Google ADK Agentic RAG router with Opik |
+| [integrations/langchain_langgraph](integrations/langchain_langgraph/) | Trace LangChain runnables inside a LangGraph workflow with Opik |
 | [integrations/otel/offline_evaluation](integrations/otel/offline_evaluation/) | OTel tracing alongside Opik's offline evaluation workflow |
 | [integrations/otel/distributed_tracing](integrations/otel/distributed_tracing/) | Stitch out-of-process tool call spans into a single trace |
 

integrations/README.md

@@ -5,4 +5,5 @@ Examples for adding Opik to a specific framework or library. Each folder covers
 | Integration | Description |
 |---|---|
 | [google_adk/](./google_adk/) | Google ADK — Trace an Agentic RAG router with Opik |
+| [langchain_langgraph/](./langchain_langgraph/) | LangChain/LangGraph - Trace runnables and graph branches with Opik |
 | [otel/](./otel/) | OpenTelemetry — send OTel spans to Opik via OTLP |

integrations/langchain_langgraph/.env.example

@@ -0,0 +1,9 @@
+# Opik (Comet-hosted). Leave OPIK_API_KEY/OPIK_WORKSPACE unset to run in DRY_RUN.
+OPIK_API_KEY=your_opik_api_key_here
+OPIK_WORKSPACE=your_workspace
+
+# Self-hosted / on-prem only; defaults to Opik Cloud if unset
+# OPIK_URL_OVERRIDE=https://www.comet.com/opik/api
+
+# Optional project name
+# OPIK_PROJECT_NAME=langchain-langgraph

integrations/langchain_langgraph/.gitignore

@@ -0,0 +1,8 @@
+.env
+.venv
+__pycache__/
+*.pyc
+.tmp/
+*.log
+.ruff_cache/
+uv.lock

integrations/langchain_langgraph/README.md

@@ -0,0 +1,58 @@
+# LangChain + LangGraph + Opik
+
+Trace LangChain runnables inside a LangGraph workflow with Opik.
+
+## What this does
+
+This example builds a small support-router workflow with [LangGraph](https://langchain-ai.github.io/langgraph/)
+and [LangChain](https://python.langchain.com/) runnables. The graph classifies a support question,
+routes it to a branch, and generates a deterministic response. When Opik credentials are set, the
+workflow is wrapped with `track_langgraph()` and traced with `OpikTracer`.
+
+## Prerequisites
+
+This is a `uv` project - dependencies live in `pyproject.toml`.
+
+```bash
+uv sync
+```
+
+Or, with `pip`:
+
+```bash
+pip install opik langchain-core langgraph
+```
+
+| Environment variable | Required | Description |
+|---|---|---|
+| `OPIK_API_KEY` | for a live run | Opik API key from [comet.com/opik](https://www.comet.com/opik). Unset -> DRY_RUN. |
+| `OPIK_WORKSPACE` | for a live run | Your Opik workspace. Unset -> DRY_RUN. |
+| `OPIK_PROJECT_NAME` | no | Project traces are logged to (default `langchain-langgraph`). |
+| `OPIK_URL_OVERRIDE` | no | Base URL for self-hosted Opik (default: Opik Cloud). |
+
+## Running it
+
+```bash
+# Dry-run first - no credentials needed.
+uv run langchain-langgraph-opik --dry-run
+
+# Full run - set credentials, then the same command logs the graph to Opik.
+export OPIK_API_KEY="<your-key>"
+export OPIK_WORKSPACE="<your-workspace>"
+
+uv run langchain-langgraph-opik
+
+# or run it the way CI does:
+bash run.sh
+```
+
+## How it works
+
+1. **LangChain runnables** - `RunnableLambda` wraps the classification and response functions so
+   each unit is visible as a LangChain step.
+2. **LangGraph routing** - `StateGraph` routes the question to greeting, billing, technical, or
+   general response nodes based on the classification.
+3. **Opik tracing** - `OpikTracer` records the graph execution and `track_langgraph()` attaches
+   graph structure and node spans to the trace.
+4. **Dry-run fallback** - missing Opik credentials switch the script into DRY_RUN, which prints the
+   same classification and response locally without sending data.

integrations/langchain_langgraph/langchain_langgraph.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Trace LangChain runnables inside a LangGraph workflow with Opik."""
+
+import argparse
+import os
+import sys
+from typing import TypedDict
+
+OPIK_API_KEY = os.environ.get("OPIK_API_KEY")
+OPIK_WORKSPACE = os.environ.get("OPIK_WORKSPACE")
+OPIK_PROJECT_NAME = os.environ.get("OPIK_PROJECT_NAME", "langchain-langgraph")
+OPIK_URL = os.environ.get("OPIK_URL_OVERRIDE", "https://www.comet.com/opik/api")
+
+DEFAULT_QUESTION = "Hello, I need help understanding my latest invoice."
+DRY_RUN = not (OPIK_API_KEY and OPIK_WORKSPACE)
+
+
+class SupportState(TypedDict, total=False):
+    question: str
+    classification: str
+    response: str
+
+
+def classify_question(question: str) -> str:
+    text = question.casefold()
+    if any(term in text for term in ("bill", "billing", "invoice", "payment", "price", "refund")):
+        return "billing"
+    if any(term in text for term in ("bug", "crash", "error", "login", "broken", "timeout")):
+        return "technical"
+    if any(term in text for term in ("hello", "hi ", "hey", "good morning", "good afternoon")):
+        return "greeting"
+    return "general"
+
+
+def route_by_classification(state: SupportState) -> str:
+    classification = state.get("classification", "general")
+    return {
+        "greeting": "handle_greeting",
+        "billing": "handle_billing",
+        "technical": "handle_technical",
+    }.get(classification, "handle_general")
+
+
+def build_support_response(state: SupportState) -> dict[str, str]:
+    question = state.get("question", "")
+    classification = state.get("classification", "general")
+    responses = {
+        "greeting": "Greeting: welcome the customer and ask how you can help.",
+        "billing": "Billing: route the customer to invoice, payment, or refund support.",
+        "technical": "Technical: collect reproduction details and route to product support.",
+        "general": "General: acknowledge the request and ask one clarifying question.",
+    }
+    return {
+        "response": f"{responses.get(classification, responses['general'])} Original question: {question}",
+    }
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--question", default=DEFAULT_QUESTION, help="Support question to route through the graph."
+    )
+    parser.add_argument(
+        "--thread-id", default="langchain-langgraph-demo", help="Thread ID attached to the trace."
+    )
+    parser.add_argument("--dry-run", action="store_true", help="Print what would happen; do not touch Opik.")
+    return parser
+
+
+def run_dry(question: str, thread_id: str) -> SupportState:
+    classification = classify_question(question)
+    result: SupportState = {
+        "question": question,
+        "classification": classification,
+        **build_support_response({"question": question, "classification": classification}),
+    }
+    print("[DRY RUN] would trace a LangGraph support router to Opik")
+    print(f"  project: {OPIK_PROJECT_NAME}")
+    print(f"  thread_id: {thread_id}")
+    print(f"  opik_url: {OPIK_URL}")
+    print(f"  classification: {result['classification']}")
+    print(f"  response: {result['response']}")
+    return result
+
+
+def build_graph():
+    from langchain_core.runnables import RunnableLambda
+    from langgraph.graph import END, START, StateGraph
+
+    classify_chain = RunnableLambda(classify_question).with_config({"run_name": "classify_question"})
+    response_chain = RunnableLambda(build_support_response).with_config(
+        {"run_name": "build_support_response"}
+    )
+
+    def classify_node(state: SupportState) -> dict[str, str]:
+        return {"classification": classify_chain.invoke(state["question"])}
+
+    def response_node(state: SupportState) -> dict[str, str]:
+        return response_chain.invoke(state)
+
+    workflow = StateGraph(SupportState)
+    workflow.add_node("classify", classify_node)
+    workflow.add_node("handle_greeting", response_node)
+    workflow.add_node("handle_billing", response_node)
+    workflow.add_node("handle_technical", response_node)
+    workflow.add_node("handle_general", response_node)
+    workflow.add_edge(START, "classify")
+    workflow.add_conditional_edges(
+        "classify",
+        route_by_classification,
+        {
+            "handle_greeting": "handle_greeting",
+            "handle_billing": "handle_billing",
+            "handle_technical": "handle_technical",
+            "handle_general": "handle_general",
+        },
+    )
+    workflow.add_edge("handle_greeting", END)
+    workflow.add_edge("handle_billing", END)
+    workflow.add_edge("handle_technical", END)
+    workflow.add_edge("handle_general", END)
+    return workflow.compile()
+
+
+def run_live(question: str, thread_id: str) -> SupportState:
+    from opik.integrations.langchain import OpikTracer, track_langgraph
+
+    opik_tracer = OpikTracer(
+        project_name=OPIK_PROJECT_NAME,
+        tags=["langchain", "langgraph", "support-router"],
+        metadata={"example": "langchain_langgraph"},
+    )
+    app = track_langgraph(build_graph(), opik_tracer)
+    result = app.invoke(
+        {"question": question},
+        config={"configurable": {"thread_id": thread_id}},
+    )
+    opik_tracer.flush()
+    return result
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    dry_run = DRY_RUN or args.dry_run
+
+    if dry_run:
+        if not args.dry_run:
+            print("OPIK_API_KEY / OPIK_WORKSPACE not set - running in DRY_RUN.", file=sys.stderr)
+        run_dry(args.question, args.thread_id)
+        return 0
+
+    result = run_live(args.question, args.thread_id)
+    print(result["response"])
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

integrations/langchain_langgraph/pyproject.toml

@@ -0,0 +1,35 @@
+[project]
+name = "langchain-langgraph"
+version = "0.1.0"
+description = "Trace LangChain runnables and LangGraph workflows to Opik."
+readme = "README.md"
+requires-python = ">=3.12,<3.14"
+dependencies = [
+    "langchain-core",
+    "langgraph",
+    "opik>=2.0",
+]
+
+[project.scripts]
+langchain-langgraph-opik = "langchain_langgraph:main"
+
+[dependency-groups]
+dev = [
+    "pytest",
+    "ruff",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+# WHY: single top-level module (not a src/ package) - point hatchling at the one file.
+[tool.hatch.build.targets.wheel]
+include = ["langchain_langgraph.py"]
+
+[tool.ruff]
+line-length = 110
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B"]

integrations/langchain_langgraph/run.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+set -e
+
+# Entry point CI runs for this example. With no Opik credentials it falls back to
+# DRY_RUN and exits 0; with credentials set it logs the graph to Opik.
+uv sync
+export OPIK_PROJECT_NAME="langchain-langgraph"
+uv run langchain-langgraph-opik

integrations/langchain_langgraph/tests/test_langchain_langgraph.py

@@ -0,0 +1,32 @@
+from langchain_langgraph import build_support_response, classify_question, route_by_classification
+
+
+def test_classifies_greeting_questions() -> None:
+    assert classify_question("Hello, can you help me?") == "greeting"
+
+
+def test_classifies_billing_questions() -> None:
+    assert classify_question("Can I get a refund for my invoice?") == "billing"
+
+
+def test_billing_terms_take_priority_over_greeting() -> None:
+    assert classify_question("Hello, I need help with my latest invoice") == "billing"
+
+
+def test_classifies_technical_questions() -> None:
+    assert classify_question("The login page throws an error") == "technical"
+
+
+def test_routes_to_node_for_classification() -> None:
+    assert route_by_classification({"classification": "billing"}) == "handle_billing"
+
+
+def test_builds_response_from_classification() -> None:
+    result = build_support_response(
+        {
+            "question": "Can I get a refund?",
+            "classification": "billing",
+        }
+    )
+
+    assert result["response"].startswith("Billing:")