"""
Nexus-Pathfinder Intelligence Console
======================================
Enterprise Supply-Chain Risk & Compliance Copilot
AMD Developer Hackathon: ACT II — Track 1 (Hybrid Token-Efficient Routing Agent)
Architecture
------------
A five-stage local-first multi-agent pipeline sits in front of a Groq-backed
reasoning layer:
1. Scanner Agent -> local, zero-token intent/domain classification
2. Retrieval Agent -> local, zero-token FAISS semantic search over the
supply-chain knowledge base (tariffs, SLAs, ports)
3. Escalation Agent -> local, zero-token confidence gate. Only escalates
to a paid web search / LLM call when local
retrieval confidence is insufficient
4. Web Intelligence Agent-> optional, triggered only on low confidence
5. Compliance Reasoning -> remote Groq LLM (Llama-3.3-70B) call, used only
Agent for the genuinely hard synthesis / risk reasoning
step, combining KB + web context + short-term
conversation memory
This mirrors the Track 1 token-efficiency philosophy: cheap, deterministic
work (classification, vector search, gating) never touches a paid model, and
the expensive model is invoked exactly once, only when synthesis actually
requires it.
"""
from __future__ import annotations
import os
import re
import traceback
from dataclasses import dataclass, field
from typing import Any
import gradio as gr
from openai import OpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
try:
from pypdf import PdfReader
PDF_SUPPORT = True
except Exception: # pragma: no cover
PDF_SUPPORT = False
try:
from duckduckgo_search import DDGS
WEB_SEARCH_AVAILABLE = True
except Exception: # pragma: no cover
WEB_SEARCH_AVAILABLE = False
# ======================================================================
# CONFIGURATION
# ======================================================================
EMBEDDING_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
EMBEDDING_DEVICE = "cpu"
GROQ_BASE_URL = "https://api.groq.com/openai/v1"
GROQ_MODEL = "llama-3.3-70b-versatile"
GROQ_TEMPERATURE = 0.15
GROQ_MAX_TOKENS = 900
CHUNK_SIZE = 500
CHUNK_OVERLAP = 60
RETRIEVAL_K = 4
CONFIDENCE_ESCALATION_THRESHOLD = 45.0 # below this %, trigger web search
MAX_WEB_RESULTS = 4
MAX_MEMORY_TURNS = 3 # number of prior user+assistant exchanges kept as conversation memory
APP_TITLE = "Nexus-Pathfinder Intelligence Console"
APP_SUBTITLE = "Multi-Agent Supply-Chain Risk, Compliance & Decision Support — AMD Hackathon Track 1"
SYSTEM_PROMPT = (
"You are the Compliance & Logistics Reasoning Agent inside Nexus-Pathfinder, "
"an autonomous multi-agent supply-chain intelligence system.\n\n"
"You receive verified context retrieved by a Knowledge Base Agent and, when "
"necessary, a Web Intelligence Agent. Your job is to synthesize this into an "
"operational answer for a logistics/compliance decision-maker.\n\n"
"STRICT RULES:\n"
"1. Never fabricate tariff rates, contract clauses, SLA numbers, or legal citations. "
"Only use what is present in the supplied context.\n"
"2. If context is insufficient to fully answer, clearly say what is missing.\n"
"3. Always structure your reply using EXACTLY these three markdown headers, in order:\n"
"## Analysis\n"
"(A concise operational analysis of the situation grounded in the context.)\n"
"## Risk Assessment\n"
"(State a single risk level of LOW, MEDIUM, HIGH, or CRITICAL on the first line, "
"then 1-3 sentences justifying it.)\n"
"## Recommendation\n"
"(A short, decisive, actionable recommendation for the operations manager.)\n"
"4. Keep the entire reply under 260 words. Be precise, not verbose."
)
# ======================================================================
# SEED KNOWLEDGE BASE
# ======================================================================
SEED_DOCUMENTS: dict[str, str] = {
"global_trade_sanctions_2026.md": """# Global Trade & Sanction Regulations (FY 2026)
## Section 1: Electronics & HS-8542 Semiconductors
* Article 12.1 (Singapore Transit Exemption): Electronic sub-assemblies and
semiconductors classified under tariff category HS-8542 are 100% exempt from
transit customs tariffs when routed via the Port of Singapore, provided they
originate from approved ASEAN manufacturers.
* Article 4.2 (Suez Canal Surcharge): Due to heightened regional security
issues and insurance premiums, all maritime cargo transiting through the
Suez Canal is subject to a flat 15% emergency war-risk tariff.
* Article 9.4 (Taiwan Strait Restrictions): Heavy cargo transiting via the
Taiwan Strait is subject to mandatory 48-hour inspections and a temporary
geopolitical tariff surcharge of 25% unless cleared under direct diplomatic
passage agreements.
* Article 7.3 (China Land Transit Advisory): Transshipment through mainland
Chinese hubs for HS-8542 category cargo currently carries elevated
documentation risk due to pending tariff category disputes; carriers are
advised to prefer maritime bypass routes.
## Section 2: Port & Logistics Operations Restrictions
* Port of Hamburg (DEHAM): Vessels over 120,000 deadweight tonnage (DWT) face
restricted river Elbe access during low-tide intervals. Standard offloading
delay penalty is calculated at $8,500/hour.
* Suez Canal Bypass Regulations: Cargo rerouted via the Cape of Good Hope
(South Africa) bypass is completely exempt from standard Middle-Eastern
customs inspections but adds approximately 3,200 nautical miles
(~10-12 days) to standard European transit legs.
* Port of Singapore (SGSIN): Functions as the primary compliant transshipment
hub for ASEAN-origin electronics; average dwell time 18 hours.
""",
"supplier_sla_contracts.md": """# Supplier SLA Contracts Directory (Active FY 2026)
## Agreement: SLA-902 (ASEAN Semiconductor Co. - Malaysia)
* Contractor: ASEAN Semiconductor Co., Penang Facility, Malaysia.
* Scope: Emergency manufacturing and export of high-grade microchips
(HS-8542 category).
* Delivery Lead Time: Guarantees standard delivery lead times of 3 business
days under Emergency Activation Protocol Code SLA-902.
* Rates: Fixed emergency contract transit rate of $2.40 per unit for
shipments up to 500,000 units.
* Fallback Penalties: Delay penalty is waived if transport disruption is
flagged as force majeure (e.g., maritime blockades).
## Agreement: SLA-504 (EuroTech Logistics - Hamburg)
* Contractor: EuroTech Logistics GmbH, Hamburg, Germany.
* Scope: Port-to-warehouse drayage and short-term container storage.
* Guarantees: Guaranteed container offloading and sorting within 6 hours of
berth docking.
* Default Penalty: Default penalties for standard delivery delays exceed
$10,000 per idle day if drayage logistics fail due to internal scheduling
issues.
* Disaster Mitigation: Standard contract Clause 14.2 allows free drayage
rerouting to Rotterdam or Antwerp in the event of Hamburg Port congestion
strikes.
## Agreement: SLA-233 (Saigon Integrated Circuits - Vietnam)
* Contractor: Saigon Integrated Circuits, Ho Chi Minh City, Vietnam.
* Scope: Backup semiconductor sourcing for HS-8542 category components.
* Delivery Lead Time: 34 hours air-freight, 6 business days sea-freight.
* Rates: $2.90 per unit, medium regional risk rating.
* Fallback Penalties: No force-majeure waiver clause; standard 5% penalty
applies for delays beyond contracted window.
""",
"logistics_risk_playbook.md": """# Logistics Disruption & Risk Response Playbook (FY 2026)
## Canal / Chokepoint Disruption Response
* Suez Canal Closure: Default mitigation is reroute via Cape of Good Hope
(Singapore -> Cape Town -> Gibraltar -> Hamburg). Adds 10-12 days transit,
removes Suez war-risk surcharge (15%).
* Panama Canal Low-Water Restriction: Prefer rail-bridge alternatives through
Mexican inland corridors for time-sensitive cargo; air-freight bridge only
justified when idle exposure exceeds $500,000/day.
* Port Strike Protocol: Immediately flag all shipments whose route geofence
intersects the affected port radius; activate Alternative Procurement Agent
to source substitute suppliers outside the impacted region before
attempting a physical reroute.
## Risk Scoring Guidance
* LOW: Disruption resolved by rerouting alone, no new supplier required, added
cost under $50,000.
* MEDIUM: Requires supplier substitution or a route add of 5-10 days, added
cost $50,000-$500,000.
* HIGH: Requires cross-border compliance re-validation, added cost
$500,000-$1,000,000, or delay exceeding 10 days.
* CRITICAL: Multiple failed compliance passes, idle exposure exceeding
$1,000,000/day, or no compliant route found within 5 iterations.
## Escalation Rule
* Maximum of 5 negotiation iterations between Logistics Router and Compliance
Agent before the system halts and returns the best available alternative
with an explicit warning banner instead of looping indefinitely.
""",
}
# ======================================================================
# UTILITIES
# ======================================================================
def _safe_str(value: Any, fallback: str = "") -> str:
if value is None:
return fallback
return str(value)
def classify_task_locally(query: str) -> dict[str, str]:
"""Zero-token local intent classifier (Scanner Agent).
Pure keyword heuristics -- runs entirely on CPU with no model call,
mirroring the Track 1 hybrid-routing philosophy of keeping cheap,
well-structured classification work off the paid inference path.
"""
q = query.lower()
domains = {
"compliance_legal": ["tariff", "custom", "compliance", "sanction", "hs-8542",
"hs code", "duty", "regulation", "legal", "law", "exempt"],
"logistics_routing": ["route", "reroute", "transit", "port", "shipping", "delay",
"eta", "logistics", "vessel", "detour", "canal", "strike",
"blockade", "weather"],
"procurement_sourcing": ["supplier", "vendor", "sourcing", "contract", "sla",
"lead time", "price", "cost", "negotiat", "penalty"],
"risk_analysis": ["risk", "exposure", "impact", "loss", "downtime", "critical"],
}
scores = {domain: sum(1 for kw in kws if kw in q) for domain, kws in domains.items()}
best_domain = max(scores, key=scores.get)
if scores[best_domain] == 0:
best_domain = "general_inquiry"
labels = {
"compliance_legal": "Trade Compliance & Customs",
"logistics_routing": "Logistics & Route Disruption",
"procurement_sourcing": "Procurement & Supplier Sourcing",
"risk_analysis": "Risk & Financial Exposure",
"general_inquiry": "General Supply-Chain Inquiry",
}
return {"domain_key": best_domain, "domain_label": labels[best_domain]}
def confidence_from_distance(l2_distance: float) -> float:
"""Convert a FAISS L2 distance into a 0-100 confidence percentage."""
confidence = 1.0 / (1.0 + max(l2_distance, 0.0))
return round(min(max(confidence * 100.0, 0.0), 100.0), 1)
def confidence_badge_color(confidence: float) -> str:
if confidence >= 70:
return "#16a34a" # green
if confidence >= CONFIDENCE_ESCALATION_THRESHOLD:
return "#d97706" # amber
return "#dc2626" # red
def risk_badge_color(risk_level: str) -> str:
mapping = {
"LOW": "#16a34a",
"MEDIUM": "#d97706",
"HIGH": "#ea580c",
"CRITICAL": "#dc2626",
}
return mapping.get(risk_level.upper(), "#6b7280")
# ======================================================================
# EMBEDDING + VECTOR STORE
# ======================================================================
class EmbeddingService:
_instance: "EmbeddingService | None" = None
def __init__(self) -> None:
self.embeddings = HuggingFaceEmbeddings(
model_name=EMBEDDING_MODEL_NAME,
model_kwargs={"device": EMBEDDING_DEVICE},
)
@classmethod
def get_instance(cls) -> "EmbeddingService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
class VectorStoreManager:
"""In-memory FAISS knowledge base seeded from SEED_DOCUMENTS, with
runtime document ingestion support (file upload)."""
def __init__(self) -> None:
self.embedding_service = EmbeddingService.get_instance()
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP
)
self.vector_db: FAISS | None = None
self.indexed_files: list[str] = []
self._build_seed_index()
def _build_seed_index(self) -> None:
docs: list[Document] = []
for filename, text in SEED_DOCUMENTS.items():
chunks = self.text_splitter.split_text(text.strip())
for i, chunk in enumerate(chunks):
docs.append(Document(page_content=chunk, metadata={"source": filename, "chunk": i}))
self.vector_db = FAISS.from_documents(docs, self.embedding_service.embeddings)
self.indexed_files = list(SEED_DOCUMENTS.keys())
def add_uploaded_file(self, file_path: str) -> str:
if not file_path or not os.path.exists(file_path):
return "No valid file supplied."
filename = os.path.basename(file_path)
ext = filename.lower().rsplit(".", 1)[-1] if "." in filename else ""
try:
if ext == "pdf":
if not PDF_SUPPORT:
return "PDF support is unavailable in this deployment (pypdf missing)."
reader = PdfReader(file_path)
text = "\n".join((page.extract_text() or "") for page in reader.pages)
elif ext in {"txt", "md"}:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
text = f.read()
else:
return f"Unsupported file type '.{ext}'. Please upload .txt, .md, or .pdf."
if not text.strip():
return f"No extractable text found in '{filename}'."
chunks = self.text_splitter.split_text(text)
docs = [
Document(page_content=chunk, metadata={"source": filename, "chunk": i})
for i, chunk in enumerate(chunks)
]
if self.vector_db is None:
self.vector_db = FAISS.from_documents(docs, self.embedding_service.embeddings)
else:
self.vector_db.add_documents(docs)
if filename not in self.indexed_files:
self.indexed_files.append(filename)
return f"Indexed '{filename}' into the knowledge base ({len(chunks)} chunks)."
except Exception as exc: # pragma: no cover
return f"Failed to index '{filename}': {exc}"
def search(self, query: str, k: int = RETRIEVAL_K) -> list[tuple[Document, float]]:
if self.vector_db is None:
return []
return self.vector_db.similarity_search_with_score(query, k=k)
# ======================================================================
# WEB SEARCH AGENT
# ======================================================================
def run_web_search(query: str, max_results: int = MAX_WEB_RESULTS) -> list[dict[str, str]]:
"""Web Intelligence Agent: triggered only when local KB confidence is
insufficient. Uses DuckDuckGo (no API key required)."""
if not WEB_SEARCH_AVAILABLE:
return []
try:
with DDGS() as ddgs:
raw_results = list(ddgs.text(query, max_results=max_results))
results = []
for r in raw_results:
results.append({
"title": _safe_str(r.get("title"), "Untitled result"),
"href": _safe_str(r.get("href"), ""),
"body": _safe_str(r.get("body"), ""),
})
return results
except Exception:
return []
# ======================================================================
# GROQ REASONING AGENT
# ======================================================================
def get_groq_client() -> OpenAI:
api_key = os.environ.get("GROQ_API_KEY", "").strip()
if not api_key:
raise RuntimeError(
"GROQ_API_KEY is not configured. Add it under this Space's Settings -> "
"Repository secrets, then restart the Space."
)
return OpenAI(api_key=api_key, base_url=GROQ_BASE_URL)
def run_reasoning_agent(
query: str,
combined_context: str,
conversation_memory: list[dict[str, str]] | None = None,
) -> str:
"""Compliance Reasoning Agent -- the single remote (paid) call in the
pipeline. `conversation_memory` carries a short window of prior
user/assistant turns so follow-up questions ("what about the Vietnam
option instead?") resolve correctly without re-stating context."""
client = get_groq_client()
messages: list[dict[str, str]] = [{"role": "system", "content": SYSTEM_PROMPT}]
for turn in (conversation_memory or []):
role = turn.get("role")
content = turn.get("content")
if role in {"user", "assistant"} and content:
messages.append({"role": role, "content": content})
messages.append(
{"role": "user", "content": f"CONTEXT:\n{combined_context}\n\nDECISION-MAKER QUERY:\n{query}"}
)
completion = client.chat.completions.create(
model=GROQ_MODEL,
messages=messages,
temperature=GROQ_TEMPERATURE,
max_tokens=GROQ_MAX_TOKENS,
)
return completion.choices[0].message.content or ""
def parse_structured_reply(raw_reply: str) -> dict[str, str]:
"""Split the LLM's ## Analysis / ## Risk Assessment / ## Recommendation reply."""
sections = {"analysis": "", "risk_assessment": "", "recommendation": ""}
pattern = re.split(r"##\s*(Analysis|Risk Assessment|Recommendation)\s*\n", raw_reply, flags=re.IGNORECASE)
if len(pattern) >= 3:
it = iter(pattern[1:])
for header, body in zip(it, it):
key = header.strip().lower().replace(" ", "_")
if key in sections:
sections[key] = body.strip()
else:
sections["analysis"] = raw_reply.strip()
risk_level = "MEDIUM"
match = re.search(r"\b(LOW|MEDIUM|HIGH|CRITICAL)\b", sections["risk_assessment"].upper())
if match:
risk_level = match.group(1)
sections["risk_level"] = risk_level
return sections
# ======================================================================
# ORCHESTRATOR (MULTI-AGENT PIPELINE)
# ======================================================================
@dataclass
class AgentStep:
name: str
status: str # done | active | skipped | error
detail: str
tokens: str = "0 tokens (local)"
@dataclass
class PipelineResult:
steps: list[AgentStep] = field(default_factory=list)
kb_hits: list[tuple[Document, float]] = field(default_factory=list)
web_hits: list[dict[str, str]] = field(default_factory=list)
kb_confidence: float = 0.0
source_mode: str = "Knowledge Base"
analysis: str = ""
risk_level: str = "MEDIUM"
risk_text: str = ""
recommendation: str = ""
error: str | None = None
def run_pipeline(
query: str,
vsm: VectorStoreManager,
conversation_memory: list[dict[str, str]] | None = None,
) -> PipelineResult:
result = PipelineResult()
# --- Agent 1: Scanner (local, 0 tokens) ---
classification = classify_task_locally(query)
memory_note = (
f" Carrying forward {len(conversation_memory) // 2} prior exchange(s) as conversation memory."
if conversation_memory else ""
)
result.steps.append(AgentStep(
name="Scanner Agent",
status="done",
detail=f"Classified query as **{classification['domain_label']}** using local keyword routing.{memory_note}",
))
# --- Agent 2: Knowledge Retrieval (local, 0 tokens) ---
try:
hits = vsm.search(query, k=RETRIEVAL_K)
except Exception as exc:
result.error = f"Knowledge base search failed: {exc}"
result.steps.append(AgentStep("Retrieval Agent", "error", result.error))
return result
result.kb_hits = hits
max_conf = max((confidence_from_distance(dist) for _, dist in hits), default=0.0)
result.kb_confidence = max_conf
result.steps.append(AgentStep(
name="Retrieval Agent",
status="done",
detail=(
f"Retrieved {len(hits)} chunk(s) from the local FAISS knowledge base. "
f"Top match confidence: **{max_conf:.1f}%**."
),
))
# --- Agent 3: Escalation gate (local, 0 tokens) ---
needs_web = max_conf < CONFIDENCE_ESCALATION_THRESHOLD or len(hits) == 0
if needs_web:
result.steps.append(AgentStep(
name="Escalation Agent",
status="done",
detail=(
f"Confidence {max_conf:.1f}% is below the "
f"{CONFIDENCE_ESCALATION_THRESHOLD:.0f}% threshold — escalating to "
"Web Intelligence Agent."
),
))
else:
result.steps.append(AgentStep(
name="Escalation Agent",
status="done",
detail=(
f"Confidence {max_conf:.1f}% clears the "
f"{CONFIDENCE_ESCALATION_THRESHOLD:.0f}% threshold — knowledge base "
"alone is sufficient, skipping web search."
),
))
# --- Agent 4: Web Intelligence (optional) ---
web_context_blocks: list[str] = []
if needs_web:
web_hits = run_web_search(query)
result.web_hits = web_hits
if web_hits:
for w in web_hits:
web_context_blocks.append(f"[Web: {w['title']}] ({w['href']})\n{w['body']}")
result.steps.append(AgentStep(
name="Web Intelligence Agent",
status="done",
detail=f"Retrieved {len(web_hits)} supporting web result(s) to fill the knowledge gap.",
tokens="external search (no LLM tokens)",
))
else:
result.steps.append(AgentStep(
name="Web Intelligence Agent",
status="skipped",
detail="Web search returned no usable results or is unavailable in this environment.",
tokens="external search (no LLM tokens)",
))
else:
result.steps.append(AgentStep(
name="Web Intelligence Agent",
status="skipped",
detail="Not triggered — local knowledge base confidence was sufficient.",
))
if hits and web_context_blocks:
result.source_mode = "Knowledge Base + Web Search"
elif web_context_blocks:
result.source_mode = "Web Search"
else:
result.source_mode = "Knowledge Base"
# --- Agent 5: Compliance Reasoning (remote Groq call) ---
kb_context_blocks = [
f"[KB: {doc.metadata.get('source', 'unknown')} — chunk {doc.metadata.get('chunk', 0)}]\n{doc.page_content}"
for doc, _ in hits
]
combined_context = "\n\n".join(kb_context_blocks + web_context_blocks)
if not combined_context.strip():
combined_context = "No supporting context was retrieved from the knowledge base or the web."
try:
raw_reply = run_reasoning_agent(query, combined_context, conversation_memory)
parsed = parse_structured_reply(raw_reply)
result.analysis = parsed["analysis"]
result.risk_text = parsed["risk_assessment"]
result.risk_level = parsed["risk_level"]
result.recommendation = parsed["recommendation"]
result.steps.append(AgentStep(
name="Compliance Reasoning Agent",
status="done",
detail=f"Synthesized final answer via Fireworks/Groq Llama-3.3-70B ({result.source_mode}).",
tokens="remote call (Groq API)",
))
except Exception as exc:
result.error = str(exc)
result.steps.append(AgentStep(
name="Compliance Reasoning Agent",
status="error",
detail=f"Reasoning call failed: {exc}",
tokens="remote call (Groq API)",
))
return result
# ======================================================================
# RENDERING HELPERS (HTML/Markdown for the UI panels)
# ======================================================================
STATUS_ICON = {"done": "✅", "active": "🟢", "skipped": "⏭️", "error": "❌"}
def render_agent_timeline(steps: list[AgentStep]) -> str:
rows = []
for step in steps:
icon = STATUS_ICON.get(step.status, "•")
rows.append(
f"""
"""
)
return f'{"".join(rows)}
'
def render_confidence_gauge(confidence: float, source_mode: str) -> str:
color = confidence_badge_color(confidence)
return f"""
Knowledge Base Confidence
{confidence:.1f}%
Source: {source_mode}
"""
def render_citations(kb_hits: list[tuple[Document, float]], web_hits: list[dict[str, str]]) -> str:
if not kb_hits and not web_hits:
return "_No sources were retrieved for this query yet._"
blocks = []
if kb_hits:
blocks.append("### 📚 Knowledge Base Sources")
for i, (doc, dist) in enumerate(kb_hits, start=1):
conf = confidence_from_distance(dist)
source = doc.metadata.get("source", "unknown")
chunk = doc.metadata.get("chunk", 0)
snippet = doc.page_content.strip().replace("\n", " ")
if len(snippet) > 280:
snippet = snippet[:280].rstrip() + "…"
blocks.append(
f"**{i}. `{source}`** (chunk {chunk}) — confidence `{conf:.1f}%`\n\n> {snippet}\n"
)
if web_hits:
blocks.append("### 🌐 Web Search Sources")
for i, w in enumerate(web_hits, start=1):
blocks.append(f"**{i}. [{w['title']}]({w['href']})**\n\n> {w['body'][:280]}\n")
return "\n".join(blocks)
def render_recommendation_panel(result: PipelineResult) -> str:
if result.error and not result.recommendation:
return f"""
⚠️ Unable to Complete Analysis
{result.error}
"""
color = risk_badge_color(result.risk_level)
return f"""
✅ Final Recommendation
{result.recommendation or "No recommendation was generated."}
"""
def render_kb_files(files: list[str]) -> str:
items = "".join(f"{f}" for f in files)
return f'Active knowledge base files: '
# ======================================================================
# GRADIO APPLICATION
# ======================================================================
CUSTOM_CSS = """
:root {
--nx-primary: #2563eb;
--nx-primary-dark: #1d4ed8;
--nx-bg: #f8fafc;
--nx-card: #ffffff;
--nx-border: #e2e8f0;
--nx-text: #0f172a;
--nx-muted: #64748b;
}
.gradio-container {
background: var(--nx-bg) !important;
font-family: 'Inter', 'Segoe UI', system-ui, sans-serif !important;
}
#nx-header {
background: linear-gradient(135deg, #1e3a8a 0%, #2563eb 55%, #38bdf8 100%);
border-radius: 18px;
padding: 28px 32px;
color: white;
margin-bottom: 18px;
box-shadow: 0 10px 30px -12px rgba(37, 99, 235, 0.45);
}
#nx-header h1 {
margin: 0 0 6px 0;
font-size: 1.7em;
font-weight: 800;
letter-spacing: -0.02em;
}
#nx-header p {
margin: 0;
opacity: 0.92;
font-size: 1.0em;
}
.nx-header-badges {
margin-top: 14px;
display: flex;
gap: 8px;
flex-wrap: wrap;
}
.nx-badge {
background: rgba(255,255,255,0.18);
border: 1px solid rgba(255,255,255,0.35);
border-radius: 999px;
padding: 4px 12px;
font-size: 0.78em;
font-weight: 600;
}
.nx-card {
background: var(--nx-card);
border: 1px solid var(--nx-border);
border-radius: 14px;
padding: 18px;
box-shadow: 0 1px 3px rgba(15, 23, 42, 0.04);
}
.agent-timeline { display: flex; flex-direction: column; gap: 8px; }
.agent-step {
border-left: 3px solid var(--nx-border);
border-radius: 8px;
background: #f8fafc;
padding: 8px 12px;
}
.agent-step-done { border-left-color: #16a34a; }
.agent-step-error { border-left-color: #dc2626; }
.agent-step-skipped { border-left-color: #94a3b8; opacity: 0.75; }
.agent-step-header {
display: flex;
align-items: center;
gap: 8px;
font-weight: 700;
font-size: 0.9em;
color: var(--nx-text);
}
.agent-step-name { flex: 1; }
.agent-step-tokens {
font-size: 0.72em;
color: var(--nx-muted);
background: #eef2f7;
border-radius: 999px;
padding: 2px 8px;
}
.agent-step-detail {
margin-top: 4px;
font-size: 0.85em;
color: var(--nx-muted);
line-height: 1.4;
}
.confidence-card { text-align: center; }
.confidence-label { font-size: 0.85em; color: var(--nx-muted); font-weight: 600; margin-bottom: 8px; }
.confidence-bar-track {
width: 100%; height: 10px; border-radius: 999px; background: #e2e8f0; overflow: hidden;
}
.confidence-bar-fill { height: 100%; border-radius: 999px; transition: width 0.6s ease; }
.confidence-value { font-size: 1.6em; font-weight: 800; margin-top: 8px; }
.source-badge {
display: inline-block; margin-top: 6px; font-size: 0.78em; color: var(--nx-muted);
background: #f1f5f9; border-radius: 999px; padding: 3px 10px;
}
.recommendation-card {
border-radius: 14px;
padding: 18px;
background: linear-gradient(135deg, #ecfdf5 0%, #f0fdfa 100%);
border: 1px solid #a7f3d0;
}
.recommendation-error { background: #fef2f2; border-color: #fecaca; }
.recommendation-header { display: flex; gap: 8px; margin-bottom: 10px; }
.risk-badge {
color: white; font-weight: 800; font-size: 0.72em; padding: 4px 10px; border-radius: 999px;
letter-spacing: 0.03em;
}
.source-tag {
background: #e2e8f0; color: #334155; font-size: 0.72em; font-weight: 700;
padding: 4px 10px; border-radius: 999px;
}
.recommendation-title { font-weight: 800; font-size: 1.05em; margin-bottom: 6px; color: var(--nx-text); }
.recommendation-body { color: #1e293b; line-height: 1.5; font-size: 0.95em; white-space: pre-wrap; }
.kb-file-list { font-size: 0.85em; color: var(--nx-muted); }
.kb-file-list ul { margin: 6px 0 0 18px; padding: 0; }
.nx-footer {
text-align: center;
color: var(--nx-muted);
font-size: 0.82em;
padding: 18px 8px 6px 8px;
}
button.primary {
background: linear-gradient(135deg, var(--nx-primary) 0%, var(--nx-primary-dark) 100%) !important;
border: none !important;
transition: transform 0.15s ease, box-shadow 0.15s ease !important;
}
button.primary:hover {
transform: translateY(-1px);
box-shadow: 0 8px 18px -6px rgba(37, 99, 235, 0.55) !important;
}
@media (max-width: 768px) {
#nx-header { padding: 20px; }
#nx-header h1 { font-size: 1.3em; }
}
"""
EXAMPLE_PROMPTS = [
"Does HS-8542 cargo face customs tariff exemptions transiting through Singapore?",
"What is the delay penalty under SLA-902 for the Penang facility?",
"How many extra transit days does the Cape of Good Hope bypass add versus Suez?",
"The Suez Canal has just been closed by a port strike — what should we do about our Kaohsiung to Hamburg shipment?",
"Compare the risk of sourcing from Vietnam (SLA-233) versus Malaysia (SLA-902) during a regional disruption.",
]
def build_app() -> gr.Blocks:
vsm = VectorStoreManager()
with gr.Blocks(title=APP_TITLE) as demo:
gr.HTML(
f"""
"""
)
with gr.Row():
# ---------------- LEFT COLUMN: Chat + Input ----------------
with gr.Column(scale=6):
with gr.Group(elem_classes=["nx-card"]):
chatbot = gr.Chatbot(
label="Nexus-Pathfinder Copilot",
height=420,
avatar_images=(None, None),
)
with gr.Row():
query_box = gr.Textbox(
placeholder="Ask about tariffs, supplier SLAs, disruptions, or routing risk…",
label="Your question",
scale=5,
lines=2,
)
submit_btn = gr.Button("🚀 Analyze", variant="primary", scale=1)
with gr.Row():
clear_btn = gr.Button("🗑️ Clear conversation", size="sm")
gr.Examples(examples=EXAMPLE_PROMPTS, inputs=[query_box], label="💡 Example prompts")
with gr.Accordion("📥 Add a document to the knowledge base", open=False, elem_classes=["nx-card"]):
file_upload = gr.File(
label="Upload a .txt, .md, or .pdf compliance/SLA document",
file_types=[".txt", ".md", ".pdf"],
type="filepath",
)
upload_status = gr.Markdown("Status: ready to receive a document.")
kb_files_panel = gr.HTML(render_kb_files(vsm.indexed_files))
# ---------------- RIGHT COLUMN: Agent Intelligence ----------------
with gr.Column(scale=5):
with gr.Group(elem_classes=["nx-card"]):
gr.Markdown("### 🤖 Live Agent Activity")
agent_panel = gr.HTML(
'Awaiting your first query…
'
)
with gr.Group(elem_classes=["nx-card"]):
confidence_panel = gr.HTML(render_confidence_gauge(0.0, "—"))
with gr.Group(elem_classes=["nx-card"]):
recommendation_panel = gr.HTML(
''
"✅ Final Recommendation
"
"Run a query to see the AI's recommendation here.
Awaiting your first query…
',
render_confidence_gauge(0.0, "—"),
gr.update(),
"_No sources retrieved yet._",
"_No analysis generated yet._",
)
return
# Conversation memory: last few completed exchanges, captured
# BEFORE we append the new (not-yet-answered) turn below.
memory_turns = [m for m in history if m.get("role") in {"user", "assistant"}]
memory_turns = memory_turns[-(MAX_MEMORY_TURNS * 2):]
history.append({"role": "user", "content": query})
history.append({"role": "assistant", "content": "⏳ Coordinating agents — analyzing your request…"})
yield (
history, "",
'🟢 Scanner Agent classifying task…
',
render_confidence_gauge(0.0, "—"),
gr.update(),
"_Retrieving sources…_",
"_Generating analysis…_",
)
try:
result = run_pipeline(query, vsm, memory_turns)
except Exception as exc: # pragma: no cover
error_msg = f"Unexpected pipeline error: {exc}\n{traceback.format_exc(limit=1)}"
history[-1] = {"role": "assistant", "content": f"❌ {error_msg}"}
yield (
history, "",
'Pipeline failed before completion.
',
render_confidence_gauge(0.0, "—"),
render_recommendation_panel(PipelineResult(error=error_msg)),
"_No sources retrieved due to an error._",
"_No analysis available due to an error._",
)
return
if result.error and not result.recommendation:
answer_text = f"⚠️ {result.error}"
gr.Warning(result.error)
else:
answer_text = (
f"**Risk level: {result.risk_level}**\n\n"
f"{result.recommendation}\n\n"
f"_Source: {result.source_mode} · Confidence: {result.kb_confidence:.1f}%_"
)
gr.Info(f"Analysis complete — source: {result.source_mode}, risk: {result.risk_level}.")
history[-1] = {"role": "assistant", "content": answer_text}
reasoning_md = result.analysis or "_No detailed analysis was generated._"
if result.risk_text:
reasoning_md += f"\n\n**Risk detail:** {result.risk_text}"
yield (
history, "",
render_agent_timeline(result.steps),
render_confidence_gauge(result.kb_confidence, result.source_mode),
render_recommendation_panel(result),
render_citations(result.kb_hits, result.web_hits),
reasoning_md,
)
submit_btn.click(
on_submit,
inputs=[query_box, chatbot],
outputs=[chatbot, query_box, agent_panel, confidence_panel, recommendation_panel,
citations_panel, reasoning_panel],
)
query_box.submit(
on_submit,
inputs=[query_box, chatbot],
outputs=[chatbot, query_box, agent_panel, confidence_panel, recommendation_panel,
citations_panel, reasoning_panel],
)
def on_clear():
return (
[], "",
'Awaiting your first query…
',
render_confidence_gauge(0.0, "—"),
''
"✅ Final Recommendation
"
"Run a query to see the AI's recommendation here.