Callbacks and Observability

🎧 Callbacks & Observability in LangChain

Your AI’s Personal Detective System

The Big Picture: What Are We Learning?

Imagine you have a magical robot helper doing tasks for you. But sometimes you wonder:

• “What is my robot doing right now?”
• “Did something go wrong?”
• “How long did that take?”

Callbacks and Observability are like putting a tiny detective inside your robot. This detective watches everything the robot does and tells you about it!

graph TD
    A["🤖 Your AI Agent"] --> B["📡 Callbacks"]
    B --> C["👀 You See Everything!"]
    C --> D["✅ Success?"]
    C --> E["❌ Error?"]
    C --> F["⏱️ How Long?"]

1. Callbacks System Overview

🎬 The Story

Think of callbacks like a sports announcer at a game. When a player scores, the announcer tells everyone. When there’s a foul, the announcer explains it. The announcer doesn’t play the game—they just report what happens.

In LangChain, callbacks work the same way:

• When your AI starts working → callback announces it
• When AI finishes → callback reports the result
• When something breaks → callback alerts you

What Is a Callback?

A callback is a special function that runs automatically when something happens in your AI system.

Simple Example:

# This is like telling someone:
# "Hey, call me when dinner is ready!"

def my_callback(event):
    print(f"Something happened: {event}")

Why Do We Need Callbacks?

2. Built-in Callback Handlers

🎬 The Story

LangChain comes with pre-made detectives already built in! You don’t need to create them from scratch. It’s like buying a toy with batteries included.

Meet the Built-in Handlers

1. StdOutCallbackHandler (The Printer)

Prints everything to your screen. Like a parrot that repeats everything it sees!

from langchain.callbacks import (
    StdOutCallbackHandler
)

# Create the "printer" handler
handler = StdOutCallbackHandler()

# Use it with your AI
llm.invoke("Hello!", callbacks=[handler])
# Output: Prints what's happening!

2. StreamingStdOutCallbackHandler

Shows text as it’s being created, word by word. Like watching someone type a message in real-time!

from langchain.callbacks.streaming_stdout import (
    StreamingStdOutCallbackHandler
)

# Watch AI think in real-time!
handler = StreamingStdOutCallbackHandler()

3. FileCallbackHandler

Saves everything to a file. Like a diary that writes itself!

from langchain.callbacks import (
    FileCallbackHandler
)

# Save to a log file
handler = FileCallbackHandler("my_log.txt")

Quick Comparison

3. Custom Callback Handlers

🎬 The Story

What if the built-in detectives aren’t enough? What if you want a detective that sends you a text message when something happens? You can build your own!

It’s like building a custom alarm system for your house. You decide what triggers the alarm and what happens when it goes off.

Building Your First Custom Handler

from langchain.callbacks.base import (
    BaseCallbackHandler
)

class MyCustomHandler(BaseCallbackHandler):
    """My special detective!"""

    def on_llm_start(self, *args, **kwargs):
        print("🚀 AI is starting to think...")

    def on_llm_end(self, response, **kwargs):
        print("✅ AI finished thinking!")

    def on_llm_error(self, error, **kwargs):
        print(f"❌ Oops! Error: {error}")

Using Your Custom Handler

# Create your custom detective
my_handler = MyCustomHandler()

# Put the detective to work
llm.invoke(
    "Tell me a joke",
    callbacks=[my_handler]
)

# Output:
# 🚀 AI is starting to think...
# ✅ AI finished thinking!

Real-World Example: Slack Notifier

class SlackNotifier(BaseCallbackHandler):
    """Sends alerts to Slack!"""

    def on_llm_error(self, error, **kwargs):
        # Send message to Slack
        send_to_slack(f"⚠️ AI Error: {error}")

4. Callback Lifecycle Events

🎬 The Story

Imagine watching a chef cook a meal. There are different moments:

1. 👨‍🍳 Chef starts cooking (START)
2. 🍳 Chef is cooking (IN PROGRESS)
3. 🍽️ Food is ready (END)
4. 🔥 Kitchen fire! (ERROR)

Callbacks have the same lifecycle—different events happen at different stages.

The Complete Lifecycle

graph TD
    A["on_llm_start"] --> B["on_llm_new_token"]
    B --> B
    B --> C{Success?}
    C -->|Yes| D["on_llm_end"]
    C -->|No| E["on_llm_error"]

All the Events You Can Catch

Example: Tracking Time

import time

class TimerHandler(BaseCallbackHandler):
    def on_llm_start(self, *args, **kwargs):
        self.start_time = time.time()
        print("⏱️ Timer started!")

    def on_llm_end(self, *args, **kwargs):
        duration = time.time() - self.start_time
        print(f"⏱️ Took {duration:.2f} seconds")

5. LangSmith Integration

🎬 The Story

LangSmith is like having a professional surveillance system instead of just a home camera. It’s made by the same people who created LangChain, so it works perfectly together!

Think of it as upgrading from a flashlight to a lighthouse. Both give light, but one shows you MUCH more!

What Is LangSmith?

LangSmith is a platform for tracking, testing, and debugging your AI applications. It’s like:

• 📊 A dashboard for your AI
• 🔍 A magnifying glass for bugs
• 📈 A report card for performance

Setting Up LangSmith

# Step 1: Set environment variables
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

# Step 2: That's it! LangSmith now tracks
# everything automatically! 🎉

What You See in LangSmith

graph LR
    A["Your App"] -->|Traces| B["LangSmith"]
    B --> C["Dashboard"]
    B --> D["Debug Tools"]
    B --> E["Analytics"]

6. Tracing and Observability

🎬 The Story

Imagine you’re a detective solving a mystery. You need clues—footprints, fingerprints, witnesses. Tracing gives you all the clues about what your AI did!

Observability means you can see inside your system. It’s like having X-ray vision for your code.

What Is a Trace?

A trace is a record of everything that happened during one AI request. Like a receipt that shows every item you bought at a store.

TRACE: "Tell me about cats"
├── LLM Call (2.3 seconds)
│   ├── Input: "Tell me about cats"
│   └── Output: "Cats are..."
├── Tool Call: search (0.5 seconds)
│   └── Result: [cat facts...]
└── Final Response (0.1 seconds)

The Three Pillars of Observability

graph TD
    A["Observability"] --> B["📊 Metrics"]
    A --> C["📝 Logs"]
    A --> D["🔗 Traces"]
    B --> E["Numbers: latency, cost"]
    C --> F["Text: errors, events"]
    D --> G["Paths: request flow"]

Why Tracing Matters

Enabling Tracing

# Method 1: Environment variable
os.environ["LANGCHAIN_TRACING_V2"] = "true"

# Method 2: In code
from langchain.callbacks.tracers import (
    LangChainTracer
)

tracer = LangChainTracer(project_name="my-app")
llm.invoke("Hello", callbacks=[tracer])

7. Manual Tracing Instrumentation

🎬 The Story

Sometimes automatic tracing isn’t enough. What if you want to track your own custom code too? That’s where manual instrumentation comes in.

It’s like adding your own notes to a detective’s report: “I was here at 3pm and saw this.”

What Is Manual Instrumentation?

Manual instrumentation means you add tracing code yourself, exactly where you want it. You decide what to track!

Using @traceable Decorator

The easiest way to add manual tracing:

from langsmith import traceable

@traceable(name="my_custom_function")
def process_data(data):
    # Your code here
    result = do_something(data)
    return result

# Now this function is tracked!
process_data(my_data)

Creating Custom Spans

For more control, create spans manually:

from langsmith import trace

def my_complex_function():
    # Start a custom trace
    with trace("step-1-prepare"):
        data = prepare_data()

    # Another traced section
    with trace("step-2-process"):
        result = process(data)

    return result

Real Example: Tracing a Pipeline

from langsmith import traceable

@traceable(name="full-pipeline")
def run_pipeline(question):

    # Step 1: Research
    with trace("research"):
        docs = search_docs(question)

    # Step 2: AI Processing
    with trace("ai-answer"):
        answer = llm.invoke(question)

    # Step 3: Format
    with trace("formatting"):
        final = format_response(answer)

    return final

What You Get

TRACE: full-pipeline
├── research (0.3s)
├── ai-answer (2.1s)
└── formatting (0.05s)
Total: 2.45s

🎯 Putting It All Together

Here’s how all the pieces connect:

graph TD
    A["Your LangChain App"] --> B["Callbacks"]
    B --> C["Built-in Handlers"]
    B --> D["Custom Handlers"]
    C --> E["StdOut/File/Stream"]
    D --> F["Your Logic"]
    B --> G["LangSmith"]
    G --> H["Traces"]
    H --> I["Dashboard"]
    I --> J["Debug & Improve"]

Quick Reference

🚀 You Did It!

Now you understand:

• ✅ What callbacks are and why they matter
• ✅ How to use built-in handlers
• ✅ How to create custom handlers
• ✅ The lifecycle of callback events
• ✅ How LangSmith provides pro-level observability
• ✅ How tracing helps debug and optimize
• ✅ How to manually instrument your own code

You’re now a LangChain Observability Expert! 🎉

Remember: The best AI systems aren’t just smart—they’re transparent. With callbacks and tracing, you’ll always know exactly what your AI is doing!