In this tutorial, we build a production-style Route Optimizer Agent for a logistics dispatch center using the latest LangChain agent APIs. We design a tool-driven workflow in which the agent reliably computes distances, ETAs, and optimal routes rather than guessing, and we enforce structured outputs to make the results directly usable in downstream systems. We integrate geographic calculations, configurable speed profiles, traffic buffers, and multi-stop route optimization, ensuring the agent behaves deterministically while still reasoning flexibly through tools.
!pip -q install -U langchain langchain-openai pydantic
import os
from getpass import getpass
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass("Enter OPENAI_API_KEY (input hidden): ")
from typing import Dict, List, Optional, Tuple, Any
from math import radians, sin, cos, sqrt, atan2
from pydantic import BaseModel, Field, ValidationError
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_agentWe set up the execution environment and ensure all required libraries are installed and imported correctly. We securely load the OpenAI API key so the agent can interact with the language model without hardcoding credentials. We also prepare the core dependencies that power tools, agents, and structured outputs.
SITES: Dict[str, Dict[str, Any]] = {
"Rig_A": {"lat": 23.5880, "lon": 58.3829, "type": "rig"},
"Rig_B": {"lat": 23.6100, "lon": 58.5400, "type": "rig"},
"Rig_C": {"lat": 23.4500, "lon": 58.3000, "type": "rig"},
"Yard_Main": {"lat": 23.5700, "lon": 58.4100, "type": "yard"},
"Depot_1": {"lat": 23.5200, "lon": 58.4700, "type": "depot"},
"Depot_2": {"lat": 23.6400, "lon": 58.4300, "type": "depot"},
}
SPEED_PROFILES: Dict[str, float] = {
"highway": 90.0,
"arterial": 65.0,
"local": 45.0,
}
DEFAULT_TRAFFIC_MULTIPLIER = 1.10
def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
R = 6371.0
dlat = radians(lat2 - lat1)
dlon = radians(lon2 - lon1)
a = sin(dlat / 2) ** 2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2
return R * cWe define the core domain data representing rigs, yards, and depots along with their geographic coordinates. We establish speed profiles and a default traffic multiplier to reflect realistic driving conditions. We also implement the Haversine distance function, which serves as the mathematical backbone of all routing decisions.
def _normalize_site_name(name: str) -> str:
return name.strip()
def _assert_site_exists(name: str) -> None:
if name not in SITES:
raise ValueError(f"Unknown site '{name}'. Use list_sites() or suggest_site().")
def _distance_between(a: str, b: str) -> float:
_assert_site_exists(a)
_assert_site_exists(b)
sa, sb = SITES[a], SITES[b]
return float(haversine_km(sa["lat"], sa["lon"], sb["lat"], sb["lon"]))
def _eta_minutes(distance_km: float, speed_kmph: float, traffic_multiplier: float) -> float:
speed = max(float(speed_kmph), 1e-6)
base_minutes = (distance_km / speed) * 60.0
return float(base_minutes * max(float(traffic_multiplier), 0.0))
def compute_route_metrics(path: List[str], speed_kmph: float, traffic_multiplier: float) -> Dict[str, Any]:
if len(path) < 2:
raise ValueError("Route path must include at least origin and destination.")
for s in path:
_assert_site_exists(s)
legs = []
total_km = 0.0
total_min = 0.0
for i in range(len(path) - 1):
a, b = path[i], path[i + 1]
d_km = _distance_between(a, b)
t_min = _eta_minutes(d_km, speed_kmph, traffic_multiplier)
legs.append({"from": a, "to": b, "distance_km": d_km, "eta_minutes": t_min})
total_km += d_km
total_min += t_min
return {"route": path, "distance_km": float(total_km), "eta_minutes": float(total_min), "legs": legs}We build the low-level utility functions that validate site names and compute distances and travel times. We implement logic to calculate per-leg and total route metrics deterministically. This ensures that every ETA and distance returned by the agent is based on explicit computation rather than inference.
def _all_paths_with_waypoints(origin: str, destination: str, waypoints: List[str], max_stops: int) -> List[List[str]]:
from itertools import permutations
waypoints = [w for w in waypoints if w not in (origin, destination)]
max_stops = int(max(0, max_stops))
candidates = []
for k in range(0, min(len(waypoints), max_stops) + 1):
for perm in permutations(waypoints, k):
candidates.append([origin, *perm, destination])
if [origin, destination] not in candidates:
candidates.insert(0, [origin, destination])
return candidates
def find_best_route(origin: str, destination: str, allowed_waypoints: Optional[List[str]], max_stops: int, speed_kmph: float, traffic_multiplier: float, objective: str, top_k: int) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
destination = _normalize_site_name(destination)
_assert_site_exists(origin)
_assert_site_exists(destination)
allowed_waypoints = allowed_waypoints or []
for w in allowed_waypoints:
_assert_site_exists(_normalize_site_name(w))
objective = (objective or "eta").strip().lower()
if objective not in {"eta", "distance"}:
raise ValueError("objective must be one of: 'eta', 'distance'")
top_k = max(1, int(top_k))
candidates = _all_paths_with_waypoints(origin, destination, allowed_waypoints, max_stops=max_stops)
scored = []
for path in candidates:
metrics = compute_route_metrics(path, speed_kmph=speed_kmph, traffic_multiplier=traffic_multiplier)
score = metrics["eta_minutes"] if objective == "eta" else metrics["distance_km"]
scored.append((score, metrics))
scored.sort(key=lambda x: x[0])
best = scored[0][1]
alternatives = [m for _, m in scored[1:top_k]]
return {"best": best, "alternatives": alternatives, "objective": objective}We introduce multi-stop routing logic by generating candidate paths with optional waypoints. We evaluate each candidate route against a clear optimization objective, such as ETA or distance. We then rank routes and extract the best option along with a set of strong alternatives.
@tool
def list_sites(site_type: Optional[str] = None) -> List[str]:
if site_type:
st = site_type.strip().lower()
return sorted([k for k, v in SITES.items() if str(v.get("type", "")).lower() == st])
return sorted(SITES.keys())
@tool
def get_site_details(site: str) -> Dict[str, Any]:
s = _normalize_site_name(site)
_assert_site_exists(s)
return {"site": s, **SITES[s]}
@tool
def suggest_site(query: str, max_suggestions: int = 5) -> List[str]:
q = (query or "").strip().lower()
max_suggestions = max(1, int(max_suggestions))
scored = []
for name in SITES.keys():
n = name.lower()
common = len(set(q) & set(n))
bonus = 5 if q and q in n else 0
scored.append((common + bonus, name))
scored.sort(key=lambda x: x[0], reverse=True)
return [name for _, name in scored[:max_suggestions]]
@tool
def compute_direct_route(origin: str, destination: str, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
destination = _normalize_site_name(destination)
rc = (road_class or "arterial").strip().lower()
if rc not in SPEED_PROFILES:
raise ValueError(f"Unknown road_class '{road_class}'. Use one of: {sorted(SPEED_PROFILES.keys())}")
speed = SPEED_PROFILES[rc]
return compute_route_metrics([origin, destination], speed_kmph=speed, traffic_multiplier=float(traffic_multiplier))
@tool
def optimize_route(origin: str, destination: str, allowed_waypoints: Optional[List[str]] = None, max_stops: int = 2, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER, objective: str = "eta", top_k: int = 3) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
destination = _normalize_site_name(destination)
rc = (road_class or "arterial").strip().lower()
if rc not in SPEED_PROFILES:
raise ValueError(f"Unknown road_class '{road_class}'. Use one of: {sorted(SPEED_PROFILES.keys())}")
speed = SPEED_PROFILES[rc]
allowed_waypoints = allowed_waypoints or []
allowed_waypoints = [_normalize_site_name(w) for w in allowed_waypoints]
return find_best_route(origin, destination, allowed_waypoints, int(max_stops), float(speed), float(traffic_multiplier), str(objective), int(top_k))We expose the routing and discovery logic as callable tools for the agent. We allow the agent to list sites, inspect site details, resolve ambiguous names, and compute both direct and optimized routes. This tool layer ensures that the agent always reasons by calling verified functions rather than hallucinating results.
class RouteLeg(BaseModel):
from_site: str
to_site: str
distance_km: float
eta_minutes: float
class RoutePlan(BaseModel):
route: List[str]
distance_km: float
eta_minutes: float
legs: List[RouteLeg]
objective: str
class RouteDecision(BaseModel):
chosen: RoutePlan
alternatives: List[RoutePlan] = []
assumptions: Dict[str, Any] = {}
notes: str = ""
audit: List[str] = []
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
SYSTEM_PROMPT = (
"You are the Route Optimizer Agent for a logistics dispatch center.\n"
"You MUST use tools for any distance/ETA calculation.\n"
"Return ONLY the structured RouteDecision."
)
route_agent = create_agent(
model=llm,
tools=[list_sites, get_site_details, suggest_site, compute_direct_route, optimize_route],
system_prompt=SYSTEM_PROMPT,
response_format=RouteDecision,
)
def get_route_decision(origin: str, destination: str, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER, allowed_waypoints: Optional[List[str]] = None, max_stops: int = 2, objective: str = "eta", top_k: int = 3) -> RouteDecision:
user_msg = {
"role": "user",
"content": (
f"Optimize the route from {origin} to {destination}.\n"
f"road_class={road_class}, traffic_multiplier={traffic_multiplier}\n"
f"objective={objective}, top_k={top_k}\n"
f"allowed_waypoints={allowed_waypoints}, max_stops={max_stops}\n"
"Return the structured RouteDecision only."
),
}
result = route_agent.invoke({"messages": [user_msg]})
return result["structured_response"]
decision1 = get_route_decision("Yard_Main", "Rig_B", road_class="arterial", traffic_multiplier=1.12)
print(decision1.model_dump())
decision2 = get_route_decision("Rig_C", "Rig_B", road_class="highway", traffic_multiplier=1.08, allowed_waypoints=["Depot_1", "Depot_2", "Yard_Main"], max_stops=2, objective="eta", top_k=3)
print(decision2.model_dump())We define strict Pydantic schemas to enforce structured, machine-readable outputs from the agent. We initialize the language model and create the agent with a clear system prompt and response format. We then demonstrate how to invoke the agent and obtain reliable route decisions ready for real logistics workflows.
In conclusion, we have implemented a robust, extensible route optimization agent that selects the best path between sites while clearly explaining its assumptions and alternatives. We demonstrated how combining deterministic routing logic with a tool-calling LLM produces reliable, auditable decisions suitable for real logistics operations. This foundation allows us to easily extend the system with live traffic data, fleet constraints, or cost-based objectives, making the agent a practical component in a larger dispatch or fleet-management platform.
Check out the Full Codes here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
