{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "15c4bd28",
   "metadata": {},
   "source": [
    "# How to stream custom data\n",
    "\n",
    "<div class=\"admonition tip\">\n",
    "    <p class=\"admonition-title\">Prerequisites</p>\n",
    "    <p>\n",
    "        This guide assumes familiarity with the following:\n",
    "        <ul>\n",
    "            <li>            \n",
    "                <a href=\"https://langchain-ai.github.io/langgraphjs/concepts/streaming/\">\n",
    "                    Streaming\n",
    "                </a>\n",
    "            </li>\n",
    "            <li>\n",
    "                <a href=\"https://js.langchain.com/docs/how_to/streaming#using-stream-events\">\n",
    "                    streamEvents API\n",
    "                </a>\n",
    "            </li>\n",
    "            <li>\n",
    "                <a href=\"https://js.langchain.com/docs/concepts/chat_models\">\n",
    "                    Chat Models\n",
    "                </a>\n",
    "            </li>\n",
    "            <li>\n",
    "                <a href=\"https://js.langchain.com/docs/concepts/tools\">\n",
    "                    Tools\n",
    "                </a>\n",
    "            </li>\n",
    "        </ul>\n",
    "    </p>\n",
    "</div>\n",
    "\n",
    "The most common use case for streaming from inside a node is to stream LLM tokens, but you may also want to stream custom data.\n",
    "\n",
    "For example, if you have a long-running tool call, you can dispatch custom events between the steps and use these custom events to monitor progress. You could also surface these custom events to an end user of your application to show them how the current task is progressing.\n",
    "\n",
    "You can do so in two ways:\n",
    "\n",
    "* using your graph's `.stream` method with `streamMode: \"custom\"`\n",
    "* emitting custom events using [`dispatchCustomEvents`](https://js.langchain.com/docs/how_to/callbacks_custom_events/) with `streamEvents`.\n",
    "\n",
    "Below we'll see how to use both APIs.\n",
    "\n",
    "## Setup\n",
    "\n",
    "First, let's install our required packages:\n",
    "\n",
    "```bash\n",
    "npm install @langchain/langgraph @langchain/core\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "12297071",
   "metadata": {},
   "source": [
    "<div class=\"admonition tip\">\n",
    "    <p class=\"admonition-title\">Set up <a href=\"https://smith.langchain.com\">LangSmith</a> for LangGraph development</p>\n",
    "    <p style=\"padding-top: 5px;\">\n",
    "        Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph — read more about how to get started <a href=\"https://docs.smith.langchain.com\">here</a>. \n",
    "    </p>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "29814253-ca9b-4844-a8a5-d6b19fbdbdba",
   "metadata": {},
   "source": [
    "## Stream custom data using .stream\n",
    "\n",
    "<div class=\"admonition tip\">\n",
    "    <p class=\"admonition-title\">Compatibility</p>\n",
    "    <p>\n",
    "        This section requires <code>@langchain/langgraph>=0.2.20</code>. For help upgrading, see <a href=\"/langgraphjs/how-tos/manage-ecosystem-dependencies/\">this guide</a>.\n",
    "    </p>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b729644a-b65f-4e69-ad45-f2e88ffb4e9d",
   "metadata": {},
   "source": [
    "### Define the graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "9731c40f-5ce7-460d-b2ad-33185529c99d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import {\n",
    "  StateGraph,\n",
    "  MessagesAnnotation,\n",
    "  LangGraphRunnableConfig,\n",
    "} from \"@langchain/langgraph\";\n",
    "\n",
    "const myNode = async (\n",
    "  _state: typeof MessagesAnnotation.State,\n",
    "  config: LangGraphRunnableConfig\n",
    ") => {\n",
    "  const chunks = [\n",
    "    \"Four\",\n",
    "    \"score\",\n",
    "    \"and\",\n",
    "    \"seven\",\n",
    "    \"years\",\n",
    "    \"ago\",\n",
    "    \"our\",\n",
    "    \"fathers\",\n",
    "    \"...\",\n",
    "  ];\n",
    "  for (const chunk of chunks) {\n",
    "    // write the chunk to be streamed using streamMode=custom\n",
    "    // Only populated if one of the passed stream modes is \"custom\".\n",
    "    config.writer?.(chunk);\n",
    "  }\n",
    "  return {\n",
    "    messages: [{\n",
    "      role: \"assistant\",\n",
    "      content: chunks.join(\" \"),\n",
    "    }],\n",
    "  };\n",
    "};\n",
    "\n",
    "const graph = new StateGraph(MessagesAnnotation)\n",
    "  .addNode(\"model\", myNode)\n",
    "  .addEdge(\"__start__\", \"model\")\n",
    "  .compile();"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ecd69eed-9624-4640-b0af-c9f82b190900",
   "metadata": {},
   "source": [
    "### Stream content"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "00a91b15-82c7-443c-acb6-a7406df15cee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Four\n",
      "score\n",
      "and\n",
      "seven\n",
      "years\n",
      "ago\n",
      "our\n",
      "fathers\n",
      "...\n"
     ]
    }
   ],
   "source": [
    "const inputs = [{\n",
    "  role: \"user\",\n",
    "  content: \"What are you thinking about?\",\n",
    "}];\n",
    "\n",
    "const stream = await graph.stream(\n",
    "  { messages: inputs },\n",
    "  { streamMode: \"custom\" }\n",
    ");\n",
    "\n",
    "for await (const chunk of stream) {\n",
    "  console.log(chunk);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c7b9f1f0-c170-40dc-9c22-289483dfbc99",
   "metadata": {},
   "source": [
    "You will likely need to use [multiple streaming modes](https://langchain-ai.github.io/langgraphjs/how-tos/stream-multiple/) as you will\n",
    "want access to both the custom data and the state updates."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "f8ed22d4-6ce6-4b04-a68b-2ea516e3ab15",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[ 'custom', 'Four' ]\n",
      "[ 'custom', 'score' ]\n",
      "[ 'custom', 'and' ]\n",
      "[ 'custom', 'seven' ]\n",
      "[ 'custom', 'years' ]\n",
      "[ 'custom', 'ago' ]\n",
      "[ 'custom', 'our' ]\n",
      "[ 'custom', 'fathers' ]\n",
      "[ 'custom', '...' ]\n",
      "[ 'updates', { model: { messages: [Array] } } ]\n"
     ]
    }
   ],
   "source": [
    "const streamMultiple = await graph.stream(\n",
    "  { messages: inputs },\n",
    "  { streamMode: [\"custom\", \"updates\"] }\n",
    ");\n",
    "\n",
    "for await (const chunk of streamMultiple) {\n",
    "  console.log(chunk);\n",
    "}"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "ca976d6a-7c64-4603-8bb4-dee95428c33d",
   "metadata": {},
   "source": [
    "## Stream custom data using .streamEvents\n",
    "\n",
    "If you are already using graph's `.streamEvents` method in your workflow, you can also stream custom data by emitting custom events using `dispatchCustomEvents`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b390a9fe-2d5f-4e82-a1ea-c7c0186b8559",
   "metadata": {},
   "source": [
    "### Define the graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "486a01a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "import { dispatchCustomEvent } from \"@langchain/core/callbacks/dispatch\";\n",
    "\n",
    "const graphNode = async (_state: typeof MessagesAnnotation.State) => {\n",
    "  const chunks = [\n",
    "    \"Four\",\n",
    "    \"score\",\n",
    "    \"and\",\n",
    "    \"seven\",\n",
    "    \"years\",\n",
    "    \"ago\",\n",
    "    \"our\",\n",
    "    \"fathers\",\n",
    "    \"...\",\n",
    "  ];\n",
    "  for (const chunk of chunks) {\n",
    "    await dispatchCustomEvent(\"my_custom_event\", { chunk });\n",
    "  }\n",
    "  return {\n",
    "    messages: [{\n",
    "      role: \"assistant\",\n",
    "      content: chunks.join(\" \"),\n",
    "    }],\n",
    "  };\n",
    "};\n",
    "\n",
    "const graphWithDispatch = new StateGraph(MessagesAnnotation)\n",
    "  .addNode(\"model\", graphNode)\n",
    "  .addEdge(\"__start__\", \"model\")\n",
    "  .compile();"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7dcded03-6776-405e-afae-005a3212d3e4",
   "metadata": {},
   "source": [
    "### Stream content"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ce773a40",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Four|\n",
      "score|\n",
      "and|\n",
      "seven|\n",
      "years|\n",
      "ago|\n",
      "our|\n",
      "fathers|\n",
      "...|\n"
     ]
    }
   ],
   "source": [
    "const eventStream = await graphWithDispatch.streamEvents(\n",
    "  {\n",
    "    messages: [{\n",
    "      role: \"user\",\n",
    "      content: \"What are you thinking about?\",\n",
    "    }]\n",
    "  },\n",
    "  {\n",
    "    version: \"v2\",\n",
    "  },\n",
    ");\n",
    "\n",
    "for await (const { event, name, data } of eventStream) {\n",
    "  if (event === \"on_custom_event\" && name === \"my_custom_event\") {\n",
    "    console.log(`${data.chunk}|`);\n",
    "  }\n",
    "}"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "TypeScript",
   "language": "typescript",
   "name": "tslab"
  },
  "language_info": {
   "codemirror_mode": {
    "mode": "typescript",
    "name": "javascript",
    "typescript": true
   },
   "file_extension": ".ts",
   "mimetype": "text/typescript",
   "name": "typescript",
   "version": "3.7.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}