How to add agent analytics to a LangChain app

LangChain gives you the model, the tools, and the loop between them, and its callback system sees every model call and tool call that loop makes. Agent analytics turns those calls into answers about cost, quality, and where users give up.

In this tutorial, I will pass Amplitude's callback handler to a LangChain agent, give each complete invocation its own agent session and cover async calls, and make sure buffered events ship before your process stops.

Prerequisites

• Python 3.10 or later, an authenticated user ID, and a LangChain 1.x app that builds its agent with create_agent. The examples use OpenAI through langchain-openai.
• An Amplitude project with agent analytics enabled, and its project API key. In Amplitude, go to Settings > Projects > your project > General. Agent data can live in the same project as your product analytics, which is the simplest setup.
• The View Agent Analytics Objects permission on your role, which you need to open agent analytics at all.

No Amplitude account yet? A free account is enough for this tutorial; every plan includes a free allotment of agent sessions. Sign up for free to get your project API key.

1. Install the SDK with the LangChain extra

The Python package is amplitude-ai, and it installs framework support as extras. The langchain extra adds the LangChain integration. This tutorial uses amplitude-ai 1.20.1 or later. Install LangChain and its OpenAI provider package next to it. For every option, see the agent analytics SDK reference.

uv add "amplitude-ai[langchain]>=1.20.1" langchain langchain-openai

The docs name the variable AMPLITUDE_AI_API_KEY. It holds your standard project API key. Set it and your OpenAI key in the environment your app runs in.

# .env
AMPLITUDE_AI_API_KEY=your-project-api-key
OPENAI_API_KEY=your-openai-key

Prefer to let a coding agent do the instrumentation? Install the package, then tell your agent to run amplitude-ai --print-guide and follow the guide it prints. The steps below are what it should produce for a LangChain agent.

2. Initialize once in a module of its own

Create one module that constructs the Amplitude client and names your agent at import time. Everything else imports from it.

# app/agent_analytics.py
import os

from amplitude_ai import AIConfig, AmplitudeAI

ai = AmplitudeAI(
    api_key=os.environ["AMPLITUDE_AI_API_KEY"],
    config=AIConfig(
        content_mode="full",  # prompt and response text, with PII redaction
        redact_pii=True,
        debug=os.environ.get("AMPLITUDE_AI_DEBUG") == "true",
    ),
)

# Reuse the named agent and its configuration across requests.
support_agent = ai.agent("support-assistant")

Add one helper for the user ID, and use the same format your product analytics already sends.

# app/identity.py
def amplitude_user_id(user_id):
    # This example serves authenticated users. Never invent a user ID.
    if user_id is None:
        raise ValueError("Pass an authenticated user ID")
    # Prefix integer primary keys: Amplitude's HTTP V2 API requires IDs of
    # at least 5 characters by default, and an ID of 42 is two.
    return f"user-{user_id}"

content_mode="full" is the default and sends prompt and response text after scrubbing emails, phone numbers, SSNs, card numbers, and IP addresses. redact_pii defaults to True from version 1.17.0 and to False before it, so the explicit True keeps older installs safe. The SDK applies the privacy mode to framework integrations too, so the LangChain handler follows these settings. If your conversations are regulated, use "metadata_only": you still get tokens, latency, model, cost, and session grouping, with no text. With AMPLITUDE_AI_DEBUG=true, every tracked event prints to stderr.

3. Build the LangChain agent once

Build the agent at import time, the same way you would without analytics. Don't attach the Amplitude handler here: the agent is shared by every user, and the handler carries one user's identity.

# app/assistant.py
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver


@tool
def lookup_order(order_id: str) -> str:
    """Look up the status of an order by its ID."""
    # Replace with your order service.
    return f"Order {order_id} shipped on Monday."


assistant = create_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[lookup_order],
    # Keeps each conversation's history under its thread_id.
    # Swap in a persistent checkpointer for production.
    checkpointer=InMemorySaver(),
)

Use the canonical model ID, gpt-4o-mini, not a gateway alias. Cost comes from token counts multiplied against a public model-price catalog, and a name the catalog cannot price gets no cost. This example uses completed, non-streaming OpenAI calls through invoke or ainvoke.

4. Pass the handler in each call's config

LangChain takes callback handlers at run time through the callbacks key of a call's config. Create an AmplitudeCallbackHandler per call, pass it there, and make the call inside an agent session keyed by this complete job.

# app/assistant.py (add to the file)
from amplitude_ai.integrations.langchain import AmplitudeCallbackHandler

from .agent_analytics import ai, support_agent
from .identity import amplitude_user_id


def run_turn(user_id, conversation_id, message):
    uid = amplitude_user_id(user_id)

    with support_agent.session(user_id=uid, session_id=conversation_id):
        handler = AmplitudeCallbackHandler(ai, user_id=uid)
        result = assistant.invoke(
            {"messages": [{"role": "user", "content": message}]},
            config={
                "callbacks": [handler],
                "configurable": {"thread_id": conversation_id},
            },
        )

    return result["messages"][-1].content

The job ID does two jobs. LangChain uses it as the thread_id for this invocation's history, and Amplitude uses it as the [Agent] Session ID. One invocation includes the model calls and tool calls needed to answer one request. Generate a fresh ID for each new job and never share it across users.

The with block closes the session after the complete invocation. Amplitude enriches a session once, after its first close, so finish the job before closing. If your application keeps one job open across several HTTP requests, use an explicit session lifecycle instead of opening and closing this scope for every request. Pass a stable user ID or device ID on every session; this example requires an authenticated user ID.

5. Use the same pattern for async calls

In async code, open the session with async with and call ainvoke. The handler goes in the same place.

# app/assistant.py (add to the file)
async def arun_turn(user_id, conversation_id, message):
    uid = amplitude_user_id(user_id)

    async with support_agent.session(user_id=uid, session_id=conversation_id):
        handler = AmplitudeCallbackHandler(ai, user_id=uid)
        result = await assistant.ainvoke(
            {"messages": [{"role": "user", "content": message}]},
            config={
                "callbacks": [handler],
                "configurable": {"thread_id": conversation_id},
            },
        )

    return result["messages"][-1].content

6. Flush before the process exits

The SDK batches events in memory. A script, a notebook cell, a CLI, or a worker that exits right after the agent answers can take the last batch with it. Call ai.shutdown() once when the process is done; it flushes and then closes the analytics client.

# main.py
import uuid

from app.agent_analytics import ai
from app.assistant import run_turn


def main():
    conversation_id = str(uuid.uuid4())
    try:
        print(run_turn(42, conversation_id, "Where is order A-1001?"))
    finally:
        # Ship everything still buffered, then close the client.
        ai.shutdown()


if __name__ == "__main__":
    main()

In a long-running server, call ai.shutdown() from your shutdown hook instead: a SIGTERM handler, or the FastAPI lifespan. Don't call it per request, since it closes the client. On AWS Lambda and other serverless platforms the SDK detects, sessions flush on exit by themselves.

Verify it worked

Run AMPLITUDE_AI_DEBUG=true uv run python main.py, then check in this order:

1. Terminal. Every tracked event prints to stderr as a one-line summary, for example [amplitude-ai] [Agent] AI Response | user=user-42 session=... agent=support-assistant model=gpt-4o-mini .... Nothing printing means the handler never reached the run, usually a config without callbacks. You can also run amplitude-ai doctor, or amplitude-ai verify for local checks.
2. Ingestion Debugger. In Amplitude, click Data, select Sources, then open the Ingestion Debugger tab. The successful requests chart should show your app's uploads. It shows counts, not individual events.
3. User lookup. Go to Users > User Profiles and search user-42. Within ten seconds to a minute the event stream shows [Agent] AI Response events for the model calls and an [Agent] Tool Call for lookup_order. Open an AI Response and confirm [Agent] Model Name, [Agent] Input Tokens, [Agent] Output Tokens, [Agent] Cost USD, [Agent] Latency Ms, [Agent] Session ID (your job ID), and [Agent] Agent ID (support-assistant) are set.
4. Agent analytics. Open agent analytics in the left nav and find the session by your job ID. Quality signals arrive on [Agent] Session Record typically 15 to 20 minutes after the first close, and cover the events received before it.

Troubleshooting

ImportError: cannot import name 'AmplitudeAI' from partially initialized module

A file in your project is named amplitude_ai.py and shadows the SDK package. Rename it, for example to agent_analytics.py, and delete any stale __pycache__ copy.

Nothing prints in the terminal

The handler isn't in the call's config, or the call ran outside the session. Pass config={"callbacks": [handler]} on the invoke or ainvoke call itself, make the call inside the with or async with block, and check that a wrapper function isn't dropping the config argument on its way to the agent.

One user's events show up under another user

The handler was created once and shared. Create AmplitudeCallbackHandler inside the function that serves the request, as in step 4, never at module scope or on the agent.

Events from one invocation have different [Agent] Session ID values

The job ID changes inside the invocation. Generate it once before calling run_turn or arun_turn, and pass the same value as both session_id and thread_id for that complete run.

The last events before exit never arrive

The process ended with a batch still buffered. Call ai.shutdown() in a finally for scripts and workers, or from your server's shutdown hook.

[Agent] Cost USD is empty

The model name isn't one the price catalog knows, such as a gateway alias or a fine-tuned ft: model. Pass the canonical provider model ID. Check that the completed response includes provider token usage.

Full code reference

Every file from this tutorial in one block, the version to hand to an agent.

# -- .env --
# AMPLITUDE_AI_API_KEY=your-project-api-key
# OPENAI_API_KEY=your-openai-key

# -- app/__init__.py --
# (empty)

# -- app/agent_analytics.py --
import os

from amplitude_ai import AIConfig, AmplitudeAI

ai = AmplitudeAI(
    api_key=os.environ["AMPLITUDE_AI_API_KEY"],
    config=AIConfig(
        content_mode="full",
        redact_pii=True,
        debug=os.environ.get("AMPLITUDE_AI_DEBUG") == "true",
    ),
)

support_agent = ai.agent("support-assistant")

# -- app/identity.py --
def amplitude_user_id(user_id):
    if user_id is None:
        raise ValueError("Pass an authenticated user ID")
    return f"user-{user_id}"

# -- app/assistant.py --
from amplitude_ai.integrations.langchain import AmplitudeCallbackHandler
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

from .agent_analytics import ai, support_agent
from .identity import amplitude_user_id


@tool
def lookup_order(order_id: str) -> str:
    """Look up the status of an order by its ID."""
    return f"Order {order_id} shipped on Monday."


assistant = create_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[lookup_order],
    checkpointer=InMemorySaver(),
)


def run_turn(user_id, conversation_id, message):
    uid = amplitude_user_id(user_id)
    with support_agent.session(user_id=uid, session_id=conversation_id):
        handler = AmplitudeCallbackHandler(ai, user_id=uid)
        result = assistant.invoke(
            {"messages": [{"role": "user", "content": message}]},
            config={
                "callbacks": [handler],
                "configurable": {"thread_id": conversation_id},
            },
        )
    return result["messages"][-1].content


async def arun_turn(user_id, conversation_id, message):
    uid = amplitude_user_id(user_id)
    async with support_agent.session(user_id=uid, session_id=conversation_id):
        handler = AmplitudeCallbackHandler(ai, user_id=uid)
        result = await assistant.ainvoke(
            {"messages": [{"role": "user", "content": message}]},
            config={
                "callbacks": [handler],
                "configurable": {"thread_id": conversation_id},
            },
        )
    return result["messages"][-1].content



# -- main.py --
import uuid

from app.agent_analytics import ai
from app.assistant import run_turn


def main():
    conversation_id = str(uuid.uuid4())
    try:
        print(run_turn(42, conversation_id, "Where is order A-1001?"))
    finally:
        ai.shutdown()


if __name__ == "__main__":
    main()

Hand it to your agent

Prefer to have Cursor, Claude Code, or Codex CLI do the wiring? This prompt points the agent at this page and pins the decisions that matter.

Add Amplitude agent analytics to this LangChain app by following
https://amplitude.com/build/langchain-agent-analytics exactly. Install
"amplitude-ai[langchain]>=1.20.1". Create app/agent_analytics.py (never name it
amplitude_ai.py) that builds AmplitudeAI from
os.environ["AMPLITUDE_AI_API_KEY"] and calls ai.agent() once at module
level; never call ai.agent() per request. Keep the LangChain agent
built once at import, with ChatOpenAI(model="gpt-4o-mini") as its
model, and don't attach the Amplitude handler to it.
For each complete invoke or ainvoke job, open
agent.session(user_id, session_id) with a fresh job ID as
session_id ("async with" in async code), create
AmplitudeCallbackHandler(ai, user_id=...) inside the session, and pass
it as config={"callbacks": [handler]} on that call, with the same
job ID as the thread_id. Use the same user ID format as our product
analytics, prefixed so it's at least 5 characters, and require an authenticated user for this example. Never invent
a placeholder user ID. Call ai.shutdown() once on
process exit, never per request. Don't hardcode keys. Report every
file you changed.

What this unlocks

• Product analytics funnels can run from [Agent] Tool Call for lookup_order to the Support Ticket Created event your app already sends, because both carry the same user-42 user ID.
• Feature Experiment can test gpt-4o-mini against another model with [Agent] Cost USD and [Agent] Response Quality as metrics.
• Guides and surveys can reach users whose [Agent] Session Record shows [Agent] Has User Friction true.

FAQ

Does the callback replace LangChain?

No. LangChain still calls the model and tools. The callback records their activity in Amplitude.

When does this example close an agent session?

After one complete invocation, including its model and tool calls. Use a fresh job ID for the next invocation.

Next steps