433 lines
19 KiB
Python
433 lines
19 KiB
Python
"""Operating mission builder for PMFlow AI multi-agent runtime."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from typing import Any
|
||
|
||
from ...schemas.pmflow_ai_operating_runtime import PMFlowAIOperatingMission, PMFlowAIOperatingScope
|
||
from ...schemas.rag import RAGAskRequest
|
||
from ..ai_function_runtime_profile_service import AIFunctionRuntimeProfileService
|
||
from ..llm_service import llm_service
|
||
from ..pmflow_ai_context_pack_service import slim_pmflow_context_pack
|
||
from ..prompt_runtime_service import PromptRuntimeRenderResult, PromptRuntimeService
|
||
from ..runtime_audit_context import bind_agent_audit_context
|
||
from ..pmflow_ai_thread_context_service import slim_thread_context_pack
|
||
from .agent_registry import PMFlowAIOperatingAgentRegistry
|
||
|
||
|
||
PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION = "PROMPT_PMFLOW_AI_OPERATING_MISSION"
|
||
FUNCTION_NAME_PMFLOW_AI_OPERATING_MISSION = "pmflow_ai_operating_mission"
|
||
SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME = "space_pmflow_ai_operating_runtime"
|
||
|
||
|
||
class PMFlowAIOperatingMissionService:
|
||
"""Build a mission contract from user input and customer-safe context."""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
db: Any | None = None,
|
||
tenant_id: int | None = None,
|
||
generator: Any | None = None,
|
||
agent_registry: PMFlowAIOperatingAgentRegistry | None = None,
|
||
) -> None:
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id or 1)
|
||
self.generator = generator or llm_service
|
||
self.agent_registry = agent_registry or PMFlowAIOperatingAgentRegistry()
|
||
|
||
async def build_mission(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
current_user: Any,
|
||
conversation_context: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
) -> PMFlowAIOperatingMission:
|
||
question = " ".join(str(request.question or "").strip().split())
|
||
fallback = self._fallback_mission(
|
||
question=question,
|
||
request=request,
|
||
current_user=current_user,
|
||
conversation_context=conversation_context,
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
generate = getattr(self.generator, "generate", None)
|
||
if not callable(generate) or not callable(getattr(self.db, "execute", None)):
|
||
return fallback
|
||
try:
|
||
prompt_result = await self._render_prompt(
|
||
question=question,
|
||
conversation_context=conversation_context,
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
runtime_profile = await AIFunctionRuntimeProfileService(self.db).resolve_profile(
|
||
function_name=FUNCTION_NAME_PMFLOW_AI_OPERATING_MISSION,
|
||
scene_key=SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME,
|
||
prompt_key=PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION,
|
||
)
|
||
with bind_agent_audit_context(agent_step_key="operating_mission"):
|
||
raw = await generate(
|
||
prompt_result.prompt,
|
||
db=self.db,
|
||
tenant_id=self.tenant_id,
|
||
function_name=FUNCTION_NAME_PMFLOW_AI_OPERATING_MISSION,
|
||
use_prompt_profile=False,
|
||
temperature=runtime_profile.temperature,
|
||
response_mode=runtime_profile.response_mode,
|
||
max_tokens=runtime_profile.max_tokens,
|
||
routing_meta={
|
||
"scene_key": SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME,
|
||
"runtime": "operating_mission_runtime",
|
||
"scene_contract": "operating_mission_generation",
|
||
**runtime_profile.routing_meta(),
|
||
"allow_low_to_high_escalation": False,
|
||
"disable_thinking": True,
|
||
**prompt_result.routing_meta(),
|
||
},
|
||
)
|
||
if raw is None or (isinstance(raw, str) and not raw.strip()):
|
||
raise ValueError("operating_mission_empty_llm_response")
|
||
mission = PMFlowAIOperatingMission.model_validate(
|
||
{
|
||
**fallback.model_dump(mode="json", exclude_none=True),
|
||
**_extract_json_object(raw),
|
||
"original_user_input": question,
|
||
"project_scope": fallback.project_scope.model_dump(mode="json", exclude_none=True),
|
||
}
|
||
)
|
||
return _repair_mission(mission=mission, fallback=fallback, registry=self.agent_registry)
|
||
except Exception:
|
||
return fallback
|
||
|
||
def _fallback_mission(
|
||
self,
|
||
*,
|
||
question: str,
|
||
request: RAGAskRequest,
|
||
current_user: Any,
|
||
conversation_context: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
) -> PMFlowAIOperatingMission:
|
||
project_id = _current_project_id(
|
||
request=request,
|
||
conversation_context=conversation_context,
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
object_refs = _object_refs(
|
||
project_id=project_id,
|
||
conversation_context=conversation_context,
|
||
tenant_project_context=tenant_project_context,
|
||
)
|
||
mission_goal = _mission_goal(question=question, object_refs=object_refs)
|
||
desired_outcome = _desired_outcome(object_refs=object_refs)
|
||
scope = PMFlowAIOperatingScope(
|
||
tenant_id=int(getattr(current_user, "tenant_id", 0) or 0) or None,
|
||
user_id=int(getattr(current_user, "id", 0) or 0) or None,
|
||
session_id=int(request.session_id) if request.session_id is not None else None,
|
||
object_refs=object_refs,
|
||
current_project_id=project_id,
|
||
surface=_surface_name(request.surface_context),
|
||
)
|
||
target_objects = _target_objects_from_refs(object_refs)
|
||
unbound = not object_refs
|
||
# Fallback only runs when the LLM mission planner is unavailable. Keep it
|
||
# capability-contract based and scope-aware so the product still behaves,
|
||
# but do not let it override active LLM-authored specialist selection.
|
||
required_agents = (
|
||
[
|
||
"idea_incubation_agent",
|
||
"product_strategy_agent",
|
||
"learning_agent",
|
||
]
|
||
if unbound
|
||
else [
|
||
"product_strategy_agent",
|
||
"mvp_scope_agent",
|
||
"pm_breakdown_agent",
|
||
"tech_architect_agent",
|
||
"qa_risk_agent",
|
||
"growth_agent",
|
||
"learning_agent",
|
||
]
|
||
)
|
||
return PMFlowAIOperatingMission(
|
||
mission_goal=mission_goal,
|
||
stage="research" if unbound else "execution_bridge",
|
||
desired_outcome=desired_outcome,
|
||
desired_business_outcome=desired_outcome,
|
||
original_user_input=question,
|
||
project_scope=scope,
|
||
target_objects=target_objects,
|
||
deliverable_type="opportunity_brief" if unbound else "operating_decision",
|
||
action_ladder_level="prepare_object" if unbound else "prepare_execution",
|
||
success_criteria=[
|
||
"明确本轮 Operating Mission 的目标、对象范围和业务交付物。",
|
||
"组织 specialist agents 产出结构化 work products，并优先形成产品对象投影。",
|
||
"把可推进动作拆成 safe suggestions、prepared changes、requires confirmation 和 blocked actions。",
|
||
"保留证据、边界、下一轮触发和 replay 所需运行证据。",
|
||
],
|
||
autonomy_level="prepare_changes",
|
||
decision_boundary=[
|
||
"Project Memory 写入、ChangeSet commit、任务变更、Execution Bridge handoff、发布、外部动作和权限变更必须经过用户确认。",
|
||
"本轮只能准备经营对象、计划、执行交接包或验证动作，不直接执行写操作。",
|
||
],
|
||
blocked_boundaries=[
|
||
"不得用单段回答替代需要沉淀的经营对象。",
|
||
"不得在证据不足时把假设包装成已确认事实。",
|
||
"不得自动写入 Project Memory、提交 ChangeSet、创建正式任务、发起 handoff 或发布。",
|
||
],
|
||
required_agents=required_agents,
|
||
next_cycle_policy={
|
||
"trigger_policy": "when_user_confirms_or_new_signal_arrives",
|
||
"follow_up_triggers": ["user_confirmation", "market_signal", "execution_result", "experiment_result"],
|
||
"memory_write_policy": "record_learning_only_until_user_confirms",
|
||
},
|
||
)
|
||
|
||
async def _render_prompt(
|
||
self,
|
||
*,
|
||
question: str,
|
||
conversation_context: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
) -> PromptRuntimeRenderResult:
|
||
variables = {
|
||
"question": question[:1600],
|
||
"conversation_context_json": _dump_json(_mission_safe_conversation_context(conversation_context), limit=5000),
|
||
"tenant_project_context_json": _dump_json(_customer_safe_surface_context(tenant_project_context), limit=8000),
|
||
"operating_mission_schema_json": _dump_json(_operating_mission_prompt_contract(), limit=5000),
|
||
"specialist_contracts_json": _dump_json(
|
||
[item.model_dump(mode="json", exclude_none=True) for item in self.agent_registry.list_contracts()],
|
||
limit=12000,
|
||
),
|
||
}
|
||
return await PromptRuntimeService(self.db).render(
|
||
tenant_id=self.tenant_id,
|
||
prompt_key=PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION,
|
||
**variables,
|
||
)
|
||
|
||
|
||
def _mission_goal(*, question: str, object_refs: list[str]) -> str:
|
||
if object_refs:
|
||
return f"围绕 {', '.join(object_refs[:3])} 推进用户本轮目标：{question or '继续推进当前项目'}"
|
||
return f"把用户输入转成可推进的 PMFlow operating mission：{question or '继续推进当前空间'}"
|
||
|
||
|
||
def _operating_mission_prompt_contract() -> dict[str, Any]:
|
||
schema = PMFlowAIOperatingMission.model_json_schema()
|
||
properties = schema.get("properties") if isinstance(schema.get("properties"), dict) else {}
|
||
|
||
def _field(name: str) -> dict[str, Any]:
|
||
item = properties.get(name) if isinstance(properties.get(name), dict) else {}
|
||
result: dict[str, Any] = {"type": item.get("type") or "value"}
|
||
enum_values = item.get("enum") if isinstance(item.get("enum"), list) else []
|
||
if enum_values:
|
||
result["allowed_values"] = enum_values
|
||
return result
|
||
|
||
return {
|
||
"source": "PMFlowAIOperatingMission schema summary",
|
||
"fields": {
|
||
"mission_goal": _field("mission_goal"),
|
||
"stage": _field("stage"),
|
||
"desired_outcome": _field("desired_outcome"),
|
||
"desired_business_outcome": _field("desired_business_outcome"),
|
||
"target_objects": {"type": "array", "item": "customer-safe business object reference"},
|
||
"deliverable_type": _field("deliverable_type"),
|
||
"action_ladder_level": _field("action_ladder_level"),
|
||
"required_agents": {"type": "array", "item": "agent_id selected from specialist_contracts"},
|
||
"success_criteria": _field("success_criteria"),
|
||
"blocked_boundaries": _field("blocked_boundaries"),
|
||
"decision_boundary": _field("decision_boundary"),
|
||
"next_cycle_policy": _field("next_cycle_policy"),
|
||
},
|
||
}
|
||
|
||
|
||
def _desired_outcome(*, object_refs: list[str]) -> str:
|
||
if object_refs:
|
||
return "形成面向当前项目对象的经营判断、专业 agent 产物、可确认操作包和下一轮推进计划。"
|
||
return "形成面向当前空间的经营判断、专业 agent 产物、可确认操作包和下一轮推进计划。"
|
||
|
||
|
||
def _target_objects_from_refs(object_refs: list[str]) -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"object_ref": ref,
|
||
"object_type": ref.split(":", 1)[0] if ":" in ref else "business_object",
|
||
"title": ref,
|
||
"binding_strength": "operating_mission",
|
||
}
|
||
for ref in object_refs[:12]
|
||
]
|
||
|
||
|
||
def _surface_name(surface_context: dict[str, Any] | None) -> str:
|
||
surface = surface_context if isinstance(surface_context, dict) else {}
|
||
for key in ("surface", "surface_name", "entry", "page"):
|
||
text = str(surface.get(key) or "").strip()
|
||
if text:
|
||
return text[:120]
|
||
return "space"
|
||
|
||
|
||
def _current_project_id(
|
||
*,
|
||
request: RAGAskRequest,
|
||
conversation_context: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
) -> int | None:
|
||
for value in (
|
||
getattr(request, "project_id", None),
|
||
(conversation_context or {}).get("current_project_id"),
|
||
_project_id_from_surface(tenant_project_context),
|
||
):
|
||
try:
|
||
if value is not None and str(value).strip().isdigit():
|
||
return int(value)
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _project_id_from_surface(tenant_project_context: dict[str, Any] | None) -> int | None:
|
||
surface = (tenant_project_context or {}).get("surface_context")
|
||
if not isinstance(surface, dict):
|
||
return None
|
||
focus = surface.get("current_focus_object")
|
||
if isinstance(focus, dict):
|
||
for key in ("object_id", "project_id", "id"):
|
||
value = focus.get(key)
|
||
if value is not None and str(value).strip().isdigit():
|
||
return int(value)
|
||
ref = str(focus.get("object_ref") or "").strip()
|
||
prefix, _, raw_id = ref.partition(":")
|
||
if prefix == "project" and raw_id.strip().isdigit():
|
||
return int(raw_id)
|
||
return None
|
||
|
||
|
||
def _object_refs(
|
||
*,
|
||
project_id: int | None,
|
||
conversation_context: dict[str, Any] | None,
|
||
tenant_project_context: dict[str, Any] | None,
|
||
) -> list[str]:
|
||
refs: list[str] = []
|
||
if project_id is not None:
|
||
refs.append(f"project:{int(project_id)}")
|
||
source = (conversation_context or {}).get("object_ledger")
|
||
if isinstance(source, list):
|
||
for item in source[:12]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
ref = str(item.get("object_ref") or "").strip()
|
||
binding_strength = str(item.get("binding_strength") or "").strip()
|
||
if ref and binding_strength in {"explicit", "session_object_ledger", "current_focus"} and ref not in refs:
|
||
refs.append(ref[:160])
|
||
return refs[:12]
|
||
|
||
|
||
def _repair_mission(
|
||
*,
|
||
mission: PMFlowAIOperatingMission,
|
||
fallback: PMFlowAIOperatingMission,
|
||
registry: PMFlowAIOperatingAgentRegistry,
|
||
) -> PMFlowAIOperatingMission:
|
||
contracts = {item.agent_id for item in registry.list_contracts()}
|
||
required_agents = [item for item in mission.required_agents if item in contracts]
|
||
if not required_agents:
|
||
required_agents = list(fallback.required_agents)
|
||
decision_boundary = [*list(mission.decision_boundary or []), *list(fallback.decision_boundary or [])]
|
||
blocked_boundaries = [*list(mission.blocked_boundaries or []), *list(fallback.blocked_boundaries or [])]
|
||
return mission.model_copy(
|
||
update={
|
||
"mission_goal": mission.mission_goal or fallback.mission_goal,
|
||
"desired_outcome": mission.desired_outcome or fallback.desired_outcome,
|
||
"desired_business_outcome": mission.desired_business_outcome or mission.desired_outcome or fallback.desired_business_outcome,
|
||
"target_objects": mission.target_objects or fallback.target_objects,
|
||
"required_agents": required_agents[:12],
|
||
"success_criteria": list(mission.success_criteria or fallback.success_criteria)[:16],
|
||
"decision_boundary": _dedupe(decision_boundary)[:16],
|
||
"blocked_boundaries": _dedupe(blocked_boundaries)[:16],
|
||
"next_cycle_policy": mission.next_cycle_policy or fallback.next_cycle_policy,
|
||
}
|
||
)
|
||
|
||
|
||
def _extract_json_object(raw: Any) -> dict[str, Any]:
|
||
if isinstance(raw, dict):
|
||
return raw
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
return {}
|
||
try:
|
||
parsed = json.loads(text)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except Exception:
|
||
pass
|
||
match = re.search(r"\{.*\}", text, flags=re.S)
|
||
if not match:
|
||
return {}
|
||
parsed = json.loads(match.group(0))
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
|
||
|
||
def _dump_json(value: Any, *, limit: int) -> str:
|
||
return json.dumps(value, ensure_ascii=False, default=str)[:limit]
|
||
|
||
|
||
def _customer_safe_surface_context(tenant_project_context: dict[str, Any] | None) -> dict[str, Any]:
|
||
context = dict(tenant_project_context or {})
|
||
surface = context.get("surface_context")
|
||
return {"surface_context": surface if isinstance(surface, dict) else {}}
|
||
|
||
|
||
def _mission_safe_conversation_context(conversation_context: dict[str, Any] | None) -> dict[str, Any]:
|
||
context = conversation_context if isinstance(conversation_context, dict) else {}
|
||
safe = {}
|
||
for key in ("session_id", "conversation_scope", "current_project_id", "surface", "source_channel"):
|
||
value = context.get(key)
|
||
if value not in (None, "", [], {}):
|
||
safe[key] = value
|
||
capsules = []
|
||
for item in list(context.get("context_capsules") or [])[:8]:
|
||
if isinstance(item, dict):
|
||
capsule_type = str(item.get("capsule_type") or item.get("type") or "").strip()
|
||
if capsule_type in {"pmflow_ai_object_ledger", "object_ledger", "session_object_ledger"}:
|
||
capsules.append(
|
||
{
|
||
"capsule_type": capsule_type,
|
||
"current_focus_object": item.get("current_focus_object"),
|
||
"recent_objects": list(item.get("recent_objects") or [])[:4],
|
||
}
|
||
)
|
||
if capsules:
|
||
safe["context_capsules"] = capsules
|
||
context_pack = slim_pmflow_context_pack(context.get("context_pack"), limit_chars=18000)
|
||
if context_pack:
|
||
safe["context_pack"] = context_pack
|
||
thread_context_pack = {} if context_pack else slim_thread_context_pack(context.get("thread_context_pack"), recent_limit=8, relevant_limit=8)
|
||
if thread_context_pack:
|
||
safe["thread_context_pack"] = thread_context_pack
|
||
return safe
|
||
|
||
|
||
def _dedupe(rows: list[str]) -> list[str]:
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for row in rows:
|
||
text = " ".join(str(row or "").strip().split())
|
||
key = text.lower()
|
||
if text and key not in seen:
|
||
seen.add(key)
|
||
out.append(text)
|
||
return out
|
||
|
||
|
||
__all__ = ["PMFlowAIOperatingMissionService", "PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION"]
|