Spaces:
Sleeping
Sleeping
File size: 3,474 Bytes
d746025 e0b9681 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 404ed37 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 01eb587 d746025 | 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 | import chainlit as cl
import httpx
import asyncio
import os
API_URL = "https://tlong-ds-sentiment-analysis-api.hf.space"
# -----------------------------
# Chat Service (async HTTP calls)
# -----------------------------
class ChatService:
@staticmethod
async def respond_to_chat(message: str) -> str:
"""Send message to backend API and return formatted response"""
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{API_URL}/chat",
data={"message": message}
)
if response.status_code == 200:
raw_response = str(response.text).strip()
if raw_response:
if raw_response.startswith('"') and raw_response.endswith('"'):
raw_response = raw_response[1:-1]
formatted_response = (
raw_response.replace("\\n", "\n")
.replace('\\"', '"')
.replace("\\*", "*")
)
return formatted_response
else:
return "**No response received from the API.**"
else:
return f"**API Error {response.status_code}:**\n\n```\n{response.text}\n```"
except Exception as e:
return f"**Error:** {str(e)}"
# -----------------------------
# Typing animation
# -----------------------------
async def run_typing_animation(msg: cl.Message):
frames = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
idx = 0
try:
while True:
msg.content = f"{frames[idx % len(frames)]} Analyzing sentiment..."
await msg.update()
await asyncio.sleep(0.25)
idx += 1
except asyncio.CancelledError:
# Expected when we cancel after API finishes
raise
# -----------------------------
# Chainlit Handlers
# -----------------------------
@cl.on_chat_start
async def on_chat_start():
"""Show intro message when chat starts"""
await cl.Message(
content=(
"**Sentiment Analysis**\n\n"
"Hi there! 😊 I'm Sen, your friendly sentiment analysis assistant! "
"I'm here to help you understand the emotions and feelings behind your words. "
"Share anything with me and I'll cheerfully analyze the sentiment for you! 🌟\n\n"
"Feel free to ask me about the sentiment of your text. Let's get started! 🚀"
),
author="assistant"
).send()
@cl.on_message
async def main(message: cl.Message):
"""Handle user messages with animation + API call"""
# Placeholder message (starts typing animation)
msg = cl.Message(content="⠋ Analyzing sentiment...", author="assistant")
await msg.send()
# Run animation + API call concurrently
animation_task = asyncio.create_task(run_typing_animation(msg))
api_task = asyncio.create_task(ChatService.respond_to_chat(message.content))
try:
response = await api_task # wait for API result
finally:
animation_task.cancel() # stop animation
try:
await asyncio.wait_for(animation_task, timeout=0.1)
except (asyncio.CancelledError, asyncio.TimeoutError):
pass
# Update final content
msg.content = response
await msg.update() |