simple_tool_calling_graph.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from langchain.prompts import (\n",
    "    ChatPromptTemplate,\n",
    "    HumanMessagePromptTemplate,\n",
    "    MessagesPlaceholder,\n",
    "    SystemMessagePromptTemplate,\n",
    ")\n",
    "from helpers import llm\n",
    "from langchain.tools import tool\n",
    "\n",
    "# OpenAI API Wrapper erstellen\n",
    "llm = llm()\n",
    "\n",
    "# Custom Function als Tool definieren\n",
    "\n",
    "\n",
    "@tool\n",
    "def test_tool(question: str):\n",
    "    \"This tool provides information about Fritz Karuugaa. Give a question as input.\"\n",
    "    return \"Fritz Karuugaa is a software developer\"\n",
    "\n",
    "\n",
    "# Tools müssen immer als list übergeben werden\n",
    "tools = [test_tool]\n",
    "\n",
    "# Tools an das llm binden. Dadurch werden automatisch die vollständige Tool beschreibung in das Prompting des LLM übergeben. (Funktioniert nur mit den großen LLM's OpenAI, Claude, etc.)\n",
    "llm_with_tools = llm.bind_tools(tools)\n",
    "\n",
    "# Prompt Template erstellen um dem Agenten einen System prompt mitzugeben.\n",
    "prompt = ChatPromptTemplate.from_messages(\n",
    "    [\n",
    "        SystemMessagePromptTemplate.from_template(\n",
    "            \"Du bist ein hilfsbereiter Assistent. Löse deine Aufgaben auf eine humorvolle Art und weiße. Antworte zusätzlich immer mit einem Witz.\",\n",
    "        ),\n",
    "        HumanMessagePromptTemplate.from_template(\"{input}\"),\n",
    "        MessagesPlaceholder(\"intermediate_steps\"),\n",
    "    ]\n",
    ")\n",
    "\n",
    "# Einfache LCEL Chain die das Prompt Template erstellt und in den LLM call injectet.\n",
    "chain = prompt | llm_with_tools"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "state {'input': 'Was weißt du über Fritz Karuugaa?', 'intermediate_steps': [], 'answer': None}\n",
      "state {'input': 'Was weißt du über Fritz Karuugaa?', 'intermediate_steps': [AIMessage(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_53h7yTYcMPV5oj2t37KNRpLF', 'function': {'arguments': '{\"question\":\"Was weißt du über Fritz Karuugaa?\"}', 'name': 'test_tool'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls'}, id='run-40db209c-c59e-43b1-a461-e28167109a25-0', tool_calls=[{'name': 'test_tool', 'args': {'question': 'Was weißt du über Fritz Karuugaa?'}, 'id': 'call_53h7yTYcMPV5oj2t37KNRpLF'}]), ToolMessage(content='Fritz Karuugaa is a software developer', tool_call_id='call_53h7yTYcMPV5oj2t37KNRpLF')], 'answer': None}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'input': 'Was weißt du über Fritz Karuugaa?',\n",
       " 'intermediate_steps': [AIMessage(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_53h7yTYcMPV5oj2t37KNRpLF', 'function': {'arguments': '{\"question\":\"Was weißt du über Fritz Karuugaa?\"}', 'name': 'test_tool'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls'}, id='run-40db209c-c59e-43b1-a461-e28167109a25-0', tool_calls=[{'name': 'test_tool', 'args': {'question': 'Was weißt du über Fritz Karuugaa?'}, 'id': 'call_53h7yTYcMPV5oj2t37KNRpLF'}]),\n",
       "  ToolMessage(content='Fritz Karuugaa is a software developer', tool_call_id='call_53h7yTYcMPV5oj2t37KNRpLF'),\n",
       "  AIMessage(content='Fritz Karuugaa ist ein Softwareentwickler. Aber warte, hier ist noch etwas: Warum ließ der Softwareentwickler seine Arbeit im Stich? Weil er nicht genug \"Java\" hatte, um durchzustarten! ☕😄', response_metadata={'finish_reason': 'stop'}, id='run-25dec6f3-02c9-4529-ac5e-5543d27b8893-0')]}"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import operator\n",
    "from typing import Annotated, TypedDict\n",
    "\n",
    "from langchain_core.agents import AgentActionMessageLog, AgentFinish\n",
    "from langchain_core.messages import ToolMessage\n",
    "from langchain_core.runnables import RunnableLambda\n",
    "from langgraph.graph import END, StateGraph\n",
    "\n",
    "# State für den Graph definieren\n",
    "\n",
    "\n",
    "class AgentState(TypedDict):\n",
    "    input: str\n",
    "    intermediate_steps: Annotated[list[tuple[AgentActionMessageLog, str]], operator.add]\n",
    "    answer: str\n",
    "\n",
    "# Continue Condition die entscheidet prüft ob ein Tool ausgeben werden soll oder der Graph beendet wird.\n",
    "\n",
    "\n",
    "def should_continue(state):\n",
    "    return \"continue\" if state[\"intermediate_steps\"][-1].tool_calls else \"end\"\n",
    "\n",
    "# Function um das LLM zu triggern und Agentenlogik auszuführen.\n",
    "\n",
    "\n",
    "def call_model(state, config):\n",
    "    response = chain.invoke(state, config=config)\n",
    "    print(\"state\", state)\n",
    "    if isinstance(response, AgentFinish):\n",
    "        return {\"answer\": response.answer}\n",
    "    else:\n",
    "        return {\"intermediate_steps\": [response]}\n",
    "    # return {\"intermediate_steps\": [chain.invoke(state, config=config)]}\n",
    "\n",
    "# Function um lokal die Tools auszuführen.\n",
    "\n",
    "\n",
    "def _invoke_tool(tool_call):\n",
    "    tool = {tool.name: tool for tool in tools}[tool_call[\"name\"]]\n",
    "    return ToolMessage(tool.invoke(tool_call[\"args\"]), tool_call_id=tool_call[\"id\"])\n",
    "\n",
    "\n",
    "tool_executor = RunnableLambda(_invoke_tool)\n",
    "\n",
    "# Function um die Tool Calls auszulesen und parallel auszuführen.\n",
    "\n",
    "\n",
    "def call_tools(state):\n",
    "    last_message = state[\"intermediate_steps\"][-1]\n",
    "    return {\"intermediate_steps\": tool_executor.batch(last_message.tool_calls)}\n",
    "\n",
    "\n",
    "# Gestaltung des Grapen Workflows\n",
    "workflow = StateGraph(AgentState)\n",
    "workflow.add_node(\"agent\", call_model)\n",
    "workflow.add_node(\"action\", call_tools)\n",
    "workflow.set_entry_point(\"agent\")\n",
    "workflow.add_conditional_edges(\n",
    "    \"agent\",\n",
    "    should_continue,\n",
    "    {\n",
    "        \"continue\": \"action\",\n",
    "        \"end\": END,\n",
    "    },\n",
    ")\n",
    "workflow.add_edge(\"action\", \"agent\")\n",
    "\n",
    "# Graph kompilieren.\n",
    "graph = workflow.compile()\n",
    "\n",
    "# Graph mit input ausführen.\n",
    "graph.invoke({\"input\": \"Was weißt du über Fritz Karuugaa?\"})"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}