Enterprise Dynamic Pricing using Reinforcement Learning with Multi-Agent LangGraph and RAG

Dynamic pricing is no longer just about applying static rules or simple heuristic markups. In modern enterprise environments, pricing requires real-time adaptation to market shifts, competitor movements, and inventory constraints. While Reinforcement Learning (RL) excels at mathematical optimization in sequential decision-making, Large Language Models (LLMs) via Multi-Agent frameworks (like LangGraph) excel at unstructured reasoning, policy compliance, and contextual retrieval (RAG). This article explores how to architect an enterprise-grade dynamic pricing system that fuses an RL pricing engine with a Multi-Agent LangGraph orchestrator, complete with RAG, memory, and tool usage.

Part 1: The Reinforcement Learning Formulation

To implement RL for dynamic pricing, we must define the Markov Decision Process (MDP).

1. The State Space

The state represents the environment's current context. In an enterprise setting, this is a mix of internal metrics and external market signals.

Internal State: Current inventory levels, time of day/seasonality, current base price.
External State: Competitor pricing, macroeconomic indicators, real-time demand signals.
Implementation: We normalize and discretize these into a state vector (e.g., [Inventory_Level, Time_Bucket, Competitor_Price_Ratio]).

2. The Action Space

Instead of predicting an absolute price (which can lead to wild, unsafe outputs), the RL agent outputs a price adjustment multiplier.

Discrete Actions: [-10%, -5%, 0%, +5%, +10%] applied to the base price.
Why discrete? It stabilizes training and allows the business to map these actions to specific psychological price tiers (e.g., $99.99, $104.99).

3. The Reward Function

The reward function is the most critical component. It must align the RL agent's goals with business objectives.

Rt = α(Profit) − β(Stockout_Penalty) − γ(Price_Volatility)

Profit: (Price − Cost) × Units_Sold
Stockout Penalty: A heavy negative reward if inventory drops to zero (lost future sales).
Price Volatility: A small penalty for changing prices too drastically between time steps, preserving customer trust.

4. Exploration vs. Exploitation

Exploitation: Choosing the price adjustment that historically yielded the highest reward.
Exploration: Trying a sub-optimal price to discover if market conditions have changed.
Strategy: We use an ϵ-greedy approach with decay. In production, "exploration" is risky. Therefore, we constrain the action space using business guardrails (handled by our LLM agents) so exploration never breaches minimum margin requirements.

Part 2: The Enterprise Architecture (LangGraph + RAG)

An RL model cannot operate in a vacuum. It needs context, and its outputs must be governed by business rules. This is where LangGraph comes in.

We will build a Multi-Agent System with the following roles:

Supervisor Agent: Routes the workflow and manages the global state.
Market Intelligence Agent (RAG): Retrieves real-time competitor data and market news.
RL Pricing Agent (Tool/Engine): Runs the Q-Learning algorithm to determine the optimal price adjustment.
Compliance & Guardrail Agent (RAG): Retrieves company pricing policies to ensure the RL's decision doesn't violate minimum margins or price-matching rules.
Memory & State: LangGraph's Checkpointer will maintain the conversation and transaction state, while a custom Replay Buffer will store past pricing episodes to continuously retrain the RL model offline.

Part 3: Real-Time Use Case & Code Implementation

Use Case

TechNova, an enterprise electronics retailer, needs to dynamically price a high-demand GPU. The system must adjust prices hourly based on competitor moves while strictly adhering to corporate margin policies.

Prerequisites

pip install langgraph langchain-core langchain-community faiss-cpu numpy

The Complete Implementation

Below is the end-to-end implementation. We use a Tabular Q-Learning agent for the RL engine (for lightweight execution) and mock LLMs/RAG to ensure the code runs out-of-the-box without API keys.

import numpy as np
import random
from typing import TypedDict, Annotated, List, Dict, Any
from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.embeddings import FakeEmbeddings

# ==========================================
# 1. The Reinforcement Learning Engine
# ==========================================
class DynamicPricingRL:
    """Tabular Q-Learning Agent for Dynamic Pricing"""
    def __init__(self, n_states=27, n_actions=5, learning_rate=0.1, discount_factor=0.95, epsilon=1.0):
        self.q_table = np.zeros((n_states, n_actions))
        self.lr = learning_rate
        self.gamma = discount_factor
        self.epsilon = epsilon
        self.epsilon_decay = 0.995
        self.epsilon_min = 0.05
        # Actions: -10%, -5%, 0%, +5%, +10%
        self.actions = [-0.10, -0.05, 0.0, 0.05, 0.10]
        self.replay_buffer = []

    def discretize_state(self, inventory_ratio, time_bucket, competitor_price_ratio):
        # Discretize continuous variables into state indices (0-2 for each)
        inv_idx = min(2, int(inventory_ratio * 3))
        time_idx = time_bucket % 3
        comp_idx = min(2, int(competitor_price_ratio * 3))
        return inv_idx * 9 + time_idx * 3 + comp_idx

    def choose_action(self, state_idx):
        if random.uniform(0, 1) < self.epsilon:
            return random.randint(0, len(self.actions) - 1) # Explore
        else:
            return np.argmax(self.q_table[state_idx]) # Exploit

    def update(self, state_idx, action_idx, reward, next_state_idx):
        # Q-Learning update rule
        predict = self.q_table[state_idx, action_idx]
        target = reward + self.gamma * np.max(self.q_table[next_state_idx])
        self.q_table[state_idx, action_idx] += self.lr * (target - predict)

        # Decay epsilon
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

    def save_to_replay_buffer(self, experience):
        self.replay_buffer.append(experience)

# ==========================================
# 2. RAG Setup (Mocked for demonstration)
# ==========================================
class MockRAGSystem:
    def __init__(self):
        self.embeddings = FakeEmbeddings(size=32)
        self.vectorstore = InMemoryVectorStore(self.embeddings)

        # Ingest Competitor Data
        self.vectorstore.add_texts(
            texts=["Competitor A is selling the GPU for $850 today due to high demand."],
            ids=["comp_data_1"]
        )

        # Ingest Company Policy
        self.vectorstore.add_texts(
            texts=["Company Policy: Minimum gross margin must be 15%. Price cannot drop below $750."],
            ids=["policy_1"]
        )

    def get_competitor_price(self, query: str) -> float:
        # Simulated retrieval: extracts price from context or returns default
        return 850.0

    def get_pricing_policy(self, query: str) -> Dict[str, float]:
        # Simulated retrieval: returns policy constraints
        return {"min_margin_pct": 0.15, "absolute_price_floor": 750.0}

# ==========================================
# 3. LangGraph State and Tools
# ==========================================
class PricingState(TypedDict):
    messages: Annotated[list, lambda x, y: x + y] # LangGraph message reducer
    step: int
    base_cost: float
    current_inventory: int
    max_inventory: int
    base_price: float
    competitor_price: float
    rl_state_idx: int
    proposed_action_idx: int
    final_price: float
    reward: float
    policy_constraints: dict

# Initialize Global Components
rl_agent = DynamicPricingRL()
rag_system = MockRAGSystem()

# --- Agent Nodes ---

def market_intelligence_agent(state: PricingState) -> PricingState:
    """Retrieves competitor pricing via RAG"""
    comp_price = rag_system.get_competitor_price("current GPU market price")

    # Calculate state features for RL
    inv_ratio = state["current_inventory"] / state["max_inventory"]
    time_bucket = state["step"] % 3 # Mock time of day
    comp_ratio = comp_price / state["base_price"]

    rl_state_idx = rl_agent.discretize_state(inv_ratio, time_bucket, comp_ratio)

    return {
        "messages": [AIMessage(content=f"Market Intel: Competitor price is ${comp_price}.")],
        "competitor_price": comp_price,
        "rl_state_idx": rl_state_idx
    }

def rl_pricing_agent(state: PricingState) -> PricingState:
    """Runs the RL algorithm to decide price adjustment"""
    state_idx = state["rl_state_idx"]
    action_idx = rl_agent.choose_action(state_idx)
    action_multiplier = rl_agent.actions[action_idx]

    proposed_price = state["base_price"] * (1 + action_multiplier)

    return {
        "messages": [AIMessage(content=f"RL Agent: Proposing {action_multiplier*100}% adjustment. New price: ${proposed_price:.2f}")],
        "proposed_action_idx": action_idx,
        "final_price": proposed_price
    }

def compliance_agent(state: PricingState) -> PricingState:
    """Checks RL decision against company policy via RAG"""
    constraints = rag_system.get_pricing_policy("pricing rules")
    final_price = state["final_price"]

    # Enforce absolute floor
    if final_price < constraints["absolute_price_floor"]:
        final_price = constraints["absolute_price_floor"]

    # Enforce minimum margin
    min_allowed_price = state["base_cost"] * (1 + constraints["min_margin_pct"])
    if final_price < min_allowed_price:
        final_price = min_allowed_price

    return {
        "messages": [AIMessage(content=f"Compliance: Adjusted price to ${final_price:.2f} to meet margin/floor rules.")],
        "final_price": final_price,
        "policy_constraints": constraints
    }

def environment_and_reward_node(state: PricingState) -> PricingState:
    """Simulates market response, calculates reward, and updates RL Q-Table"""
    price = state["final_price"]
    comp_price = state["competitor_price"]

    # Simulate Demand (Elasticity: if we are more expensive than competitor, demand drops)
    price_ratio = price / comp_price
    base_demand = 100
    if price_ratio > 1.05:
        demand = int(base_demand * 0.5)
    elif price_ratio < 0.95:
        demand = int(base_demand * 1.5)
    else:
        demand = base_demand

    # Simulate Sales (bounded by inventory)
    sales = min(demand, state["current_inventory"])
    revenue = sales * price
    cost = sales * state["base_cost"]
    profit = revenue - cost

    # Calculate Reward
    stockout_penalty = -500 if state["current_inventory"] - sales <= 0 else 0
    reward = profit + stockout_penalty

    # Update Inventory
    new_inventory = state["current_inventory"] - sales

    # Update RL Agent
    next_inv_ratio = new_inventory / state["max_inventory"]
    next_time_bucket = (state["step"] + 1) % 3
    next_comp_ratio = comp_price / state["base_price"]
    next_state_idx = rl_agent.discretize_state(next_inv_ratio, next_time_bucket, next_comp_ratio)

    rl_agent.update(state["rl_state_idx"], state["proposed_action_idx"], reward, next_state_idx)
    rl_agent.save_to_replay_buffer({
        "state": state["rl_state_idx"],
        "action": state["proposed_action_idx"],
        "reward": reward,
        "next_state": next_state_idx
    })

    return {
        "messages": [AIMessage(content=f"Environment: Sold {sales} units. Profit: ${profit:.2f}. Reward: {reward:.2f}")],
        "step": state["step"] + 1,
        "current_inventory": new_inventory,
        "reward": reward
    }

# ==========================================
# 4. Build the LangGraph
# ==========================================
def build_pricing_graph():
    workflow = StateGraph(PricingState)

    # Add Nodes
    workflow.add_node("market_intel", market_intelligence_agent)
    workflow.add_node("rl_pricing", rl_pricing_agent)
    workflow.add_node("compliance", compliance_agent)
    workflow.add_node("environment", environment_and_reward_node)

    # Define Edges (Sequential Flow for this episode)
    workflow.add_edge(START, "market_intel")
    workflow.add_edge("market_intel", "rl_pricing")
    workflow.add_edge("rl_pricing", "compliance")
    workflow.add_edge("compliance", "environment")
    workflow.add_edge("environment", END)

    # Compile with Memory (Checkpointer)
    memory = MemorySaver()
    return workflow.compile(checkpointer=memory)

# ==========================================
# 5. Execution & Simulation Loop
# ==========================================
if __name__ == "__main__":
    app = build_pricing_graph()

    # Initial State
    initial_config = {
        "configurable": {"thread_id": "pricing_thread_1"}
    }

    current_state = {
        "messages": [HumanMessage(content="Initiate dynamic pricing cycle.")],
        "step": 0,
        "base_cost": 600.0,
        "current_inventory": 50,
        "max_inventory": 100,
        "base_price": 800.0,
        "competitor_price": 0.0,
        "rl_state_idx": 0,
        "proposed_action_idx": 0,
        "final_price": 800.0,
        "reward": 0.0,
        "policy_constraints": {}
    }

    print("--- Starting Enterprise Dynamic Pricing Simulation ---\n")

    # Simulate 15 time steps (e.g., hours)
    for i in range(15):
        print(f"\n{'='*20} TIME STEP {i+1} {'='*20}")

        # Replenish inventory slightly to keep the simulation going
        if current_state["current_inventory"] < 20:
            current_state["current_inventory"] += 30
            print("*(Inventory Restocked)*")

        # Invoke the graph
        result = app.invoke(current_state, initial_config)

        # Print agent messages
        for msg in result["messages"]:
            if isinstance(msg, AIMessage):
                print(f" -> {msg.content}")

        print(f" [Final Price Executed: ${result['final_price']:.2f} | Reward: {result['reward']:.2f}]")

        # Prepare state for next step
        current_state = result
        # Clear messages for the next step to keep context window clean, but keep state
        current_state["messages"] = [HumanMessage(content="Next pricing cycle.")]

    print("\n--- Simulation Complete ---")
    print(f"Final RL Epsilon (Exploration Rate): {rl_agent.epsilon:.3f}")
    print(f"Total Experiences in Replay Buffer: {len(rl_agent.replay_buffer)}")

Part 4: Architectural Breakdown & Enterprise Considerations

1. Why Multi-Agent Instead of a Single LLM?

If you ask a single LLM to "set the price", it will hallucinate numbers based on its pre-training. By separating the workflow:

The Market Agent grounds the LLM in real-time data via RAG.
The RL Agent does the mathematical heavy lifting (optimizing the reward function) without LLM hallucination.
The Compliance Agent acts as a deterministic guardrail, ensuring the RL's mathematical optimum doesn't violate legal or corporate policies.

2. The Role of RAG and Memory

RAG (Retrieval-Augmented Generation): Used by the Market and Compliance agents to fetch unstructured data (competitor news, PDF pricing policies) and convert them into structured constraints for the RL state and guardrails.
LangGraph Memory (MemorySaver): Maintains the state of the current pricing thread. If a human approver needs to step in (Human-in-the-Loop), the graph pauses, saves the state, and resumes exactly where it left off.
RL Replay Buffer: The save_to_replay_buffer method stores state transitions. In an enterprise setting, this buffer is written to a database (like Redis or PostgreSQL) and used to nightly retrain a Deep Q-Network (DQN) or PPO model.

3. Safe Exploration in Production

The biggest fear with RL in production is the agent "exploring" by dropping prices to $0.

Mitigation 1: The Action Space is restricted to small multipliers (e.g., ±10%).
Mitigation 2: The Compliance Agent (via RAG) enforces hard floors. Even if the RL agent outputs a -10% adjustment, if it breaches the 15% margin policy retrieved from the vector store, the Compliance agent overrides it.
Mitigation 3: Epsilon decay. As seen in the code, epsilon decays over time. The agent explores heavily in the beginning, then shifts to exploitation as it learns the market dynamics.

Summary

Fusing Reinforcement Learning with Multi-Agent LangGraph architectures represents the cutting edge of enterprise AI. By allowing LLMs to handle context retrieval (RAG), reasoning, and compliance, while delegating the mathematical optimization to an RL engine, businesses can achieve dynamic pricing systems that are not only highly profitable but also safe, explainable, and aligned with corporate governance.