File size: 3,780 Bytes
732e77c | 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 | # /agenticcore/chatbot/services.py
from __future__ import annotations
import json
import os
from dataclasses import dataclass
from typing import Dict
# Delegate sentiment to the unified provider layer
# If you put providers_unified.py under agenticcore/chatbot/, change the import to:
# from agenticcore.chatbot.providers_unified import analyze_sentiment
from agenticcore.providers_unified import analyze_sentiment
from ..providers_unified import analyze_sentiment
def _trim(s: str, max_len: int = 2000) -> str:
s = (s or "").strip()
return s if len(s) <= max_len else s[: max_len - 1] + "…"
@dataclass(frozen=True)
class SentimentResult:
label: str # "positive" | "neutral" | "negative" | "mixed" | "unknown"
confidence: float # 0.0 .. 1.0
class ChatBot:
"""
Minimal chatbot that uses provider-agnostic sentiment via providers_unified.
Public API:
- reply(text: str) -> Dict[str, object]
- capabilities() -> Dict[str, object]
"""
def __init__(self, system_prompt: str = "You are a concise helper.") -> None:
self._system_prompt = _trim(system_prompt, 800)
# Expose which provider is intended/active (for diagnostics)
self._mode = os.getenv("AI_PROVIDER") or "auto"
def capabilities(self) -> Dict[str, object]:
"""List what this bot can do."""
return {
"system": "chatbot",
"mode": self._mode, # "auto" or a pinned provider (hf/azure/openai/cohere/deepai/offline)
"features": ["text-input", "sentiment-analysis", "help"],
"commands": {"help": "Describe capabilities and usage."},
}
def reply(self, text: str) -> Dict[str, object]:
"""Produce a reply and sentiment for one user message."""
user = _trim(text)
if not user:
return self._make_response(
"I didn't catch that. Please provide some text.",
SentimentResult("unknown", 0.0),
)
if user.lower() in {"help", "/help"}:
return {"reply": self._format_help(), "capabilities": self.capabilities()}
s = analyze_sentiment(user) # -> {"provider", "label", "score", ...}
sr = SentimentResult(label=str(s.get("label", "neutral")), confidence=float(s.get("score", 0.5)))
return self._make_response(self._compose(sr), sr)
# ---- internals ----
def _format_help(self) -> str:
caps = self.capabilities()
feats = ", ".join(caps["features"])
return f"I can analyze sentiment and respond concisely. Features: {feats}. Send any text or type 'help'."
@staticmethod
def _make_response(reply: str, s: SentimentResult) -> Dict[str, object]:
return {"reply": reply, "sentiment": s.label, "confidence": round(float(s.confidence), 2)}
@staticmethod
def _compose(s: SentimentResult) -> str:
if s.label == "positive":
return "Thanks for sharing. I detected a positive sentiment."
if s.label == "negative":
return "I hear your concern. I detected a negative sentiment."
if s.label == "neutral":
return "Noted. The sentiment appears neutral."
if s.label == "mixed":
return "Your message has mixed signals. Can you clarify?"
return "I could not determine the sentiment. Please rephrase."
# Optional: local REPL for quick manual testing
def _interactive_loop() -> None:
bot = ChatBot()
try:
while True:
msg = input("> ").strip()
if msg.lower() in {"exit", "quit"}:
break
print(json.dumps(bot.reply(msg), ensure_ascii=False))
except (EOFError, KeyboardInterrupt):
pass
if __name__ == "__main__":
_interactive_loop()
|