324 lines
13 KiB
Python
324 lines
13 KiB
Python
"""Thread-level context pack builder for Space PMFlow AI.
|
|
|
|
The agent runtime should not infer multi-turn continuity from product objects
|
|
alone. This service builds a customer-safe context pack from the persisted
|
|
thread transcript and uses generic text relevance to surface prior turns.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import and_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..models import PMFlowAIMessage, PMFlowAIThread
|
|
|
|
|
|
class PMFlowAIThreadContextService:
|
|
"""Build durable, model-ready context from PMFlow AI thread messages."""
|
|
|
|
def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
|
|
self.db = db
|
|
self.tenant_id = int(tenant_id)
|
|
|
|
async def build_context_pack(
|
|
self,
|
|
*,
|
|
thread: PMFlowAIThread | None,
|
|
user_question: str,
|
|
recent_limit: int = 12,
|
|
search_limit: int = 8,
|
|
scan_limit: int = 80,
|
|
) -> dict[str, Any]:
|
|
if thread is None or not getattr(thread, "id", None):
|
|
return {}
|
|
messages = await self._load_messages(thread=thread, limit=scan_limit)
|
|
if not messages:
|
|
return {}
|
|
recent_messages = messages[-max(1, recent_limit):]
|
|
relevant_messages = _rank_relevant_messages(
|
|
question=user_question,
|
|
messages=messages,
|
|
limit=search_limit,
|
|
)
|
|
return {
|
|
"packet_version": "pmflow_ai_thread_context_pack.v1",
|
|
"thread_id": str(thread.public_thread_id or ""),
|
|
"message_count": len(messages),
|
|
"context_policy": {
|
|
"same_thread_history_is_primary_for_multi_turn_recall": True,
|
|
"recent_messages_are_customer_safe_transcript": True,
|
|
"relevant_messages_are_generic_text_retrieval_not_scene_keywords": True,
|
|
"project_state_must_not_replace_thread_history": True,
|
|
},
|
|
"recent_messages": [_message_for_context(item) for item in recent_messages],
|
|
"relevant_messages": [_message_for_context(item) for item in relevant_messages],
|
|
"rolling_summary": _rolling_summary_from_thread(thread),
|
|
"thread_outline": _thread_outline(messages),
|
|
}
|
|
|
|
async def _load_messages(self, *, thread: PMFlowAIThread, limit: int) -> list[PMFlowAIMessage]:
|
|
stmt = (
|
|
select(PMFlowAIMessage)
|
|
.where(
|
|
and_(
|
|
PMFlowAIMessage.tenant_id == self.tenant_id,
|
|
PMFlowAIMessage.thread_id == int(thread.id),
|
|
)
|
|
)
|
|
.order_by(PMFlowAIMessage.message_index.desc(), PMFlowAIMessage.id.desc())
|
|
.limit(max(1, limit))
|
|
)
|
|
rows = list((await self.db.execute(stmt)).scalars().all())
|
|
return list(reversed(rows))
|
|
|
|
|
|
def slim_thread_context_pack(value: Any, *, recent_limit: int = 10, relevant_limit: int = 8) -> dict[str, Any]:
|
|
"""Return a prompt-safe projection of a thread context pack."""
|
|
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
recent = [_slim_message(item) for item in list(value.get("recent_messages") or [])[:recent_limit]]
|
|
relevant = [_slim_message(item) for item in list(value.get("relevant_messages") or [])[:relevant_limit]]
|
|
outline = value.get("thread_outline") if isinstance(value.get("thread_outline"), dict) else {}
|
|
result = {
|
|
"packet_version": value.get("packet_version") or "pmflow_ai_thread_context_pack.v1",
|
|
"thread_id": str(value.get("thread_id") or "")[:80],
|
|
"message_count": _safe_int(value.get("message_count")),
|
|
"context_policy": {
|
|
"same_thread_history_is_primary_for_multi_turn_recall": True,
|
|
"project_state_must_not_replace_thread_history": True,
|
|
},
|
|
"recent_messages": [item for item in recent if item],
|
|
"relevant_messages": [item for item in relevant if item],
|
|
"thread_outline": {
|
|
"first_user_message": str(outline.get("first_user_message") or "")[:500],
|
|
"last_user_message": str(outline.get("last_user_message") or "")[:500],
|
|
"last_assistant_message": str(outline.get("last_assistant_message") or "")[:700],
|
|
},
|
|
}
|
|
rolling_summary = value.get("rolling_summary") if isinstance(value.get("rolling_summary"), dict) else {}
|
|
if rolling_summary:
|
|
result["rolling_summary"] = {
|
|
"summary_version": str(rolling_summary.get("summary_version") or "pmflow_ai_thread_summary.v1")[:80],
|
|
"covered_message_count": _safe_int(rolling_summary.get("covered_message_count")),
|
|
"durable_points": [
|
|
{
|
|
"kind": str(item.get("kind") or "")[:40],
|
|
"summary": str(item.get("summary") or "")[:700],
|
|
"source_message_indexes": list(item.get("source_message_indexes") or [])[:6],
|
|
}
|
|
for item in list(rolling_summary.get("durable_points") or [])[:16]
|
|
if isinstance(item, dict) and str(item.get("summary") or "").strip()
|
|
],
|
|
}
|
|
return {key: item for key, item in result.items() if item not in ("", [], {}, None)}
|
|
|
|
|
|
def update_thread_context_summary_metadata(
|
|
*,
|
|
thread_metadata: Any,
|
|
user_message: str,
|
|
assistant_answer: str,
|
|
user_message_index: int,
|
|
assistant_message_index: int,
|
|
) -> dict[str, Any]:
|
|
"""Append a compact durable turn summary into thread metadata."""
|
|
|
|
metadata = dict(thread_metadata or {}) if isinstance(thread_metadata, dict) else {}
|
|
existing = metadata.get("thread_context_summary") if isinstance(metadata.get("thread_context_summary"), dict) else {}
|
|
points = [dict(item) for item in list(existing.get("durable_points") or []) if isinstance(item, dict)]
|
|
user_text = _clean_text(user_message, limit=500)
|
|
answer_text = _clean_text(assistant_answer, limit=900)
|
|
if user_text:
|
|
points.append(
|
|
{
|
|
"kind": "user_turn",
|
|
"summary": user_text,
|
|
"source_message_indexes": [int(user_message_index)],
|
|
}
|
|
)
|
|
if answer_text:
|
|
points.append(
|
|
{
|
|
"kind": "assistant_turn",
|
|
"summary": answer_text,
|
|
"source_message_indexes": [int(assistant_message_index)],
|
|
}
|
|
)
|
|
summary = {
|
|
"summary_version": "pmflow_ai_thread_summary.v1",
|
|
"covered_message_count": max(
|
|
_safe_int(existing.get("covered_message_count")),
|
|
int(assistant_message_index or 0),
|
|
),
|
|
"durable_points": _dedupe_summary_points(points)[-24:],
|
|
"updated_at": datetime.utcnow().isoformat(),
|
|
}
|
|
metadata["thread_context_summary"] = summary
|
|
return metadata
|
|
|
|
|
|
def thread_context_claim_rows(value: Any, *, limit: int = 8) -> list[dict[str, str]]:
|
|
"""Project relevant thread messages as generic claim rows for answer runtime."""
|
|
|
|
pack = slim_thread_context_pack(value, recent_limit=6, relevant_limit=limit)
|
|
rows: list[dict[str, str]] = []
|
|
seen: set[str] = set()
|
|
for item in [
|
|
*list(pack.get("relevant_messages") or []),
|
|
*list(pack.get("recent_messages") or [])[-4:],
|
|
]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
content = str(item.get("content") or "").strip()
|
|
if not content:
|
|
continue
|
|
key = f"{item.get('message_index')}:{item.get('role')}:{content[:80]}"
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
rows.append(
|
|
{
|
|
"message_id": str(item.get("message_id") or "")[:80],
|
|
"role": str(item.get("role") or "")[:24],
|
|
"message_index": str(item.get("message_index") or "")[:20],
|
|
"content": content[:900],
|
|
}
|
|
)
|
|
if len(rows) >= limit:
|
|
break
|
|
return rows
|
|
|
|
|
|
def _message_for_context(row: PMFlowAIMessage) -> dict[str, Any]:
|
|
content = _clean_text(getattr(row, "content", "") or "", limit=1800)
|
|
return {
|
|
"message_id": str(getattr(row, "public_message_id", "") or "")[:80],
|
|
"role": str(getattr(row, "role", "") or "")[:24],
|
|
"message_index": _safe_int(getattr(row, "message_index", None)),
|
|
"run_id": str(getattr(row, "run_public_id", "") or "")[:80],
|
|
"sent_at": _iso(getattr(row, "sent_at", None)),
|
|
"content": content,
|
|
}
|
|
|
|
|
|
def _slim_message(value: Any) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
content = _clean_text(value.get("content") or "", limit=1200)
|
|
if not content:
|
|
return {}
|
|
return {
|
|
"message_id": str(value.get("message_id") or "")[:80],
|
|
"role": str(value.get("role") or "")[:24],
|
|
"message_index": _safe_int(value.get("message_index")),
|
|
"content": content,
|
|
}
|
|
|
|
|
|
def _rank_relevant_messages(*, question: str, messages: list[PMFlowAIMessage], limit: int) -> list[PMFlowAIMessage]:
|
|
query_tokens = _semantic_tokens(question)
|
|
if not query_tokens:
|
|
return []
|
|
scored: list[tuple[float, int, PMFlowAIMessage]] = []
|
|
total = max(1, len(messages))
|
|
for index, row in enumerate(messages):
|
|
content = str(getattr(row, "content", "") or "")
|
|
tokens = _semantic_tokens(content)
|
|
if not tokens:
|
|
continue
|
|
overlap = len(query_tokens.intersection(tokens))
|
|
if overlap <= 0:
|
|
continue
|
|
density = overlap / max(8, len(query_tokens))
|
|
recency = (index + 1) / total
|
|
role_boost = 0.12 if str(getattr(row, "role", "") or "") == "assistant" else 0.0
|
|
scored.append((density + recency * 0.08 + role_boost, index, row))
|
|
scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
|
|
return [row for _, _, row in scored[: max(1, limit)]]
|
|
|
|
|
|
def _semantic_tokens(value: str) -> set[str]:
|
|
text = _clean_text(value, limit=4000).lower()
|
|
if not text:
|
|
return set()
|
|
tokens = set(re.findall(r"[a-z0-9][a-z0-9_+-]{1,}|[\u4e00-\u9fff]{2,}", text))
|
|
for chunk in re.findall(r"[\u4e00-\u9fff]{3,}", text):
|
|
tokens.update(chunk[index : index + 2] for index in range(0, max(0, len(chunk) - 1)))
|
|
tokens.update(chunk[index : index + 3] for index in range(0, max(0, len(chunk) - 2)))
|
|
return {token for token in tokens if len(token) >= 2}
|
|
|
|
|
|
def _thread_outline(messages: list[PMFlowAIMessage]) -> dict[str, Any]:
|
|
first_user = next((row for row in messages if str(getattr(row, "role", "") or "") == "user"), None)
|
|
user_rows = [row for row in messages if str(getattr(row, "role", "") or "") == "user"]
|
|
assistant_rows = [row for row in messages if str(getattr(row, "role", "") or "") == "assistant"]
|
|
return {
|
|
"first_user_message": _clean_text(getattr(first_user, "content", "") if first_user else "", limit=500),
|
|
"last_user_message": _clean_text(getattr(user_rows[-1], "content", "") if user_rows else "", limit=500),
|
|
"last_assistant_message": _clean_text(getattr(assistant_rows[-1], "content", "") if assistant_rows else "", limit=700),
|
|
}
|
|
|
|
|
|
def _rolling_summary_from_thread(thread: PMFlowAIThread) -> dict[str, Any]:
|
|
metadata = getattr(thread, "thread_metadata", None)
|
|
if not isinstance(metadata, dict):
|
|
return {}
|
|
summary = metadata.get("thread_context_summary")
|
|
return dict(summary) if isinstance(summary, dict) else {}
|
|
|
|
|
|
def _dedupe_summary_points(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
out: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for item in rows:
|
|
summary = _clean_text(item.get("summary") if isinstance(item, dict) else "", limit=900)
|
|
if not summary:
|
|
continue
|
|
key = f"{item.get('kind')}:{summary[:120].lower()}"
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(
|
|
{
|
|
"kind": str(item.get("kind") or "turn")[:40],
|
|
"summary": summary,
|
|
"source_message_indexes": [
|
|
_safe_int(value)
|
|
for value in list(item.get("source_message_indexes") or [])[:6]
|
|
if _safe_int(value) > 0
|
|
],
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def _clean_text(value: Any, *, limit: int) -> str:
|
|
text = " ".join(str(value or "").replace("\x00", " ").split())
|
|
return text[: max(0, limit)]
|
|
|
|
|
|
def _safe_int(value: Any) -> int:
|
|
try:
|
|
return int(value)
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def _iso(value: Any) -> str | None:
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
return None
|
|
|
|
|
|
__all__ = [
|
|
"PMFlowAIThreadContextService",
|
|
"slim_thread_context_pack",
|
|
"thread_context_claim_rows",
|
|
"update_thread_context_summary_metadata",
|
|
]
|