LangChain

Overview

Why monitor LangChain?

LangChain is great for building chains — but chains break in subtle ways. A retriever returns zero results. An LLM call times out silently. A prompt template produces malformed output that downstream steps accept without error. Without observability, these failures are invisible until a user complains.

NotiLens gives you two layers: auto-patch for immediate zero-code observability, and manual instrumentation for fine-grained control over what you track and when.

Auto-Patch

Installation

pip install notilens langchain langchain-openai

Setup

Add patch=True to your NotiLens init. Your existing LangChain code stays unchanged — NotiLens hooks in via LangChain's callback system.

import notilens
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

nl = notilens.init(
    name="support-bot",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True       # hooks into LangChain's callback system
)

# Existing chain — no changes needed
llm    = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "You are a helpful support agent. Answer: {question}"
)
chain  = prompt | llm

# NotiLens automatically tracks this call: start, tokens, cost, output
result = chain.invoke({"question": "How do I reset my password?"})
print(result.content)

What gets tracked automatically

Every event includes token count, cost, model name, and latency — no manual extraction needed.

Real-world example — RAG customer support pipeline

A support bot that retrieves relevant docs from a vector store, then generates an answer. With patch=True, every LLM call inside the retrieval chain is tracked — including how many tokens the retrieved context consumed and whether the answer was actually generated.

import notilens
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

nl = notilens.init(
    name="support-rag",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True
)

# Build the RAG chain
embeddings  = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./kb", embedding_function=embeddings)
retriever   = vectorstore.as_retriever(search_kwargs={"k": 4})

prompt = ChatPromptTemplate.from_template("""
You are a support agent. Use the context to answer the question.
Context: {context}
Question: {question}
""")

llm   = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
)

# NotiLens tracks: embeddings call, retrieval, LLM call, output
answer = chain.invoke("Why is my payment failing?")

# NotiLens will alert if:
# - The LLM call takes > 2x the usual latency
# - Token cost spikes (retrieved context too large)
# - output.failed fires (chain throws an exception)

Manual

Manual Instrumentation

Use manual instrumentation when you need to track steps between LLM calls — document fetching, preprocessing, validation — or when you want human review before the chain continues.

Real-world use: a content generation pipeline that fetches a brief, calls the LLM, runs a quality check, then waits for editor approval before publishing.

import notilens
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

nl  = notilens.init(name="content-pipeline", token="YOUR_TOKEN", secret="YOUR_SECRET")
run = nl.task("article-generation")
run.start()

try:
    # Step 1 — fetch brief from CMS
    run.progress("Fetching article brief from CMS")
    brief = cms.get_brief(brief_id)

    # Step 2 — build and invoke chain
    run.progress("Generating article draft")
    llm    = ChatOpenAI(model="gpt-4o-mini")
    prompt = ChatPromptTemplate.from_template(
        "Write a 500-word article based on this brief: {brief}"
    )
    chain  = prompt | llm
    result = chain.invoke({"brief": brief.content})

    # Step 3 — track cost
    if hasattr(result, "usage_metadata"):
        run.metric("tokens", result.usage_metadata.get("total_tokens", 0))
        run.metric("cost", result.usage_metadata.get("total_tokens", 0) * 0.00015 / 1000)

    # Step 4 — output ready, needs editor approval before publishing
    run.output_generated(f"Draft ready: '{brief.title}'")
    run.input_required("Editor review required before publish")

    # Step 5 — wait for decision (updated externally)
    decision = wait_for_approval(brief_id)
    if decision == "approved":
        run.input_approved("Approved — publishing")
        publish(result.content)
        run.complete("Article published")
    else:
        run.input_rejected(f"Rejected: {decision['reason']}")
        run.fail("Article rejected")

except Exception as e:
    run.fail(str(e))

When to use

Auto-Patch vs Manual — which to choose

force_send

force_send

Auto-patched events follow the same defaults as the Python SDK — output.failed always delivers, ai.call.start routes through ML. Override per-call if needed. See full defaults →