"""Shared analysis packet builder for analysis-style replies.""" from __future__ import annotations import os from typing import Any from app.schemas.analysis_packets import ( AnalysisFocusObject, ExternalImpactAnalysisPacket, FollowupExplainAnalysisPacket, InternalExplainAnalysisPacket, ProjectFitAnalysisPacket, normalize_analysis_packet_payload, ) _ENV_ENABLE_ANALYSIS_PACKET_FAMILY = "PMFLOW_ENABLE_ANALYSIS_PACKET_FAMILY" def _env_bool(name: str, default: bool = False) -> bool: raw = str(os.getenv(name, "") or "").strip().lower() if not raw: return bool(default) if raw in {"1", "true", "yes", "on"}: return True if raw in {"0", "false", "no", "off"}: return False return bool(default) def _safe_dict(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, dict) else {} def _internal_packet_section(packet: Any, key: str) -> dict[str, Any]: payload = _safe_dict(packet) value = payload.get(key) return dict(value) if isinstance(value, dict) else {} def _normalize_text(value: Any, *, max_len: int = 240) -> str: text = " ".join(str(value or "").split()).strip() if not text: return "" return text[:max_len] def _normalize_object_id(value: Any) -> int | None: try: return int(value) if value is not None else None except Exception: return None def _normalize_string_list(value: Any, *, limit: int = 6, max_len: int = 120) -> list[str]: if not isinstance(value, list): return [] rows: list[str] = [] for item in value: text = _normalize_text(item, max_len=max_len) if text and text not in rows: rows.append(text) if len(rows) >= limit: break return rows def _normalize_preferred_names( preferred_names: list[Any] | None, *, limit: int = 6, max_len: int = 220, ) -> list[str]: if not isinstance(preferred_names, list): return [] rows: list[str] = [] for item in preferred_names: text = _normalize_text(item, max_len=max_len) if not text or text in rows: continue rows.append(text) if len(rows) >= limit: break return rows def _normalize_route_mode(value: Any) -> str: text = str(value or "").strip().lower() if text in {"internal", "external", "uncertain"}: return text return "uncertain" def _first_non_empty(*values: Any, max_len: int = 240) -> str: for value in values: text = _normalize_text(value, max_len=max_len) if text: return text return "" def _project_subject( *, project_snapshot: dict[str, Any], fit_brief: dict[str, Any], project_operating_context: dict[str, Any], ) -> str: return _first_non_empty( project_snapshot.get("project_name"), fit_brief.get("project_name"), fit_brief.get("project_summary"), project_operating_context.get("project_name"), project_operating_context.get("current_repo_scope"), max_len=120, ) def _normalize_entity_detail(value: Any) -> str: text = _normalize_text(value, max_len=220) if not text: return "" for marker in ("上下文：", "上下文:", "context:", "Context:"): if marker in text: text = text.split(marker, 1)[0].rstrip("。；;，, ") break return _normalize_text(text, max_len=220) def _iter_structured_snapshot_entities( project_snapshot: dict[str, Any], *, key: str, kind: str, ) -> list[dict[str, str]]: rows = project_snapshot.get(key) if not isinstance(rows, list): return [] normalized: list[dict[str, str]] = [] for item in rows[:6]: if not isinstance(item, dict): continue name = _first_non_empty( item.get("title"), item.get("name"), item.get("display_name"), max_len=120, ) if not name: continue detail = _first_non_empty( item.get("description"), item.get("summary"), item.get("reasoning"), item.get("content"), max_len=220, ) normalized.append( { "name": name, "kind": kind, "source": f"project_snapshot_{kind}", "object_id": _normalize_object_id(item.get("id") or item.get("object_id")), "detail": detail, "status": _normalize_text(item.get("status"), max_len=40), } ) return normalized def _build_entity_catalog( *, project_operating_context: dict[str, Any], fit_brief: dict[str, Any], external_recommendation: dict[str, Any], project_snapshot: dict[str, Any], ) -> list[dict[str, str]]: catalog: list[dict[str, str]] = [] def _append(name: Any, kind: str, source: str, *, detail: Any = None, status: Any = None) -> None: normalized = _normalize_text(name, max_len=120) if not normalized: return detail_payload = dict(detail) if isinstance(detail, dict) else {} detail_text = _normalize_entity_detail(detail_payload.get("text") if detail_payload else detail) status_text = _normalize_text(status, max_len=40) object_id = _normalize_object_id(detail_payload.get("id")) for existing in catalog: if existing["name"] != normalized: continue if detail_text and not existing.get("detail"): existing["detail"] = detail_text if status_text and not existing.get("status"): existing["status"] = status_text if object_id is not None and existing.get("object_id") is None: existing["object_id"] = object_id return catalog.append( { "name": normalized, "kind": kind, "source": source, "object_id": object_id, "detail": detail_text, "status": status_text, } ) structured_tasks = _iter_structured_snapshot_entities(project_snapshot, key="tasks", kind="task") structured_signals = _iter_structured_snapshot_entities(project_snapshot, key="signals", kind="signal") for item in structured_tasks: _append( item.get("name"), "task", str(item.get("source") or "project_snapshot_task"), detail={"text": item.get("detail"), "id": item.get("object_id")}, status=item.get("status"), ) for item in structured_signals: _append( item.get("name"), "signal", str(item.get("source") or "project_snapshot_signal"), detail={"text": item.get("detail"), "id": item.get("object_id")}, status=item.get("status"), ) for item in _normalize_string_list(project_snapshot.get("task_titles"), limit=6): _append(item, "task", "project_snapshot_task") for item in _normalize_string_list(project_snapshot.get("signal_titles"), limit=6): _append(item, "signal", "project_snapshot_signal") for item in _normalize_string_list(fit_brief.get("signal_source_anchors"), limit=6): _append(item, "signal", "fit_brief_anchor") for item in _normalize_string_list(fit_brief.get("current_capabilities"), limit=4): _append(item, "capability", "fit_brief") _append(project_snapshot.get("project_name"), "project", "project_snapshot") _append(fit_brief.get("project_summary"), "project", "fit_brief") _append(project_operating_context.get("current_repo_scope"), "repo", "project_operating_context") for item in _normalize_string_list(external_recommendation.get("affected_signal_sources"), limit=4): _append(item, "signal", "external_recommendation") return catalog def _entity_detail(entity: dict[str, str] | None) -> str: if not entity: return "" return _normalize_entity_detail(entity.get("detail")) def _trim_clause(text: str) -> str: return str(text or "").strip().rstrip("。；;，,") def _describe_signal_fact(name: str, detail: str, *, suffix: str) -> str: normalized_detail = _trim_clause(detail) if not normalized_detail: return suffix return f"{name}这条线索当前指向{normalized_detail}，{suffix}" def _entity_text(entity: dict[str, str] | None) -> str: if not entity: return "" return " ".join( part for part in ( _normalize_text(entity.get("name"), max_len=120), _entity_detail(entity), _normalize_text(entity.get("status"), max_len=40), _normalize_text(entity.get("source"), max_len=40), ) if part ).strip() def _pick_entity_by_object_ref( entity_catalog: list[dict[str, str]], *, object_ref: Any, used_names: set[str] | None = None, ) -> dict[str, str] | None: payload = dict(object_ref) if isinstance(object_ref, dict) else {} if not payload: return None used = used_names or set() target_id = _normalize_object_id(payload.get("id")) target_title = _normalize_text(payload.get("title"), max_len=120) target_kind = _normalize_text(payload.get("kind"), max_len=40) for item in entity_catalog: item_name = _normalize_text(item.get("name"), max_len=120) if not item_name or item_name in used: continue item_id = _normalize_object_id(item.get("object_id")) item_kind = _normalize_text(item.get("kind"), max_len=40) if target_id is not None and item_id == target_id and (not target_kind or target_kind == item_kind): return dict(item) if target_title: return _pick_entity_by_name(entity_catalog, name=target_title, used_names=used) return None def _project_snapshot_analysis_refs(project_snapshot: dict[str, Any]) -> dict[str, Any]: refs = project_snapshot.get("analysis_object_refs") return dict(refs) if isinstance(refs, dict) else {} def _iter_entities_by_kind( entity_catalog: list[dict[str, str]], *, kind: str, used_names: set[str] | None = None, ) -> list[dict[str, str]]: used = used_names or set() return [ dict(item) for item in entity_catalog if str(item.get("kind") or "").strip() == kind and str(item.get("name") or "").strip() and str(item.get("name") or "").strip() not in used ] def _pick_entity_by_name( entity_catalog: list[dict[str, str]], *, name: Any, used_names: set[str] | None = None, ) -> dict[str, str] | None: target = _normalize_text(name, max_len=120) if not target: return None used = used_names or set() target_key = target.lower() exact = None for item in entity_catalog: item_name = _normalize_text(item.get("name"), max_len=120) if not item_name or item_name in used: continue item_key = item_name.lower() if item_key == target_key: exact = dict(item) break return exact def _pick_entity_by_preferred_names( entity_catalog: list[dict[str, str]], *, preferred_names: list[str] | None = None, used_names: set[str] | None = None, ) -> dict[str, str] | None: preferred = _normalize_preferred_names(preferred_names) if not preferred: return None for raw_name in preferred: matched = _pick_entity_by_name(entity_catalog, name=raw_name, used_names=used_names) if matched is not None: return matched return None def _pick_entity_by_exact_text_binding( entity_catalog: list[dict[str, str]], *, texts: list[Any] | None = None, used_names: set[str] | None = None, ) -> dict[str, str] | None: normalized_texts = { _normalize_text(item, max_len=220).lower() for item in list(texts or []) if _normalize_text(item, max_len=220) } if not normalized_texts: return None used = used_names or set() for item in entity_catalog: item_name = _normalize_text(item.get("name"), max_len=220) if not item_name or item_name in used: continue detail = _entity_detail(item) if item_name.lower() in normalized_texts or detail.lower() in normalized_texts: return dict(item) return None def _entity_matches_object_ref(entity: dict[str, str] | None, object_ref: Any) -> bool: if not entity: return False payload = dict(object_ref) if isinstance(object_ref, dict) else {} if not payload: return False target_id = _normalize_object_id(payload.get("id")) target_title = _normalize_text(payload.get("title"), max_len=120) target_kind = _normalize_text(payload.get("kind"), max_len=40) entity_id = _normalize_object_id(entity.get("object_id")) entity_name = _normalize_text(entity.get("name"), max_len=120) entity_kind = _normalize_text(entity.get("kind"), max_len=40) if target_id is not None and entity_id == target_id and (not target_kind or target_kind == entity_kind): return True return bool(target_title and entity_name and target_title == entity_name and (not target_kind or target_kind == entity_kind)) def _task_entity_structural_support( entity: dict[str, str] | None, *, snapshot_refs: dict[str, Any], anchor_entity: dict[str, str] | None = None, ) -> float: if not entity or _normalize_text(entity.get("kind"), max_len=40) != "task": return 0.0 score = 0.0 entity_name = _normalize_text(entity.get("name"), max_len=120) if not entity_name: return 0.0 explicit_refs = ( snapshot_refs.get("fit_candidate_task_ref"), snapshot_refs.get("fit_rejected_task_ref"), snapshot_refs.get("primary_dependent_task_ref"), ) if any(_entity_matches_object_ref(entity, ref) for ref in explicit_refs): score += 8.0 _ = anchor_entity return score def _signal_entity_structural_support( entity: dict[str, str] | None, *, snapshot_refs: dict[str, Any], ) -> float: if not entity or _normalize_text(entity.get("kind"), max_len=40) != "signal": return 0.0 score = 0.0 explicit_refs = ( snapshot_refs.get("fit_candidate_signal_ref"), snapshot_refs.get("fit_rejected_signal_ref"), snapshot_refs.get("primary_anchor_signal_ref"), ) if any(_entity_matches_object_ref(entity, ref) for ref in explicit_refs): score += 8.0 return score def _pick_anchor_entity( entity_catalog: list[dict[str, str]], *, preferred_ref: Any = None, preferred_names: list[str] | None = None, used_names: set[str] | None = None, ) -> dict[str, str] | None: explicit_entity = _pick_entity_by_preferred_names( _iter_entities_by_kind(entity_catalog, kind="signal", used_names=used_names), preferred_names=preferred_names, used_names=used_names, ) if explicit_entity is not None: return explicit_entity preferred_entity = _pick_entity_by_object_ref( entity_catalog, object_ref=preferred_ref, used_names=used_names, ) if preferred_entity is not None: return preferred_entity signals = _iter_entities_by_kind(entity_catalog, kind="signal", used_names=used_names) if len(signals) == 1: return dict(signals[0]) return None def _pick_dependent_task( entity_catalog: list[dict[str, str]], *, anchor_entity: dict[str, str] | None = None, preferred_ref: Any = None, preferred_name: Any = None, used_names: set[str] | None = None, ) -> dict[str, str] | None: tasks = _iter_entities_by_kind(entity_catalog, kind="task", used_names=used_names) preferred = _pick_entity_by_object_ref(tasks, object_ref=preferred_ref, used_names=used_names) if preferred: return preferred preferred = _pick_entity_by_name(tasks, name=preferred_name, used_names=used_names) if preferred: return preferred if len(tasks) == 1: return dict(tasks[0]) if not tasks: signals = _iter_entities_by_kind(entity_catalog, kind="signal", used_names=used_names) preferred_signal = _pick_entity_by_object_ref(signals, object_ref=preferred_ref, used_names=used_names) if preferred_signal: return preferred_signal preferred_signal = _pick_entity_by_name(signals, name=preferred_name, used_names=used_names) if preferred_signal: return preferred_signal if len(signals) == 1: return dict(signals[0]) return None _ = anchor_entity return None def _pick_fit_contrast_tasks( entity_catalog: list[dict[str, str]], *, snapshot_refs: dict[str, Any] | None = None, preferred_ref: Any = None, rejected_ref: Any = None, used_names: set[str] | None = None, ) -> tuple[dict[str, str] | None, dict[str, str] | None]: tasks = _iter_entities_by_kind(entity_catalog, kind="task", used_names=used_names) if not tasks: signals = _iter_entities_by_kind(entity_catalog, kind="signal", used_names=used_names) if not signals: return None, None preferred_signal = _pick_entity_by_object_ref(signals, object_ref=snapshot_refs.get("fit_candidate_signal_ref") if snapshot_refs else None, used_names=used_names) rejected_signal = _pick_entity_by_object_ref( signals, object_ref=snapshot_refs.get("fit_rejected_signal_ref") if snapshot_refs else None, used_names={(preferred_signal.get("name") if preferred_signal else "")} | set(used_names or set()), ) if preferred_signal or rejected_signal: return preferred_signal, rejected_signal return None, None preferred = _pick_entity_by_object_ref(tasks, object_ref=preferred_ref, used_names=used_names) rejected = _pick_entity_by_object_ref( tasks, object_ref=rejected_ref, used_names={(preferred.get("name") if preferred else "")} | set(used_names or set()), ) if preferred or rejected: return preferred, rejected return None, None def _pick_best_entity( entity_catalog: list[dict[str, str]], *, role: str, used_names: set[str] | None = None, ) -> dict[str, str] | None: used = used_names or set() if role in {"preferred_fit", "rejected_fit"}: preferred, rejected = _pick_fit_contrast_tasks(entity_catalog, used_names=used) return preferred if role == "preferred_fit" else rejected if role in {"blocked", "anchor"}: return _pick_anchor_entity(entity_catalog, used_names=used) if role == "dependent": return _pick_dependent_task(entity_catalog, used_names=used) candidates = [ dict(item) for item in entity_catalog if str(item.get("name") or "").strip() and str(item.get("name") or "").strip() not in used ] if len(candidates) == 1: return candidates[0] return None def _build_focus_object( *, entity: dict[str, str] | None, stance: str, reason: str, ) -> AnalysisFocusObject | None: if not entity: return None name = _normalize_text(entity.get("name"), max_len=120) if not name: return None normalized_reason = _normalize_text(reason, max_len=160) if not normalized_reason: return None return AnalysisFocusObject( name=name, kind=str(entity.get("kind") or "unknown").strip() or "unknown", stance=stance, # type: ignore[arg-type] reason=normalized_reason, ) def _build_project_fit_packet( *, fit_brief: dict[str, Any], external_recommendation: dict[str, Any], recommended_next_step: str, entity_catalog: list[dict[str, str]], project_subject: str, project_snapshot: dict[str, Any], ) -> dict[str, Any]: retrieval_status = str(external_recommendation.get("retrieval_status") or "").strip().lower() impact_status = str(external_recommendation.get("impact_status") or "").strip().lower() used_names: set[str] = set() focus_objects: list[AnalysisFocusObject] = [] def _append(entity: dict[str, str] | None, stance: str, reason: str) -> None: focus = _build_focus_object(entity=entity, stance=stance, reason=reason) if focus is None or focus.name in used_names or len(focus_objects) >= 2: return focus_objects.append(focus) used_names.add(focus.name) fit_verdict = "uncertain" current_subject = _normalize_text(project_subject, max_len=120) or None preferred_layer: str | None = None rejected_layer: str | None = None candidate_position: str | None = None reason_chain: list[str] = [] snapshot_refs = _project_snapshot_analysis_refs(project_snapshot) project_description = _normalize_text(project_snapshot.get("project_description"), max_len=180) human_next_step = "" suppress_generic_fallback_focus = False if retrieval_status == "evidence_found" and impact_status == "direct_match": anchor = _pick_anchor_entity( entity_catalog, preferred_ref=snapshot_refs.get("primary_anchor_signal_ref"), preferred_names=_normalize_string_list(external_recommendation.get("affected_signal_sources"), limit=4), used_names=used_names, ) dependent = _pick_dependent_task( entity_catalog, anchor_entity=anchor, preferred_ref=snapshot_refs.get("primary_dependent_task_ref"), used_names=used_names, ) _append(dependent, "dependent", "如果这条变化真有影响，最容易先落到这一步。") _append(anchor, "anchor", "这是当前能直接对上的外部来源。") fit_verdict = "fit_now" candidate_position = dependent.get("name") if dependent else anchor.get("name") if anchor else None preferred_layer = candidate_position anchor_name = _first_non_empty(anchor.get("name") if anchor else None, "当前正式锚点", max_len=120) dependent_name = _first_non_empty(dependent.get("name") if dependent else None, anchor_name, max_len=120) anchor_detail = _trim_clause(_entity_detail(anchor)) dependent_detail = _trim_clause(_entity_detail(dependent)) conclusion = ( f"放到{current_subject}里，这条外部变化已经直接打到{dependent_name}这一步。" if current_subject else f"这条外部变化已经直接打到{dependent_name}这一步。" ) reason_chain = [ ( f"{dependent_name}负责{dependent_detail}，会最先接住这条变化。" if dependent_detail else f"{dependent_name}是当前最直接会接住这条变化的项目对象。" ), ( f"对应的正式锚点是{anchor_name}，它指向{anchor_detail}，已经和外部证据建立了直接映射。" if anchor_detail else f"{anchor_name}已经和外部证据建立了直接映射。" ), ] human_next_step = f"先看这条变化会不会直接改到{dependent_name}这一步的下游动作。" missing_mapping = _first_non_empty( external_recommendation.get("external_action"), f"还缺它具体作用到{dependent_name}这一步哪个字段或触发条件的明确信息。", max_len=220, ) elif retrieval_status == "verifier_blocked": blocked = _pick_anchor_entity( entity_catalog, preferred_ref=snapshot_refs.get("primary_anchor_signal_ref"), preferred_names=_normalize_string_list(external_recommendation.get("affected_signal_sources"), limit=4), used_names=used_names, ) dependent = _pick_dependent_task( entity_catalog, anchor_entity=blocked, preferred_ref=snapshot_refs.get("primary_dependent_task_ref"), used_names=used_names, ) _append(blocked, "blocked", "这条线索还没核实清楚，暂时不能直接拿来下结论。") _append(dependent, "dependent", "如果后面真要用这条变化，还得先把字段和证据对上。") fit_verdict = "uncertain" blocked_name = _first_non_empty(blocked.get("name") if blocked else None, "这条线索", max_len=120) dependent_name = _first_non_empty(dependent.get("name") if dependent else None, "当前项目动作", max_len=120) blocked_detail = _trim_clause(_entity_detail(blocked)) dependent_detail = _trim_clause(_entity_detail(dependent)) conclusion = ( f"放到{current_subject}里，这条外部变化现在还不能直接挂项目结论。" if current_subject else "这条外部变化现在还不能直接挂项目结论。" ) if _task_entity_structural_support(dependent, snapshot_refs=snapshot_refs, anchor_entity=blocked) <= 0: reason_chain = [ ( f"现在卡在{blocked_name}这条线索上：它指向{blocked_detail}，但还没通过校验。" if blocked_detail else f"{blocked_name}这条线索还没通过校验，当前不能直接挂到项目结论上。" ), "项目里现在更多还是实现层任务，我不想硬说它会直接改到哪一步，先把这条线索核实清楚更稳。", ] human_next_step = f"先把{blocked_name}这条线索核实清楚，再确认它到底会影响这个项目里的哪个结果。" else: reason_chain = [ ( f"现在卡在{blocked_name}这条线索上：它指向{blocked_detail}，但还没通过校验。" if blocked_detail else f"{blocked_name}这条线索还没通过校验，当前不能直接挂到项目结论上。" ), ( f"如果后面真要让{dependent_name}用这条变化，还得先把对应字段和证据核实清楚。" if dependent_detail else f"{dependent_name}如果要用这条变化，还缺能对上的字段和证据。" ), ] human_next_step = f"先把{blocked_name}这条线索核实清楚，再判断要不要让{dependent_name}用它。" missing_mapping = human_next_step else: preferred, rejected = _pick_fit_contrast_tasks( entity_catalog, snapshot_refs=snapshot_refs, preferred_ref=snapshot_refs.get("fit_candidate_task_ref") or snapshot_refs.get("fit_candidate_signal_ref"), rejected_ref=snapshot_refs.get("fit_rejected_task_ref") or snapshot_refs.get("fit_rejected_signal_ref"), used_names=used_names, ) _append(preferred, "better_fit", "更适合先在这一步试，看看它到底值不值。") _append(rejected, "not_fit", "这一步要先稳定跑通，不适合一上来就改。") fit_verdict = "fit_later" if impact_status == "indirect_only" else "not_fit_yet" candidate_position = preferred.get("name") if preferred else None preferred_layer = preferred.get("name") if preferred else None rejected_layer = rejected.get("name") if rejected else None preferred_name = _first_non_empty(preferred.get("name") if preferred else None, "结果出来之后的解释层", max_len=120) rejected_name = _first_non_empty(rejected.get("name") if rejected else None, "当前主链", max_len=120) preferred_detail = _trim_clause(_entity_detail(preferred)) rejected_detail = _trim_clause(_entity_detail(rejected)) preferred_kind = str(preferred.get("kind") or "").strip() if preferred else "" rejected_kind = str(rejected.get("kind") or "").strip() if rejected else "" if preferred is None or rejected is None: conclusion = ( f"光看{current_subject}现在这些项目信息，我还不想直接说它该放哪一步。" if current_subject else "光看现在这些项目信息，我还不想直接说它该放哪一步。" ) if project_description: normalized_description = _trim_clause(project_description) reason_chain = [ f"我现在只看得到这个项目的大致目标：{normalized_description}。", "但还看不到这条链路里谁来触发、要接哪些外部输入、以及哪个结果算真的跑通。", ] human_next_step = "至少先把实际流程拆出来：谁发起、系统接什么输入、最后产出什么结果。" missing_mapping = "还缺能对上的实际流程、关键结果和外部输入。" else: reason_chain = [ "现在项目里还没有足够具体的任务或信号源，硬把它挂到某个位置上，结论很容易说偏。" ] human_next_step = "先把这个项目最关键的业务步骤补清楚，我再判断它更适合放哪里。" missing_mapping = "至少还缺一个能对上的业务步骤或信号源。" suppress_generic_fallback_focus = True elif ( preferred_kind == "task" and rejected_kind == "task" and max( _task_entity_structural_support(preferred, snapshot_refs=snapshot_refs), _task_entity_structural_support(rejected, snapshot_refs=snapshot_refs), ) <= 0 ): conclusion = ( f"光看{current_subject}现在这批任务，我还不想硬说它该放哪一步。" if current_subject else "光看现在这批任务，我还不想硬说它该放哪一步。" ) reason_chain = [ f"现在拆出来的主要还是实现层任务，比如{preferred_name}、{rejected_name}，还看不出它到底该先影响哪个业务结果。" ] human_next_step = "先定清它到底想改善哪个业务结果，再看该落到哪一步。" missing_mapping = "至少还缺一个明确的业务结果或业务动作。" suppress_generic_fallback_focus = True elif ( preferred_kind == "signal" and rejected_kind == "signal" and max( _signal_entity_structural_support(preferred, snapshot_refs=snapshot_refs), _signal_entity_structural_support(rejected, snapshot_refs=snapshot_refs), ) <= 0 ): conclusion = ( f"光看{current_subject}现在这些信号源占位，我还不想直接说它该放哪一步。" if current_subject else "光看现在这些信号源占位，我还不想直接说它该放哪一步。" ) reason_chain = [ "现在写出来的还是“增加某个信号源”这类占位，不是真正的业务步骤，硬判很容易说偏。" ] human_next_step = "先定清这个项目最后要给谁看什么结果，再判断哪个信号只是输入、哪个位置更值得先试。" missing_mapping = "至少还缺一个能落到业务动作上的对象。" suppress_generic_fallback_focus = True elif preferred_kind == "signal" or rejected_kind == "signal": conclusion = ( f"放到{current_subject}里，我更倾向先让它去帮{preferred_name}，不太会先拿它改{rejected_name}。" if current_subject else f"我更倾向先让它去帮{preferred_name}，不太会先拿它改{rejected_name}。" ) reason_chain = [ ( f"{rejected_name}现在代表的是{rejected_detail}，这类前面必须先跑通的条件，不适合直接拿来试新能力。" if rejected_detail else f"{rejected_name}更像前面必须先跑通的条件，不适合直接拿来试新能力。" ), ( f"{preferred_name}现在代表的是{preferred_detail}，先从这里试更稳，也更容易看出它有没有价值。" if preferred_detail else f"{preferred_name}更像用户最后会看到的结果，先从这里试更稳。" ), ] human_next_step = f"先定清它到底是给{preferred_name}补说明，还是会直接改到{rejected_name}。" missing_mapping = f"现在还说不准它会先帮到{preferred_name}，还是会直接改到{rejected_name}。" else: conclusion = ( f"如果真要放进{current_subject}，我会先放在{preferred_name}这一步，不太会直接改{rejected_name}这条主流程。" if current_subject else f"如果真要放进去，我会先放在{preferred_name}这一步，不太会直接改{rejected_name}这条主流程。" ) reason_chain = [ ( f"{rejected_name}现在做的是{rejected_detail}，这一步要稳，不能随便把新能力塞进去。" if rejected_detail else f"{rejected_name}这一步要稳，不适合一上来就塞新能力。" ), ( f"{preferred_name}现在做的是{preferred_detail}，先在这里试更稳，也更容易看出值不值。" if preferred_detail else f"{preferred_name}这一步更适合先试，比直接改{rejected_name}风险小。" ), ] human_next_step = f"先定清楚它是给{preferred_name}补说明，还是会直接改到{rejected_name}，再决定要不要继续做。" missing_mapping = f"现在还说不准它会先帮到{preferred_name}，还是会直接改到{rejected_name}。" if not focus_objects and not suppress_generic_fallback_focus: fallback = _pick_best_entity(entity_catalog, role="relevant", used_names=used_names) _append(fallback, "relevant", "这是当前最直接相关的项目对象。") packet = ProjectFitAnalysisPacket( conclusion=conclusion, focus_objects=focus_objects, next_step=_normalize_text(human_next_step, max_len=180) or _normalize_text(recommended_next_step, max_len=180) or None, fit_verdict=fit_verdict, # type: ignore[arg-type] current_subject=current_subject, candidate_position=candidate_position, rejected_position=rejected_layer, preferred_layer=preferred_layer, rejected_layer=rejected_layer, reason_chain=[_normalize_text(item, max_len=180) for item in reason_chain if _normalize_text(item, max_len=180)], missing_mapping=missing_mapping, ) return packet.model_dump() def _build_external_impact_packet( *, external_recommendation: dict[str, Any], recommended_next_step: str, entity_catalog: list[dict[str, str]], project_subject: str, project_snapshot: dict[str, Any], ) -> dict[str, Any]: retrieval_status = str(external_recommendation.get("retrieval_status") or "").strip().lower() or "retrieval_failed" impact_status = str(external_recommendation.get("impact_status") or "").strip().lower() or "no_direct_impact" used_names: set[str] = set() focus_objects: list[AnalysisFocusObject] = [] current_subject = _normalize_text(project_subject, max_len=120) or None reason_chain: list[str] = [] snapshot_refs = _project_snapshot_analysis_refs(project_snapshot) human_next_step = "" def _append(entity: dict[str, str] | None, stance: str, reason: str) -> None: focus = _build_focus_object(entity=entity, stance=stance, reason=reason) if focus is None or focus.name in used_names or len(focus_objects) >= 2: return focus_objects.append(focus) used_names.add(focus.name) if retrieval_status == "verifier_blocked": blocked = _pick_anchor_entity( entity_catalog, preferred_ref=snapshot_refs.get("primary_anchor_signal_ref"), preferred_names=_normalize_string_list(external_recommendation.get("affected_signal_sources"), limit=4), used_names=used_names, ) dependent = _pick_dependent_task( entity_catalog, anchor_entity=blocked, preferred_ref=snapshot_refs.get("primary_dependent_task_ref"), used_names=used_names, ) _append(blocked, "blocked", "这条线索还没核实清楚，不能直接拿来下结论。") _append(dependent, "dependent", "如果后面真要用这条变化，这一步会最先受影响。") blocked_name = _first_non_empty(blocked.get("name") if blocked else None, "这条线索", max_len=120) dependent_name = _first_non_empty(dependent.get("name") if dependent else None, "当前项目动作", max_len=120) blocked_detail = _trim_clause(_entity_detail(blocked)) dependent_detail = _trim_clause(_entity_detail(dependent)) conclusion = ( f"这条外部变化现在还不能直接挂到{current_subject}的项目结论上。" if current_subject else "这条外部变化现在还不能直接挂到当前项目结论上。" ) if _task_entity_structural_support(dependent, snapshot_refs=snapshot_refs, anchor_entity=blocked) <= 0: reason_chain = [ ( f"现在卡在{blocked_name}这条线索上：它指向{blocked_detail}，但还没通过校验。" if blocked_detail else f"{blocked_name}这条线索还没通过校验，当前不能直接挂到项目结论上。" ), "项目里现在更多还是实现层任务，我不想硬说它会直接改到哪一步，先把这条线索核实清楚更稳。", ] human_next_step = f"先把{blocked_name}这条线索核实清楚，再确认它到底会影响这个项目里的哪个结果。" else: reason_chain = [ ( f"现在卡在{blocked_name}这条线索上：它指向{blocked_detail}，但还没通过校验。" if blocked_detail else f"{blocked_name}这条线索还没通过校验，当前不能直接挂到项目结论上。" ), ( f"如果后面真要让{dependent_name}用这条变化，还得先把对应字段和证据核实清楚。" if dependent_detail else f"{dependent_name}如果要用这条变化，还缺能对上的字段和证据。" ), ] human_next_step = f"先把{blocked_name}这条线索核实清楚，再判断要不要让{dependent_name}用它。" elif retrieval_status == "evidence_found" and impact_status == "direct_match": anchor = _pick_anchor_entity( entity_catalog, preferred_ref=snapshot_refs.get("primary_anchor_signal_ref"), preferred_names=_normalize_string_list(external_recommendation.get("affected_signal_sources"), limit=4), used_names=used_names, ) dependent = _pick_dependent_task( entity_catalog, anchor_entity=anchor, preferred_ref=snapshot_refs.get("primary_dependent_task_ref"), used_names=used_names, ) _append(dependent, "dependent", "如果这条变化真有影响，最容易先落到这一步。") _append(anchor, "anchor", "这是当前能直接对上的外部来源。") anchor_name = _first_non_empty(anchor.get("name") if anchor else None, "当前正式锚点", max_len=120) dependent_name = _first_non_empty(dependent.get("name") if dependent else None, anchor_name, max_len=120) anchor_detail = _trim_clause(_entity_detail(anchor)) dependent_detail = _trim_clause(_entity_detail(dependent)) if _task_entity_structural_support(dependent, snapshot_refs=snapshot_refs, anchor_entity=anchor) <= 0: conclusion = ( f"这条外部变化已经和{current_subject}对上了，但现在还不适合硬挂到某个实现任务上。" if current_subject else "这条外部变化已经对上了，但现在还不适合硬挂到某个实现任务上。" ) reason_chain = [ ( f"能对上的外部来源是{anchor_name}，它指向{anchor_detail}，所以这次不是泛泛相关。" if anchor_detail else f"能对上的外部来源是{anchor_name}，所以这次不是泛泛相关。" ), "项目里现在更多还是实现层任务，先别急着说它会直接改到哪一步，先看它到底影响哪个业务结果。", ] human_next_step = "先确认这条变化会影响哪个业务结果，再决定要不要拆到具体任务。" else: conclusion = ( f"这条外部变化已经直接打到{current_subject}里的{dependent_name}这一步。" if current_subject else f"这条外部变化已经直接打到{dependent_name}这一步。" ) reason_chain = [ ( f"{dependent_name}负责{dependent_detail}，会最先接住这条变化，可以当成后续收口和计划讨论的正式输入。" if dependent_detail else f"{dependent_name}会最先接住这条变化，可以当成后续收口和计划讨论的正式输入。" ), ( f"能对上的外部来源是{anchor_name}，它指向{anchor_detail}，所以这次不是泛泛相关。" if anchor_detail else f"能对上的外部来源是{anchor_name}，所以这次不是泛泛相关。" ), ] human_next_step = f"下一步就看这条变化会不会直接改到{dependent_name}后面的动作或规则。" elif retrieval_status == "evidence_found" and impact_status == "indirect_only": preferred, rejected = _pick_fit_contrast_tasks( entity_catalog, snapshot_refs=snapshot_refs, preferred_ref=snapshot_refs.get("fit_candidate_task_ref") or snapshot_refs.get("fit_candidate_signal_ref"), rejected_ref=snapshot_refs.get("fit_rejected_task_ref") or snapshot_refs.get("fit_rejected_signal_ref"), used_names=used_names, ) _append(preferred, "better_fit", "如果继续看，这一步更适合先试。") _append(rejected, "not_fit", "这一步现在还得按稳定流程跑。") preferred_name = _first_non_empty(preferred.get("name") if preferred else None, "解释层", max_len=120) rejected_name = _first_non_empty(rejected.get("name") if rejected else None, "当前主链", max_len=120) preferred_detail = _trim_clause(_entity_detail(preferred)) rejected_detail = _trim_clause(_entity_detail(rejected)) conclusion = ( f"这条外部变化现在只形成背景相关，还没真正落到{current_subject}的正式对象上。" if current_subject else "这条外部变化现在只形成背景相关，还没真正落到当前项目的正式对象上。" ) if ( max( _task_entity_structural_support(preferred, snapshot_refs=snapshot_refs), _task_entity_structural_support(rejected, snapshot_refs=snapshot_refs), ) <= 0 ): reason_chain = [ f"项目里现在更多还是实现层任务，比如{preferred_name}、{rejected_name}，还看不出这条外部变化会先影响哪个业务结果。" ] human_next_step = "先看这条外部变化到底影响哪个业务结果，再决定要不要拆到具体任务。" else: reason_chain = [ ( f"{rejected_name}现在做的是{rejected_detail}，这一步还得按现在的稳定流程跑，不能直接把这条外部变化挂进去。" if rejected_detail else f"{rejected_name}这一步还得按现在的稳定流程跑，不能直接把这条外部变化挂进去。" ), ( f"如果继续看，我会先从{preferred_name}这一步试，因为{preferred_detail or f'这里比直接改{rejected_name}更容易先试出价值'}。" if preferred_detail else f"如果继续看，我会先从{preferred_name}这一步试，不会先去动{rejected_name}。" ), ] human_next_step = f"如果继续看，先看{preferred_name}能不能先用上，再决定要不要动{rejected_name}。" elif retrieval_status == "no_relevant_results": _append(_pick_best_entity(entity_catalog, role="relevant", used_names=used_names), "relevant", "当前只找到最接近的项目对象，但还没有外部证据真正落上来。") conclusion = ( f"目前还没拿到与{current_subject}直接相关的外部证据。" if current_subject else "目前还没拿到与当前项目直接相关的外部证据。" ) reason_chain = ["当前只找到最接近的项目对象，但还没有外部证据真正落上来。"] else: _append(_pick_best_entity(entity_catalog, role="relevant", used_names=used_names), "relevant", "当前只保留最接近的项目对象，等待后续补证。") conclusion = "这次外部检索没有形成可直接消费的项目结论。" reason_chain = ["当前只保留最接近的项目对象，等待后续补证。"] packet = ExternalImpactAnalysisPacket( conclusion=conclusion, focus_objects=focus_objects, next_step=_normalize_text(human_next_step, max_len=180) or _normalize_text(recommended_next_step, max_len=180) or None, retrieval_status=retrieval_status, # type: ignore[arg-type] impact_status=impact_status if impact_status in {"direct_match", "indirect_only", "no_direct_impact"} else "no_direct_impact", # type: ignore[arg-type] current_subject=current_subject, reason_chain=[_normalize_text(item, max_len=180) for item in reason_chain if _normalize_text(item, max_len=180)], anchor_objects=_normalize_string_list(external_recommendation.get("affected_signal_sources"), limit=4), blocked_reason_or_gap=_first_non_empty( human_next_step, recommended_next_step, external_recommendation.get("external_action"), external_recommendation.get("external_impact"), external_recommendation.get("external_summary"), "当前还缺能把外部变化稳定映射到项目对象上的证据。", max_len=220, ), ) return packet.model_dump() def _build_internal_explain_packet( *, answer: str, query_type: str, evidence_pack: dict[str, Any], internal_context_packet: Any, entity_catalog: list[dict[str, str]], ) -> dict[str, Any]: operation_contract = _internal_packet_section(internal_context_packet, "operation_contract") query_plan = _internal_packet_section(internal_context_packet, "query_plan") target_resolution = _internal_packet_section(internal_context_packet, "target_resolution") answer_contract = _first_non_empty( query_plan.get("answer_contract"), operation_contract.get("answer_contract"), query_type, max_len=80, ) resolved_target = _first_non_empty( target_resolution.get("primary_target_label"), target_resolution.get("target_label"), target_resolution.get("matched_target_label"), target_resolution.get("target_name"), max_len=120, ) if not resolved_target: matched_entity = _pick_entity_by_exact_text_binding( entity_catalog, texts=[answer, _safe_dict(evidence_pack.get("summary")).get("conclusion")], ) resolved_target = _normalize_text((matched_entity or {}).get("name"), max_len=120) uncertainties = _normalize_string_list(_safe_dict(evidence_pack.get("summary")).get("uncertainties"), limit=2, max_len=120) conclusion = _first_non_empty( _safe_dict(evidence_pack.get("summary")).get("conclusion"), answer, "当前内部对象已经拿到可解释结果。", max_len=220, ) current_subject = resolved_target or None reason_chain: list[str] = [] focus_objects: list[AnalysisFocusObject] = [] if resolved_target: focus_objects.append( AnalysisFocusObject( name=resolved_target, kind="internal_target", stance="relevant", reason="这次回答说的就是这个对象。", ) ) if resolved_target and answer_contract in {"status_summary", "status"}: reason_chain.append(f"这次就按{resolved_target}的当前情况回答，先不展开方案。") elif resolved_target: reason_chain.append(f"这次先把{resolved_target}现在是什么情况说清楚。") elif answer_contract: reason_chain.append("这次先直接回答当前情况，不往方案上展开。") packet = InternalExplainAnalysisPacket( conclusion=conclusion, focus_objects=focus_objects, answer_contract=answer_contract or None, current_subject=current_subject, reason_chain=[_normalize_text(item, max_len=180) for item in reason_chain if _normalize_text(item, max_len=180)], resolved_target=resolved_target or None, evidence_gap_or_uncertainty=uncertainties[0] if uncertainties else None, ) return packet.model_dump() def _build_followup_explain_packet( *, answer: str, evidence_pack: dict[str, Any], clarification_code: str, anchor_kind: str, meta_followup_kind: str, ) -> dict[str, Any]: summary = _safe_dict(evidence_pack.get("summary")) uncertainties = _normalize_string_list(summary.get("uncertainties"), limit=2, max_len=120) packet = FollowupExplainAnalysisPacket( conclusion=_first_non_empty( summary.get("conclusion"), answer, "当前回复依赖上一轮上下文。", max_len=220, ), focus_objects=[], anchor_kind=anchor_kind or None, continuation_kind=meta_followup_kind or None, clarification_state=clarification_code or None, missing_mapping=uncertainties[0] if uncertainties else None, ) return packet.model_dump() def build_analysis_packet( *, route_decision: Any = None, project_operating_context: Any = None, execution_recipe: Any = None, fit_brief: Any = None, external_recommendation: Any = None, recommended_next_step: Any = None, answer: Any = None, evidence_pack: Any = None, query_type: Any = None, clarification_code: Any = None, anchor_kind: Any = None, meta_followup_kind: Any = None, internal_context_packet: Any = None, project_snapshot: Any = None, ) -> dict[str, Any]: if not _env_bool(_ENV_ENABLE_ANALYSIS_PACKET_FAMILY, True): return {} normalized_route = _safe_dict(route_decision) normalized_project_context = _safe_dict(project_operating_context) normalized_recipe = _safe_dict(execution_recipe) normalized_fit_brief = _safe_dict(fit_brief) normalized_external = _safe_dict(external_recommendation) normalized_evidence_pack = _safe_dict(evidence_pack) normalized_project_snapshot = _safe_dict(project_snapshot) recommended_next = _normalize_text(recommended_next_step, max_len=180) normalized_query_type = str(query_type or "").strip().lower() normalized_clarification_code = str(clarification_code or "").strip().lower() normalized_anchor_kind = str(anchor_kind or "").strip().lower() normalized_followup_kind = str(meta_followup_kind or "").strip().lower() project_subject = _project_subject( project_snapshot=normalized_project_snapshot, fit_brief=normalized_fit_brief, project_operating_context=normalized_project_context, ) if normalized_clarification_code or normalized_anchor_kind or normalized_followup_kind: return _build_followup_explain_packet( answer=_normalize_text(answer, max_len=220), evidence_pack=normalized_evidence_pack, clarification_code=normalized_clarification_code, anchor_kind=normalized_anchor_kind, meta_followup_kind=normalized_followup_kind, ) effective_source_policy = _normalize_route_mode( normalized_route.get("effective_source_policy") or normalized_route.get("route_mode") ) entity_catalog = _build_entity_catalog( project_operating_context=normalized_project_context, fit_brief=normalized_fit_brief, external_recommendation=normalized_external, project_snapshot=normalized_project_snapshot, ) if effective_source_policy == "internal": return _build_internal_explain_packet( answer=_normalize_text(answer, max_len=220), query_type=normalized_query_type, evidence_pack=normalized_evidence_pack, internal_context_packet=internal_context_packet, entity_catalog=entity_catalog, ) if effective_source_policy == "external" and bool(normalized_recipe.get("fit_brief_required")): return _build_project_fit_packet( fit_brief=normalized_fit_brief, external_recommendation=normalized_external, recommended_next_step=recommended_next, entity_catalog=entity_catalog, project_subject=project_subject, project_snapshot=normalized_project_snapshot, ) if effective_source_policy == "external" and normalized_external: return _build_external_impact_packet( external_recommendation=normalized_external, recommended_next_step=recommended_next, entity_catalog=entity_catalog, project_subject=project_subject, project_snapshot=normalized_project_snapshot, ) return {} __all__ = ["build_analysis_packet", "normalize_analysis_packet_payload"]