1207 lines
54 KiB
Python
1207 lines
54 KiB
Python
"""Durable Agent OS runtime for user-facing PMFlow AI runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import math
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ...models import (
|
|
PMFlowAgentArtifact,
|
|
PMFlowAgentMemory,
|
|
PMFlowAgentObservation,
|
|
PMFlowAgentRun,
|
|
PMFlowAgentStep,
|
|
PMFlowAgentVerification,
|
|
User,
|
|
)
|
|
from ...schemas.pmflow_agent_runtime import (
|
|
PMFlowAgentArtifactPacket,
|
|
PMFlowAgentMemoryPacket,
|
|
PMFlowAgentObservationPacket,
|
|
PMFlowAgentRunPacket,
|
|
PMFlowAgentStepPacket,
|
|
PMFlowAgentVerificationPacket,
|
|
customer_safe_pmflow_agent_run_packet,
|
|
new_pmflow_agent_id,
|
|
utc_now_iso,
|
|
)
|
|
from ...schemas.rag import RAGAskRequest
|
|
from ..pmflow_ai_context_grounding_verifier import revise_answer_with_context_boundary, verify_answer_context_grounding
|
|
from ..pmflow_ai_context_pack_service import context_pack_observability_packet, context_pack_used_context_packet
|
|
from ..runtime_audit_context import bind_agent_audit_context
|
|
|
|
|
|
PMFLOW_AGENT_OS_POLICY_VERSION = "pmflow-agent-os-v2"
|
|
|
|
|
|
class PMFlowAgentRuntimeService:
|
|
"""Own the PMFlow AI Agent Run truth source.
|
|
|
|
QARecord is kept as transcript/history only. The current run state, steps,
|
|
observations, artifacts, verifications and memories are written to Agent OS
|
|
tables first, then projected into the customer payload.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
tenant_id: int,
|
|
delivery_runtime: Any | None = None,
|
|
environment_service: Any | None = None,
|
|
specialist_runtime: Any | None = None,
|
|
) -> None:
|
|
self.db = db
|
|
self.tenant_id = int(tenant_id or 1)
|
|
# The product entry now uses a dynamic observe-plan-act-critic-answer
|
|
# agent runtime. The argument name is kept for constructor compatibility;
|
|
# when present it must expose run_turn() and is treated as the dynamic
|
|
# agent delegate, not as a legacy answer override.
|
|
self.delivery_runtime = delivery_runtime
|
|
self.environment_service = environment_service
|
|
self._compat_specialist_runtime = specialist_runtime
|
|
|
|
async def run_space_turn(
|
|
self,
|
|
*,
|
|
request: RAGAskRequest,
|
|
current_user: User,
|
|
conversation_context: dict[str, Any] | None,
|
|
tenant_project_context: dict[str, Any] | None,
|
|
started_at: float | None = None,
|
|
event_sink: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
context_observability = context_pack_observability_packet((conversation_context or {}).get("context_pack"))
|
|
context_pack = (conversation_context or {}).get("context_pack") if isinstance(conversation_context, dict) else {}
|
|
context_evidence_packet = (
|
|
context_pack.get("context_evidence_packet") if isinstance(context_pack, dict) and isinstance(context_pack.get("context_evidence_packet"), dict) else {}
|
|
)
|
|
used_context_packet = context_pack_used_context_packet(context_pack)
|
|
run = PMFlowAgentRun(
|
|
tenant_id=self.tenant_id,
|
|
public_run_id=new_pmflow_agent_id("run"),
|
|
user_id=int(getattr(current_user, "id", 0) or 0) or None,
|
|
qa_session_id=_safe_int(getattr(request, "session_id", None)),
|
|
workspace_scope="space",
|
|
surface="space",
|
|
presentation_mode="pmflow_ai",
|
|
source_channel=str(getattr(request, "asked_via", "") or "pmflow")[:40],
|
|
source_ref=str(getattr(request, "source_chat_id", "") or "")[:160] or None,
|
|
title=_title_from_question(getattr(request, "question", "")),
|
|
goal=str(getattr(request, "question", "") or "").strip(),
|
|
status="active",
|
|
active_object_refs=[],
|
|
run_metadata={
|
|
"runtime": "pmflow_agent_os",
|
|
"runtime_contract": "agent_os_direct_answer_runtime",
|
|
"policy_version": PMFLOW_AGENT_OS_POLICY_VERSION,
|
|
"conversation_session_id": _safe_int((conversation_context or {}).get("session_id")),
|
|
"context_observability": context_observability,
|
|
"used_context": used_context_packet,
|
|
"context_evidence": context_evidence_packet,
|
|
},
|
|
started_at=datetime.utcnow(),
|
|
)
|
|
self.db.add(run)
|
|
await self.db.flush()
|
|
|
|
try:
|
|
run_kwargs = {
|
|
"request": request,
|
|
"current_user": current_user,
|
|
"conversation_context": conversation_context,
|
|
"tenant_project_context": tenant_project_context,
|
|
"started_at": started_at,
|
|
}
|
|
if event_sink is not None:
|
|
run_kwargs["event_sink"] = event_sink
|
|
with bind_agent_audit_context(agent_run_id=run.public_run_id):
|
|
agent_payload = await self._dynamic_agent_runtime().run_turn(**run_kwargs)
|
|
answer = str(agent_payload.get("answer") or agent_payload.get("reply_text") or "").strip()
|
|
grounding_packet = verify_answer_context_grounding(
|
|
answer=answer,
|
|
used_context_packet=used_context_packet,
|
|
)
|
|
revised_answer = revise_answer_with_context_boundary(answer=answer, grounding_packet=grounding_packet)
|
|
if revised_answer != answer:
|
|
answer = revised_answer
|
|
agent_payload = {
|
|
**dict(agent_payload or {}),
|
|
"answer": answer,
|
|
"reply_text": answer,
|
|
"grounding_status": "context_revised",
|
|
}
|
|
agent_run_packet = _dict(agent_payload.get("agent_run_packet"))
|
|
run.active_object_refs = _agent_payload_object_refs(agent_payload)
|
|
run.status = _durable_status_from_agent_payload(agent_payload, answer=answer)
|
|
run.final_answer = answer.strip() or None
|
|
run.completed_at = datetime.utcnow()
|
|
run.run_metadata = {
|
|
**dict(run.run_metadata or {}),
|
|
"runtime": "pmflow_agent_os",
|
|
"runtime_contract": "agent_os_direct_answer_runtime",
|
|
"delegated_runtime": str(agent_payload.get("answer_mode") or agent_payload.get("processing_stage") or "pmflow_ai_agent_runtime"),
|
|
"delegated_run_id": str(agent_run_packet.get("run_id") or ""),
|
|
"step_count": len(list(agent_run_packet.get("steps") or [])),
|
|
"observation_count": len(list(agent_run_packet.get("observations") or [])),
|
|
"artifact_count": 2 if answer else 1,
|
|
"verification_count": len(_agent_verifications_from_payload(agent_payload, answer=answer)),
|
|
"stop_reason": str(agent_run_packet.get("stop_reason") or ""),
|
|
"iterations": _safe_int(agent_run_packet.get("iterations")),
|
|
"tool_call_count": _safe_int(agent_run_packet.get("tool_call_count")),
|
|
"context_observability": context_observability,
|
|
"used_context": used_context_packet,
|
|
"context_evidence": context_evidence_packet,
|
|
"context_grounding": grounding_packet,
|
|
}
|
|
packet = await self._persist_agent_runtime_payload(
|
|
run=run,
|
|
agent_payload=agent_payload,
|
|
answer=answer,
|
|
)
|
|
payload = self._build_customer_payload_from_agent_runtime(
|
|
request=request,
|
|
run=run,
|
|
packet=packet,
|
|
agent_payload=agent_payload,
|
|
answer=answer,
|
|
)
|
|
if context_observability:
|
|
payload["context_observability_packet"] = context_observability
|
|
if used_context_packet:
|
|
payload["used_context_packet"] = used_context_packet
|
|
if context_evidence_packet:
|
|
payload["context_evidence_packet"] = context_evidence_packet
|
|
if grounding_packet:
|
|
payload["context_grounding_packet"] = grounding_packet
|
|
await self.db.flush()
|
|
return payload
|
|
except Exception as exc:
|
|
run.status = "failed"
|
|
run.last_error = _customer_safe_error_summary(exc)
|
|
run.completed_at = datetime.utcnow()
|
|
await self.db.flush()
|
|
raise
|
|
|
|
def _dynamic_agent_runtime(self) -> Any:
|
|
if self.delivery_runtime is not None and callable(getattr(self.delivery_runtime, "run_turn", None)):
|
|
return self.delivery_runtime
|
|
from ..pmflow_ai_operating_runtime import PMFlowAIOperatingRunService
|
|
|
|
return PMFlowAIOperatingRunService(
|
|
self.db,
|
|
tenant_id=self.tenant_id,
|
|
)
|
|
|
|
async def _persist_agent_runtime_payload(
|
|
self,
|
|
*,
|
|
run: PMFlowAgentRun,
|
|
agent_payload: dict[str, Any],
|
|
answer: str,
|
|
) -> PMFlowAgentRunPacket:
|
|
steps = _steps_from_agent_payload(agent_payload)
|
|
step_rows_by_public_id: dict[str, PMFlowAgentStep] = {}
|
|
for step in steps:
|
|
row = PMFlowAgentStep(
|
|
tenant_id=self.tenant_id,
|
|
run_id=int(run.id),
|
|
public_step_id=step.step_id,
|
|
step_index=int(step.index),
|
|
step_kind=step.kind,
|
|
status=step.status,
|
|
input_summary=(run.goal or "")[:1000],
|
|
output_summary=step.summary,
|
|
tool_name=step.tool_name,
|
|
duration_ms=step.duration_ms,
|
|
step_payload=_storage_safe_json(step.model_dump(mode="json", exclude_none=True), string_limit=1200),
|
|
completed_at=datetime.utcnow(),
|
|
)
|
|
self.db.add(row)
|
|
step_rows_by_public_id[step.step_id] = row
|
|
await self.db.flush()
|
|
|
|
observations = _observations_from_agent_payload(agent_payload, steps=steps)
|
|
for observation in observations:
|
|
source_step = step_rows_by_public_id.get(str(observation.step_id or ""))
|
|
self.db.add(
|
|
PMFlowAgentObservation(
|
|
tenant_id=self.tenant_id,
|
|
run_id=int(run.id),
|
|
step_id=int(source_step.id) if source_step is not None and getattr(source_step, "id", None) is not None else None,
|
|
public_observation_id=observation.observation_id,
|
|
source_kind=observation.source_kind,
|
|
customer_safe_summary=observation.summary,
|
|
confidence=observation.confidence,
|
|
facts=list(_storage_safe_json(observation.facts or [], string_limit=1200, list_limit=32)),
|
|
gaps=list(_storage_safe_json(observation.gaps or [], string_limit=800, list_limit=16)),
|
|
raw_payload=_storage_safe_json(
|
|
observation.model_dump(mode="json", exclude_none=True),
|
|
string_limit=1200,
|
|
list_limit=32,
|
|
),
|
|
observed_at=datetime.utcnow(),
|
|
)
|
|
)
|
|
|
|
artifacts = _artifacts_from_agent_payload(agent_payload, answer=answer)
|
|
for artifact in artifacts:
|
|
self.db.add(
|
|
PMFlowAgentArtifact(
|
|
tenant_id=self.tenant_id,
|
|
run_id=int(run.id),
|
|
step_id=None,
|
|
public_artifact_id=artifact.artifact_id,
|
|
artifact_kind=artifact.kind,
|
|
title=artifact.title,
|
|
status=artifact.status,
|
|
payload=_storage_safe_json(dict(artifact.payload or {}), string_limit=2000, list_limit=24),
|
|
customer_safe_payload=_storage_safe_json(
|
|
artifact.model_dump(mode="json", exclude_none=True),
|
|
string_limit=2000,
|
|
list_limit=24,
|
|
),
|
|
created_by_step_id=steps[-1].step_id if steps else None,
|
|
)
|
|
)
|
|
|
|
verifications = _agent_verifications_from_payload(agent_payload, answer=answer)
|
|
for verification in verifications:
|
|
self.db.add(
|
|
PMFlowAgentVerification(
|
|
tenant_id=self.tenant_id,
|
|
run_id=int(run.id),
|
|
step_id=None,
|
|
public_verification_id=verification.verification_id,
|
|
target_ref=verification.target_ref,
|
|
result=verification.result,
|
|
retry_would_change_state=verification.retry_would_change_state,
|
|
findings=list(_storage_safe_json(verification.findings or [], string_limit=800, list_limit=16)),
|
|
verification_payload=_storage_safe_json(
|
|
verification.model_dump(mode="json", exclude_none=True),
|
|
string_limit=1200,
|
|
list_limit=16,
|
|
),
|
|
verified_at=datetime.utcnow(),
|
|
)
|
|
)
|
|
|
|
memories = _agent_memories_from_payload(agent_payload)
|
|
for memory in memories:
|
|
self.db.add(
|
|
PMFlowAgentMemory(
|
|
tenant_id=self.tenant_id,
|
|
run_id=int(run.id),
|
|
memory_kind=memory.memory_kind,
|
|
scope_ref=memory.scope_ref,
|
|
outcome_signal=memory.outcome_signal,
|
|
content={"summary": _storage_safe_text(memory.summary, limit=1000)},
|
|
memory_payload=_storage_safe_json(memory.model_dump(mode="json", exclude_none=True), string_limit=1000),
|
|
)
|
|
)
|
|
|
|
return PMFlowAgentRunPacket(
|
|
run_id=run.public_run_id,
|
|
status=run.status,
|
|
goal=run.goal,
|
|
surface=run.surface or "space",
|
|
presentation_mode=run.presentation_mode or "pmflow_ai",
|
|
steps=steps,
|
|
observations=observations,
|
|
artifacts=artifacts,
|
|
verifications=verifications,
|
|
memories=memories,
|
|
command_payload=_dict(agent_payload.get("command_payload") or agent_payload.get("answer_canvas")),
|
|
created_at=utc_now_iso(),
|
|
completed_at=utc_now_iso(),
|
|
)
|
|
|
|
def _build_customer_payload_from_agent_runtime(
|
|
self,
|
|
*,
|
|
request: RAGAskRequest,
|
|
run: PMFlowAgentRun,
|
|
packet: PMFlowAgentRunPacket,
|
|
agent_payload: dict[str, Any],
|
|
answer: str,
|
|
) -> dict[str, Any]:
|
|
safe_packet = customer_safe_pmflow_agent_run_packet(packet)
|
|
process_packet = _dict(agent_payload.get("answer_process_packet"))
|
|
run_duration_ms = _agent_run_duration_ms(run)
|
|
if process_packet:
|
|
process_packet = {
|
|
**process_packet,
|
|
"total_duration_ms": run_duration_ms,
|
|
"wall_clock_duration_ms": run_duration_ms,
|
|
}
|
|
answer_canvas = _dict(agent_payload.get("answer_canvas")) or _answer_canvas(answer=answer, packet=packet)
|
|
customer_evidence_packet = _dict(agent_payload.get("customer_evidence_packet")) or _customer_evidence_packet(
|
|
answer=answer,
|
|
packet=packet,
|
|
)
|
|
deliverable_packet = _dict(agent_payload.get("deliverable_packet"))
|
|
answer_runtime_packet = _dict(agent_payload.get("answer_runtime_packet"))
|
|
terminal_state = _dict(agent_payload.get("terminal_answer_state"))
|
|
payload = {
|
|
"answer": answer,
|
|
"reply_text": answer,
|
|
"final_answer": answer,
|
|
"run_id": run.public_run_id,
|
|
"sources": list(agent_payload.get("sources") or []),
|
|
"sources_used": list(agent_payload.get("sources_used") or agent_payload.get("sources") or []),
|
|
"thinking_mode": getattr(request, "thinking_mode", None) or "normal",
|
|
"processing_stage": "pmflow_agent_runtime",
|
|
"answer_mode": "pmflow_agent_runtime",
|
|
"grounding_status": str(agent_payload.get("grounding_status") or ("grounded" if packet.observations else "partially_grounded")),
|
|
"policy_version": PMFLOW_AGENT_OS_POLICY_VERSION,
|
|
"pmflow_agent_run": safe_packet,
|
|
"agent_os_environment": {
|
|
"packet_version": "pmflow_agent_environment.v1",
|
|
"visibility": "customer",
|
|
"steps": safe_packet.get("steps") or [],
|
|
"observations": safe_packet.get("observations") or [],
|
|
},
|
|
"agent_os_run_id": run.public_run_id,
|
|
"agent_os_steps": safe_packet.get("steps") or [],
|
|
"agent_os_observations": safe_packet.get("observations") or [],
|
|
"agent_os_artifacts": safe_packet.get("artifacts") or [],
|
|
"agent_os_verifications": safe_packet.get("verifications") or [],
|
|
"agent_os_memories": safe_packet.get("memories") or [],
|
|
"operating_mission": _dict(agent_payload.get("operating_mission")),
|
|
"project_world_state_packet": _dict(agent_payload.get("project_world_state_packet")),
|
|
"operating_run_packet": _dict(agent_payload.get("operating_run_packet")),
|
|
"agent_os_specialists": {"runtime": "mission_first_operating_runtime"} if _dict(agent_payload.get("operating_mission")) else {},
|
|
"agent_os_specialist_assignments": list(agent_payload.get("agent_assignments") or []),
|
|
"agent_os_work_products": list(agent_payload.get("agent_work_products") or []),
|
|
"agent_os_agent_council_review": _dict(agent_payload.get("agent_council_review")),
|
|
"agent_os_operating_plan": _dict(agent_payload.get("operating_plan")),
|
|
"agent_os_autonomy_gate": _dict(agent_payload.get("autonomy_gate")),
|
|
"agent_os_operating_cycle": _dict(agent_payload.get("operating_cycle")),
|
|
"agent_os_consumed_learning_ledger": _dict(agent_payload.get("consumed_learning_ledger")),
|
|
"agent_os_outcome_observations": list(agent_payload.get("outcome_observations") or []),
|
|
"agent_os_learning_ledger": _dict(agent_payload.get("learning_ledger")),
|
|
"agent_os_learning_updates": list(agent_payload.get("learning_updates") or []),
|
|
"agent_os_next_cycle_triggers": list(agent_payload.get("next_cycle_triggers") or []),
|
|
"agent_os_prepared_action_packets": _prepared_actions_from_agent_payload(agent_payload),
|
|
"agent_os_confirmation_requests": _confirmation_requests_from_agent_payload(agent_payload),
|
|
"agent_os_execution_preparations": list(agent_payload.get("execution_preparations") or []),
|
|
"agent_os_product_object_projections": list(agent_payload.get("product_object_projections") or []),
|
|
"agent_os_command_payload": _dict(agent_payload.get("command_payload")),
|
|
"agent_assignments": list(agent_payload.get("agent_assignments") or []),
|
|
"agent_work_products": list(agent_payload.get("agent_work_products") or []),
|
|
"agent_council_review": _dict(agent_payload.get("agent_council_review")),
|
|
"autonomy_gate": _dict(agent_payload.get("autonomy_gate")),
|
|
"prepared_action_packets": _prepared_actions_from_agent_payload(agent_payload),
|
|
"confirmation_requests": _confirmation_requests_from_agent_payload(agent_payload),
|
|
"execution_preparations": list(agent_payload.get("execution_preparations") or []),
|
|
"product_object_projections": list(agent_payload.get("product_object_projections") or []),
|
|
"operating_cycle": _dict(agent_payload.get("operating_cycle")),
|
|
"consumed_learning_ledger": _dict(agent_payload.get("consumed_learning_ledger")),
|
|
"outcome_observations": list(agent_payload.get("outcome_observations") or []),
|
|
"learning_ledger": _dict(agent_payload.get("learning_ledger")),
|
|
"learning_updates": list(agent_payload.get("learning_updates") or []),
|
|
"next_cycle_triggers": list(agent_payload.get("next_cycle_triggers") or []),
|
|
"operating_plan": _dict(agent_payload.get("operating_plan")),
|
|
"command_payload": _dict(agent_payload.get("command_payload")),
|
|
"answer_process_packet": process_packet or _answer_process_packet(packet, total_duration_ms=run_duration_ms),
|
|
"customer_evidence_packet": customer_evidence_packet,
|
|
"answer_canvas": answer_canvas,
|
|
"deliverable_packet": deliverable_packet,
|
|
"answer_runtime_packet": answer_runtime_packet,
|
|
"terminal_answer_state": terminal_state,
|
|
"turn_presentation_contract": _dict(agent_payload.get("turn_presentation_contract")),
|
|
"customer_interaction_context_packet": _customer_interaction_context_from_agent_payload(agent_payload),
|
|
# Kept for customer payload sanitizers and debug-only product gates.
|
|
# The PMFlow AI workspace does not render these raw packets.
|
|
"agent_run_packet": _dict(agent_payload.get("agent_run_packet")),
|
|
"research_workspace": _dict(agent_payload.get("research_workspace")),
|
|
}
|
|
return payload
|
|
|
|
|
|
def _answer_process_packet(packet: PMFlowAgentRunPacket, *, total_duration_ms: int | None = None) -> dict[str, Any]:
|
|
duration = total_duration_ms if total_duration_ms is not None else sum(int(step.duration_ms or 0) for step in packet.steps)
|
|
return {
|
|
"packet_version": "customer_answer_process.v1",
|
|
"visibility": "customer",
|
|
"answer_type": "pmflow_agent_runtime",
|
|
"route_display": "pmflow_agent_runtime",
|
|
"steps": [
|
|
{
|
|
"key": step.kind,
|
|
"label": step.label,
|
|
"status": step.status,
|
|
"display_summary": step.summary,
|
|
"duration_ms": step.duration_ms,
|
|
}
|
|
for step in packet.steps
|
|
],
|
|
"total_duration_ms": max(0, int(duration or 0)),
|
|
"wall_clock_duration_ms": max(0, int(duration or 0)),
|
|
}
|
|
|
|
|
|
def _agent_run_duration_ms(run: PMFlowAgentRun) -> int:
|
|
if getattr(run, "started_at", None) is not None and getattr(run, "completed_at", None) is not None:
|
|
return max(0, int((run.completed_at - run.started_at).total_seconds() * 1000))
|
|
return 0
|
|
|
|
|
|
def _steps_from_agent_payload(agent_payload: dict[str, Any]) -> list[PMFlowAgentStepPacket]:
|
|
agent_run = _dict(agent_payload.get("agent_run_packet"))
|
|
rows = list(agent_run.get("steps") or [])
|
|
if not rows:
|
|
process = _dict(agent_payload.get("answer_process_packet"))
|
|
rows = list(process.get("steps") or [])
|
|
steps: list[PMFlowAgentStepPacket] = []
|
|
for index, row in enumerate(rows[:32]):
|
|
if not isinstance(row, dict):
|
|
continue
|
|
step_key = str(row.get("key") or row.get("step_id") or f"agent_step_{index + 1}").strip()[:80]
|
|
selected_tool = str(row.get("selected_tool") or row.get("tool_name") or "").strip()[:120] or None
|
|
step_type = str(row.get("step_type") or row.get("type") or row.get("key") or "agent_step").strip()[:64]
|
|
kind = selected_tool if step_type == "tool" and selected_tool else step_type
|
|
label = str(row.get("label") or _agent_step_label(kind) or f"推进第 {index + 1} 步").strip()[:160]
|
|
summary = str(row.get("display_summary") or row.get("summary") or row.get("description") or "").strip()[:1000]
|
|
status = _durable_step_status(row.get("status"), blocking=bool(row.get("blocking")))
|
|
steps.append(
|
|
PMFlowAgentStepPacket(
|
|
step_id=str(row.get("step_id") or step_key or new_pmflow_agent_id("step"))[:80],
|
|
index=index,
|
|
kind=kind or "agent_step",
|
|
status=status,
|
|
label=label,
|
|
summary=summary or label,
|
|
tool_name=selected_tool,
|
|
duration_ms=_safe_int(row.get("duration_ms")),
|
|
)
|
|
)
|
|
if not steps:
|
|
steps.append(
|
|
PMFlowAgentStepPacket(
|
|
step_id=new_pmflow_agent_id("step"),
|
|
index=0,
|
|
kind="agent_answer_runtime",
|
|
status="completed" if str(agent_payload.get("answer") or "").strip() else "blocked",
|
|
label="生成回答",
|
|
summary="已根据本轮客户安全状态组织回答。",
|
|
)
|
|
)
|
|
return _dedupe_steps(steps)
|
|
|
|
|
|
def _observations_from_agent_payload(agent_payload: dict[str, Any], *, steps: list[PMFlowAgentStepPacket]) -> list[PMFlowAgentObservationPacket]:
|
|
agent_run = _dict(agent_payload.get("agent_run_packet"))
|
|
rows = list(agent_run.get("observations") or [])
|
|
if not rows:
|
|
workspace = _dict(agent_payload.get("agent_workspace"))
|
|
rows = list(workspace.get("observations") or [])
|
|
step_id_by_observation_id = {
|
|
str(row.get("observation_id")): str(row.get("step_id"))
|
|
for row in list(agent_run.get("steps") or [])
|
|
if isinstance(row, dict) and row.get("observation_id") and row.get("step_id")
|
|
}
|
|
tool_steps = [step for step in steps if step.tool_name]
|
|
fallback_step_id = (tool_steps[-1].step_id if tool_steps else (steps[-1].step_id if steps else None))
|
|
observations: list[PMFlowAgentObservationPacket] = []
|
|
for index, row in enumerate(rows[:48]):
|
|
if not isinstance(row, dict):
|
|
continue
|
|
facts = [
|
|
_storage_safe_json(dict(item), string_limit=1200, list_limit=12, dict_limit=32)
|
|
for item in list(row.get("facts") or [])[:32]
|
|
if isinstance(item, dict)
|
|
]
|
|
observation_id = str(row.get("observation_id") or new_pmflow_agent_id("obs"))[:80]
|
|
observations.append(
|
|
PMFlowAgentObservationPacket(
|
|
observation_id=observation_id,
|
|
step_id=str(row.get("step_id") or step_id_by_observation_id.get(observation_id) or fallback_step_id or "")[:80] or None,
|
|
source_kind=str(row.get("source") or row.get("source_kind") or "agent_runtime")[:80],
|
|
summary=_storage_safe_text(row.get("summary"), limit=1200),
|
|
confidence=_storage_safe_text(row.get("confidence") or "unknown", limit=32),
|
|
facts=facts,
|
|
gaps=_gaps_from_agent_observation(row, facts=facts),
|
|
)
|
|
)
|
|
if not observations:
|
|
observations.append(
|
|
PMFlowAgentObservationPacket(
|
|
step_id=fallback_step_id,
|
|
source_kind="agent_runtime",
|
|
summary="本轮没有产生额外工具观察。",
|
|
confidence="unknown",
|
|
facts=[],
|
|
gaps=[],
|
|
)
|
|
)
|
|
return _dedupe_observations(observations)
|
|
|
|
|
|
def _artifacts_from_agent_payload(agent_payload: dict[str, Any], *, answer: str) -> list[PMFlowAgentArtifactPacket]:
|
|
artifacts: list[PMFlowAgentArtifactPacket] = []
|
|
for projection in list(agent_payload.get("product_object_projections") or [])[:8]:
|
|
if not isinstance(projection, dict):
|
|
continue
|
|
artifacts.append(
|
|
PMFlowAgentArtifactPacket(
|
|
kind="product_object_projection",
|
|
status=str(projection.get("status") or "ready_for_review")[:32],
|
|
title=str(projection.get("title") or projection.get("object_type") or "经营对象候选")[:240],
|
|
summary=str(projection.get("summary") or "")[:1000],
|
|
payload=projection,
|
|
)
|
|
)
|
|
for product in list(agent_payload.get("agent_work_products") or [])[:12]:
|
|
if not isinstance(product, dict):
|
|
continue
|
|
artifacts.append(
|
|
PMFlowAgentArtifactPacket(
|
|
kind="specialist_work_product",
|
|
status=str(product.get("status") or "completed")[:32],
|
|
title=_customer_work_product_title(product),
|
|
summary=str(product.get("summary") or "")[:1000],
|
|
payload=product,
|
|
)
|
|
)
|
|
deliverable = _dict(agent_payload.get("deliverable_packet"))
|
|
if deliverable:
|
|
artifacts.append(
|
|
PMFlowAgentArtifactPacket(
|
|
kind="agent_deliverable",
|
|
status=str(deliverable.get("status") or "completed")[:32],
|
|
title=str(deliverable.get("title") or "本轮结果")[:240],
|
|
summary=str(deliverable.get("summary") or deliverable.get("answer_text") or "")[:1000],
|
|
payload=deliverable,
|
|
)
|
|
)
|
|
research_workspace = _dict(agent_payload.get("research_workspace"))
|
|
if research_workspace and (research_workspace.get("candidate_items") or research_workspace.get("research_goal")):
|
|
artifacts.append(
|
|
PMFlowAgentArtifactPacket(
|
|
kind="research_workspace",
|
|
status="completed",
|
|
title=str(research_workspace.get("topic") or "研究工作区")[:240],
|
|
summary=str(research_workspace.get("research_goal") or research_workspace.get("topic") or "")[:1000],
|
|
payload=research_workspace,
|
|
)
|
|
)
|
|
quality_findings = list(_dict(agent_payload.get("answer_runtime_packet")).get("quality_findings") or [])
|
|
artifacts.append(_final_answer_artifact(answer, status="blocked" if quality_findings else None))
|
|
return _dedupe_artifacts(artifacts)
|
|
|
|
|
|
def _agent_verifications_from_payload(agent_payload: dict[str, Any], *, answer: str) -> list[PMFlowAgentVerificationPacket]:
|
|
agent_run = _dict(agent_payload.get("agent_run_packet"))
|
|
events = [item for item in list(agent_run.get("event_log") or []) if isinstance(item, dict)]
|
|
verifications: list[PMFlowAgentVerificationPacket] = []
|
|
for event in events[:40]:
|
|
event_type = str(event.get("event_type") or "").strip()
|
|
if event_type not in {"critic_review", "runtime_stop"}:
|
|
continue
|
|
is_stop = event_type == "runtime_stop"
|
|
stop_reason = str(event.get("stop_reason") or "").strip()
|
|
blocked = is_stop and stop_reason not in {"planner_ready_to_answer", "ready_to_answer", "planner_answer", "completed"}
|
|
verifications.append(
|
|
PMFlowAgentVerificationPacket(
|
|
target_ref=f"{event_type}:{event.get('iteration', 0)}",
|
|
result="blocked" if blocked else "passed",
|
|
retry_would_change_state=bool(event.get("state_changed") or blocked),
|
|
findings=[
|
|
{
|
|
"type": stop_reason or event_type,
|
|
"summary": str(event.get("summary") or stop_reason or event_type)[:500],
|
|
}
|
|
],
|
|
)
|
|
)
|
|
quality_findings = list(_dict(agent_payload.get("answer_runtime_packet")).get("quality_findings") or [])
|
|
verifications.append(
|
|
PMFlowAgentVerificationPacket(
|
|
target_ref="final_answer",
|
|
result="passed" if answer.strip() and not quality_findings else "blocked",
|
|
retry_would_change_state=False,
|
|
findings=[
|
|
{"type": str(item or "quality_finding")[:120], "summary": "回答出口命中质量门禁。"}
|
|
for item in quality_findings[:8]
|
|
],
|
|
)
|
|
)
|
|
return _dedupe_verifications(verifications)
|
|
|
|
|
|
def _agent_memories_from_payload(agent_payload: dict[str, Any]) -> list[PMFlowAgentMemoryPacket]:
|
|
learning = _dict(agent_payload.get("learning_ledger"))
|
|
learning_items = list(learning.get("learning_items") or [])
|
|
memories: list[PMFlowAgentMemoryPacket] = []
|
|
for item in learning_items[:8]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
memories.append(
|
|
PMFlowAgentMemoryPacket(
|
|
memory_kind="operating_learning",
|
|
scope_ref=str(learning.get("scope_ref") or item.get("scope_ref") or "")[:160] or None,
|
|
outcome_signal=str(item.get("outcome_status") or item.get("learning_key") or "observed")[:80],
|
|
summary=str(item.get("summary") or item.get("learning_key") or "")[:1000],
|
|
)
|
|
)
|
|
if memories:
|
|
return _dedupe_memories(memories)
|
|
workspace = _dict(agent_payload.get("agent_workspace"))
|
|
deliverable = _dict(agent_payload.get("deliverable_packet"))
|
|
summary = str(deliverable.get("summary") or workspace.get("current_goal") or "").strip()
|
|
if not summary:
|
|
return []
|
|
return [
|
|
PMFlowAgentMemoryPacket(
|
|
memory_kind="agent_workspace_snapshot",
|
|
scope_ref=str(workspace.get("workspace_id") or "")[:160] or None,
|
|
outcome_signal=str(deliverable.get("status") or workspace.get("status") or "observed")[:80],
|
|
summary=summary[:1000],
|
|
)
|
|
]
|
|
|
|
|
|
def _gaps_from_agent_observation(row: dict[str, Any], *, facts: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
gaps: list[dict[str, Any]] = []
|
|
observation_type = str(row.get("observation_type") or "").strip()
|
|
if observation_type in {"tool_rejected", "external_search_failed", "external_search_missing_query", "project_world_state_missing_scope"}:
|
|
gaps.append(
|
|
{
|
|
"type": observation_type,
|
|
"summary": str(row.get("summary") or "本轮观察未完整完成。")[:500],
|
|
"user_actionable": False,
|
|
}
|
|
)
|
|
for fact in facts[:16]:
|
|
fact_type = str(fact.get("type") or "").strip()
|
|
if fact_type.startswith("missing_") or fact_type in {"tool_rejected", "tool_error", "no_object_candidates"}:
|
|
gaps.append(
|
|
{
|
|
"type": fact_type,
|
|
"summary": str(fact.get("summary") or fact.get("reason") or row.get("summary") or fact_type)[:500],
|
|
"user_actionable": bool(fact.get("user_actionable", False)),
|
|
}
|
|
)
|
|
return gaps[:12]
|
|
|
|
|
|
def _agent_payload_object_refs(agent_payload: dict[str, Any]) -> list[str]:
|
|
refs: list[str] = []
|
|
mission = _dict(agent_payload.get("operating_mission"))
|
|
scope = _dict(mission.get("project_scope"))
|
|
for item in list(scope.get("object_refs") or []):
|
|
if isinstance(item, str) and item.strip():
|
|
refs.append(item.strip()[:160])
|
|
for item in list(mission.get("target_objects") or []):
|
|
if isinstance(item, dict):
|
|
object_ref = str(item.get("object_ref") or "").strip()
|
|
if object_ref:
|
|
refs.append(object_ref[:160])
|
|
for source in (
|
|
_dict(agent_payload.get("deliverable_packet")).get("object_refs"),
|
|
_dict(agent_payload.get("answer_canvas")).get("object_refs"),
|
|
):
|
|
for item in list(source or []):
|
|
if isinstance(item, str) and item.strip():
|
|
refs.append(item.strip()[:160])
|
|
elif isinstance(item, dict):
|
|
object_ref = str(item.get("object_ref") or "").strip()
|
|
if object_ref:
|
|
refs.append(object_ref[:160])
|
|
workspace = _dict(agent_payload.get("agent_workspace"))
|
|
for item in list(workspace.get("object_ledger") or [])[:12]:
|
|
if isinstance(item, dict):
|
|
object_ref = str(item.get("object_ref") or "").strip()
|
|
if object_ref:
|
|
refs.append(object_ref[:160])
|
|
return _dedupe([item for item in refs if item])[:12]
|
|
|
|
|
|
def _customer_interaction_context_from_agent_payload(agent_payload: dict[str, Any]) -> dict[str, Any]:
|
|
refs = _agent_payload_object_refs(agent_payload)
|
|
objects: list[dict[str, Any]] = []
|
|
workspace = _dict(agent_payload.get("agent_workspace"))
|
|
for item in list(workspace.get("object_ledger") or [])[:12]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
object_ref = str(item.get("object_ref") or "").strip()
|
|
if not object_ref:
|
|
continue
|
|
objects.append(
|
|
{
|
|
"object_ref": object_ref[:160],
|
|
"object_type": str(item.get("object_type") or object_ref.split(":", 1)[0] or "object")[:80],
|
|
"title": str(item.get("title") or object_ref)[:240],
|
|
"binding_strength": str(item.get("binding_strength") or "candidate")[:80],
|
|
}
|
|
)
|
|
if not objects:
|
|
objects = [
|
|
{
|
|
"object_ref": ref,
|
|
"object_type": ref.split(":", 1)[0] if ":" in ref else "object",
|
|
"title": ref,
|
|
"binding_strength": "derived_from_agent_output",
|
|
}
|
|
for ref in refs
|
|
]
|
|
return {
|
|
"packet_version": "customer_interaction_context.v1",
|
|
"visibility": "customer",
|
|
"current_focus_object": objects[0] if objects else None,
|
|
"last_shown_objects": objects[:12],
|
|
"recent_objects": objects[:12],
|
|
}
|
|
|
|
|
|
def _prepared_actions_from_agent_payload(agent_payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
operating_rows = agent_payload.get("prepared_action_packets")
|
|
if isinstance(operating_rows, list) and operating_rows:
|
|
out: list[dict[str, Any]] = []
|
|
for index, item in enumerate(operating_rows[:8], start=1):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
title = str(item.get("title") or "").strip()
|
|
if not _is_customer_visible_action_title(title):
|
|
continue
|
|
out.append(
|
|
{
|
|
"action_id": str(item.get("prepared_action_id") or f"operating_action_{index}")[:80],
|
|
"title": title[:120],
|
|
"summary": _customer_visible_action_detail(
|
|
item.get("summary")
|
|
or _dict(item.get("payload")).get("title")
|
|
or "已准备为待确认动作，确认前不会写入或执行。"
|
|
),
|
|
"href": str(item.get("href") or "")[:300],
|
|
"target_ref": str(item.get("target_ref") or "")[:160] or None,
|
|
"status": str(item.get("status") or "prepared_waiting_confirmation")[:80],
|
|
"confirmation_required": bool(item.get("confirmation_required", True)),
|
|
"source": "operating_prepared_action",
|
|
}
|
|
)
|
|
return out
|
|
canvas = _dict(agent_payload.get("answer_canvas"))
|
|
rows = canvas.get("action_candidates")
|
|
if not isinstance(rows, list):
|
|
rows = []
|
|
answer_text = str(agent_payload.get("answer") or agent_payload.get("reply_text") or canvas.get("direct_answer") or "").strip()
|
|
valid_rows = [
|
|
item for item in rows
|
|
if isinstance(item, dict)
|
|
and _is_customer_visible_action_title(item.get("title"))
|
|
and str(item.get("href") or "").strip()
|
|
]
|
|
if len(valid_rows) > 1:
|
|
mentioned = [
|
|
item for item in valid_rows
|
|
if isinstance(item, dict)
|
|
and str(item.get("title") or "").strip()
|
|
and str(item.get("title") or "").strip() in answer_text
|
|
]
|
|
rows = mentioned if mentioned else valid_rows[:1]
|
|
else:
|
|
rows = valid_rows
|
|
out: list[dict[str, Any]] = []
|
|
for index, item in enumerate(rows[:8], start=1):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
title = str(item.get("title") or "").strip()
|
|
row = dict(item)
|
|
href = str(row.get("href") or "").strip()[:300]
|
|
if not _is_customer_visible_action_title(title) or not href:
|
|
continue
|
|
summary = _customer_visible_action_detail(row.get("reason") or row.get("completion_criteria") or row.get("summary"))
|
|
out.append(
|
|
{
|
|
"action_id": str(row.get("action_id") or f"agent_action_{index}")[:80],
|
|
"title": title[:120],
|
|
"summary": summary,
|
|
"href": href,
|
|
"target_ref": str(row.get("target_ref") or row.get("object_ref") or "")[:160] or None,
|
|
"status": "prepared_waiting_confirmation" if bool(row.get("requires_confirmation") or row.get("requires_user_confirm")) else "suggested",
|
|
"confirmation_required": bool(row.get("requires_confirmation") or row.get("requires_user_confirm")),
|
|
"source": "deliverable_action_candidate",
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def _is_customer_visible_action_title(value: object) -> bool:
|
|
text = " ".join(str(value or "").strip().split())
|
|
if not text or len(text) > 120:
|
|
return False
|
|
lowered = text.lower()
|
|
internal_terms = (
|
|
"goalframe",
|
|
"top_next_actions",
|
|
"turnstate",
|
|
"claim ledger",
|
|
"boundary ledger",
|
|
"agent_os",
|
|
"runtime",
|
|
"planner",
|
|
"critic",
|
|
"workspace",
|
|
"target_ref",
|
|
"object_ref",
|
|
"tool",
|
|
)
|
|
if any(term in lowered for term in internal_terms):
|
|
return False
|
|
if re.search(r"\b(project|approval|change_set|task):\d+\b", lowered):
|
|
return False
|
|
if "；对象" in text or "；依据" in text or "; object" in lowered:
|
|
return False
|
|
latin_count = sum(1 for ch in text if ("a" <= ch.lower() <= "z"))
|
|
cjk_count = sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff")
|
|
return not (latin_count > 20 and cjk_count == 0)
|
|
|
|
|
|
def _customer_visible_action_detail(value: object) -> str:
|
|
text = " ".join(str(value or "").strip().split())[:240]
|
|
if not text:
|
|
return ""
|
|
lowered = text.lower()
|
|
if any(term in lowered for term in ("goalframe", "top_next_actions", "turnstate", "target_ref", "object_ref")):
|
|
return ""
|
|
if re.search(r"\b(project|approval|change_set|task):\d+\b", lowered):
|
|
return ""
|
|
return text
|
|
|
|
|
|
def _customer_work_product_title(product: dict[str, Any]) -> str:
|
|
labels = {
|
|
"IdeaSeed": "想法种子",
|
|
"ResearchNote": "研究记录",
|
|
"CompetitorMap": "竞品地图",
|
|
"OpportunityBrief": "机会简报",
|
|
"MVPOption": "MVP 选项",
|
|
"TaskDraft": "任务草案",
|
|
"TechnicalArchitectureAssessment": "技术评估",
|
|
"QAPlan": "质量与风险计划",
|
|
"GrowthExperiment": "增长实验",
|
|
"RevenueSignal": "商业化信号",
|
|
"LearningUpdate": "学习记录",
|
|
}
|
|
product_type = str(product.get("product_type") or "").strip()
|
|
return labels.get(product_type, "专业 Agent 产物")[:240]
|
|
|
|
|
|
def _confirmation_requests_from_agent_payload(agent_payload: dict[str, Any]) -> list[dict[str, Any]]:
|
|
operating_rows = agent_payload.get("confirmation_requests")
|
|
if isinstance(operating_rows, list) and operating_rows:
|
|
out: list[dict[str, Any]] = []
|
|
for index, item in enumerate(operating_rows[:8], start=1):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
out.append(
|
|
{
|
|
"confirmation_id": str(item.get("confirmation_request_id") or f"operating_confirmation_{index}")[:80],
|
|
"title": str(item.get("title") or "需要确认的动作")[:240],
|
|
"summary": str(item.get("confirmation_reason") or item.get("summary") or "")[:500],
|
|
"status": str(item.get("status") or "waiting_for_user_confirmation")[:80],
|
|
"requested_for": [str(value)[:120] for value in list(item.get("requested_for") or []) if str(value).strip()][:8],
|
|
"target_ref": str(item.get("target_ref") or "")[:160],
|
|
"action_label": str(item.get("action_label") or "")[:120],
|
|
"action_type": str(item.get("action_type") or "")[:120],
|
|
"action_payload": dict(item.get("action_payload") or {}) if isinstance(item.get("action_payload"), dict) else {},
|
|
"can_execute_after_confirmation": bool(item.get("can_execute_after_confirmation")),
|
|
}
|
|
)
|
|
return out
|
|
deliverable = _dict(agent_payload.get("deliverable_packet"))
|
|
rows = deliverable.get("pending_confirmations") if isinstance(deliverable.get("pending_confirmations"), list) else []
|
|
out: list[dict[str, Any]] = []
|
|
for index, item in enumerate(rows[:8], start=1):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
out.append(
|
|
{
|
|
"confirmation_id": str(item.get("confirmation_id") or f"agent_confirmation_{index}")[:80],
|
|
"title": str(item.get("title") or item.get("summary") or "需要确认的动作")[:240],
|
|
"summary": str(item.get("summary") or item.get("description") or "")[:500],
|
|
"status": str(item.get("status") or "waiting_for_user_confirmation")[:80],
|
|
"action_label": str(item.get("action_label") or "")[:120],
|
|
"action_type": str(item.get("action_type") or "")[:120],
|
|
"action_payload": dict(item.get("action_payload") or {}) if isinstance(item.get("action_payload"), dict) else {},
|
|
"can_execute_after_confirmation": bool(item.get("can_execute_after_confirmation")),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def _durable_status_from_agent_payload(agent_payload: dict[str, Any], *, answer: str) -> str:
|
|
if not answer.strip():
|
|
return "blocked"
|
|
agent_status = str(_dict(agent_payload.get("agent_run_packet")).get("status") or "").strip()
|
|
deliverable_status = str(_dict(agent_payload.get("deliverable_packet")).get("status") or "").strip()
|
|
if agent_status in {"blocked", "waiting_for_user"} or deliverable_status in {"blocked", "needs_customer_input"}:
|
|
return "blocked"
|
|
return "completed"
|
|
|
|
|
|
def _durable_step_status(value: Any, *, blocking: bool = False) -> str:
|
|
if blocking:
|
|
return "blocked"
|
|
text = str(value or "").strip().lower()
|
|
if text in {"completed", "done", "passed"}:
|
|
return "completed"
|
|
if text in {"active", "running", "pending"}:
|
|
return "active"
|
|
if text in {"blocked", "failed"}:
|
|
return "blocked"
|
|
if text in {"skipped"}:
|
|
return "skipped"
|
|
return "completed"
|
|
|
|
|
|
def _agent_step_label(kind: str) -> str:
|
|
labels = {
|
|
"understand_goal": "承接目标",
|
|
"workspace_snapshot": "建立工作区",
|
|
"planner": "规划下一步",
|
|
"critic": "检查状态",
|
|
"answer": "生成回答",
|
|
"external_search": "读取外部公开信息",
|
|
"inspect_space_collections": "读取空间状态",
|
|
"inspect_project_operating_state": "读取项目运行状态",
|
|
"inspect_pmflow_capability_boundary": "读取能力边界",
|
|
"inspect_saved_ideas": "读取暂存想法",
|
|
"inspect_attention_queue": "读取注意力队列",
|
|
"inspect_object_affordances": "读取对象可用动作",
|
|
"inspect_project_execution_package": "读取项目执行包",
|
|
"inspect_source_detail": "读取公开来源详情",
|
|
"resolve_customer_object": "解析对象锚点",
|
|
"inspect_object_detail": "读取对象详情",
|
|
}
|
|
return labels.get(str(kind or "").strip(), "")
|
|
|
|
|
|
def _storage_safe_text(value: Any, *, limit: int) -> str:
|
|
text = str(value or "")
|
|
if not text:
|
|
return ""
|
|
# PostgreSQL jsonb/text cannot store NUL bytes. External pages can contain
|
|
# copied control characters, so normalize once at the Agent OS persistence
|
|
# boundary instead of letting a single bad source roll back the whole turn.
|
|
text = text.replace("\x00", "")
|
|
text = text.encode("utf-8", errors="replace").decode("utf-8", errors="replace")
|
|
return " ".join(text.strip().split())[:limit]
|
|
|
|
|
|
def _storage_safe_json(
|
|
value: Any,
|
|
*,
|
|
string_limit: int = 2000,
|
|
list_limit: int = 24,
|
|
dict_limit: int = 64,
|
|
depth: int = 0,
|
|
max_depth: int = 5,
|
|
) -> Any:
|
|
if value is None or isinstance(value, (bool, int)):
|
|
return value
|
|
if isinstance(value, float):
|
|
return value if math.isfinite(value) else None
|
|
if isinstance(value, str):
|
|
return _storage_safe_text(value, limit=string_limit)
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
if depth >= max_depth:
|
|
return _storage_safe_text(value, limit=min(string_limit, 500))
|
|
if isinstance(value, dict):
|
|
out: dict[str, Any] = {}
|
|
for raw_key, raw_item in list(value.items())[:dict_limit]:
|
|
key = _storage_safe_text(raw_key, limit=120)
|
|
if not key:
|
|
continue
|
|
out[key] = _storage_safe_json(
|
|
raw_item,
|
|
string_limit=string_limit,
|
|
list_limit=list_limit,
|
|
dict_limit=dict_limit,
|
|
depth=depth + 1,
|
|
max_depth=max_depth,
|
|
)
|
|
return out
|
|
if isinstance(value, (list, tuple, set)):
|
|
return [
|
|
_storage_safe_json(
|
|
item,
|
|
string_limit=string_limit,
|
|
list_limit=list_limit,
|
|
dict_limit=dict_limit,
|
|
depth=depth + 1,
|
|
max_depth=max_depth,
|
|
)
|
|
for item in list(value)[:list_limit]
|
|
]
|
|
dump = getattr(value, "model_dump", None)
|
|
if callable(dump):
|
|
try:
|
|
return _storage_safe_json(
|
|
dump(mode="json", exclude_none=True),
|
|
string_limit=string_limit,
|
|
list_limit=list_limit,
|
|
dict_limit=dict_limit,
|
|
depth=depth + 1,
|
|
max_depth=max_depth,
|
|
)
|
|
except Exception:
|
|
return _storage_safe_text(value, limit=min(string_limit, 500))
|
|
return _storage_safe_text(value, limit=min(string_limit, 500))
|
|
|
|
|
|
def _dict(value: Any) -> dict[str, Any]:
|
|
if isinstance(value, dict):
|
|
return dict(value)
|
|
dump = getattr(value, "model_dump", None)
|
|
if callable(dump):
|
|
try:
|
|
return dump(mode="json", exclude_none=True)
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _dedupe(rows: list[str]) -> list[str]:
|
|
out: list[str] = []
|
|
seen: set[str] = set()
|
|
for row in rows:
|
|
text = str(row or "").strip()
|
|
key = " ".join(text.lower().split())
|
|
if not text or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(text)
|
|
return out
|
|
|
|
|
|
def _customer_evidence_packet(*, answer: str, packet: PMFlowAgentRunPacket) -> dict[str, Any]:
|
|
return {
|
|
"packet_version": "customer_evidence_packet.v1",
|
|
"visibility": "customer",
|
|
"answer_type": "pmflow_agent_runtime",
|
|
"confidence": "high" if packet.observations else "medium",
|
|
"answer": answer,
|
|
"facts": [
|
|
fact
|
|
for observation in packet.observations[:8]
|
|
for fact in list(observation.facts or [])[:6]
|
|
if isinstance(fact, dict)
|
|
][:24],
|
|
"limitations": [
|
|
finding.get("summary")
|
|
for verification in packet.verifications
|
|
for finding in list(verification.findings or [])[:4]
|
|
if isinstance(finding, dict) and str(finding.get("summary") or "").strip()
|
|
][:12],
|
|
}
|
|
|
|
|
|
def _answer_canvas(*, answer: str, packet: PMFlowAgentRunPacket) -> dict[str, Any]:
|
|
return {
|
|
"packet_version": "answer_canvas.v1",
|
|
"visibility": "customer",
|
|
"direct_answer": answer,
|
|
"next_actions": list(packet.prepared_action_packets or [])[:6],
|
|
"confirmation_requests": list(packet.confirmation_requests or [])[:6],
|
|
"process_steps": [step.model_dump(mode="json", exclude_none=True) for step in packet.steps[:12]],
|
|
}
|
|
|
|
|
|
def _final_answer_artifact(answer: str, *, status: str | None = None) -> PMFlowAgentArtifactPacket:
|
|
return PMFlowAgentArtifactPacket(
|
|
kind="final_answer",
|
|
status=status or ("completed" if answer.strip() else "blocked"),
|
|
title="PMFlow AI 回答",
|
|
summary=answer[:1000],
|
|
payload={"answer": answer},
|
|
)
|
|
|
|
|
|
def _dedupe_steps(steps: list[PMFlowAgentStepPacket]) -> list[PMFlowAgentStepPacket]:
|
|
out: list[PMFlowAgentStepPacket] = []
|
|
seen: set[str] = set()
|
|
for step in steps:
|
|
step_id = str(step.step_id or new_pmflow_agent_id("step"))[:80]
|
|
if step_id in seen:
|
|
step_id = f"{step_id[:64]}_{len(out)}"
|
|
seen.add(step_id)
|
|
out.append(step.model_copy(update={"step_id": step_id, "index": len(out)}))
|
|
return out[:32]
|
|
|
|
|
|
def _dedupe_observations(observations: list[PMFlowAgentObservationPacket]) -> list[PMFlowAgentObservationPacket]:
|
|
out: list[PMFlowAgentObservationPacket] = []
|
|
seen: set[str] = set()
|
|
for observation in observations:
|
|
observation_id = str(observation.observation_id or new_pmflow_agent_id("obs"))[:80]
|
|
if observation_id in seen:
|
|
observation_id = f"{observation_id[:64]}_{len(out)}"
|
|
seen.add(observation_id)
|
|
out.append(observation.model_copy(update={"observation_id": observation_id}))
|
|
return out[:48]
|
|
|
|
|
|
def _dedupe_artifacts(artifacts: list[PMFlowAgentArtifactPacket]) -> list[PMFlowAgentArtifactPacket]:
|
|
out: list[PMFlowAgentArtifactPacket] = []
|
|
seen: set[str] = set()
|
|
for artifact in artifacts:
|
|
key = f"{artifact.kind}:{artifact.title}:{artifact.summary}"[:500]
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(artifact)
|
|
return out[:32]
|
|
|
|
|
|
def _dedupe_verifications(verifications: list[PMFlowAgentVerificationPacket]) -> list[PMFlowAgentVerificationPacket]:
|
|
out: list[PMFlowAgentVerificationPacket] = []
|
|
seen: set[str] = set()
|
|
for verification in verifications:
|
|
verification_id = str(verification.verification_id or new_pmflow_agent_id("ver"))[:80]
|
|
if verification_id in seen:
|
|
verification_id = f"{verification_id[:64]}_{len(out)}"
|
|
seen.add(verification_id)
|
|
out.append(verification.model_copy(update={"verification_id": verification_id}))
|
|
return out[:40]
|
|
|
|
|
|
def _dedupe_memories(memories: list[PMFlowAgentMemoryPacket]) -> list[PMFlowAgentMemoryPacket]:
|
|
out: list[PMFlowAgentMemoryPacket] = []
|
|
seen: set[str] = set()
|
|
for memory in memories:
|
|
key = f"{memory.memory_kind}:{memory.scope_ref}:{memory.summary}"[:500]
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(memory)
|
|
return out[:12]
|
|
|
|
|
|
def _title_from_question(question: Any) -> str:
|
|
text = " ".join(str(question or "").strip().split())
|
|
return text[:80] or "PMFlow AI Run"
|
|
|
|
|
|
def _customer_safe_error_summary(exc: Exception) -> str:
|
|
return f"{exc.__class__.__name__}: runtime failed"[:500]
|
|
|
|
|
|
def _safe_int(value: Any) -> int | None:
|
|
try:
|
|
if value is None or value == "":
|
|
return None
|
|
return int(value)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
__all__ = ["PMFLOW_AGENT_OS_POLICY_VERSION", "PMFlowAgentRuntimeService"]
|