workflows-py/examples/observability/workflow_context_logging.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "29188397",
   "metadata": {},
   "source": [
    "Llama Index Dispatcher context fields are passed through to workflow runs. Additionally, workflow runs track their individual `run_id`s as a context field. Tracking these fields in logs can be useful to differentiate runs when running with concurrency, and associate them back to a trace.\n",
    "\n",
    "This notebook demonstrates how to integrate with standard library logging, as well as with structlog, for including these fields in logs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d1b89a92",
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install structlog llama-index-workflows"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6bb3c1e0",
   "metadata": {},
   "source": [
    "Set up imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e1942819",
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "from typing import Any, MutableMapping\n",
    "\n",
    "import structlog\n",
    "from llama_index_instrumentation.dispatcher import (\n",
    "    active_instrument_tags,\n",
    "    instrument_tags,\n",
    ")\n",
    "from workflows import Context, Workflow, step\n",
    "from workflows.events import StartEvent, StopEvent"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "552d7606",
   "metadata": {},
   "source": [
    "set up structlog to read from the dispatcher context:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "e5995e5d",
   "metadata": {},
   "outputs": [],
   "source": [
    "def merge_custom_context(\n",
    "    _logger: structlog.BoundLogger,\n",
    "    _method_name: str,\n",
    "    event_dict: MutableMapping[str, Any],\n",
    ") -> MutableMapping[str, Any]:\n",
    "    \"\"\"\n",
    "    Merge values from your ContextVar dict into structlog's event_dict.\n",
    "    Later processors (e.g., JSONRenderer) will see these keys as if bound.\n",
    "    \"\"\"\n",
    "    ctx = active_instrument_tags.get()\n",
    "    if ctx:\n",
    "        # don't clobber explicitly-set event keys unless you want to:\n",
    "        for k, v in ctx.items():\n",
    "            event_dict.setdefault(k, v)\n",
    "            # or: event_dict[k] = v  # if you want your ctx to win\n",
    "    return event_dict\n",
    "\n",
    "\n",
    "structlog.configure(\n",
    "    processors=[\n",
    "        merge_custom_context,  # <------------- Add this to add llama index dispatcher tags to structlog\n",
    "        structlog.processors.add_log_level,\n",
    "        structlog.processors.TimeStamper(fmt=\"%Y-%m-%d %H:%M:%S\", utc=False),\n",
    "        structlog.dev.ConsoleRenderer(),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8e8a195b",
   "metadata": {},
   "source": [
    "Set up structlog to read from the dispatcher context:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "a5e51f78",
   "metadata": {},
   "outputs": [],
   "source": [
    "def merge_custom_context(\n",
    "    _logger: structlog.BoundLogger,\n",
    "    _method_name: str,\n",
    "    event_dict: MutableMapping[str, Any],\n",
    ") -> MutableMapping[str, Any]:\n",
    "    \"\"\"\n",
    "    Merge values from your ContextVar dict into structlog's event_dict.\n",
    "    Later processors (e.g., JSONRenderer) will see these keys as if bound.\n",
    "    \"\"\"\n",
    "    ctx = active_instrument_tags.get()\n",
    "    if ctx:\n",
    "        # don't clobber explicitly-set event keys unless you want to:\n",
    "        for k, v in ctx.items():\n",
    "            event_dict.setdefault(k, v)\n",
    "            # or: event_dict[k] = v  # if you want your ctx to win\n",
    "    return event_dict\n",
    "\n",
    "\n",
    "structlog.configure(\n",
    "    processors=[\n",
    "        merge_custom_context,  # <------------- Add this to add llama index dispatcher tags to structlog\n",
    "        structlog.processors.add_log_level,\n",
    "        structlog.processors.TimeStamper(fmt=\"%Y-%m-%d %H:%M:%S\", utc=False),\n",
    "        structlog.dev.ConsoleRenderer(),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c7f3672d",
   "metadata": {},
   "source": [
    "Set up stdlib logging to include the run_id from the dispatcher context. Note that stdlib logging is much harder to configure correctly, and has difficulty with extra fields being optional or overwritten."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "f1f724fc",
   "metadata": {},
   "outputs": [],
   "source": [
    "old_factory = logging.getLogRecordFactory()\n",
    "\n",
    "\n",
    "def record_factory(*args, **kwargs):\n",
    "    record = old_factory(*args, **kwargs)  # get the unmodified record\n",
    "    record.run_id = active_instrument_tags.get().get(\"run_id\", \"\")\n",
    "    return record\n",
    "\n",
    "\n",
    "logging.setLogRecordFactory(record_factory)\n",
    "\n",
    "logging.basicConfig(level=logging.INFO, format=\"%(message)s run_id=%(run_id)s\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "759295ee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[2m2025-11-05 22:48:59\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mHello from structlog          \u001b[0m\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Hello from stdlib run_id=\n"
     ]
    }
   ],
   "source": [
    "structlog_logger = structlog.get_logger()\n",
    "regular_logger = logging.getLogger()\n",
    "structlog_logger.info(\"Hello from structlog\")\n",
    "regular_logger.info(\"Hello from stdlib\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb68f22b",
   "metadata": {},
   "source": [
    "Set up an example workflow that demonstrates log context:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "83a19c90",
   "metadata": {},
   "outputs": [],
   "source": [
    "class LoggingWorkflow(Workflow):\n",
    "    \"\"\"A workflow that demonstrates log context.\"\"\"\n",
    "\n",
    "    @step\n",
    "    async def log_step(self, ctx: Context, ev: StartEvent) -> StopEvent:\n",
    "        # Any fields bound here will also appear alongside dispatcher tags\n",
    "        structlog_logger.info(\"structlog processing step\")\n",
    "        # Without a more complex wrappers, the fields must be manually passed into standard logging\n",
    "        regular_logger.info(\"regular processing step\")\n",
    "\n",
    "        return StopEvent(result=\"ok\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c8681a0",
   "metadata": {},
   "source": [
    "And run it! Try multiple times and see the run_id change."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "baa2b7d6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[2m2025-11-05 22:51:47\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mstructlog processing step     \u001b[0m \u001b[36mrequest_id\u001b[0m=\u001b[35mreq-123\u001b[0m \u001b[36mrun_id\u001b[0m=\u001b[35mlBUAX17ywM\u001b[0m \u001b[36muser\u001b[0m=\u001b[35malice\u001b[0m\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "regular processing step run_id=lBUAX17ywM\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[2m2025-11-05 22:51:47\u001b[0m [\u001b[32m\u001b[1minfo     \u001b[0m] \u001b[1mfinal result 'ok'             \u001b[0m\n"
     ]
    }
   ],
   "source": [
    "# Tags set outside the workflow run will be captured in all logs emitted\n",
    "# during the run (together with run_id injected by the broker).\n",
    "wf = LoggingWorkflow()\n",
    "\n",
    "with instrument_tags({\"request_id\": \"req-123\", \"user\": \"alice\"}):\n",
    "    result = await wf.run()\n",
    "structlog_logger.info(f\"final result '{result}'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c64826c9",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.14.0b4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}