{
 "cells": [
  {
   "attachments": {
    "a108ffc8-6136-4cd7-a6f9-579e41a5a786.png": {
     "image/png": "[encoded data, 152344 characters]"
    }
   },
   "cell_type": "markdown",
   "id": "95a87145-34d0-4f97-b45f-5c9fd8532c8a",
   "metadata": {},
   "source": [
    "# How to create map-reduce branches for parallel execution\n",
    "\n",
    "[Map-reduce](https://en.wikipedia.org/wiki/MapReduce) operations are essential for efficient task decomposition and parallel processing. This approach involves breaking a task into smaller sub-tasks, processing each sub-task in parallel, and aggregating the results across all of the completed sub-tasks. \n",
    "\n",
    "Consider this example: given a general topic from the user, generate a list of related subjects, generate a joke for each subject, and select the best joke from the resulting list. In this design pattern, a first node may generate a list of objects (e.g., related subjects) and we want to apply some other node (e.g., generate a joke) to all those objects (e.g., subjects). However, two main challenges arise.\n",
    " \n",
    "(1) the number of objects (e.g., subjects) may be unknown ahead of time (meaning the number of edges may not be known) when we lay out the graph and (2) the input State to the downstream Node should be different (one for each generated object).\n",
    "  \n",
    "LangGraph addresses these challenges [through its `Send` API](https://langchain-ai.github.io/langgraph/concepts/low_level/#send). By utilizing conditional edges, `Send` can distribute different states (e.g., subjects) to multiple instances of a node (e.g., joke generation). Importantly, the sent state can differ from the core graph's state, allowing for flexible and dynamic workflow management. \n",
    "\n",
    "![Screenshot 2024-07-12 at 9.45.40 AM.png](attachment:a108ffc8-6136-4cd7-a6f9-579e41a5a786.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3eb04cd1",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%capture --no-stderr\n",
    "%pip install -U langchain-anthropic langgraph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "dc292321",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import getpass\n",
    "\n",
    "\n",
    "def _set_env(name: str):\n",
    "    if not os.getenv(name):\n",
    "        os.environ[name] = getpass.getpass(f\"{name}: \")\n",
    "\n",
    "\n",
    "_set_env(\"ANTHROPIC_API_KEY\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "0f0f78e4-423d-4e2d-aa1a-01efaec4715f",
   "metadata": {},
   "outputs": [],
   "source": [
    "import operator\n",
    "from typing import Annotated, TypedDict\n",
    "\n",
    "from langchain_core.pydantic_v1 import BaseModel\n",
    "from langchain_anthropic import ChatAnthropic\n",
    "\n",
    "from langgraph.constants import Send\n",
    "from langgraph.graph import END, StateGraph, START\n",
    "\n",
    "# Model and prompts\n",
    "# Define model and prompts we will use\n",
    "subjects_prompt = \"\"\"Generate a comma separated list of between 2 and 5 examples related to: {topic}.\"\"\"\n",
    "joke_prompt = \"\"\"Generate a joke about {subject}\"\"\"\n",
    "best_joke_prompt = \"\"\"Below are a bunch of jokes about {topic}. Select the best one! Return the ID of the best one.\n",
    "\n",
    "{jokes}\"\"\"\n",
    "\n",
    "\n",
    "class Subjects(BaseModel):\n",
    "    subjects: list[str]\n",
    "\n",
    "\n",
    "class Joke(BaseModel):\n",
    "    joke: str\n",
    "\n",
    "\n",
    "class BestJoke(BaseModel):\n",
    "    id: int\n",
    "\n",
    "\n",
    "model = ChatAnthropic(model=\"claude-3-5-sonnet-20240620\")\n",
    "\n",
    "# Graph components: define the components that will make up the graph\n",
    "\n",
    "\n",
    "# This will be the overall state of the main graph.\n",
    "# It will contain a topic (which we expect the user to provide)\n",
    "# and then will generate a list of subjects, and then a joke for\n",
    "# each subject\n",
    "class OverallState(TypedDict):\n",
    "    topic: str\n",
    "    subjects: list\n",
    "    # Notice here we use the operator.add\n",
    "    # This is because we want combine all the jokes we generate\n",
    "    # from individual nodes back into one list - this is essentially\n",
    "    # the \"reduce\" part\n",
    "    jokes: Annotated[list, operator.add]\n",
    "    best_selected_joke: str\n",
    "\n",
    "\n",
    "# This will be the state of the node that we will \"map\" all\n",
    "# subjects to in order to generate a joke\n",
    "class JokeState(TypedDict):\n",
    "    subject: str\n",
    "\n",
    "\n",
    "# This is the function we will use to generate the subjects of the jokes\n",
    "def generate_topics(state: OverallState):\n",
    "    prompt = subjects_prompt.format(topic=state[\"topic\"])\n",
    "    response = model.with_structured_output(Subjects).invoke(prompt)\n",
    "    return {\"subjects\": response.subjects}\n",
    "\n",
    "\n",
    "# Here we generate a joke, given a subject\n",
    "def generate_joke(state: JokeState):\n",
    "    prompt = joke_prompt.format(subject=state[\"subject\"])\n",
    "    response = model.with_structured_output(Joke).invoke(prompt)\n",
    "    return {\"jokes\": [response.joke]}\n",
    "\n",
    "\n",
    "# Here we define the logic to map out over the generated subjects\n",
    "# We will use this an edge in the graph\n",
    "def continue_to_jokes(state: OverallState):\n",
    "    # We will return a list of `Send` objects\n",
    "    # Each `Send` object consists of the name of a node in the graph\n",
    "    # as well as the state to send to that node\n",
    "    return [Send(\"generate_joke\", {\"subject\": s}) for s in state[\"subjects\"]]\n",
    "\n",
    "\n",
    "# Here we will judge the best joke\n",
    "def best_joke(state: OverallState):\n",
    "    jokes = \"\\n\\n\".join(state[\"jokes\"])\n",
    "    prompt = best_joke_prompt.format(topic=state[\"topic\"], jokes=jokes)\n",
    "    response = model.with_structured_output(BestJoke).invoke(prompt)\n",
    "    return {\"best_selected_joke\": state[\"jokes\"][response.id]}\n",
    "\n",
    "\n",
    "# Construct the graph: here we put everything together to construct our graph\n",
    "graph = StateGraph(OverallState)\n",
    "graph.add_node(\"generate_topics\", generate_topics)\n",
    "graph.add_node(\"generate_joke\", generate_joke)\n",
    "graph.add_node(\"best_joke\", best_joke)\n",
    "graph.add_edge(START, \"generate_topics\")\n",
    "graph.add_conditional_edges(\"generate_topics\", continue_to_jokes, [\"generate_joke\"])\n",
    "graph.add_edge(\"generate_joke\", \"best_joke\")\n",
    "graph.add_edge(\"best_joke\", END)\n",
    "app = graph.compile()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "37ed1f71-63db-416f-b715-4617b33d4b7f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/jpeg": "[encoded data, 14648 characters]",
      "text/plain": [
       "<IPython.core.display.Image object>"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from IPython.display import Image\n",
    "\n",
    "Image(app.get_graph().draw_mermaid_png())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "fd90cace",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'generate_topics': {'subjects': ['lion', 'elephant', 'penguin', 'dolphin']}}\n",
      "{'generate_joke': {'jokes': [\"Why don't elephants use computers? They're afraid of the mouse!\"]}}\n",
      "{'generate_joke': {'jokes': [\"Why don't dolphins use smartphones? They're afraid of phishing!\"]}}\n",
      "{'generate_joke': {'jokes': [\"Why don't lions like fast food? Because they can't catch it!\"]}}\n",
      "{'generate_joke': {'jokes': [\"Why don't you see penguins in Britain? Because they're afraid of Wales!\"]}}\n",
      "{'best_joke': {'best_selected_joke': \"Why don't you see penguins in Britain? Because they're afraid of Wales!\"}}\n"
     ]
    }
   ],
   "source": [
    "# Call the graph: here we call it to generate a list of jokes\n",
    "for s in app.stream({\"topic\": \"animals\"}):\n",
    "    print(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f28eaf56",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
