{ "cells": [ { "cell_type": "code", "execution_count": null, "id": "4b2cf5f5", "metadata": {}, "outputs": [], "source": [ "import asyncio\n", "import json\n", "import os\n", "from typing import Annotated, Any, Never\n", "\n", "from agent_framework import (\n", " AgentExecutor,\n", " AgentExecutorRequest,\n", " AgentExecutorResponse,\n", " Message,\n", " WorkflowBuilder,\n", " WorkflowContext,\n", " executor,\n", " tool,\n", ")\n", "from agent_framework.foundry import FoundryChatClient\n", "from azure.identity import AzureCliCredential\n", "from dotenv import load_dotenv\n", "from IPython.display import HTML, display\n", "from pydantic import BaseModel\n", "\n", "print(\"✅ All imports successful!\")\n" ] }, { "cell_type": "markdown", "id": "001c224e", "metadata": {}, "source": [ "## Step 1: Define Pydantic Models for Structured Outputs\n", "\n", "These models define the **schema** that agents will return. Using `response_format` with Pydantic ensures:\n", "- ✅ Type-safe data extraction\n", "- ✅ Automatic validation\n", "- ✅ No parsing errors from free-text responses\n", "- ✅ Easy conditional routing based on fields" ] }, { "cell_type": "code", "execution_count": null, "id": "6c2ef582", "metadata": {}, "outputs": [], "source": [ "class BookingCheckResult(BaseModel):\n", " \"\"\"Result from checking hotel availability at a destination.\"\"\"\n", "\n", " destination: str\n", " has_availability: bool\n", " message: str\n", "\n", "\n", "class AlternativeResult(BaseModel):\n", " \"\"\"Suggested alternative destination when no rooms available.\"\"\"\n", "\n", " alternative_destination: str\n", " reason: str\n", "\n", "\n", "class BookingConfirmation(BaseModel):\n", " \"\"\"Booking suggestion when rooms are available.\"\"\"\n", "\n", " destination: str\n", " action: str\n", " message: str\n", "\n", "\n", "print(\"✅ Pydantic models defined:\")\n", "print(\" - BookingCheckResult (availability check)\")\n", "print(\" - AlternativeResult (alternative suggestion)\")\n", "print(\" - BookingConfirmation (booking confirmation)\")" ] }, { "cell_type": "markdown", "id": "48423ecc", "metadata": {}, "source": [ "## Step 2: Create the Hotel Booking Tool\n", "\n", "This tool is what the **availability_agent** will call to check if rooms are available. We use the `@ai_function` decorator to:\n", "- Convert a Python function into an AI-callable tool\n", "- Automatically generate JSON schema for the LLM\n", "- Handle parameter validation\n", "- Enable automatic invocation by agents\n", "\n", "For this demo:\n", "- **Stockholm, Seattle, Tokyo, London, Amsterdam** → Have rooms ✅\n", "- **All other cities** → No rooms ❌" ] }, { "cell_type": "code", "execution_count": null, "id": "aad7e7ec", "metadata": {}, "outputs": [], "source": [ "@tool(description=\"Check hotel room availability for a destination city\")\n", "def hotel_booking(destination: Annotated[str, \"The destination city to check for hotel rooms\"]) -> str:\n", " \"\"\"\n", " Simulates checking hotel room availability.\n", "\n", " Returns JSON string with availability status.\n", " \"\"\"\n", " display(\n", " HTML(f\"\"\"\n", "
\n", " 🔍 Tool Invoked: hotel_booking(\"{destination}\")\n", "
\n", " \"\"\")\n", " )\n", "\n", " # Simulate availability check\n", " cities_with_rooms = [\"stockholm\", \"seattle\", \"tokyo\", \"london\", \"amsterdam\"]\n", " has_rooms = destination.lower() in cities_with_rooms\n", "\n", " result = {\"has_availability\": has_rooms, \"destination\": destination}\n", "\n", " return json.dumps(result)\n", "\n", "\n", "print(\"✅ hotel_booking tool created with @tool decorator\")" ] }, { "cell_type": "markdown", "id": "134c54b0", "metadata": {}, "source": [ "## Step 3: Define Condition Functions for Routing\n", "\n", "These functions inspect the agent's response and determine which path to take in the workflow.\n", "\n", "**Key Pattern:**\n", "1. Check if the message is `AgentExecutorResponse`\n", "2. Parse the structured output (Pydantic model)\n", "3. Return `True` or `False` to control routing\n", "\n", "The workflow will evaluate these conditions on **edges** to decide which executor to invoke next." ] }, { "cell_type": "code", "execution_count": null, "id": "6960edd1", "metadata": {}, "outputs": [], "source": [ "def has_availability_condition(message: Any) -> bool:\n", " \"\"\"\n", " Condition for routing when hotels ARE available.\n", " \n", " Returns True if the destination has hotel rooms.\n", " \"\"\"\n", " if not isinstance(message, AgentExecutorResponse):\n", " return True # Default to True if unexpected type\n", "\n", " try:\n", " result = BookingCheckResult.model_validate_json(message.agent_run_response.text)\n", "\n", " display(\n", " HTML(f\"\"\"\n", "
\n", " ✅ Condition Check: has_availability = {result.has_availability} for {result.destination}\n", "
\n", " \"\"\")\n", " )\n", "\n", " return result.has_availability\n", " except Exception as e:\n", " display(\n", " HTML(f\"\"\"\n", "
\n", " ⚠️ Error: {str(e)}\n", "
\n", " \"\"\")\n", " )\n", " return False\n", "\n", "\n", "def no_availability_condition(message: Any) -> bool:\n", " \"\"\"\n", " Condition for routing when hotels are NOT available.\n", " \n", " Returns True if the destination has no hotel rooms.\n", " \"\"\"\n", " if not isinstance(message, AgentExecutorResponse):\n", " return False\n", "\n", " try:\n", " result = BookingCheckResult.model_validate_json(message.agent_run_response.text)\n", "\n", " display(\n", " HTML(f\"\"\"\n", "
\n", " ❌ Condition Check: no_availability for {result.destination}\n", "
\n", " \"\"\")\n", " )\n", "\n", " return not result.has_availability\n", " except Exception as e:\n", " return False\n", "\n", "\n", "print(\"✅ Condition functions defined:\")\n", "print(\" - has_availability_condition (routes when rooms exist)\")\n", "print(\" - no_availability_condition (routes when no rooms)\")" ] }, { "cell_type": "markdown", "id": "9dc783ba", "metadata": {}, "source": [ "## Step 4: Create Custom Display Executor\n", "\n", "Executors are workflow components that perform transformations or side effects. We use the `@executor` decorator to create a custom executor that displays the final result.\n", "\n", "**Key Concepts:**\n", "- `@executor(id=\"...\")` - Registers a function as a workflow executor\n", "- `WorkflowContext[Never, str]` - Type hints for input/output\n", "- `ctx.yield_output(...)` - Yields the final workflow result" ] }, { "cell_type": "code", "execution_count": null, "id": "c67f6b55", "metadata": {}, "outputs": [], "source": [ "@executor(id=\"display_result\")\n", "async def display_result(response: AgentExecutorResponse, ctx: WorkflowContext[Never, str]) -> None:\n", " \"\"\"\n", " Display the final result as workflow output.\n", " \n", " This executor receives the final agent response and yields it as the workflow output.\n", " \"\"\"\n", " display(\n", " HTML(\"\"\"\n", "
\n", " 📤 Display Executor: Yielding workflow output\n", "
\n", " \"\"\")\n", " )\n", "\n", " await ctx.yield_output(response.agent_run_response.text)\n", "\n", "\n", "print(\"✅ display_result executor created with @executor decorator\")" ] }, { "cell_type": "markdown", "id": "7de6eb90", "metadata": {}, "source": [ "## Step 5: Load Environment Variables\n", "\n", "Configure the LLM client. This example works with:\n", "- **GitHub Models** (Free tier with GitHub token)\n", "- **Azure OpenAI**\n", "- **OpenAI**" ] }, { "cell_type": "code", "execution_count": null, "id": "1e8f0d88", "metadata": {}, "outputs": [], "source": [ "# Load environment variables\n", "load_dotenv()\n", "\n", "# Configure the Microsoft Foundry provider with keyless authentication\n", "provider = FoundryChatClient(\n", " project_endpoint=os.environ[\"AZURE_AI_PROJECT_ENDPOINT\"],\n", " model=os.environ[\"AZURE_AI_MODEL_DEPLOYMENT_NAME\"],\n", " credential=AzureCliCredential(),\n", ")\n" ] }, { "cell_type": "markdown", "id": "3fc61fe7", "metadata": {}, "source": [ "## Step 6: Create AI Agents with Structured Outputs\n", "\n", "We create **three specialized agents**, each wrapped in an `AgentExecutor`:\n", "\n", "1. **availability_agent** - Checks hotel availability using the tool\n", "2. **alternative_agent** - Suggests alternative cities (when no rooms)\n", "3. **booking_agent** - Encourages booking (when rooms available)\n", "\n", "**Key Features:**\n", "- `tools=[hotel_booking]` - Provides the tool to the agent\n", "- `response_format=PydanticModel` - Forces structured JSON output\n", "- `AgentExecutor(..., id=\"...\")` - Wraps agent for workflow use" ] }, { "cell_type": "code", "execution_count": null, "id": "66466dda", "metadata": {}, "outputs": [], "source": [ "# Agent 1: Check availability with tool\n", "availability_agent = AgentExecutor(\n", " provider.as_agent(\n", " name=\"availability-agent\",\n", " instructions=(\n", " \"You are a hotel booking assistant that checks room availability. \"\n", " \"Use the hotel_booking tool to check if rooms are available at the destination. \"\n", " \"Return JSON with fields: destination (string), has_availability (bool), and message (string). \"\n", " \"The message should summarize the availability status.\"\n", " ),\n", " tools=[hotel_booking],\n", " default_options={\"response_format\": BookingCheckResult},\n", " ),\n", " id=\"availability_agent\",\n", ")\n", "\n", "# Agent 2: Suggest alternative (when no rooms)\n", "alternative_agent = AgentExecutor(\n", " provider.as_agent(\n", " name=\"alternative-agent\",\n", " instructions=(\n", " \"You are a helpful travel assistant. When a user cannot find hotels in their requested city, \"\n", " \"suggest an alternative nearby city that has availability. \"\n", " \"Return JSON with fields: alternative_destination (string) and reason (string). \"\n", " \"Make your suggestion sound appealing and helpful.\"\n", " ),\n", " default_options={\"response_format\": AlternativeResult},\n", " ),\n", " id=\"alternative_agent\",\n", ")\n", "\n", "# Agent 3: Suggest booking (when rooms available)\n", "booking_agent = AgentExecutor(\n", " provider.as_agent(\n", " name=\"booking-agent\",\n", " instructions=(\n", " \"You are a booking assistant. The user has found available hotel rooms. \"\n", " \"Encourage them to book by highlighting the destination's appeal. \"\n", " \"Return JSON with fields: destination (string), action (string), and message (string). \"\n", " \"The action should be 'book_now' and message should be encouraging.\"\n", " ),\n", " default_options={\"response_format\": BookingConfirmation},\n", " ),\n", " id=\"booking_agent\",\n", ")\n", "\n", "display(\n", " HTML(\"\"\"\n", "
\n", " ✅ Created 3 Agents:\n", " \n", "
\n", "\"\"\")\n", ")\n" ] }, { "cell_type": "markdown", "id": "7879a0cb", "metadata": {}, "source": [ "## Step 7: Build the Workflow with Conditional Edges\n", "\n", "Now we use `WorkflowBuilder` to construct the graph with conditional routing:\n", "\n", "**Workflow Structure:**\n", "```\n", "availability_agent (START)\n", " ↓\n", " Evaluate conditions\n", " ↙ ↘\n", "[no_availability] [has_availability]\n", " ↓ ↓\n", "alternative_agent booking_agent\n", " ↓ ↓\n", " display_result ←───┘\n", "```\n", "\n", "**Key Methods:**\n", "- `.set_start_executor(...)` - Sets the entry point\n", "- `.add_edge(from, to, condition=...)` - Adds conditional edge\n", "- `.build()` - Finalizes the workflow" ] }, { "cell_type": "code", "execution_count": null, "id": "90bb29dd", "metadata": {}, "outputs": [], "source": [ "# Build the workflow with conditional routing\n", "workflow = (\n", " WorkflowBuilder(\n", " start_executor=availability_agent,\n", " output_executors=[display_result],\n", " )\n", " # NO AVAILABILITY PATH\n", " .add_edge(availability_agent, alternative_agent, condition=no_availability_condition)\n", " .add_edge(alternative_agent, display_result)\n", " # HAS AVAILABILITY PATH\n", " .add_edge(availability_agent, booking_agent, condition=has_availability_condition)\n", " .add_edge(booking_agent, display_result)\n", " .build()\n", ")\n", "\n", "display(\n", " HTML(\"\"\"\n", "
\n", "

✅ Workflow Built Successfully!

\n", "

\n", " Conditional Routing:
\n", " • If NO availability → alternative_agent → display_result
\n", " • If availability → booking_agent → display_result\n", "

\n", "
\n", "\"\"\")\n", ")" ] }, { "cell_type": "markdown", "id": "0a3ad845", "metadata": {}, "source": [ "## Step 8: Run Test Case 1 - City WITHOUT Availability (Paris)\n", "\n", "Let's test the **no availability** path by requesting hotels in Paris (which has no rooms in our simulation)." ] }, { "cell_type": "code", "execution_count": null, "id": "af1538cd", "metadata": {}, "outputs": [], "source": [ "display(\n", " HTML(\"\"\"\n", "
\n", "

🧪 TEST CASE 1: Paris (No Availability)

\n", "

Expected workflow path: availability_agent → alternative_agent → display_result

\n", "
\n", "\"\"\")\n", ")\n", "\n", "# Create request for Paris\n", "request_paris = AgentExecutorRequest(\n", " messages=[Message(role=\"user\", text=\"I want to book a hotel in Paris\")], should_respond=True\n", ")\n", "\n", "# Run the workflow\n", "events_paris = await workflow.run(request_paris)\n", "outputs_paris = events_paris.get_outputs()\n", "\n", "# Display results\n", "if outputs_paris:\n", " result_paris = AlternativeResult.model_validate_json(outputs_paris[0])\n", "\n", " display(\n", " HTML(f\"\"\"\n", "
\n", "

🏆 WORKFLOW RESULT (Paris)

\n", "
\n", "

Status: ❌ No rooms in Paris

\n", "

Alternative Suggestion: 🏨 {result_paris.alternative_destination}

\n", "

Reason: {result_paris.reason}

\n", "
\n", "
\n", " \"\"\")\n", " )" ] }, { "cell_type": "markdown", "id": "408a3f60", "metadata": {}, "source": [ "## Step 9: Run Test Case 2 - City WITH Availability (Stockholm)\n", "\n", "Now let's test the **availability** path by requesting hotels in Stockholm (which has rooms in our simulation)." ] }, { "cell_type": "code", "execution_count": null, "id": "e1471000", "metadata": {}, "outputs": [], "source": [ "display(\n", " HTML(\"\"\"\n", "
\n", "

🧪 TEST CASE 2: Stockholm (Has Availability)

\n", "

Expected workflow path: availability_agent → booking_agent → display_result

\n", "
\n", "\"\"\")\n", ")\n", "\n", "# Create request for Stockholm\n", "request_stockholm = AgentExecutorRequest(\n", " messages=[Message(role=\"user\", text=\"I want to book a hotel in Stockholm\")], should_respond=True\n", ")\n", "\n", "# Run the workflow\n", "events_stockholm = await workflow.run(request_stockholm)\n", "outputs_stockholm = events_stockholm.get_outputs()\n", "\n", "# Display results\n", "if outputs_stockholm:\n", " result_stockholm = BookingConfirmation.model_validate_json(outputs_stockholm[0])\n", "\n", " display(\n", " HTML(f\"\"\"\n", "
\n", "

🏆 WORKFLOW RESULT (Stockholm)

\n", "
\n", "

Status: ✅ Rooms Available!

\n", "

Destination: 🏨 {result_stockholm.destination}

\n", "

Action: {result_stockholm.action}

\n", "

Message: {result_stockholm.message}

\n", "
\n", "
\n", " \"\"\")\n", " )" ] }, { "cell_type": "markdown", "id": "a415537c", "metadata": {}, "source": [ "## Key Takeaways and Next Steps\n", "\n", "### ✅ What You've Learned:\n", "\n", "1. **WorkflowBuilder Pattern**\n", " - Use `.set_start_executor()` to define entry point\n", " - Use `.add_edge(from, to, condition=...)` for conditional routing\n", " - Call `.build()` to finalize the workflow\n", "\n", "2. **Conditional Routing**\n", " - Condition functions inspect `AgentExecutorResponse`\n", " - Parse structured outputs to make routing decisions\n", " - Return `True` to activate an edge, `False` to skip it\n", "\n", "3. **Tool Integration**\n", " - Use `@ai_function` to convert Python functions into AI tools\n", " - Agents call tools automatically when needed\n", " - Tools return JSON that agents can parse\n", "\n", "4. **Structured Outputs**\n", " - Use Pydantic models for type-safe data extraction\n", " - Set `response_format=MyModel` when creating agents\n", " - Parse responses with `Model.model_validate_json()`\n", "\n", "5. **Custom Executors**\n", " - Use `@executor(id=\"...\")` to create workflow components\n", " - Executors can transform data or perform side effects\n", " - Use `ctx.yield_output()` to produce workflow results\n", "\n", "### 🚀 Real-World Applications:\n", "\n", "- **Travel Booking**: Check availability, suggest alternatives, compare options\n", "- **Customer Service**: Route based on issue type, sentiment, priority\n", "- **E-commerce**: Check inventory, suggest alternatives, process orders\n", "- **Content Moderation**: Route based on toxicity scores, user flags\n", "- **Approval Workflows**: Route based on amount, user role, risk level\n", "- **Multi-stage Processing**: Route based on data quality, completeness\n", "\n", "### 📚 Next Steps:\n", "\n", "- Add more complex conditions (multiple criteria)\n", "- Implement loops with workflow state management\n", "- Add sub-workflows for reusable components\n", "- Integrate with real APIs (hotel booking, inventory systems)\n", "- Add error handling and fallback paths\n", "- Visualize workflows with the built-in visualization tools" ] } ], "metadata": { "kernelspec": { "display_name": ".venv (3.12.12)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.12" } }, "nbformat": 4, "nbformat_minor": 5 }