{ "cells": [ { "cell_type": "markdown", "id": "c62413a1", "metadata": {}, "source": [ "# πŸ€– Building AI Agents - From LLMs to Autonomous Systems!\n", "\n", "## πŸ€” What Are Agents?\n", "\n", "An **AI Agent** is an LLM that can **reason**, **plan**, and **take actions** by using tools. Instead of just generating text, agents can interact with the real world!\n", "\n", "**The Evolution:**\n", "- πŸ“ LLM alone β†’ Can only generate text\n", "- 🧠 LLM + Memory β†’ Can remember past conversations\n", "- πŸ”§ LLM + Tools β†’ Can take actions (search, calculate, call APIs)\n", "- πŸ€– LLM + Memory + Tools + Reasoning = **Agent!**\n", "\n", "---\n", "\n", "### 🌟 Why Agents Are Revolutionary:\n", "\n", "| Plain LLM | Agent |\n", "|-----------|-------|\n", "| πŸ”€ Generates text only | πŸ”§ Can use tools & APIs |\n", "| ❌ No memory between calls | βœ… Remembers conversation context |\n", "| 🚫 Can't take actions | 🎯 Can search, calculate, retrieve data |\n", "| πŸ“ Limited to training data | 🌐 Can access live information |\n", "\n", "---\n", "\n", "### πŸŽ“ What You'll Learn:\n", "\n", "1. πŸ’¬ **Memory** - Give your LLM conversation history\n", "2. πŸ”§ **Tool Calling** - Let the LLM use functions\n", "3. πŸ”„ **The ReAct Pattern** - Reason + Act in a loop\n", "4. πŸ—οΈ **LangGraph Agents** - Build agents with frameworks\n", "5. 🧩 **Multi-Tool Agents** - Combine multiple capabilities\n", "6. πŸš€ **Advanced Patterns** - Structured output, RAG agents\n", "\n", "All examples use **Mistral AI** models! πŸ‡«πŸ‡·\n", "\n", "Let's dive in! πŸš€" ] }, { "cell_type": "markdown", "id": "0a957e7e", "metadata": {}, "source": [ "> **πŸ“ Note:** You'll need a `MISTRAL_API_KEY` to follow along.\n", ">\n", "> **πŸ”‘ Get Your Free API Key:**\n", "> 1. Go to [Mistral AI Console](https://console.mistral.ai/home)\n", "> 2. Sign up or log in\n", "> 3. Navigate to \"API Keys\" section\n", "> 4. Create a new API key\n", "> 5. Copy your key and keep it safe! πŸ”’" ] }, { "cell_type": "markdown", "id": "e7be997e", "metadata": {}, "source": [ "## πŸ”‘ Step 1: Environment Setup\n", "\n", "First, let's install the dependencies and set up our API key! πŸ”’" ] }, { "cell_type": "code", "execution_count": 207, "id": "0161884a", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… API key configured!\n" ] } ], "source": [ "import os\n", "import time\n", "\n", "os.environ[\"MISTRAL_API_KEY\"] = \"EIdRWsLoyAH6ATXuYgHShsVd4n9Z4JPl\"\n", "\n", "print(\"βœ… API key configured!\")\n" ] }, { "cell_type": "code", "execution_count": 208, "id": "2f70718f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸŽ‰ Model is working!\n", "πŸ’¬ \"Hello there!\"\n" ] } ], "source": [ "from langchain_mistralai import ChatMistralAI\n", "\n", "# Initialize our base model - we'll reuse this throughout the notebook\n", "model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", "\n", "# Quick test\n", "response = model.invoke(\"Say hello in one sentence!\")\n", "print(\"πŸŽ‰ Model is working!\")\n", "print(f\"πŸ’¬ {response.content}\")\n" ] }, { "cell_type": "markdown", "id": "e1c46522", "metadata": {}, "source": [ "---\n", "\n", "# 🧠 Part 1: Memory - Giving Your LLM a Brain\n", "\n", "LLMs are **stateless** β€” they have no idea about previous interactions. Every API call starts from scratch!\n", "\n", "```\n", "Without Memory: With Memory:\n", "β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”\n", "β”‚ User: Hi! β”‚ β”‚ User: Hi! β”‚\n", "β”‚ AI: Hello! β”‚ β”‚ AI: Hello! β”‚\n", "β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ User: Name? β”‚\n", "β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ AI: I'm... β”‚\n", "β”‚ User: Name? β”‚ β”‚ User: ??? β”‚\n", "β”‚ AI: ??? β”‚ ← Forgot! β”‚ AI: Knows! β”‚ ← Remembers!\n", "β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜\n", "```\n", "\n", "**Memory = passing conversation history back to the LLM on each call.**" ] }, { "cell_type": "markdown", "id": "33028603", "metadata": {}, "source": [ "### πŸ” The Problem: LLMs Are Stateless\n", "\n", "Let's see this in action β€” the model forgets between calls:" ] }, { "cell_type": "code", "execution_count": 209, "id": "15d79a97", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“© Call 1: That's awesome, Alice! πŸŽ‰ Python is such a versatile and beginner-friendly language, and it's great to hear you're passionate about it.\n", "\n", "If you'd like, I can help you with:\n", "- **Learning Python** (concepts, best practices, or project ideas)\n", "- **Debugging code** (share snippets, and I’ll help!)\n", "- **Exploring libraries** (e.g., Django, Flask, NumPy, Pandas, etc.)\n", "- **Career advice** (how to level up your Python skills for jobs)\n", "\n", "What would you like to focus on today? 😊\n", "\n", "*(P.S. If you have a fun Python project you're working on, I’d love to hear about it!)*\n", "\n", "============================================================\n", "\n", "πŸ“© Call 2: I don’t have access to personal information about you unless you’ve shared it with me during our conversation. If you’d like, you can tell me your name and what you love, and I’ll remember it for our chat! 😊\n", "\n", "For example, you could say:\n", "*\"My name is [Your Name], and I love [Your Passion]!\"*\n", "\n", "Then I can respond with something like:\n", "*\"Nice to meet you, [Your Name]! That’s awesome that you love [Your Passion]β€”I’d love to hear more about it!\"*\n", "\n", "❌ The model forgot everything from the first call!\n" ] } ], "source": [ "from langchain_core.messages import AIMessage, HumanMessage, SystemMessage\n", "\n", "# First call\n", "response1 = model.invoke(\"My name is Alice and I love Python programming.\")\n", "print(\"πŸ“© Call 1:\", response1.content)\n", "\n", "print(\"\\n\" + \"=\"*60 + \"\\n\")\n", "\n", "# Second call β€” the model has NO idea about the first call!\n", "response2 = model.invoke(\"What is my name and what do I love?\")\n", "print(\"πŸ“© Call 2:\", response2.content)\n", "print(\"\\n❌ The model forgot everything from the first call!\")\n" ] }, { "cell_type": "markdown", "id": "137f5b83", "metadata": {}, "source": [ "### βœ… The Solution: Pass Conversation History\n", "\n", "We manually pass all previous messages each time:" ] }, { "cell_type": "code", "execution_count": 210, "id": "f87a71d4", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“© With history: Your name is Alice and you love Python programming.\n", "\n", "βœ… The model remembers because we passed the full conversation!\n" ] } ], "source": [ "# Build conversation history manually\n", "conversation = [\n", " SystemMessage(content=\"You are a friendly assistant. Remember everything the user tells you.\"),\n", " HumanMessage(content=\"My name is Alice and I love Python programming.\"),\n", " AIMessage(content=\"Nice to meet you, Alice! Python is a great language.\"),\n", " HumanMessage(content=\"What is my name and what do I love?\"),\n", "]\n", "\n", "response = model.invoke(conversation)\n", "print(\"πŸ“© With history:\", response.content)\n", "print(\"\\nβœ… The model remembers because we passed the full conversation!\")\n" ] }, { "cell_type": "markdown", "id": "365eafb0", "metadata": {}, "source": [ "### πŸ—οΈ Building a Simple Memory Manager\n", "\n", "Let's create a reusable `ChatBot` class with memory:" ] }, { "cell_type": "code", "execution_count": 211, "id": "b3bd9fac", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… ChatBot class created!\n" ] } ], "source": [ "\n", "from langchain_core.messages import BaseMessage\n", "\n", "\n", "class ChatBot:\n", " \"\"\"A simple chatbot with conversation memory.\"\"\"\n", "\n", " def __init__(self, system_prompt: str = \"You are a helpful assistant.\") -> None:\n", " \"\"\"Initialize the chatbot with a system prompt and an empty history.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.history: list[BaseMessage] = [\n", " SystemMessage(content=system_prompt),\n", " ]\n", "\n", " def chat(self, user_message: str) -> str | list:\n", " \"\"\"Send a message and get a response, maintaining history.\"\"\"\n", " # Add user message to history\n", " self.history.append(HumanMessage(content=user_message))\n", "\n", " # Call the LLM with FULL history\n", " response = self.model.invoke(self.history)\n", "\n", " # Add AI response to history\n", " self.history.append(AIMessage(content=response.content))\n", "\n", " return response.content\n", "\n", " def get_history_length(self) -> int:\n", " \"\"\"Get the number of messages in the conversation history.\"\"\"\n", " return len(self.history)\n", "\n", "\n", "print(\"βœ… ChatBot class created!\")\n" ] }, { "cell_type": "code", "execution_count": 212, "id": "cc6d4792", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ User: My name is Bob and I'm learning Python.\n", "πŸ€– Bot: Hello Bob! I'm here to help you learn Python. Let's start with the basics. What would you like to learn first?\n", " πŸ“Š History size: 3 messages\n", "\n", "πŸ‘€ User: What's my name and what am I learning?\n", "πŸ€– Bot: Your name is Bob and you're learning Python.\n", " πŸ“Š History size: 5 messages\n", "\n", "πŸ‘€ User: Give me a tip about what I'm learning.\n", "πŸ€– Bot: Tip: In Python, indentation matters! Use 4 spaces or a tab to indent your code blocks.\n", " πŸ“Š History size: 7 messages\n" ] } ], "source": [ "# Test our chatbot with memory!\n", "bot = ChatBot(system_prompt=\"You are a friendly coding tutor. Keep answers short and clear.\")\n", "\n", "# Turn 1\n", "print(\"πŸ‘€ User: My name is Bob and I'm learning Python.\")\n", "print(f\"πŸ€– Bot: {bot.chat('My name is Bob and I am learning Python.')}\")\n", "print(f\" πŸ“Š History size: {bot.get_history_length()} messages\\n\")\n", "\n", "# Turn 2 β€” Does it remember?\n", "print(\"πŸ‘€ User: What's my name and what am I learning?\")\n", "print(f\"πŸ€– Bot: {bot.chat('What is my name and what am I learning?')}\")\n", "print(f\" πŸ“Š History size: {bot.get_history_length()} messages\\n\")\n", "\n", "# Turn 3 β€” Continuing the conversation\n", "print(\"πŸ‘€ User: Give me a tip about what I'm learning.\")\n", "print(f\"πŸ€– Bot: {bot.chat('Give me a tip about what I am learning.')}\")\n", "print(f\" πŸ“Š History size: {bot.get_history_length()} messages\")\n" ] }, { "cell_type": "markdown", "id": "c659186a", "metadata": {}, "source": [ "### ⚠️ The Sliding Window Problem\n", "\n", "As conversations grow, you'll hit the model's **context window limit**. A common solution is a **sliding window** β€” keep only the last N messages:\n", "\n", "```\n", "Full History: [msg1, msg2, msg3, msg4, msg5, msg6, msg7, msg8]\n", "Window (last 4): [ msg5, msg6, msg7, msg8]\n", "```" ] }, { "cell_type": "code", "execution_count": 213, "id": "0a5ce995", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ My favorite color is blue.\n", "πŸ€– That's great! Blue is a wonderful color. It's often associated with calmness, tranquility, and stability. Do you have a specific shade of blue that you like the most, such as sky blue, navy blue, or perhaps something else?\n", "\n", "πŸ‘€ I live in Paris.\n", "πŸ€– Paris is a beautiful city! With its rich history, stunning architecture, and vibrant culture, there's always something to see and do. Since blue is your favorite color, you might enjoy visiting places like the MusΓ©e d'Orsay, which has a beautiful blue dome, or taking a stroll along the Seine River, which is often reflected in shades of blue. Have you been to any of these places, or do you have a favorite spot in Paris?\n", "\n", "πŸ‘€ I work as a data scientist.\n", "πŸ€– That's fascinating! Data science is a field that combines statistics, computer science, and domain expertise to extract insights from structured and unstructured data. As a data scientist in Paris, you're likely surrounded by a thriving tech community and have access to many exciting opportunities.\n", "\n", "Here are a few questions to help me understand your work better:\n", "\n", "1. What aspects of data science do you enjoy the most? (e.g., data cleaning, exploratory data analysis, machine learning, visualization, etc.)\n", "2. Do you work in a specific industry, such as finance, healthcare, or e-commerce?\n", "3. Are there any particular tools, programming languages, or libraries that you prefer to use in your work?\n", "4. How do you stay updated with the latest trends and developments in data science?\n", "\n", "I'm here to help with any questions or discussions related to data science, Paris, or any other topic you're interested in.\n", "\n", "πŸ‘€ What's my favorite color?\n", "πŸ€– Based on our previous conversation, I mentioned that blue is your favorite color. Is that correct, or would you like to share your actual favorite color? I'm here to help and learn more about your preferences!\n", "\n" ] } ], "source": [ "class SlidingWindowChatBot:\n", " \"\"\"ChatBot with a sliding window memory to avoid exceeding context limits.\"\"\"\n", "\n", " def __init__(self, system_prompt: str = \"You are a helpful assistant.\", max_messages: int = 10) -> None:\n", " \"\"\"Initialize the chatbot with a system prompt, empty history, and max message limit.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.system_message = SystemMessage(content=system_prompt)\n", " self.history: list[BaseMessage] = []\n", " self.max_messages = max_messages # Max user+AI messages to keep\n", "\n", " def chat(self, user_message: str) -> str:\n", " \"\"\"Send a message and get a response, maintaining a sliding window of history.\"\"\"\n", " self.history.append(HumanMessage(content=user_message))\n", "\n", " # Apply sliding window β€” always keep system message + last N messages\n", " windowed_history = [self.system_message, *self.history[-self.max_messages:]]\n", "\n", " response = self.model.invoke(windowed_history)\n", " self.history.append(AIMessage(content=response.content))\n", "\n", " return response.content\n", "\n", "\n", "# Test with a small window\n", "bot = SlidingWindowChatBot(max_messages=4)\n", "\n", "messages = [\n", " \"My favorite color is blue.\",\n", " \"I live in Paris.\",\n", " \"I work as a data scientist.\",\n", " \"What's my favorite color?\", # This might be forgotten with window=4!\n", "]\n", "\n", "for msg in messages:\n", " print(f\"πŸ‘€ {msg}\")\n", " print(f\"πŸ€– {bot.chat(msg)}\\n\")\n" ] }, { "cell_type": "markdown", "id": "8e987f1c", "metadata": {}, "source": [ "---\n", "\n", "# πŸ”§ Part 2: Tool Calling - Extending LLM Capabilities\n", "\n", "Agents need to interact with the real world, and **tools** are the foundation that allows LLMs to take actions beyond generating text.\n", "\n", "```\n", "Without Tools: With Tools:\n", "β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”\n", "β”‚ User: What's β”‚ β”‚ User: What's β”‚\n", "β”‚ 847 Γ— 293? β”‚ β”‚ 847 Γ— 293? β”‚\n", "β”‚ β”‚ β”‚ β”‚\n", "β”‚ LLM: Hmm... let β”‚ β”‚ LLM: I'll use β”‚\n", "β”‚ me try... 248071? β”‚ ← Wrong! β”‚ the calculator β”‚\n", "β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β†’ calc(847*293) β”‚\n", " β”‚ β†’ 248,171 βœ… β”‚\n", " β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜\n", "```\n", "\n", "### 🎯 How Tool Calling Works:\n", "\n", "1. You **define tools** (functions with descriptions)\n", "2. You send the tool definitions to the LLM along with the user's query\n", "3. The LLM **decides** which tool(s) to use and with what arguments\n", "4. You **execute** the tool and send results back to the LLM\n", "5. The LLM generates a **final response** using the tool results" ] }, { "cell_type": "markdown", "id": "0c2a94c1", "metadata": {}, "source": [ "### πŸ“ Defining Tools with LangChain\n", "\n", "LangChain makes it easy to define tools using the `@tool` decorator. The docstring becomes the tool description that the LLM reads!" ] }, { "cell_type": "code", "execution_count": 214, "id": "5d74f69b", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… Tools defined!\n", "\n", "πŸ“ Calculator tool: calculator\n", " Description: Perform a mathematical calculation.\n", "\n", "Args:\n", " operation: The operation to perform. One of: add, subtract, multiply, divide\n", " x: The first number\n", " y: The second number\n", "\n", "🌀️ Weather tool: get_weather\n", " Description: Get the current weather for a given city.\n", "\n", "Args:\n", " city: The name of the city to get weather for\n" ] } ], "source": [ "from langchain_core.tools import tool\n", "\n", "\n", "@tool\n", "def calculator(operation: str, x: float, y: float) -> str:\n", " \"\"\"Perform a mathematical calculation.\n", "\n", " Args:\n", " operation: The operation to perform. One of: add, subtract, multiply, divide\n", " x: The first number\n", " y: The second number\n", "\n", " \"\"\"\n", " operations = {\n", " \"add\": x + y,\n", " \"subtract\": x - y,\n", " \"multiply\": x * y,\n", " \"divide\": x / y if y != 0 else \"Error: Division by zero\",\n", " }\n", " result = operations.get(operation, f\"Unknown operation: {operation}\")\n", " return f\"{x} {operation} {y} = {result}\"\n", "\n", "\n", "@tool\n", "def get_weather(city: str) -> str:\n", " \"\"\"Get the current weather for a given city.\n", "\n", " Args:\n", " city: The name of the city to get weather for\n", "\n", " \"\"\"\n", " # Simulated weather data (in production, you'd call a real API)\n", " weather_data = {\n", " \"paris\": \"β˜€οΈ 22Β°C, Sunny with light clouds\",\n", " \"london\": \"🌧️ 15Β°C, Rainy\",\n", " \"new york\": \"β›… 28Β°C, Partly cloudy\",\n", " \"tokyo\": \"🌀️ 25Β°C, Clear skies\",\n", " \"san francisco\": \"🌫️ 18Β°C, Foggy\",\n", " }\n", " city_lower = city.lower()\n", " if city_lower in weather_data:\n", " return f\"Weather in {city}: {weather_data[city_lower]}\"\n", " return f\"Weather in {city}: 🌑️ 20Β°C, Fair weather (simulated)\"\n", "\n", "\n", "print(\"βœ… Tools defined!\")\n", "print(f\"\\nπŸ“ Calculator tool: {calculator.name}\")\n", "print(f\" Description: {calculator.description}\")\n", "print(f\"\\n🌀️ Weather tool: {get_weather.name}\")\n", "print(f\" Description: {get_weather.description}\")\n" ] }, { "cell_type": "markdown", "id": "c7335672", "metadata": {}, "source": [ "### πŸ”— Binding Tools to the Model\n", "\n", "We tell the model which tools are available using `bind_tools()`:" ] }, { "cell_type": "code", "execution_count": 215, "id": "edefc405", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“© Response content: \n", "\n", "πŸ”§ Tool calls: [{'name': 'calculator', 'args': {'operation': 'multiply', 'x': 847, 'y': 293}, 'id': '7xFRkWGpc', 'type': 'tool_call'}]\n", "\n", "πŸ’‘ The model didn't answer directly β€” it wants to use the calculator tool!\n" ] } ], "source": [ "# Bind tools to the model\n", "tools = [calculator, get_weather]\n", "model_with_tools = model.bind_tools(tools)\n", "\n", "# Now when we invoke the model, it can CHOOSE to use tools!\n", "response = model_with_tools.invoke(\"What is 847 multiplied by 293?\")\n", "\n", "print(\"πŸ“© Response content:\", response.content)\n", "print(\"\\nπŸ”§ Tool calls:\", response.tool_calls)\n", "print(\"\\nπŸ’‘ The model didn't answer directly β€” it wants to use the calculator tool!\")\n" ] }, { "cell_type": "markdown", "id": "eafb052c", "metadata": {}, "source": [ "### πŸ” Understanding Tool Call Response\n", "\n", "When the model wants to use a tool, it returns a `tool_calls` list instead of a direct answer:\n", "\n", "```python\n", "response.tool_calls = [\n", " {\n", " 'name': 'calculator', # Which tool to call\n", " 'args': { # With what arguments\n", " 'operation': 'multiply',\n", " 'x': 847,\n", " 'y': 293\n", " },\n", " 'id': 'call_abc123' # Unique ID for this call\n", " }\n", "]\n", "```\n", "\n", "The LLM **doesn't execute the tool** β€” it just tells you which tool to call and with what arguments. **You** execute the tool and send results back." ] }, { "cell_type": "code", "execution_count": 216, "id": "41b4ac58", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Step 1 β€” LLM decides to use a tool:\n", " Tool: calculator\n", " Args: {'operation': 'multiply', 'x': 847, 'y': 293}\n", "\n", "Step 2 β€” We execute the tool:\n", " Result: 847.0 multiply 293.0 = 248171.0\n", "\n", "Step 3 β€” LLM generates final answer:\n", " πŸ’¬ The result of 847 multiplied by 293 is 248,171.\n" ] } ], "source": [ "# Let's manually execute the tool and complete the conversation\n", "from langchain_core.messages import ToolMessage\n", "\n", "# Step 1: Ask the model\n", "user_msg = HumanMessage(content=\"What is 847 multiplied by 293?\")\n", "ai_response = model_with_tools.invoke([user_msg])\n", "\n", "print(\"Step 1 β€” LLM decides to use a tool:\")\n", "print(f\" Tool: {ai_response.tool_calls[0]['name']}\")\n", "print(f\" Args: {ai_response.tool_calls[0]['args']}\")\n", "\n", "# Step 2: Execute the tool\n", "tool_call = ai_response.tool_calls[0]\n", "tool_result = calculator.invoke(tool_call[\"args\"])\n", "\n", "print(\"\\nStep 2 β€” We execute the tool:\")\n", "print(f\" Result: {tool_result}\")\n", "\n", "# Step 3: Send the tool result back to the LLM\n", "tool_msg = ToolMessage(content=str(tool_result), tool_call_id=tool_call[\"id\"])\n", "\n", "final_response = model_with_tools.invoke([user_msg, ai_response, tool_msg])\n", "\n", "print(\"\\nStep 3 β€” LLM generates final answer:\")\n", "print(f\" πŸ’¬ {final_response.content}\")\n" ] }, { "cell_type": "markdown", "id": "543fba17", "metadata": {}, "source": [ "### 🌀️ Testing with the Weather Tool\n", "\n", "The model **chooses** the right tool based on the question:" ] }, { "cell_type": "code", "execution_count": 217, "id": "3dcbbd72", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ”§ Tool chosen: get_weather\n", " Args: {'city': 'Paris'}\n", "\n", "🌀️ Tool result: Weather in Paris: β˜€οΈ 22Β°C, Sunny with light clouds\n", "\n", "πŸ’¬ Final answer: The weather in Paris is currently sunny with light clouds and a temperature of 22Β°C.\n" ] } ], "source": [ "# The model will pick the weather tool this time!\n", "user_msg = HumanMessage(content=\"What's the weather like in Paris?\")\n", "ai_response = model_with_tools.invoke([user_msg])\n", "\n", "print(f\"πŸ”§ Tool chosen: {ai_response.tool_calls[0]['name']}\")\n", "print(f\" Args: {ai_response.tool_calls[0]['args']}\")\n", "\n", "# Execute the tool\n", "tool_call = ai_response.tool_calls[0]\n", "tool_result = get_weather.invoke(tool_call[\"args\"])\n", "print(f\"\\n🌀️ Tool result: {tool_result}\")\n", "\n", "# Get final response\n", "tool_msg = ToolMessage(content=str(tool_result), tool_call_id=tool_call[\"id\"])\n", "final_response = model_with_tools.invoke([user_msg, ai_response, tool_msg])\n", "print(f\"\\nπŸ’¬ Final answer: {final_response.content}\")\n" ] }, { "cell_type": "markdown", "id": "adbc7577", "metadata": {}, "source": [ "### πŸ’‘ When the Model Doesn't Need Tools\n", "\n", "If the question doesn't require a tool, the model answers directly:" ] }, { "cell_type": "code", "execution_count": 218, "id": "be9dac02", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ”§ Tool calls: []\n", "πŸ’¬ Direct answer: The capital of France is Paris.\n", "\n", "βœ… The model is smart enough to know it doesn't need a tool here!\n" ] } ], "source": [ "# A question that doesn't need any tool\n", "response = model_with_tools.invoke(\"What is the capital of France?\")\n", "\n", "print(f\"πŸ”§ Tool calls: {response.tool_calls}\")\n", "print(f\"πŸ’¬ Direct answer: {response.content}\")\n", "print(\"\\nβœ… The model is smart enough to know it doesn't need a tool here!\")\n" ] }, { "cell_type": "markdown", "id": "51538937", "metadata": {}, "source": [ "---\n", "\n", "# πŸ”„ Part 3: The ReAct Pattern - Reason + Act\n", "\n", "The **ReAct** (Reasoning + Acting) pattern is the most common way to build agents. The idea is simple:\n", "\n", "```\n", "β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”\n", "β”‚ ReAct Loop β”‚\n", "β”‚ β”‚\n", "β”‚ 1. 🧠 REASON: Analyze the question β”‚\n", "β”‚ ↓ β”‚\n", "β”‚ 2. πŸ”§ ACT: Choose & call a tool β”‚\n", "β”‚ ↓ β”‚\n", "β”‚ 3. πŸ‘€ OBSERVE: Get the tool result β”‚\n", "β”‚ ↓ β”‚\n", "β”‚ 4. πŸ”„ REPEAT or βœ… ANSWER β”‚\n", "β”‚ β”‚\n", "β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜\n", "```\n", "\n", "The agent keeps looping until it has enough information to give a final answer.\n", "\n", "### 🎯 Key Insight\n", "This is essentially what we did manually in the tool calling section β€” but automated in a **while loop**!" ] }, { "cell_type": "markdown", "id": "c8265450", "metadata": {}, "source": [ "### πŸ—οΈ Building a ReAct Agent from Scratch\n", "\n", "Let's build our own ReAct agent in vanilla Python!\n", "This is the same pattern used by LangGraph, CrewAI, and other frameworks under the hood." ] }, { "cell_type": "code", "execution_count": 219, "id": "e732c547", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… ReActAgent class created!\n" ] } ], "source": [ "from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage\n", "from langchain_core.tools import BaseTool\n", "\n", "\n", "class ReActAgent:\n", " \"\"\"A ReAct agent built from scratch.\n", "\n", " Implements the Reason-Act-Observe loop with tool calling.\n", " \"\"\"\n", "\n", " def __init__(\n", " self,\n", " tools: list[BaseTool],\n", " system_prompt: str = \"You are a helpful assistant with access to tools. Use them when needed.\",\n", " max_iterations: int = 5,\n", " ) -> None:\n", " \"\"\"Initialize the agent with a model, tools, system prompt, and max iterations.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.tools = {t.name: t for t in tools} # Map tool name β†’ tool object\n", " self.model_with_tools = self.model.bind_tools(tools)\n", " self.system_prompt = system_prompt\n", " self.max_iterations = max_iterations\n", " self.conversation: list[BaseMessage] = []\n", "\n", " def _execute_tools(self, ai_message: AIMessage) -> list[ToolMessage]:\n", " \"\"\"Execute all tool calls from the AI message.\"\"\"\n", " tool_messages = []\n", " for tool_call in ai_message.tool_calls:\n", " tool_name = tool_call[\"name\"]\n", " tool_args = tool_call[\"args\"]\n", "\n", " print(f\" πŸ”§ Calling tool: {tool_name}({tool_args})\")\n", "\n", " # Execute the tool\n", " tool = self.tools[tool_name]\n", " result = tool.invoke(tool_args)\n", "\n", " print(f\" πŸ“‹ Result: {result}\")\n", "\n", " tool_messages.append(\n", " ToolMessage(content=str(result), tool_call_id=tool_call[\"id\"]),\n", " )\n", " return tool_messages\n", "\n", " def invoke(self, user_message: str) -> str:\n", " \"\"\"Run the ReAct loop until the agent produces a final answer.\"\"\"\n", " # Initialize conversation\n", " self.conversation = [\n", " SystemMessage(content=self.system_prompt),\n", " HumanMessage(content=user_message),\n", " ]\n", "\n", " print(f\"πŸ‘€ User: {user_message}\\n\")\n", "\n", " for i in range(self.max_iterations):\n", " print(f\"πŸ”„ Iteration {i + 1}:\")\n", "\n", " # REASON: Call the LLM\n", " response = self.model_with_tools.invoke(self.conversation)\n", " self.conversation.append(response)\n", "\n", " # Check if the model wants to use tools\n", " if response.tool_calls:\n", " # ACT: Execute tools\n", " tool_messages = self._execute_tools(response)\n", "\n", " # OBSERVE: Add tool results to conversation\n", " self.conversation.extend(tool_messages)\n", " print()\n", " continue # Loop back for more reasoning\n", "\n", " # No tool calls β†’ final answer!\n", " print(f\"\\nβœ… Final answer: {response.content}\")\n", " return response.content\n", "\n", " return \"⚠️ Max iterations reached without a final answer.\"\n", "\n", "\n", "print(\"βœ… ReActAgent class created!\")\n" ] }, { "cell_type": "code", "execution_count": 220, "id": "a75b43c0", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ User: What is 1234 multiplied by 5678?\n", "\n", "πŸ”„ Iteration 1:\n", " πŸ”§ Calling tool: calculator({'operation': 'multiply', 'x': 1234, 'y': 5678})\n", " πŸ“‹ Result: 1234.0 multiply 5678.0 = 7006652.0\n", "\n", "πŸ”„ Iteration 2:\n", "\n", "βœ… Final answer: The result of 1234 multiplied by 5678 is 7,006,652.\n" ] } ], "source": [ "# Create our agent with both tools\n", "agent = ReActAgent(tools=[calculator, get_weather])\n", "\n", "# Test 1: Math question β†’ should use calculator\n", "result = agent.invoke(\"What is 1234 multiplied by 5678?\")\n" ] }, { "cell_type": "code", "execution_count": 221, "id": "c2ddef7e", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ User: What's the weather in Tokyo?\n", "\n", "πŸ”„ Iteration 1:\n", " πŸ”§ Calling tool: get_weather({'city': 'Tokyo'})\n", " πŸ“‹ Result: Weather in Tokyo: 🌀️ 25Β°C, Clear skies\n", "\n", "πŸ”„ Iteration 2:\n", "\n", "βœ… Final answer: The weather in Tokyo is currently clear with a temperature of 25Β°C.\n" ] } ], "source": [ "# Test 2: Weather question β†’ should use get_weather\n", "result = agent.invoke(\"What's the weather in Tokyo?\")\n" ] }, { "cell_type": "code", "execution_count": 222, "id": "10dc7c90", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ User: Tell me a fun fact about cats.\n", "\n", "πŸ”„ Iteration 1:\n", "\n", "βœ… Final answer: Did you know that cats can make over 100 different sounds, while dogs can only make around 10? This is one of the many fascinating facts about our feline friends.\n" ] } ], "source": [ "# Test 3: No tool needed\n", "result = agent.invoke(\"Tell me a fun fact about cats.\")\n" ] }, { "cell_type": "code", "execution_count": 223, "id": "bd83d429", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ User: What's the weather in Paris and London? Also, what is 42 divided by 7?\n", "\n", "πŸ”„ Iteration 1:\n", " πŸ”§ Calling tool: get_weather({'city': 'Paris'})\n", " πŸ“‹ Result: Weather in Paris: β˜€οΈ 22Β°C, Sunny with light clouds\n", " πŸ”§ Calling tool: get_weather({'city': 'London'})\n", " πŸ“‹ Result: Weather in London: 🌧️ 15Β°C, Rainy\n", " πŸ”§ Calling tool: calculator({'operation': 'divide', 'x': 42, 'y': 7})\n", " πŸ“‹ Result: 42.0 divide 7.0 = 6.0\n", "\n", "πŸ”„ Iteration 2:\n", "\n", "βœ… Final answer: The weather in Paris is β˜€οΈ 22Β°C, Sunny with light clouds.\n", "\n", "The weather in London is 🌧️ 15Β°C, Rainy.\n", "\n", "42 divided by 7 is 6.\n" ] } ], "source": [ "# Test 4: Multi-step reasoning β€” requires multiple tool calls\n", "result = agent.invoke(\n", " \"What's the weather in Paris and London? Also, what is 42 divided by 7?\",\n", ")\n" ] }, { "cell_type": "markdown", "id": "763dffcc", "metadata": {}, "source": [ "### πŸŽ‰ What Did We Just Build?\n", "\n", "We built a **ReAct agent** from scratch! This is the same pattern that powers:\n", "- LangGraph's `create_react_agent()`\n", "- OpenAI's function calling agents\n", "- Most modern agent frameworks\n", "\n", "The core idea is just a **while loop** + **tool calling**!" ] }, { "cell_type": "markdown", "id": "92ef9729", "metadata": {}, "source": [ "---\n", "\n", "# πŸ—οΈ Part 4: Using LangGraph for Agents\n", "\n", "While building from scratch is educational, frameworks like **LangGraph** make it much easier to build production-ready agents.\n", "\n", "LangGraph provides:\n", "\n", "| Feature | Benefit |\n", "|---------|--------|\n", "| `create_react_agent()` | One-line agent creation |\n", "| State management | Built-in conversation memory |\n", "| Graph-based flow | Visual, debuggable workflows |\n", "| Streaming | Real-time response streaming |\n", "\n", "Let's recreate our agent in just a few lines!" ] }, { "cell_type": "code", "execution_count": 224, "id": "2b317a98", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… LangGraph agent created in one line!\n" ] } ], "source": [ "from langchain.agents import create_agent\n", "\n", "# Create a ReAct agent in ONE line!\n", "langgraph_agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=0),\n", " tools=[calculator, get_weather],\n", ")\n", "\n", "print(\"βœ… LangGraph agent created in one line!\")\n" ] }, { "cell_type": "code", "execution_count": 225, "id": "520b7e9f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“œ Full conversation trace:\n", "\n", "πŸ‘€ HumanMessage: What is 99 multiplied by 88?\n", "πŸ€– AIMessage: [tool call β†’ calculator({'operation': 'multiply', 'x': 99, 'y': 88})]\n", "πŸ€– ToolMessage: 99.0 multiply 88.0 = 8712.0\n", "πŸ€– AIMessage: The result of 99 multiplied by 88 is 8712.\n" ] } ], "source": [ "# Invoke the LangGraph agent\n", "response = langgraph_agent.invoke(\n", " {\"messages\": [(\"user\", \"What is 99 multiplied by 88?\")]},\n", ")\n", "\n", "# The response contains the full message history\n", "print(\"πŸ“œ Full conversation trace:\\n\")\n", "for msg in response[\"messages\"]:\n", " role = msg.__class__.__name__\n", " if hasattr(msg, \"tool_calls\") and msg.tool_calls:\n", " print(f\"πŸ€– {role}: [tool call β†’ {msg.tool_calls[0]['name']}({msg.tool_calls[0]['args']})]\")\n", " elif hasattr(msg, \"content\") and msg.content:\n", " print(f\"{'πŸ‘€' if role == 'HumanMessage' else 'πŸ€–'} {role}: {msg.content[:200]}\")\n", " else:\n", " print(f\"πŸ“‹ {role}: {msg}\")\n" ] }, { "cell_type": "code", "execution_count": 226, "id": "fa0d155c", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ’¬ Final answer: The weather in Paris is β˜€οΈ 22Β°C, Sunny with light clouds.\n", "\n", "The weather in New York is β›… 28Β°C, Partly cloudy.\n" ] } ], "source": [ "# Test with weather question\n", "response = langgraph_agent.invoke(\n", " {\"messages\": [(\"user\", \"What's the weather in Paris and New York?\")]},\n", ")\n", "\n", "# Print just the final answer\n", "final_message = response[\"messages\"][-1]\n", "print(f\"πŸ’¬ Final answer: {final_message.content}\")\n" ] }, { "cell_type": "markdown", "id": "d33bc6c1", "metadata": {}, "source": [ "### πŸ” LangGraph with Streaming\n", "\n", "One of the big advantages of LangGraph is built-in streaming support. You can see the agent's reasoning in real time:" ] }, { "cell_type": "code", "execution_count": 227, "id": "fe268975", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "🌊 Streaming agent execution:\n", "\n", "πŸ“ Node: model\n", " πŸ”§ Tool call: calculator({'operation': 'multiply', 'x': 15, 'y': 23})\n", " πŸ”§ Tool call: get_weather({'city': 'London'})\n", "\n", "πŸ“ Node: tools\n", " πŸ’¬ Weather in London: 🌧️ 15Β°C, Rainy\n", "\n", "πŸ“ Node: tools\n", " πŸ’¬ 15.0 multiply 23.0 = 345.0\n", "\n", "πŸ“ Node: model\n", " πŸ’¬ The result of 15 times 23 is 345. The weather in London is currently rainy with a temperature of 15Β°C.\n", "\n" ] } ], "source": [ "# Stream the agent's execution step by step\n", "print(\"🌊 Streaming agent execution:\\n\")\n", "\n", "for step in langgraph_agent.stream(\n", " {\"messages\": [(\"user\", \"What is 15 times 23, and what's the weather in London?\")]},\n", "):\n", " # Each step is a dict with the node name as key\n", " for node_name, node_output in step.items():\n", " print(f\"πŸ“ Node: {node_name}\")\n", " for msg in node_output.get(\"messages\", []):\n", " if hasattr(msg, \"tool_calls\") and msg.tool_calls:\n", " for tc in msg.tool_calls:\n", " print(f\" πŸ”§ Tool call: {tc['name']}({tc['args']})\")\n", " elif hasattr(msg, \"content\") and msg.content:\n", " content_preview = msg.content[:150]\n", " print(f\" πŸ’¬ {content_preview}\")\n", " print()\n" ] }, { "cell_type": "markdown", "id": "778a011d", "metadata": {}, "source": [ "---\n", "\n", "# 🧩 Part 5: Building a Multi-Tool Agent\n", "\n", "Let's build a more capable agent with multiple specialized tools. This is closer to what you'd build in production!\n", "\n", "We'll add:\n", "- πŸ” A **search** tool (simulated)\n", "- πŸ“ A **note-taking** tool\n", "- πŸ• A **time** tool" ] }, { "cell_type": "code", "execution_count": 228, "id": "0b88c81c", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… All tools defined!\n", " πŸ“ calculator\n", " 🌀️ get_weather\n", " πŸ” search_knowledge_base\n", " πŸ“ save_note\n", " πŸ“‹ get_notes\n", " πŸ• get_current_time\n" ] } ], "source": [ "from datetime import UTC, datetime\n", "\n", "notes_store: dict[str, str] = {}\n", "\n", "\n", "@tool\n", "def search_knowledge_base(query: str) -> str:\n", " \"\"\"Search a knowledge base for information about programming and technology.\n", "\n", " Args:\n", " query: The search query\n", "\n", " \"\"\"\n", " knowledge = {\n", " \"python\": \"Python is a high-level programming language created by Guido van Rossum in 1991. It emphasizes code readability and supports multiple paradigms.\",\n", " \"langchain\": \"LangChain is a framework for developing applications powered by language models. It provides tools for chains, agents, and retrieval.\",\n", " \"mistral\": \"Mistral AI is a French AI company founded in 2023. They develop open-weight LLMs including Mistral 7B, Mixtral, and Mistral Large.\",\n", " \"rag\": \"RAG (Retrieval Augmented Generation) is a technique that enhances LLMs by retrieving relevant documents before generating answers.\",\n", " \"transformer\": \"Transformers are neural network architectures introduced in the 'Attention is All You Need' paper (2017). They use self-attention mechanisms.\",\n", " }\n", " query_lower = query.lower()\n", " for key, value in knowledge.items():\n", " if key in query_lower:\n", " return value\n", " return f\"No specific results found for '{query}'. Try searching for: python, langchain, mistral, rag, or transformer.\"\n", "\n", "\n", "@tool\n", "def save_note(title: str, content: str) -> str:\n", " \"\"\"Save a note with a title and content for later retrieval.\n", "\n", " Args:\n", " title: The title of the note\n", " content: The content of the note\n", "\n", " \"\"\"\n", " notes_store[title] = content\n", " return f\"βœ… Note '{title}' saved successfully!\"\n", "\n", "\n", "@tool\n", "def get_notes() -> str:\n", " \"\"\"Retrieve all saved notes.\"\"\"\n", " if not notes_store:\n", " return \"πŸ“­ No notes saved yet.\"\n", " result = \"πŸ“ Saved notes:\\n\"\n", " for title, content in notes_store.items():\n", " result += f\" - {title}: {content}\\n\"\n", " return result\n", "\n", "\n", "@tool\n", "def get_current_time() -> str:\n", " \"\"\"Get the current date and time.\"\"\"\n", " now = datetime.now(UTC)\n", " return f\"πŸ• Current UTC time: {now.strftime('%Y-%m-%d %H:%M:%S %Z')}\"\n", "\n", "\n", "print(\"βœ… All tools defined!\")\n", "print(\" πŸ“ calculator\")\n", "print(\" 🌀️ get_weather\")\n", "print(\" πŸ” search_knowledge_base\")\n", "print(\" πŸ“ save_note\")\n", "print(\" πŸ“‹ get_notes\")\n", "print(\" πŸ• get_current_time\")\n" ] }, { "cell_type": "code", "execution_count": 229, "id": "d1d4ec57", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… Multi-tool agent ready with 6 tools!\n" ] } ], "source": [ "# Create a powerful multi-tool agent\n", "all_tools = [calculator, get_weather, search_knowledge_base, save_note, get_notes, get_current_time]\n", "\n", "multi_agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=0),\n", " tools=all_tools,\n", " system_prompt=\"You are a helpful research assistant. You have access to multiple tools including a calculator, weather service, knowledge base search, note-taking, and a clock. Use the appropriate tools to help the user. Always be thorough and use multiple tools if needed.\",\n", ")\n", "\n", "print(\"βœ… Multi-tool agent ready with 6 tools!\")\n" ] }, { "cell_type": "code", "execution_count": 230, "id": "5bc24dc9", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“œ Agent trace:\n", "\n", " πŸ‘€ User: Search for information about Mistral AI and save it as a note titled 'Mistral Info'. Then tell me the current time.\n", " πŸ”§ Tool: search_knowledge_base({'query': 'Mistral AI'})\n", " πŸ”§ Tool: get_current_time({})\n", " πŸ“‹ Result: Mistral AI is a French AI company founded in 2023. They develop open-weight LLMs including Mistral 7...\n", " πŸ“‹ Result: πŸ• Current UTC time: 2026-03-04 17:18:07 UTC...\n", " πŸ”§ Tool: save_note({'title': 'Mistral Info', 'content': 'Mistral AI is a French AI company founded in 2023. They develop open-weight LLMs including Mistral 7B, Mixtral, and Mistral Large.'})\n", " πŸ“‹ Result: βœ… Note 'Mistral Info' saved successfully!...\n", "\n", "πŸ’¬ Final: I have saved the information about Mistral AI as a note titled 'Mistral Info'. The current time is 2026-03-04 17:18:07 UTC.\n" ] } ], "source": [ "# Test: Complex multi-tool query\n", "response = multi_agent.invoke(\n", " {\"messages\": [(\"user\", \"Search for information about Mistral AI and save it as a note titled 'Mistral Info'. Then tell me the current time.\")]},\n", ")\n", "\n", "# Print the conversation trace\n", "print(\"πŸ“œ Agent trace:\\n\")\n", "for msg in response[\"messages\"]:\n", " role = msg.__class__.__name__\n", " if hasattr(msg, \"tool_calls\") and msg.tool_calls:\n", " for tc in msg.tool_calls:\n", " print(f\" πŸ”§ Tool: {tc['name']}({tc['args']})\")\n", " elif role == \"ToolMessage\":\n", " print(f\" πŸ“‹ Result: {msg.content[:100]}...\")\n", " elif role == \"HumanMessage\":\n", " print(f\" πŸ‘€ User: {msg.content}\")\n", " elif msg.content:\n", " print(f\"\\nπŸ’¬ Final: {msg.content}\")\n" ] }, { "cell_type": "code", "execution_count": 231, "id": "30d41686", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“ Notes in store:\n", " Mistral Info: Mistral AI is a French AI company founded in 2023. They develop open-weight LLMs including Mistral 7B, Mixtral, and Mistral Large.\n" ] } ], "source": [ "# Verify the note was saved\n", "print(\"πŸ“ Notes in store:\")\n", "for title, content in notes_store.items():\n", " print(f\" {title}: {content}\")\n" ] }, { "cell_type": "markdown", "id": "183d7992", "metadata": {}, "source": [ "---\n", "\n", "# 🎯 Part 6: Structured Output with Tool Calling\n", "\n", "A powerful pattern is using tool calling to get **structured output** from the LLM. Instead of parsing free-text, we force the model to return data in a specific Pydantic schema.\n", "\n", "```\n", "Unstructured: Structured:\n", "\"The sentiment is positive {\"sentiment\": \"positive\",\n", " with confidence around 90%\" \"confidence\": 0.9,\n", " \"keywords\": [\"great\", \"love\"]}\n", "```" ] }, { "cell_type": "code", "execution_count": 232, "id": "5cd4545d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“Š Sentiment Analysis Result:\n", " Sentiment: positive\n", " Confidence: 0.95\n", " Keywords: ['love', 'easier', 'fantastic']\n", " Summary: The text expresses strong positive sentiment about a new Python library, highlighting ease of use and excellent documentation.\n", "\n", " Type: SentimentAnalysis ← It's a proper Pydantic object!\n" ] } ], "source": [ "from pydantic import BaseModel, Field\n", "\n", "\n", "class SentimentAnalysis(BaseModel):\n", " \"\"\"Analyze the sentiment of a piece of text.\"\"\"\n", "\n", " sentiment: str = Field(description=\"The sentiment: positive, negative, or neutral\")\n", " confidence: float = Field(description=\"Confidence score between 0 and 1\")\n", " keywords: list[str] = Field(description=\"Key words that influenced the sentiment\")\n", " summary: str = Field(description=\"One-sentence summary of the analysis\")\n", "\n", "\n", "def to_sentiment_result(raw: object) -> SentimentAnalysis:\n", " \"\"\"Normalize structured output into a SentimentAnalysis model.\"\"\"\n", " if isinstance(raw, SentimentAnalysis):\n", " return raw\n", " if isinstance(raw, BaseModel):\n", " return SentimentAnalysis.model_validate(raw.model_dump())\n", " if isinstance(raw, dict):\n", " return SentimentAnalysis.model_validate(raw)\n", " return SentimentAnalysis.model_validate_json(str(raw))\n", "\n", "\n", "structured_model = model.with_structured_output(SentimentAnalysis)\n", "raw_result = structured_model.invoke(\n", " \"I absolutely love this new Python library! It makes everything so much easier and the documentation is fantastic.\",\n", ")\n", "result = to_sentiment_result(raw_result)\n", "\n", "print(\"πŸ“Š Sentiment Analysis Result:\")\n", "print(f\" Sentiment: {result.sentiment}\")\n", "print(f\" Confidence: {result.confidence}\")\n", "print(f\" Keywords: {result.keywords}\")\n", "print(f\" Summary: {result.summary}\")\n", "print(f\"\\n Type: {type(result).__name__} ← It's a proper Pydantic object!\")\n" ] }, { "cell_type": "code", "execution_count": 233, "id": "7f842ae2", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ‘€ Extracted Person Info:\n", " Name: Marie\n", " Age: 28\n", " Occupation: Machine Learning Engineer\n", " Location: Lyon\n", " Interests: ['hiking', 'reading science fiction novels', 'contributing to open source projects']\n" ] } ], "source": [ "class PersonInfo(BaseModel):\n", " \"\"\"Extract personal information from text.\"\"\"\n", "\n", " name: str = Field(description=\"The person's full name\")\n", " age: int | None = Field(description=\"The person's age if mentioned\")\n", " occupation: str | None = Field(description=\"The person's job or occupation if mentioned\")\n", " location: str | None = Field(description=\"Where the person lives if mentioned\")\n", " interests: list[str] = Field(description=\"The person's interests or hobbies\")\n", "\n", "\n", "def to_person_info(raw: object) -> PersonInfo:\n", " \"\"\"Normalize structured output into a PersonInfo model.\"\"\"\n", " if isinstance(raw, PersonInfo):\n", " return raw\n", " if isinstance(raw, BaseModel):\n", " return PersonInfo.model_validate(raw.model_dump())\n", " if isinstance(raw, dict):\n", " return PersonInfo.model_validate(raw)\n", " return PersonInfo.model_validate_json(str(raw))\n", "\n", "\n", "person_extractor = model.with_structured_output(PersonInfo)\n", "raw_result = person_extractor.invoke(\n", " \"Marie is a 28-year-old machine learning engineer living in Lyon. \"\n", " \"She enjoys hiking, reading science fiction novels, and contributing to open source projects.\",\n", ")\n", "result = to_person_info(raw_result)\n", "\n", "print(\"πŸ‘€ Extracted Person Info:\")\n", "print(f\" Name: {result.name}\")\n", "print(f\" Age: {result.age}\")\n", "print(f\" Occupation: {result.occupation}\")\n", "print(f\" Location: {result.location}\")\n", "print(f\" Interests: {result.interests}\")\n" ] }, { "cell_type": "markdown", "id": "8412b02a", "metadata": {}, "source": [ "---\n", "\n", "# 🧠 Part 7: Putting It All Together β€” A Complete Agent\n", "\n", "Let's build a complete agent that combines **memory**, **tools**, and **structured output**. This agent acts as a personal research assistant.\n", "\n", "```\n", "β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”\n", "β”‚ Complete Research Agent β”‚\n", "β”‚ β”‚\n", "β”‚ πŸ“ System Prompt (personality & rules) β”‚\n", "β”‚ 🧠 Memory (conversation history) β”‚\n", "β”‚ πŸ”§ Tools: β”‚\n", "β”‚ β”œβ”€β”€ πŸ“ Calculator β”‚\n", "β”‚ β”œβ”€β”€ 🌀️ Weather β”‚\n", "β”‚ β”œβ”€β”€ πŸ” Knowledge Base β”‚\n", "β”‚ β”œβ”€β”€ πŸ“ Note Taking β”‚\n", "β”‚ └── πŸ• Clock β”‚\n", "β”‚ πŸ”„ ReAct Loop (reason β†’ act β†’ observe β†’ repeat) β”‚\n", "β”‚ β”‚\n", "β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜\n", "```" ] }, { "cell_type": "code", "execution_count": 234, "id": "5f125a1e", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "βœ… ResearchAssistant ready!\n" ] } ], "source": [ "\n", "from langchain_core.messages import HumanMessage\n", "\n", "\n", "class ResearchAssistant:\n", " \"\"\"A complete research assistant agent with memory, tools, and conversation tracking.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the research assistant with a model, tools, system prompt, and empty conversation history.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.tools = [calculator, get_weather, search_knowledge_base, save_note, get_notes, get_current_time]\n", "\n", " system_prompt = (\n", " \"You are a research assistant powered by Mistral AI. \"\n", " \"You can search for information, do calculations, check the weather, \"\n", " \"and take notes. Always be thorough β€” use tools when they can help. \"\n", " \"When you search for information, save important findings as notes for future reference.\"\n", " )\n", "\n", " self.agent = create_agent(\n", " model=self.model,\n", " tools=self.tools,\n", " system_prompt=system_prompt,\n", " )\n", "\n", " # Conversation memory\n", " self.messages: list = []\n", "\n", " def chat(self, user_message: str) -> str:\n", " \"\"\"Send a message to the agent and get a response.\"\"\"\n", " self.messages.append((\"user\", user_message))\n", "\n", " response = self.agent.invoke({\"messages\": self.messages})\n", "\n", " # Extract the final AI message\n", " final_msg = response[\"messages\"][-1].content\n", " self.messages.append((\"assistant\", final_msg))\n", "\n", " return final_msg\n", "\n", " def get_conversation_length(self) -> int:\n", " \"\"\"Get the number of messages in the conversation history.\"\"\"\n", " return len(self.messages)\n", "\n", "\n", "print(\"βœ… ResearchAssistant ready!\")\n" ] }, { "cell_type": "code", "execution_count": 236, "id": "aaeb08e8", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "======================================================================\n", "πŸ‘€ User: Hello! Can you search for information about RAG and save it as a note?\n", "πŸ€– Agent: I've saved a note about RAG.\n", "\n", "======================================================================\n", "πŸ‘€ User: Great! Now what's 256 divided by 16?\n", "πŸ€– Agent: The result of 256 divided by 16 is 16.\n", "\n", "======================================================================\n", "πŸ‘€ User: What's the weather in Paris right now?\n", "πŸ€– Agent: The current weather in Paris is sunny with light clouds and a temperature of 22Β°C.\n", "\n", "======================================================================\n", "πŸ‘€ User: Can you show me all the notes we've saved so far?\n", "πŸ€– Agent: Here are the notes we've saved so far:\n", "\n", "1. Title: RAG\n", "Content: RAG (Red-Ambler-Green) is a project management and performance reporting tool that provides a visual indication of the status of tasks or projects using three colors: Red (R), Amber (A), and Green (G). It helps teams and managers quickly understand the current state of various projects or tasks. Red indicates a critical issue or significant delay, Amber signifies a warning or slight delay, and Green shows that everything is on track and progressing as planned. RAG status is often used in project management, risk assessment, and performance reporting to facilitate decision-making and communication.\n", "\n", "2. Title: Weather in Paris\n", "Content: The current weather in Paris is sunny with light clouds and a temperature of 22Β°C.\n", "\n", "πŸ“Š Conversation length: 8 messages\n" ] } ], "source": [ "# Create the assistant\n", "assistant = ResearchAssistant()\n", "\n", "# Multi-turn conversation with the agent\n", "queries = [\n", " \"Hello! Can you search for information about RAG and save it as a note?\",\n", " \"Great! Now what's 256 divided by 16?\",\n", " \"What's the weather in Paris right now?\",\n", " \"Can you show me all the notes we've saved so far?\",\n", "]\n", "\n", "for query in queries:\n", " print(f\"\\n{'='*70}\")\n", " print(f\"πŸ‘€ User: {query}\")\n", " response = assistant.chat(query)\n", " print(f\"πŸ€– Agent: {response}\")\n", "\n", "print(f\"\\nπŸ“Š Conversation length: {assistant.get_conversation_length()} messages\")\n" ] }, { "cell_type": "markdown", "id": "e29ccc60", "metadata": {}, "source": [ "---\n", "\n", "## πŸ“ Summary\n", "\n", "Congratulations! πŸŽ‰ You've learned to build AI agents from scratch!\n", "\n", "βœ… **Memory** β€” Maintaining conversation history across turns \n", "βœ… **Tool Calling** β€” Letting LLMs use functions to take actions \n", "βœ… **ReAct Pattern** β€” The reason-act-observe loop \n", "βœ… **LangGraph** β€” Building agents with frameworks \n", "βœ… **Multi-Tool Agents** β€” Combining multiple capabilities \n", "βœ… **Structured Output** β€” Getting typed data from LLMs \n", "\n", "### πŸ”‘ Key Takeaways:\n", "\n", "| Concept | One-liner |\n", "|---------|----------|\n", "| **Agent** | An LLM that can reason AND act |\n", "| **Memory** | Pass conversation history on each call |\n", "| **Tools** | Functions the LLM can decide to call |\n", "| **ReAct** | While loop: reason β†’ use tool β†’ observe β†’ repeat |\n", "| **LangGraph** | Framework that handles the loop for you |\n", "| **Structured Output** | Force LLM to return Pydantic models |\n", "\n", "### πŸš€ Next Steps:\n", "\n", "- Add **real API tools** (web search, database queries)\n", "- Explore **multi-agent systems** (agents that coordinate with each other)\n", "- Learn about **Model Context Protocol (MCP)** for standardized tool interfaces\n", "- Build **RAG agents** that retrieve from real vector databases\n", "\n", "### πŸ“š Resources:\n", "\n", "- [LangGraph Documentation](https://langchain-ai.github.io/langgraph/)\n", "- [Mistral AI Function Calling](https://docs.mistral.ai/capabilities/function_calling/)\n", "- [LangChain Agents](https://python.langchain.com/docs/concepts/agents/)\n", "- [ReAct Paper](https://arxiv.org/abs/2210.03629)\n", "\n", "---\n", "\n", "**Happy agent building!** πŸš€βœ¨" ] }, { "cell_type": "markdown", "id": "b4888037", "metadata": {}, "source": [ "---\n", "\n", "## πŸ‹οΈ Exercises\n", "\n", "Time to practice! The exercises below go from fundamental memory concepts to advanced agent patterns. Each exercise builds on what you learned in this notebook." ] }, { "cell_type": "markdown", "id": "8300e042", "metadata": {}, "source": [ "### πŸ“š Exercise 1: Summarized Memory ChatBot\n", "\n", "Instead of a sliding window that **drops** old messages, implement a chatbot that **summarizes** them. When the history exceeds `max_messages`, use the LLM itself to create a summary, then continue the conversation with the summary as context.\n", "\n", "**Requirements:**\n", "- Implement `_summarize_history()` using the LLM to condense old messages\n", "- Replace old messages with a `SystemMessage` containing the summary\n", "- Keep the most recent messages intact\n", "- The bot should still remember key facts after summarization" ] }, { "cell_type": "code", "execution_count": 237, "id": "49b3e472", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "πŸ‘€ My name is Alice and I love machine learning.\n", "πŸ€– Hello Alice! It's great to meet you. With your interest in machine learning, I'm sure you're always looking for new ways to learn and apply your skills. How can I assist you today? Are you looking for resources, project ideas, or help with a specific concept in machine learning?\n", " πŸ“Š History: 2 messages | Summary: No\n", "\n", "πŸ‘€ I work at Google as a research scientist.\n", "πŸ€– That's impressive, Alice! Working as a research scientist at Google in the field of machine learning must be incredibly exciting and rewarding. With your expertise, I'm sure you're involved in cutting-edge projects and research. How can I assist you today? Are you looking for information on a specific topic, collaboration ideas, or perhaps help with a technical challenge you're facing? I'm here to help!\n", " πŸ“Š History: 4 messages | Summary: No\n", "\n", "πŸ‘€ I recently published a paper on transformers.\n", "πŸ€– That's great to hear! Congratulations on your publication. Transformers are a fascinating area of research in machine learning. How can I assist you further with your work on transformers or any other aspect of your research?\n", " πŸ“Š History: 2 messages | Summary: Yes\n", "\n", "πŸ‘€ I also enjoy hiking on weekends.\n", "πŸ€– That's wonderful! Hiking is a great way to unwind and connect with nature. If you ever need recommendations for hiking trails or tips on gear, I'd be happy to help. As for your work on transformers, is there any specific aspect you'd like to discuss or explore further? Perhaps you're looking for new research directions, or you might be interested in practical applications of transformers in your projects at Google.\n", " πŸ“Š History: 4 messages | Summary: Yes\n", "\n", "πŸ‘€ What do you know about me so far?\n", "πŸ€– Based on our previous conversation, here's what I know about you:\n", "\n", "1. **Professional Background:**\n", " - You recently published a paper on transformers.\n", " - You work at Google.\n", "\n", "2. **Interests:**\n", " - You're interested in research on transformers, including potential new research directions and practical applications.\n", " - You enjoy hiking on weekends and are interested in hiking trail recommendations and gear tips.\n", "\n", "3. **Assistance Offered:**\n", " - I can help with transformer-related research or applications.\n", " - I can provide hiking trail recommendations and gear tips.\n", " πŸ“Š History: 2 messages | Summary: Yes\n", "\n", "πŸ‘€ What was my paper about?\n", "πŸ€– I'm afraid I don't have access to specific details about your recently published paper on transformers, as our conversation didn't include that information. If you'd like to share more about it, I'd be happy to discuss it with you or help with any related questions you might have!\n", " πŸ“Š History: 4 messages | Summary: Yes\n" ] } ], "source": [ "class SummarizedMemoryChatBot:\n", " \"\"\"A chatbot that summarizes old messages instead of dropping them.\n", "\n", " When the history exceeds max_messages, it:\n", " 1. Sends old messages to the LLM with a summarization prompt\n", " 2. Replaces them with a summary context\n", " 3. Keeps recent messages intact\n", " \"\"\"\n", "\n", " def __init__(self, system_prompt: str = \"You are a helpful assistant.\", max_messages: int = 6) -> None:\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.system_prompt = system_prompt\n", " self.history: list[BaseMessage] = []\n", " self.max_messages = max_messages\n", " self.summary: str = \"\"\n", "\n", " @staticmethod\n", " def _to_text(content: str | list[str | dict]) -> str:\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", "\n", " def _summarize_history(self) -> str:\n", " \"\"\"Summarize older conversation turns into a concise memory.\"\"\"\n", " if not self.history:\n", " return \"\"\n", "\n", " conversation_text = \"\\n\".join(\n", " f\"{'User' if isinstance(m, HumanMessage) else 'Assistant'}: {self._to_text(m.content)}\"\n", " for m in self.history\n", " if isinstance(m, (HumanMessage, AIMessage))\n", " )\n", "\n", " summary_messages = [\n", " SystemMessage(\n", " content=(\n", " \"You summarize conversations. Keep only key facts, preferences, goals, \"\n", " \"and commitments that the assistant should remember.\"\n", " ),\n", " ),\n", " HumanMessage(content=conversation_text),\n", " ]\n", " summary_response = self.model.invoke(summary_messages)\n", " return self._to_text(summary_response.content)\n", "\n", " def chat(self, user_message: str) -> str:\n", " \"\"\"Chat with memory. When history gets too long, summarize older messages.\"\"\"\n", " if len(self.history) >= self.max_messages:\n", " self.summary = self._summarize_history()\n", " self.history = []\n", "\n", " full_messages: list[BaseMessage] = [SystemMessage(content=self.system_prompt)]\n", " if self.summary:\n", " full_messages.append(\n", " SystemMessage(\n", " content=f\"Summary of previous conversation:\\n{self.summary}\",\n", " ),\n", " )\n", " full_messages.extend(self.history)\n", " full_messages.append(HumanMessage(content=user_message))\n", "\n", " response = self.model.invoke(full_messages)\n", " response_text = self._to_text(response.content)\n", "\n", " self.history.append(HumanMessage(content=user_message))\n", " self.history.append(AIMessage(content=response_text))\n", " return response_text\n", "\n", "\n", "bot = SummarizedMemoryChatBot(max_messages=4)\n", "\n", "test_messages = [\n", " \"My name is Alice and I love machine learning.\",\n", " \"I work at Google as a research scientist.\",\n", " \"I recently published a paper on transformers.\",\n", " \"I also enjoy hiking on weekends.\",\n", " \"What do you know about me so far?\",\n", " \"What was my paper about?\",\n", "]\n", "\n", "for msg in test_messages:\n", " print(f\"\\nπŸ‘€ {msg}\")\n", " response = bot.chat(msg)\n", " print(f\"πŸ€– {response}\")\n", " print(f\" πŸ“Š History: {len(bot.history)} messages | Summary: {'Yes' if bot.summary else 'No'}\")\n" ] }, { "cell_type": "markdown", "id": "5ffa780d", "metadata": {}, "source": [ "### πŸ”§ Exercise 2: Build Your Own Custom Tools\n", "\n", "Create 3 custom tools and wire them into a LangGraph agent:\n", "\n", "1. **`translate`** β€” Translates text to a target language (use the LLM itself as the translation engine)\n", "2. **`string_analyzer`** β€” Returns stats about a string (word count, character count, most common word)\n", "3. **`random_fact`** β€” Returns a random fun fact from a hardcoded list\n", "\n", "Then test your agent with:\n", "- \"Translate 'Hello, how are you?' to French\"\n", "- \"Analyze the string: 'The quick brown fox jumps over the lazy dog'\"\n", "- \"Tell me a random fact and then translate it to Spanish\"" ] }, { "cell_type": "code", "execution_count": 241, "id": "37821465", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "πŸ‘€ Translate 'Hello, how are you?' to French\n", "πŸ€– Bonjour, comment Γ§a va ?\n", "\n", "Alternatively, you could also say:\n", "- Salut, comment Γ§a va ? (more casual)\n", "- Bonjour, comment allez-vous ? (more formal)\n", "\n", "πŸ‘€ Analyze the string: 'The quick brown fox jumps over the lazy dog'\n", "πŸ€– The analysis of the string \"The quick brown fox jumps over the lazy dog\" is as follows:\n", "\n", "- **Word count:** 9\n", "- **Character count:** 43\n", "- **Unique words:** 9\n", "- **Most common word:** \"The\" (appears 1 time)\n", "\n", "πŸ‘€ Tell me a random fact and then translate it to Spanish\n", "πŸ€– Here is a random fact: \"Bananas are berries, but strawberries are not.\"\n", "\n", "And here is the translation to Spanish:\n", "\n", "\"Los plΓ‘tanos son bayas, pero las fresas no lo son.\"\n" ] } ], "source": [ "import secrets\n", "from collections import Counter\n", "\n", "from langchain_core.tools import tool\n", "\n", "\n", "def _content_to_text(content: str | list[str | dict]) -> str:\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", "\n", "\n", "@tool\n", "def translate(text: str, target_language: str) -> str:\n", " \"\"\"Translate text to a target language.\n", "\n", " Args:\n", " text: The text to translate\n", " target_language: The language to translate to (e.g., French, Spanish, German)\n", "\n", " \"\"\"\n", " prompt = f\"Translate the following text to {target_language}:\\n\\n{text}\"\n", " response = ChatMistralAI(model=\"mistral-small-latest\", temperature=0).invoke(prompt)\n", " return _content_to_text(response.content).strip()\n", "\n", "\n", "@tool\n", "def string_analyzer(text: str) -> str:\n", " \"\"\"Analyze a string and return statistics.\n", "\n", " Args:\n", " text: The text to analyze\n", "\n", " \"\"\"\n", " words = text.split()\n", " word_count = len(words)\n", " char_count = len(text)\n", " counts = Counter(words)\n", " most_common_word, occurrences = counts.most_common(1)[0] if counts else (\"\", 0)\n", " return (\n", " f\"String analysis:\\n\"\n", " f\" - Word count: {word_count}\\n\"\n", " f\" - Character count: {char_count}\\n\"\n", " f\" - Unique words: {len(counts)}\\n\"\n", " f\" - Most common word: '{most_common_word}' (appears {occurrences} times)\"\n", " )\n", "\n", "\n", "@tool\n", "def random_fact() -> str:\n", " \"\"\"Return a random fun fact.\"\"\"\n", " facts = [\n", " \"Honey never spoils. Archaeologists have found 3000-year-old honey in Egyptian tombs that was still edible.\",\n", " \"Octopuses have three hearts and blue blood.\",\n", " \"A group of flamingos is called a 'flamboyance'.\",\n", " \"The shortest war in history lasted 38 minutes between Britain and Zanzibar in 1896.\",\n", " \"Bananas are berries, but strawberries are not.\",\n", " ]\n", " return secrets.choice(facts)\n", "\n", "\n", "custom_agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=0),\n", " tools=[translate, string_analyzer, random_fact],\n", ")\n", "\n", "test_queries = [\n", " \"Translate 'Hello, how are you?' to French\",\n", " \"Analyze the string: 'The quick brown fox jumps over the lazy dog'\",\n", " \"Tell me a random fact and then translate it to Spanish\",\n", "]\n", "\n", "for query in test_queries:\n", " response = custom_agent.invoke({\"messages\": [(\"user\", query)]})\n", " final = _content_to_text(response[\"messages\"][-1].content)\n", " print(f\"\\nπŸ‘€ {query}\")\n", " print(f\"πŸ€– {final}\")\n", "\n" ] }, { "cell_type": "markdown", "id": "338cfebe", "metadata": {}, "source": [ "### πŸ”„ Exercise 3: Manual ReAct Loop with Trace Logging\n", "\n", "Extend the `ReActAgent` class from Part 3 to produce a detailed **trace log** of every step the agent takes. The trace should record each iteration, the tool chosen, the arguments, the result, and the final answer.\n", "\n", "**Requirements:**\n", "- Add a `trace: List[dict]` attribute to the agent\n", "- Each trace entry should have: `iteration`, `action` (tool name or \"final_answer\"), `input`, `output`\n", "- Add a `print_trace()` method that displays the trace in a formatted table\n", "- Test with a query that requires multiple tool calls" ] }, { "cell_type": "code", "execution_count": 242, "id": "0acc8be4", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "The result of 15 times 23 is 345.\n", "\n", "The weather in Tokyo is currently clear skies with a temperature of 25Β°C.\n", "Step | Action | Input | Output\n", "------------------------------------------------------------------------------------------------------------------------\n", "1 | calculator | {'operation': 'multiply', 'x': 15, 'y': | 15.0 multiply 23.0 = 345.0\n", "1 | get_weather | {'city': 'Tokyo'} | Weather in Tokyo: 🌀️ 25Β°C, Clear skies\n", "2 | final_answer | - | The result of 15 times 23 is 345.\n", "\n", "The weather in Tokyo is currently clear skies\n" ] } ], "source": [ "class TracedReActAgent:\n", " \"\"\"A ReAct agent that logs a detailed trace of every step.\"\"\"\n", "\n", " def __init__(\n", " self,\n", " tools: list[BaseTool],\n", " system_prompt: str = \"You are a helpful assistant with access to tools.\",\n", " max_iterations: int = 5,\n", " ) -> None:\n", " \"\"\"Initialize the agent with a model, tools, system prompt, max iterations, and empty conversation and trace.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.tools = {t.name: t for t in tools}\n", " self.model_with_tools = self.model.bind_tools(tools)\n", " self.system_prompt = system_prompt\n", " self.max_iterations = max_iterations\n", " self.conversation: list[BaseMessage] = []\n", " self.trace: list[dict] = []\n", "\n", " @staticmethod\n", " def _to_text(content: str | list[str | dict]) -> str:\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", "\n", " def _execute_tools(self, ai_message: AIMessage, iteration: int) -> list[ToolMessage]:\n", " \"\"\"Execute all tool calls from the AI message and log the trace.\"\"\"\n", " tool_messages: list[ToolMessage] = []\n", " for index, tool_call in enumerate(ai_message.tool_calls):\n", " tool_name = tool_call[\"name\"]\n", " tool_args = tool_call[\"args\"]\n", " tool = self.tools[tool_name]\n", " result = tool.invoke(tool_args)\n", " result_text = self._to_text(result)\n", "\n", " self.trace.append(\n", " {\n", " \"iteration\": iteration,\n", " \"action\": tool_name,\n", " \"input\": tool_args,\n", " \"output\": result_text,\n", " },\n", " )\n", "\n", " tool_call_id = tool_call.get(\"id\")\n", " if not isinstance(tool_call_id, str) or not tool_call_id:\n", " tool_call_id = f\"call_{iteration}_{index}\"\n", "\n", " tool_messages.append(\n", " ToolMessage(content=result_text, name=tool_name, tool_call_id=tool_call_id),\n", " )\n", " return tool_messages\n", "\n", " def invoke(self, user_message: str) -> str:\n", " \"\"\"Run the ReAct loop until the agent produces a final answer, while logging each step in the trace.\"\"\"\n", " self.trace = []\n", " self.conversation = [\n", " SystemMessage(content=self.system_prompt),\n", " HumanMessage(content=user_message),\n", " ]\n", "\n", " for i in range(self.max_iterations):\n", " response = self.model_with_tools.invoke(self.conversation)\n", " self.conversation.append(response)\n", "\n", " if response.tool_calls:\n", " tool_messages = self._execute_tools(response, iteration=i + 1)\n", " self.conversation.extend(tool_messages)\n", " continue\n", "\n", " final_text = self._to_text(response.content)\n", " self.trace.append(\n", " {\n", " \"iteration\": i + 1,\n", " \"action\": \"final_answer\",\n", " \"input\": \"-\",\n", " \"output\": final_text,\n", " },\n", " )\n", " return final_text\n", "\n", " max_iter_msg = \"⚠️ Max iterations reached without a final answer.\"\n", " self.trace.append(\n", " {\n", " \"iteration\": self.max_iterations,\n", " \"action\": \"final_answer\",\n", " \"input\": \"-\",\n", " \"output\": max_iter_msg,\n", " },\n", " )\n", " return max_iter_msg\n", "\n", " def print_trace(self) -> None:\n", " \"\"\"Print the agent's trace in a readable format.\"\"\"\n", " print(f\"{'Step':<5} | {'Action':<12} | {'Input':<40} | Output\")\n", " print(\"-\" * 120)\n", " for entry in self.trace:\n", " step = entry[\"iteration\"]\n", " action = entry[\"action\"]\n", " input_str = str(entry[\"input\"])\n", " output_str = str(entry[\"output\"])\n", " print(f\"{step:<5} | {action:<12} | {input_str[:40]:<40} | {output_str[:80]}\")\n", "\n", "\n", "traced_agent = TracedReActAgent(tools=[calculator, get_weather])\n", "max_retries = 5\n", "\n", "for attempt in range(max_retries):\n", " try:\n", " result = traced_agent.invoke(\"What is 15 times 23, and what's the weather in Tokyo?\")\n", " break\n", " except Exception as exc:\n", " error_msg = str(exc).lower()\n", " if \"429\" in error_msg or \"rate limit\" in error_msg:\n", " wait_seconds = min(2**attempt, 30)\n", " print(f\"Rate limit atteint. Nouvelle tentative dans {wait_seconds}s...\")\n", " time.sleep(wait_seconds)\n", " continue\n", " raise\n", "else:\n", " result = \"⚠️ Γ‰chec aprΓ¨s plusieurs tentatives (rate limit).\"\n", "\n", "print(result)\n", "traced_agent.print_trace()\n" ] }, { "cell_type": "markdown", "id": "3cdf0a94", "metadata": {}, "source": [ "### 🎯 Exercise 4: Structured Data Extraction Pipeline\n", "\n", "Build an agent that extracts structured information from unstructured text using `with_structured_output()`:\n", "\n", "1. Define a `MovieReview` Pydantic model with fields: `title`, `rating` (1-5), `sentiment`, `pros: List[str]`, `cons: List[str]`, `recommended: bool`\n", "2. Create a function that takes a raw text review and returns a `MovieReview`\n", "3. Process a batch of 3 reviews and display results in a formatted table\n", "\n", "**Bonus:** Add a `ReviewSummary` model that takes a list of `MovieReview`s and produces an overall recommendation." ] }, { "cell_type": "code", "execution_count": 243, "id": "a7645dec", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "🎬 Inception β€” ⭐⭐⭐⭐⭐ β€” positive\n", " βœ… Pros: ['Layers of dreams within dreams', \"Nolan's direction\", 'Stellar cast', 'Haunting ending']\n", " ❌ Cons: ['Can be confusing on first watch']\n", " πŸ‘ Recommended: True\n", "\n", "🎬 The Room β€” ⭐⭐ β€” mixed\n", " βœ… Pros: [\"It's so bad it's entertaining\"]\n", " ❌ Cons: ['Terrible acting', 'Plot makes no sense', 'Laughable dialogue']\n", " πŸ‘ Recommended: False\n", "\n", "🎬 Interstellar β€” ⭐⭐⭐⭐ β€” mixed\n", " βœ… Pros: ['visually stunning', 'fascinating science', 'moving father-daughter relationship']\n", " ❌ Cons: ['pacing drags in the middle', 'some plot points feel forced']\n", " πŸ‘ Recommended: True\n", "\n" ] } ], "source": [ "from pydantic import BaseModel, Field\n", "\n", "\n", "# Step 1: Define the MovieReview model\n", "class MovieReview(BaseModel):\n", " \"\"\"Extract structured information from a movie review.\"\"\"\n", "\n", " title: str = Field(description=\"The movie title\")\n", " rating: int = Field(description=\"Rating from 1 to 5 stars\")\n", " sentiment: str = Field(description=\"Overall sentiment: positive, negative, or mixed\")\n", " pros: list[str] = Field(description=\"List of positive points mentioned\")\n", " cons: list[str] = Field(description=\"List of negative points mentioned\")\n", " recommended: bool = Field(description=\"Whether the reviewer recommends the movie\")\n", "\n", "# Step 2: Create the extraction function\n", "def extract_review(raw_text: str) -> MovieReview:\n", " \"\"\"Extract structured data from a raw movie review.\"\"\"\n", " structured_model = model.with_structured_output(MovieReview)\n", " result = structured_model.invoke(raw_text)\n", " if isinstance(result, MovieReview):\n", " return result\n", " return MovieReview.model_validate(result)\n", "\n", "# Step 3: Test with sample reviews\n", "raw_reviews = [\n", " \"Inception is a masterpiece! The layers of dreams within dreams blew my mind. \"\n", " \"Nolan's direction is impeccable and the cast is stellar. The ending still haunts me. \"\n", " \"Only downside is it can be confusing on first watch. 5/5, must see!\",\n", "\n", " \"I watched The Room last night. Wow, what a disaster. The acting is terrible, \"\n", " \"the plot makes no sense, and the dialogue is laughable. But honestly? \"\n", " \"It's so bad it's entertaining. I'd give it 2/5.\",\n", "\n", " \"Interstellar was visually stunning and the science was fascinating. \"\n", " \"The relationship between father and daughter was moving. However, \"\n", " \"the pacing drags in the middle and some plot points feel forced. \"\n", " \"Overall a good watch. 4/5.\",\n", "]\n", "\n", "for review_text in raw_reviews:\n", " review = extract_review(review_text)\n", " print(f\"🎬 {review.title} β€” {'⭐' * review.rating} β€” {review.sentiment}\")\n", " print(f\" βœ… Pros: {review.pros}\")\n", " print(f\" ❌ Cons: {review.cons}\")\n", " print(f\" πŸ‘ Recommended: {review.recommended}\\n\")\n", "\n", "# BONUS: Create a ReviewSummary model\n", "class ReviewSummary(BaseModel):\n", " \"\"\"Summarize multiple movie reviews.\"\"\"\n", "\n", " total_reviews: int\n", " average_rating: float\n", " best_movie: str\n", " overall_recommendation: str\n", "\n" ] }, { "cell_type": "markdown", "id": "3781ae62", "metadata": {}, "source": [ "### 🌑️ Exercise 5: Temperature Impact on Agent Behavior\n", "\n", "Investigate how `temperature` affects an agent's **tool-calling decisions** and **final answers**.\n", "\n", "**Requirements:**\n", "1. Create the same LangGraph agent (with `calculator` and `get_weather`) at 3 different temperatures: `0.0`, `0.5`, `1.0`\n", "2. Send the same ambiguous query to all 3: \"I'm going to Paris next week, anything I should know?\"\n", "3. Compare: Does the agent decide to call tools? Which ones? How does the answer change?\n", "4. Run each temperature 3 times and track how consistent the tool choices are\n", "5. Print a comparison table" ] }, { "cell_type": "code", "execution_count": 244, "id": "a05d3d92", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Temperature: 0.0 | Run: 1 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 0.0 | Run: 2 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 0.0 | Run: 3 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 0.5 | Run: 1 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 0.5 | Run: 2 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 0.5 | Run: 3 | Tools Used: [] | Answer Preview: Paris is a beautiful city with a rich history and ...\n", "Temperature: 1.0 | Run: 1 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 1.0 | Run: 2 | Tools Used: [] | Answer Preview: That's exciting! Paris is a beautiful city with a ...\n", "Temperature: 1.0 | Run: 3 | Tools Used: [] | Answer Preview: That sounds like an exciting trip! Paris, the city...\n" ] } ], "source": [ "def run_temperature_experiment(query: str, temperatures: list[float], runs_per_temp: int = 3) -> None:\n", " \"\"\"Test how temperature affects agent tool-calling decisions.\n", "\n", " For each temperature:\n", " 1. Create a LangGraph agent with that temperature\n", " 2. Send the same query `runs_per_temp` times\n", " 3. Track which tools were called each time\n", " 4. Print a comparison table\n", "\n", " Args:\n", " query: The user query to test\n", " temperatures: List of temperature values\n", " runs_per_temp: Number of times to run each temperature\n", "\n", " \"\"\"\n", " for temp in temperatures:\n", " agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=temp),\n", " tools=[calculator, get_weather],\n", " )\n", " for run in range(1, runs_per_temp + 1):\n", " response = agent.invoke({\"messages\": [(\"user\", query)]})\n", " tool_calls = []\n", " for msg in response[\"messages\"]:\n", " if hasattr(msg, \"tool_calls\") and msg.tool_calls:\n", " tool_calls.extend([tc[\"name\"] for tc in msg.tool_calls])\n", " answer_preview = response[\"messages\"][-1].content[:50] + \"...\"\n", " print(f\"Temperature: {temp:<5} | Run: {run} | Tools Used: {tool_calls} | Answer Preview: {answer_preview}\")\n", "\n", "\n", "# Test it\n", "run_temperature_experiment(\n", " query=\"I'm going to Paris next week, anything I should know?\",\n", " temperatures=[0.0, 0.5, 1.0],\n", " runs_per_temp=3,\n", ")\n", "\n" ] }, { "cell_type": "markdown", "id": "0c43a6cb", "metadata": {}, "source": [ "### πŸ›‘οΈ Exercise 6: Agent with Error Handling & Retries\n", "\n", "Real-world tools can fail! Build an agent that **gracefully handles tool errors** and retries.\n", "\n", "**Requirements:**\n", "1. Create a `flaky_api` tool that randomly fails 50% of the time (raises an exception)\n", "2. Build a custom ReAct agent (not LangGraph) that:\n", " - Catches tool execution errors\n", " - Sends the error message back to the LLM as a `ToolMessage` with the error\n", " - Lets the LLM decide to retry or use a different approach\n", " - Has a maximum of 3 retries per tool call\n", "3. Test with a query that triggers the flaky tool" ] }, { "cell_type": "code", "execution_count": 245, "id": "1aff3292", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "πŸ’¬ It seems that the stock price service is currently unavailable. Would you like me to try again, or is there something else I can assist you with?\n" ] } ], "source": [ "\n", "\n", "# Step 1: Create a flaky tool\n", "@tool\n", "def flaky_stock_price(ticker: str) -> str:\n", " \"\"\"Get the current stock price for a company ticker symbol.\n", "\n", " Args:\n", " ticker: Stock ticker symbol (e.g., AAPL, GOOGL, MSFT)\n", "\n", " \"\"\"\n", " # 50% chance of failure\n", " if secrets.randbelow(2) == 0:\n", " message = f\"API timeout: Could not reach stock service for {ticker}\"\n", " raise ConnectionError(message)\n", "\n", " ticker_upper = ticker.upper()\n", " prices: dict[str, float] = {\n", " \"AAPL\": 187.50,\n", " \"GOOGL\": 142.30,\n", " \"MSFT\": 415.80,\n", " \"AMZN\": 185.60,\n", " }\n", "\n", " if ticker_upper in prices:\n", " price = prices[ticker_upper]\n", " else:\n", " # pseudo-random fallback price in [50.00, 500.00]\n", " cents = 5000 + secrets.randbelow(45001)\n", " price = cents / 100\n", "\n", " return f\"{ticker_upper}: ${price:.2f}\"\n", "\n", "\n", "# Step 2: Build a RobustReActAgent\n", "class RobustReActAgent:\n", " \"\"\"A ReAct agent that handles tool execution errors gracefully.\n", "\n", " When a tool fails, it sends the error back to the LLM and lets it decide what to do.\n", " \"\"\"\n", "\n", " def __init__(self, tools: list[BaseTool], max_iterations: int = 8, max_retries: int = 3) -> None:\n", " \"\"\"Initialize the agent with a model, tools, max iterations, max retries, and conversation history.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.tools = {t.name: t for t in tools}\n", " self.model_with_tools = self.model.bind_tools(tools)\n", " self.max_iterations = max_iterations\n", " self.max_retries = max_retries\n", " self.conversation: list[BaseMessage] = []\n", " self.retry_counts: dict[str, int] = {}\n", "\n", " @staticmethod\n", " def _content_to_text(content: str | list[str | dict]) -> str:\n", " \"\"\"Normalize model/tool content to plain string.\"\"\"\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", "\n", " def _execute_tools(self, ai_message: AIMessage) -> list[ToolMessage]:\n", " \"\"\"Execute tool calls with retry-aware error feedback.\"\"\"\n", " tool_messages: list[ToolMessage] = []\n", "\n", " for index, tool_call in enumerate(ai_message.tool_calls):\n", " tool_name = tool_call.get(\"name\")\n", " raw_args = tool_call.get(\"args\", {})\n", " tool_args = raw_args if isinstance(raw_args, dict) else {}\n", "\n", " raw_call_id = tool_call.get(\"id\")\n", " call_id = raw_call_id if isinstance(raw_call_id, str) and raw_call_id else f\"tool_call_{index}\"\n", "\n", " if tool_name is None or tool_name not in self.tools:\n", " tool_messages.append(\n", " ToolMessage(\n", " content=f\"Unknown tool: {tool_name}. Please answer without this tool.\",\n", " name=tool_name or \"unknown_tool\",\n", " tool_call_id=call_id,\n", " ),\n", " )\n", " continue\n", "\n", " tool = self.tools[tool_name]\n", "\n", " try:\n", " result = tool.invoke(tool_args)\n", " tool_messages.append(\n", " ToolMessage(\n", " content=self._content_to_text(result),\n", " name=tool_name,\n", " tool_call_id=call_id,\n", " ),\n", " )\n", " except (ConnectionError, TimeoutError, ValueError, TypeError, KeyError, RuntimeError) as exc:\n", " retries = self.retry_counts.get(call_id, 0) + 1\n", " self.retry_counts[call_id] = retries\n", "\n", " if retries <= self.max_retries:\n", " content = f\"Error executing {tool_name}: {exc!s}. You can retry this tool.\"\n", " else:\n", " content = (\n", " f\"Error executing {tool_name}: {exc!s}. \"\n", " f\"Tool failed after {self.max_retries} retries. Please answer without this tool.\"\n", " )\n", "\n", " tool_messages.append(\n", " ToolMessage(\n", " content=content,\n", " name=tool_name,\n", " tool_call_id=call_id,\n", " ),\n", " )\n", "\n", " return tool_messages\n", "\n", " def invoke(self, user_message: str) -> str:\n", " \"\"\"Run a ReAct loop with robust tool error handling.\"\"\"\n", " self.retry_counts = {}\n", " self.conversation = [HumanMessage(content=user_message)]\n", "\n", " for _ in range(self.max_iterations):\n", " ai_message = self.model_with_tools.invoke(self.conversation)\n", " self.conversation.append(ai_message)\n", "\n", " if not ai_message.tool_calls:\n", " return self._content_to_text(ai_message.content)\n", "\n", " tool_messages = self._execute_tools(ai_message)\n", " self.conversation.extend(tool_messages)\n", "\n", " return \"⚠️ Max iterations reached without a final answer.\"\n", "\n", "\n", "# Test it!\n", "robust_agent = RobustReActAgent(tools=[flaky_stock_price, calculator])\n", "result = robust_agent.invoke(\"What's the stock price of Apple (AAPL)?\")\n", "print(f\"\\nπŸ’¬ {result}\")\n" ] }, { "cell_type": "markdown", "id": "f63c2405", "metadata": {}, "source": [ "### πŸ“‹ Exercise 7: System Prompt Engineering for Agents\n", "\n", "System prompts **dramatically** change how an agent behaves β€” which tools it prefers, how verbose it is, how it reasons.\n", "\n", "**Requirements:**\n", "1. Create the same LangGraph agent (with `calculator`, `get_weather`, `search_knowledge_base`) 3 times, each with a different system prompt:\n", " - **\"Concise Agent\"**: Always use tools when possible. Give short, direct answers.\n", " - **\"Verbose Agent\"**: Explain your reasoning step by step. Always use tools to verify.\n", " - **\"Lazy Agent\"**: Only use tools when absolutely necessary. Prefer answering from your own knowledge.\n", "2. Send the same 3 queries to each agent\n", "3. Compare: answer length, number of tool calls, quality of answers\n", "4. Present results in a formatted comparison" ] }, { "cell_type": "code", "execution_count": 246, "id": "5c314dbf", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Prompt: Concise | Query: What is 42 times 58? | Tools Used: ['calculator'] | Tool Calls: 1 | Answer Length: 4\n", "Prompt: Concise | Query: What's the weather in London? | Tools Used: ['get_weather'] | Tool Calls: 1 | Answer Length: 49\n", "Prompt: Concise | Query: Tell me about Python programming | Tools Used: ['search_knowledge_base'] | Tool Calls: 1 | Answer Length: 41\n", "Prompt: Verbose | Query: What is 42 times 58? | Tools Used: ['calculator'] | Tool Calls: 1 | Answer Length: 35\n", "Prompt: Verbose | Query: What's the weather in London? | Tools Used: ['get_weather'] | Tool Calls: 1 | Answer Length: 66\n", "Prompt: Verbose | Query: Tell me about Python programming | Tools Used: ['search_knowledge_base'] | Tool Calls: 1 | Answer Length: 2352\n", "Prompt: Lazy | Query: What is 42 times 58? | Tools Used: [] | Tool Calls: 0 | Answer Length: 386\n", "Prompt: Lazy | Query: What's the weather in London? | Tools Used: ['get_weather'] | Tool Calls: 1 | Answer Length: 68\n", "Prompt: Lazy | Query: Tell me about Python programming | Tools Used: [] | Tool Calls: 0 | Answer Length: 483\n" ] } ], "source": [ "system_prompts = {\n", " \"Concise\": (\n", " \"You are a concise assistant. Always use your tools when they are relevant. \"\n", " \"Give short, direct answers β€” no more than 2 sentences.\"\n", " ),\n", " \"Verbose\": (\n", " \"You are a thorough assistant. Explain your reasoning step by step. \"\n", " \"Always use tools to verify facts before answering. Provide detailed explanations.\"\n", " ),\n", " \"Lazy\": (\n", " \"You are an efficient assistant. Only use tools when absolutely necessary β€” \"\n", " \"prefer answering from your own knowledge. Don't call tools for things you already know.\"\n", " ),\n", "}\n", "\n", "test_queries = [\n", " \"What is 42 times 58?\",\n", " \"What's the weather in London?\",\n", " \"Tell me about Python programming\",\n", "]\n", "\n", "\n", "def _result_to_text(result: object) -> str:\n", " if isinstance(result, str):\n", " return result\n", " if hasattr(result, \"content\"):\n", " content = result.content\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", " return str(result)\n", "\n", "\n", "def compare_system_prompts(prompts: dict[str, str], queries: list[str]) -> None:\n", " \"\"\"Create an agent for each system prompt, run all queries, and compare results.\"\"\"\n", " for prompt_name, prompt_text in prompts.items():\n", " agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=0),\n", " tools=[calculator, get_weather, search_knowledge_base],\n", " system_prompt=prompt_text,\n", " )\n", " for query in queries:\n", " response = agent.invoke({\"messages\": [(\"user\", query)]})\n", " tool_calls = []\n", " for msg in response[\"messages\"]:\n", " if hasattr(msg, \"tool_calls\") and msg.tool_calls:\n", " tool_calls.extend([tc[\"name\"] for tc in msg.tool_calls])\n", " final_answer = _result_to_text(response[\"messages\"][-1])\n", " answer_length = len(final_answer)\n", " print(\n", " f\"Prompt: {prompt_name:<8} | Query: {query:<25} | \"\n", " f\"Tools Used: {tool_calls} | Tool Calls: {len(tool_calls)} | Answer Length: {answer_length}\",\n", " )\n", "\n", "\n", "compare_system_prompts(system_prompts, test_queries)\n" ] }, { "cell_type": "markdown", "id": "f3f20e09", "metadata": {}, "source": [ "### πŸ—οΈ Exercise 8: Research & Report Agent\n", "\n", "Build a two-phase agent pipeline:\n", "1. **Phase 1 β€” Research:** Use a tool-calling agent to gather information (search knowledge base, calculate, etc.)\n", "2. **Phase 2 β€” Report:** Feed the gathered information into a `with_structured_output()` call to produce a clean `ResearchReport` Pydantic object\n", "\n", "**Requirements:**\n", "- Define a `ResearchReport` model with: `title`, `summary`, `key_facts: List[str]`, `related_topics: List[str]`, `confidence_score: float`\n", "- The research phase should actually call tools (search, etc.)\n", "- The report phase should format the agent's findings into the structured model\n", "- Test with: \"What is RAG in machine learning?\"" ] }, { "cell_type": "code", "execution_count": 247, "id": "5fc15855", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "πŸ“„ Title: Retrieval-Augmented Generation (RAG): Enhancing Language Models with Retrieval Mechanisms\n", "πŸ“ Summary: Retrieval-Augmented Generation (RAG) is a machine learning technique that combines retrieval-based methods with generative models to improve the performance of language models. By fetching relevant information from a large corpus before generating responses, RAG enhances accuracy and reduces hallucination. This approach is scalable and flexible, applicable in various NLP tasks such as question answering, dialogue systems, and summarization.\n", "πŸ”‘ Key Facts:\n", " β€’ RAG combines retrieval-based methods with generative models to improve language model performance.\n", " β€’ The retrieval module fetches relevant documents using techniques like vector similarity search.\n", " β€’ The generation module produces coherent responses based on retrieved information.\n", " β€’ RAG enhances accuracy, reduces hallucination, and is scalable and flexible.\n", " β€’ Applications include question answering, dialogue systems, summarization, and content generation.\n", "πŸ”— Related Topics: ['Information Retrieval Techniques', 'Generative Models in NLP', 'Question Answering Systems', 'Dialogue Systems and Chatbots', 'Natural Language Processing Applications']\n", "πŸ“Š Confidence: 95%\n" ] } ], "source": [ "from pydantic import BaseModel, Field\n", "\n", "\n", "class ResearchReport(BaseModel):\n", " \"\"\"A structured research report.\"\"\"\n", "\n", " title: str = Field(description=\"Title of the research report\")\n", " summary: str = Field(description=\"A 2-3 sentence summary of findings\")\n", " key_facts: list[str] = Field(description=\"3-5 key facts discovered\")\n", " related_topics: list[str] = Field(description=\"Related topics worth exploring\")\n", " confidence_score: float = Field(description=\"Confidence in findings, 0.0 to 1.0\")\n", "\n", "\n", "def research_and_report(topic: str) -> ResearchReport:\n", " \"\"\"Two-phase pipeline.\n", "\n", " Phase 1: Use a tool-calling agent to research the topic\n", " Phase 2: Use structured output to format findings into a ResearchReport\n", " \"\"\"\n", " research_agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=0),\n", " tools=[search_knowledge_base, get_current_time],\n", " )\n", " response = research_agent.invoke({\"messages\": [(\"user\", f\"Research this topic thoroughly: {topic}\")]})\n", " raw_findings = response[\"messages\"][-1].content\n", " findings_text = raw_findings if isinstance(raw_findings, str) else str(raw_findings)\n", "\n", " report_model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0).with_structured_output(ResearchReport)\n", " report_result = report_model.invoke(\n", " f\"Based on these research findings, create a structured report:\\n\\n{findings_text}\",\n", " )\n", "\n", " if isinstance(report_result, ResearchReport):\n", " return report_result\n", " if isinstance(report_result, BaseModel):\n", " return ResearchReport.model_validate(report_result.model_dump())\n", " if isinstance(report_result, dict):\n", " return ResearchReport.model_validate(report_result)\n", " return ResearchReport.model_validate_json(str(report_result))\n", "\n", "\n", "report = research_and_report(\"What is RAG in machine learning?\")\n", "print(f\"πŸ“„ Title: {report.title}\")\n", "print(f\"πŸ“ Summary: {report.summary}\")\n", "print(\"πŸ”‘ Key Facts:\")\n", "for fact in report.key_facts:\n", " print(f\" β€’ {fact}\")\n", "print(f\"πŸ”— Related Topics: {report.related_topics}\")\n", "print(f\"πŸ“Š Confidence: {report.confidence_score:.0%}\")\n" ] }, { "cell_type": "markdown", "id": "585f2a7d", "metadata": {}, "source": [ "### πŸ’¬ Exercise 9: Conversational Agent with Persistent Memory\n", "\n", "Build a `PersistentChatAgent` that combines **tools** and **multi-turn memory** in a single class using LangGraph.\n", "\n", "**Requirements:**\n", "1. The agent should maintain conversation history across `.chat()` calls\n", "2. It should have access to `calculator`, `get_weather`, and `save_note` / `get_notes` tools\n", "3. Test a multi-turn scenario:\n", " - \"What's 144 divided by 12?\" β†’ uses calculator\n", " - \"Save that result as a note titled 'Math Result'\" β†’ uses save_note\n", " - \"What's the weather in Tokyo?\" β†’ uses get_weather\n", " - \"Save the Tokyo weather as a note too\" β†’ uses save_note\n", " - \"Show me all my notes\" β†’ uses get_notes\n", " - \"What was the first thing I asked you?\" β†’ must remember from conversation history" ] }, { "cell_type": "code", "execution_count": 248, "id": "59fbcfb1", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "πŸ‘€ What's 144 divided by 12?\n", "πŸ€– The answer is 12.\n", " πŸ“Š History: 2 messages\n", "\n", "πŸ‘€ Save that result as a note titled 'Math Result'\n", "πŸ€– I've saved the result as a note titled 'Math Result'.\n", " πŸ“Š History: 4 messages\n", "\n", "πŸ‘€ What's the weather in Tokyo?\n", "πŸ€– The weather in Tokyo is currently clear with a temperature of 25Β°C.\n", " πŸ“Š History: 6 messages\n", "\n", "πŸ‘€ Save the Tokyo weather as a note too\n", "πŸ€– What title would you like to give to this note?\n", " πŸ“Š History: 8 messages\n", "\n", "πŸ‘€ Show me all my notes\n", "πŸ€– Here are your saved notes:\n", "\n", "1. Math Result: 144 divided by 12 equals 12\n", "2. Tokyo Weather: The weather in Tokyo is currently clear with a temperature of 25Β°C\n", " πŸ“Š History: 10 messages\n", "\n", "πŸ‘€ What was the first thing I asked you?\n", "πŸ€– The first thing you asked me was \"What's 144 divided by 12?\"\n", " πŸ“Š History: 12 messages\n" ] } ], "source": [ "class PersistentChatAgent:\n", " \"\"\"An agent that remembers the full conversation AND uses tools.\n", "\n", " Each call to .chat() appends to the running message history.\n", " \"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the agent with a model, tools, system prompt, and empty message history.\"\"\"\n", " # Your code here:\n", " # 1. Initialize the ChatMistralAI model\n", " # 2. Define the tools list\n", " # 3. Create the LangGraph react agent\n", " # 4. Initialize an empty message history list\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.tools = [calculator, get_weather, search_knowledge_base, save_note, get_notes, get_current_time]\n", " self.agent = create_agent(\n", " model=self.model,\n", " tools=self.tools,\n", " system_prompt=(\n", " \"You are a helpful assistant that remembers the entire conversation. \"\n", " \"Use tools whenever they can help you answer the user's questions. \"\n", " \"Maintain a running history of the conversation and use it to inform your answers.\"\n", " ),\n", " )\n", " self.messages: list = [] # This will store the full conversation history\n", "\n", " def chat(self, user_message: str) -> str:\n", " \"\"\"Send a message, get a response, and maintain conversation history.\"\"\"\n", " # Your code here:\n", " # 1. Append (\"user\", user_message) to self.messages\n", " # 2. Invoke the agent with the FULL message history\n", " # 3. Extract the final AI message content\n", " # 4. Append (\"assistant\", answer) to self.messages\n", " # 5. Return the answer\n", " self.messages.append((\"user\", user_message))\n", " response = self.agent.invoke({\"messages\": self.messages})\n", " final_msg = response[\"messages\"][-1].content\n", " self.messages.append((\"assistant\", final_msg))\n", " return final_msg\n", "\n", " def get_history_length(self) -> int:\n", " \"\"\"Get the number of messages in the conversation history.\"\"\"\n", " # Return the number of messages\n", " return len(self.messages)\n", "\n", "\n", "# Test the multi-turn scenario\n", "notes_store.clear() # Reset notes\n", "agent = PersistentChatAgent()\n", "\n", "conversation_flow = [\n", " \"What's 144 divided by 12?\",\n", " \"Save that result as a note titled 'Math Result'\",\n", " \"What's the weather in Tokyo?\",\n", " \"Save the Tokyo weather as a note too\",\n", " \"Show me all my notes\",\n", " \"What was the first thing I asked you?\",\n", "]\n", "\n", "for msg in conversation_flow:\n", " print(f\"\\nπŸ‘€ {msg}\")\n", " response = agent.chat(msg)\n", " print(f\"πŸ€– {response}\")\n", " print(f\" πŸ“Š History: {agent.get_history_length()} messages\")\n", "\n" ] }, { "cell_type": "markdown", "id": "a0864b7c", "metadata": {}, "source": [ "### πŸ”€ Exercise 10: Agent Router β€” Classify & Dispatch\n", "\n", "Build a system that **classifies** user queries and **routes** them to different specialized agents.\n", "\n", "**Requirements:**\n", "1. Create a `QueryClassifier` using `with_structured_output()` that classifies queries into:\n", " - `\"math\"` β€” send to a calculator agent\n", " - `\"weather\"` β€” send to a weather agent\n", " - `\"knowledge\"` β€” send to a knowledge base agent\n", " - `\"general\"` β€” answer directly without tools\n", "2. Create a specialized mini-agent for each category\n", "3. Build a `RouterAgent` that:\n", " - Classifies the incoming query\n", " - Dispatches to the right specialized agent\n", " - Returns the result along with which route was taken\n", "4. Test with a variety of queries" ] }, { "cell_type": "code", "execution_count": null, "id": "9903101d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "πŸ‘€ Query: What is 256 divided by 8?\n", " 🏷️ Category: math (The query involves a mathematical operation (division).)\n", " πŸ’¬ Answer: The result of 256 divided by 8 is 32.\n", "\n", "πŸ‘€ Query: What's the weather in San Francisco?\n", " 🏷️ Category: weather (The query asks for weather information, specifically about San Francisco.)\n", " πŸ’¬ Answer: The weather in San Francisco is currently foggy with a temperature of 18Β°C.\n", "\n", "πŸ‘€ Query: Tell me about LangChain\n", " 🏷️ Category: knowledge (The query 'Tell me about LangChain' is asking for information about a specific topic, which falls under the 'knowledge' category.)\n", " πŸ’¬ Answer: LangChain is a framework designed to facilitate the development of applications powered by language models. It offers a range of tools and features th\n", "\n", "πŸ‘€ Query: What are the benefits of exercise?\n", " 🏷️ Category: knowledge (The query asks for information about the benefits of exercise, which is a factual and informational question. It does not pertain to math, weather, or general topics.)\n", " πŸ’¬ Answer: I couldn't find specific results for the benefits of exercise. However, I can share some general benefits based on common knowledge:\n", "\n", "1. **Improved Ph\n", "\n", "πŸ‘€ Query: Calculate 99 times 101\n", " 🏷️ Category: math (The query involves a mathematical calculation, specifically multiplying two numbers.)\n", " πŸ’¬ Answer: The result of 99 times 101 is 9,999.\n", "\n", "πŸ‘€ Query: Is it raining in London?\n", " 🏷️ Category: weather (The query asks about the current weather conditions in a specific location, which falls under the 'weather' category.)\n", " πŸ’¬ Answer: Yes, it is currently raining in London. The temperature is 15Β°C.\n" ] } ], "source": [ "from enum import Enum\n", "\n", "from pydantic import BaseModel, Field\n", "\n", "\n", "class QueryCategory(str, Enum):\n", " \"\"\"Categories for classifying user queries.\"\"\"\n", "\n", " math = \"math\"\n", " weather = \"weather\"\n", " knowledge = \"knowledge\"\n", " general = \"general\"\n", "\n", "\n", "class QueryClassification(BaseModel):\n", " \"\"\"Classify a user query into the appropriate category.\"\"\"\n", "\n", " category: QueryCategory = Field(description=\"The category this query belongs to\")\n", " reasoning: str = Field(description=\"Brief explanation for the classification\")\n", "\n", "\n", "def _to_text(value: object) -> str:\n", " if isinstance(value, str):\n", " return value\n", " if hasattr(value, \"content\"):\n", " content = value.content\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", " return str(value)\n", "\n", "\n", "class RouterAgent:\n", " \"\"\"Classifies incoming queries and routes them to specialized agents.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the router with a classification model and specialized agents for math, weather, and knowledge queries.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0)\n", " self.classifier = self.model.with_structured_output(QueryClassification)\n", " self.math_agent = create_agent(\n", " model=self.model,\n", " tools=[calculator],\n", " system_prompt=\"You are a math assistant. Use the calculator tool to solve math problems.\",\n", " )\n", " self.weather_agent = create_agent(\n", " model=self.model,\n", " tools=[get_weather],\n", " system_prompt=\"You are a weather assistant. Use the get_weather tool to provide current weather information.\",\n", " )\n", " self.knowledge_agent = create_agent(\n", " model=self.model,\n", " tools=[search_knowledge_base],\n", " system_prompt=\"You are a knowledge assistant. Use the search_knowledge_base tool to find information on various topics.\",\n", " )\n", "\n", " def classify(self, query: str) -> QueryClassification:\n", " \"\"\"Classify the query into a category using the classifier model.\"\"\"\n", " result = self.classifier.invoke(\n", " f\"Classify this query into one of: math, weather, knowledge, general. Query: {query}\",\n", " )\n", " if isinstance(result, QueryClassification):\n", " return result\n", " if isinstance(result, BaseModel):\n", " return QueryClassification.model_validate(result.model_dump())\n", " if isinstance(result, dict):\n", " return QueryClassification.model_validate(result)\n", " return QueryClassification.model_validate_json(str(result))\n", "\n", " def route(self, query: str) -> dict[str, str]:\n", " \"\"\"Classify the query and route it to the appropriate agent.\"\"\"\n", " classification = self.classify(query)\n", " category = classification.category\n", " reasoning = classification.reasoning\n", "\n", " if category == QueryCategory.math:\n", " answer = _to_text(self.math_agent.invoke({\"messages\": [(\"user\", query)]})[\"messages\"][-1])\n", " elif category == QueryCategory.weather:\n", " answer = _to_text(self.weather_agent.invoke({\"messages\": [(\"user\", query)]})[\"messages\"][-1])\n", " elif category == QueryCategory.knowledge:\n", " answer = _to_text(self.knowledge_agent.invoke({\"messages\": [(\"user\", query)]})[\"messages\"][-1])\n", " else:\n", " answer = _to_text(self.model.invoke(query))\n", "\n", " return {\"category\": category.value, \"reasoning\": reasoning, \"answer\": answer}\n", "\n", "\n", "test_queries = [\n", " \"What is 256 divided by 8?\",\n", " \"What's the weather in San Francisco?\",\n", " \"Tell me about LangChain\",\n", " \"What are the benefits of exercise?\",\n", " \"Calculate 99 times 101\",\n", " \"Is it raining in London?\",\n", "]\n", "\n", "router = RouterAgent()\n", "for query in test_queries:\n", " result = router.route(query)\n", " print(f\"\\nπŸ‘€ Query: {query}\")\n", " print(f\" 🏷️ Category: {result['category']} ({result['reasoning']})\")\n", " print(f\" πŸ’¬ Answer: {result['answer'][:150]}\")\n" ] }, { "cell_type": "markdown", "id": "c4483b53", "metadata": {}, "source": [ "### 🌊 Exercise 11: Streaming Agent with Progress Callbacks\n", "\n", "Build an agent wrapper that provides **real-time progress callbacks** during execution, so a user interface could display what the agent is doing.\n", "\n", "**Requirements:**\n", "1. Use LangGraph's `.stream()` method (shown in Part 4)\n", "2. Create a `StreamingAgentRunner` class that:\n", " - Accepts callback functions: `on_thinking`, `on_tool_call`, `on_tool_result`, `on_final_answer`\n", " - Streams the agent execution and fires the callbacks at each step\n", " - Tracks total execution time and number of steps\n", "3. Test with a query that triggers multiple tool calls\n", "4. Implement simple callbacks that print progress with timestamps" ] }, { "cell_type": "code", "execution_count": 250, "id": "d1be6374", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "🌊 Streaming agent execution:\n", "\n", " ⏳ [17:20:54 UTC] Agent thinking: choosing tool\n", " πŸ”§ [17:20:54 UTC] Calling: calculator({'operation': 'multiply', 'x': 25, 'y': 17})\n", " πŸ”§ [17:20:54 UTC] Calling: get_weather({'city': 'Paris'})\n", " πŸ“‹ [17:20:54 UTC] Result: 25.0 multiply 17.0 = 425.0\n", " πŸ“‹ [17:20:54 UTC] Result: Weather in Paris: β˜€οΈ 22Β°C, Sunny with light clouds\n", " βœ… [17:20:54 UTC] Final answer ready!\n", "\n", "πŸ“Š Summary: 4 steps in 0.9s\n", " Tools used: ['calculator', 'get_weather']\n", " Answer: The result of 25 multiplied by 17 is 425. The weather in Paris is currently sunny with light clouds and a temperature of 22Β°C.\n" ] } ], "source": [ "from collections.abc import Callable\n", "\n", "\n", "class StreamingAgentRunner:\n", " \"\"\"Wraps a LangGraph agent and fires callbacks during execution.\"\"\"\n", "\n", " def __init__(\n", " self,\n", " tools: list,\n", " on_thinking: Callable[[str], None] | None = None,\n", " on_tool_call: Callable[[str, dict], None] | None = None,\n", " on_tool_result: Callable[[str], None] | None = None,\n", " on_final_answer: Callable[[str], None] | None = None,\n", " ) -> None:\n", " \"\"\"Initialize the runner with tools and optional callbacks.\"\"\"\n", " self.agent = create_agent(\n", " model=ChatMistralAI(model=\"mistral-small-latest\", temperature=0),\n", " tools=tools,\n", " system_prompt=(\n", " \"You are a helpful assistant that explains your reasoning step by step. \"\n", " \"Use tools whenever they can help you answer the user's question.\"\n", " ),\n", " )\n", " self.on_thinking = on_thinking\n", " self.on_tool_call = on_tool_call\n", " self.on_tool_result = on_tool_result\n", " self.on_final_answer = on_final_answer\n", " self.steps = 0\n", " self.tools_used: set[str] = set()\n", "\n", " @staticmethod\n", " def _to_text(content: str | list[str | dict]) -> str:\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", "\n", " @staticmethod\n", " def _extract_messages(node_output: dict[str, object]) -> list[BaseMessage]:\n", " \"\"\"Extract message list from a node output.\"\"\"\n", " messages = node_output.get(\"messages\", [])\n", " return messages if isinstance(messages, list) else []\n", "\n", " def _handle_tool_calls(self, tool_calls: list[dict[str, object]]) -> None:\n", " \"\"\"Handle tool call callbacks and tracking.\"\"\"\n", " if self.on_thinking:\n", " self.on_thinking(\"choosing tool\")\n", "\n", " for tool_call in tool_calls:\n", " tool_name = str(tool_call.get(\"name\", \"unknown\"))\n", " tool_args = tool_call.get(\"args\", {})\n", " self.tools_used.add(tool_name)\n", "\n", " if self.on_tool_call:\n", " self.on_tool_call(tool_name, tool_args if isinstance(tool_args, dict) else {})\n", "\n", " def _handle_message(self, msg: BaseMessage, final_answer: str) -> str:\n", " \"\"\"Process one message and return updated final answer.\"\"\"\n", " tool_calls = getattr(msg, \"tool_calls\", None)\n", " if tool_calls:\n", " self._handle_tool_calls(tool_calls)\n", " return final_answer\n", "\n", " if isinstance(msg, ToolMessage):\n", " if self.on_tool_result:\n", " self.on_tool_result(self._to_text(msg.content))\n", " return final_answer\n", "\n", " content = getattr(msg, \"content\", \"\")\n", " return self._to_text(content) if content else final_answer\n", "\n", " def run(self, user_message: str) -> dict[str, object]:\n", " \"\"\"Stream agent execution with callbacks.\"\"\"\n", " start_time = time.time()\n", " self.steps = 0\n", " self.tools_used.clear()\n", " final_answer = \"\"\n", "\n", " for step in self.agent.stream({\"messages\": [(\"user\", user_message)]}):\n", " self.steps += 1\n", " for node_output in step.values():\n", " for msg in self._extract_messages(node_output):\n", " final_answer = self._handle_message(msg, final_answer)\n", "\n", " if self.on_final_answer:\n", " self.on_final_answer(final_answer)\n", "\n", " return {\n", " \"answer\": final_answer,\n", " \"steps\": self.steps,\n", " \"elapsed_seconds\": time.time() - start_time,\n", " \"tools_used\": sorted(self.tools_used),\n", " }\n", "\n", "\n", "def print_thinking(node_name: str) -> None:\n", " \"\"\"Print a message when the agent is thinking in a node.\"\"\"\n", " print(f\" ⏳ [{datetime.now(UTC).strftime('%H:%M:%S %Z')}] Agent thinking: {node_name}\")\n", "\n", "\n", "def print_tool_call(tool_name: str, tool_args: dict) -> None:\n", " \"\"\"Print a message when the agent calls a tool.\"\"\"\n", " print(f\" πŸ”§ [{datetime.now(UTC).strftime('%H:%M:%S %Z')}] Calling: {tool_name}({tool_args})\")\n", "\n", "\n", "def print_tool_result(result: str) -> None:\n", " \"\"\"Print a message when the agent receives a tool result.\"\"\"\n", " print(f\" πŸ“‹ [{datetime.now(UTC).strftime('%H:%M:%S %Z')}] Result: {result[:100]}\")\n", "\n", "\n", "def print_final(_answer: str) -> None:\n", " \"\"\"Print a message when the agent produces the final answer.\"\"\"\n", " print(f\" βœ… [{datetime.now(UTC).strftime('%H:%M:%S %Z')}] Final answer ready!\")\n", "\n", "\n", "runner = StreamingAgentRunner(\n", " tools=[calculator, get_weather, search_knowledge_base],\n", " on_thinking=print_thinking,\n", " on_tool_call=print_tool_call,\n", " on_tool_result=print_tool_result,\n", " on_final_answer=print_final,\n", ")\n", "print(\"🌊 Streaming agent execution:\\n\")\n", "summary = runner.run(\"What is 25 * 17, and what's the weather in Paris?\")\n", "print(f\"\\nπŸ“Š Summary: {summary['steps']} steps in {summary['elapsed_seconds']:.1f}s\")\n", "print(f\" Tools used: {summary['tools_used']}\")\n", "print(f\" Answer: {summary['answer']}\")\n" ] }, { "cell_type": "markdown", "id": "da13d2f2", "metadata": {}, "source": [ "### 🀝 Exercise 12: Reflection Agent (Generate β†’ Critique β†’ Improve)\n", "\n", "Build an agent that uses a **reflection loop** to iteratively improve its output. This is a common advanced pattern where one LLM call generates content and another critiques it.\n", "\n", "**Requirements:**\n", "1. Create a `ReflectionAgent` with two internal chains:\n", " - **Generator**: Writes content based on a user request\n", " - **Critic**: Reviews the generated content and provides feedback\n", "2. The agent loops `max_rounds` times:\n", " - Generate β†’ Critique β†’ Revise based on critique β†’ Critique again β†’ ...\n", "3. Stop early if the critic rates the content above a threshold (e.g., 8/10)\n", "4. Use structured output for the critic: `CritiqueResult` with `score: int`, `feedback: str`, `suggestions: List[str]`\n", "5. Test with: \"Write a concise explanation of how neural networks learn\"" ] }, { "cell_type": "code", "execution_count": 251, "id": "21437c5f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Round 1: Score 8/10\n", "\n", "======================================================================\n", "πŸ“ Final Content (after 1 rounds, score: 8/10):\n", "\n", "**How Neural Networks Learn**\n", "\n", "Neural networks learn by adjusting their internal parameters (weights and biases) to minimize prediction errors. Here’s the process:\n", "\n", "1. **Forward Pass**: Input data flows through the network, producing an output.\n", "2. **Loss Calculation**: The output is compared to the true value using a loss function (e.g., mean squared error).\n", "3. **Backpropagation**: The network calculates how much each weight contributed to the error and adjusts them in the opposite direction (gradient descent).\n", "4. **Iteration**: Steps 1–3 repeat with new data until the network’s predictions improve.\n", "\n", "Over time, the network refines its weights to make accurate predictions.\n", "\n", "πŸ“Š Improvement History:\n", " Round 1: Score 8/10 β€” The content is clear and concise, effectively explaining the basic process of ho...\n" ] } ], "source": [ "from pydantic import BaseModel, Field\n", "\n", "\n", "class CritiqueResult(BaseModel):\n", " \"\"\"Structured critique of generated content.\"\"\"\n", "\n", " score: int = Field(description=\"Quality score from 1 to 10\")\n", " feedback: str = Field(description=\"Overall feedback on the content\")\n", " suggestions: list[str] = Field(description=\"Specific suggestions for improvement\")\n", "\n", "\n", "class ReflectionAgent:\n", " \"\"\"An agent that generates content and iteratively improves it using self-critique.\"\"\"\n", "\n", " def __init__(self, max_rounds: int = 3, quality_threshold: int = 8) -> None:\n", " \"\"\"Initialize the agent with a model, critic model, max rounds, quality threshold, and history.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0.3)\n", " self.critic_model = self.model.with_structured_output(CritiqueResult)\n", " self.max_rounds = max_rounds\n", " self.quality_threshold = quality_threshold\n", " self.history: list[dict] = []\n", "\n", " @staticmethod\n", " def _to_text(content: str | list[str | dict]) -> str:\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", "\n", " def generate(self, request: str, previous_critique: str = \"\") -> str:\n", " \"\"\"Generate or revise content based on the request and optional critique.\"\"\"\n", " if not previous_critique:\n", " prompt = f\"Generate concise, clear content for this request:\\n\\n{request}\"\n", " else:\n", " prompt = (\n", " \"Revise the content for this request based on the critique below. \"\n", " \"Apply the suggestions explicitly.\\n\\n\"\n", " f\"Critique:\\n{previous_critique}\\n\\n\"\n", " f\"Request:\\n{request}\"\n", " )\n", " response = self.model.invoke(prompt)\n", " return self._to_text(response.content)\n", "\n", " def critique(self, content: str, original_request: str) -> CritiqueResult:\n", " \"\"\"Critique the generated content.\"\"\"\n", " prompt = (\n", " f\"Rate and critique this content written for: '{original_request}'. \"\n", " \"Return score (1-10), feedback, and concrete suggestions.\\n\\n\"\n", " f\"Content:\\n{content}\"\n", " )\n", " result = self.critic_model.invoke(prompt)\n", " if isinstance(result, CritiqueResult):\n", " return result\n", " if isinstance(result, BaseModel):\n", " return CritiqueResult.model_validate(result.model_dump())\n", " if isinstance(result, dict):\n", " return CritiqueResult.model_validate(result)\n", " return CritiqueResult.model_validate_json(str(result))\n", "\n", " def run(self, request: str) -> dict[str, object]:\n", " \"\"\"Run the full reflection loop.\"\"\"\n", " self.history = []\n", " content = self.generate(request)\n", "\n", " for round_num in range(self.max_rounds):\n", " critique = self.critique(content, request)\n", " self.history.append(\n", " {\n", " \"round\": round_num + 1,\n", " \"content\": content,\n", " \"score\": critique.score,\n", " \"feedback\": critique.feedback,\n", " \"suggestions\": critique.suggestions,\n", " },\n", " )\n", " print(f\"Round {round_num + 1}: Score {critique.score}/10\")\n", "\n", " if critique.score >= self.quality_threshold:\n", " break\n", "\n", " improvement_notes = critique.feedback\n", " if critique.suggestions:\n", " improvement_notes += \"\\n\\nSuggestions:\\n- \" + \"\\n- \".join(critique.suggestions)\n", " content = self.generate(request, previous_critique=improvement_notes)\n", "\n", " return {\n", " \"final_content\": content,\n", " \"rounds\": len(self.history),\n", " \"final_score\": self.history[-1][\"score\"] if self.history else 0,\n", " \"history\": self.history,\n", " }\n", "\n", "\n", "agent = ReflectionAgent(max_rounds=3, quality_threshold=8)\n", "result = agent.run(\"Write a concise explanation of how neural networks learn\")\n", "\n", "print(f\"\\n{'='*70}\")\n", "print(f\"πŸ“ Final Content (after {result['rounds']} rounds, score: {result['final_score']}/10):\\n\")\n", "print(result[\"final_content\"])\n", "print(\"\\nπŸ“Š Improvement History:\")\n", "for i, round_info in enumerate(result[\"history\"]):\n", " print(f\" Round {i+1}: Score {round_info['score']}/10 β€” {round_info['feedback'][:80]}...\")\n" ] }, { "cell_type": "markdown", "id": "d9a92759", "metadata": {}, "source": [ "### πŸ—οΈ Exercise 13: Hierarchical Multi-Agent System (Planner + Executors)\n", "\n", "Build a **two-layer agent architecture** where a **Planner** agent decomposes a complex task into sub-tasks and dispatches them to **Specialist Executor** agents.\n", "\n", "**Requirements:**\n", "1. Define a structured `Plan` model with a list of `SubTask` items (each with `agent`, `instruction`, `depends_on`)\n", "2. Create three specialist agents as simple functions or classes:\n", " - **ResearcherAgent**: Answers factual questions (uses `search_knowledge_base` or a stub)\n", " - **AnalystAgent**: Performs numerical analysis / comparisons (uses `calculator` or logic)\n", " - **WriterAgent**: Synthesises findings into polished prose\n", "3. Build a `PlannerAgent` that:\n", " - Receives a high-level goal from the user\n", " - Returns a structured `Plan` using `with_structured_output`\n", "4. Build an `OrchestratorAgent` that:\n", " - Calls `PlannerAgent` to get the plan\n", " - Executes sub-tasks in dependency order (sequential for simplicity)\n", " - Passes the output of each step as context to dependent steps\n", " - Returns the final writer output and an execution log\n", "5. Test with: `\"Compare supervised vs unsupervised learning and write a short summary\"`\n" ] }, { "cell_type": "code", "execution_count": 253, "id": "5760c75d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "======================================================================\n", "🎯 Goal: Compare supervised vs unsupervised learning and write a short summary\n", "\n", "πŸ“‹ Plan (4 steps):\n", " [step_1] RESEARCHER: Find key characteristics of supervised learning...\n", " [step_2] RESEARCHER: Find key characteristics of unsupervised learning...\n", " [step_3] ANALYST: Compare and contrast the characteristics of supervised and u... (depends on: ['step_1', 'step_2'])\n", " [step_4] WRITER: Write a short summary comparing supervised and unsupervised ... (depends on: ['step_3'])\n", "\n", "πŸ“ Execution Log:\n", " Step step_1 output: Supervised learning is a type of machine learning where a model is trained on a labeled dataset, mea...\n", " Step step_2 output: Unsupervised learning is a type of machine learning where the model learns patterns and structures f...\n", " Step step_3 output: Here’s a structured comparison and contrast of **supervised** and **unsupervised learning** based on...\n", " Step step_4 output: **Summary: Supervised vs. Unsupervised Learning**\n", "\n", "Supervised and unsupervised learning are two core...\n", "\n", "✍️ Final Output:\n", "\n", "**Summary: Supervised vs. Unsupervised Learning**\n", "\n", "Supervised and unsupervised learning are two core approaches in machine learning, differing primarily in their use of labeled data and objectives.\n", "\n", "- **Supervised learning** relies on **labeled datasets** (input-output pairs) to train models for tasks like classification or regression, with clear evaluation metrics (e.g., accuracy, MSE). It is **goal-driven** and requires human effort for data labeling.\n", "\n", "- **Unsupervised learning** works with **unlabeled data**, focusing on discovering hidden patterns, clustering, or dimensionality reduction (e.g., K-means, PCA). Evaluation is less straightforward, often relying on domain-specific metrics.\n", "\n", "While supervised learning is **task-specific** and interpretable, unsupervised learning is **exploratory** and flexible. Both methods are often combined in workflowsβ€”for example, using unsupervised clustering to preprocess data before supervised modeling. The choice depends on the problem, data availability, and desired outcomes.\n" ] } ], "source": [ "from pydantic import BaseModel, Field\n", "\n", "\n", "class SubTask(BaseModel):\n", " \"\"\"A single step in the execution plan.\"\"\"\n", "\n", " id: str = Field(description=\"Unique identifier, e.g. 'step_1'\")\n", " agent: str = Field(\n", " description=\"Which agent to use: 'researcher', 'analyst', or 'writer'\",\n", " )\n", " instruction: str = Field(description=\"Detailed instruction for the agent\")\n", " depends_on: list[str] = Field(\n", " default_factory=list,\n", " description=\"List of step ids whose output this step needs as input\",\n", " )\n", "\n", "\n", "class Plan(BaseModel):\n", " \"\"\"The full execution plan produced by the Planner.\"\"\"\n", "\n", " goal: str = Field(description=\"The high-level user goal\")\n", " steps: list[SubTask] = Field(description=\"Ordered list of sub-tasks to execute\")\n", "\n", "\n", "def _extract_text(response: object) -> str:\n", " \"\"\"Convert model output to plain text.\"\"\"\n", " if isinstance(response, str):\n", " return response\n", " if isinstance(response, BaseMessage):\n", " content = response.content\n", " if isinstance(content, str):\n", " return content\n", " return str(content)\n", " return str(response)\n", "\n", "\n", "class ResearcherAgent:\n", " \"\"\"Answers factual questions by querying the LLM directly.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the researcher model.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0.1)\n", "\n", " def run(self, instruction: str, context: str = \"\") -> str:\n", " \"\"\"Return a factual answer.\"\"\"\n", " prompt = (\n", " \"You are a research assistant.\\n\\n\"\n", " f\"Context:\\n{context or 'None'}\\n\\n\"\n", " f\"Instruction: {instruction}\\n\\n\"\n", " \"Answer:\"\n", " )\n", " response = self.model.invoke(prompt)\n", " return _extract_text(response)\n", "\n", "\n", "class AnalystAgent:\n", " \"\"\"Performs analysis, comparisons and reasoning over provided information.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the analyst model.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0.1)\n", "\n", " def run(self, instruction: str, context: str = \"\") -> str:\n", " \"\"\"Return structured analytical insights.\"\"\"\n", " prompt = (\n", " \"You are an analytical assistant.\\n\\n\"\n", " f\"Context:\\n{context or 'None'}\\n\\n\"\n", " f\"Instruction: {instruction}\\n\\n\"\n", " \"Analysis:\"\n", " )\n", " response = self.model.invoke(prompt)\n", " return _extract_text(response)\n", "\n", "\n", "class WriterAgent:\n", " \"\"\"Synthesises research and analysis into polished prose.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the writer model.\"\"\"\n", " self.model = ChatMistralAI(model=\"mistral-small-latest\", temperature=0.7)\n", "\n", " def run(self, instruction: str, context: str = \"\") -> str:\n", " \"\"\"Return a well-written piece of text.\"\"\"\n", " prompt = (\n", " \"You are a skilled technical writer.\\n\\n\"\n", " f\"Context:\\n{context or 'None'}\\n\\n\"\n", " f\"Instruction: {instruction}\\n\\n\"\n", " \"Write:\"\n", " )\n", " response = self.model.invoke(prompt)\n", " return _extract_text(response)\n", "\n", "\n", "class PlannerAgent:\n", " \"\"\"Decomposes a high-level goal into an ordered list of specialist sub-tasks.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize the planner model with structured output.\"\"\"\n", " self.model = ChatMistralAI(\n", " model=\"mistral-small-latest\", temperature=0.1,\n", " ).with_structured_output(Plan)\n", "\n", " def plan(self, goal: str) -> Plan:\n", " \"\"\"Generate a Plan for the given goal.\"\"\"\n", " system_prompt = (\n", " \"You are a planner agent that breaks down a high-level goal into a structured \"\n", " \"execution plan. Available agents: \"\n", " \"'researcher' (facts), 'analyst' (reasoning/comparison), \"\n", " \"'writer' (final synthesis). \"\n", " \"Each step must include id, agent, instruction, and depends_on.\"\n", " )\n", " messages = [SystemMessage(content=system_prompt), HumanMessage(content=goal)]\n", " raw_plan = self.model.invoke(messages)\n", "\n", " if isinstance(raw_plan, Plan):\n", " return raw_plan\n", " if isinstance(raw_plan, BaseModel):\n", " return Plan.model_validate(raw_plan.model_dump())\n", " if isinstance(raw_plan, dict):\n", " return Plan.model_validate(raw_plan)\n", "\n", " msg = \"Planner returned an unsupported output type.\"\n", " raise TypeError(msg)\n", "\n", "\n", "class OrchestratorAgent:\n", " \"\"\"Coordinates planning and execution across specialist agents.\"\"\"\n", "\n", " def __init__(self) -> None:\n", " \"\"\"Initialize all agents.\"\"\"\n", " self.planner = PlannerAgent()\n", " self.researcher = ResearcherAgent()\n", " self.analyst = AnalystAgent()\n", " self.writer = WriterAgent()\n", "\n", " def _get_agent(self, agent_name: str) -> ResearcherAgent | AnalystAgent | WriterAgent:\n", " \"\"\"Return the specialist agent by name.\"\"\"\n", " if agent_name == \"researcher\":\n", " return self.researcher\n", " if agent_name == \"analyst\":\n", " return self.analyst\n", " if agent_name == \"writer\":\n", " return self.writer\n", " msg = f\"Unknown agent: {agent_name}\"\n", " raise ValueError(msg)\n", "\n", " def _resolve_context(self, depends_on: list[str], results: dict[str, str]) -> str:\n", " \"\"\"Build context from dependency outputs.\"\"\"\n", " parts: list[str] = []\n", " for step_id in depends_on:\n", " if step_id not in results:\n", " msg = f\"Missing dependency output for step '{step_id}'.\"\n", " raise ValueError(msg)\n", " parts.append(f\"=== {step_id} output ===\\n{results[step_id]}\")\n", " return \"\\n\\n\".join(parts)\n", "\n", " def run(self, goal: str) -> dict[str, object]:\n", " \"\"\"Execute the full multi-agent pipeline.\"\"\"\n", " plan = self.planner.plan(goal)\n", "\n", " results: dict[str, str] = {}\n", " log: list[str] = []\n", "\n", " for step in plan.steps:\n", " context = self._resolve_context(step.depends_on, results)\n", " agent = self._get_agent(step.agent)\n", " output = agent.run(step.instruction, context)\n", " results[step.id] = output\n", " log.append(f\"Step {step.id} output: {output[:100]}...\")\n", "\n", " writer_step_id = next(\n", " (s.id for s in reversed(plan.steps) if s.agent == \"writer\" and s.id in results),\n", " None,\n", " )\n", " final_output = results[writer_step_id] if writer_step_id else list(results.values())[-1]\n", "\n", " return {\n", " \"goal\": goal,\n", " \"plan\": plan,\n", " \"results\": results,\n", " \"final_output\": final_output,\n", " \"log\": log,\n", " }\n", "\n", "\n", "orchestrator = OrchestratorAgent()\n", "output = orchestrator.run(\n", " \"Compare supervised vs unsupervised learning and write a short summary\",\n", ")\n", "\n", "plan_result = output.get(\"plan\")\n", "plan_steps = plan_result.steps if isinstance(plan_result, Plan) else []\n", "\n", "print(\"\\n\" + \"=\" * 70)\n", "print(f\"🎯 Goal: {output['goal']}\")\n", "print(f\"\\nπŸ“‹ Plan ({len(plan_steps)} steps):\")\n", "for step in plan_steps:\n", " deps = f\" (depends on: {step.depends_on})\" if step.depends_on else \"\"\n", " print(f\" [{step.id}] {step.agent.upper()}: {step.instruction[:60]}...{deps}\")\n", "\n", "print(\"\\nπŸ“ Execution Log:\")\n", "log_entries = output.get(\"log\", [])\n", "if isinstance(log_entries, list):\n", " for entry in log_entries:\n", " print(f\" {entry}\")\n", "\n", "print(\"\\n✍️ Final Output:\\n\")\n", "print(output[\"final_output\"])\n" ] }, { "cell_type": "code", "execution_count": null, "id": "ca675f78", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "studies (3.13.9)", "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.13.9" } }, "nbformat": 4, "nbformat_minor": 5 }