llmentary

Python (pytest) client for llmentary: deterministic trace-based integration tests for LLM apps.

llmentary is the Python/pytest client for the barzoj-co/llmentary recording proxy. It gives tests deterministic, trace-based assertions for OpenAI-compatible Chat Completions and Responses calls. There is no LLM judge: assertions inspect the requests, responses, tool calls, usage, timings, retries, and errors the proxy recorded. The control API contract is the OpenAPI description of /healthz, /sessions, and /sessions/{id}/wait. After the next proxy release the same file is served at GET /openapi.yaml on the control listener and ships next to the bundled binary.

Published platform wheels bundle proxy v0.3.0; see docs/binaries.md for how it is resolved and how to override it.

Install

pip install llmentary

Platform wheels bundle the version-pinned proxy binary (PROXY_VERSION, currently 0.3.0). The sdist is client-only and does not include a binary. Install a wheel with pip install --only-binary=:all: llmentary, or set LLMENTARY_BIN, or put llmentary on PATH. Run python -m llmentary doctor to see what this install resolved.

Supported wheel tags: manylinux2014_x86_64, manylinux2014_aarch64, macosx_11_0_x86_64, macosx_11_0_arm64, win_amd64.

For pytest fixture support:

pip install "llmentary[pytest]"

For the quickstart using openai-python:

pip install "llmentary[pytest,openai]"

Quickstart

Point pytest at an existing proxy, or let the fixture start one:

export LLMENTARY_PROXY_URL=http://127.0.0.1:7777
export LLMENTARY_CONTROL_URL=http://127.0.0.1:7778
pytest

export LLMENTARY_BIN=/path/to/llmentary
export LLMENTARY_UPSTREAM_BASE_URL=https://api.openai.com/v1
export LLMENTARY_UPSTREAM_API_KEY="$OPENAI_API_KEY"
pytest

In-process tests that can set headers:

from openai import OpenAI


def test_tool_flow(llmentary):
    client = OpenAI(
        api_key="test-key",
        base_url=llmentary.openai_base_url,
        default_headers=llmentary.headers,
    )

    first = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": "Weather in Paris?"}],
        tools=[
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "parameters": {
                        "type": "object",
                        "properties": {"city": {"type": "string"}},
                        "required": ["city"],
                    },
                },
            }
        ],
    )
    tool_call = first.choices[0].message.tool_calls[0]

    client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "user", "content": "Weather in Paris?"},
            first.choices[0].message,
            {"role": "tool", "tool_call_id": tool_call.id, "content": '{"weather":"sunny"}'},
        ],
    )

    llmentary.expect().call_count(2).tool_called("get_weather").args_subset(
        "get_weather", {"city": "Paris"}
    ).tool_result_fed_back("get_weather").no_errors()

Black-box / import-time clients that cannot send headers: set the service base URL to llmentary.path_base_url() inside the test, then import the app. The session id exists only after the fixture runs, so a module-level import is too early.

import importlib


def test_import_time_client(llmentary, monkeypatch):
    monkeypatch.setenv("OPENAI_BASE_URL", llmentary.path_base_url())
    app = importlib.import_module("myapp")
    app.client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": "ping"}],
    )
    llmentary.expect().prompt_contains("ping").no_errors()

A path prefix isolates client processes, not conversations. One process with one env-configured SDK client is one session. For N parallel conversations use one of: (a) one worker process per session; (b) construct the SDK client per conversation with base_url=lm.path_base_url(); (c) headers/baggage when the app can set them; (d) run conversations sequentially with DELETE between them. A second bus worker needs its own topic or partition; two consumer groups on one topic both get every event.

Prove isolation with isolated_from, not a substring search over request JSON (system prompts and tool names will false-positive):

other.expect().isolated_from(llmentary)
llmentary.expect().isolated_from(other)

See the proxy black-box guide and assertions guide.

The function-scoped llmentary fixture returns a Session: it creates a unique session id, exposes headers with X-Test-Session, waits for proxy quiescence before assertions, and deletes the session during teardown.

Waiting for calls

wait asks the proxy to return the session trace once the session has been idle for quiet, or to fail when timeout elapses. The proxy only reads these query parameters:

Override the defaults with environment variables. Each accepts a Go duration (45s, 200ms) or a bare number of seconds (45, 0.2):

The client HTTP deadline is timeout + quiet + 5s, so the client cannot abort before the proxy answers. A timeout is an error: wait timed out after 30s on session X (k calls so far, still active: n). Raise LLMENTARY_WAIT_TIMEOUT ….

wait on a session that was never created (or was DELETEd):

from llmentary import SessionNotFound, WaitTimeoutError

try:
    llmentary.wait(timeout=3, quiet=0.2, min_calls=0)
except SessionNotFound:
    pass

try:
    llmentary.wait(timeout=3, quiet=0.2, min_calls=1)
except WaitTimeoutError:
    pass

expect() treats that SessionNotFound as zero calls. A never-created session that will be filled by an async worker should wait with min_calls=1 (or higher), not 0.

Assertions

llmentary.expect().call_count(1).tool_called("search").called_times("search", 1)
llmentary.query(api_type="responses", model="gpt-4.1-mini").expect().no_errors()
llmentary.query().without_tools().expect().call_count(1).request_param(
    "response_format.type", "json_schema"
)
mark = llmentary.mark()
llmentary.expect(since=mark).final_response_mentions_any("20", "sunny")

with llmentary.soft_expect() as expect:
    expect.tool_called("search")
    expect.response_contains("done")

Supported deterministic assertions include call count (call_count), tool calls (tool_called, not_called, called_times, args_subset, args_matcher, order, parallel_tool_calls, tool_result_fed_back), request and finish metadata (model, request_param, finish_reason), usage and timing (tokens, latency, ttft, retries, no_retries), errors (no_errors, has_error, mid_stream_error), prompt or response text (prompt_contains, prompt_regex, response_contains, response_regex, response_json_valid, response_json_schema), and spec observability (no_spec_violations, no_ignored_fields). Failures raise AssertionError with the session trace summary. A filter or since= that matches zero calls fails no_errors() / not_called() unless you call allow_empty() first. Use llmentary.list_sessions() when a wait or expect fails with zero calls, so the message can name the known session ids.

Call id values (call_3) are unique within one proxy process and can interleave across sessions. CapturedCall.seq is 1-based per session (sequence_number() is an alias). The trace summary prints #seq (id).

See docs/api.md, docs/pytest.md, and the proxy assertions guide.

llmentary repositories

License

Licensed under MIT, see LICENSE