From 7006296226879fd8e71250cba7627e5fd19bbfc8 Mon Sep 17 00:00:00 2001 From: Mai Duc Duy Date: Fri, 17 Apr 2026 23:22:03 +0700 Subject: [PATCH] feat: implement planner agent with multi-strategy plan generation, validation, and repair logic --- agents/planner.py | 504 +++++------ app_api/services.py | 47 +- llm/service.py | 1613 +++++++++++++++++----------------- policies/strategic_prompt.py | 62 +- 4 files changed, 1104 insertions(+), 1122 deletions(-) diff --git a/agents/planner.py b/agents/planner.py index d0b40b6..8c1fb06 100644 --- a/agents/planner.py +++ b/agents/planner.py @@ -3,16 +3,16 @@ from uuid import uuid4 from agents.base import BaseAgent -from core.enums import ActionType, ApprovalStatus, PlanStatus -from core.models import ( - Action, - AgentProposal, - CandidatePlanDraft, - CandidatePlanEvaluation, - DecisionLog, - Event, - Plan, - SystemState, +from core.enums import ActionType, ApprovalStatus, PlanStatus +from core.models import ( + Action, + AgentProposal, + CandidatePlanDraft, + CandidatePlanEvaluation, + DecisionLog, + Event, + Plan, + SystemState, PlanMetadata ) from core.logger import get_logger @@ -22,80 +22,80 @@ build_winning_factors, explain_rejected_actions, ) -from policies.constraints import evaluate_plan_constraints, mode_rationale, evaluate_hard_constraints, evaluate_soft_constraints +from policies.constraints import evaluate_plan_constraints, mode_rationale, evaluate_hard_constraints, evaluate_soft_constraints from policies.guardrails import approval_required from policies.scoring import compute_score from policies.strategic_prompt import ( - retrieve_relevant_cases, - compute_memory_influence, + retrieve_relevant_cases, + compute_memory_influence, derive_strategy_rationale, - build_strategic_prompt, + build_memory_prompt, ) from simulation.evaluator import evaluate_candidate_plan logger = get_logger(__name__) - + STRATEGY_ORDER = ("cost_first", "balanced", "resilience_first") - - -def _dedupe_actions(actions: list[Action], limit: int) -> list[Action]: - selected: list[Action] = [] - seen: set[tuple[str, str]] = set() - for action in actions: - key = (action.action_type.value, action.target_id) - if key in seen: - continue - selected.append(action) - seen.add(key) - if len(selected) >= limit: - break - return selected - - -def _fallback_sort_key(strategy_label: str, action: Action) -> tuple[float, float, float, float]: - if strategy_label == "cost_first": - return ( - action.estimated_cost_delta, - action.estimated_recovery_hours, - -action.estimated_service_delta, - action.estimated_risk_delta, - ) - if strategy_label == "resilience_first": - return ( - action.estimated_risk_delta, - action.estimated_recovery_hours, - -action.estimated_service_delta, - action.estimated_cost_delta, - ) - return ( - -action.priority, - action.estimated_risk_delta, - -action.estimated_service_delta, - action.estimated_cost_delta, - ) - - -def _strategy_reason(strategy_label: str) -> str: - if strategy_label == "cost_first": - return "fallback cost-first strategy favors lower incremental cost and operational simplicity" - if strategy_label == "resilience_first": - return "fallback resilience-first strategy favors risk reduction and faster recovery" - return "fallback balanced strategy blends specialist priority, risk reduction, and service protection" - - + + +def _dedupe_actions(actions: list[Action], limit: int) -> list[Action]: + selected: list[Action] = [] + seen: set[tuple[str, str]] = set() + for action in actions: + key = (action.action_type.value, action.target_id) + if key in seen: + continue + selected.append(action) + seen.add(key) + if len(selected) >= limit: + break + return selected + + +def _fallback_sort_key(strategy_label: str, action: Action) -> tuple[float, float, float, float]: + if strategy_label == "cost_first": + return ( + action.estimated_cost_delta, + action.estimated_recovery_hours, + -action.estimated_service_delta, + action.estimated_risk_delta, + ) + if strategy_label == "resilience_first": + return ( + action.estimated_risk_delta, + action.estimated_recovery_hours, + -action.estimated_service_delta, + action.estimated_cost_delta, + ) + return ( + -action.priority, + action.estimated_risk_delta, + -action.estimated_service_delta, + action.estimated_cost_delta, + ) + + +def _strategy_reason(strategy_label: str) -> str: + if strategy_label == "cost_first": + return "fallback cost-first strategy favors lower incremental cost and operational simplicity" + if strategy_label == "resilience_first": + return "fallback resilience-first strategy favors risk reduction and faster recovery" + return "fallback balanced strategy blends specialist priority, risk reduction, and service protection" + + def _ensure_actions(candidate_actions: list[Action]) -> list[Action]: ensured = list(candidate_actions) if any(action.action_type == ActionType.NO_OP for action in ensured): return ensured - ensured.append( - Action( - action_id="act_no_op", - action_type=ActionType.NO_OP, - target_id="system", - reason="no action required", - priority=0.1, - ) + ensured.append( + Action( + action_id="act_no_op", + action_type=ActionType.NO_OP, + target_id="system", + reason="no action required", + priority=0.1, + ) ) return ensured @@ -136,57 +136,57 @@ def _should_suppress_no_op(state: SystemState) -> bool: or pressure["critical_skus"] >= 8 or pressure["below_safety_stock"] >= 2 ) - - -def _fallback_drafts(candidate_actions: list[Action], action_limit: int) -> list[CandidatePlanDraft]: - drafts: list[CandidatePlanDraft] = [] - for strategy_label in STRATEGY_ORDER: - sorted_actions = sorted(candidate_actions, key=lambda action: _fallback_sort_key(strategy_label, action)) - selected = _dedupe_actions(sorted_actions, action_limit) - drafts.append( - CandidatePlanDraft( - strategy_label=strategy_label, - action_ids=[action.action_id for action in selected], - rationale=_strategy_reason(strategy_label), - llm_used=False, - ) - ) - return drafts - - + + +def _fallback_drafts(candidate_actions: list[Action], action_limit: int) -> list[CandidatePlanDraft]: + drafts: list[CandidatePlanDraft] = [] + for strategy_label in STRATEGY_ORDER: + sorted_actions = sorted(candidate_actions, key=lambda action: _fallback_sort_key(strategy_label, action)) + selected = _dedupe_actions(sorted_actions, action_limit) + drafts.append( + CandidatePlanDraft( + strategy_label=strategy_label, + action_ids=[action.action_id for action in selected], + rationale=_strategy_reason(strategy_label), + llm_used=False, + ) + ) + return drafts + + def _normalize_drafts( drafts: list[CandidatePlanDraft], candidate_actions: list[Action], action_limit: int, ) -> tuple[list[CandidatePlanDraft], int]: - allowed_ids = {action.action_id for action in candidate_actions} - by_id = {action.action_id: action for action in candidate_actions} - normalized: list[CandidatePlanDraft] = [] - repaired_count = 0 - fallback_map = {draft.strategy_label: draft for draft in _fallback_drafts(candidate_actions, action_limit)} - draft_map = {draft.strategy_label: draft for draft in drafts if draft.strategy_label in STRATEGY_ORDER} - - for strategy_label in STRATEGY_ORDER: - draft = draft_map.get(strategy_label) - if draft is None: - normalized.append(fallback_map[strategy_label]) - repaired_count += 1 - continue - action_ids = [action_id for action_id in draft.action_ids if action_id in allowed_ids] - selected_actions = _dedupe_actions( - [by_id[action_id] for action_id in action_ids if action_id in by_id], - action_limit, - ) - if not selected_actions: - normalized.append(fallback_map[strategy_label]) - repaired_count += 1 - continue - normalized.append( - CandidatePlanDraft( - strategy_label=strategy_label, - action_ids=[action.action_id for action in selected_actions], - rationale=draft.rationale or _strategy_reason(strategy_label), - llm_used=draft.llm_used, + allowed_ids = {action.action_id for action in candidate_actions} + by_id = {action.action_id: action for action in candidate_actions} + normalized: list[CandidatePlanDraft] = [] + repaired_count = 0 + fallback_map = {draft.strategy_label: draft for draft in _fallback_drafts(candidate_actions, action_limit)} + draft_map = {draft.strategy_label: draft for draft in drafts if draft.strategy_label in STRATEGY_ORDER} + + for strategy_label in STRATEGY_ORDER: + draft = draft_map.get(strategy_label) + if draft is None: + normalized.append(fallback_map[strategy_label]) + repaired_count += 1 + continue + action_ids = [action_id for action_id in draft.action_ids if action_id in allowed_ids] + selected_actions = _dedupe_actions( + [by_id[action_id] for action_id in action_ids if action_id in by_id], + action_limit, + ) + if not selected_actions: + normalized.append(fallback_map[strategy_label]) + repaired_count += 1 + continue + normalized.append( + CandidatePlanDraft( + strategy_label=strategy_label, + action_ids=[action.action_id for action in selected_actions], + rationale=draft.rationale or _strategy_reason(strategy_label), + llm_used=draft.llm_used, ) ) return normalized, repaired_count @@ -322,31 +322,31 @@ def _decision_priority_score( def _evaluate_candidate( - *, - state: SystemState, - event: Event | None, - before_kpis, - strategy_label: str, - actions: list[Action], - rationale: str, - llm_used: bool, -) -> CandidatePlanEvaluation: - selected_mode_rationale = mode_rationale(state, event) - violations = evaluate_plan_constraints( - state=state, - event=event, - actions=actions, - ) - feasible = not violations + *, + state: SystemState, + event: Event | None, + before_kpis, + strategy_label: str, + actions: list[Action], + rationale: str, + llm_used: bool, +) -> CandidatePlanEvaluation: + selected_mode_rationale = mode_rationale(state, event) + violations = evaluate_plan_constraints( + state=state, + event=event, + actions=actions, + ) + feasible = not violations if not feasible: return CandidatePlanEvaluation( strategy_label=strategy_label, action_ids=[action.action_id for action in actions], score=0.0, - score_breakdown={ - "service_level": 0.0, - "total_cost": 0.0, - "disruption_risk": 0.0, + score_breakdown={ + "service_level": 0.0, + "total_cost": 0.0, + "disruption_risk": 0.0, "recovery_speed": 0.0, }, projected_kpis=before_kpis.model_copy(deep=True), @@ -376,14 +376,14 @@ def _evaluate_candidate( mode=state.mode, baseline_cost=before_kpis.total_cost, ) - transient_plan = Plan( - plan_id=f"plan_eval_{uuid4().hex[:8]}", - mode=state.mode, - trigger_event_ids=[event.event_id] if event else [], - actions=actions, - score=score, - score_breakdown=breakdown, - strategy_label=strategy_label, + transient_plan = Plan( + plan_id=f"plan_eval_{uuid4().hex[:8]}", + mode=state.mode, + trigger_event_ids=[event.event_id] if event else [], + actions=actions, + score=score, + score_breakdown=breakdown, + strategy_label=strategy_label, generated_by="llm" if llm_used else "deterministic_fallback", planner_reasoning=rationale, status=PlanStatus.PROPOSED, @@ -410,7 +410,7 @@ def _evaluate_candidate( mode_rationale=selected_mode_rationale, approval_required=needs_approval, approval_reason=reason if needs_approval else "no approval required: thresholds not triggered", - rationale=rationale, + rationale=rationale, llm_used=llm_used, ) @@ -739,9 +739,9 @@ def _select_best_evaluation( -item.projected_kpis.total_cost, 1 if item.strategy_label == "balanced" else 0, ), - ) - - + ) + + def _selection_reason( selected: CandidatePlanEvaluation, evaluations: list[CandidatePlanEvaluation], @@ -773,7 +773,7 @@ def _selection_reason( item.projected_kpis.service_level, -item.projected_kpis.total_cost, ), - reverse=True, + reverse=True, ) runner_up = ordered[1] if len(ordered) > 1 else None base = ( @@ -809,24 +809,24 @@ def _selection_reason( f"{base}. It ranked ahead of {runner_up.strategy_label} ({runner_up_priority_score:.4f}) after comparing " f"service protection, cost, disruption risk, and recovery speed.{coverage_note}" ) - - -def _safe_hold_evaluation( - *, - state: SystemState, - event: Event | None, - before_kpis, - safe_action: Action, - infeasible_count: int, -) -> CandidatePlanEvaluation: - score, breakdown = compute_score( - service_level=before_kpis.service_level, - total_cost=before_kpis.total_cost, - disruption_risk=before_kpis.disruption_risk, - recovery_speed=before_kpis.recovery_speed, - mode=state.mode, - baseline_cost=before_kpis.total_cost, - ) + + +def _safe_hold_evaluation( + *, + state: SystemState, + event: Event | None, + before_kpis, + safe_action: Action, + infeasible_count: int, +) -> CandidatePlanEvaluation: + score, breakdown = compute_score( + service_level=before_kpis.service_level, + total_cost=before_kpis.total_cost, + disruption_risk=before_kpis.disruption_risk, + recovery_speed=before_kpis.recovery_speed, + mode=state.mode, + baseline_cost=before_kpis.total_cost, + ) return CandidatePlanEvaluation( strategy_label="safe_hold", action_ids=[safe_action.action_id], @@ -843,17 +843,17 @@ def _safe_hold_evaluation( mode_rationale=mode_rationale(state, event), approval_required=False, approval_reason="no approval required: retaining current state because all candidate plans were infeasible", - rationale=( - f"no candidate plan satisfied hard constraints; retaining the current network state after excluding " - f"{infeasible_count} infeasible candidate plan(s)" - ), - llm_used=False, - ) - - -class PlannerAgent(BaseAgent): - name = "planner" - + rationale=( + f"no candidate plan satisfied hard constraints; retaining the current network state after excluding " + f"{infeasible_count} infeasible candidate plan(s)" + ), + llm_used=False, + ) + + +class PlannerAgent(BaseAgent): + name = "planner" + def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: proposal = AgentProposal(agent=self.name) @@ -868,32 +868,32 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: infeasible_reasons = {} for act in candidate_actions: dummy_plan = Plan(plan_id="tmp", mode=state.mode, score=0, score_breakdown={}, actions=[act]) - is_feas, vios = evaluate_hard_constraints(dummy_plan, state) - if is_feas: - feasible_candidates.append(act) - else: - infeasible_reasons[act.action_id] = vios - - if not feasible_candidates and candidate_actions: - feasible_candidates = [ + is_feas, vios = evaluate_hard_constraints(dummy_plan, state) + if is_feas: + feasible_candidates.append(act) + else: + infeasible_reasons[act.action_id] = vios + + if not feasible_candidates and candidate_actions: + feasible_candidates = [ Action( action_id="act_no_op_fallback", action_type=ActionType.NO_OP, target_id="system", reason="all candidate actions violated hard constraints", priority=0.0 ) ] action_limit = _action_limit_for_mode(state, len(feasible_candidates)) - - effective_event = event or (state.active_events[-1] if state.active_events else None) - historical_cases = retrieve_relevant_cases(effective_event, state.memory, top_k=3) - memory_influence = compute_memory_influence(historical_cases) - strategic_prompt = build_strategic_prompt( - mode=state.mode.value, event=effective_event, - historical_cases=historical_cases, candidate_actions=feasible_candidates - ) - - before_kpis = state.kpis.model_copy(deep=True) + + effective_event = event or (state.active_events[-1] if state.active_events else None) + historical_cases = retrieve_relevant_cases(effective_event, state.memory, top_k=3) + memory_influence = compute_memory_influence(historical_cases) + memory_prompt = build_memory_prompt( + mode=state.mode.value, historical_cases=historical_cases + ) + + before_kpis = state.kpis.model_copy(deep=True) llm_drafts, planner_error = generate_candidate_plan_drafts( - state=state, event=event, candidate_actions=feasible_candidates + state=state, event=event, candidate_actions=feasible_candidates, + memory_prompt=memory_prompt ) logger.info( "planner candidate generation result llm_drafts=%s planner_error=%s feasible_candidates=%s suppress_no_op=%s action_limit=%s", @@ -932,8 +932,8 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: ) else: logger.info("planner used llm candidate drafts without repair") - - by_id = {action.action_id: action for action in feasible_candidates} + + by_id = {action.action_id: action for action in feasible_candidates} gap_by_sku = _inventory_gap_map(state) evaluations: list[CandidatePlanEvaluation] = [] for draft in drafts: @@ -962,12 +962,12 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: safe_action = next(action for action in feasible_candidates if action.action_type == ActionType.NO_OP) evaluations.append( _safe_hold_evaluation( - state=state, event=event, before_kpis=before_kpis, - safe_action=safe_action, infeasible_count=len(evaluations), - ) - ) - - + state=state, event=event, before_kpis=before_kpis, + safe_action=safe_action, infeasible_count=len(evaluations), + ) + ) + + selected_evaluation = _select_best_evaluation( evaluations, state=state, @@ -1026,12 +1026,12 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: simulation_horizon_days=selected_evaluation.simulation_horizon_days, worst_case_kpis=selected_evaluation.worst_case_kpis, ) - - final_plan = Plan( - plan_id=f"plan_{uuid4().hex[:8]}", - mode=state.mode, - trigger_event_ids=[event.event_id] if event else [], - actions=selected_actions, + + final_plan = Plan( + plan_id=f"plan_{uuid4().hex[:8]}", + mode=state.mode, + trigger_event_ids=[event.event_id] if event else [], + actions=selected_actions, score=selected_evaluation.score, score_breakdown=selected_evaluation.score_breakdown, feasible=selected_evaluation.feasible, @@ -1052,22 +1052,22 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: planner_reasoning=summary, status=PlanStatus.PROPOSED, ) - - is_hard_feas, hard_violations = evaluate_hard_constraints(final_plan, state) - soft_violations = evaluate_soft_constraints(final_plan, state) - - final_plan.feasible = is_hard_feas - final_plan.violations = hard_violations + soft_violations - - - strategy_rationale = derive_strategy_rationale(effective_event, historical_cases, selected_actions) - final_plan.metadata = PlanMetadata( - referenced_cases=historical_cases, - memory_influence_score=memory_influence, - strategy_rationale=strategy_rationale, - strategic_prompt=strategic_prompt - ) - + + is_hard_feas, hard_violations = evaluate_hard_constraints(final_plan, state) + soft_violations = evaluate_soft_constraints(final_plan, state) + + final_plan.feasible = is_hard_feas + final_plan.violations = hard_violations + soft_violations + + + strategy_rationale = derive_strategy_rationale(effective_event, historical_cases, selected_actions) + final_plan.metadata = PlanMetadata( + referenced_cases=historical_cases, + memory_influence_score=memory_influence, + strategy_rationale=strategy_rationale, + strategic_prompt=memory_prompt + ) + winning_factors = build_winning_factors( selected_actions, before_kpis, @@ -1075,17 +1075,17 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: selected_evaluation.score_breakdown, ) rejection_reasons = explain_rejected_actions(candidate_actions, selected_actions, action_limit) - - decision_log = DecisionLog( - decision_id=f"dec_{uuid4().hex[:8]}", - plan_id=final_plan.plan_id, - event_ids=final_plan.trigger_event_ids, - before_kpis=before_kpis, - after_kpis=selected_evaluation.projected_kpis, - selected_actions=[action.action_id for action in selected_actions], - rejected_actions=rejection_reasons, - score_breakdown=selected_evaluation.score_breakdown, - mode_rationale=selected_evaluation.mode_rationale, + + decision_log = DecisionLog( + decision_id=f"dec_{uuid4().hex[:8]}", + plan_id=final_plan.plan_id, + event_ids=final_plan.trigger_event_ids, + before_kpis=before_kpis, + after_kpis=selected_evaluation.projected_kpis, + selected_actions=[action.action_id for action in selected_actions], + rejected_actions=rejection_reasons, + score_breakdown=selected_evaluation.score_breakdown, + mode_rationale=selected_evaluation.mode_rationale, rationale=summary, candidate_evaluations=evaluations, selection_reason=_selection_reason( @@ -1095,20 +1095,20 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: by_id=by_id, ), winning_factors=winning_factors, - approval_required=selected_evaluation.approval_required, - approval_reason=selected_evaluation.approval_reason, - approval_status=ApprovalStatus.PENDING if selected_evaluation.approval_required else ApprovalStatus.AUTO_APPLIED, - planner_error=planner_error if fallback_used else None, - metadata=final_plan.metadata, - ) - - enrich_plan_and_decision(state=state, event=event, plan=final_plan, decision_log=decision_log) - - state.latest_plan = final_plan - state.latest_plan_id = final_plan.plan_id - state.pending_plan = final_plan if final_plan.approval_required else None - state.decision_logs.append(decision_log) - + approval_required=selected_evaluation.approval_required, + approval_reason=selected_evaluation.approval_reason, + approval_status=ApprovalStatus.PENDING if selected_evaluation.approval_required else ApprovalStatus.AUTO_APPLIED, + planner_error=planner_error if fallback_used else None, + metadata=final_plan.metadata, + ) + + enrich_plan_and_decision(state=state, event=event, plan=final_plan, decision_log=decision_log) + + state.latest_plan = final_plan + state.latest_plan_id = final_plan.plan_id + state.pending_plan = final_plan if final_plan.approval_required else None + state.decision_logs.append(decision_log) + proposal.observations.append( f"built {final_plan.plan_id} using {final_plan.strategy_label} with score {final_plan.score:.4f}" ) diff --git a/app_api/services.py b/app_api/services.py index b7aa431..2b48996 100644 --- a/app_api/services.py +++ b/app_api/services.py @@ -585,28 +585,29 @@ def dispatch_plan(self, plan_id: str, mode: str) -> dict[str, Any]: if plan is None: raise_not_found("plan", plan_id) + existing_commit_records = [ + ActionExecutionRecord.model_validate(item) + for item in self.store.list_action_execution_records(limit=None) + if item.get("plan_id") == plan_id + and item.get("dispatch_mode") == "commit" + ] - if mode == "commit": - existing_records = [ - ActionExecutionRecord.model_validate(item) - for item in self.store.list_action_execution_records(limit=None) - if item.get("plan_id") == plan_id - and item.get("dispatch_mode", "dry_run") == mode - ] - if existing_records: - existing_records.sort(key=lambda record: record.created_at) - return { - "plan_id": plan_id, - "dispatch_mode": mode, - "plan_execution_status": compute_plan_execution_status( - existing_records - ).value, - "overall_progress": compute_overall_progress(existing_records), - "records": [ - action_execution_record_view(r) for r in existing_records - ], - "compensation_hints": [], - } + if existing_commit_records: + existing_commit_records.sort(key=lambda record: record.created_at) + reported_mode = "dry_run" if mode == "dry_run" else "commit" + + return { + "plan_id": plan_id, + "dispatch_mode": reported_mode, + "plan_execution_status": compute_plan_execution_status( + existing_commit_records + ).value, + "overall_progress": compute_overall_progress(existing_commit_records), + "records": [ + action_execution_record_view(r) for r in existing_commit_records + ], + "compensation_hints": [], # Historic records don't have active hints + } summary = self.dispatch_service.dispatch_plan(plan, mode=mode) @@ -1574,3 +1575,7 @@ def build_event_from_envelope(envelope: EventEnvelope) -> Event: payload=envelope.payload, dedupe_key=envelope.idempotency_key, ) + + + + diff --git a/llm/service.py b/llm/service.py index 1b30c1f..267eece 100644 --- a/llm/service.py +++ b/llm/service.py @@ -1,411 +1,412 @@ -from __future__ import annotations - -import json -import re -from typing import Any - -from core.logger import get_logger -from core.models import ( - Action, - AgentProposal, - CandidatePlanDraft, - CandidatePlanEvaluation, - DecisionLog, - Event, - Plan, - ReflectionNote, - ScenarioRun, - SystemState, -) -from llm.config import load_settings -from llm.gemini_client import GeminiClient, GeminiClientError -from llm.vertex_client import VertexClientError, VertexGeminiClient -from llm.fpt_client import FPTClient, FPTClientError -from orchestrator.prompts import ( - DECISION_EXPLANATION_PROMPT, - AI_CANDIDATE_PLANNER_PROMPT, - CRITIC_PROMPT, - HUMAN_APPROVAL_PROMPT, - LLM_ENRICHMENT_PROMPT, - PLANNER_PROMPT, - REFLECTION_PROMPT, - SPECIALIST_AGENT_PROMPT, - SPECIALIST_REASONING_PROMPT, -) - -logger = get_logger(__name__) - - -def _provider_client(settings): - if settings.provider == "gemini": - return GeminiClient(settings) - if settings.provider == "vertex": - return VertexGeminiClient(settings) - if settings.provider == "fpt": - return FPTClient(settings) - return None - - -ENRICHMENT_SCHEMA = { - "type": "object", - "properties": { - "planner_narrative": {"type": "string"}, - "operator_explanation": {"type": "string"}, - "approval_summary": {"type": "string"}, - }, - "required": [ - "planner_narrative", - "operator_explanation", - "approval_summary", - ], -} - -SPECIALIST_SCHEMA = { - "type": "object", - "properties": { - "domain_summary": {"type": "string"}, - "downstream_impacts": { - "type": "array", - "items": {"type": "string"}, - }, - "recommended_action_ids": { - "type": "array", - "items": {"type": "string"}, - }, - "tradeoffs": { - "type": "array", - "items": {"type": "string"}, - }, - "notes_for_planner": {"type": "string"}, - }, - "required": [ - "domain_summary", - "downstream_impacts", - "recommended_action_ids", - "tradeoffs", - "notes_for_planner", - ], -} - -PLANNER_CANDIDATES_SCHEMA = { - "type": "object", - "properties": { - "candidate_plans": { - "type": "array", - "items": { - "type": "object", - "properties": { - "strategy_label": {"type": "string"}, - "action_ids": { - "type": "array", - "items": {"type": "string"}, - }, - "rationale": {"type": "string"}, - }, - "required": ["strategy_label", "action_ids", "rationale"], - }, - } - }, - "required": ["candidate_plans"], -} - -CRITIC_SCHEMA = { - "type": "object", - "properties": { - "summary": {"type": "string"}, - "findings": { - "type": "array", - "items": {"type": "string"}, - }, - }, - "required": ["summary", "findings"], -} - -REFLECTION_SCHEMA = { - "type": "object", - "properties": { - "summary": {"type": "string"}, - "lessons": { - "type": "array", - "items": {"type": "string"}, - }, - "pattern_tags": { - "type": "array", - "items": {"type": "string"}, - }, - "follow_up_checks": { - "type": "array", - "items": {"type": "string"}, - }, - }, - "required": ["summary", "lessons", "pattern_tags", "follow_up_checks"], -} - - -def _event_context(event: Event | None) -> dict: - if event is None: - return {} - return { - "event_id": event.event_id, - "type": event.type.value, - "severity": event.severity, - "entity_ids": event.entity_ids, - "payload": event.payload, - } - - -def _call_json_model( - *, - prompt: str, - schema: dict, - capability: str = "general", -) -> tuple[dict | None, str | None, str | None]: - settings = load_settings() - if not settings.enabled: - logger.info("llm capability=%s skipped: llm disabled", capability) - return None, None, "llm_disabled" if capability == "planner" else None - if capability == "planner" and settings.planner_mode == "deterministic": - logger.info("llm capability=%s skipped: planner mode is deterministic", capability) - return None, settings.provider, "planner_mode=deterministic" - client = _provider_client(settings) - if client is None: - logger.warning( - "llm capability=%s skipped: unsupported provider=%s", - capability, - settings.provider, - ) - return None, settings.provider, f"unsupported provider: {settings.provider}" - last_error: str | None = None - for attempt in range(1, settings.retry_attempts + 1): - logger.info( - "llm capability=%s call start provider=%s model=%s attempt=%s/%s", - capability, - settings.provider, - settings.model, - attempt, - settings.retry_attempts, - ) - try: - response = client.generate_json( - prompt=prompt, - schema=schema, - ) - except (GeminiClientError, VertexClientError, FPTClientError) as exc: - last_error = str(exc) - logger.warning( - "llm capability=%s call failed attempt=%s/%s error=%s", - capability, - attempt, - settings.retry_attempts, - last_error, - ) - if attempt >= settings.retry_attempts: - break - continue - logger.info( - "llm capability=%s call succeeded provider=%s model=%s attempt=%s/%s", - capability, - settings.provider, - settings.model, - attempt, - settings.retry_attempts, - ) - return response, settings.provider, None - return None, settings.provider, last_error - - -def _action_catalog(actions: list[Action]) -> list[dict[str, Any]]: - return [ - { - "action_id": action.action_id, - "action_type": action.action_type.value, - "target_id": action.target_id, - "reason": action.reason, - "priority": action.priority, - "estimated_cost_delta": action.estimated_cost_delta, - "estimated_service_delta": action.estimated_service_delta, - "estimated_risk_delta": action.estimated_risk_delta, - "estimated_recovery_hours": action.estimated_recovery_hours, - } - for action in actions - ] - - -def _normalize_text(value: str) -> str: - return re.sub(r"[^a-z0-9]+", "", value.strip().lower()) - - -def _normalize_strategy_label(raw_value: str, rationale: str = "") -> str: - value = raw_value.strip().lower() - normalized = _normalize_text(raw_value) - rationale_normalized = _normalize_text(rationale) - if value in {"cost_first", "balanced", "resilience_first"}: - return value - alias_map = { - "costfirst": "cost_first", - "costoptimized": "cost_first", - "costplan": "cost_first", - "plana": "cost_first", - "strategya": "cost_first", - "optiona": "cost_first", - "balanced": "balanced", - "balancedplan": "balanced", - "planb": "balanced", - "strategyb": "balanced", - "optionb": "balanced", - "resiliencefirst": "resilience_first", - "resilientfirst": "resilience_first", - "resilienceplan": "resilience_first", - "planc": "resilience_first", - "strategyc": "resilience_first", - "optionc": "resilience_first", - } - if normalized in alias_map: - return alias_map[normalized] - if "cost" in normalized: - return "cost_first" - if "balanc" in normalized: - return "balanced" - if "resilien" in normalized or "recover" in normalized: - return "resilience_first" - if "cost" in rationale_normalized: - return "cost_first" - if "balance" in rationale_normalized: - return "balanced" - if any(token in rationale_normalized for token in {"resilien", "recover", "service", "stockout"}): - return "resilience_first" - return raw_value.strip().lower() - - -def _action_aliases(action: Action) -> set[str]: - aliases = { - action.action_id, - action.action_id.lower(), - _normalize_text(action.action_id), - action.target_id.lower(), - _normalize_text(action.target_id), - f"{action.action_type.value}:{action.target_id}".lower(), - _normalize_text(f"{action.action_type.value}:{action.target_id}"), - _normalize_text(f"{action.action_type.value} {action.target_id}"), - _normalize_text(action.reason), - } - supplier_id = action.parameters.get("supplier_id") - route_id = action.parameters.get("route_id") - quantity = action.parameters.get("quantity") - if supplier_id: - aliases.add(str(supplier_id).lower()) - aliases.add(_normalize_text(str(supplier_id))) - aliases.add(_normalize_text(f"{action.target_id} {supplier_id}")) - if route_id: - aliases.add(str(route_id).lower()) - aliases.add(_normalize_text(str(route_id))) - aliases.add(_normalize_text(f"{action.target_id} {route_id}")) - if quantity is not None: - aliases.add(_normalize_text(f"{action.target_id} {quantity}")) - return {alias for alias in aliases if alias} - - -def _action_alias_map(candidate_actions: list[Action]) -> dict[str, str]: - alias_map: dict[str, str] = {} - for action in candidate_actions: - for alias in _action_aliases(action): - alias_map.setdefault(alias, action.action_id) - return alias_map - - -def _extract_candidate_action_refs(item: dict[str, Any]) -> list[str]: - refs: list[str] = [] - direct_fields = [ - item.get("action_ids"), - item.get("recommended_action_ids"), - item.get("selected_actions"), - item.get("actions"), - ] - for field in direct_fields: - if isinstance(field, list): - for value in field: - if isinstance(value, dict): - for key in ("action_id", "id", "ref", "name"): - if value.get(key): - refs.append(str(value[key])) - break - elif value is not None: - refs.append(str(value)) - return refs - - -def _build_specialist_prompt( - *, - agent_name: str, - state: SystemState, - event: Event | None, - proposal: AgentProposal, - state_slice: dict[str, Any], - custom_prompt: str | None = None, -) -> str: - context = { - "agent": agent_name, - "mode": state.mode.value, - "active_events": [_event_context(item) for item in state.active_events], - "trigger_event": _event_context(event), - "kpis": state.kpis.model_dump(mode="json"), - "state_slice": state_slice, - "current_observations": proposal.observations, - "current_risks": proposal.risks, - "candidate_actions": _action_catalog(proposal.proposals), - } - instruction_parts = [SPECIALIST_AGENT_PROMPT.format(agent_name=agent_name)] - if custom_prompt: - instruction_parts.append(custom_prompt.strip()) - instruction_parts.extend( - [ - SPECIALIST_REASONING_PROMPT.strip(), - ( - "Return JSON with keys domain_summary, downstream_impacts, " - "recommended_action_ids, tradeoffs, notes_for_planner. " - "Only reference action_ids that appear in candidate_actions. " - "If no action is recommended, return an empty recommended_action_ids array." - ), - ] - ) - instructions = "\n\n".join(instruction_parts) - return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" - - -def _build_planner_candidates_prompt( - *, - state: SystemState, - event: Event | None, - candidate_actions: list[Action], -) -> str: - context = { - "mode": state.mode.value, - "active_events": [_event_context(item) for item in state.active_events], - "trigger_event": _event_context(event), - "kpis": state.kpis.model_dump(mode="json"), - "specialist_outputs": { - name: { - "observations": output.observations, - "risks": output.risks, - "domain_summary": output.domain_summary, - "downstream_impacts": output.downstream_impacts, - "recommended_action_ids": output.recommended_action_ids, - "tradeoffs": output.tradeoffs, - "notes_for_planner": output.notes_for_planner, - } - for name, output in state.agent_outputs.items() - if name != "planner" - }, - "candidate_actions": _action_catalog(candidate_actions), - } - instructions = [ - AI_CANDIDATE_PLANNER_PROMPT.strip(), - PLANNER_PROMPT.strip(), - ] - if state.mode.value == "crisis": - from orchestrator.prompts import CRISIS_MODE_PROMPT - - instructions.append(CRISIS_MODE_PROMPT.strip()) +from __future__ import annotations + +import json +import re +from typing import Any + +from core.logger import get_logger +from core.models import ( + Action, + AgentProposal, + CandidatePlanDraft, + CandidatePlanEvaluation, + DecisionLog, + Event, + Plan, + ReflectionNote, + ScenarioRun, + SystemState, +) +from llm.config import load_settings +from llm.gemini_client import GeminiClient, GeminiClientError +from llm.vertex_client import VertexClientError, VertexGeminiClient +from llm.fpt_client import FPTClient, FPTClientError +from orchestrator.prompts import ( + DECISION_EXPLANATION_PROMPT, + AI_CANDIDATE_PLANNER_PROMPT, + CRITIC_PROMPT, + HUMAN_APPROVAL_PROMPT, + LLM_ENRICHMENT_PROMPT, + PLANNER_PROMPT, + REFLECTION_PROMPT, + SPECIALIST_AGENT_PROMPT, + SPECIALIST_REASONING_PROMPT, +) + +logger = get_logger(__name__) + + +def _provider_client(settings): + if settings.provider == "gemini": + return GeminiClient(settings) + if settings.provider == "vertex": + return VertexGeminiClient(settings) + if settings.provider == "fpt": + return FPTClient(settings) + return None + + +ENRICHMENT_SCHEMA = { + "type": "object", + "properties": { + "planner_narrative": {"type": "string"}, + "operator_explanation": {"type": "string"}, + "approval_summary": {"type": "string"}, + }, + "required": [ + "planner_narrative", + "operator_explanation", + "approval_summary", + ], +} + +SPECIALIST_SCHEMA = { + "type": "object", + "properties": { + "domain_summary": {"type": "string"}, + "downstream_impacts": { + "type": "array", + "items": {"type": "string"}, + }, + "recommended_action_ids": { + "type": "array", + "items": {"type": "string"}, + }, + "tradeoffs": { + "type": "array", + "items": {"type": "string"}, + }, + "notes_for_planner": {"type": "string"}, + }, + "required": [ + "domain_summary", + "downstream_impacts", + "recommended_action_ids", + "tradeoffs", + "notes_for_planner", + ], +} + +PLANNER_CANDIDATES_SCHEMA = { + "type": "object", + "properties": { + "candidate_plans": { + "type": "array", + "items": { + "type": "object", + "properties": { + "strategy_label": {"type": "string"}, + "action_ids": { + "type": "array", + "items": {"type": "string"}, + }, + "rationale": {"type": "string"}, + }, + "required": ["strategy_label", "action_ids", "rationale"], + }, + } + }, + "required": ["candidate_plans"], +} + +CRITIC_SCHEMA = { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "findings": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["summary", "findings"], +} + +REFLECTION_SCHEMA = { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "lessons": { + "type": "array", + "items": {"type": "string"}, + }, + "pattern_tags": { + "type": "array", + "items": {"type": "string"}, + }, + "follow_up_checks": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["summary", "lessons", "pattern_tags", "follow_up_checks"], +} + + +def _event_context(event: Event | None) -> dict: + if event is None: + return {} + return { + "event_id": event.event_id, + "type": event.type.value, + "severity": event.severity, + "entity_ids": event.entity_ids, + "payload": event.payload, + } + + +def _call_json_model( + *, + prompt: str, + schema: dict, + capability: str = "general", +) -> tuple[dict | None, str | None, str | None]: + settings = load_settings() + if not settings.enabled: + logger.info("llm capability=%s skipped: llm disabled", capability) + return None, None, "llm_disabled" if capability == "planner" else None + if capability == "planner" and settings.planner_mode == "deterministic": + logger.info("llm capability=%s skipped: planner mode is deterministic", capability) + return None, settings.provider, "planner_mode=deterministic" + client = _provider_client(settings) + if client is None: + logger.warning( + "llm capability=%s skipped: unsupported provider=%s", + capability, + settings.provider, + ) + return None, settings.provider, f"unsupported provider: {settings.provider}" + last_error: str | None = None + for attempt in range(1, settings.retry_attempts + 1): + logger.info( + "llm capability=%s call start provider=%s model=%s attempt=%s/%s", + capability, + settings.provider, + settings.model, + attempt, + settings.retry_attempts, + ) + try: + response = client.generate_json( + prompt=prompt, + schema=schema, + ) + except (GeminiClientError, VertexClientError, FPTClientError) as exc: + last_error = str(exc) + logger.warning( + "llm capability=%s call failed attempt=%s/%s error=%s", + capability, + attempt, + settings.retry_attempts, + last_error, + ) + if attempt >= settings.retry_attempts: + break + continue + logger.info( + "llm capability=%s call succeeded provider=%s model=%s attempt=%s/%s", + capability, + settings.provider, + settings.model, + attempt, + settings.retry_attempts, + ) + return response, settings.provider, None + return None, settings.provider, last_error + + +def _action_catalog(actions: list[Action]) -> list[dict[str, Any]]: + return [ + { + "action_id": action.action_id, + "action_type": action.action_type.value, + "target_id": action.target_id, + "reason": action.reason, + "priority": action.priority, + "estimated_cost_delta": action.estimated_cost_delta, + "estimated_service_delta": action.estimated_service_delta, + "estimated_risk_delta": action.estimated_risk_delta, + "estimated_recovery_hours": action.estimated_recovery_hours, + } + for action in actions + ] + + +def _normalize_text(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "", value.strip().lower()) + + +def _normalize_strategy_label(raw_value: str, rationale: str = "") -> str: + value = raw_value.strip().lower() + normalized = _normalize_text(raw_value) + rationale_normalized = _normalize_text(rationale) + if value in {"cost_first", "balanced", "resilience_first"}: + return value + alias_map = { + "costfirst": "cost_first", + "costoptimized": "cost_first", + "costplan": "cost_first", + "plana": "cost_first", + "strategya": "cost_first", + "optiona": "cost_first", + "balanced": "balanced", + "balancedplan": "balanced", + "planb": "balanced", + "strategyb": "balanced", + "optionb": "balanced", + "resiliencefirst": "resilience_first", + "resilientfirst": "resilience_first", + "resilienceplan": "resilience_first", + "planc": "resilience_first", + "strategyc": "resilience_first", + "optionc": "resilience_first", + } + if normalized in alias_map: + return alias_map[normalized] + if "cost" in normalized: + return "cost_first" + if "balanc" in normalized: + return "balanced" + if "resilien" in normalized or "recover" in normalized: + return "resilience_first" + if "cost" in rationale_normalized: + return "cost_first" + if "balance" in rationale_normalized: + return "balanced" + if any(token in rationale_normalized for token in {"resilien", "recover", "service", "stockout"}): + return "resilience_first" + return raw_value.strip().lower() + + +def _action_aliases(action: Action) -> set[str]: + aliases = { + action.action_id, + action.action_id.lower(), + _normalize_text(action.action_id), + action.target_id.lower(), + _normalize_text(action.target_id), + f"{action.action_type.value}:{action.target_id}".lower(), + _normalize_text(f"{action.action_type.value}:{action.target_id}"), + _normalize_text(f"{action.action_type.value} {action.target_id}"), + _normalize_text(action.reason), + } + supplier_id = action.parameters.get("supplier_id") + route_id = action.parameters.get("route_id") + quantity = action.parameters.get("quantity") + if supplier_id: + aliases.add(str(supplier_id).lower()) + aliases.add(_normalize_text(str(supplier_id))) + aliases.add(_normalize_text(f"{action.target_id} {supplier_id}")) + if route_id: + aliases.add(str(route_id).lower()) + aliases.add(_normalize_text(str(route_id))) + aliases.add(_normalize_text(f"{action.target_id} {route_id}")) + if quantity is not None: + aliases.add(_normalize_text(f"{action.target_id} {quantity}")) + return {alias for alias in aliases if alias} + + +def _action_alias_map(candidate_actions: list[Action]) -> dict[str, str]: + alias_map: dict[str, str] = {} + for action in candidate_actions: + for alias in _action_aliases(action): + alias_map.setdefault(alias, action.action_id) + return alias_map + + +def _extract_candidate_action_refs(item: dict[str, Any]) -> list[str]: + refs: list[str] = [] + direct_fields = [ + item.get("action_ids"), + item.get("recommended_action_ids"), + item.get("selected_actions"), + item.get("actions"), + ] + for field in direct_fields: + if isinstance(field, list): + for value in field: + if isinstance(value, dict): + for key in ("action_id", "id", "ref", "name"): + if value.get(key): + refs.append(str(value[key])) + break + elif value is not None: + refs.append(str(value)) + return refs + + +def _build_specialist_prompt( + *, + agent_name: str, + state: SystemState, + event: Event | None, + proposal: AgentProposal, + state_slice: dict[str, Any], + custom_prompt: str | None = None, +) -> str: + context = { + "agent": agent_name, + "mode": state.mode.value, + "active_events": [_event_context(item) for item in state.active_events], + "trigger_event": _event_context(event), + "kpis": state.kpis.model_dump(mode="json"), + "state_slice": state_slice, + "current_observations": proposal.observations, + "current_risks": proposal.risks, + "candidate_actions": _action_catalog(proposal.proposals), + } + instruction_parts = [SPECIALIST_AGENT_PROMPT.format(agent_name=agent_name)] + if custom_prompt: + instruction_parts.append(custom_prompt.strip()) + instruction_parts.extend( + [ + SPECIALIST_REASONING_PROMPT.strip(), + ( + "Return JSON with keys domain_summary, downstream_impacts, " + "recommended_action_ids, tradeoffs, notes_for_planner. " + "Only reference action_ids that appear in candidate_actions. " + "If no action is recommended, return an empty recommended_action_ids array." + ), + ] + ) + instructions = "\n\n".join(instruction_parts) + return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" + + +def _build_planner_candidates_prompt( + *, + state: SystemState, + event: Event | None, + candidate_actions: list[Action], + memory_prompt: str | None = None, +) -> str: + context = { + "mode": state.mode.value, + "active_events": [_event_context(item) for item in state.active_events], + "trigger_event": _event_context(event), + "kpis": state.kpis.model_dump(mode="json"), + "specialist_outputs": { + name: { + "observations": output.observations, + "risks": output.risks, + "domain_summary": output.domain_summary, + "downstream_impacts": output.downstream_impacts, + "recommended_action_ids": output.recommended_action_ids, + "tradeoffs": output.tradeoffs, + "notes_for_planner": output.notes_for_planner, + } + for name, output in state.agent_outputs.items() + if name != "planner" + }, + "candidate_actions": _action_catalog(candidate_actions), + } + instructions = [ + AI_CANDIDATE_PLANNER_PROMPT.strip(), + PLANNER_PROMPT.strip(), + ] + if state.mode.value == "crisis": + from orchestrator.prompts import CRISIS_MODE_PROMPT + + instructions.append(CRISIS_MODE_PROMPT.strip()) instructions.append( "Return JSON with candidate_plans containing exactly these strategy_label values: " "cost_first, balanced, resilience_first. Only use action ids from candidate_actions." @@ -414,394 +415,408 @@ def _build_planner_candidates_prompt( "Choose a right-sized action_ids list for each strategy. " "Avoid selecting all candidate actions unless truly necessary for that strategy." ) - instructions.append( - "Allowed planner output shape example: " - '{"candidate_plans":[{"strategy_label":"cost_first","action_ids":["act_x"],"rationale":"..."},' - '{"strategy_label":"balanced","action_ids":["act_y"],"rationale":"..."},' - '{"strategy_label":"resilience_first","action_ids":["act_z"],"rationale":"..."}]}' - ) - instruction_text = "\n\n".join(instructions) - return f"{instruction_text}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" - - -def _build_critic_prompt( - *, - state: SystemState, - event: Event | None, - selected_plan: Plan, - evaluations: list[CandidatePlanEvaluation], -) -> str: - context = { - "mode": state.mode.value, - "active_events": [_event_context(item) for item in state.active_events], - "trigger_event": _event_context(event), - "selected_plan": { - "plan_id": selected_plan.plan_id, - "strategy_label": selected_plan.strategy_label, - "approval_required": selected_plan.approval_required, - "approval_reason": selected_plan.approval_reason, - "score": selected_plan.score, - "score_breakdown": selected_plan.score_breakdown, - "action_ids": [action.action_id for action in selected_plan.actions], - }, - "candidate_evaluations": [evaluation.model_dump(mode="json") for evaluation in evaluations], - } - instructions = "\n\n".join( - [ - CRITIC_PROMPT.strip(), - "Return JSON with keys summary and findings. Keep findings concise and grounded in the evaluated candidates.", - ] - ) - return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" - - -def _build_reflection_prompt( - *, - state: SystemState, - run: ScenarioRun, - decision_log: DecisionLog | None, -) -> str: - context = { - "scenario_run": run.model_dump(mode="json"), - "active_events": [_event_context(item) for item in state.active_events], - "current_kpis": state.kpis.model_dump(mode="json"), - "latest_plan": state.latest_plan.model_dump(mode="json") if state.latest_plan else None, - "latest_decision": decision_log.model_dump(mode="json") if decision_log else None, - } - instructions = "\n\n".join( - [ - REFLECTION_PROMPT.strip(), - ( - "Return JSON with keys summary, lessons, pattern_tags, follow_up_checks. " - "Keep each field concise and grounded in the actual outcome." - ), - ] - ) - return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" - - -def _unique_action_ids(action_ids: list[str], allowed_ids: set[str]) -> list[str]: - unique: list[str] = [] - seen: set[str] = set() - for action_id in action_ids: - if action_id in allowed_ids and action_id not in seen: - unique.append(action_id) - seen.add(action_id) - return unique - - -def _apply_ranked_actions(proposal: AgentProposal, ranked_action_ids: list[str]) -> None: - if not ranked_action_ids: - return - by_id = {action.action_id: action for action in proposal.proposals} - ranked_actions = [by_id[action_id] for action_id in ranked_action_ids if action_id in by_id] - if not ranked_actions: - return - current_max = max((action.priority for action in proposal.proposals), default=0.0) - top_priority = min(1.0, current_max + 0.1) - for index, action in enumerate(ranked_actions): - action.priority = round(max(0.0, top_priority - (index * 0.03)), 4) - unranked_actions = [action for action in proposal.proposals if action.action_id not in ranked_action_ids] - proposal.proposals = ranked_actions + unranked_actions - - -def enrich_specialist_proposal( - *, - agent_name: str, - state: SystemState, - event: Event | None, - proposal: AgentProposal, - state_slice: dict[str, Any], - custom_prompt: str | None = None, -) -> None: - prompt = _build_specialist_prompt( - agent_name=agent_name, - state=state, - event=event, - proposal=proposal, - state_slice=state_slice, - custom_prompt=custom_prompt, - ) - response, provider, error = _call_json_model( - prompt=prompt, - schema=SPECIALIST_SCHEMA, - capability="specialist", - ) - proposal.llm_used = False - proposal.llm_error = error - if response is None: - return - - proposal.domain_summary = str(response.get("domain_summary", "")).strip() - proposal.downstream_impacts = [ - str(item).strip() - for item in response.get("downstream_impacts", []) - if str(item).strip() - ] - proposal.tradeoffs = [ - str(item).strip() - for item in response.get("tradeoffs", []) - if str(item).strip() - ] - notes_for_planner = str(response.get("notes_for_planner", "")).strip() - if notes_for_planner: - proposal.notes_for_planner = notes_for_planner - - allowed_ids = {action.action_id for action in proposal.proposals} - ranked_action_ids = _unique_action_ids( - [str(item).strip() for item in response.get("recommended_action_ids", []) if str(item).strip()], - allowed_ids, - ) - if response.get("recommended_action_ids") and not ranked_action_ids and allowed_ids: - proposal.llm_error = "llm returned no valid action ids" - return - - proposal.recommended_action_ids = ranked_action_ids - _apply_ranked_actions(proposal, ranked_action_ids) - proposal.llm_used = any( - [ - proposal.domain_summary, - proposal.downstream_impacts, - proposal.tradeoffs, - proposal.notes_for_planner, - proposal.recommended_action_ids, - ] - ) - if not proposal.llm_used and provider: - proposal.llm_error = proposal.llm_error or f"{provider} returned an empty specialist response" - - -def generate_candidate_plan_drafts( - *, - state: SystemState, - event: Event | None, - candidate_actions: list[Action], -) -> tuple[list[CandidatePlanDraft], str | None]: - prompt = _build_planner_candidates_prompt( - state=state, - event=event, - candidate_actions=candidate_actions, - ) - response, _, error = _call_json_model( - prompt=prompt, - schema=PLANNER_CANDIDATES_SCHEMA, - capability="planner", - ) - if response is None: - logger.info("planner llm path not used: %s", error or "unknown_reason") - return [], error - - allowed_ids = {action.action_id for action in candidate_actions} - alias_map = _action_alias_map(candidate_actions) - drafts: list[CandidatePlanDraft] = [] - for item in response.get("candidate_plans", []): - rationale = str(item.get("rationale", "")).strip() - strategy_label = _normalize_strategy_label(str(item.get("strategy_label", "")).strip(), rationale) - extracted_refs = _extract_candidate_action_refs(item) - action_ids = _unique_action_ids( - [ - alias_map.get(candidate_ref, alias_map.get(_normalize_text(candidate_ref), candidate_ref)) - for candidate_ref in extracted_refs - if candidate_ref.strip() - ], - allowed_ids, - ) - drafts.append( - CandidatePlanDraft( - strategy_label=strategy_label, - action_ids=action_ids, - rationale=rationale, - llm_used=True, - ) - ) - logger.info( - "planner llm returned candidate_plans=%s normalized_drafts=%s", - len(response.get("candidate_plans", [])), - len(drafts), - ) - return drafts, None - - -def critique_candidate_plans( - *, - state: SystemState, - event: Event | None, - selected_plan: Plan, - evaluations: list[CandidatePlanEvaluation], -) -> tuple[str | None, list[str], bool, str | None]: - prompt = _build_critic_prompt( - state=state, - event=event, - selected_plan=selected_plan, - evaluations=evaluations, - ) - response, _, error = _call_json_model( - prompt=prompt, - schema=CRITIC_SCHEMA, - capability="critic", - ) - if response is None: - return None, [], False, error - summary = str(response.get("summary", "")).strip() or None - findings = [str(item).strip() for item in response.get("findings", []) if str(item).strip()] - used = bool(summary or findings) - return summary, findings, used, None - - -def generate_reflection_note( - *, - state: SystemState, - run: ScenarioRun, - decision_log: DecisionLog | None, -) -> tuple[ReflectionNote | None, str | None]: - prompt = _build_reflection_prompt( - state=state, - run=run, - decision_log=decision_log, - ) - response, _, error = _call_json_model( - prompt=prompt, - schema=REFLECTION_SCHEMA, - capability="reflection", - ) - if response is None: - return None, error - note = ReflectionNote( - note_id=f"note_{run.run_id}", - run_id=run.run_id, - scenario_id=run.scenario_id, - plan_id=run.result_plan_id, - event_types=[event.type.value for event in run.events], - mode=state.mode.value, - approval_status=run.approval_status or "unknown", - summary=str(response.get("summary", "")).strip(), - lessons=[str(item).strip() for item in response.get("lessons", []) if str(item).strip()], - pattern_tags=[str(item).strip() for item in response.get("pattern_tags", []) if str(item).strip()], - follow_up_checks=[ - str(item).strip() for item in response.get("follow_up_checks", []) if str(item).strip() - ], - llm_used=True, - llm_error=None, - ) - return note, None - - -def _build_prompt( - *, - state: SystemState, - event: Event | None, - plan: Plan, - decision_log: DecisionLog, -) -> str: - context = { - "mode": state.mode.value, - "active_events": [_event_context(item) for item in state.active_events], - "trigger_event": _event_context(event), - "plan": { - "plan_id": plan.plan_id, - "score": plan.score, - "score_breakdown": plan.score_breakdown, - "approval_required": plan.approval_required, - "approval_reason": plan.approval_reason, - "selected_actions": [ - { - "action_id": action.action_id, - "action_type": action.action_type.value, - "target_id": action.target_id, - "reason": action.reason, - "estimated_cost_delta": action.estimated_cost_delta, - "estimated_service_delta": action.estimated_service_delta, - "estimated_risk_delta": action.estimated_risk_delta, - "estimated_recovery_hours": action.estimated_recovery_hours, - } - for action in plan.actions - ], - }, - "decision_log": { - "decision_id": decision_log.decision_id, - "before_kpis": decision_log.before_kpis.model_dump(mode="json"), - "after_kpis": decision_log.after_kpis.model_dump(mode="json"), - "rejected_actions": decision_log.rejected_actions, - "winning_factors": decision_log.winning_factors, - "deterministic_rationale": decision_log.rationale, - }, - } - instructions = "\n\n".join( - [ - PLANNER_PROMPT.strip(), - DECISION_EXPLANATION_PROMPT.strip(), - HUMAN_APPROVAL_PROMPT.strip(), - LLM_ENRICHMENT_PROMPT.strip(), - ( - "Return JSON with keys planner_narrative, operator_explanation, approval_summary. " - "Use only the provided facts. Keep each field concise. If no approval is required, " - "set approval_summary to an empty string." - ), - ] - ) - return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" - - -def enrich_plan_and_decision( - *, - state: SystemState, - event: Event | None, - plan: Plan, - decision_log: DecisionLog, -) -> None: - settings = load_settings() - if not settings.enabled: - decision_log.llm_used = False - decision_log.llm_provider = None - decision_log.llm_model = None - decision_log.llm_error = None - plan.llm_planner_narrative = None - return - - decision_log.llm_provider = settings.provider - decision_log.llm_model = settings.model - client = _provider_client(settings) - if client is None: - decision_log.llm_used = False - decision_log.llm_error = f"unsupported provider: {settings.provider}" - return - - prompt = _build_prompt( - state=state, event=event, plan=plan, decision_log=decision_log - ) - try: - logger.info( - "llm capability=enrichment call start provider=%s model=%s", - settings.provider, - settings.model, - ) - response = client.generate_json( - prompt=prompt, - schema=ENRICHMENT_SCHEMA, - ) - except (GeminiClientError, VertexClientError, FPTClientError) as exc: - logger.warning( - "llm capability=enrichment call failed provider=%s model=%s error=%s", - settings.provider, - settings.model, - exc, - ) - decision_log.llm_used = False - decision_log.llm_error = str(exc) - plan.llm_planner_narrative = None - return - logger.info( - "llm capability=enrichment call succeeded provider=%s model=%s", - settings.provider, - settings.model, - ) - - planner_narrative = str(response.get("planner_narrative", "")).strip() - operator_explanation = str(response.get("operator_explanation", "")).strip() - approval_summary = str(response.get("approval_summary", "")).strip() - - plan.llm_planner_narrative = planner_narrative or None - decision_log.llm_operator_explanation = operator_explanation or None - decision_log.llm_approval_summary = approval_summary or None - decision_log.llm_used = any([planner_narrative, operator_explanation, approval_summary]) - decision_log.llm_error = None + instructions.append( + "Allowed planner output shape example: " + '{"candidate_plans":[{"strategy_label":"cost_first","action_ids":["act_x"],"rationale":"..."},' + '{"strategy_label":"balanced","action_ids":["act_y"],"rationale":"..."},' + '{"strategy_label":"resilience_first","action_ids":["act_z"],"rationale":"..."}]}' + ) + instruction_text = "\n\n".join(instructions) + + prompt_parts = [] + # 1. Base instructions + prompt_parts.append(instruction_text) + + # 2. Memory Context (Historical Analogies, Directives) + if memory_prompt: + prompt_parts.append("## STRATEGIC MEMORY & DIRECTIVES\n" + memory_prompt) + + # 3. Operational Data (JSON) + prompt_parts.append("## CURRENT OPERATIONAL DATA\n" + json.dumps(context, ensure_ascii=True, indent=2)) + + return "\n\n".join(prompt_parts) + + +def _build_critic_prompt( + *, + state: SystemState, + event: Event | None, + selected_plan: Plan, + evaluations: list[CandidatePlanEvaluation], +) -> str: + context = { + "mode": state.mode.value, + "active_events": [_event_context(item) for item in state.active_events], + "trigger_event": _event_context(event), + "selected_plan": { + "plan_id": selected_plan.plan_id, + "strategy_label": selected_plan.strategy_label, + "approval_required": selected_plan.approval_required, + "approval_reason": selected_plan.approval_reason, + "score": selected_plan.score, + "score_breakdown": selected_plan.score_breakdown, + "action_ids": [action.action_id for action in selected_plan.actions], + }, + "candidate_evaluations": [evaluation.model_dump(mode="json") for evaluation in evaluations], + } + instructions = "\n\n".join( + [ + CRITIC_PROMPT.strip(), + "Return JSON with keys summary and findings. Keep findings concise and grounded in the evaluated candidates.", + ] + ) + return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" + + +def _build_reflection_prompt( + *, + state: SystemState, + run: ScenarioRun, + decision_log: DecisionLog | None, +) -> str: + context = { + "scenario_run": run.model_dump(mode="json"), + "active_events": [_event_context(item) for item in state.active_events], + "current_kpis": state.kpis.model_dump(mode="json"), + "latest_plan": state.latest_plan.model_dump(mode="json") if state.latest_plan else None, + "latest_decision": decision_log.model_dump(mode="json") if decision_log else None, + } + instructions = "\n\n".join( + [ + REFLECTION_PROMPT.strip(), + ( + "Return JSON with keys summary, lessons, pattern_tags, follow_up_checks. " + "Keep each field concise and grounded in the actual outcome." + ), + ] + ) + return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" + + +def _unique_action_ids(action_ids: list[str], allowed_ids: set[str]) -> list[str]: + unique: list[str] = [] + seen: set[str] = set() + for action_id in action_ids: + if action_id in allowed_ids and action_id not in seen: + unique.append(action_id) + seen.add(action_id) + return unique + + +def _apply_ranked_actions(proposal: AgentProposal, ranked_action_ids: list[str]) -> None: + if not ranked_action_ids: + return + by_id = {action.action_id: action for action in proposal.proposals} + ranked_actions = [by_id[action_id] for action_id in ranked_action_ids if action_id in by_id] + if not ranked_actions: + return + current_max = max((action.priority for action in proposal.proposals), default=0.0) + top_priority = min(1.0, current_max + 0.1) + for index, action in enumerate(ranked_actions): + action.priority = round(max(0.0, top_priority - (index * 0.03)), 4) + unranked_actions = [action for action in proposal.proposals if action.action_id not in ranked_action_ids] + proposal.proposals = ranked_actions + unranked_actions + + +def enrich_specialist_proposal( + *, + agent_name: str, + state: SystemState, + event: Event | None, + proposal: AgentProposal, + state_slice: dict[str, Any], + custom_prompt: str | None = None, +) -> None: + prompt = _build_specialist_prompt( + agent_name=agent_name, + state=state, + event=event, + proposal=proposal, + state_slice=state_slice, + custom_prompt=custom_prompt, + ) + response, provider, error = _call_json_model( + prompt=prompt, + schema=SPECIALIST_SCHEMA, + capability="specialist", + ) + proposal.llm_used = False + proposal.llm_error = error + if response is None: + return + + proposal.domain_summary = str(response.get("domain_summary", "")).strip() + proposal.downstream_impacts = [ + str(item).strip() + for item in response.get("downstream_impacts", []) + if str(item).strip() + ] + proposal.tradeoffs = [ + str(item).strip() + for item in response.get("tradeoffs", []) + if str(item).strip() + ] + notes_for_planner = str(response.get("notes_for_planner", "")).strip() + if notes_for_planner: + proposal.notes_for_planner = notes_for_planner + + allowed_ids = {action.action_id for action in proposal.proposals} + ranked_action_ids = _unique_action_ids( + [str(item).strip() for item in response.get("recommended_action_ids", []) if str(item).strip()], + allowed_ids, + ) + if response.get("recommended_action_ids") and not ranked_action_ids and allowed_ids: + proposal.llm_error = "llm returned no valid action ids" + return + + proposal.recommended_action_ids = ranked_action_ids + _apply_ranked_actions(proposal, ranked_action_ids) + proposal.llm_used = any( + [ + proposal.domain_summary, + proposal.downstream_impacts, + proposal.tradeoffs, + proposal.notes_for_planner, + proposal.recommended_action_ids, + ] + ) + if not proposal.llm_used and provider: + proposal.llm_error = proposal.llm_error or f"{provider} returned an empty specialist response" + + +def generate_candidate_plan_drafts( + *, + state: SystemState, + event: Event | None, + candidate_actions: list[Action], + memory_prompt: str | None = None, +) -> tuple[list[CandidatePlanDraft], str | None]: + prompt = _build_planner_candidates_prompt( + state=state, + event=event, + candidate_actions=candidate_actions, + memory_prompt=memory_prompt + ) + response, _, error = _call_json_model( + prompt=prompt, + schema=PLANNER_CANDIDATES_SCHEMA, + capability="planner", + ) + if response is None: + logger.info("planner llm path not used: %s", error or "unknown_reason") + return [], error + + allowed_ids = {action.action_id for action in candidate_actions} + alias_map = _action_alias_map(candidate_actions) + drafts: list[CandidatePlanDraft] = [] + for item in response.get("candidate_plans", []): + rationale = str(item.get("rationale", "")).strip() + strategy_label = _normalize_strategy_label(str(item.get("strategy_label", "")).strip(), rationale) + extracted_refs = _extract_candidate_action_refs(item) + action_ids = _unique_action_ids( + [ + alias_map.get(candidate_ref, alias_map.get(_normalize_text(candidate_ref), candidate_ref)) + for candidate_ref in extracted_refs + if candidate_ref.strip() + ], + allowed_ids, + ) + drafts.append( + CandidatePlanDraft( + strategy_label=strategy_label, + action_ids=action_ids, + rationale=rationale, + llm_used=True, + ) + ) + logger.info( + "planner llm returned candidate_plans=%s normalized_drafts=%s", + len(response.get("candidate_plans", [])), + len(drafts), + ) + return drafts, None + + +def critique_candidate_plans( + *, + state: SystemState, + event: Event | None, + selected_plan: Plan, + evaluations: list[CandidatePlanEvaluation], +) -> tuple[str | None, list[str], bool, str | None]: + prompt = _build_critic_prompt( + state=state, + event=event, + selected_plan=selected_plan, + evaluations=evaluations, + ) + response, _, error = _call_json_model( + prompt=prompt, + schema=CRITIC_SCHEMA, + capability="critic", + ) + if response is None: + return None, [], False, error + summary = str(response.get("summary", "")).strip() or None + findings = [str(item).strip() for item in response.get("findings", []) if str(item).strip()] + used = bool(summary or findings) + return summary, findings, used, None + + +def generate_reflection_note( + *, + state: SystemState, + run: ScenarioRun, + decision_log: DecisionLog | None, +) -> tuple[ReflectionNote | None, str | None]: + prompt = _build_reflection_prompt( + state=state, + run=run, + decision_log=decision_log, + ) + response, _, error = _call_json_model( + prompt=prompt, + schema=REFLECTION_SCHEMA, + capability="reflection", + ) + if response is None: + return None, error + note = ReflectionNote( + note_id=f"note_{run.run_id}", + run_id=run.run_id, + scenario_id=run.scenario_id, + plan_id=run.result_plan_id, + event_types=[event.type.value for event in run.events], + mode=state.mode.value, + approval_status=run.approval_status or "unknown", + summary=str(response.get("summary", "")).strip(), + lessons=[str(item).strip() for item in response.get("lessons", []) if str(item).strip()], + pattern_tags=[str(item).strip() for item in response.get("pattern_tags", []) if str(item).strip()], + follow_up_checks=[ + str(item).strip() for item in response.get("follow_up_checks", []) if str(item).strip() + ], + llm_used=True, + llm_error=None, + ) + return note, None + + +def _build_prompt( + *, + state: SystemState, + event: Event | None, + plan: Plan, + decision_log: DecisionLog, +) -> str: + context = { + "mode": state.mode.value, + "active_events": [_event_context(item) for item in state.active_events], + "trigger_event": _event_context(event), + "plan": { + "plan_id": plan.plan_id, + "score": plan.score, + "score_breakdown": plan.score_breakdown, + "approval_required": plan.approval_required, + "approval_reason": plan.approval_reason, + "selected_actions": [ + { + "action_id": action.action_id, + "action_type": action.action_type.value, + "target_id": action.target_id, + "reason": action.reason, + "estimated_cost_delta": action.estimated_cost_delta, + "estimated_service_delta": action.estimated_service_delta, + "estimated_risk_delta": action.estimated_risk_delta, + "estimated_recovery_hours": action.estimated_recovery_hours, + } + for action in plan.actions + ], + }, + "decision_log": { + "decision_id": decision_log.decision_id, + "before_kpis": decision_log.before_kpis.model_dump(mode="json"), + "after_kpis": decision_log.after_kpis.model_dump(mode="json"), + "rejected_actions": decision_log.rejected_actions, + "winning_factors": decision_log.winning_factors, + "deterministic_rationale": decision_log.rationale, + }, + } + instructions = "\n\n".join( + [ + PLANNER_PROMPT.strip(), + DECISION_EXPLANATION_PROMPT.strip(), + HUMAN_APPROVAL_PROMPT.strip(), + LLM_ENRICHMENT_PROMPT.strip(), + ( + "Return JSON with keys planner_narrative, operator_explanation, approval_summary. " + "Use only the provided facts. Keep each field concise. If no approval is required, " + "set approval_summary to an empty string." + ), + ] + ) + return f"{instructions}\n\nContext:\n{json.dumps(context, ensure_ascii=True, indent=2)}" + + +def enrich_plan_and_decision( + *, + state: SystemState, + event: Event | None, + plan: Plan, + decision_log: DecisionLog, +) -> None: + settings = load_settings() + if not settings.enabled: + decision_log.llm_used = False + decision_log.llm_provider = None + decision_log.llm_model = None + decision_log.llm_error = None + plan.llm_planner_narrative = None + return + + decision_log.llm_provider = settings.provider + decision_log.llm_model = settings.model + client = _provider_client(settings) + if client is None: + decision_log.llm_used = False + decision_log.llm_error = f"unsupported provider: {settings.provider}" + return + + prompt = _build_prompt( + state=state, event=event, plan=plan, decision_log=decision_log + ) + try: + logger.info( + "llm capability=enrichment call start provider=%s model=%s", + settings.provider, + settings.model, + ) + response = client.generate_json( + prompt=prompt, + schema=ENRICHMENT_SCHEMA, + ) + except (GeminiClientError, VertexClientError, FPTClientError) as exc: + logger.warning( + "llm capability=enrichment call failed provider=%s model=%s error=%s", + settings.provider, + settings.model, + exc, + ) + decision_log.llm_used = False + decision_log.llm_error = str(exc) + plan.llm_planner_narrative = None + return + logger.info( + "llm capability=enrichment call succeeded provider=%s model=%s", + settings.provider, + settings.model, + ) + + planner_narrative = str(response.get("planner_narrative", "")).strip() + operator_explanation = str(response.get("operator_explanation", "")).strip() + approval_summary = str(response.get("approval_summary", "")).strip() + + plan.llm_planner_narrative = planner_narrative or None + decision_log.llm_operator_explanation = operator_explanation or None + decision_log.llm_approval_summary = approval_summary or None + decision_log.llm_used = any([planner_narrative, operator_explanation, approval_summary]) + decision_log.llm_error = None diff --git a/policies/strategic_prompt.py b/policies/strategic_prompt.py index 75ee03f..095e1f1 100644 --- a/policies/strategic_prompt.py +++ b/policies/strategic_prompt.py @@ -110,71 +110,33 @@ def _format_proposals_block(actions: list[Action]) -> str: return "\n\n".join(lines) -def build_strategic_prompt( +def build_memory_prompt( mode: str, - event: Event | None, historical_cases: list[HistoricalCase], - candidate_actions: list[Action], ) -> str: """ - Build the full Strategic Planner Agent reasoning prompt. + Build the Strategic Planner Agent reasoning context focusing on directives and memory. """ - event_block = _format_event_block(event) history_block = _format_historical_block(historical_cases) - proposals_block = _format_proposals_block(candidate_actions) mode_instruction = ( - "Prioritize SERVICE LEVEL and RECOVERY SPEED above all else." + "Prioritize SERVICE LEVEL and RECOVERY SPEED above all else. Aggressively mitigate risks." if mode == "crisis" - else "Prioritize COST EFFICIENCY and LEAN INVENTORY." + else "Prioritize COST EFFICIENCY and LEAN INVENTORY while maintaining baseline service." ) return f""" -======================================================================= -STRATEGIC SUPPLY CHAIN PLANNER — REASONING CONTEXT -======================================================================= - -OPERATING MODE: {mode.upper()} -DIRECTIVE : {mode_instruction} - ------------------------------------------------------------------------ -[BLOCK 1] CURRENT DISRUPTION ------------------------------------------------------------------------ -{event_block} +### STRATEGIC DIRECTIVES +- **Current Operating Mode**: {mode.upper()} +- **Core Objective**: {mode_instruction} ------------------------------------------------------------------------ -[BLOCK 2] HISTORICAL MEMORY (Retrieved by Similarity) ------------------------------------------------------------------------ +### HISTORICAL MEMORY (Lessons Learned) {history_block} ------------------------------------------------------------------------ -[BLOCK 3] SPECIALIST AGENT PROPOSALS ------------------------------------------------------------------------ -{proposals_block} - ------------------------------------------------------------------------ -[BLOCK 4] HARD CONSTRAINTS (MUST BE SATISFIED) ------------------------------------------------------------------------ - - Only use suppliers/routes that EXIST and are NOT blocked. - - Reorder quantities must NOT exceed warehouse capacity. - - Respect Minimum Order Quantity (MOQ) where specified. - ------------------------------------------------------------------------ -TASK INSTRUCTIONS ------------------------------------------------------------------------ -1. ANALYZE PATTERNS : Compare current disruption with historical cases. - Identify which past strategies succeeded and which failed. - -2. EVALUATE PROPOSALS: Filter proposals. Cross-reference with lessons learned. - -3. CONSTRUCT PLAN : Select actions that minimize disruption_risk and - maintain service_level within constraints. - -4. JUSTIFY : Explicitly cite Case IDs from memory that influenced - your decision. Example: "Chose REORDER because in Case_001, a similar - action prevented a stockout." - -======================================================================= +### REASONING GUIDELINES +1. ANALYZE PATTERNS: Compare current operational data with historical cases above. +2. LEVERAGE SUCCESS: Prioritize actions that led to positive outcome KPIs in similar past scenarios. +3. JUSTIFY DECISIONS: Explicitly cite Case IDs that influenced your choices. For example: "Selected REORDER because in Case_001, a similar action prevented a stockout." """.strip()