545 lines
22 KiB
Python
545 lines
22 KiB
Python
"""Operating cycle triggers, outcome observation and learning ledger."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from typing import Any
|
||
|
||
from ...schemas.pmflow_ai_operating_runtime import (
|
||
PMFlowAIAgentCouncilReview,
|
||
PMFlowAIAutonomyDecision,
|
||
PMFlowAILearningLedger,
|
||
PMFlowAIOperatingCycleState,
|
||
PMFlowAIOperatingCycleTrigger,
|
||
PMFlowAIOperatingMission,
|
||
PMFlowAIOperatingPlan,
|
||
PMFlowAIOutcomeObservation,
|
||
PMFlowAIProjectWorldStatePacket,
|
||
clean_text,
|
||
)
|
||
from ...schemas.rag import RAGAskRequest
|
||
|
||
|
||
SUPPORTED_OPERATING_TRIGGERS = [
|
||
{
|
||
"trigger_type": "user_turn",
|
||
"display_label": "用户对话",
|
||
"contract": "run_immediately_from_customer_turn",
|
||
},
|
||
{
|
||
"trigger_type": "signal",
|
||
"display_label": "项目/市场信号",
|
||
"contract": "run_when_new_signal_changes_project_context",
|
||
},
|
||
{
|
||
"trigger_type": "project_creation",
|
||
"display_label": "项目创建",
|
||
"contract": "run_after_project_truth_state_created",
|
||
},
|
||
{
|
||
"trigger_type": "task_update",
|
||
"display_label": "任务更新",
|
||
"contract": "run_after_task_or_changeset_state_changes",
|
||
},
|
||
{
|
||
"trigger_type": "execution_result",
|
||
"display_label": "执行结果",
|
||
"contract": "run_after_execution_bridge_reports_result",
|
||
},
|
||
{
|
||
"trigger_type": "daily_review",
|
||
"display_label": "每日复盘",
|
||
"contract": "scheduled_operating_review",
|
||
},
|
||
{
|
||
"trigger_type": "weekly_strategy_review",
|
||
"display_label": "每周策略复盘",
|
||
"contract": "scheduled_strategy_review",
|
||
},
|
||
]
|
||
|
||
|
||
class PMFlowAIOperatingCycleScheduler:
|
||
"""Build trigger contracts for operating runs."""
|
||
|
||
def build_trigger(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
mission: PMFlowAIOperatingMission,
|
||
trigger_type: str = "user_turn",
|
||
source_payload: dict[str, Any] | None = None,
|
||
) -> PMFlowAIOperatingCycleTrigger:
|
||
normalized_type = clean_text(trigger_type, limit=80)
|
||
supported = {item["trigger_type"] for item in SUPPORTED_OPERATING_TRIGGERS}
|
||
if normalized_type not in supported:
|
||
normalized_type = "user_turn"
|
||
project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else ""
|
||
source_ref = f"qa_session:{request.session_id}" if request.session_id is not None else "space_ai_turn"
|
||
return PMFlowAIOperatingCycleTrigger(
|
||
trigger_type=normalized_type,
|
||
source_ref=source_ref,
|
||
project_ref=project_ref,
|
||
payload={
|
||
"question_ref": _stable_question_ref(request.question),
|
||
"asked_via": clean_text(request.asked_via or "pmflow", limit=40),
|
||
"source_payload": dict(source_payload or {}),
|
||
},
|
||
)
|
||
|
||
def supported_triggers(self) -> list[dict[str, Any]]:
|
||
return [dict(item) for item in SUPPORTED_OPERATING_TRIGGERS]
|
||
|
||
def next_cycle_triggers(
|
||
self,
|
||
*,
|
||
mission: PMFlowAIOperatingMission,
|
||
autonomy_gate: PMFlowAIAutonomyDecision,
|
||
learning_ledger: PMFlowAILearningLedger,
|
||
) -> list[dict[str, Any]]:
|
||
project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else ""
|
||
triggers = [
|
||
{
|
||
"trigger_type": "user_turn",
|
||
"reason": "用户继续追问、确认或补充后重新运行 operating cycle。",
|
||
"project_ref": project_ref,
|
||
},
|
||
{
|
||
"trigger_type": "execution_result",
|
||
"reason": "执行交接或任务结果返回后更新 learning ledger 并重排计划。",
|
||
"project_ref": project_ref,
|
||
},
|
||
{
|
||
"trigger_type": "daily_review",
|
||
"reason": "每日读取项目运行态和学习账本，检查阻塞、确认项和下一步。",
|
||
"project_ref": project_ref,
|
||
},
|
||
{
|
||
"trigger_type": "weekly_strategy_review",
|
||
"reason": "每周基于结果、偏好和实验信号重新判断方向。",
|
||
"project_ref": project_ref,
|
||
},
|
||
]
|
||
if autonomy_gate.confirmation_requests:
|
||
triggers.append(
|
||
{
|
||
"trigger_type": "task_update",
|
||
"reason": "确认请求被接受/拒绝后更新决策结果。",
|
||
"project_ref": project_ref,
|
||
}
|
||
)
|
||
if learning_ledger.experiment_results:
|
||
triggers.append(
|
||
{
|
||
"trigger_type": "signal",
|
||
"reason": "实验结果或市场信号进入后刷新客户验证判断。",
|
||
"project_ref": project_ref,
|
||
}
|
||
)
|
||
return triggers[:8]
|
||
|
||
|
||
class OutcomeObserver:
|
||
"""Observe decisions, execution results, experiments and preferences."""
|
||
|
||
def observe(
|
||
self,
|
||
*,
|
||
mission: PMFlowAIOperatingMission,
|
||
project_world_state: PMFlowAIProjectWorldStatePacket,
|
||
council_review: PMFlowAIAgentCouncilReview,
|
||
autonomy_gate: PMFlowAIAutonomyDecision,
|
||
operating_plan: PMFlowAIOperatingPlan,
|
||
external_outcomes: list[dict[str, Any]] | None = None,
|
||
) -> list[PMFlowAIOutcomeObservation]:
|
||
observations: list[PMFlowAIOutcomeObservation] = []
|
||
observations.extend(
|
||
DecisionOutcomeService().observe_pending_confirmations(
|
||
mission=mission,
|
||
autonomy_gate=autonomy_gate,
|
||
)
|
||
)
|
||
observations.extend(
|
||
ExperimentResultService().observe_pending_experiments(
|
||
autonomy_gate=autonomy_gate,
|
||
)
|
||
)
|
||
observations.extend(
|
||
ExecutionOutcomeService().observe_execution_preparations(
|
||
autonomy_gate=autonomy_gate,
|
||
)
|
||
)
|
||
observations.extend(
|
||
UserPreferenceLearningService().observe_preferences(
|
||
mission=mission,
|
||
operating_plan=operating_plan,
|
||
council_review=council_review,
|
||
)
|
||
)
|
||
observations.extend(self.from_external_outcomes(external_outcomes or [], project_world_state=project_world_state))
|
||
return observations[:24]
|
||
|
||
def from_external_outcomes(
|
||
self,
|
||
rows: list[dict[str, Any]],
|
||
*,
|
||
project_world_state: PMFlowAIProjectWorldStatePacket,
|
||
) -> list[PMFlowAIOutcomeObservation]:
|
||
observations: list[PMFlowAIOutcomeObservation] = []
|
||
for item in rows[:24]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
outcome_type = clean_text(item.get("outcome_type"), limit=80)
|
||
if outcome_type not in {"recommendation_decision", "execution_result", "experiment_result", "user_preference"}:
|
||
continue
|
||
observations.append(
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type=outcome_type,
|
||
outcome_status=_normalize_outcome_status(item.get("outcome_status") or item.get("status")),
|
||
source_ref=clean_text(item.get("source_ref"), limit=180),
|
||
target_ref=clean_text(item.get("target_ref") or project_world_state.primary_project_ref, limit=160),
|
||
summary=clean_text(item.get("summary") or item.get("title"), limit=800),
|
||
evidence_refs=[clean_text(ref, limit=220) for ref in list(item.get("evidence_refs") or [])[:16]],
|
||
learning_signals=dict(item.get("learning_signals") or {}) if isinstance(item.get("learning_signals"), dict) else {},
|
||
)
|
||
)
|
||
return observations
|
||
|
||
|
||
class DecisionOutcomeService:
|
||
def observe_pending_confirmations(
|
||
self,
|
||
*,
|
||
mission: PMFlowAIOperatingMission,
|
||
autonomy_gate: PMFlowAIAutonomyDecision,
|
||
) -> list[PMFlowAIOutcomeObservation]:
|
||
observations: list[PMFlowAIOutcomeObservation] = []
|
||
project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else ""
|
||
for request in list(autonomy_gate.confirmation_requests or [])[:12]:
|
||
if not isinstance(request, dict):
|
||
continue
|
||
observations.append(
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type="recommendation_decision",
|
||
outcome_status="pending",
|
||
source_ref=clean_text(request.get("confirmation_request_id"), limit=120),
|
||
target_ref=clean_text(request.get("target_ref") or project_ref, limit=160),
|
||
summary=f"确认请求待用户决策：{clean_text(request.get('title'), limit=240)}",
|
||
learning_signals={
|
||
"decision_state": "pending_user_confirmation",
|
||
"requested_for": list(request.get("requested_for") or []),
|
||
},
|
||
)
|
||
)
|
||
return observations
|
||
|
||
|
||
class ExecutionOutcomeService:
|
||
def observe_execution_preparations(
|
||
self,
|
||
*,
|
||
autonomy_gate: PMFlowAIAutonomyDecision,
|
||
) -> list[PMFlowAIOutcomeObservation]:
|
||
observations: list[PMFlowAIOutcomeObservation] = []
|
||
for item in list(autonomy_gate.execution_preparations or [])[:8]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
observations.append(
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type="execution_result",
|
||
outcome_status="pending",
|
||
source_ref=clean_text(item.get("execution_preparation_id"), limit=120),
|
||
target_ref=clean_text(item.get("target_ref"), limit=160),
|
||
summary="执行交接包已准备，但尚未发起 handoff，等待用户确认。",
|
||
learning_signals={
|
||
"execution_state": "prepared_waiting_confirmation",
|
||
"readiness": clean_text(item.get("readiness"), limit=80),
|
||
},
|
||
)
|
||
)
|
||
return observations
|
||
|
||
|
||
class ExperimentResultService:
|
||
def observe_pending_experiments(
|
||
self,
|
||
*,
|
||
autonomy_gate: PMFlowAIAutonomyDecision,
|
||
) -> list[PMFlowAIOutcomeObservation]:
|
||
observations: list[PMFlowAIOutcomeObservation] = []
|
||
for item in list(autonomy_gate.prepared_action_packets or [])[:12]:
|
||
if not isinstance(item, dict) or item.get("action_family") != "experiment_plan":
|
||
continue
|
||
observations.append(
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type="experiment_result",
|
||
outcome_status="pending",
|
||
source_ref=clean_text(item.get("prepared_action_id"), limit=120),
|
||
target_ref=clean_text(item.get("target_ref"), limit=160),
|
||
summary=f"用户验证实验待确认：{clean_text(item.get('title'), limit=240)}",
|
||
evidence_refs=list(item.get("evidence_refs") or []),
|
||
learning_signals={
|
||
"experiment_state": "prepared_waiting_confirmation",
|
||
"action_family": item.get("action_family"),
|
||
},
|
||
)
|
||
)
|
||
return observations
|
||
|
||
|
||
class UserPreferenceLearningService:
|
||
def observe_preferences(
|
||
self,
|
||
*,
|
||
mission: PMFlowAIOperatingMission,
|
||
operating_plan: PMFlowAIOperatingPlan,
|
||
council_review: PMFlowAIAgentCouncilReview,
|
||
) -> list[PMFlowAIOutcomeObservation]:
|
||
project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else ""
|
||
observations = [
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type="user_preference",
|
||
outcome_status="observed",
|
||
source_ref=mission.mission_id,
|
||
target_ref=project_ref,
|
||
summary="用户偏好体现为：需要 PMFlow AI 主动推进项目，而不是只回答问题。",
|
||
learning_signals={
|
||
"preference_key": "operating_partner_not_qa",
|
||
"preferred_behavior": "prepare_plan_with_confirmation_gate",
|
||
"risk_threshold": "do_not_execute_without_explicit_confirmation",
|
||
},
|
||
)
|
||
]
|
||
if operating_plan.priority_order:
|
||
observations.append(
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type="user_preference",
|
||
outcome_status="observed",
|
||
source_ref=str(council_review.review_id),
|
||
target_ref=project_ref,
|
||
summary="本轮计划保留单一优先级，后续运行应继续消费 priority_order 而不是散列建议。",
|
||
learning_signals={
|
||
"preference_key": "single_priority_order",
|
||
"preferred_behavior": "compile_one_ranked_plan",
|
||
"source_effectiveness_policy": "future_runs_should_reuse_evidence_types_that_survive_council_review",
|
||
},
|
||
)
|
||
)
|
||
if operating_plan.product_object_projections:
|
||
observations.append(
|
||
PMFlowAIOutcomeObservation(
|
||
outcome_type="user_preference",
|
||
outcome_status="observed",
|
||
source_ref=mission.mission_id,
|
||
target_ref=project_ref,
|
||
summary="本轮已把回答升级为经营对象候选；后续应跟踪用户是否保存、确认、修改或拒绝该对象。",
|
||
learning_signals={
|
||
"preference_key": "product_object_first",
|
||
"preferred_behavior": "project_opportunity_or_research_outputs_before_freeform_answer",
|
||
},
|
||
)
|
||
)
|
||
return observations
|
||
|
||
|
||
class LearningLedgerService:
|
||
"""Load and update durable learning from session capsules."""
|
||
|
||
def ledger_from_context(
|
||
self,
|
||
*,
|
||
conversation_context: dict[str, Any] | None,
|
||
mission: PMFlowAIOperatingMission,
|
||
) -> PMFlowAILearningLedger:
|
||
capsules = list((conversation_context or {}).get("context_capsules") or [])
|
||
project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else ""
|
||
for item in reversed(capsules):
|
||
if not isinstance(item, dict) or item.get("capsule_type") != "operating_learning_ledger":
|
||
continue
|
||
ledger = item.get("learning_ledger") if isinstance(item.get("learning_ledger"), dict) else {}
|
||
if not ledger:
|
||
continue
|
||
if project_ref and ledger.get("scope_ref") not in {"", project_ref}:
|
||
continue
|
||
return PMFlowAILearningLedger.model_validate(ledger)
|
||
return PMFlowAILearningLedger(scope_ref=project_ref)
|
||
|
||
def update_ledger(
|
||
self,
|
||
*,
|
||
consumed_ledger: PMFlowAILearningLedger,
|
||
outcome_observations: list[PMFlowAIOutcomeObservation],
|
||
) -> PMFlowAILearningLedger:
|
||
decision_outcomes = list(consumed_ledger.decision_outcomes or [])
|
||
experiment_results = list(consumed_ledger.experiment_results or [])
|
||
execution_outcomes = list(consumed_ledger.execution_outcomes or [])
|
||
user_preferences = list(consumed_ledger.user_preferences or [])
|
||
learning_items = list(consumed_ledger.learning_items or [])
|
||
for observation in outcome_observations:
|
||
row = _observation_row(observation)
|
||
if observation.outcome_type == "recommendation_decision":
|
||
decision_outcomes.append(row)
|
||
elif observation.outcome_type == "experiment_result":
|
||
experiment_results.append(row)
|
||
elif observation.outcome_type == "execution_result":
|
||
execution_outcomes.append(row)
|
||
elif observation.outcome_type == "user_preference":
|
||
user_preferences.append(row)
|
||
learning_items.append(_learning_item_from_observation(observation))
|
||
return PMFlowAILearningLedger(
|
||
ledger_id=consumed_ledger.ledger_id,
|
||
scope_ref=consumed_ledger.scope_ref,
|
||
learning_items=_dedupe_rows(learning_items, key="learning_key")[-40:],
|
||
decision_outcomes=_dedupe_rows(decision_outcomes, key="source_ref")[-40:],
|
||
experiment_results=_dedupe_rows(experiment_results, key="source_ref")[-40:],
|
||
execution_outcomes=_dedupe_rows(execution_outcomes, key="source_ref")[-40:],
|
||
user_preferences=_dedupe_rows(user_preferences, key="learning_key")[-40:],
|
||
consumed_memory_refs=_dedupe_texts(
|
||
list(consumed_ledger.consumed_memory_refs or []) + [consumed_ledger.ledger_id]
|
||
)[-24:],
|
||
)
|
||
|
||
def capsule(self, *, learning_ledger: PMFlowAILearningLedger) -> dict[str, Any]:
|
||
return {
|
||
"capsule_version": "pmflow_ai_learning_ledger_capsule.v1",
|
||
"capsule_type": "operating_learning_ledger",
|
||
"scope_ref": learning_ledger.scope_ref,
|
||
"ledger_id": learning_ledger.ledger_id,
|
||
"updated_at": learning_ledger.updated_at,
|
||
"learning_ledger": learning_ledger.model_dump(mode="json", exclude_none=True),
|
||
}
|
||
|
||
|
||
class OperatingCycleService:
|
||
def __init__(self) -> None:
|
||
self.scheduler = PMFlowAIOperatingCycleScheduler()
|
||
self.observer = OutcomeObserver()
|
||
self.ledger_service = LearningLedgerService()
|
||
|
||
def build_cycle_state(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
mission: PMFlowAIOperatingMission,
|
||
project_world_state: PMFlowAIProjectWorldStatePacket,
|
||
council_review: PMFlowAIAgentCouncilReview,
|
||
autonomy_gate: PMFlowAIAutonomyDecision,
|
||
operating_plan: PMFlowAIOperatingPlan,
|
||
conversation_context: dict[str, Any] | None,
|
||
) -> PMFlowAIOperatingCycleState:
|
||
consumed_ledger = self.ledger_service.ledger_from_context(
|
||
conversation_context=conversation_context,
|
||
mission=mission,
|
||
)
|
||
trigger = self.scheduler.build_trigger(request=request, mission=mission)
|
||
external_outcomes = _external_outcomes_from_context(conversation_context)
|
||
observations = self.observer.observe(
|
||
mission=mission,
|
||
project_world_state=project_world_state,
|
||
council_review=council_review,
|
||
autonomy_gate=autonomy_gate,
|
||
operating_plan=operating_plan,
|
||
external_outcomes=external_outcomes,
|
||
)
|
||
learning_ledger = self.ledger_service.update_ledger(
|
||
consumed_ledger=consumed_ledger,
|
||
outcome_observations=observations,
|
||
)
|
||
return PMFlowAIOperatingCycleState(
|
||
trigger=trigger,
|
||
supported_triggers=self.scheduler.supported_triggers(),
|
||
consumed_learning_ledger=consumed_ledger,
|
||
outcome_observations=observations,
|
||
learning_ledger=learning_ledger,
|
||
next_cycle_triggers=self.scheduler.next_cycle_triggers(
|
||
mission=mission,
|
||
autonomy_gate=autonomy_gate,
|
||
learning_ledger=learning_ledger,
|
||
),
|
||
)
|
||
|
||
|
||
def _external_outcomes_from_context(conversation_context: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
context = conversation_context or {}
|
||
rows = context.get("operating_outcomes")
|
||
return [dict(item) for item in rows if isinstance(item, dict)] if isinstance(rows, list) else []
|
||
|
||
|
||
def _normalize_outcome_status(value: Any) -> str:
|
||
status = clean_text(value, limit=80)
|
||
if status in {"pending", "accepted", "rejected", "succeeded", "failed", "partial", "observed"}:
|
||
return status
|
||
return "observed"
|
||
|
||
|
||
def _observation_row(observation: PMFlowAIOutcomeObservation) -> dict[str, Any]:
|
||
return {
|
||
"observation_id": observation.observation_id,
|
||
"outcome_type": observation.outcome_type,
|
||
"outcome_status": observation.outcome_status,
|
||
"source_ref": observation.source_ref,
|
||
"target_ref": observation.target_ref,
|
||
"summary": observation.summary,
|
||
"learning_signals": dict(observation.learning_signals or {}),
|
||
"observed_at": observation.observed_at,
|
||
}
|
||
|
||
|
||
def _learning_item_from_observation(observation: PMFlowAIOutcomeObservation) -> dict[str, Any]:
|
||
signals = dict(observation.learning_signals or {})
|
||
learning_key = clean_text(signals.get("preference_key") or observation.source_ref or observation.outcome_type, limit=180)
|
||
return {
|
||
"learning_key": learning_key,
|
||
"outcome_type": observation.outcome_type,
|
||
"outcome_status": observation.outcome_status,
|
||
"summary": observation.summary,
|
||
"target_ref": observation.target_ref,
|
||
"updated_at": observation.observed_at,
|
||
}
|
||
|
||
|
||
def _stable_question_ref(question: Any) -> str:
|
||
text = clean_text(question, limit=80)
|
||
if not text:
|
||
return "question:empty"
|
||
digest = hashlib.sha1(text.encode("utf-8")).hexdigest()[:12]
|
||
return f"question:{digest}"
|
||
|
||
|
||
def _dedupe_rows(rows: list[dict[str, Any]], *, key: str) -> list[dict[str, Any]]:
|
||
seen: set[str] = set()
|
||
out: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
value = clean_text(row.get(key), limit=220)
|
||
if not value:
|
||
value = clean_text(row.get("summary"), limit=220)
|
||
marker = value.lower()
|
||
if not marker or marker in seen:
|
||
continue
|
||
seen.add(marker)
|
||
out.append(row)
|
||
return out
|
||
|
||
|
||
def _dedupe_texts(rows: list[str]) -> list[str]:
|
||
out: list[str] = []
|
||
for item in rows:
|
||
text = clean_text(item, limit=220)
|
||
if text and text not in out:
|
||
out.append(text)
|
||
return out
|
||
|
||
|
||
__all__ = [
|
||
"DecisionOutcomeService",
|
||
"ExecutionOutcomeService",
|
||
"ExperimentResultService",
|
||
"LearningLedgerService",
|
||
"OperatingCycleService",
|
||
"OutcomeObserver",
|
||
"PMFlowAIOperatingCycleScheduler",
|
||
"UserPreferenceLearningService",
|
||
]
|