Spaces:
Running
Running
File size: 3,365 Bytes
7552a11 7e9c3a6 7552a11 7e9c3a6 30664f4 7e9c3a6 7552a11 30664f4 7552a11 7e9c3a6 30664f4 7552a11 7e9c3a6 7552a11 7e9c3a6 2cc1487 7552a11 7e9c3a6 7552a11 d6098e4 |
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 |
from transformers import pipeline
from fastapi import FastAPI, Request
import uvicorn
from uagents import Agent, Context, Bureau, Model
# ─── uAgent I/O MODEL ─────────────────────────────────────────────────────
class TextInput(Model):
text: str
# ─── LOAD EMOTION PIPELINE ─────────────────────────────────────────────────
emotion_model = pipeline(
"text-classification",
model="bhadresh-savani/distilbert-base-uncased-emotion"
)
# ─── CUSTOM ANALYSIS LOGIC ─────────────────────────────────────────────────
def analyze_text_metrics(text):
results = emotion_model(text)
t = text.lower()
suicide_keywords = ["kill myself", "suicidal", "die", "ending it", "pills", "overdose", "no way out"]
psychosis_keywords = ["voices", "hallucinate", "not real", "they’re watching me", "i’m not me"]
metrics = {
"self_harm": 0.0,
"homicidal": 0.0,
"distress": 0.0,
"psychosis": 0.0
}
# aggregate raw scores
for res in results:
label = res["label"]
score = res["score"]
if label == "sadness":
metrics["self_harm"] += score
metrics["distress"] += score * 0.6
elif label in ("anger", "fear"):
metrics["homicidal"] += score
metrics["distress"] += score * 0.5
elif label == "joy":
metrics["psychosis"] += score * 0.3
elif label == "surprise":
metrics["psychosis"] += score * 0.5
# keyword overrides
if any(kw in t for kw in suicide_keywords):
metrics["self_harm"] = max(metrics["self_harm"], 0.8)
if any(kw in t for kw in psychosis_keywords):
metrics["psychosis"] = max(metrics["psychosis"], 0.8)
# clamp into [0.01, 0.99]
for k in metrics:
val = metrics[k]
clamped = max(min(val, 0.99), 0.01)
metrics[k] = round(clamped, 2)
return metrics
# ─── uAgent DEFINITION ────────────────────────────────────────────────────
agent = Agent(name="sentiment_agent")
@agent.on_message(model=TextInput)
async def handle_message(ctx: Context, sender: str, msg: TextInput):
flags = analyze_text_metrics(msg.text)
await ctx.send(sender, flags)
# ─── FASTAPI HTTP ENDPOINT ───────────────────────────────────────────────
app = FastAPI()
@app.post("/")
async def analyze_text(request: Request):
data = await request.json()
return analyze_text_metrics(data.get("text", ""))
# ─── START BOTH ───────────────────────────────────────────────────────────
if __name__ == "__main__":
bureau = Bureau()
bureau.add(agent)
bureau.run_in_thread() # serve the agent on Agentverse
uvicorn.run(app, host="0.0.0.0", port=8000)
|