Python Integration Guide

LangWatch library is the easiest way to integrate your Python application with LangWatch, the messages are synced on the background so it doesn’t intercept or block your LLM calls.

Protip: wanna to get started even faster? Copy our llms.txt and ask an AI to do this integration

Prerequisites

Obtain your LANGWATCH_API_KEY from the LangWatch dashboard.

Installation

pip install langwatch

Configuration

Ensure LANGWATCH_API_KEY is set:

export LANGWATCH_API_KEY='your_api_key_here'

export LANGWATCH_API_KEY='your_api_key_here'

You can set LANGWATCH_API_KEY globally at runtime:

import langwatch
import os

langwatch.api_key = os.getenv("LANGWATCH_API_KEY")

Or on the specific trace being tracked:

import langwatch
import os

@langwatch.trace(api_key=os.getenv("LANGWATCH_API_KEY"))
def main():
    ...

Capturing Messages

Each message triggering your LLM pipeline as a whole is captured with a Trace.
A Trace contains multiple Spans, which are the steps inside your pipeline.
- A span can be an LLM call, a database query for a RAG retrieval, or a simple function transformation.
- Different types of Spans capture different parameters.
- Spans can be nested to capture the pipeline structure.
Traces can be grouped together on LangWatch Dashboard by having the same thread_id in their metadata, making the individual messages become part of a conversation.
- It is also recommended to provide the user_id metadata to track user analytics.

Create a Trace

To capture traces and spans, start by adding the @langwatch.trace() decorator to the function that starts your LLM pipeline. Here it is represented by the main() function, but it can be your endpoint call or your class method that triggers the whole generation.

import langwatch

@langwatch.trace()
def main():
    ...

This is the main entry point for your trace, and all spans called from here will be collected automatically to LangWatch in the background.

On short-live environments like Lambdas or Serverless Functions, be sure to call
langwatch.get_current_trace().send_spans() before your trace function ends to wait for all pending requests to be sent before the runtime is destroyed.

Capturing LLM Spans

LangWatch provides some utilities to automatically capture spans for popular LLM frameworks.

For OpenAI, you can use the autotrack_openai_calls() function to automatically capture LLM spans for OpenAI calls for the current trace.

import langwatch
from openai import OpenAI

client = OpenAI()

@langwatch.trace()
def main():
    langwatch.get_current_trace().autotrack_openai_calls(client)
    ...

That’s enough to have your OpenAI calls collected and visible on LangWatch dashboard:

For OpenAI, you can use the autotrack_openai_calls() function to automatically capture LLM spans for OpenAI calls for the current trace.

import langwatch
from openai import OpenAI

client = OpenAI()

@langwatch.trace()
def main():
    langwatch.get_current_trace().autotrack_openai_calls(client)
    ...

That’s enough to have your OpenAI calls collected and visible on LangWatch dashboard:

For Azure OpenAI, you can use the autotrack_openai_calls() function to automatically capture LLM spans for Azure OpenAI calls for the current trace.

import langwatch
from openai import AzureOpenAI

client = AzureOpenAI()

@langwatch.trace()
def main():
    langwatch.get_current_trace().autotrack_openai_calls(client)
    ...

That’s enough to have your Azure OpenAI calls collected and visible on LangWatch dashboard:

You can use LiteLLM to call OpenAI, Anthropic, Gemini, Groq Llama 3 and over 100+ LLM models.

And for tracking it all with LangWatch, you can use the autotrack_litellm_calls() function to automatically capture LLM spans for LiteLLM calls for the current trace.

import langwatch
import litellm

@langwatch.trace()
def main():
    langwatch.get_current_trace().autotrack_litellm_calls(litellm)

    response = litellm.completion(
        ...
    )

Since we patch the completion method of the litellm module, you must use litellm.completion() instead of just completion() when calling your LLM, otherwise LangWatch will not be able to capture the spans.

That’s enough to have your LiteLLM calls collected and visible on LangWatch dashboard:

DSPy is the LLM framework that automatically optimizes prompts, you can use LangWatch both for visualizing the optimization process, and for tracking the calls during inference as this guide shows.

To track DSPy programs, you can use the autotrack_dspy() function to automatically capture DSPy modules forward pass, retrievers and LLM calls for the current trace.

import langwatch
import dspy

@langwatch.trace()
def main():
    langwatch.get_current_trace().autotrack_dspy()

    program = MyDspyProgram()
    response = program(
        ...
    )

That’s enough to have your DSPy traces collected and visible on LangWatch dashboard:

For LangChain, you can automatically capture every step of your chain as a span by getting a LangChain callback for the current trace with get_langchain_callback().

import langwatch

@langwatch.trace()
def main():
    ...
    chain.invoke(
        {"input": user_input},
        # Add the LangWatch callback when invoking your chain
        {"callbacks": [langwatch.get_current_trace().get_langchain_callback()]},
    )

That’s enough to have your LangChain calls collected and visible on LangWatch dashboard:

Check out for more python integration examples on the examples folder on our GitHub repo.

Adding metadata

You can add metadata to track the user_id and current conversation thread_id, this is highly recommended to unlock better conversation grouping and user analytics on LangWatch.

import langwatch

@langwatch.trace()
def main():
    langwatch.get_current_trace().update(metadata={"user_id": "user_id", "thread_id": "thread_id"})
    ...

You can also add custom labels to your trace to help you better filter and group your traces, or even trigger specific evaluations and alerts.

import langwatch

@langwatch.trace()
def main():
    langwatch.get_current_trace().update(metadata={"labels": ["production"]})
    ...

Check out the reference to see all the available trace properties.

Changing the Message Input and Output

By default, the main input and output of the trace displayed on LangWatch is captured from the arguments and return value of the top-level decorated function and heuristics try to extract the human-readable message from it automatically.

However, sometimes more complex structures are used and the messages might not end up very human-readable on LangWatch, for example:

To make the messages really easy to read in the list and through the whole conversation, you can manually set what should the input and output of the trace be, by calling .update(input=...) and .update(output=...) on the current trace:

import langwatch

@langwatch.trace()
def main(inputs):
    # Update the input of the trace with the user message or any other human-readable text
    langwatch.get_current_trace().update(input=inputs.question)

    ...

    # Then, before returning, update the output of the trace with final response
    langwatch.get_current_trace().update(output=response)

    return response

This will make the messages on LangWatch look like this:

Capturing a RAG span

RAG is a combination of a retrieval and a generation step, LangWatch provides a special span type for RAG that captures both steps separately which allows to capture the contexts being used by the LLM on your pipeline. By capturing the contexts, you unlock various uses of it on LangWatch, like RAG evaluators such as Faitfhfulness and Context Relevancy, and analytics on which documents are being used the most.

To capture a RAG span, you can use the @langwatch.span(type="rag") decorator, along with a call to .update() to add the contexts to the span:

@langwatch.span(type="rag")
def rag_retrieval():
    # the documents you retrieved from your vector database
    search_results = ["France is a country in Europe.", "Paris is the capital of France."]

    # capture them on the span contexts before returning
    langwatch.get_current_span().update(contexts=search_results)

    return search_results

If you have document or chunk ids from the results, we recommend you can to capture them along with the id using RAGChunk, as this allows them to be grouped together and generate documents analytics on LangWatch dashboard:

from langwatch.types import RAGChunk

@langwatch.span(type="rag")
def rag_retrieval():
    # the documents you retrieved from your vector database
    search_results = [
        {
            "id": "doc-1",
            "content": "France is a country in Europe.",
        },
        {
            "id": "doc-2",
            "content": "Paris is the capital of France.",
        },
    ]

    # capture then on the span contexts with RAGChunk before returning
    langwatch.get_current_span().update(
        contexts=[
            RAGChunk(
                document_id=document["id"],
                content=document["content"],
            )
            for document in search_results
        ]
    )

    return search_results

Then you’ll be able to see the captured contexts that will also be used later on for evaluatios on LangWatch dashboard:

To capture a RAG span, you can use the @langwatch.span(type="rag") decorator, along with a call to .update() to add the contexts to the span:

@langwatch.span(type="rag")
def rag_retrieval():
    # the documents you retrieved from your vector database
    search_results = ["France is a country in Europe.", "Paris is the capital of France."]

    # capture them on the span contexts before returning
    langwatch.get_current_span().update(contexts=search_results)

    return search_results

from langwatch.types import RAGChunk

@langwatch.span(type="rag")
def rag_retrieval():
    # the documents you retrieved from your vector database
    search_results = [
        {
            "id": "doc-1",
            "content": "France is a country in Europe.",
        },
        {
            "id": "doc-2",
            "content": "Paris is the capital of France.",
        },
    ]

    # capture then on the span contexts with RAGChunk before returning
    langwatch.get_current_span().update(
        contexts=[
            RAGChunk(
                document_id=document["id"],
                content=document["content"],
            )
            for document in search_results
        ]
    )

    return search_results

Then you’ll be able to see the captured contexts that will also be used later on for evaluatios on LangWatch dashboard:

When using LangChain, generally your RAG happens by calling a Retriever.

We provide a utility langwatch.langchain.capture_rag_from_retriever to capture the documents found by the retriever and convert it into a LangWatch compatible format for tracking. For that you need to pass the retriever as first argument, and then a function to map each document to a RAGChunk, like in the example below:

import langwatch
from langwatch.types import RAGChunk

@langwatch.trace()
def main():
    retriever = ...
    retriever_tool = create_retriever_tool(
        langwatch.langchain.capture_rag_from_retriever(
            retriever,
            lambda document: RAGChunk(
                document_id=document.metadata["source"],
                content=document.page_content
            ),
        ),
        "langwatch_search",
        "Search for information about LangWatch. For any questions about LangWatch, use this tool if you didn't already",
    )

    tools = [retriever_tool]
    model = ChatOpenAI(streaming=True)
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You are a helpful assistant that only reply in short tweet-like responses, using lots of emojis and use tools only once.\n\n{agent_scratchpad}",
            ),
            ("human", "{question}"),
        ]
    )
    agent = create_tool_calling_agent(model, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    return executor.invoke(user_input, config=RunnableConfig(
        callbacks=[langwatch.get_current_trace().get_langchain_callback()]
    ))

Alternatively, if you don’t use retrievers, but still want to capture the context for example from a tool call that you do, we also provide a utility langwatch.langchain.capture_rag_from_tool to capture RAG contexts around a tool. For that you need to pass the tool as first argument, and then a function to map the tool’s output to RAGChunks, like in the example below:

import langwatch
from langwatch.types import RAGChunk

@langwatch.trace()
def main():
    my_custom_tool = ...
    wrapped_tool = langwatch.langchain.capture_rag_from_tool(
        my_custom_tool, lambda response: [
          RAGChunk(
            document_id=response["id"], # optional
            chunk_id=response["chunk_id"], # optional
            content=response["content"]
          )
        ]
    )

    tools = [wrapped_tool] # use the new wrapped tool in your agent instead of the original one
    model = ChatOpenAI(streaming=True)
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You are a helpful assistant that only reply in short tweet-like responses, using lots of emojis and use tools only once.\n\n{agent_scratchpad}",
            ),
            ("human", "{question}"),
        ]
    )
    agent = create_tool_calling_agent(model, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    return executor.invoke(user_input, config=RunnableConfig(
        callbacks=[langWatchCallback]
    ))

Then you’ll be able to see the captured contexts that will also be used later on for evaluatios on LangWatch dashboard:

Capturing other spans

To be able to inspect and debug each step of your pipeline along with the LLM calls, you can use the @langwatch.span() decorator. You can pass in different types to categorize your spans.

import langwatch

@langwatch.span()
def database_query():
    ...

@langwatch.span(type="tool")
def weather_forecast(city: str):
    ...

@langwatch.span(type="rag")
def rag_retrieval():
    ...

# You can manually track llm calls too if the automatic capture is not enough for your use case
@langwatch.span(type="llm")
def llm_call():
    ...

@langwatch.trace()
def main():
    ...

The input and output of the decorated function are automatically captured in the span, to disable that, you can set capture_input and capture_output to False:

@langwatch.span(capture_input=False, capture_output=False)
def database_query():
    ...

You can also modify the current spans attributes, either on the decorator by using .update() on the current span:

@langwatch.span(type="llm", name="custom_name")
def llm_call():
    langwatch.get_current_span().update(model="my-custom-model")
    ...

Check out the reference to see all the available span properties.

Capturing custom evaluation results

LangWatch Evaluators can run automatically on your traces, but if you have an in-house custom evaluator, you can also capture the evaluation results of your custom evaluator on the current trace or span by using the .add_evaluation method:

import langwatch

@langwatch.span(type="evaluation")
def evaluation_step():
    ... # your custom evaluation logic

    langwatch.get_current_span().add_evaluation(
        name="custom evaluation", # required
        passed=True,
        score=0.5,
        label="category_detected",
        details="explanation of the evaluation results",
    )

The evaluation name is required and must be a string. The other fields are optional, but at least one of passed, score or label must be provided.

Synchronizing your message IDs with LangWatch traces

If you store the messages in a database on your side as well, you set the trace_id of the current trace to the same one of the message on your side, this way your system will be in sync with LangWatch traces, making it easier to investigate later on.

@langwatch.trace()
def main():
    ...
    langwatch.get_current_trace().update(trace_id=message_id)
    ...

Optimization Studio

Monitoring Integration

Evaluations

Guardrails

User Events

DSPy Visualization

More Features

API Endpoints

Support

Python Integration Guide

Prerequisites

Installation

Configuration

Capturing Messages

Create a Trace

Capturing LLM Spans

Adding metadata

Changing the Message Input and Output

Capturing a RAG span

Capturing other spans

Capturing custom evaluation results

Synchronizing your message IDs with LangWatch traces

Optimization Studio

Monitoring Integration

Evaluations

Guardrails

User Events

DSPy Visualization

More Features

API Endpoints

Support

​Prerequisites

​Installation

​Configuration

​Capturing Messages

​Create a Trace

​Capturing LLM Spans

​Adding metadata

​Changing the Message Input and Output

​Capturing a RAG span

​Capturing other spans

​Capturing custom evaluation results

​Synchronizing your message IDs with LangWatch traces

Prerequisites

Installation

Configuration

Capturing Messages

Create a Trace

Capturing LLM Spans

Adding metadata

Changing the Message Input and Output

Capturing a RAG span

Capturing other spans

Capturing custom evaluation results

Synchronizing your message IDs with LangWatch traces