456 lines
20 KiB
Python
456 lines
20 KiB
Python
"""WorkingState revision for the PMFlow AI turn runtime.
|
||
|
||
This module does not classify raw language and does not render answers. It only
|
||
turns structured observations and verification feedback into state patches so
|
||
the runtime loop can keep probing, reading, revising, and verifying.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from ..schemas.pmflow_ai_turn_runtime import (
|
||
PMFlowAIDisclosureState,
|
||
PMFlowAIEvidenceGap,
|
||
PMFlowAIGoalFrame,
|
||
PMFlowAIOperationGap,
|
||
PMFlowAIObservation,
|
||
PMFlowAIObservationItem,
|
||
PMFlowAIStateRevision,
|
||
PMFlowAITargetCollection,
|
||
PMFlowAITargetScope,
|
||
PMFlowAITurnObjectRef,
|
||
PMFlowAITurnState,
|
||
PMFlowAIVerificationFailure,
|
||
)
|
||
from .pmflow_ai_turn_contract import normalize_turn_hypothesis_contract
|
||
|
||
|
||
class PMFlowAITurnStateReviser:
|
||
"""Revise the WorkingState from observations and verification feedback."""
|
||
|
||
def initialize_working_state(self, *, turn_state: PMFlowAITurnState, hypothesis: dict[str, Any]) -> PMFlowAITurnState:
|
||
hypothesis = normalize_turn_hypothesis_contract(hypothesis)
|
||
frames = _frames_from_hypothesis(turn_state=turn_state, hypothesis=hypothesis)
|
||
if frames and not turn_state.candidate_frames:
|
||
turn_state.candidate_frames = frames
|
||
if turn_state.candidate_frames and turn_state.selected_frame is None:
|
||
turn_state.selected_frame = turn_state.candidate_frames[0]
|
||
turn_state.disclosure_state = _disclosure_state_from_hypothesis(hypothesis)
|
||
turn_state.operation_gaps = _operation_gaps_from_hypothesis(hypothesis) or _operation_gaps_from_evidence_gaps(turn_state.evidence_state.evidence_gaps)
|
||
return turn_state
|
||
|
||
def observe_payload(
|
||
self,
|
||
*,
|
||
turn_state: PMFlowAITurnState,
|
||
action_type: str,
|
||
payload: dict[str, Any] | None,
|
||
) -> PMFlowAIObservation | None:
|
||
observation = _observation_from_payload(turn_state=turn_state, action_type=action_type, payload=payload)
|
||
if observation is not None:
|
||
turn_state.observations.append(observation)
|
||
return observation
|
||
|
||
def revise_from_observation(
|
||
self,
|
||
*,
|
||
turn_state: PMFlowAITurnState,
|
||
observation: PMFlowAIObservation | None,
|
||
) -> PMFlowAITurnState:
|
||
if observation is None:
|
||
return turn_state
|
||
patch: dict[str, Any] = {}
|
||
selected = turn_state.selected_frame
|
||
if selected is None:
|
||
selected = _select_frame_for_observation(turn_state=turn_state, observation=observation)
|
||
if selected is not None:
|
||
turn_state.selected_frame = selected
|
||
patch["selected_frame_id"] = selected.frame_id
|
||
reframed = _reframe_from_collection_observation(turn_state=turn_state, observation=observation)
|
||
if reframed is not None:
|
||
selected = reframed
|
||
patch["selected_frame_id"] = selected.frame_id
|
||
patch["reframed_from_observation"] = True
|
||
if selected and selected.operation_kind in observation.can_answer_operations:
|
||
turn_state.sufficiency_state.can_answer_now = True
|
||
turn_state.sufficiency_state.needs_user_input = False
|
||
patch["can_answer_now"] = True
|
||
patch["must_ask_user"] = False
|
||
if observation.remaining_gaps:
|
||
turn_state.operation_gaps = _merge_operation_gaps(turn_state.operation_gaps, observation.remaining_gaps)
|
||
if observation.observation_type == "collection_snapshot" and selected and selected.operation_kind == "list_collection":
|
||
turn_state.customer_packet_draft.update(
|
||
{
|
||
"operation_kind": "list_collection",
|
||
"object_type": observation.object_type,
|
||
"scope_type": observation.scope.scope_type,
|
||
"items": [item.model_dump(mode="json", exclude_none=True) for item in observation.items[:20]],
|
||
"count": observation.count,
|
||
}
|
||
)
|
||
turn_state.sufficiency_state.needs_user_input = False
|
||
patch["customer_packet_draft"] = {"operation_kind": "list_collection", "object_type": observation.object_type}
|
||
if patch:
|
||
turn_state.state_revisions.append(
|
||
PMFlowAIStateRevision(
|
||
revision_id=f"rev_{len(turn_state.state_revisions) + 1}",
|
||
source_observation_id=observation.observation_id,
|
||
selected_frame_id=(turn_state.selected_frame.frame_id if turn_state.selected_frame else ""),
|
||
reason="Observation revised WorkingState.",
|
||
state_patch=patch,
|
||
)
|
||
)
|
||
return turn_state
|
||
|
||
def revise_from_verification(
|
||
self,
|
||
*,
|
||
turn_state: PMFlowAITurnState,
|
||
failure: PMFlowAIVerificationFailure | None,
|
||
) -> PMFlowAITurnState:
|
||
if failure is None:
|
||
return turn_state
|
||
turn_state.verification_failures.append(failure)
|
||
patch = dict(failure.state_patch or {})
|
||
if patch.get("must_ask_user") is False:
|
||
turn_state.sufficiency_state.needs_user_input = False
|
||
if patch.get("can_answer_now") is True:
|
||
turn_state.sufficiency_state.can_answer_now = True
|
||
if patch.get("can_answer_now") is False:
|
||
turn_state.sufficiency_state.can_answer_now = False
|
||
turn_state.completed = False
|
||
if patch.get("needs_internal_evidence") is True:
|
||
turn_state.sufficiency_state.needs_internal_evidence = True
|
||
if patch.get("needs_external_evidence") is True:
|
||
turn_state.sufficiency_state.needs_external_evidence = True
|
||
if isinstance(patch.get("disclosure_state"), dict):
|
||
turn_state.disclosure_state = _disclosure_state_from_hypothesis({"disclosure_policy": patch.get("disclosure_state")})
|
||
next_action = str(failure.next_action_hint or "").strip()
|
||
customer_reason = str(failure.customer_safe_summary or "").strip()
|
||
if not customer_reason:
|
||
customer_reason = "这轮内部校验没有形成可安全展示的结论，需要先补齐可验证依据或改用安全边界回答。"
|
||
if (
|
||
next_action
|
||
and bool(failure.retry_would_change_state)
|
||
and next_action not in {"render_or_fail_closed", "select_minimal_read_or_render_action"}
|
||
):
|
||
turn_state.operation_gaps = _merge_operation_gaps(
|
||
turn_state.operation_gaps,
|
||
[
|
||
PMFlowAIOperationGap(
|
||
key=f"verification_{next_action}",
|
||
gap_type="system_evidence_gap",
|
||
blocking=True,
|
||
user_actionable=False,
|
||
system_action_hint=next_action,
|
||
reason=customer_reason,
|
||
)
|
||
],
|
||
)
|
||
turn_state.state_revisions.append(
|
||
PMFlowAIStateRevision(
|
||
revision_id=f"rev_{len(turn_state.state_revisions) + 1}",
|
||
reason=customer_reason,
|
||
selected_frame_id=(turn_state.selected_frame.frame_id if turn_state.selected_frame else ""),
|
||
state_patch=patch,
|
||
)
|
||
)
|
||
return turn_state
|
||
|
||
|
||
def _frames_from_hypothesis(*, turn_state: PMFlowAITurnState, hypothesis: dict[str, Any]) -> list[PMFlowAIGoalFrame]:
|
||
direct_frames = _frames_from_goal_frame_payload(turn_state=turn_state, hypothesis=hypothesis)
|
||
if direct_frames:
|
||
return direct_frames
|
||
return [
|
||
PMFlowAIGoalFrame(
|
||
frame_id="fallback_general_frame",
|
||
operation_kind="answer_general",
|
||
target_scope=PMFlowAITargetScope(scope_type="current_space", binding_strength="surface", source="contract_fallback"),
|
||
target_object_requirement="none",
|
||
answer_shape=turn_state.answer_state.answer_shape,
|
||
source_policy="internal",
|
||
confidence=0.2,
|
||
assumptions=["缺少直接 GoalFrame，按通用回答合同降级。"],
|
||
success_criteria=list(turn_state.turn_goal.success_criteria or []),
|
||
)
|
||
]
|
||
|
||
|
||
def _disclosure_state_from_hypothesis(hypothesis: dict[str, Any]) -> PMFlowAIDisclosureState:
|
||
raw = hypothesis.get("disclosure_policy") if isinstance(hypothesis.get("disclosure_policy"), dict) else {}
|
||
try:
|
||
return PMFlowAIDisclosureState.model_validate(raw)
|
||
except Exception:
|
||
return PMFlowAIDisclosureState()
|
||
|
||
|
||
def _frames_from_goal_frame_payload(*, turn_state: PMFlowAITurnState, hypothesis: dict[str, Any]) -> list[PMFlowAIGoalFrame]:
|
||
raw_frames = hypothesis.get("goal_frames")
|
||
rows = raw_frames if isinstance(raw_frames, list) else []
|
||
frames: list[PMFlowAIGoalFrame] = []
|
||
for index, raw in enumerate(rows[:6], start=1):
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
scope = _target_scope_from_payload(raw.get("target_scope"))
|
||
collection = None
|
||
raw_collection = raw.get("target_collection")
|
||
if isinstance(raw_collection, dict):
|
||
collection = PMFlowAITargetCollection(
|
||
object_type=str(raw_collection.get("object_type") or "")[:80],
|
||
required=bool(raw_collection.get("required")),
|
||
scope=_target_scope_from_payload(raw_collection.get("scope")) if isinstance(raw_collection.get("scope"), dict) else scope,
|
||
)
|
||
candidate_object = None
|
||
raw_candidate = raw.get("candidate_object")
|
||
if isinstance(raw_candidate, dict):
|
||
try:
|
||
candidate_object = PMFlowAITurnObjectRef.model_validate(raw_candidate)
|
||
except Exception:
|
||
candidate_object = turn_state.object_state.primary_object
|
||
try:
|
||
frames.append(
|
||
PMFlowAIGoalFrame(
|
||
frame_id=str(raw.get("frame_id") or f"goal_frame_{index}")[:80],
|
||
operation_kind=str(raw.get("operation_kind") or "answer_general"),
|
||
target_scope=scope,
|
||
target_collection=collection,
|
||
target_object_requirement=str(raw.get("target_object_requirement") or "none"),
|
||
candidate_object=candidate_object,
|
||
source_policy=str(raw.get("source_policy") or "none"),
|
||
answer_shape=str(raw.get("answer_shape") or turn_state.answer_state.answer_shape),
|
||
confidence=float(raw.get("confidence") or 0.0),
|
||
assumptions=list(raw.get("assumptions") or []),
|
||
success_criteria=list(raw.get("success_criteria") or turn_state.turn_goal.success_criteria or []),
|
||
)
|
||
)
|
||
except Exception:
|
||
continue
|
||
return frames
|
||
|
||
|
||
def _reframe_from_collection_observation(
|
||
*,
|
||
turn_state: PMFlowAITurnState,
|
||
observation: PMFlowAIObservation,
|
||
) -> PMFlowAIGoalFrame | None:
|
||
selected = turn_state.selected_frame
|
||
if observation.observation_type != "collection_snapshot":
|
||
return None
|
||
if not observation.items and observation.count in {None, 0}:
|
||
return None
|
||
if "list_collection" not in observation.can_answer_operations:
|
||
return None
|
||
if selected is not None and selected.operation_kind == "list_collection":
|
||
return None
|
||
if str(turn_state.answer_state.answer_shape or "") != "fact_list":
|
||
return None
|
||
scope = observation.scope or PMFlowAITargetScope(scope_type="current_space", binding_strength="surface", source="observation_reframe")
|
||
frame = PMFlowAIGoalFrame(
|
||
frame_id=f"observation_collection_{len(turn_state.candidate_frames) + 1}",
|
||
operation_kind="list_collection",
|
||
target_scope=scope,
|
||
target_collection=PMFlowAITargetCollection(
|
||
object_type=str(observation.object_type or "object")[:80],
|
||
required=True,
|
||
scope=scope,
|
||
),
|
||
target_object_requirement="none",
|
||
answer_shape="fact_list",
|
||
source_policy="internal",
|
||
confidence=0.78,
|
||
assumptions=["根据已完成的集合观察修正本轮目标框架。"],
|
||
success_criteria=list(turn_state.turn_goal.success_criteria or []),
|
||
)
|
||
turn_state.candidate_frames.insert(0, frame)
|
||
turn_state.selected_frame = frame
|
||
turn_state.answer_state.answer_shape = "fact_list"
|
||
turn_state.answer_state.required_sections = ["匹配结果", "依据", "下一步"]
|
||
return frame
|
||
|
||
|
||
def _target_scope_from_payload(value: Any) -> PMFlowAITargetScope:
|
||
raw = value if isinstance(value, dict) else {}
|
||
try:
|
||
return PMFlowAITargetScope(
|
||
scope_type=str(raw.get("scope_type") or "unspecified"),
|
||
binding_strength=str(raw.get("binding_strength") or "none"),
|
||
source=str(raw.get("source") or "goal_frame"),
|
||
)
|
||
except Exception:
|
||
return PMFlowAITargetScope()
|
||
|
||
|
||
def _operation_gaps_from_hypothesis(hypothesis: dict[str, Any]) -> list[PMFlowAIOperationGap]:
|
||
raw_gaps = hypothesis.get("operation_gaps")
|
||
rows = raw_gaps if isinstance(raw_gaps, list) else []
|
||
gaps: list[PMFlowAIOperationGap] = []
|
||
for index, raw in enumerate(rows[:20], start=1):
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
try:
|
||
gaps.append(
|
||
PMFlowAIOperationGap(
|
||
key=str(raw.get("key") or f"operation_gap_{index}")[:120],
|
||
gap_type=str(raw.get("gap_type") or "system_evidence_gap"),
|
||
blocking=bool(raw.get("blocking")),
|
||
user_actionable=bool(raw.get("user_actionable")),
|
||
customer_question=str(raw.get("customer_question") or "")[:240],
|
||
system_action_hint=str(raw.get("system_action_hint") or "")[:160],
|
||
reason=str(raw.get("reason") or raw.get("label") or "")[:400],
|
||
)
|
||
)
|
||
except Exception:
|
||
continue
|
||
return gaps
|
||
|
||
|
||
def _operation_gaps_from_evidence_gaps(gaps: list[PMFlowAIEvidenceGap]) -> list[PMFlowAIOperationGap]:
|
||
out: list[PMFlowAIOperationGap] = []
|
||
for gap in list(gaps or [])[:20]:
|
||
out.append(
|
||
PMFlowAIOperationGap(
|
||
key=gap.key,
|
||
gap_type=gap.gap_type,
|
||
blocking=gap.severity == "blocking",
|
||
user_actionable=gap.user_actionable,
|
||
customer_question=gap.customer_question,
|
||
system_action_hint=gap.system_action_hint,
|
||
reason=gap.reason or gap.label,
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _observation_from_payload(
|
||
*,
|
||
turn_state: PMFlowAITurnState,
|
||
action_type: str,
|
||
payload: dict[str, Any] | None,
|
||
) -> PMFlowAIObservation | None:
|
||
if not isinstance(payload, dict):
|
||
return PMFlowAIObservation(
|
||
observation_id=f"obs_{len(turn_state.observations) + 1}",
|
||
observation_type="verification_feedback",
|
||
action_phase="verify",
|
||
confidence="unknown",
|
||
remaining_gaps=[
|
||
PMFlowAIOperationGap(
|
||
key=f"{action_type}_no_payload",
|
||
gap_type="system_evidence_gap",
|
||
blocking=False,
|
||
system_action_hint="select_next_action",
|
||
reason="Action returned no payload.",
|
||
)
|
||
],
|
||
)
|
||
raw_observation = payload.get("runtime_observation")
|
||
if isinstance(raw_observation, dict):
|
||
try:
|
||
observation = PMFlowAIObservation.model_validate(raw_observation)
|
||
if not observation.observation_id:
|
||
observation.observation_id = f"obs_{len(turn_state.observations) + 1}"
|
||
return observation
|
||
except Exception:
|
||
pass
|
||
tool_result = payload.get("runtime_evidence_tool_result") if isinstance(payload.get("runtime_evidence_tool_result"), dict) else {}
|
||
findings = [item for item in list(tool_result.get("findings") or []) if isinstance(item, dict)]
|
||
items = [
|
||
PMFlowAIObservationItem(
|
||
object_type=_object_type_from_action(action_type),
|
||
title=str(item.get("fact") or "")[:200],
|
||
summary=str(item.get("implication") or item.get("fact") or "")[:400],
|
||
)
|
||
for item in findings[:20]
|
||
]
|
||
can_answer = _can_answer_operations_from_action(action_type=action_type, turn_state=turn_state, payload=payload)
|
||
confidence = str(tool_result.get("confidence") or ("high" if payload.get("_runtime_payload_kind") == "final_answer" else "unknown"))
|
||
return PMFlowAIObservation(
|
||
observation_id=f"obs_{len(turn_state.observations) + 1}",
|
||
observation_type=_observation_type_from_action(action_type=action_type, payload=payload),
|
||
action_phase="read" if action_type != "synthesize_answer" else "draft",
|
||
scope=(turn_state.selected_frame.target_scope if turn_state.selected_frame else PMFlowAITargetScope()),
|
||
object_type=_object_type_from_action(action_type),
|
||
items=items,
|
||
count=len(items) if items else None,
|
||
can_answer_operations=can_answer,
|
||
cannot_answer_operations=[],
|
||
confidence=confidence,
|
||
remaining_gaps=_operation_gaps_from_raw(tool_result.get("remaining_gaps")),
|
||
summary=str(tool_result.get("summary") or payload.get("answer") or payload.get("reply_text") or "")[:500],
|
||
)
|
||
|
||
|
||
def _observation_type_from_action(*, action_type: str, payload: dict[str, Any]) -> str:
|
||
if payload.get("_runtime_payload_kind") == "final_answer":
|
||
return "verification_feedback"
|
||
if action_type == "inspect_surface_context":
|
||
return "collection_snapshot"
|
||
if action_type in {"resolve_project_objects", "inspect_project_status", "inspect_internal_projects"}:
|
||
return "object_snapshot"
|
||
if action_type == "inspect_saved_ideas":
|
||
return "collection_snapshot"
|
||
if action_type in {"external_search", "external_fetch"}:
|
||
return "external_evidence"
|
||
return "scope_probe"
|
||
|
||
|
||
def _object_type_from_action(action_type: str) -> str:
|
||
if action_type == "inspect_saved_ideas":
|
||
return "saved_idea"
|
||
if action_type in {"inspect_surface_context", "resolve_project_objects", "inspect_project_status", "inspect_internal_projects"}:
|
||
return "project"
|
||
if action_type in {"external_search", "external_fetch"}:
|
||
return "external_source"
|
||
return "object"
|
||
|
||
|
||
def _can_answer_operations_from_action(*, action_type: str, turn_state: PMFlowAITurnState, payload: dict[str, Any]) -> list[str]:
|
||
selected = turn_state.selected_frame.operation_kind if turn_state.selected_frame else "answer_general"
|
||
if payload.get("_runtime_payload_kind") == "final_answer":
|
||
return [selected]
|
||
if action_type in {"inspect_surface_context", "inspect_saved_ideas", "resolve_project_objects"}:
|
||
return ["list_collection"] if selected == "list_collection" else []
|
||
if action_type in {"inspect_project_status", "inspect_internal_projects"}:
|
||
return ["inspect_object"] if selected == "inspect_object" else []
|
||
if action_type in {"external_search", "external_fetch"}:
|
||
return ["judge_direction", "compare_objects", "answer_general"]
|
||
return []
|
||
|
||
|
||
def _operation_gaps_from_raw(values: Any) -> list[PMFlowAIOperationGap]:
|
||
rows = values if isinstance(values, list) else []
|
||
out: list[PMFlowAIOperationGap] = []
|
||
for index, value in enumerate(rows[:10], start=1):
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
continue
|
||
out.append(
|
||
PMFlowAIOperationGap(
|
||
key=f"tool_gap_{index}",
|
||
gap_type="system_evidence_gap",
|
||
blocking=False,
|
||
system_action_hint="select_next_action",
|
||
reason=text,
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _select_frame_for_observation(*, turn_state: PMFlowAITurnState, observation: PMFlowAIObservation) -> PMFlowAIGoalFrame | None:
|
||
for frame in list(turn_state.candidate_frames or []):
|
||
if frame.operation_kind in observation.can_answer_operations:
|
||
return frame
|
||
return turn_state.candidate_frames[0] if turn_state.candidate_frames else None
|
||
|
||
|
||
def _merge_operation_gaps(existing: list[PMFlowAIOperationGap], incoming: list[PMFlowAIOperationGap]) -> list[PMFlowAIOperationGap]:
|
||
by_key = {gap.key: gap for gap in list(existing or [])}
|
||
for gap in incoming:
|
||
by_key[gap.key] = gap
|
||
return list(by_key.values())[:20]
|
||
|
||
|
||
__all__ = ["PMFlowAITurnStateReviser"]
|