> ## Documentation Index
> Fetch the complete documentation index at: https://openlayer.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Open WebUI
> Trace Open WebUI chats, agents, tool calls, files, and user feedback with Openlayer
[Open WebUI](https://openwebui.com/) is a self-hosted AI chat platform that works with
Ollama and OpenAI-compatible APIs. Openlayer connects to it through a
[Filter Function](https://docs.openwebui.com/features/extensibility/plugin/functions/filter),
Open WebUI's built-in plugin system, so there is nothing extra to deploy. Every chat turn
becomes a trace in Openlayer with:
* The prompt and response, model, latency, token usage, and cost
* A step for each tool call, including MCP tools and Open WebUI's built-in tools such as
web search and image generation
* A retrieval step with the knowledge and web search sources the answer used
* The Open WebUI user and chat, so you can follow [users and sessions](/docs/monitoring/sessions-and-users)
* Custom models, sub-agents, and Automations, each tagged with how the turn started
* Optionally, the images, documents, and audio in the conversation, and your users'
ratings
Requests sent to Open WebUI's API are traced the same way as chats in the UI.
## Prerequisites
* Open WebUI 0.11 or later, and an admin account
* An Openlayer [API key](/docs/workspace-and-projects/find-your-api-key)
* A data source in your Openlayer project to receive the traces
To find your data source ID in Openlayer, open your project and copy the ID
from the data source settings. The function calls it the inference pipeline
ID.
## Step 1: Set your Openlayer credentials
The function reads your credentials from environment variables on the Open WebUI server.
Add them to the command you run Open WebUI with:
```bash theme={null}
docker run -d -p 3000:8080 \
-e OPENLAYER_API_KEY=YOUR_OPENLAYER_API_KEY \
-e OPENLAYER_INFERENCE_PIPELINE_ID=YOUR_DATA_SOURCE_ID \
-v open-webui:/app/backend/data \
--name open-webui --restart always \
ghcr.io/open-webui/open-webui:main
```
For self-hosted Openlayer, also set `OPENLAYER_BASE_URL` (for example,
`https://openlayer.example.com/v1`).
You can enter the API key in the function's valves instead (Step 3), but Open
WebUI shows valve values in plain text to every admin. Environment variables
keep the key out of the UI.
## Step 2: Add the Openlayer function
1. In Open WebUI, open the user menu and go to **Admin Panel → Functions**, then click **Create**.
2. Name the function `Openlayer Tracing`, replace the contents of the editor with the code
below, and click **Save & Create**.
```python theme={null}
"""
title: Openlayer Tracing
author: Openlayer
author_url: https://www.openlayer.com
version: 1.0.0
required_open_webui_version: 0.11.0
requirements: openlayer>=0.36.0
license: MIT
description: Sends each Open WebUI chat turn to Openlayer as a trace, including tool calls, knowledge sources, token usage, user, session, and (optionally) the images, audio, and files in the conversation.
"""
import asyncio
import base64
import json
import logging
import mimetypes
import os
import re
import time
from pathlib import Path
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
log = logging.getLogger("openlayer-filter")
_START_KEY = "openlayer_start_time"
_INLET_MESSAGES_KEY = "openlayer_inlet_messages"
_TURN_FILES_KEY = "openlayer_turn_files"
_DATA_URL = re.compile(r"^data:([^;,]+)[^,]*;base64,(.+)$", re.DOTALL)
_FILE_ID = re.compile(r"/api/v1/files/([0-9a-fA-F-]{36})")
def _text(content: Any) -> str:
"""Flatten OpenAI-style message content (str or list of parts) to text."""
if isinstance(content, str):
return content
if isinstance(content, list):
parts = []
for part in content:
if not isinstance(part, dict):
continue
if part.get("type") in ("text", "input_text", "output_text"):
parts.append(part.get("text", ""))
elif part.get("type") in ("image_url", "input_image"):
parts.append("[image]")
elif part.get("type") == "input_audio":
parts.append("[audio]")
elif part.get("type") == "file":
parts.append("[file]")
return " ".join(p for p in parts if p)
return "" if content is None else str(content)
def _file_refs(entries: Any) -> List[tuple]:
"""(file id, name) pairs from Open WebUI file entries; skips collections and web pages."""
refs = []
for entry in entries or []:
if not isinstance(entry, dict) or entry.get("type") not in (None, "file", "image"):
continue
info = entry.get("file") if isinstance(entry.get("file"), dict) else {}
file_id = info.get("id") or entry.get("id")
if file_id:
refs.append((file_id, info.get("filename") or entry.get("name") or "file"))
return refs
def _usage(usage: Dict[str, Any]) -> Dict[str, Optional[int]]:
"""Read Open WebUI's normalized usage.
Since v0.11, input_tokens/output_tokens/total_tokens are cumulative for the
whole turn (every model call in a tool loop), while prompt_tokens and
completion_tokens only describe the last call. Prefer the cumulative fields.
"""
usage = usage or {}
prompt = usage.get("input_tokens", usage.get("prompt_tokens", usage.get("prompt_eval_count")))
completion = usage.get("output_tokens", usage.get("completion_tokens", usage.get("eval_count")))
total = usage.get("total_tokens")
if total is None and prompt is not None and completion is not None:
total = int(prompt) + int(completion)
return {"prompt": prompt, "completion": completion, "total": total}
def _stamp(step, start: float, end: Optional[float] = None) -> None:
"""Pin a step's timing (create_step only fills in times that are unset)."""
step.start_time = start
step.end_time = end if end is not None else start
step.latency = (step.end_time - start) * 1000
class _Media:
"""Turns Open WebUI media into Openlayer attachments, once per unique file."""
def __init__(self, max_bytes: int, user_id: Optional[str]):
from openlayer.lib.tracing import Attachment
from openlayer.lib.tracing.content import AudioContent, FileContent, ImageContent
self._attachment = Attachment
self._kinds = {"image": ImageContent, "audio": AudioContent}
self._file = FileContent
self.max_bytes = max_bytes
self.user_id = user_id
self.by_checksum: Dict[str, Any] = {}
self.skipped: List[str] = []
def item(self, data: bytes, name: str, media_type: str):
"""Wrap bytes in the content item the platform renders for this media type."""
if len(data) > self.max_bytes:
self.skipped.append(f"{name} ({len(data)} bytes)")
return None
# from_bytes keeps the bytes out of the published trace if the upload fails.
attachment = self._attachment.from_bytes(data, name=name, media_type=media_type)
attachment = self.by_checksum.setdefault(attachment.checksum_md5, attachment)
content_class = self._kinds.get(media_type.split("/")[0], self._file)
return content_class(attachment)
def from_data_url(self, url: str, kind: str):
match = _DATA_URL.match(url or "")
if not match:
return None
media_type = match.group(1)
try:
data = base64.b64decode(match.group(2))
except (ValueError, TypeError):
return None
extension = (mimetypes.guess_extension(media_type) or ".bin").lstrip(".")
return self.item(data, f"{kind}.{extension}", media_type)
async def from_file_id(self, file_id: str, fallback_name: str = "file"):
"""Read a file from Open WebUI's storage (local disk, S3, GCS, or Azure)."""
from open_webui.models.files import Files
from open_webui.storage.provider import Storage
record = await Files.get_file_by_id(file_id)
# Only the chat user's own files: a prompt or tool result could mention any file id.
if not record or not record.path or record.user_id != self.user_id:
return None
meta = record.meta or {}
size = meta.get("size")
name = record.filename or meta.get("name") or fallback_name
if isinstance(size, int) and size > self.max_bytes:
self.skipped.append(f"{name} ({size} bytes)")
return None
media_type = meta.get("content_type") or mimetypes.guess_type(name)[0] or "application/octet-stream"
local_path = await asyncio.to_thread(Storage.get_file, record.path)
data = await asyncio.to_thread(Path(local_path).read_bytes)
return self.item(data, name, media_type)
class Filter:
class Valves(BaseModel):
priority: int = Field(default=0, description="Run order among filters (lower runs first).")
openlayer_api_key: str = Field(
default_factory=lambda: os.getenv("OPENLAYER_API_KEY", ""),
description="Openlayer API key. Keep on Default to use the OPENLAYER_API_KEY environment variable.",
)
inference_pipeline_id: str = Field(
default_factory=lambda: os.getenv("OPENLAYER_INFERENCE_PIPELINE_ID", ""),
description="ID of the Openlayer data source that receives traces.",
)
base_url: str = Field(
default_factory=lambda: os.getenv("OPENLAYER_BASE_URL", ""),
description="Only for self-hosted Openlayer, e.g. https://openlayer.example.com/v1.",
)
capture_tool_calls: bool = Field(default=True, description="Add a step for every tool call and its result.")
capture_sources: bool = Field(default=True, description="Add a retrieval step with knowledge and web search sources.")
upload_attachments: bool = Field(
default=False,
description="Upload the images, audio, and files in each chat turn to your Openlayer workspace storage so they show up in the trace.",
)
max_attachment_mb: int = Field(default=20, description="Skip attachments larger than this.")
routes: str = Field(
default="",
description=(
"Optional JSON map from a model ID or a run type (subagent, automation, api, ui) to a data source ID, "
'e.g. {"subagent": "", "support-agent": ""}. Turns that match nothing go to the default data source.'
),
)
debug: bool = Field(default=False, description="Log what is sent to Openlayer.")
def __init__(self):
self.valves = self.Valves()
self._configured_for = None
def _ready(self) -> bool:
v = self.valves
if not v.openlayer_api_key or not v.inference_pipeline_id:
return False
key = (v.openlayer_api_key, v.inference_pipeline_id, v.base_url, v.upload_attachments)
if key != self._configured_for:
from openlayer.lib import init
kwargs = {
"api_key": v.openlayer_api_key,
"inference_pipeline_id": v.inference_pipeline_id,
"attachment_upload_enabled": v.upload_attachments,
# Open WebUI makes its own provider calls; don't patch SDK clients here.
"auto_instrument": False,
}
if v.base_url:
kwargs["base_url"] = v.base_url
init(**kwargs)
self._configured_for = key
return True
async def inlet(
self, body: dict, __metadata__: Optional[dict] = None, __model__: Optional[dict] = None
) -> dict:
# __metadata__ is the same dict object in outlet, so values stashed here travel with the turn.
if __metadata__ is not None:
__metadata__.setdefault(_START_KEY, time.time())
# Outlet only gets text content, so keep the request's media parts
# (Open WebUI has already inlined images as data URLs at this point).
__metadata__[_INLET_MESSAGES_KEY] = [dict(m) for m in body.get("messages") or []]
if self.valves.upload_attachments:
# Open WebUI resends every file in the chat on each turn; keep only this turn's.
user_message = __metadata__.get("user_message")
if isinstance(user_message, dict): # UI chats
turn_files = _file_refs(user_message.get("files"))
else: # API callers: the request's own files, minus the model's knowledge
knowledge = ((__model__ or {}).get("info") or {}).get("meta", {}).get("knowledge") or []
knowledge_ids = {k.get("id") for k in knowledge if isinstance(k, dict)}
turn_files = [r for r in _file_refs(body.get("files")) if r[0] not in knowledge_ids]
__metadata__[_TURN_FILES_KEY] = turn_files
return body
async def outlet(
self,
body: dict,
__user__: Optional[dict] = None,
__metadata__: Optional[dict] = None,
__model__: Optional[dict] = None,
) -> dict:
try:
if self._ready():
await self._trace(body, __user__ or {}, __metadata__ or {}, __model__ or {})
except Exception:
# Never break the chat because tracing failed.
log.exception("Openlayer: failed to trace chat turn")
finally:
if __metadata__ is not None:
__metadata__.pop(_INLET_MESSAGES_KEY, None)
__metadata__.pop(_TURN_FILES_KEY, None)
return body
def _routes(self) -> Dict[str, str]:
if not self.valves.routes.strip():
return {}
try:
routes = json.loads(self.valves.routes)
except ValueError:
log.error("Openlayer: the routes valve is not valid JSON; using the default data source")
return {}
return {str(k): str(v) for k, v in routes.items() if v} if isinstance(routes, dict) else {}
@staticmethod
async def _run_context(metadata: dict, chat_id: str) -> Dict[str, Any]:
"""What started this turn: a sub-agent, an automation, an API call, or the UI."""
if metadata.get("automation_id"):
return {"run_type": "automation", "automation_id": metadata["automation_id"]}
if metadata.get("internal") and chat_id:
try:
from open_webui.models.chats import Chats
chat = await Chats.get_chat_by_id(chat_id)
chat_meta = (chat.meta if chat else None) or {}
except Exception:
chat_meta = {}
if chat_meta.get("type") == "subagent":
return {
"run_type": "subagent",
"parent_chat_id": chat_meta.get("parent_chat_id"),
"parent_message_id": chat_meta.get("parent_message_id"),
"delegation_id": chat_meta.get("delegation_id"),
}
return {"run_type": "ui" if metadata.get("session_id") else "api"}
async def _prompt(self, messages: List[dict], metadata: dict, media: Optional[_Media]) -> List[dict]:
"""Build the prompt, with media as content items when uploads are on."""
from openlayer.lib.tracing.content import TextContent
prompt = []
for message in messages:
content = message.get("content")
if media is None or not isinstance(content, list):
prompt.append({"role": message.get("role"), "content": _text(content)})
continue
items = []
for part in content:
kind = part.get("type") if isinstance(part, dict) else None
item = None
if kind in ("text", "input_text"):
item = TextContent(text=part.get("text", ""))
elif kind in ("image_url", "input_image"):
image = part.get("image_url")
url = image.get("url") if isinstance(image, dict) else image
item = media.from_data_url(url or "", "image")
elif kind == "input_audio":
audio = part.get("input_audio") or {}
fmt = audio.get("format") or "wav"
item = media.from_data_url(f"data:audio/{fmt};base64,{audio.get('data', '')}", "audio")
elif kind == "file":
item = media.from_data_url((part.get("file") or {}).get("file_data", ""), "file")
items.append(item or TextContent(text=_text([part])))
prompt.append({"role": message.get("role"), "content": items})
if media is not None:
# Files the user attached in this turn (documents, audio, images) go on the latest user message.
files = []
for file_id, name in metadata.get(_TURN_FILES_KEY) or []:
item = await media.from_file_id(file_id, name)
if item and all(f.attachment is not item.attachment for f in files):
files.append(item)
last_user = next((p for p in reversed(prompt) if p["role"] == "user"), None)
if last_user and files:
if isinstance(last_user["content"], str):
last_user["content"] = [TextContent(text=last_user["content"])]
present = {id(getattr(i, "attachment", None)) for i in last_user["content"]}
last_user["content"].extend(f for f in files if id(f.attachment) not in present)
return prompt
async def _tool_output(self, result: dict, media: Optional[_Media]):
"""Tool results as text, plus any files the tool produced (e.g. generated images)."""
from openlayer.lib.tracing.content import TextContent
text = _text(result.get("output"))
if media is None:
return text
# Built-in tools (e.g. generate_image) reference their files inside the result text
# as /api/v1/files//content; some tools also return a separate files list.
urls = [e.get("url", "") if isinstance(e, dict) else str(e) for e in result.get("files") or []]
file_ids = list(dict.fromkeys(_FILE_ID.findall(" ".join(urls) + " " + text)))
items = []
for file_id in file_ids:
item = await media.from_file_id(file_id, "generated")
if item:
items.append(item)
for url in urls:
if url.startswith("data:"):
item = media.from_data_url(url, "image")
if item:
items.append(item)
return [TextContent(text=text), *items] if items else text
async def _trace(self, body: dict, user: dict, metadata: dict, model: dict) -> None:
from openlayer.lib import update_current_trace, update_trace_user_session
from openlayer.lib.tracing import tracer
from openlayer.lib.tracing.enums import StepType
messages: List[dict] = body.get("messages") or []
if not messages or messages[-1].get("role") != "assistant":
return
assistant = messages[-1]
end = time.time()
start = metadata.get(_START_KEY) or end
media = (
_Media(self.valves.max_attachment_mb * 1024 * 1024, user.get("id"))
if self.valves.upload_attachments
else None
)
prompt = await self._prompt(metadata.get(_INLET_MESSAGES_KEY) or messages[:-1], metadata, media)
question = next((_text_of(p["content"]) for p in reversed(prompt) if p["role"] == "user"), "")
answer = _text(assistant.get("content"))
chat_id = body.get("chat_id") or metadata.get("chat_id") or ""
message_id = body.get("id") or metadata.get("message_id")
model_id = body.get("model") or model.get("id") or "unknown"
# Custom Workspace models ("agents") wrap a base model; price the call by the base model.
base_model_id = (model.get("info") or {}).get("base_model_id") or model_id
provider = model.get("owned_by") or "unknown"
run = await self._run_context(metadata, chat_id)
routes = self._routes()
destination = (
(routes.get(run["run_type"]) if run["run_type"] in ("subagent", "automation") else None)
or routes.get(model_id)
or routes.get(run["run_type"])
or self.valves.inference_pipeline_id
)
usage = _usage(assistant.get("usage") or {})
output_items = assistant.get("output") or []
sources = assistant.get("sources") or metadata.get("sources") or []
reasoning = [
_text(item.get("content") or item.get("summary"))
for item in output_items
if item.get("type") == "reasoning"
]
results = {
item.get("call_id"): item for item in output_items if item.get("type") == "function_call_output"
}
tool_calls = []
if self.valves.capture_tool_calls:
for item in output_items:
if item.get("type") != "function_call":
continue
try:
arguments = json.loads(item.get("arguments") or "{}")
except (TypeError, ValueError):
arguments = {"arguments": item.get("arguments")}
result = results.get(item.get("call_id"), {})
tool_calls.append((item, arguments, result, await self._tool_output(result, media)))
root_metadata = {
"chat_id": chat_id,
"message_id": message_id,
"model": model_id,
"model_name": model.get("name"),
"features": {k: v for k, v in (metadata.get("features") or {}).items() if v},
"tool_ids": metadata.get("tool_ids") or [],
"files": [
(f.get("file") or f).get("filename") or (f.get("file") or f).get("name")
for f in metadata.get("files") or []
if isinstance(f, dict)
],
**{k: v for k, v in run.items() if v},
"user_name": user.get("name"),
"user_role": user.get("role"),
}
if media is not None and media.skipped:
root_metadata["attachments_skipped"] = media.skipped
with tracer.create_step(
name="Open WebUI chat turn",
step_type=StepType.USER_CALL,
inputs={"prompt": prompt},
metadata=root_metadata,
# Always explicit: the SDK default is process-wide and shared with other functions.
inference_pipeline_id=destination,
) as root:
_stamp(root, start, end)
update_trace_user_session(
user_id=user.get("email") or user.get("id") or "anonymous",
# Sub-agent runs join the session of the chat that delegated them.
session_id=run.get("parent_chat_id") or chat_id or None,
)
if message_id:
# Lets the feedback Event function update this exact row later.
update_current_trace(inferenceId=message_id)
if self.valves.capture_sources and sources:
with tracer.create_step(
name="Knowledge retrieval",
step_type=StepType.RETRIEVER,
inputs={"query": question},
) as step:
_stamp(step, start)
step.log(output=[self._source(s) for s in sources])
with tracer.create_step(
name="LLM chat completion",
step_type=StepType.CHAT_COMPLETION,
inputs={"prompt": prompt},
metadata={"reasoning": "\n\n".join(reasoning)} if reasoning else None,
) as llm:
_stamp(llm, start + 0.001, end)
llm.model = base_model_id
llm.provider = provider
llm.prompt_tokens = usage["prompt"]
llm.completion_tokens = usage["completion"]
llm.tokens = usage["total"]
llm.log(output=answer)
for index, (item, arguments, result, output) in enumerate(tool_calls):
with tracer.create_step(
name=item.get("name") or "tool",
step_type=StepType.TOOL,
inputs=arguments,
metadata={"call_id": item.get("call_id"), "status": result.get("status")},
) as step:
# Outlet has no per-call timings; keep tool steps in call order.
_stamp(step, start + 0.002 + 0.001 * index)
step.log(output=output)
root.log(output=answer)
if self.valves.debug:
count = len(media.by_checksum) if media is not None else 0
log.info(
"Openlayer: traced %s message %s to %s (%s tokens, %d attachments)",
run["run_type"], message_id, destination, usage["total"], count,
)
@staticmethod
def _source(source: dict) -> dict:
info = source.get("source") or {}
return {
"name": info.get("name") or info.get("id"),
"type": info.get("type"),
"documents": [str(d)[:500] for d in (source.get("document") or [])][:5],
}
def _text_of(content: Any) -> str:
"""Text of a prompt message whose content may hold Openlayer content items."""
if isinstance(content, list):
return " ".join(getattr(item, "text", "") for item in content if getattr(item, "text", None))
return _text(content)
```
Open WebUI installs the `openlayer` package listed in the function's `requirements` the first
time it loads the function. If your deployment sets
`ENABLE_PIP_INSTALL_FRONTMATTER_REQUIREMENTS=false`, install `openlayer` in your Open WebUI
image instead.
Functions run arbitrary code on your Open WebUI server. Only install functions
from sources you trust.
## Step 3: Turn the function on
1. On the **Functions** page, switch the **Openlayer Tracing** toggle on.
2. Open the function's **⋯** menu and select **Global**, so it runs for every model.
The function's valves (the gear icon) control what it captures. To change one, switch it
from **Default** to **Custom** and save.
| Valve | Description |
| - | - |
| **Openlayer Api Key** | Keep on Default to use `OPENLAYER_API_KEY`. |
| **Inference Pipeline Id** | Keep on Default to use `OPENLAYER_INFERENCE_PIPELINE_ID`. |
| **Base Url** | Only for self-hosted Openlayer. Keep on Default to use `OPENLAYER_BASE_URL`. |
| **Capture Tool Calls** | Adds a step for each tool call and its result. On by default. |
| **Capture Sources** | Adds a retrieval step with knowledge and web search sources. On by default. |
| **Upload Attachments** | Uploads the images, documents, and audio in each turn. Off by default. See [Log images, files, and audio](#log-images-files-and-audio). |
| **Max Attachment Mb** | Skips attachments larger than this. Defaults to 20. |
| **Routes** | Sends some turns to other data sources. See [Send agents to their own data sources](#send-agents-to-their-own-data-sources). |
| **Debug** | Logs each traced turn to the Open WebUI logs. |
| **Priority** | Run order relative to your other filters. |
## Step 4: Enable usage reporting
Go to **Admin Panel → Settings → Models**, open each model you want to track, and enable
the **Usage** capability. Without it, chats streamed in the UI arrive in Openlayer without
token counts or cost.
## View traces in Openlayer
Send a message in Open WebUI, then open your data source in Openlayer. Each chat turn is
one trace:
* **Open WebUI chat turn**: the root step, with the conversation as its prompt and the
reply as its output
* **Knowledge retrieval**: when the answer used knowledge, attached files, or web search
* **LLM chat completion**: the model, provider, and token usage for the whole turn,
including every model call in a tool loop
* One step per **tool call**, with its arguments and result
The trace's session is the Open WebUI chat, its user is the Open WebUI user's email, and
its inference ID is the Open WebUI message ID. Its metadata records how the turn started in
`run_type`: `ui`, `api`, `automation`, or `subagent`.
## Agents and sub-agents
The function also traces the agents your users build in Open WebUI:
* **Custom models** (**Workspace → Models**) are traced like any other model. The trace
keeps the custom model's ID in its metadata, and the LLM step uses the base model, so
cost is calculated from the base model's pricing.
* **Sub-agents** that a model starts with Open WebUI's `delegate_task` tool produce two
traces. The parent turn has a `delegate_task` step with the sub-agent's result. The
sub-agent's own run is a separate trace with its own tool calls and token usage, in the
parent chat's session, with `run_type: subagent`, `parent_chat_id`, and
`parent_message_id` (the parent turn's inference ID) in its metadata.
* **Automations** are traced with `run_type: automation` and their `automation_id`.
* **Pipe Functions** that run their own agent logic appear as a single turn with the Pipe's
final answer. The function doesn't see the model calls inside a Pipe. To trace those,
instrument the Pipe's code with the [Openlayer SDK](/docs/monitoring/instrument). Its steps
are published as their own trace.
## Send agents to their own data sources
By default, every turn goes to one data source. To send some turns somewhere else, such as
all sub-agent runs or one agent's turns, set the **Routes** valve to a JSON object that maps
a model ID or a run type to a data source ID:
```json theme={null}
{
"subagent": "YOUR_SUBAGENT_DATA_SOURCE_ID",
"support-agent": "YOUR_SUPPORT_AGENT_DATA_SOURCE_ID"
}
```
The function sends each turn to the first match, in this order:
1. The run type `subagent` or `automation`
2. The model ID, such as a custom model's ID
3. The run type `ui` or `api`
4. The default data source
Each trace goes to a single data source, so a `delegate_task` step stays in the parent
turn's trace. A sub-agent run routed elsewhere stays linked to its parent through its
`parent_chat_id` and `parent_message_id` metadata. Every route uses the function's API key,
so each data source must be one that key can write to. If you
[capture user feedback](#capture-user-feedback), set the same value in the Openlayer
Feedback function's **Routes** valve.
## Log images, files, and audio
Turn on the **Upload Attachments** valve to add the media in each chat turn to its trace:
* Images in the user's message, shown inside the prompt
* Documents and audio files the user attached in that turn
* Files a tool produced, such as a generated image, on that tool's step
Openlayer shows images inline, PDFs in a page viewer, and audio in a player. See
[Trace multimodal data](/docs/monitoring/multimodal) for how attachments work.
Uploading attachments sends your users' files to your Openlayer workspace
storage, so enable it only when your privacy requirements allow it. On
self-hosted Openlayer, attachment uploads require Amazon S3 storage.
## Capture user feedback
When users rate a response in Open WebUI (thumbs up or down, with an optional score,
reason, and comment), an
[Event Function](https://docs.openwebui.com/features/extensibility/plugin/functions/event)
can add the rating to that response's trace:
1. Go to **Admin Panel → Functions** and click **Create**. Name the function
`Openlayer Feedback`, replace the contents of the editor with the code below, and click
**Save & Create**.
2. Switch its toggle on. Event functions handle every event, so they don't need **Global**.
```python theme={null}
"""
title: Openlayer Feedback
author: Openlayer
author_url: https://www.openlayer.com
version: 1.0.0
required_open_webui_version: 0.11.0
requirements: openlayer>=0.36.0
license: MIT
description: Adds Open WebUI message ratings (thumbs up/down, reason, comment) to the matching Openlayer trace.
"""
import asyncio
import json
import logging
import os
from typing import Optional
from pydantic import BaseModel, Field
log = logging.getLogger("openlayer-feedback")
class Event:
class Valves(BaseModel):
openlayer_api_key: str = Field(
default_factory=lambda: os.getenv("OPENLAYER_API_KEY", ""),
description="Openlayer API key. Keep on Default to use the OPENLAYER_API_KEY environment variable.",
)
inference_pipeline_id: str = Field(
default_factory=lambda: os.getenv("OPENLAYER_INFERENCE_PIPELINE_ID", ""),
description="Same data source as the Openlayer Tracing filter.",
)
base_url: str = Field(
default_factory=lambda: os.getenv("OPENLAYER_BASE_URL", ""),
description="Only for self-hosted Openlayer, e.g. https://openlayer.example.com/v1.",
)
routes: str = Field(
default="",
description="Same routes JSON as the Openlayer Tracing filter, if you use it. Ratings are matched to traces in any of these data sources.",
)
def __init__(self):
self.valves = self.Valves()
async def event(self, event: dict, __event_name__: Optional[str] = None) -> None:
if __event_name__ not in ("feedback.created", "feedback.updated"):
return
if not self.valves.openlayer_api_key or not self.valves.inference_pipeline_id:
return
from open_webui.models.feedbacks import Feedbacks
feedback_id = (event.get("subject") or {}).get("id")
feedback = await Feedbacks.get_feedback_by_id(feedback_id) if feedback_id else None
if not feedback:
return
message_id = (feedback.meta or {}).get("message_id")
data = feedback.data or {}
if not message_id or data.get("rating") is None:
return
row = {
"user_rating": data.get("rating"), # 1 = thumbs up, -1 = thumbs down
"user_rating_reason": data.get("reason") or None,
"user_rating_comment": data.get("comment") or None,
"user_rating_details": (data.get("details") or {}).get("rating"), # optional 1-10 score
}
try:
await asyncio.to_thread(self._update, message_id, {k: v for k, v in row.items() if v is not None})
except Exception:
log.exception("Openlayer: failed to attach feedback to %s", message_id)
def _destinations(self) -> list:
destinations = [self.valves.inference_pipeline_id]
try:
routes = json.loads(self.valves.routes) if self.valves.routes.strip() else {}
except ValueError:
routes = {}
if isinstance(routes, dict):
destinations += [str(v) for v in routes.values() if v]
return list(dict.fromkeys(destinations))
def _update(self, inference_id: str, row: dict) -> None:
from openlayer import NotFoundError, Openlayer
kwargs = {"api_key": self.valves.openlayer_api_key}
if self.valves.base_url:
kwargs["base_url"] = self.valves.base_url
client = Openlayer(**kwargs)
# The trace may have been routed to any configured data source; the others answer 404.
for pipeline_id in self._destinations():
try:
client.inference_pipelines.rows.update(
inference_pipeline_id=pipeline_id, inference_id=inference_id, row=row
)
return
except NotFoundError:
continue
log.warning("Openlayer: no trace found for message %s in any configured data source", inference_id)
```
The rating is added to the trace as `user_rating` (`1` or `-1`), along with
`user_rating_reason`, `user_rating_comment`, and `user_rating_details` (the optional 1–10
score). The function uses the same environment variables as the tracing function and finds
the trace by its message ID. If you use [routes](#send-agents-to-their-own-data-sources),
set the same value in this function's **Routes** valve so ratings reach traces in any of
those data sources. See [Update existing traces](/docs/monitoring/updating-data).
## Limitations
* Background tasks, such as title, tag, and follow-up generation, search query generation,
and autocomplete, don't pass through filters, so they aren't traced.
* Requests to Open WebUI's direct provider routes (`/openai/...` and `/ollama/...`) skip
filters and aren't traced. Requests to `/api/chat/completions` are traced unless
`ENABLE_API_OUTLET_FILTERS` is set to `false`.
* Tool steps don't have individual timings, because Open WebUI reports one duration for the
whole turn.
## Troubleshooting
### Traces don't appear in Openlayer
1. Check that the function is switched on and set to **Global**.
2. Check the API key and data source ID, in the environment variables or the valves.
3. Turn on the **Debug** valve and check the Open WebUI logs (`docker logs open-webui`).
Each traced turn logs `Openlayer: traced message`, and errors log
`Openlayer: failed to trace chat turn`.
4. Make sure the Open WebUI server can reach `api.openlayer.com`, or your
`OPENLAYER_BASE_URL`.
### Token counts and cost are missing
Enable the **Usage** capability for the model (Step 4).
### Each turn is traced twice
If you set up an earlier version of this integration that used Open WebUI Pipelines,
remove that pipeline in **Admin Panel → Settings → Pipelines**.
## Learn more
* [Open WebUI Functions](https://docs.openwebui.com/features/extensibility/plugin/functions/)
* [Track users and sessions](/docs/monitoring/sessions-and-users)
* [Trace multimodal data](/docs/monitoring/multimodal)
* [Openlayer monitoring overview](/docs/monitoring/overview)
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.