Metadata-Version: 2.5
Name: uselemma-tracing
Version: 7.11.0
Summary: HTTP tracing SDK for Lemma
Project-URL: Homepage, https://github.com/uselemma/lemma/tree/main/packages/py/tracing#readme
Project-URL: Repository, https://github.com/uselemma/lemma
Project-URL: Issues, https://github.com/uselemma/lemma/issues
License-Expression: MIT
License-File: LICENSE
Keywords: llm,monitoring,observability,sdk,tracing
Requires-Python: >=3.9
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Provides-Extra: langchain
Requires-Dist: langchain>=0.3.0; extra == 'langchain'
Provides-Extra: langgraph
Requires-Dist: langchain>=0.3.0; extra == 'langgraph'
Requires-Dist: langgraph>=0.2.0; extra == 'langgraph'
Provides-Extra: openai-agents
Requires-Dist: openai-agents>=0.17.0; (python_version >= '3.10') and extra == 'openai-agents'
Description-Content-Type: text/markdown

# uselemma-tracing

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

## Installation

```bash
pip install uselemma-tracing
```

## Quick Start

```python
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

```python
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:

```python
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:

```python
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:

```python
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()`:

```python
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.

## One turn across processes

`thread_id` correlates **turns** of a conversation. It is not how you glue a
host process and an E2B-style sandbox into one turn. The host mints a versioned
context token, the child records a serializable journal without a Lemma API key,
and the host applies the journal then `ingest()`s once.

```python
import json
from uselemma_tracing import Lemma, attach_turn

lemma = Lemma()
turn = lemma.start_turn(
    "agent-turn",
    input=user_message,
    thread_id=conversation_id,
)
sandbox = turn.start_span(name="e2b-sandbox")
token = json.dumps(turn.export(parent_span_id=sandbox.id))
# pass token to the child on the existing channel, then:
turn.apply(child_journal)
sandbox.end()
turn.end(output=answer)  # strict ingest
```

In the child, do not construct `Lemma` and do not call `/traces/ingest`:

```python
import json
import os
from uselemma_tracing import attach_turn

local = attach_turn(os.environ["LEMMA_TURN"])
local.record_tool(name="search_docs", input=query, output=docs)
print(json.dumps(local.records()))
```

The journal uses the same camelCase schema as the TypeScript SDK so a TS host
can apply a Python child's dump (and the reverse). `assemble_turn(token, journal)`
builds a `TraceContext` when the coordinator already has the dump; then call
`ingest()` once. Re-applying the same journal is idempotent (stable span ids).
If the sandbox dies before a clean dump, end the host sandbox span as `ERROR`;
tools that started and never ended are left incomplete.

## OpenAI Agents SDK

Install the OpenAI Agents extra and register the Lemma processor:

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

```python
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:

```python
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.

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

```python
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`):

```bash
pip install "uselemma-tracing[langgraph]"
```

```python
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

| Option       | Environment variable | Default                   |
| ------------ | -------------------- | ------------------------- |
| `api_key`    | `LEMMA_API_KEY`      | Required                  |
| `project_id` | `LEMMA_PROJECT_ID`   | Required                  |
| `base_url`   | none                 | `https://api.uselemma.ai` |

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

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

```python
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:

```python
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
