Lab 02 · Northwind support agent v0

langchain==1.4.2
langchain-core==1.6.4
langchain-openai==1.6.2
python-dotenv==1.2.3
langgraph==1.2.11
langgraph-checkpoint==4.2.0
pydantic==2.13.5

# Azure OpenAI settings. Your instructor gives you these three values.
AZURE_OPENAI_ENDPOINT=https://<your-resource>.openai.azure.com/openai/v1
AZURE_OPENAI_API_KEY=<paste-your-key-here>
AZURE_OPENAI_DEPLOYMENT=<your-chat-deployment-name>

# Only used when your endpoint does NOT end in /openai/v1.
OPENAI_API_VERSION=2025-04-01-preview

"""Azure OpenAI model setup shared by every lab.

Import it in lab code as:  from model_setup import get_model, get_embeddings
"""

import asyncio
import os
import sys

from dotenv import load_dotenv
from langchain_core.embeddings import Embeddings
from langchain_core.language_models import BaseChatModel

# Read AZURE_OPENAI_* values from the .env file next to your lab code.
load_dotenv()

# Windows consoles default to cp1252 and crash on the curly quotes models love to print.
for _stream in (sys.stdout, sys.stderr):
    if hasattr(_stream, "reconfigure"):
        _stream.reconfigure(encoding="utf-8", errors="replace")

def _setting(name: str) -> str:
    """Return a required setting from .env, or stop with a readable message."""
    value = os.getenv(name, "").strip()
    if not value:
        sys.exit(f"Missing {name} in your .env file. Open .env and fill it in.")
    return value


def _uses_v1_api(endpoint: str) -> bool:
    """True for the OpenAI-compatible Azure endpoint that ends in /openai/v1."""
    return endpoint.rstrip("/").endswith("/openai/v1")

def get_model(deployment: str | None = None, **kwargs) -> BaseChatModel:
    """Return a chat model for your Azure OpenAI deployment."""
    endpoint = _setting("AZURE_OPENAI_ENDPOINT")
    api_key = _setting("AZURE_OPENAI_API_KEY")
    name = deployment or _setting("AZURE_OPENAI_DEPLOYMENT")

    if _uses_v1_api(endpoint):
        from langchain_openai import ChatOpenAI

        return ChatOpenAI(model=name, base_url=endpoint, api_key=api_key, **kwargs)

    from langchain_openai import AzureChatOpenAI

    return AzureChatOpenAI(
        azure_deployment=name,
        azure_endpoint=endpoint,
        api_key=api_key,
        api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
        **kwargs,
    )

def get_embeddings(deployment: str | None = None) -> Embeddings:
    """Return an embeddings model for your Azure OpenAI embeddings deployment."""
    endpoint = _setting("AZURE_OPENAI_ENDPOINT")
    api_key = _setting("AZURE_OPENAI_API_KEY")
    name = deployment or _setting("AZURE_OPENAI_EMBEDDING_DEPLOYMENT")

    if _uses_v1_api(endpoint):
        from langchain_openai import OpenAIEmbeddings

        return OpenAIEmbeddings(model=name, base_url=endpoint, api_key=api_key)

    from langchain_openai import AzureOpenAIEmbeddings

    return AzureOpenAIEmbeddings(
        azure_deployment=name,
        azure_endpoint=endpoint,
        api_key=api_key,
        api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
    )

def use_stable_event_loop() -> None:
    """On Windows, use the Selector event loop so async MCP code exits cleanly."""
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

"""Lab 02: Northwind support agent v0.

Four tools, a custom state, a per-run context and a typed Resolution answer.

Run:  python support_agent.py
"""

import re
from dataclasses import dataclass
from typing import Any, Literal, NotRequired

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import (
    ModelCallLimitMiddleware,
    ToolCallRequest,
    ToolErrorMiddleware,
)
from langchain.agents.structured_output import ToolStrategy
from langchain.messages import ToolMessage
from langchain.tools import ToolException, ToolRuntime, tool
from langchain_core.language_models import BaseChatModel
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from pydantic import BaseModel, Field

from model_setup import get_model

ORDERS = {
    "A-100": {"status": "shipped via DHL", "eta": "2026-09-15", "total": 249.00,
              "items": "Northwind Router X2"},
    "A-101": {"status": "processing", "eta": "2026-09-18", "total": 1299.50,
              "items": "Northwind Edge Gateway"},
    "A-102": {"status": "delivered by UPS", "eta": "2026-09-02", "total": 89.99,
              "items": "Northwind Sensor Pack"},
}
FAQ = {
    "returns-window": "Hardware can be returned within 30 days of delivery for a full refund.",
    "shipping-times": "Standard shipping is 3-5 business days; express is next business day.",
    "warranty": "All Northwind hardware carries a 24-month limited warranty.",
    "cancellation": "Orders can be cancelled free of charge while they are still processing.",
}
TIER_ENTITLEMENTS = {
    "standard": "email support, next-business-day response, paid return shipping",
    "gold": "24/5 support, 4-hour response, free return shipping, advance replacement",
    "enterprise": "24/7 support, 1-hour response, free returns, a named CSM",
}
ORDER_ID_RE = re.compile(r"^[A-Z]-\d{3}$")

@dataclass
class SupportContext:
    """Per-run, read-only facts about who we are serving. Supplied by the application."""

    customer_id: str
    tier: Literal["standard", "gold", "enterprise"] = "standard"
    locale: str = "en-US"


class SupportState(AgentState):
    """Agent state plus the escalation facts our tools write back into the loop."""

    escalated: NotRequired[bool]
    escalation_reason: NotRequired[str]
    ticket_id: NotRequired[str]

class Resolution(BaseModel):
    """The structured outcome of a support conversation, for the CRM and for evals."""

    answer: str = Field(description="The customer-facing reply, 1-3 sentences.")
    intent: Literal["order_status", "faq", "escalation", "other"] = Field(
        description="What the customer actually wanted."
    )
    order_id: str | None = Field(default=None, description="Order id, or null if none applies.")
    needs_human: bool = Field(default=False, description="True when a human must take over.")
    citations: list[str] = Field(
        default_factory=list,
        description="Ids of the facts used, e.g. 'order:A-100' or 'faq:returns-window'.",
    )

class OrderLookup(BaseModel):
    """Arguments for looking up an order."""

    order_id: str = Field(description="Order id in the form 'A-123'.")
    include_items: bool = Field(default=False, description="Also list the ordered items.")


@tool("lookup_order", args_schema=OrderLookup)
def lookup_order(order_id: str, include_items: bool = False) -> str:
    """Look up a Northwind order. Use whenever the customer mentions an order id."""
    order_id = order_id.strip().upper()
    if not ORDER_ID_RE.match(order_id):
        raise ToolException(f"{order_id!r} is not a valid order id; expected the form 'A-123'.")
    order = ORDERS.get(order_id)
    if order is None:
        # A business miss is a normal result, not an exception.
        return f"No order found with id {order_id}."
    items = f" Items: {order['items']}." if include_items else ""
    return (
        f"Order {order_id} is {order['status']}; ETA {order['eta']}; "
        f"total {order['total']:.2f} USD.{items} [order:{order_id}]"
    )

@tool(parse_docstring=True)
def search_faq(query: str, top_k: int = 2) -> str:
    """Search the Northwind help centre for policy answers, not for a specific order.

    Args:
        query: The customer's question, in their own words.
        top_k: How many help-centre passages to return.
    """
    words = re.findall(r"[a-z]{4,}", query.lower())
    hits = [
        f"{text} [faq:{key}]"
        for key, text in FAQ.items()
        if any(word in f"{key} {text}".lower() for word in words)
    ]
    return "\n".join(hits[:top_k]) or "No help-centre article matched that question."

@tool
def check_entitlement(runtime: ToolRuntime[SupportContext, SupportState]) -> str:
    """Report the current customer's plan and what it includes. Takes no arguments."""
    ctx = runtime.context
    return (
        f"Customer {ctx.customer_id} is on the {ctx.tier} plan: "
        f"{TIER_ENTITLEMENTS[ctx.tier]}. [tier:{ctx.tier}]"
    )

@tool
def escalate_to_human(reason: str, runtime: ToolRuntime[SupportContext, SupportState]) -> Command:
    """Open a human ticket. Use when the customer asks for a person or a tool failure blocks you."""
    # One ticket per customer, so calling this twice cannot open a second one.
    ticket_id = f"ESC-{runtime.context.customer_id}"
    return Command(
        update={
            "escalated": True,
            "escalation_reason": reason,
            "ticket_id": ticket_id,
            # A tool result is a prompt: a final tool has to say that the work is over.
            "messages": [
                ToolMessage(
                    f"Escalated as ticket {ticket_id} ({reason}). This is the final action for "
                    "this request: do not call any other tool. Emit the Resolution now, with "
                    "needs_human=true and the ticket id in the answer.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

TOOLS = [lookup_order, search_faq, check_entitlement, escalate_to_human]

SYSTEM_PROMPT = (
    "You are the Northwind Cloud support agent. Ground every factual claim in a tool result: "
    "lookup_order for orders, search_faq for policy, check_entitlement before promising "
    "anything that depends on the customer's plan. Never invent order ids, dates or totals. "
    "Escalate when the customer asks for a human, and as soon as escalate_to_human returns, "
    "stop gathering facts. Finish every request with the Resolution schema, copying every "
    "[bracketed] id you relied on into its citations."
)

def on_tool_error(exc: Exception, request: ToolCallRequest) -> str | None:
    """Surface bad tool arguments to the model; let unexpected bugs halt the run."""
    if isinstance(exc, ToolException):
        return (
            f"Tool '{request.tool_call['name']}' rejected the call: {exc} "
            "Correct the arguments and call it again."
        )
    return None  # propagate: never send an unknown internal error to the model

def build_agent(
    model: BaseChatModel | None = None, checkpointer: BaseCheckpointSaver | None = None
) -> Any:
    """Build the support agent. Pass a model to override the Azure OpenAI default."""
    return create_agent(
        model=model or get_model(),
        tools=TOOLS,
        system_prompt=SYSTEM_PROMPT,
        middleware=[
            # The backstop: end the run cleanly if the model never converges.
            ModelCallLimitMiddleware(run_limit=8, exit_behavior="end"),
            ToolErrorMiddleware(on_tool_error),
        ],
        response_format=ToolStrategy(
            Resolution,
            tool_message_content="Resolution recorded.",
            handle_errors=(
                "That Resolution was invalid. Re-emit it with every required field: "
                "answer, intent, order_id (or null), needs_human, citations."
            ),
        ),
        state_schema=SupportState,
        context_schema=SupportContext,
        checkpointer=checkpointer,
        name="northwind-support-v0",
    )

def tools_called(messages: list[Any]) -> list[str]:
    """Names of the real tools the model called (the Resolution tool is left out)."""
    names = [call["name"] for m in messages for call in getattr(m, "tool_calls", [])]
    return [name for name in names if name != Resolution.__name__]


if __name__ == "__main__":
    agent = build_agent(checkpointer=InMemorySaver())
    config = {"configurable": {"thread_id": "northwind-lab-02"}}
    context = SupportContext(customer_id="C-4471", tier="gold")

    for turn in [
        "Where is order A-100 and what did it cost?",
        "And what about my other order, AB-1000?",
        "It arrived damaged. What is your return window, and do I pay return shipping?",
        "That is not good enough. Put me through to a person.",
    ]:
        before = len(agent.get_state(config).values.get("messages", []))
        result = agent.invoke(
            {"messages": [{"role": "user", "content": turn}]}, config=config, context=context
        )
        new_messages = result["messages"][before:]
        errors = [m.name for m in new_messages if getattr(m, "status", None) == "error"]
        print(f"\ncustomer  : {turn}")
        print(f"tools     : {', '.join(tools_called(new_messages)) or 'none'}")
        if errors:
            print(f"tool error: {', '.join(errors)} (the model received the error message instead of a crash)")
        resolution = result.get("structured_response")
        if resolution is None:
            print("answer    : (no Resolution: the run hit the model-call limit)")
            continue
        print(f"answer    : {resolution.answer}")
        print(f"metadata  : intent={resolution.intent} needs_human={resolution.needs_human} "
              f"citations={resolution.citations}")

    values = agent.get_state(config).values
    print(f"\nthread state: escalated={values.get('escalated', False)} "
          f"ticket_id={values.get('ticket_id', '(no escalation happened)')}")

customer  : Where is order A-100 and what did it cost?
tools     : lookup_order
answer    : Order A-100 was shipped via DHL with an ETA of 2026-09-15; the total cost was 249.00 USD. The order includes: Northwind Router X2.
metadata  : intent=order_status needs_human=False citations=['order:A-100']

customer  : And what about my other order, AB-1000?
tools     : lookup_order
tool error: lookup_order (the model received the error message instead of a crash)
answer    : I couldn’t look up “AB-1000” because the order ID must be in the form “A-123” and the lookup rejected that ID; please confirm the correct order ID (for example “A-123”) and I’ll check its status right away.
metadata  : intent=order_status needs_human=False citations=['order:A-100']

customer  : It arrived damaged. What is your return window, and do I pay return shipping?
tools     : search_faq, check_entitlement
answer    : Sorry this arrived damaged. Hardware can be returned within 30 days of delivery for a full refund, and since your account is on the Gold plan you have free return shipping and an advance replacement—would you like me to start the return for order A-100?
metadata  : intent=faq needs_human=False citations=['faq:returns-window', 'tier:gold', 'order:A-100']

customer  : That is not good enough. Put me through to a person.
tools     : escalate_to_human
answer    : I’ve escalated your damaged order A-100 to our specialist team and opened ticket ESC-C-4471; a human agent will take over (your Gold plan includes a 4-hour response).
metadata  : intent=escalation needs_human=True citations=['order:A-100', 'faq:returns-window', 'tier:gold', 'ESC-C-4471']

thread state: escalated=True ticket_id=ESC-C-4471