Manage Conversation DB Skill

This skill provides guidance for implementing conversation persistence in the database.

Purpose

Handle stateless conversation persistence:

• Create/load conversation by ID
• Save user/assistant messages with role and content
• Fetch history for agent input
• Async queries for performance

When to Use

Use this skill when:

• Implementing conversation storage layer
• Building message history retrieval
• Creating conversation management utilities
• Setting up async database operations for chat

Capabilities

• Conversation Lifecycle: Create new or load existing conversations
• Message Storage: Save messages with role, content, and metadata
• History Retrieval: Fetch conversation messages in chronological order
• Async Operations: Non-blocking database queries
• User Isolation: Conversations tied to user_id

Database Schema

from sqlmodel import SQLModel, Field, Relationship
from datetime import datetime
from typing import Optional, List

class Conversation(SQLModel, table=True):
    id: str = Field(primary_key=True)
    user_id: str = Field(index=True)
    title: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
    messages: List["Message"] = Relationship(back_populates="conversation")

class Message(SQLModel, table=True):
    id: int = Field(primary_key=True, autoincrement=True)
    conversation_id: str = Field(foreign_key="conversation.id", index=True)
    role: str = Field(index=True)  # "user", "assistant", "tool"
    content: str
    tool_name: Optional[str] = None
    tool_call_id: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
    conversation: Optional[Conversation] = Relationship(back_populates="messages")

Implementation Pattern

Conversation Service

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, desc

class ConversationService:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_or_create_conversation(
        self,
        conversation_id: Optional[str],
        user_id: str
    ) -> Conversation:
        """Get existing or create new conversation."""
        if conversation_id:
            result = await self.session.execute(
                select(Conversation).where(
                    Conversation.id == conversation_id,
                    Conversation.user_id == user_id
                )
            )
            conversation = result.scalar_one_or_none()
            if conversation:
                return conversation

        # Create new conversation
        conversation = Conversation(
            id=conversation_id or str(uuid.uuid4()),
            user_id=user_id,
            title="New Chat"
        )
        self.session.add(conversation)
        await self.session.commit()
        await self.session.refresh(conversation)
        return conversation

    async def save_message(
        self,
        conversation_id: str,
        role: str,
        content: str,
        tool_name: Optional[str] = None,
        tool_call_id: Optional[str] = None
    ) -> Message:
        """Save a message to the conversation."""
        message = Message(
            conversation_id=conversation_id,
            role=role,
            content=content,
            tool_name=tool_name,
            tool_call_id=tool_call_id
        )
        self.session.add(message)
        await self.session.commit()
        await self.session.refresh(message)
        return message

    async def get_conversation_history(
        self,
        conversation_id: str,
        user_id: str,
        limit: int = 50
    ) -> List[Message]:
        """Fetch conversation messages for agent input."""
        result = await self.session.execute(
            select(Message)
            .where(Message.conversation_id == conversation_id)
            .order_by(Message.created_at.asc())
            .limit(limit)
        )
        messages = result.scalars().all()

        # Verify user owns this conversation
        conv_result = await self.session.execute(
            select(Conversation).where(Conversation.id == conversation_id)
        )
        conversation = conv_result.scalar_one_or_none()
        if not conversation or conversation.user_id != user_id:
            raise PermissionError("Conversation not found")

        return messages

    async def save_conversation_messages(
        self,
        conversation_id: str,
        user_id: str,
        messages: List[dict]
    ) -> None:
        """Save multiple messages atomically."""
        for msg in messages:
            await self.save_message(
                conversation_id=conversation_id,
                role=msg["role"],
                content=msg["content"],
                tool_name=msg.get("tool_name"),
                tool_call_id=msg.get("tool_call_id")
            )

        # Update conversation timestamp
        await self.session.execute(
            select(Conversation)
            .where(Conversation.id == conversation_id)
        )
        # ... update updated_at

Helper Functions

async def build_message_array(
    db_messages: List[Message],
    user_message: str
) -> List[dict]:
    """Convert DB messages to agent input format."""
    message_array = [{"role": "system", "content": SYSTEM_PROMPT}]

    for msg in db_messages:
        message_array.append({
            "role": msg.role,
            "content": msg.content
        })

    message_array.append({
        "role": "user",
        "content": user_message
    })

    return message_array

def message_to_dict(message: Message) -> dict:
    """Convert Message to dictionary."""
    return {
        "id": message.id,
        "role": message.role,
        "content": message.content,
        "tool_name": message.tool_name,
        "tool_call_id": message.tool_call_id,
        "created_at": message.created_at.isoformat()
    }

Async Session Management

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

@asynccontextmanager
async def get_db_session():
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Verification Checklist

• Conversations can be created and loaded by ID
• Messages saved with correct role and content
• History retrieved in chronological order
• User isolation enforced on conversations
• Async queries work without blocking
• Tool calls stored with metadata