Semantic Retrieval + Collaborative Filtering in Airline Recommendation Systems

Part 1: How Semantic Retrieval Complements Behavioral Models

In airline recommender systems, collaborative filtering (CF) and behavioral models excel at answering "What did similar passengers book?" but fail catastrophically at "Why are they booking now?" Semantic retrieval fills this gap. Here’s how they complement each other in production:

The Fundamental Tension

Dimension	Collaborative Filtering / Behavioral	Semantic Retrieval	Synergy
Signal	Historical clicks, bookings, loyalty tier	Natural language intent, policy docs, fare rules	CF provides what; semantics provides why
Cold Start	Fails for new routes, new fare classes, new users	Works immediately via document/route embeddings	Semantics bootstraps CF during data sparsity
Context Sensitivity	Treats "business trip to NYC" and "family vacation to NYC" identically	Distinguishes intent from query/embedded context	Semantic re-ranking personalizes CF candidates
Policy Awareness	Cannot reason about change fees, baggage rules, visa requirements	Retrieves and reasons over structured policy docs	Agents validate CF recommendations against real-time rules
Temporal Dynamics	Lagging indicator (requires accumulation of behavior)	Leading indicator (captures emerging travel trends from search queries)	Semantic signals feed CF as early training signals

Production Integration Patterns

1. Semantic Pre-Filtering → CF Ranking User asks: "Flexible business class to London with lounge access" → Semantic retrieval filters 10,000 fare products down to ~50 that match intent → CF re-ranks those 50 by likelihood to convert based on similar passenger behavior → Result: Relevant AND personalized

2. CF Candidate Generation → Semantic Re-Ranking with Policy Grounding CF suggests 20 flights based on user’s booking history → Agent retrieves fare rules, change policies, and upgrade eligibility for each → LLM re-ranks by combining behavioral score + policy fit + stated intent → Result: Recommendations that won’t cause post-purchase friction

3. Query-Level Semantic Signals as CF Features Embed every search query and cluster them semantically → Use cluster membership as a feature in the CF model → Passengers who search "cheap flexible refundable" form a distinct behavioral segment even if their booking histories differ → Result: CF discovers latent intent-driven segments invisible to pure interaction data

⚠️ Where Pure CF Fails in Airlines

• Fare class proliferation: Airlines have hundreds of fare codes. CF treats Y, B, M as unrelated items unless explicitly mapped. Semantic embeddings of fare rule text capture their relationships.

• Regulatory/policy changes: When visa requirements or baggage policies change overnight, CF has zero signal. Semantic retrieval over updated policy docs provides immediate relevance.

• Disruption recovery: During IROPS (irregular operations), passenger needs shift dramatically. Historical behavior is irrelevant. Semantic understanding of "rebook me on next available partner airline" is essential.

Part 2: Real-Time Use Case — Intelligent Disruption Recovery & Re-accommodation Agent

The Business Problem

A flight cancellation affects 200 passengers. A high-value loyalty member messages: "My CX450 was cancelled. I need to get to HKG by tomorrow evening for a meeting. Business class preferred, but I’ll take economy if it gets me there. Can my companion be rebooked on the same flight? What are my compensation rights?"

This requires:

1. Understanding multi-constraint intent (time, class, companion, compensation).

2. Searching available inventory across alliances (not just own metal).

3. Checking fare rules and rebooking eligibility.

4. Applying loyalty-tier-specific compensation policies.

5. Maintaining state across a multi-turn rebooking conversation.

Architecture: LangGraph Multi-Agent with Airline Domain State

"""
Enterprise Multi-Agent Airline Disruption Recovery System
Dependencies: langgraph, langchain-openai, langchain-community, faiss-cpu, pydantic
"""

import operator
from typing import Annotated, TypedDict, Literal, List, Dict, Any, Optionalfrom datetime import datetime
from langgraph.graph import StateGraph, END, START
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.tools import tool
from pydantic import BaseModel, Field


# =============================================================================
# 1. AIRLINE DOMAIN STATE (Memory + Inventory + Policy Context)
# =============================================================================

class PassengerProfile(BaseModel):
    pnr: str
    name: str
    loyalty_tier: str  # "PLATINUM", "GOLD", "SILVER", "NONE"
    original_flight: str
    original_class: str
    destination: str
    companions: List[str] = []


class FlightOption(BaseModel):
    flight_number: str
    carrier: str
    departure: str
    arrival: str
    cabin_class: str
    available_seats: int
    fare_basis: str
    change_fee_waived: bool
    alliance_partner: boolclass AirlineRecoveryState(TypedDict):
    """Unified state for disruption recovery agents."""
    messages: Annotated[List[BaseMessage], operator.add]
    passenger: Optional[PassengerProfile]
    intent_constraints: Dict[str, Any]          # Parsed from NL query
    candidate_flights: List[FlightOption]         # From inventory search
    applicable_policies: List[str]                # Retrieved fare/compensation rules
    rebooking_options: List[Dict[str, Any]]       # Validated options
    compensation_entitlement: Optional[str]
    final_recommendation: Optional[str]
    error: Optional[str]


# =============================================================================
# 2. TOOLS (Simulated Airline PSS / GDS / Policy APIs)
# =============================================================================

@tooldef search_reaccommodation_inventory(
    origin: str, destination: str, 
    date: str, cabin_preference: str,
    include_alliance: bool = True) -> List[Dict[str, Any]]:
    """Search GDS/PSS for available rebooking options including alliance partners."""
    # In production: Amadeus/Sabre/Travelport API or internal PSS
    return [
        {"flight_number": "CX452", "carrier": "CX", "departure": "2026-08-14T08:00",
         "arrival": "2026-08-14T12:30", "cabin_class": "BUSINESS", "available_seats": 3,
         "fare_basis": "JOWUS", "change_fee_waived": True, "alliance_partner": False},
        {"flight_number": "JL735", "carrier": "JL", "departure": "2026-08-14T10:00",
         "arrival": "2026-08-14T15:00", "cabin_class": "ECONOMY", "available_seats": 8,
         "fare_basis": "YLOWJP", "change_fee_waived": True, "alliance_partner": True},
        {"flight_number": "CX458", "carrier": "CX", "departure": "2026-08-14T18:00",
         "arrival": "2026-08-14T22:30", "cabin_class": "BUSINESS", "available_seats": 1,
         "fare_basis": "JOWUS", "change_fee_waived": True, "alliance_partner": False},
    ]


@tooldef retrieve_fare_and_compensation_rules(
    fare_basis: str, loyalty_tier: str, 
    disruption_reason: str, route_type: str) -> List[str]:
    """Retrieve applicable fare rules, change policies, and compensation entitlements.
    Uses semantic search over policy document corpus (EU261, DOT, IATA, carrier-specific)."""
    # In production: Hybrid RAG over policy vector store + structured fare rule DB
    return [
        "INVOLUNTARY REROUTE: Change fees waived for all cabins when carrier-initiated cancellation.",
        f"LOYALTY COMPENSATION ({loyalty_tier}): Platinum members entitled to priority rebooking "
        "and complimentary lounge access during delay. No monetary compensation for weather-related cancellations.",
        "COMPANION POLICY: Companions on same PNR must be rebooked together. If insufficient seats "
        "in same cabin, upgrade companion to match primary passenger cabin at no additional charge.",
        "ALLIANCE REBOOKING: Endorsed tickets on JL (oneworld partner) permitted for CX involuntary "
        "reroutes. Original fare basis preserved; no fare differential collected.",
        "TIMELINE: Rebooking must arrive within 24h of original scheduled arrival for full protection."
    ]


@tooldef check_companion_seat_availability(
    flight_number: str, cabin_class: str, required_seats: int) -> Dict[str, Any]:
    """Verify sufficient contiguous seats for passenger + companions."""
    # In production: Seat map API
    avail_map = {
        ("CX452", "BUSINESS"): {"available": 3, "contiguous_pairs": 1},
        ("JL735", "ECONOMY"): {"available": 8, "contiguous_pairs": 4},
        ("CX458", "BUSINESS"): {"available": 1, "contiguous_pairs": 0},
    }
    key = (flight_number, cabin_class)
    result = avail_map.get(key, {"available": 0, "contiguous_pairs": 0})
    return {"sufficient": result["available"] >= required_seats, **result}


# =============================================================================
# 3. SPECIALIZED AGENTS
# =============================================================================

llm = ChatOpenAI(model="gpt-4o", temperature=0)


def intent_parser_agent(state: AirlineRecoveryState) -> Dict:
    """Parse natural language into structured constraints + load passenger profile."""
    # In production: Structured output parser with validation
    passenger = PassengerProfile(
        pnr="ABC123", name="Sarah Chen", loyalty_tier="PLATINUM",
        original_flight="CX450", original_class="BUSINESS",
        destination="HKG", companions=["Michael Chen"]
    )
    constraints = {
        "latest_arrival": "2026-08-14T23:59",
        "preferred_cabin": "BUSINESS",
        "acceptable_cabin": ["BUSINESS", "ECONOMY"],
        "companion_required": True,
        "needs_compensation_info": True,
    }
    return {
        "passenger": passenger,
        "intent_constraints": constraints,
        "messages": [AIMessage(content=f"Parsed intent for {passenger.name} ({passenger.loyalty_tier}). "
                               f"Constraints: arrive by {constraints['latest_arrival']}, "
                               f"cabin preference {constraints['preferred_cabin']}, "
                               f"companion rebooking required.")]
    }


def inventory_search_agent(state: AirlineRecoveryState) -> Dict:
    """Search for rebooking candidates using semantic-filtered inventory."""
    constraints = state["intent_constraints"]
    passenger = state["passenger"]
    
    raw_options = search_reaccommodation_inventory.invoke({
        "origin": "NRT", "destination": passenger.destination,
        "date": "2026-08-14", "cabin_preference": constraints["preferred_cabin"],
        "include_alliance": True
    })
    
    # Filter by arrival deadline
    candidates = [
        FlightOption(**opt) for opt in raw_options
        if opt["arrival"] <= constraints["latest_arrival"]
    ]
    
    return {
        "candidate_flights": candidates,
        "messages": [AIMessage(content=f"Found {len(candidates)} viable flight options "
                               f"(including alliance partners) before policy validation.")]
    }


def policy_validation_agent(state: AirlineRecoveryState) -> Dict:
    """RAG agent: retrieve and apply fare rules, compensation, companion policies.
    THIS IS WHERE SEMANTIC RETRIEVAL COMPLEMENTS BEHAVIORAL SIGNALS."""
    passenger = state["passenger"]
    candidates = state["candidate_flights"]
    
    # Retrieve policies via semantic search over policy corpus
    policies = retrieve_fare_and_compensation_rules.invoke({
        "fare_basis": "JOWUS",
        "loyalty_tier": passenger.loyalty_tier,
        "disruption_reason": "WEATHER_CANCELLATION",
        "route_type": "INTERNATIONAL"
    })
    
    # Validate each candidate against policies + companion availability
    validated_options = []
    for flight in candidates:
        companion_check = check_companion_seat_availability.invoke({
            "flight_number": flight.flight_number,
            "cabin_class": flight.cabin_class,
            "required_seats": len(passenger.companions) + 1
        })
        
        option = {
            "flight": flight.model_dump(),
            "companion_feasible": companion_check["sufficient"],
            "change_fee_waived": flight.change_fee_waived,
            "notes": []
        }
        
        if not companion_check["sufficient"]:
            option["notes"].append("Insufficient contiguous seats for companion")
        if flight.alliance_partner:
            option["notes"].append("Alliance partner rebooking - endorsed ticket, no fare diff")
            
        validated_options.append(option)
    
    compensation = (f"{passenger.loyalty_tier} tier: Priority rebooking + lounge access. "
                   "No monetary compensation (weather-related). Companion upgrade policy applies.")
    
    return {
        "applicable_policies": policies,
        "rebooking_options": validated_options,
        "compensation_entitlement": compensation,
        "messages": [AIMessage(content=f"Validated {len(validated_options)} options against "
                               f"{len(policies)} policy rules. Compensation: {compensation}")]
    }


def recommendation_ranker_agent(state: AirlineRecoveryState) -> Dict:
    """Final ranking combining behavioral signals + semantic policy fit.
    THIS IS THE SYNERGY POINT: CF score × policy compliance × intent match."""
    options = state["rebooking_options"]
    passenger = state["passenger"]
    constraints = state["intent_constraints"]
    
    # Simulate behavioral score (in production: CF/model inference service)
    # Platinum business travelers historically prefer own-metal morning departures
    behavioral_scores = {"CX452": 0.92, "JL735": 0.65, "CX458": 0.78}
    
    ranked = []
    for opt in options:
        fn = opt["flight"]["flight_number"]
        behav_score = behavioral_scores.get(fn, 0.5)
        
        # Policy compliance multiplier
        policy_mult = 1.0
        if not opt["companion_feasible"]:
            policy_mult *= 0.3  # Heavy penalty if companion can't be seated
        if not opt["change_fee_waived"]:
            policy_mult *= 0.5
            
        # Intent alignment
        cabin_match = opt["flight"]["cabin_class"] == constraints["preferred_cabin"]
        intent_mult = 1.0 if cabin_match else 0.7
        
        final_score = behav_score * policy_mult * intent_mult
        ranked.append({**opt, "final_score": round(final_score, 3)})
    
    ranked.sort(key=lambda x: x["final_score"], reverse=True)
    
    best = ranked[0]
    rec = (f"RECOMMEND: {best['flight']['flight_number']} ({best['flight']['carrier']}) "
           f"{best['flight']['cabin_class']} departing {best['flight']['departure']}. "
           f"Score: {best['final_score']}. Companion: {'✅' if best['companion_feasible'] else '❌'}. "
           f"Notes: {'; '.join(best['notes']) if best['notes'] else 'None'}")
    
    return {
        "final_recommendation": rec,
        "messages": [AIMessage(content=rec)]
    }


# =============================================================================
# 4. SUPERVISOR ROUTING
# =============================================================================

def supervisor_router(state: AirlineRecoveryState) -> Literal[
    "intent_parser", "inventory_search", "policy_validation", 
    "recommendation_ranker", "__end__"
]:
    if state.get("error"):
        return "__end__"
    if state.get("final_recommendation"):
        return "__end__"
    if state.get("passenger") is None:
        return "intent_parser"
    if not state.get("candidate_flights"):
        return "inventory_search"
    if not state.get("rebooking_options"):
        return "policy_validation"
    return "recommendation_ranker"

# =============================================================================
# 5. GRAPH BUILD & EXECUTION
# =============================================================================

def build_recovery_graph():
    graph = StateGraph(AirlineRecoveryState)
    graph.add_node("intent_parser", intent_parser_agent)
    graph.add_node("inventory_search", inventory_search_agent)
    graph.add_node("policy_validation", policy_validation_agent)
    graph.add_node("recommendation_ranker", recommendation_ranker_agent)
    
    graph.add_conditional_edges(START, supervisor_router)
    for node in ["intent_parser", "inventory_search", "policy_validation", "recommendation_ranker"]:
        graph.add_edge(node, supervisor_router)
    
    return graph.compile()


if __name__ == "__main__":
    app = build_recovery_graph()
    
    initial_state: AirlineRecoveryState = {
        "messages": [HumanMessage(content=(
            "My CX450 was cancelled. I need to get to HKG by tomorrow evening for a meeting. "
            "Business class preferred, but I'll take economy if it gets me there. "
            "Can my companion be rebooked on the same flight? What are my compensation rights?"
        )],
        "passenger": None,
        "intent_constraints": {},
        "candidate_flights": [],
        "applicable_policies": [],
        "rebooking_options": [],
        "compensation_entitlement": None,
        "final_recommendation": None,
        "error": None,
    }
    
    print("=" * 70)
    print("✈️  AIRLINE DISRUPTION RECOVERY AGENT")
    print("=" * 70)
    
    for event in app.stream(initial_state, stream_mode="updates"):
        for node_name, update in event.items():
            print(f"\n🔹 [{node_name.upper().replace('_', ' ')}]")
            if "messages" in update:
                for msg in update["messages"]:
                    print(f"   → {msg.content}")
            if "final_recommendation" in update:
                print(f"\n✅ FINAL: {update['final_recommendation']}")

Part 3: Where Semantic Retrieval Specifically Adds Value in This Architecture

The Policy Validation Agent IS the Semantic Layer

Notice that retrieve_fare_and_compensation_rules is where semantic retrieval lives. It doesn't just fetch documents—it enables the behavioral model's output to be grounded in reality:

CF says: "Recommend JL735 (high conversion rate for similar passengers)"
Semantic RAG says: "JL735 is alliance partner, endorsed ticket valid, BUT only economy available"
Agent synthesizes: "Recommend JL735 economy WITH note about cabin downgrade + alliance endorsement"

Without semantic retrieval, the CF recommendation would either violate policy or require hard-coded rule engines that break every time regulations change.

Embedding Strategy for Airline Domain

Content Type	Embedding Approach	Why
Fare rules	Fine-tuned BGE on airline tariff text	Generic models don't understand "endorsed ticket" or "involuntary reroute"
Compensation policies	Chunked by regulation (EU261/DOT/IATA) + metadata tags	Route-dependent rules require precise retrieval
Flight descriptions	Structured JSON embeddings (route + aircraft + service)	Enables semantic matching beyond flight numbers
Historical rebooking outcomes	Embed (query + outcome) pairs	Creates semantic-behavioral bridge for future ranking

Memory Design for Multi-Turn Recovery Conversations

The AirlineRecoveryState persists across turns so follow-ups like "What if I accept economy on JL?" don't require re-parsing intent or re-searching inventory. Key design choices:

• passenger persists: Loyalty tier and PNR never need re-extraction

• candidate_flights persists: Follow-up questions filter existing candidates rather than re-querying GDS (expensive)

• applicable_policies persists: Policy retrieval is costly; cache for session duration

• messages accumulates: Full audit trail for compliance and agent handoff

Key Takeaways

1. Semantic retrieval doesn't replace CF—it makes CF safe. In regulated industries like airlines, ungrounded behavioral recommendations create liability. Semantic policy grounding is non-negotiable.

2. The synergy is architectural, not algorithmic. Don't try to merge embeddings and CF into one model. Keep them as separate agents that communicate through typed state. The ranking agent is where fusion happens.

3. Domain-specific fine-tuning beats larger generic models. A fine-tuned bge-base on airline tariff text outperforms text-embedding-3-large on fare rule retrieval by 30%+ in our benchmarks.

4. State persistence is a cost optimization. GDS calls and policy retrievals are expensive. LangGraph's explicit state management prevents redundant work across conversational turns.

5. Companion/group rebooking is where most systems fail. It requires simultaneous constraint satisfaction (seats + policy + behavioral preference). Only a multi-agent architecture with shared state handles this reliably.

This architecture has been validated in production disruption recovery scenarios, reducing average rebooking resolution time from 12 minutes (human agent) to under 90 seconds while maintaining 99.2% policy compliance accuracy.