Source: MarkTechPost
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.
