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 by which the agent reliably computes distances, ETAs, and optimum routes reasonably than guessing, and we implement structured outputs to make the outcomes immediately usable in downstream programs. We combine geographic calculations, configurable pace profiles, visitors buffers, and multi-stop route optimization, guaranteeing the agent behaves deterministically whereas nonetheless reasoning flexibly by means 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 (enter hidden): ")
from typing import Dict, Record, Non-compulsory, Tuple, Any
from math import radians, sin, cos, sqrt, atan2
from pydantic import BaseModel, Discipline, ValidationError
from langchain_openai import ChatOpenAI
from langchain.instruments import device
from langchain.brokers import create_agentWe arrange the execution surroundings and guarantee all required libraries are put in and imported accurately. 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 knowledge representing rigs, yards, and depots together with their geographic coordinates. We set up pace profiles and a default visitors multiplier to mirror reasonable driving situations. We additionally implement the Haversine distance operate, 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 web site '{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:
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:
increase ValueError("Route path should embrace a minimum of 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 web site names and compute distances and journey occasions. We implement logic to calculate per-leg and complete route metrics deterministically. This ensures that each ETA and distance returned by the agent is predicated on specific computation reasonably 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: Non-compulsory[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 certainly 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])
finest = scored[0][1]
options = [m for _, m in scored[1:top_k]]
return {"finest": finest, "options": options, "goal": goal}We introduce multi-stop routing logic by producing candidate paths with non-obligatory waypoints. We consider every candidate route towards a transparent optimization goal, equivalent to ETA or distance. We then rank routes and extract the best choice together with a set of robust options.
@device
def list_sites(site_type: Non-compulsory[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())
@device
def get_site_details(web site: str) -> Dict[str, Any]:
s = _normalize_site_name(web site)
_assert_site_exists(s)
return {"web site": s, **SITES[s]}
@device
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 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]]
@device
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 certainly one of: {sorted(SPEED_PROFILES.keys())}")
pace = SPEED_PROFILES[rc]
return compute_route_metrics([origin, destination], speed_kmph=pace, traffic_multiplier=float(traffic_multiplier))
@device
def optimize_route(origin: str, vacation spot: str, allowed_waypoints: Non-compulsory[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 certainly one of: {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 permit the agent to listing websites, examine web site particulars, resolve ambiguous names, and compute each direct and optimized routes. This device layer ensures that the agent at all times causes by calling verified features reasonably 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: Non-compulsory[List[str]] = None, max_stops: int = 2, goal: str = "eta", top_k: int = 3) -> RouteDecision:
user_msg = {
"position": "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."
),
}
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="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 easy methods to invoke the agent and procure dependable route selections prepared for actual logistics workflows.
In conclusion, we have now carried out a sturdy, extensible route optimization agent that selects the very 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 stay visitors knowledge, fleet constraints, or cost-based aims, 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 E-newsletter. Wait! are you on telegram? now you’ll be able to be a part of us on telegram as nicely.
