diff --git a/agents/planner.py b/agents/planner.py index 15a0857..372acc1 100644 --- a/agents/planner.py +++ b/agents/planner.py @@ -34,7 +34,7 @@ retrieve_relevant_cases, compute_memory_influence, derive_strategy_rationale, - build_strategic_prompt, + build_memory_prompt, ) from simulation.evaluator import evaluate_candidate_plan @@ -910,7 +910,6 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: feasible_candidates.append(act) else: infeasible_reasons[act.action_id] = vios - if not feasible_candidates and candidate_actions: feasible_candidates = [ Action( @@ -923,23 +922,17 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: ] 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 - ) + 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, + 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", @@ -1149,7 +1142,7 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: referenced_cases=historical_cases, memory_influence_score=memory_influence, strategy_rationale=strategy_rationale, - strategic_prompt=strategic_prompt, + strategic_prompt=memory_prompt ) winning_factors = build_winning_factors( @@ -1198,7 +1191,6 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: 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 adcc2ac..ea4c023 100644 --- a/app_api/services.py +++ b/app_api/services.py @@ -592,28 +592,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) @@ -1712,3 +1713,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()