chore: Finalize cleanup phases (docstrings, utility consolidation, dev app isolation)
This commit is contained in:
@@ -1,12 +1,39 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional, Any
|
||||
from typing import Optional, Any, List
|
||||
from jose import JWTError, jwt
|
||||
from pydantic import BaseModel
|
||||
from langchain_core.messages import BaseMessage
|
||||
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
|
||||
from ea_chatbot.config import Settings
|
||||
from ea_chatbot.history.models import Message
|
||||
|
||||
settings = Settings()
|
||||
|
||||
def map_db_messages_to_langchain(db_messages: List[Message]) -> List[BaseMessage] :
|
||||
"""
|
||||
Converts a list of database Message models to LangChain BaseMessage objects.
|
||||
|
||||
Args:
|
||||
db_messages: List of Message objects from the database.
|
||||
|
||||
Returns:
|
||||
List of HumanMessage, AIMessage, or SystemMessage objects.
|
||||
"""
|
||||
lc_messages: List[BaseMessage] = []
|
||||
|
||||
for m in db_messages:
|
||||
role = m.role.lower()
|
||||
if role == "user":
|
||||
lc_messages.append(HumanMessage(content=m.content))
|
||||
elif role == "assistant":
|
||||
lc_messages.append(AIMessage(content=m.content))
|
||||
elif role == "system":
|
||||
lc_messages.append(SystemMessage(content=m.content))
|
||||
else:
|
||||
# Default to HumanMessage for unknown roles
|
||||
lc_messages.append(HumanMessage(content=m.content))
|
||||
|
||||
return lc_messages
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
||||
"""
|
||||
Create a JWT access token.
|
||||
|
||||
Reference in New Issue
Block a user