Integrate Lens with agent frameworks

Trace LangGraph, OpenAI Agents SDK, Google ADK, and your own agent loop with Lens, across Agent Memory, Agent Playbooks, and Context Gateway.

To trace an agent run in Lens, open a turn around it. Each step the agent takes inside the run, such as a memory recall or a model call, is recorded in that turn. For more about turns and steps, see Lens concepts.

Before you start, get the Lens connection values from the quickstart. The framework examples also need a model provider key, such as OPENAI_API_KEY.

Where the turn opens

Framework Open the turn around
LangGraph app.invoke() or app.ainvoke(). With an Iris service attached, attach() opens the turn for you.
OpenAI Agents SDK Runner.run()
Google Agent Development Kit (ADK) runner.run_async()
Your own loop The handler for one message

To trace model calls inside the turn, install redis-iris[llm-tracing] and call lens.init() at startup. See Trace model calls.

Trace one turn across Iris services

When your agent uses Agent Memory, Agent Playbooks, and Context Gateway, a single Lens turn shows how each Iris service contributed to the agent's response. This example needs the connection values for each Iris service.

workflows.turn() runs each service's context step before the model and its write-back after. Every service call joins the turn:

from redis_iris import Iris, lens, workflows
from redis_iris.hooks import stack

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

iris = Iris.from_env()
memory = iris.agent_memory(assistant_id="support-bot")
playbook = iris.agent_playbooks()
gateway = iris.context_gateway()

hooks = stack(memory.hooks(), playbook.disclosure(audience="support"))


def run_agent(context, message, tools):
    # Your model and tool loop goes here. It receives the Iris context and tools.
    return "Your refund was issued on May 2."


def handle(owner, message, thread_id):
    with lens.turn(thread_id, user_id=owner) as outer:
        with workflows.turn(hooks, message, owner=owner, session=thread_id) as turn:
            tools = [*gateway.model_tools(), *hooks.tools()]
            answer = run_agent(turn.context, message, tools)
            turn.record(answer)
        outer.set_outcome("completed")
    return answer


print(handle("user-1042", "Where can I see my refund status?", "thread-1"))

The turn shows the memory recall and the Agent Playbooks discovery, then any Context Gateway tool calls your agent makes, and the write-back. In asynchronous code, workflows.aturn() runs the services' context steps at the same time. The outer lens.turn() sets the outcome. When a turn is already open, workflows.turn() and attach() add their steps to it rather than starting another.

LangGraph

Install LangGraph and the LangChain OpenAI integration:

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

Wrap the graph run in lens.turn(). LangGraph's thread ID identifies the conversation, so pass it as the Lens session ID:

from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, MessagesState, StateGraph

from redis_iris import lens

lens.init(agent_id="support-bot")
model = ChatOpenAI(model="gpt-4o-mini")


def assistant(state: MessagesState):
    return {"messages": [model.invoke(state["messages"])]}


graph = StateGraph(MessagesState)
graph.add_node("assistant", assistant)
graph.add_edge(START, "assistant")
graph.add_edge("assistant", END)
app = graph.compile()

thread_id = "thread-1"
with lens.turn(thread_id, user_id="user-1042") as turn:
    result = app.invoke(
        {"messages": [("user", "Where can I see my refund status?")]},
        {"configurable": {"thread_id": thread_id}},
    )
    turn.set_outcome("completed")

print(result["messages"][-1].content)

Each model call in the graph joins the turn with its token counts.

Attach an Iris service

When the graph uses Agent Memory or Agent Playbooks, attach() adds the service's context before your first node and its write-back after your last. With Lens on, every run of an attached graph is one turn. The session comes from thread_id and the user from user_id in the run's configuration:

from redis_iris import AgentMemory
from redis_iris.integrations.langgraph import attach

memory = AgentMemory.from_env()
app = attach(memory, graph).compile()

result = app.invoke(
    {"messages": [("user", "Where can I see my refund status?")]},
    {"configurable": {"user_id": "user-1042", "thread_id": "thread-1"}},
)

attach() needs the langgraph extra: python -m pip install 'redis-iris[langgraph]'.

OpenAI Agents SDK

Install the OpenAI Agents SDK:

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

Wrap Runner.run() in lens.turn():

import asyncio

from agents import Agent, Runner

from redis_iris import lens

lens.init(agent_id="support-bot")
agent = Agent(name="Support", instructions="Answer refund questions in two sentences.")


async def handle(session_id, user_id, message):
    with lens.turn(session_id, user_id=user_id) as turn:
        result = await Runner.run(agent, message)
        turn.set_outcome("completed")
    return result.final_output


print(asyncio.run(handle("session-1", "user-1042", "Where can I see my refund status?")))

The OpenAI Agents SDK calls models through the OpenAI client, so each model call joins the turn with its token counts.

Google ADK

Install the Google ADK:

python -m pip install redis-iris google-adk

Wrap the runner.run_async() loop in lens.turn():

import asyncio

from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

from redis_iris import lens

lens.init(agent_id="support-bot")
agent = Agent(
    name="support",
    model="gemini-2.0-flash",
    instruction="Answer refund questions in two sentences.",
)
runner = InMemoryRunner(agent, app_name="support")


async def handle(session_id, user_id, message):
    await runner.session_service.create_session(
        app_name="support", user_id=user_id, session_id=session_id
    )
    content = types.Content(role="user", parts=[types.Part(text=message)])
    answer = ""
    with lens.turn(session_id, user_id=user_id) as turn:
        async for event in runner.run_async(
            user_id=user_id, session_id=session_id, new_message=content
        ):
            if event.is_final_response() and event.content:
                answer = event.content.parts[0].text
        turn.set_outcome("completed")
    return answer


print(asyncio.run(handle("session-1", "user-1042", "Where can I see my refund status?")))

The turn records the agent run, its outcome, and every Iris service call the agent makes. The example uses a Gemini model and needs GOOGLE_API_KEY.

Your own agent loop

Wrap the handler for one message in lens.turn(), and set the outcome on each path:

from redis_iris import lens


def call_model(message, context):
    # Your model call goes here.
    return "Your refund was issued on May 2."


def handle(session_id, user_id, message):
    with lens.turn(session_id, user_id=user_id) as turn:
        try:
            answer = call_model(message, context=[])
        except Exception:
            turn.set_outcome("failed")
            raise
        turn.set_outcome("completed")
    return answer


print(handle("session-1", "user-1042", "Where can I see my refund status?"))

Each call to handle() records one turn under the session and user IDs you pass. The turn is marked completed when the model answers, or failed when the call raises. Record your retrievals and prompt assembly inside the same turn to see which context reached the model.

Next steps

RATE THIS PAGE
Back to top ↑