diff --git a/README.md b/README.md index ce144e3..60bed88 100644 --- a/README.md +++ b/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 | diff --git a/integrations/README.md b/integrations/README.md index d8ed601..5fcf3c5 100644 --- a/integrations/README.md +++ b/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 | diff --git a/integrations/langchain_langgraph/.env.example b/integrations/langchain_langgraph/.env.example new file mode 100644 index 0000000..5c0a14d --- /dev/null +++ b/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 diff --git a/integrations/langchain_langgraph/.gitignore b/integrations/langchain_langgraph/.gitignore new file mode 100644 index 0000000..b40e307 --- /dev/null +++ b/integrations/langchain_langgraph/.gitignore @@ -0,0 +1,8 @@ +.env +.venv +__pycache__/ +*.pyc +.tmp/ +*.log +.ruff_cache/ +uv.lock diff --git a/integrations/langchain_langgraph/README.md b/integrations/langchain_langgraph/README.md new file mode 100644 index 0000000..05e65d9 --- /dev/null +++ b/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="" +export OPIK_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. diff --git a/integrations/langchain_langgraph/langchain_langgraph.py b/integrations/langchain_langgraph/langchain_langgraph.py new file mode 100644 index 0000000..3041a52 --- /dev/null +++ b/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()) diff --git a/integrations/langchain_langgraph/pyproject.toml b/integrations/langchain_langgraph/pyproject.toml new file mode 100644 index 0000000..bfad73a --- /dev/null +++ b/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"] diff --git a/integrations/langchain_langgraph/run.sh b/integrations/langchain_langgraph/run.sh new file mode 100755 index 0000000..5354cc3 --- /dev/null +++ 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 diff --git a/integrations/langchain_langgraph/tests/test_langchain_langgraph.py b/integrations/langchain_langgraph/tests/test_langchain_langgraph.py new file mode 100644 index 0000000..ae6f6f1 --- /dev/null +++ b/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:")