1350 lines
56 KiB
Python
1350 lines
56 KiB
Python
"""Read-only evidence tools for the PMFlow AI turn runtime.
|
||
|
||
The registry is intentionally small: it maps an already selected runtime
|
||
action to a bounded evidence reader. It does not classify the user question,
|
||
does not decide answer shape, and never writes business state.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import and_, desc, or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..models import Project, User, UserProject
|
||
from ..schemas.pmflow_ai_turn_runtime import (
|
||
PMFlowAIEvidenceFinding,
|
||
PMFlowAIEvidenceToolResult,
|
||
PMFlowAIOperationGap,
|
||
PMFlowAIObservation,
|
||
PMFlowAIObservationItem,
|
||
PMFlowAITargetScope,
|
||
PMFlowAITurnObjectRef,
|
||
PMFlowAITurnState,
|
||
RuntimeActionKind,
|
||
)
|
||
from ..schemas.rag import RAGAskRequest
|
||
from .adaptive_rag_service import AdaptiveRAGService
|
||
from .auth import has_any_role
|
||
from .external_query_service import ExternalQueryGateway, ExternalProviderRegistry, ExternalResearchRuntime
|
||
from .external_query_service.adapters import (
|
||
LiveSearchFallbackAdapter,
|
||
MCPExternalAdapter,
|
||
build_dashscope_responses_adapter_from_runtime,
|
||
)
|
||
from .external_query_service.schemas import ExternalEvidenceItem, ExternalQueryPlan
|
||
from .pmflow_ai_project_execution_observation import (
|
||
build_project_execution_findings,
|
||
build_project_operating_evidence_packet,
|
||
)
|
||
from .pmflow_os_projection_service import PMFlowOSProjectionService
|
||
|
||
|
||
class PMFlowAIEvidenceToolRegistry:
|
||
"""Run whitelisted read-only evidence tools selected by TurnState."""
|
||
|
||
def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id or 1)
|
||
|
||
async def run(
|
||
self,
|
||
*,
|
||
action_type: RuntimeActionKind,
|
||
request: RAGAskRequest,
|
||
current_user: User,
|
||
turn_state: PMFlowAITurnState,
|
||
conversation_context: dict[str, Any] | None = None,
|
||
route_plan: dict[str, Any] | None = None,
|
||
started_at: float | None = None,
|
||
) -> dict[str, Any] | None:
|
||
_ = conversation_context
|
||
if action_type == "inspect_surface_context":
|
||
result = self._inspect_surface_context(request=request, turn_state=turn_state)
|
||
elif action_type == "inspect_saved_ideas":
|
||
result = self._inspect_saved_ideas(request=request, turn_state=turn_state)
|
||
elif action_type == "inspect_internal_projects":
|
||
result = await self._inspect_internal_projects(
|
||
request=request,
|
||
current_user=current_user,
|
||
turn_state=turn_state,
|
||
)
|
||
elif action_type == "resolve_project_objects":
|
||
result = await self._resolve_project_objects(
|
||
request=request,
|
||
current_user=current_user,
|
||
turn_state=turn_state,
|
||
)
|
||
elif action_type == "inspect_project_status":
|
||
result = await self._inspect_project_status(
|
||
request=request,
|
||
current_user=current_user,
|
||
turn_state=turn_state,
|
||
)
|
||
elif action_type in {"external_search", "external_fetch"}:
|
||
return await self._run_external_evidence_action(
|
||
request=request,
|
||
route_plan=route_plan or {},
|
||
turn_state=turn_state,
|
||
started_at=started_at,
|
||
)
|
||
else:
|
||
return None
|
||
observation = _observation_from_tool_result(
|
||
action_type=action_type,
|
||
result=result,
|
||
turn_state=turn_state,
|
||
request=request,
|
||
)
|
||
return _payload_from_result(result, observation=observation)
|
||
|
||
def _inspect_surface_context(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> PMFlowAIEvidenceToolResult:
|
||
surface = _dict(request.surface_context)
|
||
workbench = _workbench(surface)
|
||
findings: list[PMFlowAIEvidenceFinding] = []
|
||
sources = [_source("当前空间上下文", "surface_context")]
|
||
|
||
project_rows = _list(workbench, "projects", "visible_projects", "recent_projects")
|
||
if project_rows:
|
||
findings.append(
|
||
_finding(
|
||
fact=f"当前空间可见项目 {len(project_rows)} 个。",
|
||
source="当前空间上下文",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="high",
|
||
)
|
||
)
|
||
for item in project_rows[:12]:
|
||
title = _text(item.get("title") or item.get("name"), limit=160)
|
||
if not title:
|
||
continue
|
||
status = _text(item.get("status") or item.get("state"), limit=80)
|
||
findings.append(
|
||
_finding(
|
||
fact=f"项目：{title}{f'，状态：{status}' if status else ''}。",
|
||
source="当前空间上下文",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="high",
|
||
)
|
||
)
|
||
|
||
for key, label in (
|
||
("top_next_actions", "下一步候选"),
|
||
("pending_confirmations", "待确认事项"),
|
||
("ai_running_items", "AI 进行中事项"),
|
||
):
|
||
rows = _list(workbench, key)
|
||
if rows:
|
||
findings.append(
|
||
_finding(
|
||
fact=f"当前空间有 {len(rows)} 个{label}。",
|
||
source="当前空间上下文",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="high",
|
||
)
|
||
)
|
||
|
||
capacity_findings, capacity_sources = _capacity_findings_from_surface(
|
||
surface=surface,
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
)
|
||
findings.extend(capacity_findings)
|
||
sources.extend(capacity_sources)
|
||
|
||
gaps = [] if findings else ["surface_context_missing_or_empty"]
|
||
return PMFlowAIEvidenceToolResult(
|
||
tool_name="surface_context_tool",
|
||
action_type="inspect_surface_context",
|
||
findings=findings,
|
||
sources=sources,
|
||
remaining_gaps=gaps,
|
||
confidence="high" if findings else "unknown",
|
||
summary=f"读取当前空间上下文，得到 {len(findings)} 条事实。",
|
||
)
|
||
|
||
def _inspect_saved_ideas(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> PMFlowAIEvidenceToolResult:
|
||
workbench = _workbench(_dict(request.surface_context))
|
||
rows = _list(workbench, "saved_ideas", "savedIdeas", "recent_saved_ideas")
|
||
findings: list[PMFlowAIEvidenceFinding] = []
|
||
for item in rows[:6]:
|
||
title = _text(item.get("title") or item.get("name") or item.get("raw_text") or item.get("summary"), limit=160)
|
||
summary = _text(item.get("summary") or item.get("description") or item.get("raw_text"), limit=260)
|
||
if not title and not summary:
|
||
continue
|
||
findings.append(
|
||
_finding(
|
||
fact=f"暂存想法：{title or summary}",
|
||
source="当前空间暂存想法",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="medium",
|
||
)
|
||
)
|
||
return PMFlowAIEvidenceToolResult(
|
||
tool_name="saved_ideas_tool",
|
||
action_type="inspect_saved_ideas",
|
||
findings=findings,
|
||
sources=[_source("当前空间暂存想法", "saved_ideas")],
|
||
remaining_gaps=[] if findings else ["saved_ideas_missing_or_empty"],
|
||
confidence="medium" if findings else "unknown",
|
||
summary=f"读取暂存想法，得到 {len(findings)} 条事实。",
|
||
)
|
||
|
||
async def _inspect_internal_projects(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
current_user: User,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> PMFlowAIEvidenceToolResult:
|
||
surface = _dict(request.surface_context)
|
||
workbench = _workbench(surface)
|
||
project_rows = _list(workbench, "projects", "visible_projects", "recent_projects")
|
||
loaded_rows = await self._load_visible_project_rows(current_user=current_user)
|
||
if loaded_rows:
|
||
project_rows = loaded_rows
|
||
findings: list[PMFlowAIEvidenceFinding] = []
|
||
primary = turn_state.object_state.primary_object
|
||
if _is_strong_object(primary):
|
||
matched = _match_project_rows(project_rows=project_rows, target=primary)
|
||
title = _text(primary.title or (matched or {}).get("title") or (matched or {}).get("name"), limit=180)
|
||
status = _text((matched or {}).get("status") or (matched or {}).get("state"), limit=80)
|
||
findings.append(
|
||
_finding(
|
||
fact=f"已绑定内部项目：{title or primary.object_ref}{f'，状态：{status}' if status else ''}。",
|
||
source="PMFlow 项目上下文",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="high",
|
||
)
|
||
)
|
||
else:
|
||
context_projects = [
|
||
item
|
||
for item in list(turn_state.object_state.context_objects or [])
|
||
if str(item.object_type or "").lower() == "project" or str(item.object_ref or "").startswith("project")
|
||
]
|
||
if context_projects:
|
||
titles = [_text(item.title or item.object_ref, limit=80) for item in context_projects[:6]]
|
||
findings.append(
|
||
_finding(
|
||
fact=f"当前空间有可见项目上下文：{'、'.join(title for title in titles if title)}。",
|
||
source="PMFlow 项目上下文",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="medium",
|
||
)
|
||
)
|
||
elif project_rows:
|
||
findings.append(
|
||
_finding(
|
||
fact=f"当前空间可见项目 {len(project_rows)} 个，但本轮没有强项目锚点。",
|
||
source="PMFlow 项目上下文",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="medium",
|
||
)
|
||
)
|
||
|
||
gaps = [] if findings else ["internal_project_context_missing"]
|
||
return PMFlowAIEvidenceToolResult(
|
||
tool_name="internal_project_evidence_tool",
|
||
action_type="inspect_internal_projects",
|
||
findings=findings,
|
||
sources=[_source("PMFlow 项目上下文", "internal_projects")],
|
||
remaining_gaps=gaps,
|
||
confidence="high" if _is_strong_object(primary) and findings else ("medium" if findings else "unknown"),
|
||
summary=f"读取内部项目证据，得到 {len(findings)} 条事实。",
|
||
)
|
||
|
||
async def _resolve_project_objects(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
current_user: User,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> PMFlowAIEvidenceToolResult:
|
||
project_rows = await self._project_rows_for_turn(request=request, current_user=current_user)
|
||
target = _object_query_from_turn(turn_state)
|
||
matches = _rank_project_rows(project_rows=project_rows, target=target, context_objects=turn_state.object_state.context_objects)
|
||
findings: list[PMFlowAIEvidenceFinding] = []
|
||
for row in matches[:6]:
|
||
title = _text(row.get("title") or row.get("name"), limit=160)
|
||
if not title:
|
||
continue
|
||
status = _text(row.get("status") or row.get("state"), limit=80)
|
||
reason = _text(row.get("match_reason"), limit=220)
|
||
findings.append(
|
||
_finding(
|
||
fact=f"匹配到内部项目：{title}{f'，状态：{status}' if status else ''}。",
|
||
source="PMFlow 项目对象解析",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="high" if int(row.get("match_score") or 0) >= 80 else "medium",
|
||
)
|
||
)
|
||
if not findings and project_rows:
|
||
findings.append(
|
||
_finding(
|
||
fact=f"当前空间可见项目 {len(project_rows)} 个，但没有解析到足够强的项目对象匹配。",
|
||
source="PMFlow 项目对象解析",
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="medium",
|
||
)
|
||
)
|
||
return PMFlowAIEvidenceToolResult(
|
||
tool_name="project_object_resolver_tool",
|
||
action_type="resolve_project_objects",
|
||
findings=findings,
|
||
sources=[_source("PMFlow 项目对象解析", "project_object_resolver")],
|
||
remaining_gaps=[] if findings else ["project_object_not_resolved"],
|
||
confidence="high" if matches else ("medium" if findings else "unknown"),
|
||
summary=f"解析项目对象，得到 {len(findings)} 条证据。",
|
||
)
|
||
|
||
async def _inspect_project_status(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
current_user: User,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> PMFlowAIEvidenceToolResult:
|
||
project_rows = await self._project_rows_for_turn(request=request, current_user=current_user)
|
||
project_ids = _project_ids_for_status(turn_state=turn_state, project_rows=project_rows)
|
||
findings: list[PMFlowAIEvidenceFinding] = []
|
||
gaps: list[str] = []
|
||
if not project_ids:
|
||
return PMFlowAIEvidenceToolResult(
|
||
tool_name="project_status_evidence_tool",
|
||
action_type="inspect_project_status",
|
||
findings=[],
|
||
sources=[_source("PMFlow 项目状态证据", "project_status")],
|
||
remaining_gaps=["project_object_required_for_status"],
|
||
confidence="unknown",
|
||
summary="没有强项目对象，不能读取项目状态证据。",
|
||
)
|
||
projection = PMFlowOSProjectionService(self.db, tenant_id=self.tenant_id)
|
||
row_by_id = {_safe_int(row.get("id") or row.get("project_id")): row for row in project_rows}
|
||
packets: list[dict[str, Any]] = []
|
||
for project_id in project_ids[:6]:
|
||
row = row_by_id.get(project_id) or {}
|
||
title = _text(row.get("title") or row.get("name") or f"项目 {project_id}", limit=160)
|
||
status = _text(row.get("status") or row.get("state"), limit=80)
|
||
task_counts = _dict(row.get("task_counts") or row.get("tasks") or {})
|
||
truth_payload: dict[str, Any] = {}
|
||
execution_payload: dict[str, Any] = {}
|
||
try:
|
||
truth = await projection.get_truth_state(project_id=project_id, current_user_id=int(getattr(current_user, "id", 0) or 0))
|
||
execution = await projection.get_execution_summary(project_id=project_id, current_user_id=int(getattr(current_user, "id", 0) or 0))
|
||
truth_payload = truth.model_dump(mode="json") if hasattr(truth, "model_dump") else _dict(truth)
|
||
execution_payload = execution.model_dump(mode="json") if hasattr(execution, "model_dump") else _dict(execution)
|
||
except Exception:
|
||
gaps.append(f"project_status_projection_unavailable:{project_id}")
|
||
packet = build_project_operating_evidence_packet(
|
||
project_id=project_id,
|
||
title=title,
|
||
status=status,
|
||
truth_payload=truth_payload,
|
||
execution_payload=execution_payload,
|
||
fallback_task_counts=task_counts,
|
||
object_binding=_object_binding_snapshot(turn_state=turn_state),
|
||
)
|
||
packets.append(packet.model_dump(mode="json", exclude_none=True))
|
||
project_findings, project_gaps = build_project_execution_findings(
|
||
project_id=project_id,
|
||
title=title,
|
||
status=status,
|
||
truth_payload=truth_payload,
|
||
execution_payload=execution_payload,
|
||
fallback_task_counts=task_counts,
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
)
|
||
findings.extend(project_findings)
|
||
gaps.extend(project_gaps)
|
||
return PMFlowAIEvidenceToolResult(
|
||
tool_name="project_status_evidence_tool",
|
||
action_type="inspect_project_status",
|
||
findings=findings,
|
||
structured_evidence={
|
||
"project_operating_evidence_packets": packets,
|
||
"project_operating_evidence_packet": packets[0] if len(packets) == 1 else {},
|
||
},
|
||
sources=[_source("PMFlow 项目执行态证据", "project_status")],
|
||
remaining_gaps=gaps[:6],
|
||
confidence="high" if findings and not gaps else ("medium" if findings else "unknown"),
|
||
summary=f"读取项目状态证据，得到 {len(findings)} 条事实。",
|
||
)
|
||
|
||
async def _run_external_evidence_action(
|
||
self,
|
||
*,
|
||
request: RAGAskRequest,
|
||
route_plan: dict[str, Any],
|
||
turn_state: PMFlowAITurnState,
|
||
started_at: float | None,
|
||
) -> dict[str, Any]:
|
||
started = started_at if isinstance(started_at, (int, float)) else time.perf_counter()
|
||
rag_service = AdaptiveRAGService(self.db, tenant_id=self.tenant_id)
|
||
planner = _TurnRuntimeExternalPlanner(
|
||
question=str(request.question or "").strip(),
|
||
route_plan=route_plan,
|
||
turn_state=turn_state,
|
||
)
|
||
adapters = {
|
||
"dashscope_responses": await build_dashscope_responses_adapter_from_runtime(
|
||
db=self.db,
|
||
tenant_id=1,
|
||
),
|
||
"mcp": MCPExternalAdapter(
|
||
search_callable=getattr(rag_service, "_execute_external_mcp_search", None),
|
||
fetch_text_callable=getattr(rag_service, "_fetch_url_text", None),
|
||
chunk_text_callable=getattr(rag_service, "_chunk_text", None),
|
||
score_chunk_callable=getattr(rag_service, "_score_live_external_chunk", None),
|
||
max_pages=3,
|
||
),
|
||
"live_search": LiveSearchFallbackAdapter(
|
||
build_queries_callable=None,
|
||
search_results_callable=getattr(rag_service, "_search_live_external_results", None),
|
||
fetch_text_callable=getattr(rag_service, "_fetch_url_text", None),
|
||
chunk_text_callable=getattr(rag_service, "_chunk_text", None),
|
||
score_search_result_callable=getattr(rag_service, "_score_live_external_search_result", None),
|
||
score_chunk_callable=getattr(rag_service, "_score_live_external_chunk", None),
|
||
max_pages=3,
|
||
),
|
||
}
|
||
execution = await ExternalResearchRuntime(
|
||
planner=planner,
|
||
registry=ExternalProviderRegistry(
|
||
adapters=[adapters["dashscope_responses"], adapters["mcp"], adapters["live_search"]]
|
||
),
|
||
adapter_timeout_seconds=15.0,
|
||
max_parallel_adapters=3,
|
||
).execute(
|
||
question=str(request.question or "").strip(),
|
||
query_semantics=_external_query_semantics(turn_state=turn_state),
|
||
fit_brief=None,
|
||
)
|
||
evidence_models = _filter_relevant_external_evidence(
|
||
evidence=list(execution.evidence or []),
|
||
turn_state=turn_state,
|
||
)
|
||
evidence_models = _filter_fresh_external_evidence(
|
||
evidence=evidence_models,
|
||
plan=execution.plan,
|
||
turn_state=turn_state,
|
||
)
|
||
context_packet = execution.to_context_packet(question=str(request.question or "").strip()).model_dump(mode="json")
|
||
context_packet["verified_evidence"] = [
|
||
item.model_dump(mode="json", exclude_none=True) for item in evidence_models[:6]
|
||
]
|
||
context_packet["retrieval_status"] = "evidence_found" if evidence_models else "verifier_blocked"
|
||
evidence = [item.model_dump(mode="json", exclude_none=True) for item in evidence_models[:6]]
|
||
findings = [
|
||
_finding(
|
||
fact=str(item.title or item.source_label or item.snippet or item.target_url or "外部公开证据")[:500],
|
||
source=str(item.source_label or item.title or item.target_url or "外部公开来源")[:200],
|
||
relevance=turn_state.turn_goal.user_goal,
|
||
confidence="high"
|
||
if float(item.authority_score or 0.0) >= 0.6 or float(item.freshness_score or 0.0) >= 0.6
|
||
else "medium",
|
||
).model_dump(mode="json", exclude_none=True)
|
||
for item in evidence_models[:6]
|
||
]
|
||
return {
|
||
"_runtime_payload_kind": "evidence",
|
||
"processing_stage": "turn_runtime_external_evidence",
|
||
"answer_mode": "turn_runtime_evidence_tool",
|
||
"grounding_status": "grounded" if evidence_models else "insufficient_evidence",
|
||
"source_policy": "external",
|
||
"runtime_evidence_tool_result": {
|
||
"packet_version": "pmflow_ai_evidence_tool_result.v1",
|
||
"tool_name": "external_query_evidence_tool",
|
||
"action_type": "external_search",
|
||
"findings": findings,
|
||
"sources": _sources_from_external_evidence(evidence_models),
|
||
"remaining_gaps": [] if evidence_models else ["fresh_external_evidence_missing"],
|
||
"confidence": "high" if evidence_models else "unknown",
|
||
"summary": f"查询外部公开来源，得到 {len(evidence_models)} 条可用证据。",
|
||
},
|
||
"runtime_observation": PMFlowAIObservation(
|
||
observation_id="obs_external_search",
|
||
observation_type="external_evidence",
|
||
action_phase="read",
|
||
scope=_observation_scope(turn_state),
|
||
object_type="external_source",
|
||
items=[
|
||
PMFlowAIObservationItem(
|
||
object_type="external_source",
|
||
object_ref=str(item.target_url or "")[:160],
|
||
title=str(item.title or item.source_label or item.target_url or "外部公开证据")[:200],
|
||
href=str(item.target_url or "")[:240],
|
||
summary=str(item.snippet or item.source_label or "")[:400],
|
||
)
|
||
for item in evidence_models[:6]
|
||
],
|
||
count=len(evidence_models),
|
||
can_answer_operations=_can_answer_operations_from_tool_result(
|
||
action_type="external_search",
|
||
has_findings=bool(evidence_models),
|
||
turn_state=turn_state,
|
||
),
|
||
confidence="high" if evidence_models else "unknown",
|
||
remaining_gaps=[] if evidence_models else [
|
||
PMFlowAIOperationGap(
|
||
key="fresh_external_evidence_missing",
|
||
gap_type="system_evidence_gap",
|
||
blocking=False,
|
||
system_action_hint="select_next_action",
|
||
reason="未读取到可用外部新近证据。",
|
||
)
|
||
],
|
||
summary=f"查询外部公开来源，得到 {len(evidence_models)} 条可用证据。",
|
||
).model_dump(mode="json", exclude_none=True),
|
||
"research_contract": {
|
||
"contract_version": "turn_runtime_external_research.v1",
|
||
"effective_lane": "external",
|
||
"external_sources_required": True,
|
||
"project_fit_required": False,
|
||
"completion_surface": "space_ai_concierge",
|
||
},
|
||
"external_query_plan": execution.plan.model_dump(mode="json", exclude_none=True),
|
||
"external_context_packet": context_packet,
|
||
"external_evidence_items": evidence,
|
||
"sources": _sources_from_external_evidence(evidence_models),
|
||
"sources_used": _sources_used_from_external_evidence(evidence_models),
|
||
"response_time_ms": int((time.perf_counter() - started) * 1000),
|
||
"policy_version": "pmflow-turn-runtime-external-evidence-tool-v1",
|
||
}
|
||
|
||
async def _project_rows_for_turn(self, *, request: RAGAskRequest, current_user: User) -> list[dict[str, Any]]:
|
||
loaded = await self._load_visible_project_rows(current_user=current_user)
|
||
if loaded:
|
||
return loaded
|
||
return _list(_workbench(_dict(request.surface_context)), "projects", "visible_projects", "recent_projects")
|
||
|
||
async def _load_visible_project_rows(self, *, current_user: User) -> list[dict[str, Any]]:
|
||
try:
|
||
stmt = (
|
||
select(Project)
|
||
.where(
|
||
and_(
|
||
Project.tenant_id == self.tenant_id,
|
||
Project.status != "archived",
|
||
self._visible_project_filter(current_user=current_user),
|
||
)
|
||
)
|
||
.order_by(desc(Project.updated_at), desc(Project.id))
|
||
.limit(100)
|
||
)
|
||
projects = list((await self.db.execute(stmt)).scalars().all())
|
||
except Exception:
|
||
return []
|
||
rows: list[dict[str, Any]] = []
|
||
for project in projects:
|
||
rows.append(
|
||
{
|
||
"id": int(project.id),
|
||
"project_id": int(project.id),
|
||
"object_ref": f"project:{int(project.id)}",
|
||
"title": _text(getattr(project, "title", "") or getattr(project, "name", ""), limit=180),
|
||
"status": _text(getattr(project, "status", ""), limit=80),
|
||
"description": _text(getattr(project, "description", "") or getattr(project, "summary", ""), limit=500),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
def _visible_project_filter(self, *, current_user: User):
|
||
if has_any_role(str(getattr(current_user, "role", "") or ""), ["OWNER", "REVIEWER"]):
|
||
return True
|
||
return or_(
|
||
Project.owner_id == int(getattr(current_user, "id", 0) or 0),
|
||
Project.id.in_(
|
||
select(UserProject.project_id).where(
|
||
and_(
|
||
UserProject.tenant_id == self.tenant_id,
|
||
UserProject.user_id == int(getattr(current_user, "id", 0) or 0),
|
||
)
|
||
)
|
||
),
|
||
)
|
||
|
||
|
||
def _object_binding_snapshot(*, turn_state: PMFlowAITurnState) -> dict[str, Any]:
|
||
primary = turn_state.object_state.primary_object
|
||
if primary is None:
|
||
return {}
|
||
return {
|
||
"object_type": _text(primary.object_type, limit=80),
|
||
"object_ref": _text(primary.object_ref, limit=160),
|
||
"title": _text(primary.title, limit=200),
|
||
"binding_strength": _text(primary.binding_strength, limit=80),
|
||
"role": _text(primary.role, limit=80),
|
||
}
|
||
|
||
|
||
def _payload_from_result(
|
||
result: PMFlowAIEvidenceToolResult,
|
||
*,
|
||
observation: PMFlowAIObservation,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"_runtime_payload_kind": "evidence",
|
||
"processing_stage": f"turn_runtime_{result.action_type}",
|
||
"answer_mode": "turn_runtime_evidence_tool",
|
||
"grounding_status": "grounded" if result.findings else "insufficient_evidence",
|
||
"source_policy": "internal",
|
||
"runtime_evidence_tool_result": result.model_dump(mode="json", exclude_none=True),
|
||
"runtime_observation": observation.model_dump(mode="json", exclude_none=True),
|
||
"sources": list(result.sources or []),
|
||
"policy_version": "pmflow-turn-runtime-evidence-tool-v1",
|
||
}
|
||
|
||
|
||
def _observation_from_tool_result(
|
||
*,
|
||
action_type: RuntimeActionKind,
|
||
result: PMFlowAIEvidenceToolResult,
|
||
turn_state: PMFlowAITurnState,
|
||
request: RAGAskRequest | None = None,
|
||
) -> PMFlowAIObservation:
|
||
findings = list(result.findings or [])[:20]
|
||
structured_items = _structured_observation_items_for_tool(
|
||
action_type=action_type,
|
||
request=request,
|
||
result=result,
|
||
)
|
||
items = structured_items or [
|
||
PMFlowAIObservationItem(
|
||
object_type=_object_type_for_tool_action(action_type),
|
||
title=item.fact[:200],
|
||
summary=(item.implication or item.fact)[:400],
|
||
business_values={
|
||
"source": item.source,
|
||
"relevance_to_goal": item.relevance_to_goal,
|
||
"confidence": item.confidence,
|
||
},
|
||
)
|
||
for item in findings
|
||
]
|
||
return PMFlowAIObservation(
|
||
observation_id=f"obs_{action_type}",
|
||
observation_type=_observation_type_for_tool_action(action_type),
|
||
action_phase="read",
|
||
scope=_observation_scope(turn_state),
|
||
object_type=_object_type_for_tool_action(action_type),
|
||
items=items,
|
||
count=len(items),
|
||
can_answer_operations=_can_answer_operations_from_tool_result(
|
||
action_type=action_type,
|
||
has_findings=bool(findings or items),
|
||
turn_state=turn_state,
|
||
),
|
||
confidence=result.confidence,
|
||
remaining_gaps=[
|
||
PMFlowAIOperationGap(
|
||
key=str(value or f"{action_type}_gap")[:120],
|
||
gap_type="system_evidence_gap",
|
||
blocking=False,
|
||
system_action_hint="select_next_action",
|
||
reason=str(value or "")[:400],
|
||
)
|
||
for value in list(result.remaining_gaps or [])[:10]
|
||
if str(value or "").strip()
|
||
],
|
||
summary=result.summary,
|
||
)
|
||
|
||
|
||
def _structured_observation_items_for_tool(
|
||
*,
|
||
action_type: RuntimeActionKind,
|
||
request: RAGAskRequest | None,
|
||
result: PMFlowAIEvidenceToolResult,
|
||
) -> list[PMFlowAIObservationItem]:
|
||
if action_type == "inspect_surface_context":
|
||
return _project_items_from_request_surface(request)
|
||
if action_type == "inspect_saved_ideas":
|
||
return _saved_idea_items_from_request_surface(request)
|
||
if action_type in {"resolve_project_objects", "inspect_internal_projects", "inspect_project_status"}:
|
||
packet_items = _project_items_from_operating_packets(result)
|
||
if packet_items:
|
||
return packet_items
|
||
return _project_items_from_findings(result)
|
||
return []
|
||
|
||
|
||
def _project_items_from_operating_packets(result: PMFlowAIEvidenceToolResult) -> list[PMFlowAIObservationItem]:
|
||
structured = _dict(result.structured_evidence)
|
||
packets = structured.get("project_operating_evidence_packets")
|
||
rows = packets if isinstance(packets, list) else []
|
||
out: list[PMFlowAIObservationItem] = []
|
||
for packet in rows[:12]:
|
||
if not isinstance(packet, dict):
|
||
continue
|
||
title = _text(packet.get("project_title"), limit=180)
|
||
project_ref = _text(packet.get("project_ref"), limit=160)
|
||
status = _text(packet.get("project_status"), limit=100)
|
||
coverage = _dict(packet.get("coverage"))
|
||
facets = [
|
||
_facet_summary(packet, "command_state"),
|
||
_facet_summary(packet, "workbench_state"),
|
||
_facet_summary(packet, "execution_state"),
|
||
_facet_summary(packet, "quality_test_state"),
|
||
_facet_summary(packet, "release_growth_state"),
|
||
_facet_summary(packet, "risk_blocker_state"),
|
||
_facet_summary(packet, "project_memory_state"),
|
||
]
|
||
out.append(
|
||
PMFlowAIObservationItem(
|
||
object_type="project",
|
||
object_ref=project_ref,
|
||
title=title or project_ref,
|
||
status=status,
|
||
summary="；".join(item for item in facets if item)[:400],
|
||
business_values={
|
||
"source": "PMFlow 项目全流程证据包",
|
||
"confidence": result.confidence,
|
||
"coverage_score": coverage.get("coverage_score"),
|
||
"covered_facets": coverage.get("covered_facets") or [],
|
||
"missing_facets": coverage.get("missing_facets") or [],
|
||
"remaining_gaps": packet.get("remaining_gaps") or [],
|
||
},
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _facet_summary(packet: dict[str, Any], key: str) -> str:
|
||
facet = _dict(packet.get(key))
|
||
label = _text(facet.get("display_label"), limit=60)
|
||
summary = _text(facet.get("summary"), limit=160)
|
||
if not summary:
|
||
return ""
|
||
return f"{label}：{summary}" if label else summary
|
||
|
||
|
||
def _project_items_from_request_surface(request: RAGAskRequest | None) -> list[PMFlowAIObservationItem]:
|
||
surface = _dict(getattr(request, "surface_context", None))
|
||
rows = _list(_workbench(surface), "projects", "visible_projects", "recent_projects")
|
||
out: list[PMFlowAIObservationItem] = []
|
||
for row in rows[:20]:
|
||
title = _text(row.get("title") or row.get("name"), limit=160)
|
||
if not title:
|
||
continue
|
||
object_ref = _text(row.get("object_ref") or row.get("objectRef"), limit=120)
|
||
if not object_ref and row.get("id") is not None:
|
||
object_ref = f"project:{row.get('id')}"
|
||
status = _text(row.get("status") or row.get("state"), limit=80)
|
||
out.append(
|
||
PMFlowAIObservationItem(
|
||
object_type="project",
|
||
object_ref=object_ref,
|
||
title=title,
|
||
status=status,
|
||
summary=f"{title}{f'，状态：{status}' if status else ''}",
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _saved_idea_items_from_request_surface(request: RAGAskRequest | None) -> list[PMFlowAIObservationItem]:
|
||
surface = _dict(getattr(request, "surface_context", None))
|
||
rows = _list(_workbench(surface), "saved_ideas", "savedIdeas", "recent_saved_ideas")
|
||
out: list[PMFlowAIObservationItem] = []
|
||
for index, row in enumerate(rows[:20], start=1):
|
||
title = _text(row.get("title") or row.get("name") or row.get("raw_text") or row.get("summary"), limit=160)
|
||
if not title:
|
||
continue
|
||
object_ref = _text(row.get("object_ref") or row.get("objectRef"), limit=120)
|
||
if not object_ref:
|
||
object_ref = f"saved_idea:{row.get('id') or index}"
|
||
out.append(
|
||
PMFlowAIObservationItem(
|
||
object_type="saved_idea",
|
||
object_ref=object_ref,
|
||
title=title,
|
||
summary=_text(row.get("summary") or row.get("description") or row.get("raw_text"), limit=400),
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _project_items_from_findings(result: PMFlowAIEvidenceToolResult) -> list[PMFlowAIObservationItem]:
|
||
out: list[PMFlowAIObservationItem] = []
|
||
for index, finding in enumerate(list(result.findings or [])[:20], start=1):
|
||
fact = _text(finding.fact, limit=260)
|
||
title = _project_title_from_fact(fact)
|
||
if not title:
|
||
continue
|
||
out.append(
|
||
PMFlowAIObservationItem(
|
||
object_type="project",
|
||
object_ref=f"project_match:{index}",
|
||
title=title,
|
||
summary=fact,
|
||
business_values={"source": finding.source, "confidence": finding.confidence},
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _project_title_from_fact(value: str) -> str:
|
||
text = _text(value, limit=260)
|
||
for prefix in ("项目：", "项目概况：", "匹配到内部项目：", "已绑定内部项目："):
|
||
if text.startswith(prefix):
|
||
text = text.removeprefix(prefix).strip()
|
||
break
|
||
if not text:
|
||
return ""
|
||
for sep in ("，", "。", " 当前状态证据："):
|
||
if sep in text:
|
||
text = text.split(sep, 1)[0].strip()
|
||
return text[:160]
|
||
|
||
|
||
def _observation_scope(turn_state: PMFlowAITurnState) -> PMFlowAITargetScope:
|
||
if turn_state.selected_frame is not None:
|
||
return turn_state.selected_frame.target_scope
|
||
primary = turn_state.object_state.primary_object
|
||
if _is_strong_object(primary):
|
||
return PMFlowAITargetScope(scope_type="current_project", binding_strength="explicit", source="tool_observation")
|
||
return PMFlowAITargetScope(scope_type="current_space", binding_strength="surface", source="tool_observation")
|
||
|
||
|
||
def _observation_type_for_tool_action(action_type: RuntimeActionKind) -> str:
|
||
if action_type in {"inspect_surface_context", "inspect_saved_ideas"}:
|
||
return "collection_snapshot"
|
||
if action_type in {"resolve_project_objects", "inspect_project_status", "inspect_internal_projects"}:
|
||
return "object_snapshot"
|
||
if action_type in {"external_search", "external_fetch"}:
|
||
return "external_evidence"
|
||
return "scope_probe"
|
||
|
||
|
||
def _object_type_for_tool_action(action_type: RuntimeActionKind) -> 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_tool_result(
|
||
*,
|
||
action_type: RuntimeActionKind,
|
||
has_findings: bool,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> list[str]:
|
||
if not has_findings:
|
||
return []
|
||
selected = turn_state.selected_frame.operation_kind if turn_state.selected_frame else "answer_general"
|
||
out: list[str] = []
|
||
if action_type in {"inspect_surface_context", "inspect_saved_ideas", "resolve_project_objects"} and selected == "list_collection":
|
||
out.append("list_collection")
|
||
elif action_type in {"inspect_surface_context", "inspect_saved_ideas", "resolve_project_objects"} and turn_state.answer_state.answer_shape == "fact_list":
|
||
out.append("list_collection")
|
||
if action_type in {"inspect_project_status", "inspect_internal_projects"} and selected in {"inspect_object", "guide_execution"}:
|
||
out.append(selected)
|
||
if action_type in {"external_search", "external_fetch"} and selected in {"judge_direction", "compare_objects", "answer_general"}:
|
||
out.append(selected)
|
||
if selected == "answer_general":
|
||
out.append("answer_general")
|
||
return list(dict.fromkeys(out))
|
||
|
||
|
||
def _finding(
|
||
*,
|
||
fact: str,
|
||
source: str,
|
||
relevance: str,
|
||
confidence: str,
|
||
) -> PMFlowAIEvidenceFinding:
|
||
return PMFlowAIEvidenceFinding(
|
||
fact=fact,
|
||
source=source,
|
||
relevance_to_goal=relevance,
|
||
confidence=confidence if confidence in {"high", "medium", "low", "unknown"} else "unknown",
|
||
)
|
||
|
||
|
||
def _source(label: str, source: str) -> dict[str, Any]:
|
||
return {"source": source, "display_label": label, "label": label, "confidence": "high"}
|
||
|
||
|
||
def _capacity_findings_from_surface(
|
||
*,
|
||
surface: dict[str, Any],
|
||
relevance: str,
|
||
) -> tuple[list[PMFlowAIEvidenceFinding], list[dict[str, Any]]]:
|
||
operating = _space_operating_packet(surface)
|
||
capacity_packet = _dict(operating.get("capacity_packet"))
|
||
project_capacity = _list(capacity_packet, "project_capacity")
|
||
if not project_capacity:
|
||
return [], []
|
||
findings: list[PMFlowAIEvidenceFinding] = []
|
||
for row in project_capacity[:8]:
|
||
title = _text(row.get("project_title") or row.get("title") or row.get("project_id"), limit=120)
|
||
personal = _dict(row.get("personal"))
|
||
team = _dict(row.get("team"))
|
||
planned = _safe_int(personal.get("planned_minutes") or team.get("planned_minutes"))
|
||
effective = _safe_int(personal.get("effective_minutes") or team.get("effective_minutes"))
|
||
remaining = _safe_int(personal.get("remaining_minutes"))
|
||
status = _text(personal.get("status") or team.get("status"), limit=80)
|
||
fact_parts = [f"容量证据：{title or '当前项目'}"]
|
||
if planned is not None and effective is not None:
|
||
fact_parts.append(f"已计划 {planned} 分钟 / 有效 {effective} 分钟")
|
||
if remaining is not None:
|
||
fact_parts.append(f"剩余 {remaining} 分钟")
|
||
if status:
|
||
fact_parts.append(f"状态 {status}")
|
||
reduction = _safe_int(row.get("ai_reduction_potential_minutes"))
|
||
if reduction is not None and reduction > 0:
|
||
fact_parts.append(f"AI 可减少约 {reduction} 分钟人工处理")
|
||
recommendations = [
|
||
_text(item, limit=160)
|
||
for item in list(row.get("recommendations") or [])
|
||
if _text(item, limit=160)
|
||
]
|
||
if recommendations:
|
||
fact_parts.append(recommendations[0])
|
||
findings.append(
|
||
_finding(
|
||
fact="；".join(fact_parts) + "。",
|
||
source="当前空间容量证据",
|
||
relevance=relevance,
|
||
confidence="high",
|
||
)
|
||
)
|
||
return findings, [_source("当前空间容量证据", "space_capacity_packet")]
|
||
|
||
|
||
def _dict(value: Any) -> dict[str, Any]:
|
||
return dict(value) if isinstance(value, dict) else {}
|
||
|
||
|
||
def _space_operating_packet(surface: dict[str, Any]) -> dict[str, Any]:
|
||
runtime_context = _dict(surface.get("space_runtime_context"))
|
||
operating = _dict(runtime_context.get("space_operating_packet"))
|
||
if operating:
|
||
return operating
|
||
return _dict(surface.get("space_operating_packet"))
|
||
|
||
|
||
def _workbench(surface: dict[str, Any]) -> dict[str, Any]:
|
||
runtime_context = _dict(surface.get("space_runtime_context"))
|
||
workbench = _dict(runtime_context.get("workbench"))
|
||
if workbench:
|
||
return workbench
|
||
return _dict(surface.get("workbench"))
|
||
|
||
|
||
def _list(container: dict[str, Any], *keys: str) -> list[dict[str, Any]]:
|
||
for key in keys:
|
||
value = container.get(key)
|
||
if isinstance(value, list):
|
||
return [dict(item) for item in value if isinstance(item, dict)]
|
||
return []
|
||
|
||
|
||
def _text(value: Any, *, limit: int) -> str:
|
||
return " ".join(str(value or "").strip().split())[:limit]
|
||
|
||
|
||
def _is_strong_object(value: PMFlowAITurnObjectRef | None) -> bool:
|
||
if value is None:
|
||
return False
|
||
binding = str(value.binding_strength or "").strip().lower()
|
||
if binding in {"", "none", "context", "visible_context"}:
|
||
return False
|
||
return bool(str(value.object_ref or value.title or "").strip())
|
||
|
||
|
||
def _match_project_rows(*, project_rows: list[dict[str, Any]], target: PMFlowAITurnObjectRef | None) -> dict[str, Any] | None:
|
||
if target is None:
|
||
return None
|
||
target_ref = str(target.object_ref or "").strip()
|
||
target_title = str(target.title or "").strip()
|
||
for row in project_rows:
|
||
row_ref = str(row.get("object_ref") or row.get("objectRef") or row.get("id") or "").strip()
|
||
row_title = str(row.get("title") or row.get("name") or "").strip()
|
||
if target_ref and (target_ref == row_ref or target_ref.endswith(f":{row_ref}")):
|
||
return row
|
||
if target_title and row_title and target_title == row_title:
|
||
return row
|
||
return None
|
||
|
||
|
||
def _safe_int(value: Any) -> int | None:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _object_query_from_turn(turn_state: PMFlowAITurnState) -> str:
|
||
primary = turn_state.object_state.primary_object
|
||
values = [
|
||
getattr(primary, "title", "") if primary else "",
|
||
getattr(primary, "object_ref", "") if primary else "",
|
||
turn_state.problem_frame.subject,
|
||
]
|
||
for value in values:
|
||
text = _text(value, limit=180)
|
||
if text and not text.startswith("project:"):
|
||
return text
|
||
terms = [term for term in turn_state.problem_frame.subject_terms if _text(term, limit=60)]
|
||
return " ".join(terms[:4])
|
||
|
||
|
||
def _rank_project_rows(
|
||
*,
|
||
project_rows: list[dict[str, Any]],
|
||
target: str,
|
||
context_objects: list[PMFlowAITurnObjectRef],
|
||
) -> list[dict[str, Any]]:
|
||
target_terms = _terms(target)
|
||
context_refs = {str(item.object_ref or "").strip() for item in context_objects if str(item.object_ref or "").strip()}
|
||
scored: list[tuple[int, dict[str, Any]]] = []
|
||
for row in project_rows:
|
||
row_ref = str(row.get("object_ref") or row.get("objectRef") or row.get("id") or "").strip()
|
||
title = _text(row.get("title") or row.get("name"), limit=180)
|
||
desc = _text(row.get("description") or row.get("summary"), limit=500)
|
||
haystack = " ".join([row_ref, title, desc]).lower()
|
||
score = 0
|
||
if row_ref and row_ref in context_refs:
|
||
score += 90
|
||
if target and title and target.strip().lower() == title.lower():
|
||
score += 100
|
||
for term in target_terms:
|
||
if term.lower() in haystack:
|
||
score += 24 if term.lower() in title.lower() else 10
|
||
if score <= 0:
|
||
continue
|
||
copy = dict(row)
|
||
copy["match_score"] = score
|
||
copy["match_reason"] = f"与本轮目标对象“{target}”存在结构化匹配。" if target else "与当前上下文对象存在匹配。"
|
||
scored.append((score, copy))
|
||
scored.sort(key=lambda item: item[0], reverse=True)
|
||
return [row for _, row in scored[:12]]
|
||
|
||
|
||
def _project_ids_for_status(*, turn_state: PMFlowAITurnState, project_rows: list[dict[str, Any]]) -> list[int]:
|
||
ids: list[int] = []
|
||
primary = turn_state.object_state.primary_object
|
||
if _is_strong_object(primary):
|
||
project_id = _project_id_from_ref(primary.object_ref if primary else "")
|
||
if project_id is None:
|
||
matched = _match_project_rows(project_rows=project_rows, target=primary)
|
||
project_id = _safe_int((matched or {}).get("id") or (matched or {}).get("project_id"))
|
||
if project_id is not None:
|
||
ids.append(project_id)
|
||
if not ids:
|
||
for item in list(turn_state.object_state.context_objects or [])[:6]:
|
||
if str(item.object_type or "").lower() != "project" and not str(item.object_ref or "").startswith("project:"):
|
||
continue
|
||
project_id = _project_id_from_ref(item.object_ref)
|
||
if project_id is not None:
|
||
ids.append(project_id)
|
||
out: list[int] = []
|
||
for item in ids:
|
||
if item not in out:
|
||
out.append(item)
|
||
return out[:6]
|
||
|
||
|
||
def _project_id_from_ref(value: Any) -> int | None:
|
||
text = str(value or "").strip()
|
||
if ":" in text:
|
||
text = text.rsplit(":", 1)[-1]
|
||
return _safe_int(text)
|
||
|
||
|
||
class _TurnRuntimeExternalPlanner:
|
||
"""Evidence-gap driven planner for Turn Runtime external research."""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
question: str,
|
||
route_plan: dict[str, Any],
|
||
turn_state: PMFlowAITurnState,
|
||
) -> None:
|
||
self.question = str(question or "").strip()
|
||
_ = route_plan
|
||
self.turn_state = turn_state
|
||
|
||
async def build_plan(
|
||
self,
|
||
*,
|
||
question: str,
|
||
query_semantics: dict[str, Any] | None = None,
|
||
fit_brief: dict[str, Any] | None = None,
|
||
) -> tuple[ExternalQueryPlan, dict[str, Any]]:
|
||
_ = query_semantics, fit_brief
|
||
text = _turn_external_subject(self.turn_state) or str(question or self.question or "").strip()
|
||
queries = _build_turn_external_queries(
|
||
question=text,
|
||
route_plan={},
|
||
turn_state=self.turn_state,
|
||
)
|
||
needs_freshness = _turn_state_needs_freshness(self.turn_state)
|
||
answer_shape = str(self.turn_state.answer_state.answer_shape or "").strip()
|
||
requires_multi_source = bool(len(self.turn_state.interpretation_state.branches or []) > 1 or len(queries) > 2)
|
||
plan = ExternalQueryPlan(
|
||
plan_version="turn_runtime_external_evidence.v1",
|
||
query_kind="generic_web",
|
||
interaction_level=None,
|
||
answer_contract="summary_with_sources",
|
||
freshness_required=needs_freshness,
|
||
topic=text,
|
||
subject_entities=_extract_subject_terms(text)[:6],
|
||
time_range_hint=_fresh_time_range_hint() if needs_freshness else None,
|
||
preferred_domains=[],
|
||
search_queries=queries,
|
||
result_limit=6 if requires_multi_source else 4,
|
||
requires_multi_source=requires_multi_source,
|
||
clarify_on_ambiguity=False,
|
||
recommendation_goal=str(self.turn_state.turn_goal.user_goal or text)[:240],
|
||
)
|
||
return plan, {
|
||
"planner": "turn_runtime_evidence_tool_external_planner",
|
||
"llm_planner_bypassed": True,
|
||
"query_count": len(queries),
|
||
}
|
||
|
||
|
||
def _external_query_semantics(*, turn_state: PMFlowAITurnState) -> dict[str, Any]:
|
||
return {
|
||
"query_type": "external_research",
|
||
"object_mentions": [],
|
||
"query_profile": {
|
||
"needs_freshness": _turn_state_needs_freshness(turn_state),
|
||
"source_policy": "external",
|
||
"answer_shape": turn_state.answer_state.answer_shape,
|
||
},
|
||
}
|
||
|
||
|
||
def _build_turn_external_queries(
|
||
*,
|
||
question: str,
|
||
route_plan: dict[str, Any],
|
||
turn_state: PMFlowAITurnState,
|
||
) -> list[str]:
|
||
text = " ".join(str(_turn_external_subject(turn_state) or question or "").strip().split())
|
||
queries: list[str] = []
|
||
terms = turn_state.problem_frame.subject_terms or _extract_subject_terms(text)
|
||
compact_terms = " ".join(terms[:5])
|
||
for branch in list(turn_state.interpretation_state.branches or []):
|
||
branch_terms = " ".join([branch.name, *branch.evidence_needs[:3]])
|
||
queries.append(f"{text} {branch_terms}".strip())
|
||
_ = route_plan
|
||
if text:
|
||
queries.append(text)
|
||
if compact_terms and compact_terms != text:
|
||
queries.append(f"{text} {compact_terms}")
|
||
for gap in list(turn_state.operation_gaps or []):
|
||
reason = str(gap.reason or gap.key or "").strip()
|
||
if reason:
|
||
queries.append(f"{text} {reason[:120]}")
|
||
out: list[str] = []
|
||
for item in queries:
|
||
normalized = " ".join(str(item or "").strip().split())
|
||
if normalized and normalized not in out:
|
||
out.append(normalized)
|
||
return out[:6]
|
||
|
||
|
||
def _turn_state_needs_freshness(turn_state: PMFlowAITurnState) -> bool:
|
||
if bool(getattr(turn_state.turn_goal, "freshness_required", False)):
|
||
return True
|
||
for gap in list(turn_state.operation_gaps or []):
|
||
if str(gap.gap_type or "").strip() == "freshness_gap":
|
||
return True
|
||
if str(gap.key or "").strip() == "external_fresh":
|
||
return True
|
||
return bool(turn_state.sufficiency_state.needs_external_evidence)
|
||
|
||
|
||
def _turn_external_subject(turn_state: PMFlowAITurnState) -> str:
|
||
primary = turn_state.object_state.primary_object
|
||
title = str(getattr(primary, "title", "") or "").strip() if primary else ""
|
||
binding = str(getattr(primary, "binding_strength", "") or "").strip().lower() if primary else ""
|
||
if title and binding not in {"", "none", "context", "visible_context"}:
|
||
return title
|
||
if turn_state.thread_workspace and turn_state.thread_workspace.primary_object:
|
||
title = str(turn_state.thread_workspace.primary_object.title or "").strip()
|
||
if title:
|
||
return title
|
||
return str(turn_state.problem_frame.subject or "").strip()
|
||
|
||
|
||
def _filter_relevant_external_evidence(
|
||
*,
|
||
evidence: list[ExternalEvidenceItem],
|
||
turn_state: PMFlowAITurnState,
|
||
) -> list[ExternalEvidenceItem]:
|
||
subject = str(turn_state.problem_frame.subject or _turn_external_subject(turn_state) or "").strip()
|
||
if not subject:
|
||
return list(evidence or [])[:6]
|
||
scored: list[tuple[int, ExternalEvidenceItem]] = []
|
||
rejected: list[str] = []
|
||
for item in list(evidence or []):
|
||
score = _external_evidence_relevance_score(item=item, turn_state=turn_state)
|
||
if score > 0:
|
||
scored.append((score, item))
|
||
continue
|
||
label = str(item.title or item.source_label or item.target_url or "").strip()
|
||
if label:
|
||
rejected.append(label[:120])
|
||
if rejected:
|
||
turn_state.evidence_state.stale_or_missing.extend(
|
||
[f"off_topic_external_evidence:{item}" for item in rejected[:4]]
|
||
)
|
||
scored.sort(key=lambda row: row[0], reverse=True)
|
||
return [item for _, item in scored[:6]]
|
||
|
||
|
||
def _filter_fresh_external_evidence(
|
||
*,
|
||
evidence: list[ExternalEvidenceItem],
|
||
plan: ExternalQueryPlan,
|
||
turn_state: PMFlowAITurnState,
|
||
) -> list[ExternalEvidenceItem]:
|
||
if not ExternalQueryGateway._plan_requires_year_grounding(plan=plan):
|
||
return list(evidence or [])[:6]
|
||
year_floor = ExternalQueryGateway._freshness_year_floor(plan=plan)
|
||
fresh: list[ExternalEvidenceItem] = []
|
||
stale: list[str] = []
|
||
undated: list[str] = []
|
||
for item in list(evidence or []):
|
||
years = ExternalQueryGateway._extract_grounding_years(item)
|
||
label = str(item.title or item.source_label or item.target_url or "外部公开来源").strip()[:160]
|
||
if not years:
|
||
undated.append(label)
|
||
continue
|
||
if max(years) >= year_floor:
|
||
fresh.append(item)
|
||
elif label:
|
||
stale.append(f"{label}:{','.join(str(year) for year in sorted(years))}")
|
||
if stale:
|
||
turn_state.evidence_state.stale_or_missing.extend(
|
||
[f"stale_external_evidence:{item}" for item in stale[:6]]
|
||
)
|
||
if undated:
|
||
turn_state.evidence_state.stale_or_missing.extend(
|
||
[f"undated_external_evidence:{item}" for item in undated[:4] if item]
|
||
)
|
||
return fresh[:6]
|
||
|
||
|
||
def _external_evidence_relevance_score(
|
||
*,
|
||
item: ExternalEvidenceItem,
|
||
turn_state: PMFlowAITurnState | None = None,
|
||
subject_override: str | None = None,
|
||
) -> int:
|
||
subject_text = str(subject_override or (turn_state.problem_frame.subject if turn_state else "") or "").strip().lower()
|
||
if not subject_text and turn_state:
|
||
subject_text = _turn_external_subject(turn_state).lower()
|
||
if not subject_text:
|
||
return 1
|
||
haystack = " ".join(
|
||
str(value or "")
|
||
for value in (
|
||
item.source_label,
|
||
item.title,
|
||
item.snippet,
|
||
item.target_url,
|
||
item.raw_metadata,
|
||
)
|
||
).lower()
|
||
if not haystack:
|
||
return 0
|
||
if subject_text in haystack:
|
||
return 8
|
||
terms = list(turn_state.problem_frame.subject_terms if turn_state else _extract_subject_terms(subject_text))
|
||
branch_terms: list[str] = []
|
||
if turn_state:
|
||
for branch in list(turn_state.interpretation_state.branches or []):
|
||
branch_terms.extend(_extract_subject_terms(branch.name))
|
||
for value in [*branch.assumptions, *branch.evidence_needs, *branch.risk_hints]:
|
||
branch_terms.extend(_extract_subject_terms(value))
|
||
score = 0
|
||
for term in terms:
|
||
normalized = str(term or "").strip().lower()
|
||
if len(normalized) >= 2 and normalized in haystack:
|
||
score += 2
|
||
for term in branch_terms:
|
||
normalized = str(term or "").strip().lower()
|
||
if len(normalized) >= 2 and normalized in haystack:
|
||
score += 1
|
||
return score if score >= 3 else 0
|
||
|
||
|
||
def _extract_subject_terms(text: str) -> list[str]:
|
||
normalized = " ".join(str(text or "").strip().replace("，", " ").replace("。", " ").split())
|
||
terms: list[str] = []
|
||
for raw in normalized.split(" "):
|
||
token = raw.strip(" :：,，.。!?！？()（）[]【】\"'")
|
||
if len(token) >= 2 and token not in terms:
|
||
terms.append(token[:60])
|
||
return terms[:8]
|
||
|
||
|
||
def _terms(text: str) -> list[str]:
|
||
return _extract_subject_terms(text)
|
||
|
||
|
||
def _fresh_year_terms() -> str:
|
||
current_year = datetime.now().year
|
||
return f"{current_year} {current_year - 1}"
|
||
|
||
|
||
def _fresh_time_range_hint() -> str:
|
||
current_year = datetime.now().year
|
||
return f"{current_year - 1}-{current_year}"
|
||
|
||
|
||
def _sources_from_external_evidence(evidence: list[ExternalEvidenceItem]) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
for item in list(evidence or [])[:8]:
|
||
label = str(item.source_label or item.title or item.target_url or "外部公开来源").strip()
|
||
url = str(item.target_url or "").strip()
|
||
rows.append(
|
||
{
|
||
"source": "external",
|
||
"label": label[:180],
|
||
"display_label": label[:180],
|
||
"url": url,
|
||
"target_url": url,
|
||
"score": max(float(item.authority_score or 0.0), float(item.freshness_score or 0.0), 0.7),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def _sources_used_from_external_evidence(evidence: list[ExternalEvidenceItem]) -> list[dict[str, Any]]:
|
||
out: list[dict[str, Any]] = []
|
||
for item in list(evidence or [])[:8]:
|
||
label = str(item.source_label or item.title or item.target_url or "外部公开来源").strip()
|
||
url = str(item.target_url or "").strip()
|
||
row = {"source": "external", "display_label": label[:180]}
|
||
if url:
|
||
row["target_url"] = url
|
||
out.append(row)
|
||
return out
|
||
|
||
|
||
__all__ = ["PMFlowAIEvidenceToolRegistry"]
|