File size: 2,056 Bytes
04dc214
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
"""ChatSession model for RAG chatbot conversations."""
from datetime import datetime
from typing import Optional
from sqlalchemy import Column, String, DateTime, JSON, Text
import uuid
from src.config.database import Base


class ChatSession(Base):
    """Represents a conversation session with the RAG chatbot."""

    __tablename__ = "chat_sessions"

    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String(255), nullable=True, index=True)
    conversation_history = Column(JSON, default=list)
    context = Column(JSON, default=dict)
    source_citations = Column(JSON, default=list)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    def to_dict(self) -> dict:
        """Convert model to dictionary."""
        return {
            "id": str(self.id),
            "user_id": self.user_id,
            "conversation_history": self.conversation_history,
            "context": self.context,
            "source_citations": self.source_citations,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }

    def add_message(self, role: str, content: str) -> None:
        """Add a message to conversation history.

        Args:
            role: Message role (user, assistant, system).
            content: Message content.
        """
        if self.conversation_history is None:
            self.conversation_history = []
        self.conversation_history.append({
            "role": role,
            "content": content,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def add_citation(self, source: dict) -> None:
        """Add a source citation.

        Args:
            source: Source document information.
        """
        if self.source_citations is None:
            self.source_citations = []
        self.source_citations.append(source)