diff --git a/chat/api.py b/chat/api.py index 3f4e2ef..f591012 100644 --- a/chat/api.py +++ b/chat/api.py @@ -247,6 +247,32 @@ def add_file_to_knowledge_base(self, knowledge_id: str, file_id: str) -> Any: json={"file_id": file_id}, ) + # --- functions (user/model hooks) --------------------------------------- + def list_functions(self) -> list[dict[str, Any]]: + return self.request("GET", "/api/v1/functions/") or [] + + def get_function(self, function_id: str) -> dict[str, Any]: + return self.request("GET", f"/api/v1/functions/id/{function_id}") + + def update_function_content( + self, function_id: str, name: str, content: str + ) -> dict[str, Any]: + """Replace a function's code. Open WebUI validates + re-registers it.""" + return self.request( + "POST", f"/api/v1/functions/id/{function_id}/update", + json={"id": function_id, "name": name, "content": content, "meta": {}}, + ) + + def create_function(self, name: str, content: str, function_id: str | None = None) -> dict[str, Any]: + """Create a function; ``id`` is Open WebUI's slug for it.""" + payload = { + "id": function_id or "studio_langfuse_capture", + "name": name, + "content": content, + "meta": {}, + } + return self.request("POST", "/api/v1/functions/create", json=payload) + def create_tool(self, tool_id: str, name: str, content: str, description: str = "") -> dict[str, Any]: """Register a Python tool in Open WebUI's toolkit (the Tools page). diff --git a/chat/openwebui_functions/studio_langfuse_capture.py b/chat/openwebui_functions/studio_langfuse_capture.py new file mode 100644 index 0000000..80d1db4 --- /dev/null +++ b/chat/openwebui_functions/studio_langfuse_capture.py @@ -0,0 +1,111 @@ +""" +title: Studio Langfuse Capture +author: SimpleAudit Studio +version: 1.0.0 +description: >- + Pure tap for SimpleAudit Studio chat tracing. Forwards every chat turn to the + Studio pipelines service (inlet before the model call, outlet after the reply), + which records it in Langfuse. Works with ALL connections — no base URL changes, + no provider knowledge. Tracing failures never break chat. +requirements: httpx>=0.27 +""" + +import os +import socket +import uuid + +import httpx + + +# Base URL of the Studio pipelines service (no trailing slash). In dev / +# single-container setups Open WebUI and Studio share one host, so loopback +# works; in the Azure multi-container deploy set STUDIO_PIPELINES_URL to the +# studio container name (e.g. http://helsedir-studio:9099). Leave empty to +# disable capturing. +def _default_pipelines_url() -> str: + if os.environ.get("STUDIO_PIPELINES_URL"): + return os.environ["STUDIO_PIPELINES_URL"].rstrip("/") + try: + socket.create_connection(("helsedir-studio", 9099), timeout=0.2) + return "http://helsedir-studio:9099" + except OSError: + return "http://127.0.0.1:9099" + + +STUDIO_PIPELINES_URL = _default_pipelines_url() +PIPELINE_ID = os.environ.get("STUDIO_PIPELINES_ID", "studio_capture") +# Optional bearer key; the Studio stub accepts a missing or matching key. +STUDIO_PIPELINES_KEY = os.environ.get("STUDIO_PIPELINES_API_KEY", "") + + +class Filter: + class Valves: + # If set in the admin UI, overrides STUDIO_PIPELINES_URL at runtime. + pipelines_url: str = "" + + def __init__(self): + self.valves = self.Valves() + + def _url(self) -> str: + return (self.valves.pipelines_url or STUDIO_PIPELINES_URL).rstrip("/") + + def _headers(self) -> dict: + headers = {} + if STUDIO_PIPELINES_KEY: + headers["Authorization"] = f"Bearer {STUDIO_PIPELINES_KEY}" + return headers + + @staticmethod + def _user_payload(__user__) -> dict | None: + if not __user__: + return None + return { + "id": __user__.get("id"), + "name": __user__.get("name"), + "email": __user__.get("email"), + } + + async def user_message(self, message, history, user_message, chat_id=None, __user__=None, **kwargs): + """Before the model call: report the request to the Studio stub.""" + url = self._url() + if not url: + return user_message + payload = { + "body": { + "chat_id": chat_id or str(uuid.uuid4()), + "messages": [dict(m) for m in (history or [])] + [{"role": "user", "content": user_message}], + "model": kwargs.get("model") or {}, + }, + "user": self._user_payload(__user__), + } + try: + async with httpx.AsyncClient(timeout=5) as client: + await client.post(f"{url}/v1/{PIPELINE_ID}/filter/inlet", json=payload, headers=self._headers()) + except Exception: # noqa: S110,BLE001 - tracing must never break chat + pass + return user_message + + async def after_chat_completion(self, content, context=None, __user__=None, **kwargs): + """After the model reply: report the assistant message to the Studio stub.""" + url = self._url() + if not url: + return content + ctx = context or {} + messages = list(ctx.get("messages") or []) + if isinstance(content, str): + messages.append({"role": "assistant", "content": content}) + elif isinstance(content, dict): + messages.append(content) + payload = { + "body": { + "chat_id": ctx.get("chat_id") or "local", + "messages": messages, + }, + "user": self._user_payload(__user__), + } + try: + async with httpx.AsyncClient(timeout=5) as client: + await client.post(f"{url}/v1/{PIPELINE_ID}/filter/outlet", json=payload, headers=self._headers()) + except Exception: # noqa: S110,BLE001 - tracing must never break chat + pass + return content diff --git a/chat/pipelines_service.py b/chat/pipelines_service.py index b785498..66e4717 100644 --- a/chat/pipelines_service.py +++ b/chat/pipelines_service.py @@ -112,7 +112,7 @@ def _get_langfuse(): def _chat_id(body: dict) -> str: metadata = body.get("metadata") or {} - chat_id = metadata.get("chat_id") or "local" + chat_id = body.get("chat_id") or metadata.get("chat_id") or "local" if chat_id == "local": session_id = metadata.get("session_id") chat_id = f"temporary-session-{session_id}" if session_id else str(uuid.uuid4()) diff --git a/chat/sync.py b/chat/sync.py index 0447e41..975ec90 100644 --- a/chat/sync.py +++ b/chat/sync.py @@ -50,13 +50,18 @@ def connections_to_push() -> list[dict]: def push_now(payloads: list[dict] | None = None) -> dict[str, int]: - """Push connections (all of them by default). Raises ChatAPIError on refusal.""" + """Push connections (all of them by default). Raises ChatAPIError on refusal. + + Also installs/updates the bundled tracing function (``ensure_tracing_function``); + that part never raises — connection sync stays the only failure path here. + """ api = ChatAPI.as_user(_admin()) pushed = connections_to_push() if payloads is None else payloads result = api.push_connections(pushed) # Studio never serves Ollama; leaving it on costs a failing request per page # load and an empty section in Open WebUI's settings. api.disable_ollama() + ensure_tracing_function(api) _reconcile_model_ids(api, pushed) from chat.access_sync import reconcile_safely @@ -147,3 +152,49 @@ def _admin(): if user is None: raise NoAdminError("No superuser to act as; Open WebUI's provider config needs an admin.") return user + + +# --- tracing function (auto-installed on every push) ------------------------ + +#: The bundled Open WebUI filter that taps every chat turn to the pipelines +#: service for Langfuse tracing. Kept in sync with this file, so code changes +#: are re-pushed automatically on the next startup/signal push. +TRACING_FUNCTION_NAME = "Studio Langfuse Capture" +TRACING_FUNCTION_FILE = os.path.join( + os.path.dirname(__file__), "openwebui_functions", "studio_langfuse_capture.py" +) + + +def ensure_tracing_function(api: ChatAPI | None = None) -> str: + """Install or update the tracing function in Open WebUI. Returns a report line. + + Best-effort: any failure is logged and reported, never raised — tracing is + additive and must not block connection sync. Idempotent; content is only + pushed when it actually changed. + """ + api = api or ChatAPI.as_user(_admin()) + try: + with open(TRACING_FUNCTION_FILE, encoding="utf-8") as fh: + content = fh.read() + except OSError as exc: + return f"Tracing function unavailable ({exc})" + + try: + functions = api.list_functions() + existing = next((f for f in functions if f.get("name") == TRACING_FUNCTION_NAME), None) + if existing is None: + created = api.create_function(TRACING_FUNCTION_NAME, content) + # New functions are active for all users by default; nothing to grant. + return f"Tracing function installed ({created.get('id', '?')})." + if (existing.get("content") or "") != content: + api.update_function_content(existing["id"], TRACING_FUNCTION_NAME, content) + return "Tracing function updated." + return "Tracing function up to date." + except ChatAPIError as exc: + logger.info("Tracing function sync skipped: %s", exc) + return f"Tracing function not synced: {exc}" + except Exception: + # Tracing is additive; anything unexpected (odd API shape, chat down) + # must never break connection sync or startup. + logger.info("Tracing function sync skipped", exc_info=True) + return "Tracing function sync skipped (see logs)." diff --git a/docs/chat.md b/docs/chat.md index 930b5ee..7d448ab 100644 --- a/docs/chat.md +++ b/docs/chat.md @@ -309,16 +309,33 @@ actually send traces you must set both `LANGFUSE_SECRET_KEY` and `LANGFUSE_PUBLIC_KEY` (plus `LANGFUSE_HOST`); without them the service still runs and captures, but skips the Langfuse write. -There are two capture seams, and which one fires depends on how Open WebUI is -wired: - -- **As a provider (recommended):** add an Open WebUI connection whose base URL - is the service (`http://127.0.0.1:9099/v1`, key `0p3n-w3bu!`). Every chat - turn then flows through `POST /v1/chat/completions`, which records one trace - per chat (request + response + token usage) and forwards the call upstream. - This is what Open WebUI actually calls for an OpenAI-compatible connection. -- **As a filter:** if your client drives the Pipelines filter protocol directly, - `POST /v1/{id}/filter/inlet` (user request → span) and +There are three ways to wire Open WebUI to the service, in order of +recommendation: + +- **As an Open WebUI Function (recommended — traces ALL connections, + auto-installed):** Studio installs and keeps up to date the bundled filter + function `chat/openwebui_functions/studio_langfuse_capture.py` on every chat + sync (startup and after each connection change) via Open WebUI's functions + API — no manual step. It runs *inside* Open WebUI's own backend, so no + connection base URLs change and every provider works (OpenAI, + Anthropic-compatible, local models — anything). On each chat turn it posts + the request to `POST /v1/{id}/filter/inlet` before the model call and the + reply to `POST /v1/{id}/filter/outlet` after; both return the payload + untouched, so this is a pure tap. The function targets + `http://helsedir-studio:9099` by default (same Docker network as the studio + container); override with `STUDIO_PIPELINES_URL` or the function's + `pipelines_url` valve if your topology differs. Tracing failures never break + chat. The official Open WebUI Pipelines sidecar is *not* needed — its own + README recommends Functions over Pipelines for exactly this kind of + lightweight filter. +- **As a provider:** add an Open WebUI connection whose base URL is the + service (`http://127.0.0.1:9099/v1`, key `0p3n-w3bu!`). Every chat turn then + flows through `POST /v1/chat/completions`, which records one trace per chat + (request + response + token usage) and forwards the call upstream via + `SIMPLEAUDIT_PIPELINES_FORWARD`. Only chats using that connection are traced, + and the upstream must be OpenAI-compatible. +- **As a raw filter endpoint:** if your client drives the Pipelines filter + protocol directly, `POST /v1/{id}/filter/inlet` (user request → span) and `POST /v1/{id}/filter/outlet` (assistant reply → generation) also record traces. diff --git a/infra/management/commands/dev_server.py b/infra/management/commands/dev_server.py index 079298b..652392b 100644 --- a/infra/management/commands/dev_server.py +++ b/infra/management/commands/dev_server.py @@ -78,7 +78,11 @@ def _sync_chat_models() -> str: except ChatAPIError as exc: return f"Models not synced to chat: {exc}" kept = f", kept {result['kept']} added in chat" if result["kept"] else "" - return f"Synced {result['pushed']} model connection(s) to chat{kept}." + message = f"Synced {result['pushed']} model connection(s) to chat{kept}." + from chat.sync import ensure_tracing_function + + message += f" {ensure_tracing_function()}" + return message def _backfill_demo_chat() -> str: diff --git a/pyproject.toml b/pyproject.toml index 3e09566..b14d243 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "simpleaudit-studio" -version = "0.5.12" +version = "0.5.14" description = "Production AI model audit platform — one-liner demo mode via uvx" readme = "README.md" requires-python = ">=3.11" diff --git a/static/favicon.png b/static/favicon.png new file mode 100644 index 0000000..58b4779 Binary files /dev/null and b/static/favicon.png differ diff --git a/uv.lock b/uv.lock index 786c3f7..2a20be8 100644 --- a/uv.lock +++ b/uv.lock @@ -2686,7 +2686,7 @@ wheels = [ [[package]] name = "simpleaudit-studio" -version = "0.5.12" +version = "0.5.14" source = { virtual = "." } dependencies = [ { name = "caddyserver" },