Lab 05 · Live console

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 05 helper: the Northwind agent with one fast tool and one slow one.

The console imports this, so live_console.py is only about streaming.
"""

import time
from typing import Any

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.language_models import BaseChatModel
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.config import get_stream_writer

from model_setup import get_model

INPUTS = {"messages": [{"role": "user", "content": "Where is order A-100 right now?"}]}

@tool
def get_order_status(order_id: str) -> str:
    """Look up the recorded status and ETA of a Northwind order, such as A-100.

    This is the warehouse record. It does not know where the parcel physically is.
    """
    return f"Order {order_id} is shipped via DHL; ETA 2026-09-15."

@tool
def check_shipment(order_id: str) -> str:
    """Ask the carrier where a parcel physically is right now. Slow: takes a few seconds."""
    writer = get_stream_writer()
    for pct, message in [(10, "contacting DHL"), (60, "reading scans"), (100, "done")]:
        writer({"type": "progress", "tool": "check_shipment", "pct": pct, "message": message})
        time.sleep(0.4)
    return f"Shipment {order_id}: out for delivery in Berlin (last scan 06:12)."

def build_agent(
    model: BaseChatModel | None = None, checkpointer: BaseCheckpointSaver | None = None
) -> Any:
    """Build the Northwind agent. Pass a model to override the Azure OpenAI default."""
    return create_agent(
        model=model or get_model(),
        tools=[get_order_status, check_shipment],
        system_prompt="You are the Northwind support assistant. Be brief. Always use a tool.",
        checkpointer=checkpointer,
    )

"""Lab 05: a live console for the Northwind support agent.

One `agent.stream(...)` call feeds three channels: tool announcements from `updates`,
progress from `custom`, and the answer from `messages`, token by token.

Run:  python live_console.py
"""

from typing import Any

from langchain.messages import ToolMessage

from northwind import INPUTS, build_agent

STREAM_MODES = ["updates", "messages", "custom"]

def emit(line: str, lines: list[str]) -> None:
    """Print a trace line right away and keep it for the returned transcript."""
    print(line, flush=True)
    lines.append(line)

def show_tool_calls(data: dict[str, Any], lines: list[str]) -> None:
    """Announce the tools the model just asked for (from the `updates` channel)."""
    # The message is complete here, so the arguments are real JSON. The `messages`
    # chunks carry half-parsed `tool_call_chunks` instead.
    if "model" in data:
        for call in data["model"]["messages"][-1].tool_calls:
            emit(f"[tool] {call['name']}({call['args']})", lines)

def show_progress(data: dict[str, Any], lines: list[str]) -> None:
    """Render a progress event the slow tool wrote (from the `custom` channel)."""
    emit(f"[..] {data['pct']:>3}% {data['tool']}: {data['message']}", lines)

def show_message(data: tuple[Any, dict[str, Any]], lines: list[str], answer: list[str]) -> None:
    """Render a tool result, or print one answer token inline (from the `messages` channel)."""
    message, metadata = data
    if isinstance(message, ToolMessage):
        emit(f"[done] {message.name} -> {message.text}", lines)
    elif metadata["langgraph_node"] == "model" and message.text:
        if not answer:
            print("[answer] ", end="", flush=True)
        answer.append(message.text)
        print(message.text, end="", flush=True)

def run_console(agent: Any) -> list[str]:
    """Render one streamed run to the terminal and return the lines it wrote."""
    lines: list[str] = []
    answer: list[str] = []

    for part in agent.stream(INPUTS, stream_mode=STREAM_MODES, version="v2"):
        kind, data = part["type"], part["data"]
        if kind == "updates":
            show_tool_calls(data, lines)
        elif kind == "custom":
            show_progress(data, lines)
        elif kind == "messages":
            show_message(data, lines, answer)

    print()
    lines.append("[answer] " + "".join(answer).strip())
    return lines

if __name__ == "__main__":
    transcript = run_console(build_agent())
    print(f"\n({len(transcript)} transcript lines returned)")

[tool] get_order_status({'order_id': 'A-100'})
[tool] check_shipment({'order_id': 'A-100'})
[done] get_order_status -> Order A-100 is shipped via DHL; ETA 2026-09-15.
[..]  10% check_shipment: contacting DHL
[..]  60% check_shipment: reading scans
[..] 100% check_shipment: done
[done] check_shipment -> Shipment A-100: out for delivery in Berlin (last scan 06:12).
[answer] Order A-100 — warehouse: shipped via DHL (ETA 2026-09-15). Carrier (physical location): out for delivery in Berlin; last scan 06:12. Want me to check again later?

(8 transcript lines returned)