Spaces:
Configuration error
Configuration error
File size: 3,258 Bytes
ceaf3bb 662a0bb ceaf3bb 2c03b88 ceaf3bb | 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 | import re
from datetime import datetime
from typing import Dict, List, Any
# Global conversation storage dictionary
SYSTEM_CHART_CONVERSATIONS: Dict[str, Dict[str, Any]] = {}
def save_conversation(extraction_id: str, paper_title: str, conversation_history: List, paper_content: str, paper_type: str) -> str:
"""
Save a conversation to the global conversation storage.
Args:
extraction_id: Unique identifier for the extraction task
paper_title: Title of the paper
conversation_history: List of conversation messages
paper_content: Full content of the paper
paper_type: Type of paper (Experimental, Computational, Mixed)
Returns:
str: Unique conversation key
"""
# Initialize extraction_id entry if it doesn't exist
if extraction_id not in SYSTEM_CHART_CONVERSATIONS:
SYSTEM_CHART_CONVERSATIONS[extraction_id] = {}
# Create a safe key from paper title (matching original behavior)
safe_title = re.sub(r'[^\w\s-]', '', paper_title)[:100]
safe_title = re.sub(r'[-\s]+', '_', safe_title)
conversation_key = f"{safe_title}_{len(SYSTEM_CHART_CONVERSATIONS[extraction_id])}"
# Save the conversation
SYSTEM_CHART_CONVERSATIONS[extraction_id][conversation_key] = {
"paper_title": paper_title,
"conversation_history": conversation_history.copy(), # Make a copy to avoid modification
"paper_content": paper_content,
"paper_type": paper_type,
"created_at": datetime.now().isoformat(),
"last_updated": datetime.now().isoformat()
}
print(f"💾 Saved conversation for paper: {paper_title} (key: {conversation_key})")
return conversation_key
def get_conversations(extraction_id: str) -> Dict[str, Dict[str, Any]]:
"""
Get all conversations for a specific extraction ID.
Args:
extraction_id: Unique identifier for the extraction task
Returns:
Dict of conversation_key -> conversation_data
"""
return SYSTEM_CHART_CONVERSATIONS.get(extraction_id, {})
def get_all_conversations() -> Dict[str, Dict[str, Any]]:
"""
Get all conversations across every extraction ID.
Returns:
Dict of extraction_id -> {conversation_key -> conversation_data}
"""
return SYSTEM_CHART_CONVERSATIONS
def update_conversation(extraction_id: str, conversation_key: str, new_message, response):
"""
Update an existing conversation with a new message and response.
Args:
extraction_id: Unique identifier for the extraction task
conversation_key: Unique conversation key
new_message: New message to add
response: Response to the message
Returns:
bool: True if update was successful, False otherwise
"""
if (extraction_id in SYSTEM_CHART_CONVERSATIONS and
conversation_key in SYSTEM_CHART_CONVERSATIONS[extraction_id]):
conversation = SYSTEM_CHART_CONVERSATIONS[extraction_id][conversation_key]
conversation["conversation_history"].append(new_message)
conversation["conversation_history"].append(response)
conversation["last_updated"] = datetime.now().isoformat()
return True
return False |