feat(session): 新增会话记录相关工具类
新增了session模块，实现了SessionRecorder会话记录器以及SessionOpResult结果类，提供会话的创建、更新、关闭、重开、触达以及查询等完整会话管理能力
This commit is contained in:
parent
3b487fae57
commit
5a01c894f2
6
backend/package/yuxi/channel/session/__init__.py
Normal file
6
backend/package/yuxi/channel/session/__init__.py
Normal file
@ -0,0 +1,6 @@
|
||||
from yuxi.channel.session.recorder import SessionOpResult, SessionRecorder
|
||||
|
||||
__all__ = [
|
||||
"SessionOpResult",
|
||||
"SessionRecorder",
|
||||
]
|
||||
293
backend/package/yuxi/channel/session/recorder.py
Normal file
293
backend/package/yuxi/channel/session/recorder.py
Normal file
@ -0,0 +1,293 @@
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
|
||||
from yuxi.channel.message.models import DispatchResult, UnifiedMessage
|
||||
from yuxi.repositories.conversation_repository import ConversationRepository
|
||||
from yuxi.utils.datetime_utils import utc_isoformat
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CONVERSATION_TERMINAL_STATUSES = frozenset({"closed", "archived", "deleted"})
|
||||
|
||||
|
||||
def is_terminal_status(status: str) -> bool:
|
||||
return status in _CONVERSATION_TERMINAL_STATUSES
|
||||
|
||||
|
||||
_METADATA_SESSION_KEY = "session_key"
|
||||
_METADATA_SOURCE = "source"
|
||||
_METADATA_SOURCE_CHANNEL = "channel"
|
||||
_METADATA_CHANNEL_TYPE = "channel_type"
|
||||
_METADATA_ACCOUNT_ID = "account_id"
|
||||
_METADATA_PEER_ID = "peer_id"
|
||||
_METADATA_PEER_KIND = "peer_kind"
|
||||
_METADATA_LAST_MSG_ID = "last_msg_id"
|
||||
_METADATA_FIRST_INTERACTION_AT = "first_interaction_at"
|
||||
_METADATA_LAST_INTERACTION_AT = "last_interaction_at"
|
||||
_METADATA_RECORDED_AT = "recorded_at"
|
||||
_METADATA_CLOSED_AT = "closed_at"
|
||||
_METADATA_CLOSE_REASON = "close_reason"
|
||||
|
||||
|
||||
def _normalize_session_key(session_key: str) -> str:
|
||||
return session_key.strip().lower()
|
||||
|
||||
|
||||
def _build_base_metadata(msg: UnifiedMessage, session_key: str) -> dict:
|
||||
return {
|
||||
_METADATA_SOURCE: _METADATA_SOURCE_CHANNEL,
|
||||
_METADATA_CHANNEL_TYPE: msg.channel_type,
|
||||
_METADATA_ACCOUNT_ID: msg.account_id,
|
||||
_METADATA_PEER_ID: msg.sender.id,
|
||||
_METADATA_PEER_KIND: msg.sender.kind.value,
|
||||
_METADATA_SESSION_KEY: _normalize_session_key(session_key),
|
||||
}
|
||||
|
||||
|
||||
def _build_create_metadata(msg: UnifiedMessage, session_key: str) -> dict:
|
||||
now = utc_isoformat()
|
||||
meta = _build_base_metadata(msg, session_key)
|
||||
meta[_METADATA_LAST_MSG_ID] = msg.msg_id
|
||||
meta[_METADATA_FIRST_INTERACTION_AT] = now
|
||||
meta[_METADATA_LAST_INTERACTION_AT] = now
|
||||
meta[_METADATA_RECORDED_AT] = now
|
||||
return meta
|
||||
|
||||
|
||||
@dataclass
|
||||
class SessionOpResult:
|
||||
success: bool
|
||||
thread_id: str
|
||||
error: str | None = None
|
||||
found: bool = True
|
||||
|
||||
|
||||
class SessionRecorder:
|
||||
def __init__(self, conv_repo: ConversationRepository):
|
||||
self._conv_repo = conv_repo
|
||||
|
||||
async def record(
|
||||
self,
|
||||
msg: UnifiedMessage,
|
||||
dispatch_result: DispatchResult,
|
||||
session_key: str,
|
||||
user_id: str = "channel_user",
|
||||
title: str | None = None,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
track_task: Callable[[object], None] | None = None,
|
||||
) -> None:
|
||||
task = self._create_conversation_async(
|
||||
msg,
|
||||
dispatch_result,
|
||||
session_key,
|
||||
user_id=user_id,
|
||||
title=title,
|
||||
on_error=on_error,
|
||||
)
|
||||
if track_task is not None:
|
||||
track_task(task)
|
||||
|
||||
async def record_or_update(
|
||||
self,
|
||||
msg: UnifiedMessage,
|
||||
dispatch_result: DispatchResult,
|
||||
session_key: str,
|
||||
user_id: str = "channel_user",
|
||||
title: str | None = None,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
) -> bool:
|
||||
try:
|
||||
conv = await self._conv_repo.get_conversation_by_thread_id(
|
||||
dispatch_result.thread_id
|
||||
)
|
||||
if conv is not None and is_terminal_status(conv.status):
|
||||
logger.debug(
|
||||
"Session record_or_update skipped: thread=%s status=%s",
|
||||
dispatch_result.thread_id,
|
||||
conv.status,
|
||||
)
|
||||
return False
|
||||
|
||||
base_meta = _build_create_metadata(msg, session_key)
|
||||
update_meta = {
|
||||
_METADATA_LAST_MSG_ID: msg.msg_id,
|
||||
_METADATA_LAST_INTERACTION_AT: utc_isoformat(),
|
||||
}
|
||||
|
||||
await self._conv_repo.upsert_conversation(
|
||||
thread_id=dispatch_result.thread_id,
|
||||
user_id=user_id,
|
||||
agent_id=str(dispatch_result.agent_config_id),
|
||||
title=title,
|
||||
metadata=base_meta,
|
||||
update_metadata=update_meta,
|
||||
)
|
||||
logger.info(
|
||||
"Session upserted: msg=%s channel=%s thread=%s",
|
||||
msg.msg_id,
|
||||
msg.channel_type,
|
||||
dispatch_result.thread_id,
|
||||
)
|
||||
return True
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception(
|
||||
"Session record_or_update failed: msg=%s thread=%s",
|
||||
msg.msg_id,
|
||||
dispatch_result.thread_id,
|
||||
)
|
||||
return False
|
||||
|
||||
async def close(
|
||||
self,
|
||||
thread_id: str,
|
||||
reason: str | None = None,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
) -> SessionOpResult:
|
||||
try:
|
||||
meta = {_METADATA_CLOSED_AT: utc_isoformat()}
|
||||
if reason:
|
||||
meta[_METADATA_CLOSE_REASON] = reason
|
||||
|
||||
updated = await self._conv_repo.update_conversation(
|
||||
thread_id,
|
||||
status="closed",
|
||||
metadata=meta,
|
||||
)
|
||||
if updated is None:
|
||||
return SessionOpResult(
|
||||
success=False, thread_id=thread_id, error="not found", found=False
|
||||
)
|
||||
logger.info("Session closed: thread=%s", thread_id)
|
||||
return SessionOpResult(success=True, thread_id=thread_id)
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception("Session close failed: thread=%s", thread_id)
|
||||
return SessionOpResult(success=False, thread_id=thread_id, error=str(exc))
|
||||
|
||||
async def reopen(
|
||||
self,
|
||||
thread_id: str,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
) -> SessionOpResult:
|
||||
try:
|
||||
updated = await self._conv_repo.update_conversation(
|
||||
thread_id,
|
||||
status="active",
|
||||
metadata={_METADATA_LAST_INTERACTION_AT: utc_isoformat()},
|
||||
)
|
||||
if updated is None:
|
||||
return SessionOpResult(
|
||||
success=False, thread_id=thread_id, error="not found", found=False
|
||||
)
|
||||
logger.info("Session reopened: thread=%s", thread_id)
|
||||
return SessionOpResult(success=True, thread_id=thread_id)
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception("Session reopen failed: thread=%s", thread_id)
|
||||
return SessionOpResult(success=False, thread_id=thread_id, error=str(exc))
|
||||
|
||||
async def touch(
|
||||
self,
|
||||
msg: UnifiedMessage,
|
||||
thread_id: str,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
) -> bool:
|
||||
try:
|
||||
conv = await self._conv_repo.get_conversation_by_thread_id(thread_id)
|
||||
if conv is None:
|
||||
return False
|
||||
if is_terminal_status(conv.status):
|
||||
logger.debug(
|
||||
"Session touch skipped: thread=%s status=%s",
|
||||
thread_id,
|
||||
conv.status,
|
||||
)
|
||||
return False
|
||||
await self._conv_repo.update_conversation(
|
||||
thread_id,
|
||||
metadata={
|
||||
_METADATA_LAST_MSG_ID: msg.msg_id,
|
||||
_METADATA_LAST_INTERACTION_AT: utc_isoformat(),
|
||||
},
|
||||
)
|
||||
return True
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception("Session touch failed: thread=%s", thread_id)
|
||||
return False
|
||||
|
||||
async def find_by_session_key(
|
||||
self,
|
||||
session_key: str,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
) -> list:
|
||||
try:
|
||||
normalized = _normalize_session_key(session_key)
|
||||
return await self._conv_repo.get_conversations_by_metadata_key(
|
||||
_METADATA_SESSION_KEY, normalized
|
||||
)
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception("Session find_by_session_key failed: session_key=%s", session_key)
|
||||
return []
|
||||
|
||||
async def list_by_channel(
|
||||
self,
|
||||
channel_type: str,
|
||||
account_id: str | None = None,
|
||||
on_error: Callable[[Exception], None] | None = None,
|
||||
) -> list:
|
||||
try:
|
||||
return await self._conv_repo.get_conversations_by_channel(
|
||||
channel_type, account_id
|
||||
)
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception(
|
||||
"Session list_by_channel failed: channel=%s account=%s",
|
||||
channel_type,
|
||||
account_id,
|
||||
)
|
||||
return []
|
||||
|
||||
async def _create_conversation_async(
|
||||
self,
|
||||
msg: UnifiedMessage,
|
||||
dispatch_result: DispatchResult,
|
||||
session_key: str,
|
||||
user_id: str,
|
||||
title: str | None,
|
||||
on_error: Callable[[Exception], None] | None,
|
||||
) -> None:
|
||||
try:
|
||||
metadata = _build_create_metadata(msg, session_key)
|
||||
|
||||
await self._conv_repo.upsert_conversation(
|
||||
thread_id=dispatch_result.thread_id,
|
||||
user_id=user_id,
|
||||
agent_id=str(dispatch_result.agent_config_id),
|
||||
title=title,
|
||||
metadata=metadata,
|
||||
)
|
||||
logger.info(
|
||||
"Session recorded: msg=%s channel=%s thread=%s",
|
||||
msg.msg_id,
|
||||
msg.channel_type,
|
||||
dispatch_result.thread_id,
|
||||
)
|
||||
except Exception as exc:
|
||||
if on_error is not None:
|
||||
on_error(exc)
|
||||
logger.exception(
|
||||
"Failed to record session: msg=%s thread=%s",
|
||||
msg.msg_id,
|
||||
dispatch_result.thread_id,
|
||||
)
|
||||
Loading…
Reference in New Issue
Block a user