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 reasonably than guessing, and we implement structured outputs to make the outcomes straight usable in downstream techniques. We combine geographic calculations, configurable velocity profiles, site visitors buffers, and multi-stop route optimization, guaranteeing 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, Checklist, Non-obligatory, 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 software
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, "kind": "rig"},
"Rig_B": {"lat": 23.6100, "lon": 58.5400, "kind": "rig"},
"Rig_C": {"lat": 23.4500, "lon": 58.3000, "kind": "rig"},
"Yard_Main": {"lat": 23.5700, "lon": 58.4100, "kind": "yard"},
"Depot_1": {"lat": 23.5200, "lon": 58.4700, "kind": "depot"},
"Depot_2": {"lat": 23.6400, "lon": 58.4300, "kind": "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 velocity profiles and a default site visitors multiplier to replicate life like driving circumstances. We additionally implement the Haversine distance perform, which serves because the mathematical spine of all routing selections.
def _normalize_site_name(identify: str) -> str:
return identify.strip()
def _assert_site_exists(identify: str) -> None:
if identify not in SITES:
increase ValueError(f"Unknown website '{identify}'. 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:
increase ValueError("Route path should embrace not less than 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 capabilities that validate website names and compute distances and journey occasions. We implement logic to calculate per-leg and whole route metrics deterministically. This ensures that each ETA and distance returned by the agent relies on express computation reasonably 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 okay in vary(0, min(len(waypoints), max_stops) + 1):
for perm in permutations(waypoints, okay):
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: Non-obligatory[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"}:
increase ValueError("goal should be one in all: '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]
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 elective waypoints. We consider every candidate route in opposition to a transparent optimization goal, reminiscent of ETA or distance. We then rank routes and extract the best choice together with a set of sturdy options.
@software
def list_sites(site_type: Non-obligatory[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())
@software
def get_site_details(website: str) -> Dict[str, Any]:
s = _normalize_site_name(website)
_assert_site_exists(s)
return {"website": s, **SITES[s]}
@software
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()
frequent = len(set(q) & set(n))
bonus = 5 if q and q in n else 0
scored.append((frequent + bonus, identify))
scored.type(key=lambda x: x[0], reverse=True)
return [name for _, name in scored[:max_suggestions]]
@software
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:
increase ValueError(f"Unknown road_class '{road_class}'. Use one in all: {sorted(SPEED_PROFILES.keys())}")
velocity = SPEED_PROFILES[rc]
return compute_route_metrics([origin, destination], speed_kmph=velocity, traffic_multiplier=float(traffic_multiplier))
@software
def optimize_route(origin: str, vacation spot: str, allowed_waypoints: Non-obligatory[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:
increase ValueError(f"Unknown road_class '{road_class}'. Use one in all: {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 record websites, examine website particulars, resolve ambiguous names, and compute each direct and optimized routes. This software layer ensures that the agent all the time causes by calling verified capabilities reasonably 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
options: Checklist[RoutePlan] = []
assumptions: Dict[str, Any] = {}
notes: str = ""
audit: Checklist[str] = []
llm = ChatOpenAI(mannequin="gpt-4o-mini", temperature=0.2)
SYSTEM_PROMPT = (
"You're the Route Optimizer Agent for a logistics dispatch middle.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: Non-obligatory[List[str]] = None, max_stops: int = 2, goal: str = "eta", top_k: int = 3) -> RouteDecision:
user_msg = {
"function": "consumer",
"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 show the best way to invoke the agent and acquire dependable route selections prepared for actual logistics workflows.
In conclusion, we have now carried out a strong, extensible route optimization agent that selects one of the best 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 lengthen the system with dwell site visitors information, fleet constraints, or cost-based goals, making the agent a sensible element in a bigger dispatch or fleet-management platform.
Try the Full Codes right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you possibly can be a part of us on telegram as effectively.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the most recent breakthroughs, get unique updates, and join with a worldwide community of future-focused thinkers.
Unlock tomorrow’s developments right now: learn extra, subscribe to our publication, and grow to be a part of the NextTech group at NextTech-news.com
