454 lines
18 KiB
Python
454 lines
18 KiB
Python
"""Context evidence absorption for PMFlow AI.
|
||
|
||
This module turns already-selected thread, memory and topic context into
|
||
answer-grade obligations. It is deliberately based on source role, recency,
|
||
scope and generic semantic overlap rather than business-scene keyword routing.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
from typing import Any
|
||
|
||
from ..schemas.pmflow_ai_agent_runtime import (
|
||
PMFlowAIAgentBoundary,
|
||
PMFlowAIAgentClaim,
|
||
PMFlowAIAgentWorkspace,
|
||
clean_text,
|
||
)
|
||
|
||
|
||
CONTEXT_EVIDENCE_PACKET_VERSION = "pmflow_ai_context_evidence_packet.v1"
|
||
TOPIC_WORKSPACE_PACKET_VERSION = "pmflow_ai_topic_workspace_packet.v1"
|
||
|
||
|
||
def build_context_evidence_packet(
|
||
*,
|
||
context_pack: dict[str, Any] | None,
|
||
user_question: str,
|
||
) -> dict[str, Any]:
|
||
"""Build customer-safe context claims and obligations from a context pack."""
|
||
|
||
pack = context_pack if isinstance(context_pack, dict) else {}
|
||
question = clean_text(user_question, limit=1200)
|
||
rows = _context_rows(pack)
|
||
scored = sorted(
|
||
rows,
|
||
key=lambda row: (
|
||
-_row_score(row=row, question=question),
|
||
_source_rank(row.get("source_kind")),
|
||
-int(row.get("message_index") or 0),
|
||
),
|
||
)
|
||
claims: list[dict[str, Any]] = []
|
||
for index, row in enumerate(scored[:16], start=1):
|
||
text = clean_text(row.get("text"), limit=1000)
|
||
if not text:
|
||
continue
|
||
role = _semantic_role(row)
|
||
claim_key = f"{row.get('source_ref')}:{text[:160]}"
|
||
claim_id = f"ctxclaim_{hashlib.sha1(claim_key.encode('utf-8')).hexdigest()[:12]}"
|
||
score = _row_score(row=row, question=question)
|
||
claims.append(
|
||
{
|
||
"claim_id": claim_id,
|
||
"semantic_role": role,
|
||
"claim_text": text,
|
||
"customer_value": _customer_value_for_role(role),
|
||
"source_kind": str(row.get("source_kind") or "")[:80],
|
||
"source_role": str(row.get("source_role") or "")[:40],
|
||
"source_ref": str(row.get("source_ref") or "")[:160],
|
||
"support_refs": [str(row.get("source_ref") or "")[:160]] if row.get("source_ref") else [],
|
||
"relevance_score": round(float(score), 4),
|
||
"required_for_answer": bool(index <= 6 and score > 0),
|
||
"confidence": row.get("confidence") or ("high" if row.get("source_kind") == "thread_message" else "medium"),
|
||
}
|
||
)
|
||
if claims and not any(item.get("required_for_answer") for item in claims):
|
||
claims[0]["required_for_answer"] = True
|
||
obligations = [
|
||
{
|
||
"obligation_id": f"ctxobl_{item['claim_id'].replace('ctxclaim_', '')}",
|
||
"claim_id": item["claim_id"],
|
||
"semantic_role": item["semantic_role"],
|
||
"requirement": "The final answer must preserve this prior context unless it explicitly marks a narrower missing artifact boundary.",
|
||
"source_ref": item.get("source_ref"),
|
||
}
|
||
for item in claims
|
||
if item.get("required_for_answer")
|
||
]
|
||
topic = build_topic_workspace_packet(
|
||
user_question=question,
|
||
context_claims=claims,
|
||
context_pack=pack,
|
||
)
|
||
conflicts = _context_conflict_boundaries(pack=pack, context_claims=claims)
|
||
return {
|
||
"packet_version": CONTEXT_EVIDENCE_PACKET_VERSION,
|
||
"current_user_message": question,
|
||
"claims": claims,
|
||
"continuation_obligations": obligations[:8],
|
||
"conflict_boundaries": conflicts[:8],
|
||
"topic_workspace": topic,
|
||
"policy": {
|
||
"context_claims_are_answer_grade_evidence": True,
|
||
"tool_empty_results_cannot_negate_prior_context": True,
|
||
"visible_objects_are_candidates_until_explicitly_bound": True,
|
||
"business_scene_keyword_routing": False,
|
||
},
|
||
}
|
||
|
||
|
||
def append_context_evidence_to_workspace(
|
||
*,
|
||
workspace: PMFlowAIAgentWorkspace,
|
||
evidence_packet: dict[str, Any] | None,
|
||
) -> int:
|
||
"""Promote context evidence claims into the Agent Workspace claim ledger."""
|
||
|
||
packet = evidence_packet if isinstance(evidence_packet, dict) else {}
|
||
rows = list(packet.get("claims") or [])
|
||
if not rows:
|
||
return 0
|
||
existing_refs = {
|
||
ref
|
||
for claim in list(getattr(workspace, "claim_ledger", []) or [])
|
||
for ref in list(getattr(claim, "support_refs", []) or [])
|
||
if ref
|
||
}
|
||
claims: list[PMFlowAIAgentClaim] = []
|
||
for row in rows[:16]:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
text = clean_text(row.get("claim_text"), limit=1000)
|
||
source_ref = clean_text(row.get("source_ref"), limit=160)
|
||
if not text or (source_ref and source_ref in existing_refs):
|
||
continue
|
||
semantic_role = clean_text(row.get("semantic_role"), limit=80) or "context_evidence"
|
||
claims.append(
|
||
PMFlowAIAgentClaim(
|
||
claim_id=clean_text(row.get("claim_id"), limit=80) or "",
|
||
claim_type=f"context_{semantic_role}"[:80],
|
||
claim_text=text,
|
||
customer_value=clean_text(row.get("customer_value"), limit=500),
|
||
support_refs=list(row.get("support_refs") or ([source_ref] if source_ref else []))[:12],
|
||
source_tool="context_evidence_absorber",
|
||
source_kind=clean_text(row.get("source_kind"), limit=120),
|
||
relevance_to_user_goal="direct" if bool(row.get("required_for_answer")) else "partial",
|
||
target_match="direct" if bool(row.get("required_for_answer")) else "partial",
|
||
time_match="recent" if row.get("source_kind") == "thread_message" else "unknown",
|
||
authority_level="primary" if row.get("source_kind") == "thread_message" else "credible_secondary",
|
||
claim_kind="substantive_fact",
|
||
confidence=_claim_confidence(row),
|
||
allowed_to_state=True,
|
||
)
|
||
)
|
||
boundaries = []
|
||
for row in list(packet.get("conflict_boundaries") or [])[:8]:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
reason = clean_text(row.get("customer_safe_reason"), limit=800)
|
||
if not reason:
|
||
continue
|
||
boundaries.append(
|
||
PMFlowAIAgentBoundary(
|
||
boundary_type=clean_text(row.get("boundary_type"), limit=80) or "context_boundary",
|
||
blocked_claim=clean_text(row.get("blocked_claim"), limit=800),
|
||
customer_safe_reason=reason,
|
||
allowed_alternative=clean_text(row.get("allowed_alternative"), limit=800),
|
||
support_refs=list(row.get("support_refs") or [])[:12],
|
||
)
|
||
)
|
||
if claims:
|
||
workspace.claim_ledger = [*list(getattr(workspace, "claim_ledger", []) or []), *claims][-80:]
|
||
if boundaries:
|
||
workspace.boundary_ledger = [*list(getattr(workspace, "boundary_ledger", []) or []), *boundaries][-40:]
|
||
topic = packet.get("topic_workspace") if isinstance(packet.get("topic_workspace"), dict) else {}
|
||
if topic:
|
||
workspace.known_facts = [
|
||
*list(getattr(workspace, "known_facts", []) or []),
|
||
{
|
||
"type": "topic_workspace_context",
|
||
"summary": clean_text(topic.get("topic_summary"), limit=800),
|
||
"topic_signature": topic.get("topic_signature"),
|
||
"current_objective": clean_text(topic.get("current_objective"), limit=500),
|
||
},
|
||
][-64:]
|
||
return len(claims)
|
||
|
||
|
||
def build_topic_workspace_packet(
|
||
*,
|
||
user_question: str,
|
||
context_claims: list[dict[str, Any]],
|
||
context_pack: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
pack = context_pack if isinstance(context_pack, dict) else {}
|
||
topic_pack = _topic_pack_from_context(pack)
|
||
seed_claims = [item for item in context_claims if isinstance(item, dict) and item.get("claim_text")][:8]
|
||
seed_texts = [clean_text(user_question, limit=500), *[clean_text(item.get("claim_text"), limit=500) for item in seed_claims[:6]]]
|
||
summary = clean_text(topic_pack.get("topic_summary") or _summarize_seed_texts(seed_texts), limit=1200)
|
||
linked_memory_ids = [
|
||
str(item.get("source_ref") or "")[:80]
|
||
for item in seed_claims
|
||
if str(item.get("source_kind") or "") == "memory" and str(item.get("source_ref") or "").strip()
|
||
]
|
||
linked_thread_refs = [
|
||
str(item.get("source_ref") or "")[:80]
|
||
for item in seed_claims
|
||
if str(item.get("source_kind") or "") == "thread_message" and str(item.get("source_ref") or "").strip()
|
||
]
|
||
return {
|
||
"packet_version": TOPIC_WORKSPACE_PACKET_VERSION,
|
||
"topic_signature": _topic_signature(seed_texts),
|
||
"topic_summary": summary,
|
||
"current_objective": clean_text(user_question, limit=800),
|
||
"known_assumptions": _claims_by_role(seed_claims, {"user_historical_goal", "semantic_memory"}, limit=5),
|
||
"prior_proposals": _claims_by_role(seed_claims, {"assistant_prior_proposal"}, limit=6),
|
||
"open_questions": _topic_text_list(topic_pack.get("open_questions"), limit=6),
|
||
"monetization_hypothesis": clean_text(topic_pack.get("monetization_hypothesis"), limit=600),
|
||
"next_actionable_package": clean_text(topic_pack.get("next_actionable_package"), limit=800),
|
||
"linked_thread_refs": _dedupe(linked_thread_refs)[:12],
|
||
"linked_run_ids": _topic_text_list(topic_pack.get("linked_run_ids"), limit=12),
|
||
"linked_memory_ids": _dedupe(linked_memory_ids)[:12],
|
||
"selection_policy": {
|
||
"semantic_overlap_and_scope_not_title_equality": True,
|
||
"topic_workspace_is_not_a_project_truth_state": True,
|
||
},
|
||
}
|
||
|
||
|
||
def context_evidence_obligation_packet(value: Any) -> dict[str, Any]:
|
||
packet = value if isinstance(value, dict) else {}
|
||
claims = [item for item in list(packet.get("claims") or []) if isinstance(item, dict)]
|
||
obligations = [item for item in list(packet.get("continuation_obligations") or []) if isinstance(item, dict)]
|
||
return {
|
||
"packet_version": "pmflow_ai_context_obligation_summary.v1",
|
||
"claim_count": len(claims),
|
||
"required_claim_ids": [str(item.get("claim_id") or "")[:80] for item in claims if item.get("required_for_answer")][:12],
|
||
"obligation_count": len(obligations),
|
||
"topic_signature": str((packet.get("topic_workspace") or {}).get("topic_signature") or "")[:80]
|
||
if isinstance(packet.get("topic_workspace"), dict)
|
||
else "",
|
||
}
|
||
|
||
|
||
def _context_rows(pack: dict[str, Any]) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
thread_context = pack.get("thread_context") if isinstance(pack.get("thread_context"), dict) else {}
|
||
memory_context = pack.get("memory_context") if isinstance(pack.get("memory_context"), dict) else {}
|
||
topic_context = pack.get("topic_context") if isinstance(pack.get("topic_context"), dict) else {}
|
||
seen: set[str] = set()
|
||
for item in [*list(thread_context.get("relevant_messages") or []), *list(thread_context.get("recent_messages") or [])]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
text = clean_text(item.get("content") or item.get("summary"), limit=1200)
|
||
source_ref = str(item.get("message_id") or f"thread:{item.get('message_index') or ''}")[:160]
|
||
key = f"thread:{source_ref}:{text[:120]}"
|
||
if not text or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
rows.append(
|
||
{
|
||
"source_kind": "thread_message",
|
||
"source_role": str(item.get("role") or "")[:40],
|
||
"source_ref": source_ref,
|
||
"message_index": item.get("message_index") or 0,
|
||
"text": text,
|
||
"confidence": "high",
|
||
}
|
||
)
|
||
for item in list(memory_context.get("memories") or []):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
text = clean_text(item.get("summary"), limit=1200)
|
||
source_ref = str(item.get("memory_id") or item.get("scope_ref") or "")[:160]
|
||
key = f"memory:{source_ref}:{text[:120]}"
|
||
if not text or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
rows.append(
|
||
{
|
||
"source_kind": "memory",
|
||
"source_role": str(item.get("memory_scope") or "")[:40],
|
||
"source_ref": source_ref,
|
||
"message_index": 0,
|
||
"text": text,
|
||
"confidence": _confidence_from_score(item.get("confidence_score")),
|
||
}
|
||
)
|
||
for item in list(topic_context.get("topic_workspaces") or []):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
text = clean_text(
|
||
"；".join(
|
||
str(value)
|
||
for value in (
|
||
item.get("topic_summary"),
|
||
item.get("current_objective"),
|
||
item.get("next_actionable_package"),
|
||
)
|
||
if value
|
||
),
|
||
limit=1200,
|
||
)
|
||
source_ref = str(item.get("topic_id") or item.get("topic_signature") or "")[:160]
|
||
key = f"topic:{source_ref}:{text[:120]}"
|
||
if not text or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
rows.append(
|
||
{
|
||
"source_kind": "topic_workspace",
|
||
"source_role": "topic",
|
||
"source_ref": source_ref,
|
||
"message_index": 0,
|
||
"text": text,
|
||
"confidence": _confidence_from_score(item.get("confidence_score")),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def _semantic_role(row: dict[str, Any]) -> str:
|
||
source_kind = str(row.get("source_kind") or "")
|
||
if source_kind == "memory":
|
||
return "semantic_memory"
|
||
if source_kind == "topic_workspace":
|
||
return "topic_workspace"
|
||
role = str(row.get("source_role") or "").strip().lower()
|
||
if role == "assistant":
|
||
return "assistant_prior_proposal"
|
||
if role == "user":
|
||
return "user_historical_goal"
|
||
return "thread_context"
|
||
|
||
|
||
def _row_score(*, row: dict[str, Any], question: str) -> float:
|
||
question_terms = _text_terms(question)
|
||
text_terms = _text_terms(row.get("text"))
|
||
overlap = len(question_terms & text_terms)
|
||
union = len(question_terms | text_terms) or 1
|
||
score = overlap / union
|
||
if row.get("source_kind") == "thread_message":
|
||
score += 0.18
|
||
if row.get("source_kind") == "topic_workspace":
|
||
score += 0.15
|
||
if row.get("source_kind") == "memory":
|
||
score += 0.12
|
||
if str(row.get("source_role") or "").lower() == "assistant":
|
||
score += 0.05
|
||
try:
|
||
index = int(row.get("message_index") or 0)
|
||
except Exception:
|
||
index = 0
|
||
if index:
|
||
score += min(0.08, index / 1000)
|
||
return score
|
||
|
||
|
||
def _context_conflict_boundaries(*, pack: dict[str, Any], context_claims: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
topic_context = pack.get("topic_context") if isinstance(pack.get("topic_context"), dict) else {}
|
||
rows = list(topic_context.get("topic_workspaces") or [])
|
||
if not context_claims or not rows:
|
||
return []
|
||
return []
|
||
|
||
|
||
def _topic_pack_from_context(pack: dict[str, Any]) -> dict[str, Any]:
|
||
topic_context = pack.get("topic_context") if isinstance(pack.get("topic_context"), dict) else {}
|
||
rows = [item for item in list(topic_context.get("topic_workspaces") or []) if isinstance(item, dict)]
|
||
return rows[0] if rows else {}
|
||
|
||
|
||
def _claims_by_role(claims: list[dict[str, Any]], roles: set[str], *, limit: int) -> list[str]:
|
||
return _dedupe(
|
||
[
|
||
clean_text(item.get("claim_text"), limit=360)
|
||
for item in claims
|
||
if str(item.get("semantic_role") or "") in roles and clean_text(item.get("claim_text"), limit=360)
|
||
]
|
||
)[:limit]
|
||
|
||
|
||
def _topic_text_list(value: Any, *, limit: int) -> list[str]:
|
||
rows = value if isinstance(value, list) else []
|
||
return _dedupe([clean_text(item, limit=360) for item in rows if clean_text(item, limit=360)])[:limit]
|
||
|
||
|
||
def _summarize_seed_texts(values: list[str]) -> str:
|
||
rows = _dedupe([clean_text(item, limit=360) for item in values if clean_text(item, limit=360)])
|
||
return "；".join(rows[:4])[:1200]
|
||
|
||
|
||
def _topic_signature(values: list[str]) -> str:
|
||
terms: list[str] = []
|
||
for value in values:
|
||
terms.extend(sorted(_text_terms(value)))
|
||
key_terms = _dedupe(terms)[:24]
|
||
raw = "|".join(key_terms) or "|".join(clean_text(item, limit=80) for item in values[:3])
|
||
return f"topic_{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]}"
|
||
|
||
|
||
def _text_terms(value: Any) -> set[str]:
|
||
text = str(value or "").lower()
|
||
alnum = {item for item in re.findall(r"[a-z0-9_]{2,}", text)}
|
||
cjk = re.findall(r"[\u4e00-\u9fff]", text)
|
||
cjk_terms = {"".join(cjk[index : index + 2]) for index in range(max(0, len(cjk) - 1))}
|
||
return {item for item in [*alnum, *cjk_terms] if item}
|
||
|
||
|
||
def _source_rank(value: Any) -> int:
|
||
source = str(value or "")
|
||
return {"thread_message": 0, "topic_workspace": 1, "memory": 2}.get(source, 9)
|
||
|
||
|
||
def _customer_value_for_role(role: str) -> str:
|
||
return {
|
||
"assistant_prior_proposal": "保留上一轮已经给过用户的方案和判断，支持连续追问。",
|
||
"user_historical_goal": "保留用户此前表达过的目标、约束和关注点。",
|
||
"semantic_memory": "保留长期语义记忆，支持跨轮和跨线程连续性。",
|
||
"topic_workspace": "保留持续主题工作区，避免新线程丢失同一主题进展。",
|
||
}.get(role, "保留当前上下文中与本轮目标相关的可复用材料。")
|
||
|
||
|
||
def _confidence_from_score(value: Any) -> str:
|
||
try:
|
||
score = float(value)
|
||
except Exception:
|
||
return "medium"
|
||
if score >= 0.8:
|
||
return "high"
|
||
if score >= 0.45:
|
||
return "medium"
|
||
return "low"
|
||
|
||
|
||
def _claim_confidence(row: dict[str, Any]) -> str:
|
||
confidence = str(row.get("confidence") or "medium").strip()
|
||
return confidence if confidence in {"high", "medium", "low", "unknown"} else "medium"
|
||
|
||
|
||
def _dedupe(values: list[str]) -> list[str]:
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for value in values:
|
||
text = clean_text(value, limit=600)
|
||
key = re.sub(r"\s+", "", text).lower()
|
||
if not text or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(text)
|
||
return out
|
||
|
||
|
||
__all__ = [
|
||
"CONTEXT_EVIDENCE_PACKET_VERSION",
|
||
"TOPIC_WORKSPACE_PACKET_VERSION",
|
||
"append_context_evidence_to_workspace",
|
||
"build_context_evidence_packet",
|
||
"build_topic_workspace_packet",
|
||
"context_evidence_obligation_packet",
|
||
]
|