286 lines
12 KiB
Python
286 lines
12 KiB
Python
"""Durable semantic topic workspaces for PMFlow AI continuity."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import and_, desc, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..models import PMFlowAITopicWorkspace, PMFlowAIThread
|
||
|
||
|
||
class PMFlowAITopicWorkspaceService:
|
||
"""Persist and retrieve cross-thread topic continuity without title matching."""
|
||
|
||
def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id or 1)
|
||
|
||
async def build_topic_context_pack(
|
||
self,
|
||
*,
|
||
current_user: Any,
|
||
user_question: str,
|
||
thread_context_pack: dict[str, Any] | None,
|
||
limit: int = 4,
|
||
) -> dict[str, Any]:
|
||
user_id = _safe_int(getattr(current_user, "id", None))
|
||
if user_id is None:
|
||
return {}
|
||
result = await self.db.execute(
|
||
select(PMFlowAITopicWorkspace)
|
||
.where(
|
||
and_(
|
||
PMFlowAITopicWorkspace.tenant_id == self.tenant_id,
|
||
PMFlowAITopicWorkspace.user_id == int(user_id),
|
||
PMFlowAITopicWorkspace.status == "active",
|
||
)
|
||
)
|
||
.order_by(desc(PMFlowAITopicWorkspace.last_activated_at), desc(PMFlowAITopicWorkspace.id))
|
||
.limit(24)
|
||
)
|
||
rows = list(result.scalars().all())
|
||
seed = _retrieval_seed(user_question=user_question, thread_context_pack=thread_context_pack)
|
||
scored = [
|
||
(row, _semantic_score(seed, _topic_row_text(row)))
|
||
for row in rows
|
||
]
|
||
selected = [
|
||
row
|
||
for row, score in sorted(scored, key=lambda pair: (-pair[1], pair[0].id or 0))
|
||
if score > 0 or not seed
|
||
][: max(1, min(8, int(limit or 4)))]
|
||
if not selected:
|
||
return {}
|
||
return {
|
||
"packet_version": "pmflow_ai_topic_context_pack.v1",
|
||
"topic_workspaces": [topic_workspace_packet(row) for row in selected],
|
||
"retrieval_policy": {
|
||
"semantic_overlap_with_question_and_thread_context": True,
|
||
"title_equality_required": False,
|
||
"business_keyword_list_used": False,
|
||
},
|
||
}
|
||
|
||
async def upsert_from_turn(
|
||
self,
|
||
*,
|
||
thread: PMFlowAIThread | None,
|
||
current_user: Any,
|
||
user_question: str,
|
||
assistant_answer: str,
|
||
context_evidence_packet: dict[str, Any] | None,
|
||
run_public_id: str | None,
|
||
) -> dict[str, Any] | None:
|
||
user_id = _safe_int(getattr(current_user, "id", None))
|
||
if user_id is None:
|
||
return None
|
||
evidence = context_evidence_packet if isinstance(context_evidence_packet, dict) else {}
|
||
topic = evidence.get("topic_workspace") if isinstance(evidence.get("topic_workspace"), dict) else {}
|
||
seed_text = _clean_text(
|
||
"；".join(
|
||
item
|
||
for item in [
|
||
str(topic.get("topic_summary") or ""),
|
||
str(topic.get("current_objective") or ""),
|
||
str(user_question or ""),
|
||
str(assistant_answer or "")[:1200],
|
||
]
|
||
if item
|
||
),
|
||
limit=2600,
|
||
)
|
||
if not seed_text:
|
||
return None
|
||
row = await self._best_existing_topic(user_id=user_id, seed_text=seed_text)
|
||
now = datetime.utcnow()
|
||
public_thread_id = str(getattr(thread, "public_thread_id", "") or "")[:80]
|
||
if row is None:
|
||
row = PMFlowAITopicWorkspace(
|
||
tenant_id=self.tenant_id,
|
||
public_topic_id=f"topic_{hashlib.sha1(f'{user_id}:{seed_text[:500]}'.encode('utf-8')).hexdigest()[:16]}",
|
||
user_id=int(user_id),
|
||
workspace_scope="space",
|
||
status="active",
|
||
topic_signature=str(topic.get("topic_signature") or _topic_signature(seed_text))[:80],
|
||
topic_summary=_clean_text(topic.get("topic_summary") or seed_text, limit=1600),
|
||
current_objective=_clean_text(topic.get("current_objective") or user_question, limit=1200),
|
||
known_assumptions=_safe_text_list(topic.get("known_assumptions"), limit=12),
|
||
prior_proposals=_safe_text_list(topic.get("prior_proposals"), limit=12),
|
||
open_questions=_safe_text_list(topic.get("open_questions"), limit=12),
|
||
monetization_hypothesis=_clean_text(topic.get("monetization_hypothesis"), limit=1000),
|
||
next_actionable_package=_clean_text(topic.get("next_actionable_package") or assistant_answer, limit=1600),
|
||
linked_thread_ids=[public_thread_id] if public_thread_id else [],
|
||
linked_run_ids=[str(run_public_id)[:80]] if run_public_id else [],
|
||
linked_memory_ids=_safe_text_list(topic.get("linked_memory_ids"), limit=24),
|
||
confidence_score=0.62,
|
||
topic_payload=_safe_topic_payload(evidence),
|
||
last_activated_at=now,
|
||
)
|
||
self.db.add(row)
|
||
await self.db.flush()
|
||
return topic_workspace_packet(row)
|
||
row.topic_summary = _merge_text(row.topic_summary, topic.get("topic_summary") or seed_text, limit=1800)
|
||
row.current_objective = _clean_text(user_question or topic.get("current_objective") or row.current_objective, limit=1200)
|
||
row.known_assumptions = _merge_list(row.known_assumptions, topic.get("known_assumptions"), limit=16)
|
||
row.prior_proposals = _merge_list(row.prior_proposals, topic.get("prior_proposals"), limit=16)
|
||
row.open_questions = _merge_list(row.open_questions, topic.get("open_questions"), limit=16)
|
||
row.monetization_hypothesis = _merge_text(row.monetization_hypothesis, topic.get("monetization_hypothesis"), limit=1000)
|
||
row.next_actionable_package = _merge_text(row.next_actionable_package, topic.get("next_actionable_package") or assistant_answer, limit=1800)
|
||
row.linked_thread_ids = _merge_list(row.linked_thread_ids, [public_thread_id] if public_thread_id else [], limit=24)
|
||
row.linked_run_ids = _merge_list(row.linked_run_ids, [str(run_public_id)[:80]] if run_public_id else [], limit=24)
|
||
row.linked_memory_ids = _merge_list(row.linked_memory_ids, topic.get("linked_memory_ids"), limit=32)
|
||
row.confidence_score = max(float(row.confidence_score or 0.0), 0.68)
|
||
row.topic_payload = _safe_topic_payload(evidence)
|
||
row.last_activated_at = now
|
||
await self.db.flush()
|
||
return topic_workspace_packet(row)
|
||
|
||
async def _best_existing_topic(self, *, user_id: int, seed_text: str) -> PMFlowAITopicWorkspace | None:
|
||
result = await self.db.execute(
|
||
select(PMFlowAITopicWorkspace)
|
||
.where(
|
||
and_(
|
||
PMFlowAITopicWorkspace.tenant_id == self.tenant_id,
|
||
PMFlowAITopicWorkspace.user_id == int(user_id),
|
||
PMFlowAITopicWorkspace.status == "active",
|
||
)
|
||
)
|
||
.order_by(desc(PMFlowAITopicWorkspace.last_activated_at), desc(PMFlowAITopicWorkspace.id))
|
||
.limit(24)
|
||
)
|
||
rows = list(result.scalars().all())
|
||
if not rows:
|
||
return None
|
||
ranked = sorted(((row, _semantic_score(seed_text, _topic_row_text(row))) for row in rows), key=lambda pair: -pair[1])
|
||
best, score = ranked[0]
|
||
return best if score >= 0.08 else None
|
||
|
||
|
||
def topic_workspace_packet(row: PMFlowAITopicWorkspace) -> dict[str, Any]:
|
||
return {
|
||
"packet_version": "pmflow_ai_topic_workspace_record.v1",
|
||
"topic_id": str(row.public_topic_id or "")[:80],
|
||
"topic_signature": str(row.topic_signature or "")[:80],
|
||
"status": str(row.status or "")[:40],
|
||
"topic_summary": str(row.topic_summary or "")[:1600],
|
||
"current_objective": str(row.current_objective or "")[:1200],
|
||
"known_assumptions": list(row.known_assumptions or [])[:12],
|
||
"prior_proposals": list(row.prior_proposals or [])[:12],
|
||
"open_questions": list(row.open_questions or [])[:12],
|
||
"monetization_hypothesis": str(row.monetization_hypothesis or "")[:1000],
|
||
"next_actionable_package": str(row.next_actionable_package or "")[:1600],
|
||
"linked_thread_ids": list(row.linked_thread_ids or [])[:24],
|
||
"linked_run_ids": list(row.linked_run_ids or [])[:24],
|
||
"linked_memory_ids": list(row.linked_memory_ids or [])[:24],
|
||
"confidence_score": row.confidence_score,
|
||
"last_activated_at": row.last_activated_at.isoformat() if row.last_activated_at is not None else None,
|
||
}
|
||
|
||
|
||
def _retrieval_seed(*, user_question: str, thread_context_pack: dict[str, Any] | None) -> str:
|
||
rows: list[str] = [str(user_question or "")]
|
||
pack = thread_context_pack if isinstance(thread_context_pack, dict) else {}
|
||
for item in [*list(pack.get("relevant_messages") or [])[:6], *list(pack.get("recent_messages") or [])[-4:]]:
|
||
if isinstance(item, dict):
|
||
rows.append(str(item.get("content") or item.get("summary") or ""))
|
||
return _clean_text("；".join(row for row in rows if row), limit=3000)
|
||
|
||
|
||
def _topic_row_text(row: PMFlowAITopicWorkspace) -> str:
|
||
return _clean_text(
|
||
"；".join(
|
||
str(value)
|
||
for value in [
|
||
row.topic_summary,
|
||
row.current_objective,
|
||
row.known_assumptions,
|
||
row.prior_proposals,
|
||
row.open_questions,
|
||
row.monetization_hypothesis,
|
||
row.next_actionable_package,
|
||
]
|
||
if value
|
||
),
|
||
limit=4000,
|
||
)
|
||
|
||
|
||
def _semantic_score(left: str, right: str) -> float:
|
||
left_terms = _text_terms(left)
|
||
right_terms = _text_terms(right)
|
||
if not left_terms or not right_terms:
|
||
return 0.0
|
||
return len(left_terms & right_terms) / max(1, min(len(left_terms), len(right_terms)))
|
||
|
||
|
||
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 _topic_signature(value: str) -> str:
|
||
terms = sorted(_text_terms(value))[:24]
|
||
raw = "|".join(terms) or value[:200]
|
||
return f"topic_{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]}"
|
||
|
||
|
||
def _merge_text(existing: Any, incoming: Any, *, limit: int) -> str:
|
||
values = _dedupe_texts([str(existing or ""), str(incoming or "")])
|
||
return _clean_text("；".join(values), limit=limit)
|
||
|
||
|
||
def _merge_list(existing: Any, incoming: Any, *, limit: int) -> list[str]:
|
||
return _dedupe_texts([*_safe_text_list(existing, limit=limit), *_safe_text_list(incoming, limit=limit)])[:limit]
|
||
|
||
|
||
def _safe_text_list(value: Any, *, limit: int) -> list[str]:
|
||
rows = value if isinstance(value, list) else []
|
||
return _dedupe_texts([_clean_text(item, limit=500) for item in rows if _clean_text(item, limit=500)])[:limit]
|
||
|
||
|
||
def _dedupe_texts(values: list[str]) -> list[str]:
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for value in values:
|
||
text = _clean_text(value, limit=700)
|
||
key = re.sub(r"\s+", "", text).lower()
|
||
if not text or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(text)
|
||
return out
|
||
|
||
|
||
def _safe_topic_payload(value: dict[str, Any]) -> dict[str, Any]:
|
||
packet = value if isinstance(value, dict) else {}
|
||
return {
|
||
"packet_version": str(packet.get("packet_version") or "")[:80],
|
||
"claim_count": len(list(packet.get("claims") or [])),
|
||
"obligation_count": len(list(packet.get("continuation_obligations") or [])),
|
||
"topic_workspace": packet.get("topic_workspace") if isinstance(packet.get("topic_workspace"), dict) else {},
|
||
}
|
||
|
||
|
||
def _clean_text(value: Any, *, limit: int) -> str:
|
||
return " ".join(str(value or "").replace("\x00", " ").split())[: max(0, int(limit or 0))]
|
||
|
||
|
||
def _safe_int(value: Any) -> int | None:
|
||
try:
|
||
if value in (None, ""):
|
||
return None
|
||
return int(value)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
__all__ = ["PMFlowAITopicWorkspaceService", "topic_workspace_packet"]
|