Beliefs → Desires → Intentions → Actions 完整循环
"""
BDI Inventory Agent - 书中 2.3.6 节代码的修复版
原书代码存在3个bug:
1. update_beliefs 方法缺失
2. execute_intentions 先清空 active_intentions 再遍历，导致无执行
3. run_cycle 没有 return 语句
修复后可运行。
"""
import numpy as np
from typing import Dict, List, Set, Tuple, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import logging
# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger("InventoryBDIAgent")
@dataclass
class ProductInfo:
product_id: str
name: str
category: str
price: float
cost: float
lead_time_days: int
shelf_life_days: Optional[int] = None
supplier_id: str = ""
alternative_suppliers: List[str] = field(default_factory=list)
min_order_quantity: int = 1
@dataclass
class InventoryItem:
product_id: str
current_stock: int
reorder_point: int
optimal_stock: int
last_reorder_date: Optional[datetime] = None
expected_delivery_date: Optional[datetime] = None
pending_order_quantity: int = 0
@dataclass
class SalesData:
product_id: str
daily_sales: List[int]
def average_daily_sales(self) -> float:
if not self.daily_sales:
return 0
return sum(self.daily_sales) / len(self.daily_sales)
def trend(self) -> float:
if len(self.daily_sales) < 7:
return 0
recent_week = self.daily_sales[-7:]
previous_week = self.daily_sales[-14:-7]
if not previous_week or sum(previous_week) == 0:
return 0
return (sum(recent_week) - sum(previous_week)) / sum(previous_week)
class InventoryBDIAgent:
"""
A Belief-Desire-Intention agent for inventory management.
Beliefs: Current inventory levels, Sales data and forecasts, Supplier information, Store capacity
Desires (Goals): Minimize stockouts, Minimize excess inventory, Maximize profit margin, Ensure fresh products
Intentions (Plans): Reorder products, Discount soon-to-expire items, Promote high-margin products
"""
def __init__(self):
self.inventory: Dict[str, InventoryItem] = {}
self.products: Dict[str, ProductInfo] = {}
self.sales_data: Dict[str, SalesData] = {}
self.store_capacity = 1000
self.current_date = datetime.now()
self.goals = {
"minimize_stockouts": 0.4,
"minimize_excess_inventory": 0.3,
"maximize_profit_margin": 0.2,
"ensure_fresh_products": 0.1,
}
self.active_intentions: List[Dict] = []
logger.info("Inventory BDI Agent initialized")
# ===== 修复1: 新增 update_beliefs 方法 =====
def update_beliefs(self, new_inventory=None, new_date=None):
if new_inventory is not None:
self.inventory = new_inventory
if new_date is not None:
self.current_date = new_date
logger.info(f"Beliefs updated. Date: {self.current_date.date()}, Products: {len(self.inventory)}")
def deliberate(self) -> List[str]:
goal_utilities = {}
stockout_utility = self._evaluate_stockout_prevention()
goal_utilities["minimize_stockouts"] = stockout_utility * self.goals["minimize_stockouts"]
excess_utility = self._evaluate_excess_reduction()
goal_utilities["minimize_excess_inventory"] = excess_utility * self.goals["minimize_excess_inventory"]
profit_utility = self._evaluate_profit_maximization()
goal_utilities["maximize_profit_margin"] = profit_utility * self.goals["maximize_profit_margin"]
freshness_utility = self._evaluate_freshness()
goal_utilities["ensure_fresh_products"] = freshness_utility * self.goals["ensure_fresh_products"]
sorted_goals = sorted(goal_utilities.items(), key=lambda x: x[1], reverse=True)
logger.info(f"Goal deliberation: {[(g, round(u,4)) for g,u in sorted_goals]}")
return [goal[0] for goal in sorted_goals]
def _evaluate_stockout_prevention(self) -> float:
if not self.inventory:
return 0.0
at_risk_count = 0
for product_id, item in self.inventory.items():
if product_id not in self.sales_data:
continue
sales_data = self.sales_data[product_id]
avg_daily_sales = sales_data.average_daily_sales()
# 注意: 原书写的是 item.lead_time_days，但 InventoryItem 没有这个字段
# 应该是 product.lead_time_days，这里取 product 的
product = self.products.get(product_id)
lead_time = product.lead_time_days if product else 3
if avg_daily_sales > 0 and item.current_stock / avg_daily_sales < lead_time:
at_risk_count += 1
return at_risk_count / len(self.inventory) if self.inventory else 0.0
def _evaluate_excess_reduction(self) -> float:
if not self.inventory:
return 0.0
total_excess = 0
for product_id, item in self.inventory.items():
if item.current_stock > item.optimal_stock:
total_excess += (item.current_stock - item.optimal_stock)
total_inventory = sum(i.current_stock for i in self.inventory.values())
capacity_ratio = min(1.0, total_inventory / self.store_capacity) if self.store_capacity > 0 else 0
excess_ratio = total_excess / total_inventory if total_inventory > 0 else 0
return capacity_ratio * excess_ratio
def _evaluate_profit_maximization(self) -> float:
if not self.products:
return 0.0
high_margin_opportunities = 0
for product_id, product in self.products.items():
if product_id not in self.inventory:
continue
margin = (product.price - product.cost) / product.price
item = self.inventory[product_id]
if margin > 0.4 and item.current_stock < item.optimal_stock:
high_margin_opportunities += 1
return high_margin_opportunities / len(self.products)
def _evaluate_freshness(self) -> float:
perishable_products = [p for p in self.products.values() if p.shelf_life_days is not None]
if not perishable_products:
return 0.0
at_risk_count = 0
for product in perishable_products:
if product.product_id not in self.inventory:
continue
item = self.inventory[product.product_id]
sales_data = self.sales_data.get(product.product_id)
if sales_data:
avg_daily_sales = sales_data.average_daily_sales()
if avg_daily_sales > 0:
days_to_sell = item.current_stock / avg_daily_sales
if product.shelf_life_days and days_to_sell > product.shelf_life_days * 0.7:
at_risk_count += 1
return at_risk_count / len(perishable_products)
def generate_intentions(self, prioritized_goals: List[str]) -> None:
self.active_intentions.clear()
for goal in prioritized_goals:
if goal == "minimize_stockouts":
self._plan_reorders()
elif goal == "minimize_excess_inventory":
self._plan_inventory_reduction()
elif goal == "maximize_profit_margin":
self._plan_margin_optimization()
elif goal == "ensure_fresh_products":
self._plan_freshness_management()
logger.info(f"Generated {len(self.active_intentions)} intentions")
def _plan_reorders(self) -> None:
for product_id, item in self.inventory.items():
if product_id not in self.products or product_id not in self.sales_data:
continue
if item.pending_order_quantity > 0:
continue
product = self.products[product_id]
sales_data = self.sales_data[product_id]
avg_daily_sales = sales_data.average_daily_sales()
trend_factor = 1.0 + sales_data.trend()
projected_daily_sales = avg_daily_sales * trend_factor
if projected_daily_sales <= 0:
continue
days_of_supply = item.current_stock / projected_daily_sales
if days_of_supply <= product.lead_time_days + 3:
order_quantity = max(item.optimal_stock - item.current_stock, product.min_order_quantity)
self.active_intentions.append({
"action": "reorder",
"product_id": product_id,
"quantity": order_quantity,
"supplier_id": product.supplier_id,
"priority": 1.0 - (days_of_supply / product.lead_time_days)
if days_of_supply < product.lead_time_days else 0.5
})
logger.info(f"Created intention to reorder {order_quantity} units of {product_id}")
def _plan_inventory_reduction(self) -> None:
for product_id, item in self.inventory.items():
if product_id not in self.products:
continue
if item.current_stock > item.optimal_stock * 1.5:
excess_quantity = item.current_stock - item.optimal_stock
self.active_intentions.append({
"action": "discount",
"product_id": product_id,
"discount_percentage": min(30, 5 * (item.current_stock / item.optimal_stock)),
"priority": 0.3 * (excess_quantity / item.optimal_stock)
})
logger.info(f"Created intention to discount {product_id} to reduce overstock")
def _plan_margin_optimization(self) -> None:
for product_id, product in self.products.items():
if product_id not in self.inventory:
continue
item = self.inventory[product_id]
margin = (product.price - product.cost) / product.price
if margin > 0.4 and item.current_stock < item.optimal_stock * 0.8:
self.active_intentions.append({
"action": "promote",
"product_id": product_id,
"promotion_type": "featured",
"priority": 0.2 * margin
})
logger.info(f"Created intention to promote high-margin product {product_id}")
def _plan_freshness_management(self) -> None:
for product_id, product in self.products.items():
if product.shelf_life_days is None or product_id not in self.inventory:
continue
item = self.inventory[product_id]
sales_data = self.sales_data.get(product_id)
if not sales_data:
continue
avg_daily_sales = sales_data.average_daily_sales()
if avg_daily_sales <= 0:
continue
days_to_sell = item.current_stock / avg_daily_sales
if days_to_sell > product.shelf_life_days * 0.7:
at_risk_quantity = int(item.current_stock - (avg_daily_sales * product.shelf_life_days * 0.7))
if at_risk_quantity > 0:
self.active_intentions.append({
"action": "discount_perishable",
"product_id": product_id,
"quantity": at_risk_quantity,
"discount_percentage": 40,
"priority": 0.5 * (days_to_sell / product.shelf_life_days)
})
logger.info(f"Created intention to discount perishable {product_id}")
# ===== 修复2: execute_intentions 返回已执行动作，不清空 =====
def execute_intentions(self, prioritized_goals: List[str]) -> List[Dict]:
executed = []
for goal in prioritized_goals:
if goal == "minimize_stockouts":
executed.extend(self._execute_reorders())
elif goal == "minimize_excess_inventory":
executed.extend(self._execute_inventory_reduction())
elif goal == "maximize_profit_margin":
executed.extend(self._execute_margin_optimization())
elif goal == "ensure_fresh_products":
executed.extend(self._execute_freshness_management())
logger.info(f"Executed {len(executed)} intentions")
return executed
def _execute_reorders(self) -> List[Dict]:
executed = []
for intention in self.active_intentions:
if intention["action"] == "reorder":
if self._execute_reorder(intention):
executed.append(intention)
return executed
def _execute_inventory_reduction(self) -> List[Dict]:
executed = []
for intention in self.active_intentions:
if intention["action"] == "discount":
if self._execute_discount(intention):
executed.append(intention)
return executed
def _execute_margin_optimization(self) -> List[Dict]:
executed = []
for intention in self.active_intentions:
if intention["action"] == "promote":
if self._execute_promotion(intention):
executed.append(intention)
return executed
def _execute_freshness_management(self) -> List[Dict]:
executed = []
for intention in self.active_intentions:
if intention["action"] == "discount_perishable":
if self._execute_perishable_discount(intention):
executed.append(intention)
return executed
def _execute_reorder(self, intention) -> bool:
product_id = intention["product_id"]
quantity = intention["quantity"]
supplier_id = intention["supplier_id"]
if product_id not in self.inventory or product_id not in self.products:
logger.warning(f"Cannot reorder {product_id}: not found")
return False
product = self.products[product_id]
item = self.inventory[product_id]
item.pending_order_quantity = quantity
item.last_reorder_date = self.current_date
item.expected_delivery_date = self.current_date + timedelta(days=product.lead_time_days)
logger.info(f"Executed reorder: {quantity} units of {product_id} from supplier {supplier_id}")
logger.info(f" Expected delivery date: {item.expected_delivery_date.date()}")
return True
def _execute_discount(self, intention) -> bool:
product_id = intention["product_id"]
discount_percentage = intention["discount_percentage"]
if product_id not in self.products:
logger.warning(f"Cannot discount {product_id}")
return False
logger.info(f"Executed discount: {discount_percentage}% off {product_id}")
return True
def _execute_promotion(self, intention) -> bool:
product_id = intention["product_id"]
promotion_type = intention["promotion_type"]
if product_id not in self.products:
logger.warning(f"Cannot promote {product_id}")
return False
logger.info(f"Executed promotion: {promotion_type} for {product_id}")
return True
def _execute_perishable_discount(self, intention) -> bool:
product_id = intention["product_id"]
quantity = intention["quantity"]
discount_percentage = intention["discount_percentage"]
if product_id not in self.products:
logger.warning(f"Cannot discount perishable {product_id}")
return False
logger.info(f"Executed perishable discount: {discount_percentage}% off {quantity} units of {product_id}")
return True
# ===== 修复3: run_cycle 返回执行的动作 =====
def run_cycle(self, prioritized_goals: List[str]) -> List[Dict]:
self.update_beliefs()
self.deliberate()
self.generate_intentions(prioritized_goals)
return self.execute_intentions(prioritized_goals)
def demonstrate_bdi_agent():
agent = InventoryBDIAgent()
agent.products = {
"P001": ProductInfo(product_id="P001", name="Organic Apples", category="Produce",
price=2.99, cost=1.50, lead_time_days=2, shelf_life_days=14, supplier_id="S1"),
"P002": ProductInfo(product_id="P002", name="Whole Grain Bread", category="Bakery",
price=3.49, cost=1.25, lead_time_days=1, shelf_life_days=5, supplier_id="S2"),
"P003": ProductInfo(product_id="P003", name="Premium Coffee", category="Beverages",
price=12.99, cost=6.50, lead_time_days=5, supplier_id="S3"),
}
agent.inventory = {
"P001": InventoryItem(product_id="P001", current_stock=25, reorder_point=20, optimal_stock=50),
"P002": InventoryItem(product_id="P002", current_stock=5, reorder_point=10, optimal_stock=30),
"P003": InventoryItem(product_id="P003", current_stock=60, reorder_point=15, optimal_stock=40),
}
agent.sales_data = {
"P001": SalesData(product_id="P001", daily_sales=[
8, 7, 9, 8, 10, 12, 9, 8, 7, 6,
8, 9, 10, 11, 9, 8, 9, 10, 11, 12,
13, 11, 10, 12, 13, 14, 15, 13, 12, 11,
]),
"P002": SalesData(product_id="P002", daily_sales=[
6, 5, 7, 8, 6, 5, 4, 6, 7, 8,
6, 5, 4, 5, 6, 7, 8, 9, 7, 6,
5, 6, 7, 8, 9, 10, 8, 7, 6, 7,
]),
"P003": SalesData(product_id="P003", daily_sales=[
2, 1, 3, 2, 1, 2, 3, 2, 1, 0,
2, 3, 2, 1, 3, 2, 1, 2, 3, 4,
2, 1, 2, 3, 2, 1, 2, 1, 2, 3,
]),
}
print("\n=== BDI Agent Demonstration ===\n")
print("Initial state:")
print(f" Apples (P001): {agent.inventory['P001'].current_stock} units")
print(f" Bread (P002): {agent.inventory['P002'].current_stock} units")
print(f" Coffee (P003): {agent.inventory['P003'].current_stock} units")
executed_actions = agent.run_cycle(["minimize_stockouts", "maximize_profit_margin"])
print("\nAgent reasoning process:")
print(" 1. Updated beliefs (inventory, sales, date).")
print(" 2. Deliberated on goals (stockouts vs. profit, etc.).")
print(" 3. Generated intentions (plans) to achieve top priorities.")
print(" 4. Executed the following actions:")
for i, action in enumerate(executed_actions):
action_type = action["action"]
if action_type == "reorder":
print(f" {i + 1}. Reordered {action['quantity']} units of {action['product_id']}")
elif action_type == "discount":
print(f" {i + 1}. Discounted {action['product_id']} by {action['discount_percentage']}%")
elif action_type == "promote":
print(f" {i + 1}. Promoted {action['product_id']} ({action['promotion_type']})")
elif action_type == "discount_perishable":
print(f" {i + 1}. Marked down {action['quantity']} units of {action['product_id']}")
print("\nSimulating one day passing...")
new_inventory = {
"P001": InventoryItem(
product_id="P001",
current_stock=agent.inventory["P001"].current_stock - 13,
reorder_point=agent.inventory["P001"].reorder_point,
optimal_stock=agent.inventory["P001"].optimal_stock,
pending_order_quantity=agent.inventory["P001"].pending_order_quantity,
expected_delivery_date=agent.inventory["P001"].expected_delivery_date,
last_reorder_date=agent.inventory["P001"].last_reorder_date,
),
"P002": InventoryItem(
product_id="P002",
current_stock=agent.inventory["P002"].current_stock - 7,
reorder_point=agent.inventory["P002"].reorder_point,
optimal_stock=agent.inventory["P002"].optimal_stock,
pending_order_quantity=agent.inventory["P002"].pending_order_quantity,
expected_delivery_date=agent.inventory["P002"].expected_delivery_date,
last_reorder_date=agent.inventory["P002"].last_reorder_date,
),
"P003": InventoryItem(
product_id="P003",
current_stock=agent.inventory["P003"].current_stock - 2,
reorder_point=agent.inventory["P003"].reorder_point,
optimal_stock=agent.inventory["P003"].optimal_stock,
pending_order_quantity=agent.inventory["P003"].pending_order_quantity,
expected_delivery_date=agent.inventory["P003"].expected_delivery_date,
last_reorder_date=agent.inventory["P003"].last_reorder_date,
),
}
agent.update_beliefs(new_inventory=new_inventory, new_date=agent.current_date + timedelta(days=1))
executed_actions = agent.run_cycle(["minimize_stockouts", "maximize_profit_margin"])
print("\nUpdated state (after 1 day):")
print(f" Apples (P001): {agent.inventory['P001'].current_stock} units")
print(f" Bread (P002): {agent.inventory['P002'].current_stock} units (check if low!)")
print(f" Coffee (P003): {agent.inventory['P003'].current_stock} units")
print("\nNew actions taken by the agent:")
for i, action in enumerate(executed_actions):
action_type = action["action"]
if action_type == "reorder":
print(f" {i + 1}. Reordered {action['quantity']} units of {action['product_id']}")
elif action_type == "discount":
print(f" {i + 1}. Discounted {action['product_id']} by {action['discount_percentage']}%")
elif action_type == "promote":
print(f" {i + 1}. Promoted {action['product_id']} ({action['promotion_type']})")
elif action_type == "discount_perishable":
print(f" {i + 1}. Marked down {action['quantity']} units of {action['product_id']}")
if __name__ == "__main__":
demonstrate_bdi_agent()