bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement structured reasoning_data generation for i18n support

Browse Source 
Implemented proper reasoning data generation for purchase orders and
production batches to enable multilingual dashboard support.

Backend Strategy:
- Generate structured JSON with type codes and parameters
- Store only reasoning_data (JSONB), not hardcoded text
- Frontend will translate using i18n libraries

Changes:
1. Created shared/schemas/reasoning_types.py
   - Defined reasoning types for POs and batches
   - Created helper functions for common reasoning patterns
   - Supports multiple reasoning types (low_stock, forecast_demand, etc.)

2. Production Service (services/production/app/services/production_service.py)
   - Generate reasoning_data when creating batches from forecast
   - Include parameters: product_name, predicted_demand, current_stock, etc.
   - Structure supports frontend i18n interpolation

3. Procurement Service (services/procurement/app/services/procurement_service.py)
   - Implemented actual PO creation (was placeholder before!)
   - Groups requirements by supplier
   - Generates reasoning_data based on context (low_stock vs forecast)
   - Creates PO items automatically

Example reasoning_data:
{
  "type": "low_stock_detection",
  "parameters": {
    "supplier_name": "Harinas del Norte",
    "product_names": ["Flour Type 55", "Flour Type 45"],
    "days_until_stockout": 3,
    "current_stock": 45.5,
    "required_stock": 200
  },
  "consequence": {
    "type": "stockout_risk",
    "severity": "high",
    "impact_days": 3
  }
}

Frontend will translate:
- EN: "Low stock detected for Harinas del Norte. Stock runs out in 3 days."
- ES: "Stock bajo detectado para Harinas del Norte. Se agota en 3 días."
- CA: "Estoc baix detectat per Harinas del Norte. S'esgota en 3 dies."

Next steps:
- Remove TEXT fields (reasoning, consequence) from models
- Update dashboard service to use reasoning_data
- Create frontend i18n translation keys
- Update dashboard components to translate dynamically
This commit is contained in:
Claude
2025-11-07 18:16:44 +00:00
parent 6ee8c055ee
commit ddc4928d78
3 changed files with 446 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
services/procurement/app/services/procurement_service.py
View File

@@ -872,8 +872,172 @@ class ProcurementService:
return f"PLAN-{timestamp}"
async def _create_purchase_orders_from_plan(self, tenant_id, plan_id, auto_approve):
"""Create POs from plan (placeholder)"""
return {'success': True, 'created_pos': []}
"""
Create purchase orders from procurement plan requirements
Groups requirements by supplier and creates POs with reasoning data
"""
try:
from shared.schemas.reasoning_types import create_po_reasoning_low_stock, create_po_reasoning_forecast_demand
from collections import defaultdict
# Get plan requirements
requirements = await self.requirement_repo.get_requirements_by_plan(plan_id, tenant_id)
if not requirements:
logger.warning(f"No requirements found for plan {plan_id}")
return {'success': False, 'created_pos': [], 'error': 'No requirements found'}
# Group requirements by supplier
supplier_requirements = defaultdict(list)
for req in requirements:
supplier_id = req.preferred_supplier_id
if supplier_id:
supplier_requirements[supplier_id].append(req)
created_pos = []
# Create a PO for each supplier
for supplier_id, reqs in supplier_requirements.items():
try:
# Get supplier info
supplier = await self._get_supplier_by_id(tenant_id, supplier_id)
supplier_name = supplier.get('name', 'Unknown Supplier') if supplier else 'Unknown Supplier'
# Calculate PO totals
subtotal = sum(
(req.estimated_unit_cost or 0) * (req.net_requirement or 0)
for req in reqs
)
# Determine earliest required delivery date
required_delivery_date = min(req.required_by_date for req in reqs)
# Calculate urgency based on delivery date
days_until_delivery = (required_delivery_date - date.today()).days
# Collect product names for reasoning
product_names = [req.product_name for req in reqs[:3]] # First 3 products
if len(reqs) > 3:
product_names.append(f"and {len(reqs) - 3} more")
# Determine reasoning type based on requirements
# If any requirement has low stock, use low_stock_detection
has_low_stock = any(
(req.current_stock_level or 0) < (req.total_quantity_needed or 0) * 0.3
for req in reqs
)
if has_low_stock:
# Calculate aggregate stock data for reasoning
total_current = sum(req.current_stock_level or 0 for req in reqs)
total_required = sum(req.total_quantity_needed or 0 for req in reqs)
reasoning_data = create_po_reasoning_low_stock(
supplier_name=supplier_name,
product_names=product_names,
current_stock=float(total_current),
required_stock=float(total_required),
days_until_stockout=days_until_delivery,
threshold_percentage=20,
affected_products=[req.product_name for req in reqs]
)
else:
# Use forecast-based reasoning
reasoning_data = create_po_reasoning_forecast_demand(
supplier_name=supplier_name,
product_names=product_names,
forecast_period_days=7,
total_demand=float(sum(req.order_demand or 0 for req in reqs))
)
# Generate PO number
po_number = await self._generate_po_number()
# Determine if needs approval (based on amount or config)
requires_approval = subtotal > 500 or not auto_approve # Example threshold
# Create PO
po_data = {
'tenant_id': tenant_id,
'supplier_id': supplier_id,
'procurement_plan_id': plan_id,
'po_number': po_number,
'status': 'approved' if auto_approve and not requires_approval else 'pending_approval',
'priority': 'high' if days_until_delivery <= 2 else 'normal',
'order_date': datetime.now(timezone.utc),
'required_delivery_date': datetime.combine(required_delivery_date, datetime.min.time()),
'subtotal': Decimal(str(subtotal)),
'total_amount': Decimal(str(subtotal)), # Simplified, no tax/shipping
'currency': 'EUR',
'requires_approval': requires_approval,
'reasoning_data': reasoning_data, # NEW: Structured reasoning
'created_at': datetime.now(timezone.utc),
'updated_at': datetime.now(timezone.utc),
'created_by': uuid.uuid4(), # System user
'updated_by': uuid.uuid4()
}
po = await self.po_repo.create_purchase_order(po_data)
# Create PO items
po_items = []
for req in reqs:
item_data = {
'tenant_id': tenant_id,
'purchase_order_id': po.id,
'procurement_requirement_id': req.id,
'inventory_product_id': req.product_id,
'product_name': req.product_name,
'ordered_quantity': req.net_requirement,
'unit_of_measure': req.unit_of_measure,
'unit_price': req.estimated_unit_cost or Decimal('0'),
'line_total': (req.estimated_unit_cost or Decimal('0')) * (req.net_requirement or Decimal('0')),
'remaining_quantity': req.net_requirement,
'created_at': datetime.now(timezone.utc),
'updated_at': datetime.now(timezone.utc)
}
item = await self.po_item_repo.create_po_item(item_data)
po_items.append(item)
created_pos.append({
'id': str(po.id),
'po_number': po.po_number,
'supplier_id': str(supplier_id),
'supplier_name': supplier_name,
'total_amount': float(po.total_amount),
'status': po.status.value if hasattr(po.status, 'value') else po.status,
'items_count': len(po_items)
})
logger.info(f"Created PO {po.po_number} for supplier {supplier_name} with {len(po_items)} items")
except Exception as e:
logger.error(f"Error creating PO for supplier {supplier_id}: {e}", exc_info=True)
continue
return {
'success': True,
'created_pos': created_pos,
'count': len(created_pos)
}
except Exception as e:
logger.error(f"Error in _create_purchase_orders_from_plan: {e}", exc_info=True)
return {'success': False, 'created_pos': [], 'error': str(e)}
async def _generate_po_number(self):
"""Generate unique PO number"""
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
return f"PO-{timestamp}-{uuid.uuid4().hex[:6].upper()}"
async def _get_supplier_by_id(self, tenant_id, supplier_id):
"""Get supplier details by ID"""
try:
return await self.suppliers_client.get_supplier(str(tenant_id), str(supplier_id))
except Exception as e:
logger.warning(f"Failed to get supplier {supplier_id}: {e}")
return None
async def _publish_plan_generated_event(self, tenant_id, plan_id):
"""Publish plan generated event"""

13
services/production/app/services/production_service.py
View File

@@ -1836,6 +1836,18 @@ class ProductionService:
# Note: In a real scenario, we'd fetch recipe_id from product/inventory
# For now, we assume recipe_id = product_id or fetch from a mapping
# Generate reasoning data for JTBD dashboard
from shared.schemas.reasoning_types import create_batch_reasoning_forecast_demand
reasoning_data = create_batch_reasoning_forecast_demand(
product_name=f"Product {product_id}", # TODO: Get actual product name from inventory
predicted_demand=predicted_demand,
current_stock=current_stock,
production_needed=production_needed,
target_date=target_date.isoformat(),
confidence_score=forecast.get('confidence_score', 0.85)
)
# Create production batch
batch_data = {
'tenant_id': tenant_id,
@@ -1847,6 +1859,7 @@ class ProductionService:
'planned_start_time': datetime.combine(target_date, datetime.min.time()),
'planned_end_time': datetime.combine(target_date, datetime.max.time()),
'planned_quantity': production_needed,
'reasoning_data': reasoning_data, # NEW: Structured reasoning for i18n
'created_at': datetime.now(timezone.utc),
'updated_at': datetime.now(timezone.utc),
}

267
shared/schemas/reasoning_types.py Normal file
View File

@@ -0,0 +1,267 @@
"""
Reasoning Types for JTBD Dashboard
Defines standard reasoning data structures for Purchase Orders and Production Batches.
Backend generates structured data, frontend translates using i18n.
IMPORTANT: All text displayed to users is translated in the frontend.
Backend only stores type codes and parameters.
"""
from enum import Enum
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field
# ============================================================
# Reasoning Types
# ============================================================
class PurchaseOrderReasoningType(str, Enum):
"""Types of reasoning for why a purchase order was created"""
LOW_STOCK_DETECTION = "low_stock_detection"
FORECAST_DEMAND = "forecast_demand"
SAFETY_STOCK_REPLENISHMENT = "safety_stock_replenishment"
SUPPLIER_CONTRACT = "supplier_contract"
SEASONAL_DEMAND = "seasonal_demand"
PRODUCTION_REQUIREMENT = "production_requirement"
MANUAL_REQUEST = "manual_request"
class ProductionBatchReasoningType(str, Enum):
"""Types of reasoning for why a production batch was created"""
FORECAST_DEMAND = "forecast_demand"
CUSTOMER_ORDER = "customer_order"
STOCK_REPLENISHMENT = "stock_replenishment"
SEASONAL_PREPARATION = "seasonal_preparation"
PROMOTION_EVENT = "promotion_event"
URGENT_ORDER = "urgent_order"
REGULAR_SCHEDULE = "regular_schedule"
class ConsequenceSeverity(str, Enum):
"""Severity level of consequences if action is not taken"""
CRITICAL = "critical" # Immediate business impact
HIGH = "high" # Significant impact within 48h
MEDIUM = "medium" # Moderate impact within week
LOW = "low" # Minor impact
# ============================================================
# Reasoning Data Models
# ============================================================
class PurchaseOrderReasoningData(BaseModel):
"""
Structured reasoning data for purchase orders
Example:
{
"type": "low_stock_detection",
"parameters": {
"supplier_name": "Harinas del Norte",
"product_names": ["Flour Type 55", "Flour Type 45"],
"current_stock_kg": 45.5,
"required_stock_kg": 200,
"days_until_stockout": 3,
"threshold_percentage": 20
},
"consequence": {
"type": "stockout_risk",
"severity": "high",
"impact_days": 3,
"affected_products": ["Baguette", "Croissant"],
"estimated_lost_orders": 15
},
"metadata": {
"trigger_source": "orchestrator_auto",
"forecast_confidence": 0.85,
"ai_assisted": true
}
}
"""
type: PurchaseOrderReasoningType = Field(..., description="Type of reasoning")
parameters: Dict[str, Any] = Field(..., description="Parameters for i18n interpolation")
consequence: Optional[Dict[str, Any]] = Field(None, description="What happens if not approved")
metadata: Optional[Dict[str, Any]] = Field(None, description="Additional context")
class ProductionBatchReasoningData(BaseModel):
"""
Structured reasoning data for production batches
Example:
{
"type": "forecast_demand",
"parameters": {
"product_name": "Croissant",
"predicted_demand": 500,
"current_stock": 120,
"production_needed": 380,
"target_date": "2025-11-08",
"confidence_score": 0.87
},
"urgency": {
"level": "normal",
"ready_by_time": "08:00",
"customer_commitment": false
},
"metadata": {
"trigger_source": "orchestrator_auto",
"forecast_id": "uuid-here",
"ai_assisted": true
}
}
"""
type: ProductionBatchReasoningType = Field(..., description="Type of reasoning")
parameters: Dict[str, Any] = Field(..., description="Parameters for i18n interpolation")
urgency: Optional[Dict[str, Any]] = Field(None, description="Urgency information")
metadata: Optional[Dict[str, Any]] = Field(None, description="Additional context")
# ============================================================
# Helper Functions
# ============================================================
def create_po_reasoning_low_stock(
supplier_name: str,
product_names: list,
current_stock: float,
required_stock: float,
days_until_stockout: int,
threshold_percentage: int = 20,
affected_products: Optional[list] = None,
estimated_lost_orders: Optional[int] = None
) -> Dict[str, Any]:
"""
Create reasoning data for low stock detection
Args:
supplier_name: Name of the supplier
product_names: List of product names in the order
current_stock: Current stock level
required_stock: Required stock level
days_until_stockout: Days until stock runs out
threshold_percentage: Stock threshold percentage
affected_products: Products that will be affected
estimated_lost_orders: Estimated number of lost orders
Returns:
Reasoning data dictionary
"""
return {
"type": PurchaseOrderReasoningType.LOW_STOCK_DETECTION.value,
"parameters": {
"supplier_name": supplier_name,
"product_names": product_names,
"product_count": len(product_names),
"current_stock": current_stock,
"required_stock": required_stock,
"days_until_stockout": days_until_stockout,
"threshold_percentage": threshold_percentage,
"stock_percentage": round((current_stock / required_stock * 100), 1) if required_stock > 0 else 0
},
"consequence": {
"type": "stockout_risk",
"severity": ConsequenceSeverity.HIGH.value if days_until_stockout <= 2 else ConsequenceSeverity.MEDIUM.value,
"impact_days": days_until_stockout,
"affected_products": affected_products or [],
"estimated_lost_orders": estimated_lost_orders or 0
},
"metadata": {
"trigger_source": "orchestrator_auto",
"ai_assisted": True
}
}
def create_po_reasoning_forecast_demand(
supplier_name: str,
product_names: list,
forecast_period_days: int,
total_demand: float,
forecast_confidence: float = 0.85
) -> Dict[str, Any]:
"""Create reasoning data for forecast-based demand"""
return {
"type": PurchaseOrderReasoningType.FORECAST_DEMAND.value,
"parameters": {
"supplier_name": supplier_name,
"product_names": product_names,
"product_count": len(product_names),
"forecast_period_days": forecast_period_days,
"total_demand": total_demand,
"forecast_confidence": round(forecast_confidence * 100, 1)
},
"consequence": {
"type": "insufficient_supply",
"severity": ConsequenceSeverity.MEDIUM.value,
"impact_days": forecast_period_days
},
"metadata": {
"trigger_source": "orchestrator_auto",
"forecast_confidence": forecast_confidence,
"ai_assisted": True
}
}
def create_batch_reasoning_forecast_demand(
product_name: str,
predicted_demand: float,
current_stock: float,
production_needed: float,
target_date: str,
confidence_score: float = 0.85
) -> Dict[str, Any]:
"""Create reasoning data for forecast-based production"""
return {
"type": ProductionBatchReasoningType.FORECAST_DEMAND.value,
"parameters": {
"product_name": product_name,
"predicted_demand": round(predicted_demand, 1),
"current_stock": round(current_stock, 1),
"production_needed": round(production_needed, 1),
"target_date": target_date,
"confidence_score": round(confidence_score * 100, 1)
},
"urgency": {
"level": "normal",
"ready_by_time": "08:00",
"customer_commitment": False
},
"metadata": {
"trigger_source": "orchestrator_auto",
"confidence_score": confidence_score,
"ai_assisted": True
}
}
def create_batch_reasoning_customer_order(
product_name: str,
customer_name: str,
order_quantity: float,
delivery_date: str,
order_number: str
) -> Dict[str, Any]:
"""Create reasoning data for customer order fulfillment"""
return {
"type": ProductionBatchReasoningType.CUSTOMER_ORDER.value,
"parameters": {
"product_name": product_name,
"customer_name": customer_name,
"order_quantity": round(order_quantity, 1),
"delivery_date": delivery_date,
"order_number": order_number
},
"urgency": {
"level": "high",
"ready_by_time": "07:00",
"customer_commitment": True
},
"metadata": {
"trigger_source": "customer_order",
"ai_assisted": False
}
}