Spaces:
Running
Running
File size: 3,911 Bytes
dc3879e |
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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
"""Conversation service for managing chat sessions.
[Task]: T016
[From]: specs/004-ai-chatbot/tasks.md
This service handles conversation persistence and history loading.
"""
import uuid
from datetime import datetime
from typing import Optional, List
from sqlmodel import Session, select
from models.conversation import Conversation
from models.message import Message, MessageRole
def get_or_create_conversation(
db: Session,
user_id: uuid.UUID,
conversation_id: Optional[uuid.UUID] = None
) -> Conversation:
"""Get existing conversation or create new one.
[From]: specs/004-ai-chatbot/plan.md - Conversation Management
Args:
db: Database session
user_id: User ID who owns the conversation
conversation_id: Optional conversation ID to load
Returns:
Conversation object (existing or new)
Raises:
ValueError: If conversation_id provided but not found or doesn't belong to user
"""
if conversation_id:
# Load existing conversation
conversation = db.get(Conversation, conversation_id)
if not conversation:
raise ValueError(f"Conversation {conversation_id} not found")
if conversation.user_id != user_id:
raise ValueError("Conversation does not belong to this user")
return conversation
else:
# Create new conversation
conversation = Conversation(
id=uuid.uuid4(),
user_id=user_id,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
db.add(conversation)
db.commit()
db.refresh(conversation)
return conversation
def load_conversation_history(
db: Session,
conversation_id: uuid.UUID
) -> List[dict[str, str]]:
"""Load conversation history in OpenAI format.
[From]: specs/004-ai-chatbot/plan.md - Conversation History Loading
Args:
db: Database session
conversation_id: Conversation ID to load
Returns:
List of messages in OpenAI format:
[{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
"""
statement = select(Message).where(
Message.conversation_id == conversation_id
).order_by(Message.created_at.asc())
messages = db.exec(statement).all()
# Convert to OpenAI format (role is already a string from database)
conversation_history = [
{"role": msg.role, "content": msg.content}
for msg in messages
]
return conversation_history
def list_user_conversations(
db: Session,
user_id: uuid.UUID,
limit: int = 50,
offset: int = 0
) -> List[Conversation]:
"""List all conversations for a user.
[From]: specs/004-ai-chatbot/spec.md - US2 (Future)
Args:
db: Database session
user_id: User ID
limit: Maximum number of conversations to return
offset: Number of conversations to skip
Returns:
List of conversations ordered by updated_at (most recent first)
"""
statement = select(Conversation).where(
Conversation.user_id == user_id
).order_by(
Conversation.updated_at.desc()
).offset(offset).limit(limit)
conversations = db.exec(statement).all()
return list(conversations)
def update_conversation_timestamp(
db: Session,
conversation_id: uuid.UUID
) -> None:
"""Update conversation's updated_at timestamp.
[From]: specs/004-ai-chatbot/plan.md - Conversation Management
This is called when a new message is added to update the conversation's
position in the user's conversation list.
Args:
db: Database session
conversation_id: Conversation ID to update
"""
conversation = db.get(Conversation, conversation_id)
if conversation:
conversation.updated_at = datetime.utcnow()
db.add(conversation)
db.commit()
|