Instrument an agent

Record turns, retrievals, memory recalls, prompt assembly, model calls, and custom steps with the Redis Iris SDK.

Record each agent turn and the steps inside it with redis_iris.lens, the Lens module of the Redis Iris software development kit (SDK). These examples need the Lens connection values from the quickstart.

Turn Lens on

Call lens.init() once at startup. It turns Lens on, traces model calls, and sets the agent ID that Lens groups your turns under:

from redis_iris import lens

lens.init(agent_id="support-bot")

You can set the agent ID with IRIS_LENS_AGENT_ID instead, and call lens.init() with no arguments. An agent_id you pass to lens.init() overrides the variable.

Without lens.init(), Lens starts at the first Lens or Iris service call when IRIS_LENS_ENDPOINT is set, and records everything except model calls. You can also pass Lens settings to Iris(IrisSettings(lens=...)). See Configure Lens for every setting.

Open a turn

Wrap each agent exchange in lens.turn() and pass a session ID. The Lens UI shows only turns that have one. Optionally, pass a thread ID, a user ID, and an agent ID that overrides the process default. Pass a user ID to filter sessions by user in the Lens UI and the read API:

import uuid

session_id = f"session-{uuid.uuid4().hex[:8]}"

with lens.turn(session_id, thread_id="thread-1", user_id="user-1042") as turn:
    turn.set_attribute("app.channel", "web")
    answer = "Your refund was issued on May 2."
    turn.set_outcome("completed")

Use set_attribute() to add your own fields to the turn, and set_outcome() to record how it ended. See turn outcomes for the values.

lens.turn() works the same in asynchronous code: open it with with around the await calls of one exchange.

What Iris services record

Agent Memory, Agent Playbooks, and Context Gateway calls record their own steps inside the open turn, in both their synchronous and asynchronous forms:

These steps come from the Redis Iris service objects, such as AgentMemory and AgentPlaybooks. To record a call on a service's generated client (.client), wrap it in lens.memory_recall(), lens.retrieval(), or lens.span().

Trace model calls

Install the llm-tracing extra, then call lens.init() at startup:

python -m pip install 'redis-iris[llm-tracing]'

Lens then records calls made through the OpenAI, Anthropic, Amazon Bedrock, and LangChain client libraries, including LangGraph. Each model call records its prompt, completion, and token counts, and the turn records its token totals.

Record your own retrievals

Iris services record their own results. For other sources, such as your own vector search, wrap the call in lens.retrieval() and record its results as (id, score, text) tuples:

question = "How long do refunds take?"
documents = [
    ("doc-17", 0.88, "Refunds post within 5 to 7 business days."),
    ("doc-04", 0.61, "Store credit is available immediately."),
]

with lens.turn(session_id) as turn:
    with lens.retrieval(query=question, data_source="help-center") as search:
        search.results(documents)
    turn.set_outcome("completed")

Pass None as the score for a source that doesn't return one. Lens records it as no score rather than a score of zero.

Add custom steps

Wrap any operation in lens.span() to record it in the turn. Pass attributes as keyword arguments, or set them on the step:

with lens.turn(session_id) as turn:
    with lens.span("rank_documents", candidates=len(documents)) as step:
        ranked = sorted(documents, key=lambda d: d[1], reverse=True)
        step.set_attribute("app.top_document", ranked[0][0])
    turn.set_outcome("completed")

Custom steps are part of the turn's trace, and the read API returns them in the turn's spans.

Include other instrumentation

Lens exports the spans other OpenTelemetry instrumentation creates inside a turn, apart from a default set of low-level scopes such as database drivers. To also export spans created outside a turn, or from an excluded scope, wrap the call in lens.include(). It works as a decorator or a context manager:

@lens.include()
def search_vector_store(query):
    # Spans that another OpenTelemetry instrumentation creates in here are exported to Lens.
    return documents

Get the turn ID

Read the turn ID to link a turn to your own logs, or to fetch the turn with the read application programming interface (API). Inside a turn, call lens.current_trace_id(). After the turn, read trace_id from the turn handle:

with lens.turn(session_id) as turn:
    print(f"Turn ID: {lens.current_trace_id()}")
    turn.set_outcome("completed")
print(f"Turn ID after the turn: {turn.trace_id}")

Carry a turn across processes

When a turn continues in another process, such as a queue worker, pass its context with the work. Call lens.inject() to write the turn context into a dictionary, and lens.continue_from() on the other side to resume the turn:

def enqueue(job):
    print(f"Queued job with {len(job['lens'])} Lens headers")
    return job


def worker(job):
    with lens.continue_from(job["lens"]):
        with lens.span("summarize_case"):
            pass


with lens.turn(session_id) as turn:
    job = enqueue({"case": "case-881", "lens": lens.inject()})
    worker(job)
    turn.set_outcome("completed")

Steps recorded in the worker join the same turn. The dictionary carries the session, user, and agent IDs, so send it only to services you trust, and resume only from dictionaries your own services created.

Within one process, lens.init() with the llm-tracing extra carries the turn into threads your code starts.

Shut down

Before a short-lived process exits, call lens.shutdown():

if not lens.shutdown(timeout_millis=10_000):
    print("Lens did not finish exporting before the timeout")

shutdown() waits for the export queue to flush and returns True when the flush completes. A failed export logs a warning on the iris.lens logger. shutdown() also turns Lens off for the rest of the process, so call it once, when your program is done tracing. Long-running services export continuously and call it only when they stop.