x

docs/docs/how_to/callbacks_propagate.ipynb

@@ -0,0 +1,340 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# How to propagate callbacks to child components\n",
+    "\n",
+    ":::info Prerequisites\n",
+    "\n",
+    "This guide assumes familiarity with the following concepts:\n",
+    "\n",
+    "- [Callbacks](/docs/concepts/#callbacks)\n",
+    "- [Custom callback handlers](/docs/how_to/custom_callbacks)\n",
+    "\n",
+    ":::\n",
+    "\n",
+    "If you're creating a custom chain that's composed entirely using LCEL, then callbacks will be propagated automatically for you.\n",
+    "\n",
+    "If you're using `RunnableLambda`, `RunnableGenerator` or `@tool` to create custom components, `langchain` will attempt to propagate\n",
+    "callbacks on your behalf from those components to any child `Runnables` if possible.\n",
+    "\n",
+    ":::{.callout-important}\n",
+    "If you're working in an `async` code base and are using `python<=3.10`, you will need to propagate `config` or `callbacks`  since LangChain is unable to do this automatically. `sync` code has no such limitation.\n",
+    "::"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# | output: false\n",
+    "# | echo: false\n",
+    "\n",
+    "%pip install -qU langchain"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Sync (All Python Versions)\n",
+    "\n",
+    "You should see that the custom call back is invoked twice!"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 75,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "---\n",
+      "Invoking custom callback. Recevied inputs: hello\n",
+      "---\n",
+      "Invoking custom callback. Recevied inputs: goodbye\n"
+     ]
+    }
+   ],
+   "source": [
+    "import asyncio\n",
+    "from typing import Any, Dict, List\n",
+    "\n",
+    "from langchain_core.callbacks import BaseCallbackHandler\n",
+    "from langchain_core.runnables import RunnableLambda\n",
+    "\n",
+    "class MyCustomHandler(BaseCallbackHandler):\n",
+    "    \"\"\"Sync callback handler that can be used to handle callbacks from langchain.\"\"\"\n",
+    "    def on_chain_start(self, \n",
+    "        serialized: Dict[str, Any],\n",
+    "        inputs: Dict[str, Any],\n",
+    "        **kwargs: Any,\n",
+    "    ) -> None:\n",
+    "        \"\"\"Run when chain starts running.\"\"\"\n",
+    "        print('---')\n",
+    "        print(f'Invoking custom callback. Recevied inputs: {inputs}')\n",
+    "\n",
+    "@RunnableLambda\n",
+    "def foo(inputs):\n",
+    "    return 'world'\n",
+    "\n",
+    "@RunnableLambda\n",
+    "def bar(inputs):\n",
+    "    return foo.invoke('goodbye')\n",
+    "\n",
+    "bar.invoke('hello', {'callbacks': [MyCustomHandler()]});"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Async Python >=3.11\n",
+    "\n",
+    "The custom callback will be invoked twice!\n",
+    "\n",
+    "This code will work correctly even thought `bar` does not propagate config to `foo` does not propagate config!"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 76,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "'3.11.4 (main, Sep 25 2023, 10:06:23) [GCC 11.4.0]'"
+      ]
+     },
+     "execution_count": 76,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import sys\n",
+    "sys.version"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 77,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "---\n",
+      "Invoking custom callback. Recevied inputs: hello\n",
+      "---\n",
+      "Invoking custom callback. Recevied inputs: goodbye\n"
+     ]
+    },
+    {
+     "data": {
+      "text/plain": [
+       "'world'"
+      ]
+     },
+     "execution_count": 77,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import asyncio\n",
+    "from typing import Any, Dict, List\n",
+    "\n",
+    "from langchain_core.callbacks import AsyncCallbackHandler\n",
+    "from langchain_core.runnables import RunnableLambda\n",
+    "\n",
+    "class AsyncMyCustomHandler(AsyncCallbackHandler):\n",
+    "    \"\"\"Async callback handler that can be used to handle callbacks from langchain.\"\"\"\n",
+    "    async def on_chain_start(self, \n",
+    "        serialized: Dict[str, Any],\n",
+    "        inputs: Dict[str, Any],\n",
+    "        **kwargs: Any,\n",
+    "    ) -> None:\n",
+    "        \"\"\"Run when chain starts running.\"\"\"\n",
+    "        print('---')\n",
+    "        print(f'Invoking custom callback. Recevied inputs: {inputs}')\n",
+    "\n",
+    "@RunnableLambda\n",
+    "async def foo(inputs):\n",
+    "    return 'world'\n",
+    "\n",
+    "@RunnableLambda\n",
+    "async def bar(inputs):\n",
+    "    return await foo.ainvoke('goodbye')\n",
+    "\n",
+    "await bar.ainvoke('hello', {'callbacks': [AsyncMyCustomHandler()]})"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Async Python <=3.10\n",
+    "\n",
+    "### Incorrect\n",
+    "\n",
+    "This code will not work!"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 7,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "'3.9.6 (default, Sep 26 2023, 21:46:56) \\n[GCC 11.4.0]'"
+      ]
+     },
+     "execution_count": 7,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import sys\n",
+    "sys.version"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 8,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "---\n",
+      "Invoking custom callback. Recevied inputs: hello\n"
+     ]
+    },
+    {
+     "data": {
+      "text/plain": [
+       "'world'"
+      ]
+     },
+     "execution_count": 8,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import asyncio\n",
+    "from typing import Any, Dict, List\n",
+    "\n",
+    "from langchain_core.callbacks import AsyncCallbackHandler\n",
+    "from langchain_core.runnables import RunnableLambda\n",
+    "\n",
+    "class AsyncMyCustomHandler(AsyncCallbackHandler):\n",
+    "    \"\"\"Async callback handler that can be used to handle callbacks from langchain.\"\"\"\n",
+    "    async def on_chain_start(self, \n",
+    "        serialized: Dict[str, Any],\n",
+    "        inputs: Dict[str, Any],\n",
+    "        **kwargs: Any,\n",
+    "    ) -> None:\n",
+    "        \"\"\"Run when chain starts running.\"\"\"\n",
+    "        print('---')\n",
+    "        print(f'Invoking custom callback. Recevied inputs: {inputs}')\n",
+    "\n",
+    "@RunnableLambda\n",
+    "async def foo(inputs):\n",
+    "    return 'world'\n",
+    "\n",
+    "@RunnableLambda\n",
+    "async def bar(inputs):\n",
+    "    return await foo.ainvoke('goodbye')\n",
+    "\n",
+    "await bar.ainvoke('hello', {'callbacks': [AsyncMyCustomHandler()]})"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Correct\n",
+    "\n",
+    "This code will work correctly across all supported versions of python.\n",
+    "\n",
+    ":::{.callout-tip}\n",
+    "\n",
+    "This code exposes `config` and propagates it to child component as `foo.ainvoke('goodbye', config=config)`\n",
+    ":::"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 9,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "---\n",
+      "Invoking custom callback. Recevied inputs: hello\n",
+      "---\n",
+      "Invoking custom callback. Recevied inputs: goodbye\n"
+     ]
+    },
+    {
+     "data": {
+      "text/plain": [
+       "'world'"
+      ]
+     },
+     "execution_count": 9,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "@RunnableLambda\n",
+    "async def foo(inputs, config):\n",
+    "    return 'world'\n",
+    "\n",
+    "@RunnableLambda\n",
+    "async def bar(inputs, config):\n",
+    "    return await foo.ainvoke('goodbye', config=config)\n",
+    "\n",
+    "await bar.ainvoke('hello', {'callbacks': [AsyncMyCustomHandler()]})"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3 (ipykernel)",
+   "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.9.6"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}