On this tutorial, we construct a production-style Route Optimizer Agent for a logistics dispatch heart utilizing the newest LangChain agent APIs. We design a tool-driven workflow through which the agent reliably computes distances, ETAs, and optimum routes slightly than guessing, and we implement structured outputs to make the outcomes immediately usable in downstream techniques. We combine geographic calculations, configurable pace profiles, visitors buffers, and multi-stop route optimization, making certain the agent behaves deterministically whereas nonetheless reasoning flexibly by 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 (enter hidden): ")
from typing import Dict, Record, Optionally available, Tuple, Any
from math import radians, sin, cos, sqrt, atan2
from pydantic import BaseModel, Subject, ValidationError
from langchain_openai import ChatOpenAI
from langchain.instruments import instrument
from langchain.brokers import create_agentWe arrange the execution surroundings 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, "sort": "rig"},
"Rig_B": {"lat": 23.6100, "lon": 58.5400, "sort": "rig"},
"Rig_C": {"lat": 23.4500, "lon": 58.3000, "sort": "rig"},
"Yard_Main": {"lat": 23.5700, "lon": 58.4100, "sort": "yard"},
"Depot_1": {"lat": 23.5200, "lon": 58.4700, "sort": "depot"},
"Depot_2": {"lat": 23.6400, "lon": 58.4300, "sort": "depot"},
}
SPEED_PROFILES: Dict[str, float] = {
"freeway": 90.0,
"arterial": 65.0,
"native": 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 pace profiles and a default visitors multiplier to mirror lifelike driving circumstances. We additionally implement the Haversine distance perform, which serves because the mathematical spine of all routing selections.
def _normalize_site_name(title: str) -> str:
return title.strip()
def _assert_site_exists(title: str) -> None:
if title not in SITES:
elevate ValueError(f"Unknown website '{title}'. 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:
pace = max(float(speed_kmph), 1e-6)
base_minutes = (distance_km / pace) * 60.0
return float(base_minutes * max(float(traffic_multiplier), 0.0))
def compute_route_metrics(path: Record[str], speed_kmph: float, traffic_multiplier: float) -> Dict[str, Any]:
if len(path) < 2:
elevate ValueError("Route path should embody at the very least origin and vacation spot.")
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 website 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 express computation slightly than inference.
def _all_paths_with_waypoints(origin: str, vacation spot: str, waypoints: Record[str], max_stops: int) -> Record[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("goal should be one among: '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.kind(key=lambda x: x[0])
greatest = scored[0][1]
options = [m for _, m in scored[1:top_k]]
return {"greatest": greatest, "options": options, "goal": goal}We introduce multi-stop routing logic by producing candidate paths with non-compulsory waypoints. We consider every candidate route in opposition to a transparent optimization goal, equivalent to ETA or distance. We then rank routes and extract the most suitable choice together with a set of robust options.
@instrument
def list_sites(site_type: Optionally available[str] = None) -> Record[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(website: str) -> Dict[str, Any]:
s = _normalize_site_name(website)
_assert_site_exists(s)
return {"website": s, **SITES[s]}
@instrument
def suggest_site(question: str, max_suggestions: int = 5) -> Record[str]:
q = (question or "").strip().decrease()
max_suggestions = max(1, int(max_suggestions))
scored = []
for title in SITES.keys():
n = title.decrease()
widespread = len(set(q) & set(n))
bonus = 5 if q and q in n else 0
scored.append((widespread + bonus, title))
scored.kind(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 among: {sorted(SPEED_PROFILES.keys())}")
pace = SPEED_PROFILES[rc]
return compute_route_metrics([origin, destination], speed_kmph=pace, 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 among: {sorted(SPEED_PROFILES.keys())}")
pace = 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(pace), float(traffic_multiplier), str(goal), int(top_k))We expose the routing and discovery logic as callable instruments for the agent. We enable the agent to listing websites, examine website particulars, resolve ambiguous names, and compute each direct and optimized routes. This instrument layer ensures that the agent all the time causes by calling verified features slightly than hallucinating outcomes.
class RouteLeg(BaseModel):
from_site: str
to_site: str
distance_km: float
eta_minutes: float
class RoutePlan(BaseModel):
route: Record[str]
distance_km: float
eta_minutes: float
legs: Record[RouteLeg]
goal: str
class RouteDecision(BaseModel):
chosen: RoutePlan
options: Record[RoutePlan] = []
assumptions: Dict[str, Any] = {}
notes: str = ""
audit: Record[str] = []
llm = ChatOpenAI(mannequin="gpt-4o-mini", temperature=0.2)
SYSTEM_PROMPT = (
"You're the Route Optimizer Agent for a logistics dispatch heart.n"
"You MUST use instruments 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 = {
"function": "person",
"content material": (
f"Optimize the route from {origin} to {vacation spot}.n"
f"road_class={road_class}, traffic_multiplier={traffic_multiplier}n"
f"goal={goal}, top_k={top_k}n"
f"allowed_waypoints={allowed_waypoints}, max_stops={max_stops}n"
"Return the structured RouteDecision solely."
),
}
end result = route_agent.invoke({"messages": [user_msg]})
return end 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="freeway", 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 display methods to invoke the agent and acquire dependable route selections prepared for actual logistics workflows.
In conclusion, we now have applied a sturdy, extensible route optimization agent that selects the perfect path between websites whereas clearly explaining its assumptions and options. We demonstrated how combining deterministic routing logic with a tool-calling LLM produces dependable, auditable selections appropriate for actual logistics operations. This basis permits us to simply prolong the system with stay visitors information, fleet constraints, or cost-based goals, making the agent a sensible part in a bigger dispatch or fleet-management platform.
Take a look at the Full Codes here. Additionally, be at liberty to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
