Lab 07 · Ticket triage pipeline

langchain==1.4.2
langchain-core==1.6.4
langchain-openai==1.6.2
python-dotenv==1.2.3
langgraph==1.2.11

# 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 07: the Northwind ticket triage pipeline as one LangGraph StateGraph.

Run:  python triage_pipeline.py
"""

import operator
from dataclasses import dataclass
from typing import Annotated, Any, Callable, Literal

from langchain_core.language_models import BaseChatModel
from langgraph.graph import END, START, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.runtime import Runtime
from langgraph.types import Command
from typing_extensions import TypedDict

from model_setup import get_model

CATEGORY_KEYWORDS = {
    "billing": ("invoice", "charge", "refund", "payment"),
    "technical": ("error", "crash", "timeout", "500", "bug"),
    "account": ("login", "password", "sso", "seat", "access"),
}
URGENT_WORDS = ("outage", "urgent", "asap", "block", "production")
PLAYBOOKS = {
    "billing": "Confirm the invoice number and say whether a refund was issued.",
    "technical": "Acknowledge the error, name the component, give the next step.",
    "account": "Verify identity first, then explain the access change.",
}

class TriageState(TypedDict, total=False):
    """Everything the pipeline knows about one ticket."""

    ticket: str
    category: str  # one key per classifier, so the parallel writes never collide
    urgency: str
    sentiment: str
    handler: str
    playbook: str
    draft: str
    attempts: int
    approved: bool
    feedback: Annotated[list[str], operator.add]  # one instruction per grading round


@dataclass
class TriageContext:
    """Per-run configuration: passed as `context=`, never stored in state."""

    tenant: str = "northwind"
    agent_name: str = "Ada"
    max_revisions: int = 2

def classify_category(state: TriageState) -> dict[str, Any]:
    """Keyword scoring: deterministic, free, and good enough for a first pass."""
    text = state["ticket"].lower()
    hits = {name: sum(w in text for w in words) for name, words in CATEGORY_KEYWORDS.items()}
    best = max(hits, key=lambda name: hits[name])
    return {"category": best if hits[best] else "technical"}


def classify_urgency(state: TriageState) -> dict[str, Any]:
    """Count urgent words: two or more is high, one is normal, none is low."""
    hits = sum(word in state["ticket"].lower() for word in URGENT_WORDS)
    return {"urgency": "high" if hits >= 2 else "normal" if hits else "low"}

def pick_handler(state: TriageState) -> str:
    """Churn risk goes to the account desk whatever the ticket is about."""
    if state["urgency"] == "high" and state["sentiment"] == "negative":
        return "account"
    return state["category"] if state["category"] in PLAYBOOKS else "technical"


def triage(state: TriageState) -> Command[Literal["billing", "technical", "account"]]:
    """Write the decision and jump to it in one atomic step."""
    handler = pick_handler(state)
    return Command(update={"handler": handler}, goto=handler)

def should_revise(
    state: TriageState, runtime: Runtime[TriageContext]
) -> Literal["draft_reply", "__end__"]:
    """Two exits: the grader approved, or the revision budget is spent."""
    if state["approved"] or state["attempts"] > runtime.context.max_revisions:
        return END
    return "draft_reply"

def make_model_nodes(model: BaseChatModel) -> tuple[Callable[..., dict[str, Any]], ...]:
    """Return the three nodes that call the model: sentiment, draft and grade."""

    def classify_sentiment(state: TriageState) -> dict[str, Any]:
        """Ask the model for the tone and normalise it to one of three words."""
        question = "Classify the customer's tone. One word: positive, neutral or negative."
        answer = model.invoke(f"{question}\n\n{state['ticket']}").text.lower()
        tones = ("negative", "positive", "neutral")
        return {"sentiment": next((tone for tone in tones if tone in answer), "neutral")}

    def draft_reply(state: TriageState, runtime: Runtime[TriageContext]) -> dict[str, Any]:
        """Write (or rewrite) the reply, following the playbook and any feedback."""
        context = runtime.context
        prompt = (
            f"You are {context.agent_name}, a support agent at {context.tenant}. "
            f"Write a short reply.\nPlaybook: {state['playbook']}\nTicket: {state['ticket']}\n"
            f"Reviewer feedback to address: {'; '.join(state['feedback'])}"
        )
        return {"draft": model.invoke(prompt).text.strip(), "attempts": state["attempts"] + 1}

    def grade_reply(state: TriageState) -> dict[str, Any]:
        """Approve the draft, or send it back with one concrete instruction."""
        prompt = (
            'Reply "APPROVE" if the draft follows the playbook and answers the ticket, '
            'otherwise "REVISE: <one concrete instruction>".\n'
            f"Playbook: {state['playbook']}\nDraft: {state['draft']}"
        )
        verdict = model.invoke(prompt).text.strip()
        if verdict.upper().startswith("APPROVE"):
            return {"approved": True}
        return {"approved": False, "feedback": [verdict.partition(":")[2].strip()]}

    return classify_sentiment, draft_reply, grade_reply

def build_graph(model: BaseChatModel | None = None) -> CompiledStateGraph:
    """Compile the pipeline. Pass a model to override the Azure OpenAI default."""
    classify_sentiment, draft_reply, grade_reply = make_model_nodes(model or get_model())

    builder = StateGraph(TriageState, context_schema=TriageContext)
    builder.add_node("classify_category", classify_category)
    builder.add_node("classify_urgency", classify_urgency)
    builder.add_node("classify_sentiment", classify_sentiment)
    builder.add_node("triage", triage)
    builder.add_node("draft_reply", draft_reply)
    builder.add_node("grade_reply", grade_reply)

    for classifier in ("classify_category", "classify_urgency", "classify_sentiment"):
        builder.add_edge(START, classifier)  # fan-out: one super-step, three nodes
        builder.add_edge(classifier, "triage")  # fan-in: triage waits for all three
    for desk, playbook in PLAYBOOKS.items():
        # each desk is one node whose whole job is to choose the playbook the draft follows
        builder.add_node(desk, lambda state, text=playbook: {"playbook": text})
        builder.add_edge(desk, "draft_reply")
    builder.add_edge("draft_reply", "grade_reply")
    builder.add_conditional_edges("grade_reply", should_revise, ["draft_reply", END])

    return builder.compile(name="northwind-triage")

SAMPLE_TICKET = (
    "We were charged twice for invoice INV-2291 this morning and the refund button returns a "
    "500 error. This is blocking our month-end close. Please fix it ASAP."
)


def short(value: Any, width: int = 70) -> str:
    """One-line preview of a state value."""
    text = str(value).replace("\n", " ")
    return text if len(text) <= width else text[: width - 3] + "..."


if __name__ == "__main__":
    graph = build_graph()
    print(graph.get_graph().draw_mermaid())

    inputs: TriageState = {"ticket": SAMPLE_TICKET, "attempts": 0, "feedback": []}
    result: dict[str, Any] = {}
    for mode, chunk in graph.stream(inputs, context=TriageContext(), stream_mode=["updates", "values"]):
        if mode == "values":
            result = chunk  # the full state after each step; the last one is the final state
            continue
        for node, update in chunk.items():
            fields = ", ".join(f"{key}={short(value, 50)}" for key, value in (update or {}).items())
            print(f"[{node}] {fields}")

    print(f"\n{result['handler']} desk · {result['attempts']} draft(s) · approved={result.get('approved')}")
    print(result["draft"])

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	classify_category(classify_category)
	classify_urgency(classify_urgency)
	classify_sentiment(classify_sentiment)
	triage(triage)
	draft_reply(draft_reply)
	grade_reply(grade_reply)
	billing(billing)
	technical(technical)
	account(account)
	__end__([<p>__end__</p>]):::last
	__start__ --> classify_category;
	__start__ --> classify_sentiment;
	__start__ --> classify_urgency;
	account --> draft_reply;
	billing --> draft_reply;
	classify_category --> triage;
	classify_sentiment --> triage;
	classify_urgency --> triage;
	draft_reply --> grade_reply;
	grade_reply -.-> __end__;
	grade_reply -.-> draft_reply;
	technical --> draft_reply;
	triage -.-> account;
	triage -.-> billing;
	triage -.-> technical;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

[classify_category] category=billing
[classify_urgency] urgency=high
[classify_sentiment] sentiment=negative
[triage] handler=account
[account] playbook=Verify identity first, then explain the access ...
[draft_reply] draft=Hi — I’m Ada. I can take care of this right awa..., attempts=1
[grade_reply] approved=True

account desk · 1 draft(s) · approved=True
Hi — I’m Ada. I can take care of this right away, but first I need to verify you.

Please confirm the account email or company name on the Northwind account and the last 4 digits of the payment method used for INV-2291 (and the approximate time the two charges hit). Once you reply, I will:

- Temporarily enable a backend refund override for your account so I can issue the refund immediately (this won’t change your normal permissions and will be revoked after the issue is resolved).
- Manually process the duplicate refund and escalate the 500 error to engineering so the refund button is fixed.

I’ll aim to issue the refund within 2 hours after verification and will keep you updated.