On this tutorial, we construct a production-style Route Optimizer Agent for a logistics dispatch middle utilizing the most recent LangChain agent APIs. We design a tool-driven workflow by which the agent reliably computes distances, ETAs, and optimum routes somewhat than guessing, and we implement structured outputs to make the outcomes immediately usable in downstream techniques. We combine geographic calculations, configurable velocity profiles, visitors buffers, and multi-stop route optimization, making certain the agent behaves deterministically whereas nonetheless reasoning flexibly by way of instruments.
!pip -q set up -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, Checklist, Optionally available, Tuple, Any
from math import radians, sin, cos, sqrt, atan2
from pydantic import BaseModel, Area, ValidationError
from langchain_openai import ChatOpenAI
from langchain.instruments import instrument
from langchain.brokers import create_agentWe arrange the execution setting and guarantee all required libraries are put in and imported appropriately. We securely load the OpenAI API key so the agent can work together with the language mannequin with out hardcoding credentials. We additionally put together the core dependencies that energy instruments, brokers, 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 outline the core area information representing rigs, yards, and depots together with their geographic coordinates. We set up velocity profiles and a default visitors multiplier to replicate real looking driving circumstances. We additionally implement the Haversine distance operate, which serves because the mathematical spine of all routing choices.
def _normalize_site_name(identify: str) -> str:
return identify.strip()
def _assert_site_exists(identify: str) -> None:
if identify not in SITES:
elevate 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:
velocity = max(float(speed_kmph), 1e-6)
base_minutes = (distance_km / velocity) * 60.0
return float(base_minutes * max(float(traffic_multiplier), 0.0))
def compute_route_metrics(path: Checklist[str], speed_kmph: float, traffic_multiplier: float) -> Dict[str, Any]:
if len(path) < 2:
elevate 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 vary(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 construct the low-level utility features that validate web site names and compute distances and journey instances. We implement logic to calculate per-leg and whole route metrics deterministically. This ensures that each ETA and distance returned by the agent is predicated on specific computation somewhat than inference.
def _all_paths_with_waypoints(origin: str, vacation spot: str, waypoints: Checklist[str], max_stops: int) -> Checklist[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 ok in vary(0, min(len(waypoints), max_stops) + 1):
for perm in permutations(waypoints, ok):
candidates.append([origin, *perm, destination])
if [origin, destination] not in candidates:
candidates.insert(0, [origin, destination])
return candidates
def find_best_route(origin: str, vacation spot: str, allowed_waypoints: Optionally available[List[str]], max_stops: int, speed_kmph: float, traffic_multiplier: float, goal: str, top_k: int) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
vacation spot = _normalize_site_name(vacation spot)
_assert_site_exists(origin)
_assert_site_exists(vacation spot)
allowed_waypoints = allowed_waypoints or []
for w in allowed_waypoints:
_assert_site_exists(_normalize_site_name(w))
goal = (goal or "eta").strip().decrease()
if goal not in {"eta", "distance"}:
elevate ValueError("objective must be one of: 'eta', 'distance'")
top_k = max(1, int(top_k))
candidates = _all_paths_with_waypoints(origin, vacation spot, allowed_waypoints, max_stops=max_stops)
scored = []
for path in candidates:
metrics = compute_route_metrics(path, speed_kmph=speed_kmph, traffic_multiplier=traffic_multiplier)
rating = metrics["eta_minutes"] if goal == "eta" else metrics["distance_km"]
scored.append((rating, metrics))
scored.type(key=lambda x: x[0])
greatest = scored[0][1]
alternate options = [m for _, m in scored[1:top_k]]
return {"best": greatest, "alternatives": alternate options, "objective": goal}We introduce multi-stop routing logic by producing candidate paths with non-obligatory waypoints. We consider every candidate route in opposition to a transparent optimization goal, akin to ETA or distance. We then rank routes and extract the best choice together with a set of sturdy alternate options.
@instrument
def list_sites(site_type: Optionally available[str] = None) -> Checklist[str]:
if site_type:
st = site_type.strip().decrease()
return sorted([k for k, v in SITES.items() if str(v.get("type", "")).lower() == st])
return sorted(SITES.keys())
@instrument
def get_site_details(web site: str) -> Dict[str, Any]:
s = _normalize_site_name(web site)
_assert_site_exists(s)
return {"site": s, **SITES[s]}
@instrument
def suggest_site(question: str, max_suggestions: int = 5) -> Checklist[str]:
q = (question or "").strip().decrease()
max_suggestions = max(1, int(max_suggestions))
scored = []
for identify in SITES.keys():
n = identify.decrease()
widespread = len(set(q) & set(n))
bonus = 5 if q and q in n else 0
scored.append((widespread + bonus, identify))
scored.type(key=lambda x: x[0], reverse=True)
return [name for _, name in scored[:max_suggestions]]
@instrument
def compute_direct_route(origin: str, vacation spot: str, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
vacation spot = _normalize_site_name(vacation spot)
rc = (road_class or "arterial").strip().decrease()
if rc not in SPEED_PROFILES:
elevate ValueError(f"Unknown road_class '{road_class}'. Use one of: {sorted(SPEED_PROFILES.keys())}")
velocity = SPEED_PROFILES[rc]
return compute_route_metrics([origin, destination], speed_kmph=velocity, traffic_multiplier=float(traffic_multiplier))
@instrument
def optimize_route(origin: str, vacation spot: str, allowed_waypoints: Optionally available[List[str]] = None, max_stops: int = 2, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER, goal: str = "eta", top_k: int = 3) -> Dict[str, Any]:
origin = _normalize_site_name(origin)
vacation spot = _normalize_site_name(vacation spot)
rc = (road_class or "arterial").strip().decrease()
if rc not in SPEED_PROFILES:
elevate ValueError(f"Unknown road_class '{road_class}'. Use one of: {sorted(SPEED_PROFILES.keys())}")
velocity = SPEED_PROFILES[rc]
allowed_waypoints = allowed_waypoints or []
allowed_waypoints = [_normalize_site_name(w) for w in allowed_waypoints]
return find_best_route(origin, vacation spot, allowed_waypoints, int(max_stops), float(velocity), float(traffic_multiplier), str(goal), int(top_k))We expose the routing and discovery logic as callable instruments for the agent. We permit the agent to listing websites, examine web site particulars, resolve ambiguous names, and compute each direct and optimized routes. This instrument layer ensures that the agent at all times causes by calling verified features somewhat than hallucinating outcomes.
class RouteLeg(BaseModel):
from_site: str
to_site: str
distance_km: float
eta_minutes: float
class RoutePlan(BaseModel):
route: Checklist[str]
distance_km: float
eta_minutes: float
legs: Checklist[RouteLeg]
goal: str
class RouteDecision(BaseModel):
chosen: RoutePlan
alternate options: Checklist[RoutePlan] = []
assumptions: Dict[str, Any] = {}
notes: str = ""
audit: Checklist[str] = []
llm = ChatOpenAI(mannequin="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(
mannequin=llm,
instruments=[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, vacation spot: str, road_class: str = "arterial", traffic_multiplier: float = DEFAULT_TRAFFIC_MULTIPLIER, allowed_waypoints: Optionally available[List[str]] = None, max_stops: int = 2, goal: 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."
),
}
outcome = route_agent.invoke({"messages": [user_msg]})
return outcome["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, goal="eta", top_k=3)
print(decision2.model_dump())We outline strict Pydantic schemas to implement structured, machine-readable outputs from the agent. We initialize the language mannequin and create the agent with a transparent system immediate and response format. We then reveal methods to invoke the agent and acquire dependable route choices prepared for actual logistics workflows.
In conclusion, we now have applied a sturdy, extensible route optimization agent that selects the most effective path between websites whereas clearly explaining its assumptions and alternate options. We demonstrated how combining deterministic routing logic with a tool-calling LLM produces dependable, auditable choices appropriate for actual logistics operations. This basis permits us to simply prolong the system with dwell visitors information, fleet constraints, or cost-based targets, making the agent a sensible element in a bigger dispatch or fleet-management platform.
Try the Full Codes right here. Additionally, be happy to observe us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you may be a part of us on telegram as properly.
