uselemma-tracing

HTTP tracing SDK for AI agents. The primary API sends trace payloads directly to Lemma over HTTP.

Installation

pip install uselemma-tracing

Quick Start

from uselemma_tracing import Lemma

lemma = Lemma(release="1.8.3")  # or set LEMMA_RELEASE

def run(trace):
    docs = search_docs(user_message)
    trace.record_tool(
        name="search_docs",
        input={"query": user_message},
        output=docs,
        tool_parameters={"query": "string"},
    )

    response = call_model(user_message, docs)
    trace.record_generation(
        name="draft-reply",
        input=response.messages,
        output=response.text,
        model="gpt-4o",
        llm_input_messages=[{"role": "user", "content": user_message}],
        llm_invocation_parameters={"temperature": 0.2},
    )

    return response.text

answer = lemma.trace(
    "support-agent",
    run,
    input=user_message,
    thread_id=conversation_id,
    user_id=user.id,
)

lemma.trace() measures the trace from callback start to completion. Use async_trace() for async callbacks.

Pass release (or set LEMMA_RELEASE) to stamp the running app version on every ingest payload. An explicit constructor value wins. Empty or invalid values are omitted.

Live Spans

def run(trace):
    span = trace.start_span(name="retrieve-context", input=query)
    try:
        docs = retrieve(query)
        span.end(output={"count": len(docs)})
        return docs
    except Exception as error:
        span.end(status="ERROR", error=error)
        raise

Live handles know their start time when created and their end time when .end() is called, so you usually do not pass duration_ms. Pass duration_ms only when replaying historical work or overriding the measured duration with a value from another timer.

For one-off records where you already measured the work, pass duration_ms on the record call:

trace.record_generation(
    name="answer",
    output=text,
    model="gpt-4o",
    duration_ms=measured_model_ms,
)

User-facing messaging tools

When a tool delivers the agent's response to the end user, pass the exact display text as user_facing_message. Lemma renders that text as an assistant message while preserving the complete tool input and output in the span detail:

tool_input = {
    "message": "Your order arrives Friday.",
    "send_as_voice_note": False,
    "should_terminate": True,
}

trace.record_tool(
    name="send_whatsapp",
    input=tool_input,
    output={"delivered": True},
    user_facing_message=tool_input["message"],
)

The tool's own schema can call the value message, text, body, or anything else. Lemma never guesses which input field the user saw. Omit user_facing_message for internal tools; their payload and rendering are unchanged.

The same handle pattern is available for tool calls and generations:

tool = trace.start_tool(name="search_docs", input={"query": query})
docs = search_docs(query)
tool.end(output=docs)

generation = trace.start_generation(name="answer", input=messages)
response = call_model(messages)
generation.end(output=response.text)

Sending a Trace You Built Yourself

trace() assumes the client owns the trace lifecycle within a single process. When the producer lives elsewhere — a cross-process buffer, a queue worker, a batch backfill — build a TraceContext yourself and deliver it with ingest():

from uselemma_tracing import Lemma, TraceContext

lemma = Lemma()

context = TraceContext(
    id=turn_id,  # stable id for this execution (use for retries)
    name=prompt,
    input=prompt,
    thread_id=conversation_id,
)
context.record_tool(name="search_docs", input=query, output=docs, duration_ms=25)
context.record_generation(name="answer", model="gpt-4o", output=final_answer)
context.output(final_answer)

lemma.ingest(context, started_at=started_at)

ingest() POSTs one payload. Deliver one complete trace when the execution (agent turn) finishes: root input/output, thread/user, and all child spans in one call. This is required — patching a trace over time is not currently supported.

ingest() is not an incremental merge API: omitted root fields do not preserve prior values, and after Lemma processes the trace once, a later re-delivery does not re-run issue extraction (occasional late child spans may still append to the tree for display). Retries of the same complete payload are safe — already-stored span IDs are skipped — so a failed send can be retried as-is. It raises on a non-2xx response and never mutates the trace's status.

Automatic delivery (trace / async_trace) fails open: a Lemma ingest 4xx/5xx or network error is logged in debug mode and dropped so it cannot fail the caller's application. LangChain and OpenAI Agents flush through this path. Use ingest() when you need a failed send to raise so you can retry.

OpenAI Agents SDK

Install the OpenAI Agents extra and register the Lemma processor:

pip install "uselemma-tracing[openai-agents]" openai-agents

from agents import Agent, Runner
from uselemma_tracing import instrument_openai_agents

instrument_openai_agents()

agent = Agent(
    name="support-agent",
    instructions="Answer customer questions clearly and concisely.",
)

async def call_agent(user_message: str):
    result = await Runner.run(agent, user_message)
    return result.final_output

The processor creates one Lemma trace for each OpenAI Agents trace with root current-turn input, final output or terminal error, promoted thread_id / user_id, and wall-clock bounds from child spans. Generation/response spans become Lemma generations, function spans become Lemma tool spans, and parent IDs are preserved so tools stay nested under the generation or agent span that called them.

Pass OpenAI Agents group_id for thread_id and metadata user_id / userId for user_id. Call force_flush() / shutdown() to finalize open traces once.

Enable debug mode to validate live span shape while developing:

from uselemma_tracing import enable_debug_mode

enable_debug_mode()

Prompts, tool inputs, outputs, generated text, and error messages are always recorded — Lemma cannot show what a run consumed, produced, or why it failed without them.

LangChain and LangGraph

Install the optional integration dependency and pass langchain() as a callback handler. Each root run owns one Lemma trace with current-turn input, final output or root error, promoted thread_id / user_id, typed nested generations/tools/spans, and real wall-clock bounds. Call flush() / shutdown() to finalize open traces.

pip install "uselemma-tracing[langchain]" langchain-openai

from langchain_openai import ChatOpenAI
from uselemma_tracing import langchain

handler = langchain(
    agent_name="support-agent",
    thread_id_key="conversation_id",
    user_id_key="user_id",
)
model = ChatOpenAI(model="gpt-4o", callbacks=[handler])
response = model.invoke(
    user_message,
    config={"metadata": {"conversation_id": thread_id, "user_id": user_id}},
)
handler.flush()

langgraph() is the same LangChain callback adapter with a LangGraph default trace name (langgraph-agent):

pip install "uselemma-tracing[langgraph]"

from uselemma_tracing import langgraph

result = graph.invoke(
    {"input": user_message},
    {"callbacks": [langgraph(agent_name="support-graph")]},
)

Prompts, tool inputs, outputs, generated text, and error messages are always recorded.

Supported Contract Fields

Use native SDK keyword arguments for OpenInference-style fields:

• LLM: llm_model_name, llm_provider, llm_system, llm_invocation_parameters, llm_input_messages, llm_output_messages, llm_tools, usage / input_tokens / output_tokens / cache and reasoning kwargs (omit when the provider did not supply them — never invent zeros), and prompt template fields
• provenance: every span includes lemma.sdk.language and lemma.sdk.integration (manual by default; framework integrations override)
• tools: tool_description, tool_parameters, user_facing_message
• embeddings and rerankers: embedding_model_name, embedding_invocation_parameters, embedding_embeddings, reranker_model_name, reranker_input_documents, reranker_output_documents

Use attributes for raw attributes that do not yet have a native SDK keyword.

Configuration

The SDK sends to {base_url}/traces/ingest.

You can pass configuration directly to the constructor instead of using environment variables:

lemma = Lemma(
    api_key="sk_...",
    project_id="proj_...",
    base_url="https://api.uselemma.ai",
)

Debug Mode

Debug mode logs trace starts, span starts, span completions, send attempts, and send results as they happen:

from uselemma_tracing import enable_debug_mode

enable_debug_mode()

You can also set LEMMA_DEBUG=1 (true also works). Use this when validating that spans are created in the expected order and the SDK is sending to the intended URL.

License

MIT