From 431acd9e5d1586770f3ddb78255a10ed260404a9 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 9 Mar 2026 00:14:57 +0000 Subject: [PATCH 1/2] docs: add Cursor Cloud specific instructions to AGENTS.md Co-authored-by: rockrxf78 --- AGENTS.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 9f3617ce9c..fa33a21c3e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,3 +32,27 @@ Runs on `http://localhost:3000` by default. 1. Title format: use conventional commit messages 2. Use English to write PR title and descriptions. + +## Cursor Cloud specific instructions + +### Services overview + +| Service | Port | Start command | +|---|---|---| +| AstrBot Core (Python backend) | 6185 | `uv run main.py` | +| Dashboard (Vue dev server) | 3000 | `cd dashboard && pnpm dev` | + +The backend uses embedded SQLite — no external database needed. + +### Running the application + +1. **Backend**: `uv run main.py` starts the API server on `http://localhost:6185`. Default credentials: username `astrbot`, password `astrbot`. On first login, a password change is mandatory. +2. **Dashboard dev server**: `cd dashboard && pnpm dev` proxies `/api` to `localhost:6185`. Only needed when developing the Vue frontend. + +### Gotchas + +- `uv` must be on `PATH`. It is installed to `~/.local/bin` which is added via `~/.local/bin/env`. +- After `pnpm install` in `dashboard/`, pnpm v10+ blocks build scripts for `esbuild` and `vue-demi`. Run `pnpm install` with the `--config.confirmModulesPurge=false` flag, or configure `pnpm.onlyBuiltDependencies` in `package.json` to allowlist `["esbuild","vue-demi"]` to unblock the Vite bundler. +- Lint: `uv run ruff format --check .` and `uv run ruff check .`. +- Tests: `TESTING=true uv run pytest -q tests/` (665 tests). Create `data/{plugins,config,temp,skills}` dirs before running tests. +- PR validation: `make pr-test-neo` (quick) or `make pr-test-full` (comprehensive). See `CONTRIBUTING.md`. From 33ad8bb92fe5309514add1a629cf31041e7617bf Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 9 Mar 2026 00:39:43 +0000 Subject: [PATCH 2/2] feat: add smart customer service plugin with AI-first sales skill integration - 4-state session machine (idle/ai_serving/waiting_human/human_connected/closed) - ConversationContextManager with LRU eviction and LLM context injection - CustomerProfileManager with SQLite persistence and lead tracking - ProductManager with JSON-based product catalog and search - EscalationManager with queue management for human agent handoff - 5 LLM Tools: query_product, check_order, escalate_to_human, update_customer_profile, create_lead - Sales Skill template (SKILL.md) for AI behavior guidance - MessageRouter for session-state-based message routing - TimeoutManager for conversation and queue timeouts - Dashboard-friendly configuration schema - 31 unit tests covering all managers Co-authored-by: rockrxf78 --- .tasks/2026-03-09_smart-customer-service.md | 40 ++ .../README.md | 72 +++ .../_conf_schema.json | 39 ++ .../helpers/__init__.py | 7 + .../helpers/help_text_builder.py | 30 + .../helpers/message_router.py | 108 ++++ astrbot_plugin_smart_customer_service/main.py | 542 ++++++++++++++++++ .../managers/__init__.py | 17 + .../managers/conversation_context_manager.py | 82 +++ .../managers/customer_profile_manager.py | 170 ++++++ .../managers/escalation_manager.py | 90 +++ .../managers/product_manager.py | 161 ++++++ .../managers/session_manager.py | 169 ++++++ .../managers/timeout_manager.py | 64 +++ .../metadata.yaml | 6 + .../skills/sales_skill_template/SKILL.md | 59 ++ .../tools/__init__.py | 0 tests/test_smart_customer_service.py | 305 ++++++++++ 18 files changed, 1961 insertions(+) create mode 100644 .tasks/2026-03-09_smart-customer-service.md create mode 100644 astrbot_plugin_smart_customer_service/README.md create mode 100644 astrbot_plugin_smart_customer_service/_conf_schema.json create mode 100644 astrbot_plugin_smart_customer_service/helpers/__init__.py create mode 100644 astrbot_plugin_smart_customer_service/helpers/help_text_builder.py create mode 100644 astrbot_plugin_smart_customer_service/helpers/message_router.py create mode 100644 astrbot_plugin_smart_customer_service/main.py create mode 100644 astrbot_plugin_smart_customer_service/managers/__init__.py create mode 100644 astrbot_plugin_smart_customer_service/managers/conversation_context_manager.py create mode 100644 astrbot_plugin_smart_customer_service/managers/customer_profile_manager.py create mode 100644 astrbot_plugin_smart_customer_service/managers/escalation_manager.py create mode 100644 astrbot_plugin_smart_customer_service/managers/product_manager.py create mode 100644 astrbot_plugin_smart_customer_service/managers/session_manager.py create mode 100644 astrbot_plugin_smart_customer_service/managers/timeout_manager.py create mode 100644 astrbot_plugin_smart_customer_service/metadata.yaml create mode 100644 astrbot_plugin_smart_customer_service/skills/sales_skill_template/SKILL.md create mode 100644 astrbot_plugin_smart_customer_service/tools/__init__.py create mode 100644 tests/test_smart_customer_service.py diff --git a/.tasks/2026-03-09_smart-customer-service.md b/.tasks/2026-03-09_smart-customer-service.md new file mode 100644 index 0000000000..c971d70414 --- /dev/null +++ b/.tasks/2026-03-09_smart-customer-service.md @@ -0,0 +1,40 @@ +# Background +File: 2026-03-09_smart-customer-service +Created: 2026-03-09 +Task Branch: cursor/development-environment-setup-880d +Yolo Mode: On + +# Task Description +Develop an intelligent customer service plugin (astrbot_plugin_smart_customer_service) for AstrBot, based on the Sales Skill system. Combines AI-powered auto-response with human agent escalation. + +# Reference Plugins +- astrbot_plugin_astrbot_enhance_mode (by Axi404): LLM deep integration, Memory RAG, LLM Tools, output tag parsing +- astrbot_plugin_human_service (by Zhalslar): Human service queue, session routing, blacklist, timeout + +# Architecture +- AI-first, human-fallback hybrid model +- 4-state session machine: idle → ai_serving → waiting_human → human_connected → closed +- Sales Skill (SKILL.md) for LLM behavior guidance +- LLM Tools for product query, order check, escalation, customer profiling +- Three-layer design: Entry → Managers → Helpers/Tools + +# Implementation Checklist +1. [x] Save research report and development plan +2. [ ] Create plugin skeleton (directory, metadata.yaml, _conf_schema.json) +3. [ ] Implement SessionManager (4-state machine + SQLite) +4. [ ] Implement ConversationContextManager (history + LLM injection) +5. [ ] Implement CustomerProfileManager (CRUD + SQLite) +6. [ ] Implement ProductManager (product data + query) +7. [ ] Implement LLM Tools (query_product, check_order, escalate_to_human, update_customer_profile, create_lead) +8. [ ] Implement EscalationManager (queue + agent accept) +9. [ ] Implement MessageRouter + TimeoutManager +10. [ ] Implement main.py entry (event hooks + commands) +11. [ ] Write Sales Skill template (SKILL.md) +12. [ ] Write _conf_schema.json + HelpTextBuilder + README +13. [ ] Integration test +14. [ ] Lint, format, commit and push + +# Task Progress +[2026-03-09 00:20:00] +- Created: Research report and development plan +- Status: Confirmed diff --git a/astrbot_plugin_smart_customer_service/README.md b/astrbot_plugin_smart_customer_service/README.md new file mode 100644 index 0000000000..37d99366df --- /dev/null +++ b/astrbot_plugin_smart_customer_service/README.md @@ -0,0 +1,72 @@ +# AstrBot Smart Customer Service Plugin + +AI-powered intelligent customer service plugin with sales skill integration, customer profiling, and human agent escalation. + +## Features + +- **AI-First**: Uses LLM with Sales Skill to handle customer inquiries automatically +- **Human Escalation**: Seamless transfer to human agents with queue management +- **Customer Profiling**: SQLite-backed customer profiles with tags and lead tracking +- **Product Knowledge**: JSON-based product catalog with search capabilities +- **LLM Tools**: AI can query products, check orders, create leads, and escalate +- **Session Management**: 4-state session machine (idle → ai_serving → waiting_human → human_connected) +- **Configurable**: Dashboard-friendly configuration schema + +## Commands + +### User Commands +| Command | Description | +|---------|-------------| +| `/ask ` | Ask the AI assistant | +| `/transfer_human` | Request a human agent | +| `/cancel_queue` | Cancel queue for human agent | +| `/queue_status` | Check queue position | +| `/end_service` | End current session | +| `/cs_help` | Show help | + +### Agent Commands +| Command | Description | +|---------|-------------| +| `/accept_customer` | Accept next customer in queue | +| `/end_service` | End current customer session | +| `/queue_list` | View waiting queue | +| `/customer_info` | View current customer profile | +| `/cs_help` | Show help | + +## Configuration + +Configure via AstrBot Dashboard or `_conf_schema.json`: + +- `agent_ids`: List of human agent user IDs +- `agent_names`: Display names for agents +- `max_sessions`: Max concurrent sessions (default: 200) +- `max_context_messages`: Max messages per session (default: 50) +- `conversation_timeout`: Auto-close timeout in seconds (0 = disabled) +- `queue_timeout`: Queue timeout in seconds (0 = disabled) + +## Architecture + +``` +main.py (Entry + Event Hooks + LLM Tools) + ├── managers/ + │ ├── session_manager.py (4-state session machine) + │ ├── conversation_context_manager.py (LLM context injection) + │ ├── customer_profile_manager.py (SQLite profiles) + │ ├── product_manager.py (JSON product catalog) + │ ├── escalation_manager.py (Queue management) + │ └── timeout_manager.py (Timeout tracking) + ├── helpers/ + │ ├── message_router.py (Message routing) + │ └── help_text_builder.py (Help text) + └── skills/ + └── sales_skill_template/SKILL.md (Sales behavior guide) +``` + +## Sales Skill + +The plugin includes a Sales Skill template (`SKILL.md`) that guides the AI to: +1. Greet customers and discover needs +2. Recommend products using `query_product` tool +3. Handle objections with factual information +4. Capture leads with `create_lead` tool +5. Escalate to humans when appropriate diff --git a/astrbot_plugin_smart_customer_service/_conf_schema.json b/astrbot_plugin_smart_customer_service/_conf_schema.json new file mode 100644 index 0000000000..d9fd86824f --- /dev/null +++ b/astrbot_plugin_smart_customer_service/_conf_schema.json @@ -0,0 +1,39 @@ +{ + "agent_ids": { + "description": "List of human agent user IDs who can accept customer sessions", + "type": "list", + "default": [], + "item_type": "string" + }, + "agent_names": { + "description": "Display names for agents (same order as agent_ids)", + "type": "list", + "default": [], + "item_type": "string" + }, + "max_sessions": { + "description": "Maximum number of concurrent sessions to track", + "type": "int", + "default": 200 + }, + "max_context_messages": { + "description": "Maximum conversation messages to keep per session", + "type": "int", + "default": 50 + }, + "conversation_timeout": { + "description": "Auto-close session after this many seconds (0 = disabled)", + "type": "int", + "default": 0 + }, + "timeout_warning_seconds": { + "description": "Send warning this many seconds before timeout", + "type": "int", + "default": 120 + }, + "queue_timeout": { + "description": "Remove from queue after this many seconds (0 = disabled)", + "type": "int", + "default": 0 + } +} diff --git a/astrbot_plugin_smart_customer_service/helpers/__init__.py b/astrbot_plugin_smart_customer_service/helpers/__init__.py new file mode 100644 index 0000000000..20bfd59c95 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/helpers/__init__.py @@ -0,0 +1,7 @@ +from .help_text_builder import HelpTextBuilder +from .message_router import MessageRouter + +__all__ = [ + "HelpTextBuilder", + "MessageRouter", +] diff --git a/astrbot_plugin_smart_customer_service/helpers/help_text_builder.py b/astrbot_plugin_smart_customer_service/helpers/help_text_builder.py new file mode 100644 index 0000000000..df5bb92483 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/helpers/help_text_builder.py @@ -0,0 +1,30 @@ +"""Builds help text for users and agents.""" + +from __future__ import annotations + + +class HelpTextBuilder: + @staticmethod + def build_user_help() -> str: + return ( + "=== Smart Customer Service Help ===\n" + "Available commands:\n" + " /ask - Ask AI assistant a question\n" + " /transfer_human - Request a human agent\n" + " /cancel_queue - Cancel queue for human agent\n" + " /queue_status - Check queue position\n" + " /end_service - End current service session\n" + " /cs_help - Show this help message\n" + ) + + @staticmethod + def build_agent_help() -> str: + return ( + "=== Agent Help ===\n" + "Available commands:\n" + " /accept_customer - Accept next customer in queue\n" + " /end_service - End current customer session\n" + " /queue_list - View waiting queue\n" + " /customer_info - View current customer profile\n" + " /cs_help - Show this help message\n" + ) diff --git a/astrbot_plugin_smart_customer_service/helpers/message_router.py b/astrbot_plugin_smart_customer_service/helpers/message_router.py new file mode 100644 index 0000000000..d046186a90 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/helpers/message_router.py @@ -0,0 +1,108 @@ +"""Routes messages between customers and human agents.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from astrbot.api.event import AstrMessageEvent + +from ..managers.session_manager import SessionState + +if TYPE_CHECKING: + from ..main import SmartCustomerServicePlugin + +logger = logging.getLogger("astrbot") + + +class MessageRouter: + """Decides how to handle an incoming message based on session state.""" + + def __init__(self, plugin: SmartCustomerServicePlugin) -> None: + self.plugin = plugin + + async def route(self, event: AstrMessageEvent) -> bool: + """Route the message. Returns True if handled (event should stop).""" + sender_id = str(event.get_sender_id()) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + if self.plugin.escalation_mgr.is_agent(sender_id): + return await self._route_agent_message(event, sender_id) + + session = self.plugin.session_mgr.get_session(platform, group_id, sender_id) + if session is None: + return False + + if session.state == SessionState.AI_SERVING: + return False + + if session.state == SessionState.HUMAN_CONNECTED: + return await self._forward_to_agent(event, session) + + if session.state == SessionState.WAITING_HUMAN: + await event.send( + event.make_result().message( + "You are currently in queue for a human agent. " + "Please wait or type /cancel_queue to cancel." + ) + ) + event.stop_event() + return True + + return False + + async def _forward_to_agent(self, event: AstrMessageEvent, session) -> bool: + """Forward customer message to the connected agent.""" + if not session.agent_id: + return False + agent_name = self.plugin.escalation_mgr.get_agent_name(session.agent_id) + user_name = session.user_name or str(event.get_sender_id()) + content = event.message_str + self.plugin.context_mgr.append( + session.session_key, + "user", + content, + sender_name=user_name, + ) + try: + msg_chain = [ + {"type": "text", "data": f"[Customer: {user_name}]\n{content}"} + ] + await self.plugin.context.send_message( + event.session, + msg_chain, + ) + logger.info(f"Forwarded message from {user_name} to agent {agent_name}") + except Exception as e: + logger.error(f"Failed to forward message to agent: {e}") + event.stop_event() + return True + + async def _route_agent_message( + self, event: AstrMessageEvent, agent_id: str + ) -> bool: + """Handle message from a human agent, forward to connected customer.""" + sessions = self.plugin.session_mgr.get_sessions_by_agent(agent_id) + if not sessions: + return False + + session = sessions[0] + content = event.message_str + agent_name = self.plugin.escalation_mgr.get_agent_name(agent_id) + self.plugin.context_mgr.append( + session.session_key, + "agent", + content, + sender_name=agent_name, + ) + try: + msg_chain = [{"type": "text", "data": f"[Agent: {agent_name}]\n{content}"}] + await self.plugin.context.send_message( + event.session, + msg_chain, + ) + except Exception as e: + logger.error(f"Failed to forward agent message: {e}") + event.stop_event() + return True diff --git a/astrbot_plugin_smart_customer_service/main.py b/astrbot_plugin_smart_customer_service/main.py new file mode 100644 index 0000000000..f5115e9366 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/main.py @@ -0,0 +1,542 @@ +"""Smart Customer Service Plugin - AI-first with human escalation.""" + +from __future__ import annotations + +import logging +from pathlib import Path + +from astrbot.api import star +from astrbot.api.event import AstrMessageEvent, filter +from astrbot.core.utils.astrbot_path import get_astrbot_data_path + +from .helpers.help_text_builder import HelpTextBuilder +from .helpers.message_router import MessageRouter +from .managers.conversation_context_manager import ConversationContextManager +from .managers.customer_profile_manager import CustomerProfileManager +from .managers.escalation_manager import EscalationManager +from .managers.product_manager import ProductManager +from .managers.session_manager import SessionManager, SessionState +from .managers.timeout_manager import TimeoutManager + +logger = logging.getLogger("astrbot") + +PLUGIN_NAME = "astrbot_plugin_smart_customer_service" + + +class SmartCustomerServicePlugin(star.Star): + def __init__(self, context: star.Context, config: dict | None = None) -> None: + super().__init__(context, config) + self.context = context + self.config = config or {} + + plugin_data_dir = Path(get_astrbot_data_path()) / "plugin_data" / PLUGIN_NAME + plugin_data_dir.mkdir(parents=True, exist_ok=True) + + self.session_mgr = SessionManager() + self.context_mgr = ConversationContextManager( + max_sessions=self.config.get("max_sessions", 200), + max_messages=self.config.get("max_context_messages", 50), + ) + self.profile_mgr = CustomerProfileManager( + db_path=plugin_data_dir / "customer_profiles.db", + ) + self.product_mgr = ProductManager( + data_path=plugin_data_dir / "products.json", + ) + self.escalation_mgr = EscalationManager( + queue_timeout=self.config.get("queue_timeout", 0), + ) + self.timeout_mgr = TimeoutManager( + conversation_timeout=self.config.get("conversation_timeout", 0), + warning_seconds=self.config.get("timeout_warning_seconds", 120), + ) + self.router = MessageRouter(self) + + agent_ids = self.config.get("agent_ids", []) + agent_names = self.config.get("agent_names", []) + if agent_ids: + self.escalation_mgr.set_agents(agent_ids, agent_names) + + async def initialize(self) -> None: + logger.info(f"{PLUGIN_NAME} initialized.") + + async def terminate(self) -> None: + logger.info(f"{PLUGIN_NAME} terminated.") + + # ==================== Event Handlers ==================== + + @filter.event_message_type(filter.EventMessageType.ALL, priority=10) + async def on_message(self, event: AstrMessageEvent): + """Main message handler - routes based on session state.""" + sender_id = str(event.get_sender_id()) + sender_name = str(event.get_sender_name() or sender_id) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + self._check_timeouts(event) + + if await self.router.route( + event, + ): + return + + session = self.session_mgr.get_session(platform, group_id, sender_id) + if session and session.state == SessionState.AI_SERVING: + self.profile_mgr.record_interaction(sender_id, sender_name) + self.context_mgr.append( + session.session_key, + "user", + event.message_str, + sender_name=sender_name, + ) + + @filter.on_llm_request() + async def inject_service_context(self, event: AstrMessageEvent, req): + """Inject customer profile and conversation history into LLM request.""" + sender_id = str(event.get_sender_id()) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_session(platform, group_id, sender_id) + if session is None or session.state != SessionState.AI_SERVING: + return + + context_parts = [] + + profile = self.profile_mgr.get(sender_id) + if profile: + context_parts.append(self._format_profile_context(profile)) + + history = self.context_mgr.get_history_text(session.session_key, max_items=15) + if history: + context_parts.append(f"[Conversation History]\n{history}") + + if context_parts: + injection = "\n\n".join(context_parts) + if hasattr(req, "system_prompt") and req.system_prompt: + req.system_prompt += f"\n\n{injection}" + elif hasattr(req, "system_prompt"): + req.system_prompt = injection + + @filter.on_llm_response() + async def record_ai_response(self, event: AstrMessageEvent, resp): + """Record AI response in conversation context.""" + sender_id = str(event.get_sender_id()) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_session(platform, group_id, sender_id) + if session is None or session.state != SessionState.AI_SERVING: + return + + completion_text = "" + if hasattr(resp, "completion_text"): + completion_text = resp.completion_text or "" + elif hasattr(resp, "text"): + completion_text = resp.text or "" + if completion_text: + self.context_mgr.append( + session.session_key, + "assistant", + completion_text, + sender_name="AI Assistant", + ) + + # ==================== Commands ==================== + + @filter.command("ask") + async def cmd_ask(self, event: AstrMessageEvent): + """Start or continue an AI service session.""" + sender_id = str(event.get_sender_id()) + sender_name = str(event.get_sender_name() or sender_id) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_or_create( + platform, group_id, sender_id, sender_name + ) + if not session.is_active: + self.session_mgr.start_ai_serving(session) + self.profile_mgr.record_interaction(sender_id, sender_name) + self.timeout_mgr.start(session.session_key) + + self.context_mgr.append( + session.session_key, + "user", + event.message_str, + sender_name=sender_name, + ) + + yield event.request_llm( + prompt=event.message_str, + ) + + @filter.command("transfer_human") + async def cmd_transfer_human(self, event: AstrMessageEvent): + """Request transfer to a human agent.""" + sender_id = str(event.get_sender_id()) + sender_name = str(event.get_sender_name() or sender_id) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_or_create( + platform, group_id, sender_id, sender_name + ) + + if session.state == SessionState.HUMAN_CONNECTED: + yield event.plain_result("You are already connected to a human agent.") + return + + if session.state == SessionState.WAITING_HUMAN: + pos = self.escalation_mgr.get_position(session) + yield event.plain_result(f"You are already in queue. Position: {pos}") + return + + if session.state == SessionState.AI_SERVING: + self.session_mgr.request_human(session) + elif session.state in (SessionState.IDLE, SessionState.CLOSED): + self.session_mgr.start_ai_serving(session) + self.session_mgr.request_human(session) + + position = self.escalation_mgr.enqueue(session, reason="User requested") + summary = self.context_mgr.get_summary(session.session_key) + + agents = self.escalation_mgr.get_agents() + agent_info = ( + f"Available agents: {len(agents)}" + if agents + else "No agents currently configured." + ) + + yield event.plain_result( + f"Transfer requested. You are #{position} in queue.\n{agent_info}\n" + "A human agent will be with you shortly." + ) + + logger.info( + f"User {sender_name} ({sender_id}) requested human agent. " + f"Queue position: {position}. Summary: {summary[:100]}" + ) + + @filter.command("cancel_queue") + async def cmd_cancel_queue(self, event: AstrMessageEvent): + """Cancel queue for human agent.""" + sender_id = str(event.get_sender_id()) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_session(platform, group_id, sender_id) + if session is None or session.state != SessionState.WAITING_HUMAN: + yield event.plain_result("You are not in queue.") + return + + self.escalation_mgr.dequeue(session) + self.session_mgr.back_to_ai(session) + yield event.plain_result("Queue cancelled. You are now back to AI assistance.") + + @filter.command("queue_status") + async def cmd_queue_status(self, event: AstrMessageEvent): + """Check queue position.""" + sender_id = str(event.get_sender_id()) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_session(platform, group_id, sender_id) + if session is None or session.state != SessionState.WAITING_HUMAN: + yield event.plain_result("You are not currently in queue.") + return + + pos = self.escalation_mgr.get_position(session) + total = self.escalation_mgr.get_queue_size() + yield event.plain_result(f"Queue position: {pos}/{total}") + + @filter.command("accept_customer") + async def cmd_accept_customer(self, event: AstrMessageEvent): + """Agent accepts next customer in queue.""" + agent_id = str(event.get_sender_id()) + + if not self.escalation_mgr.is_agent(agent_id): + yield event.plain_result("You are not registered as an agent.") + return + + if self.session_mgr.is_agent_busy(agent_id): + yield event.plain_result( + "You already have an active customer session. " + "End it first with /end_service." + ) + return + + entry = self.escalation_mgr.pop_next() + if entry is None: + yield event.plain_result("No customers in queue.") + return + + session = entry.session + agent_name = self.escalation_mgr.get_agent_name(agent_id) + self.session_mgr.connect_agent(session, agent_id, agent_name) + self.timeout_mgr.start(session.session_key) + + summary = self.context_mgr.get_summary(session.session_key) + + yield event.plain_result( + f"Connected to customer: {session.user_name} ({session.user_id})\n" + f"Reason: {entry.reason}\n\n{summary}" + ) + + @filter.command("end_service") + async def cmd_end_service(self, event: AstrMessageEvent): + """End current service session (user or agent).""" + sender_id = str(event.get_sender_id()) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + if self.escalation_mgr.is_agent(sender_id): + sessions = self.session_mgr.get_sessions_by_agent(sender_id) + if sessions: + session = sessions[0] + self.session_mgr.close_session(session) + self.timeout_mgr.stop(session.session_key) + yield event.plain_result(f"Session with {session.user_name} ended.") + return + yield event.plain_result("No active customer session.") + return + + session = self.session_mgr.get_session(platform, group_id, sender_id) + if session is None or not session.is_active: + yield event.plain_result("No active service session.") + return + + self.escalation_mgr.dequeue(session) + self.session_mgr.close_session(session) + self.timeout_mgr.stop(session.session_key) + yield event.plain_result("Service session ended. Thank you for your inquiry!") + + @filter.command("queue_list") + async def cmd_queue_list(self, event: AstrMessageEvent): + """View waiting queue (agent only).""" + agent_id = str(event.get_sender_id()) + if not self.escalation_mgr.is_agent(agent_id): + yield event.plain_result("You are not registered as an agent.") + return + + queue = self.escalation_mgr.get_queue_list() + if not queue: + yield event.plain_result("Queue is empty.") + return + + lines = [f"Waiting queue ({len(queue)}):"] + for i, entry in enumerate(queue, 1): + lines.append( + f" {i}. {entry.session.user_name} - {entry.reason or 'No reason'}" + ) + yield event.plain_result("\n".join(lines)) + + @filter.command("customer_info") + async def cmd_customer_info(self, event: AstrMessageEvent): + """View current customer profile (agent only).""" + agent_id = str(event.get_sender_id()) + if not self.escalation_mgr.is_agent(agent_id): + yield event.plain_result("You are not registered as an agent.") + return + + sessions = self.session_mgr.get_sessions_by_agent(agent_id) + if not sessions: + yield event.plain_result("No active customer session.") + return + + session = sessions[0] + profile = self.profile_mgr.get(session.user_id) + if profile is None: + yield event.plain_result( + f"Customer: {session.user_name} ({session.user_id})\nNo profile data." + ) + return + + info = self._format_profile_display(profile) + yield event.plain_result(info) + + @filter.command("cs_help") + async def cmd_help(self, event: AstrMessageEvent): + """Show help text.""" + sender_id = str(event.get_sender_id()) + if self.escalation_mgr.is_agent(sender_id): + yield event.plain_result(HelpTextBuilder.build_agent_help()) + else: + yield event.plain_result(HelpTextBuilder.build_user_help()) + + # ==================== LLM Tools ==================== + + @filter.llm_tool(name="query_product") + async def tool_query_product( + self, event: AstrMessageEvent, keyword: str, category: str = "" + ) -> str: + """Search for product information by keyword and optional category. + + Args: + keyword(string): Search keyword for product name, description or features. + category(string): Optional category filter. + """ + results = self.product_mgr.search(keyword, category) + if not results: + return f"No products found matching '{keyword}'." + lines = [] + for p in results[:5]: + lines.append(self.product_mgr.format_product_info(p)) + return "\n\n".join(lines) + + @filter.llm_tool(name="check_order") + async def tool_check_order(self, event: AstrMessageEvent, order_id: str) -> str: + """Check order status by order ID. + + Args: + order_id(string): The order ID to look up. + """ + return ( + f"Order {order_id}: Status information is not available in demo mode. " + "Please configure an order system integration to enable this feature." + ) + + @filter.llm_tool(name="escalate_to_human") + async def tool_escalate_to_human( + self, event: AstrMessageEvent, reason: str = "" + ) -> str: + """Transfer the current conversation to a human agent. Use this when the customer request cannot be handled by AI. + + Args: + reason(string): Reason for escalation. + """ + sender_id = str(event.get_sender_id()) + sender_name = str(event.get_sender_name() or sender_id) + group_id = str(getattr(event, "get_group_id", lambda: "")() or "") + platform = event.session.platform_id + + session = self.session_mgr.get_or_create( + platform, group_id, sender_id, sender_name + ) + + if session.state == SessionState.HUMAN_CONNECTED: + return "Customer is already connected to a human agent." + + if session.state == SessionState.WAITING_HUMAN: + return "Customer is already in queue for a human agent." + + if session.state == SessionState.AI_SERVING: + self.session_mgr.request_human(session) + else: + self.session_mgr.start_ai_serving(session) + self.session_mgr.request_human(session) + + position = self.escalation_mgr.enqueue( + session, reason=reason or "AI escalation" + ) + return ( + f"Transfer initiated. Customer is #{position} in queue. " + f"Reason: {reason or 'AI determined human assistance needed.'}" + ) + + @filter.llm_tool(name="update_customer_profile") + async def tool_update_profile( + self, + event: AstrMessageEvent, + tag: str = "", + notes: str = "", + lead_status: str = "", + ) -> str: + """Update customer profile with tags, notes, or lead status. + + Args: + tag(string): A tag to add to the customer profile (e.g. 'high_intent', 'price_sensitive'). + notes(string): Notes about the customer interaction. + lead_status(string): Lead status: none, interested, contacted, converted. + """ + sender_id = str(event.get_sender_id()) + sender_name = str(event.get_sender_name() or sender_id) + profile = self.profile_mgr.get_or_create(sender_id, sender_name) + + updates = [] + if tag: + self.profile_mgr.add_tag(sender_id, tag) + updates.append(f"Added tag: {tag}") + if notes: + existing = profile.notes + new_notes = f"{existing}\n{notes}" if existing else notes + self.profile_mgr.update_notes(sender_id, new_notes) + updates.append("Notes updated") + if lead_status and lead_status in ( + "none", + "interested", + "contacted", + "converted", + ): + self.profile_mgr.set_lead_status(sender_id, lead_status) + updates.append(f"Lead status: {lead_status}") + + if not updates: + return "No updates specified." + return "Customer profile updated: " + ", ".join(updates) + + @filter.llm_tool(name="create_lead") + async def tool_create_lead( + self, + event: AstrMessageEvent, + product_interest: str = "", + notes: str = "", + ) -> str: + """Create a sales lead for the current customer when they express purchase intent. + + Args: + product_interest(string): The product the customer is interested in. + notes(string): Additional notes about the lead. + """ + sender_id = str(event.get_sender_id()) + sender_name = str(event.get_sender_name() or sender_id) + profile = self.profile_mgr.get_or_create(sender_id, sender_name) + + self.profile_mgr.set_lead_status(sender_id, "interested") + if product_interest: + self.profile_mgr.add_tag(sender_id, f"interest:{product_interest}") + if notes: + existing = profile.notes + lead_note = f"[Lead] {notes}" + new_notes = f"{existing}\n{lead_note}" if existing else lead_note + self.profile_mgr.update_notes(sender_id, new_notes) + + return ( + f"Sales lead created for {sender_name}. " + f"Product interest: {product_interest or 'General'}. " + f"Lead status set to 'interested'." + ) + + # ==================== Internal Helpers ==================== + + def _check_timeouts(self, event: AstrMessageEvent) -> None: + timed_out, warnings = self.timeout_mgr.check_timeouts() + for key in warnings: + remaining = self.timeout_mgr.get_remaining(key) + logger.info(f"Session {key} timeout warning: {remaining}s remaining") + for key in timed_out: + logger.info(f"Session {key} timed out") + + @staticmethod + def _format_profile_context(profile) -> str: + parts = [f"[Customer Profile: {profile.name or profile.customer_id}]"] + if profile.tags: + parts.append(f"Tags: {', '.join(profile.tags)}") + if profile.lead_status != "none": + parts.append(f"Lead Status: {profile.lead_status}") + if profile.notes: + parts.append(f"Notes: {profile.notes[-200:]}") + parts.append(f"Interactions: {profile.interaction_count}") + return "\n".join(parts) + + @staticmethod + def _format_profile_display(profile) -> str: + return ( + f"=== Customer Profile ===\n" + f"ID: {profile.customer_id}\n" + f"Name: {profile.name}\n" + f"Tags: {', '.join(profile.tags) if profile.tags else 'None'}\n" + f"Lead Status: {profile.lead_status}\n" + f"Interactions: {profile.interaction_count}\n" + f"Notes: {profile.notes or 'None'}\n" + ) diff --git a/astrbot_plugin_smart_customer_service/managers/__init__.py b/astrbot_plugin_smart_customer_service/managers/__init__.py new file mode 100644 index 0000000000..96453c9ff0 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/__init__.py @@ -0,0 +1,17 @@ +from .conversation_context_manager import ConversationContextManager +from .customer_profile_manager import CustomerProfileManager +from .escalation_manager import EscalationManager +from .product_manager import ProductManager +from .session_manager import ServiceSession, SessionManager, SessionState +from .timeout_manager import TimeoutManager + +__all__ = [ + "ConversationContextManager", + "CustomerProfileManager", + "EscalationManager", + "ProductManager", + "ServiceSession", + "SessionManager", + "SessionState", + "TimeoutManager", +] diff --git a/astrbot_plugin_smart_customer_service/managers/conversation_context_manager.py b/astrbot_plugin_smart_customer_service/managers/conversation_context_manager.py new file mode 100644 index 0000000000..bd6b33de65 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/conversation_context_manager.py @@ -0,0 +1,82 @@ +"""Manages per-session conversation context for LLM injection.""" + +from __future__ import annotations + +import logging +import time +from collections import OrderedDict +from dataclasses import dataclass, field + +logger = logging.getLogger("astrbot") + + +@dataclass +class ChatMessage: + role: str + content: str + sender_name: str = "" + timestamp: float = field(default_factory=time.time) + + +class ConversationContextManager: + """Maintains conversation history per session key, with LRU eviction.""" + + def __init__(self, max_sessions: int = 200, max_messages: int = 50) -> None: + self._histories: OrderedDict[str, list[ChatMessage]] = OrderedDict() + self._max_sessions = max_sessions + self._max_messages = max_messages + + def _touch(self, key: str) -> None: + self._histories.move_to_end(key, last=True) + while len(self._histories) > self._max_sessions: + self._histories.popitem(last=False) + + def append( + self, + session_key: str, + role: str, + content: str, + sender_name: str = "", + ) -> None: + if session_key not in self._histories: + self._histories[session_key] = [] + history = self._histories[session_key] + history.append( + ChatMessage( + role=role, + content=content, + sender_name=sender_name, + ) + ) + if len(history) > self._max_messages: + history[:] = history[-self._max_messages :] + self._touch(session_key) + + def get_history(self, session_key: str) -> list[ChatMessage]: + return self._histories.get(session_key, []) + + def get_history_text(self, session_key: str, max_items: int = 20) -> str: + history = self.get_history(session_key) + recent = history[-max_items:] + if not recent: + return "" + lines = [] + for msg in recent: + name = msg.sender_name or msg.role + lines.append(f"[{name}]: {msg.content}") + return "\n".join(lines) + + def get_summary(self, session_key: str, max_items: int = 10) -> str: + """Build a summary for agent handoff.""" + history = self.get_history(session_key) + if not history: + return "No conversation history." + recent = history[-max_items:] + lines = ["=== Conversation Summary ==="] + for msg in recent: + name = msg.sender_name or msg.role + lines.append(f"[{name}]: {msg.content}") + return "\n".join(lines) + + def clear(self, session_key: str) -> None: + self._histories.pop(session_key, None) diff --git a/astrbot_plugin_smart_customer_service/managers/customer_profile_manager.py b/astrbot_plugin_smart_customer_service/managers/customer_profile_manager.py new file mode 100644 index 0000000000..d9442b22cc --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/customer_profile_manager.py @@ -0,0 +1,170 @@ +"""Customer profile management with SQLite persistence.""" + +from __future__ import annotations + +import json +import logging +import sqlite3 +import time +from dataclasses import dataclass, field +from pathlib import Path + +logger = logging.getLogger("astrbot") + + +@dataclass +class CustomerProfile: + customer_id: str + name: str = "" + tags: list[str] = field(default_factory=list) + notes: str = "" + lead_status: str = "none" + interaction_count: int = 0 + last_interaction_time: float = 0.0 + extra: dict = field(default_factory=dict) + + +class CustomerProfileManager: + """SQLite-backed customer profile store.""" + + def __init__(self, db_path: Path) -> None: + self._db_path = db_path + db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def _init_db(self) -> None: + with sqlite3.connect(str(self._db_path)) as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS customer_profiles ( + customer_id TEXT PRIMARY KEY, + name TEXT DEFAULT '', + tags TEXT DEFAULT '[]', + notes TEXT DEFAULT '', + lead_status TEXT DEFAULT 'none', + interaction_count INTEGER DEFAULT 0, + last_interaction_time REAL DEFAULT 0, + extra TEXT DEFAULT '{}' + ) + """ + ) + conn.commit() + + def get(self, customer_id: str) -> CustomerProfile | None: + with sqlite3.connect(str(self._db_path)) as conn: + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT * FROM customer_profiles WHERE customer_id = ?", + (customer_id,), + ).fetchone() + if row is None: + return None + return self._row_to_profile(row) + + def get_or_create(self, customer_id: str, name: str = "") -> CustomerProfile: + profile = self.get(customer_id) + if profile is not None: + return profile + profile = CustomerProfile(customer_id=customer_id, name=name) + self._upsert(profile) + return profile + + def update(self, profile: CustomerProfile) -> None: + profile.last_interaction_time = time.time() + self._upsert(profile) + + def record_interaction(self, customer_id: str, name: str = "") -> CustomerProfile: + profile = self.get_or_create(customer_id, name) + profile.interaction_count += 1 + profile.last_interaction_time = time.time() + if name and not profile.name: + profile.name = name + self.update(profile) + return profile + + def add_tag(self, customer_id: str, tag: str) -> CustomerProfile | None: + profile = self.get(customer_id) + if profile is None: + return None + if tag not in profile.tags: + profile.tags.append(tag) + self.update(profile) + return profile + + def set_lead_status(self, customer_id: str, status: str) -> CustomerProfile | None: + profile = self.get(customer_id) + if profile is None: + return None + profile.lead_status = status + self.update(profile) + return profile + + def update_notes(self, customer_id: str, notes: str) -> CustomerProfile | None: + profile = self.get(customer_id) + if profile is None: + return None + profile.notes = notes + self.update(profile) + return profile + + def search_by_tag(self, tag: str) -> list[CustomerProfile]: + with sqlite3.connect(str(self._db_path)) as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT * FROM customer_profiles WHERE tags LIKE ?", + (f'%"{tag}"%',), + ).fetchall() + return [self._row_to_profile(r) for r in rows] + + def _upsert(self, profile: CustomerProfile) -> None: + with sqlite3.connect(str(self._db_path)) as conn: + conn.execute( + """ + INSERT INTO customer_profiles + (customer_id, name, tags, notes, lead_status, + interaction_count, last_interaction_time, extra) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(customer_id) DO UPDATE SET + name=excluded.name, + tags=excluded.tags, + notes=excluded.notes, + lead_status=excluded.lead_status, + interaction_count=excluded.interaction_count, + last_interaction_time=excluded.last_interaction_time, + extra=excluded.extra + """, + ( + profile.customer_id, + profile.name, + json.dumps(profile.tags, ensure_ascii=False), + profile.notes, + profile.lead_status, + profile.interaction_count, + profile.last_interaction_time, + json.dumps(profile.extra, ensure_ascii=False), + ), + ) + conn.commit() + + @staticmethod + def _row_to_profile(row: sqlite3.Row) -> CustomerProfile: + tags_raw = row["tags"] + try: + tags = json.loads(tags_raw) if tags_raw else [] + except (json.JSONDecodeError, TypeError): + tags = [] + extra_raw = row["extra"] + try: + extra = json.loads(extra_raw) if extra_raw else {} + except (json.JSONDecodeError, TypeError): + extra = {} + return CustomerProfile( + customer_id=row["customer_id"], + name=row["name"] or "", + tags=tags, + notes=row["notes"] or "", + lead_status=row["lead_status"] or "none", + interaction_count=row["interaction_count"] or 0, + last_interaction_time=row["last_interaction_time"] or 0.0, + extra=extra, + ) diff --git a/astrbot_plugin_smart_customer_service/managers/escalation_manager.py b/astrbot_plugin_smart_customer_service/managers/escalation_manager.py new file mode 100644 index 0000000000..6e839ef875 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/escalation_manager.py @@ -0,0 +1,90 @@ +"""Escalation queue management for human agent handoff.""" + +from __future__ import annotations + +import logging +import time +from collections import OrderedDict +from dataclasses import dataclass, field + +from .session_manager import ServiceSession + +logger = logging.getLogger("astrbot") + + +@dataclass +class QueueEntry: + session: ServiceSession + reason: str = "" + enqueued_at: float = field(default_factory=time.time) + + +class EscalationManager: + """Manages the queue of customers waiting for a human agent.""" + + def __init__(self, queue_timeout: int = 0) -> None: + self._queue: OrderedDict[str, QueueEntry] = OrderedDict() + self._agents: set[str] = set() + self._agent_names: dict[str, str] = {} + self.queue_timeout = queue_timeout + + def set_agents( + self, agent_ids: list[str], agent_names: list[str] | None = None + ) -> None: + self._agents = set(agent_ids) + if agent_names: + for i, aid in enumerate(agent_ids): + if i < len(agent_names): + self._agent_names[aid] = agent_names[i] + + def get_agent_name(self, agent_id: str) -> str: + return self._agent_names.get(agent_id, agent_id) + + def is_agent(self, user_id: str) -> bool: + return user_id in self._agents + + def get_agents(self) -> list[str]: + return list(self._agents) + + def enqueue(self, session: ServiceSession, reason: str = "") -> int: + key = session.session_key + if key in self._queue: + return self.get_position(session) + self._queue[key] = QueueEntry(session=session, reason=reason) + return len(self._queue) + + def dequeue(self, session: ServiceSession) -> QueueEntry | None: + return self._queue.pop(session.session_key, None) + + def pop_next(self) -> QueueEntry | None: + if not self._queue: + return None + _, entry = self._queue.popitem(last=False) + return entry + + def get_position(self, session: ServiceSession) -> int: + key = session.session_key + for i, k in enumerate(self._queue): + if k == key: + return i + 1 + return 0 + + def get_queue_size(self) -> int: + return len(self._queue) + + def get_queue_list(self) -> list[QueueEntry]: + return list(self._queue.values()) + + def check_timeouts(self) -> list[QueueEntry]: + if self.queue_timeout <= 0: + return [] + now = time.time() + timed_out = [] + keys_to_remove = [] + for key, entry in self._queue.items(): + if now - entry.enqueued_at > self.queue_timeout: + timed_out.append(entry) + keys_to_remove.append(key) + for key in keys_to_remove: + self._queue.pop(key, None) + return timed_out diff --git a/astrbot_plugin_smart_customer_service/managers/product_manager.py b/astrbot_plugin_smart_customer_service/managers/product_manager.py new file mode 100644 index 0000000000..f3fd847f4a --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/product_manager.py @@ -0,0 +1,161 @@ +"""Product knowledge base management (JSON-backed).""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from pathlib import Path + +logger = logging.getLogger("astrbot") + + +@dataclass +class Product: + id: str + name: str + category: str = "" + description: str = "" + price: float = 0.0 + currency: str = "CNY" + stock: int = -1 + features: list[str] = field(default_factory=list) + faq: list[dict[str, str]] = field(default_factory=list) + extra: dict = field(default_factory=dict) + + +class ProductManager: + """Loads and queries product data from a JSON file.""" + + def __init__(self, data_path: Path) -> None: + self._data_path = data_path + self._products: dict[str, Product] = {} + self._load() + + def _load(self) -> None: + if not self._data_path.exists(): + self._create_sample() + try: + raw = json.loads(self._data_path.read_text(encoding="utf-8")) + products = raw if isinstance(raw, list) else raw.get("products", []) + for item in products: + p = Product( + id=str(item.get("id", "")), + name=str(item.get("name", "")), + category=str(item.get("category", "")), + description=str(item.get("description", "")), + price=float(item.get("price", 0)), + currency=str(item.get("currency", "CNY")), + stock=int(item.get("stock", -1)), + features=item.get("features", []), + faq=item.get("faq", []), + extra=item.get("extra", {}), + ) + self._products[p.id] = p + logger.info(f"Loaded {len(self._products)} products.") + except Exception as e: + logger.error(f"Failed to load products: {e}") + + def reload(self) -> None: + self._products.clear() + self._load() + + def get(self, product_id: str) -> Product | None: + return self._products.get(product_id) + + def search(self, keyword: str, category: str = "") -> list[Product]: + keyword_lower = keyword.lower() + results = [] + for p in self._products.values(): + if category and p.category.lower() != category.lower(): + continue + if ( + keyword_lower in p.name.lower() + or keyword_lower in p.description.lower() + or keyword_lower in p.category.lower() + or any(keyword_lower in f.lower() for f in p.features) + ): + results.append(p) + return results + + def list_categories(self) -> list[str]: + cats = {p.category for p in self._products.values() if p.category} + return sorted(cats) + + def list_all(self) -> list[Product]: + return list(self._products.values()) + + def format_product_info(self, product: Product) -> str: + lines = [ + f"Product: {product.name} (ID: {product.id})", + f"Category: {product.category}" if product.category else "", + f"Price: {product.price} {product.currency}", + f"Stock: {'In Stock' if product.stock > 0 else 'Out of Stock' if product.stock == 0 else 'N/A'}", + f"Description: {product.description}" if product.description else "", + ] + if product.features: + lines.append("Features: " + ", ".join(product.features)) + if product.faq: + lines.append("FAQ:") + for qa in product.faq: + lines.append(f" Q: {qa.get('q', '')}") + lines.append(f" A: {qa.get('a', '')}") + return "\n".join(line for line in lines if line) + + def _create_sample(self) -> None: + self._data_path.parent.mkdir(parents=True, exist_ok=True) + sample = { + "products": [ + { + "id": "PROD-001", + "name": "AstrBot Pro License", + "category": "Software", + "description": "Full-featured AstrBot license with priority support and all premium plugins.", + "price": 299.0, + "currency": "CNY", + "stock": 999, + "features": [ + "Unlimited platforms", + "Priority support", + "Premium plugins", + "Custom branding", + ], + "faq": [ + { + "q": "How long is the license valid?", + "a": "The license is valid for 1 year with free updates.", + }, + { + "q": "Can I transfer the license?", + "a": "Yes, licenses can be transferred once per year.", + }, + ], + }, + { + "id": "PROD-002", + "name": "AstrBot Enterprise", + "category": "Software", + "description": "Enterprise edition with SLA, dedicated support, and custom deployment.", + "price": 999.0, + "currency": "CNY", + "stock": 50, + "features": [ + "Everything in Pro", + "SLA guarantee", + "Dedicated support", + "Custom deployment", + "API access", + ], + "faq": [ + { + "q": "Is there a minimum contract?", + "a": "Enterprise licenses have a 1-year minimum contract.", + } + ], + }, + ] + } + self._data_path.write_text( + json.dumps(sample, ensure_ascii=False, indent=2), + encoding="utf-8", + ) diff --git a/astrbot_plugin_smart_customer_service/managers/session_manager.py b/astrbot_plugin_smart_customer_service/managers/session_manager.py new file mode 100644 index 0000000000..4a56a9b163 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/session_manager.py @@ -0,0 +1,169 @@ +"""Session state machine for customer service conversations. + +States: idle -> ai_serving -> waiting_human -> human_connected -> closed +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field +from enum import Enum + +logger = logging.getLogger("astrbot") + + +class SessionState(Enum): + IDLE = "idle" + AI_SERVING = "ai_serving" + WAITING_HUMAN = "waiting_human" + HUMAN_CONNECTED = "human_connected" + CLOSED = "closed" + + +_VALID_TRANSITIONS: dict[SessionState, set[SessionState]] = { + SessionState.IDLE: {SessionState.AI_SERVING}, + SessionState.AI_SERVING: { + SessionState.WAITING_HUMAN, + SessionState.CLOSED, + }, + SessionState.WAITING_HUMAN: { + SessionState.HUMAN_CONNECTED, + SessionState.AI_SERVING, + SessionState.CLOSED, + }, + SessionState.HUMAN_CONNECTED: { + SessionState.AI_SERVING, + SessionState.CLOSED, + }, + SessionState.CLOSED: {SessionState.AI_SERVING}, +} + + +@dataclass +class ServiceSession: + user_id: str + user_name: str + group_id: str + platform: str + state: SessionState = SessionState.IDLE + agent_id: str = "" + agent_name: str = "" + created_at: float = field(default_factory=time.time) + state_changed_at: float = field(default_factory=time.time) + + def transition(self, new_state: SessionState) -> bool: + allowed = _VALID_TRANSITIONS.get(self.state, set()) + if new_state not in allowed: + logger.warning( + f"Invalid session transition {self.state.value} -> {new_state.value} " + f"for user {self.user_id}" + ) + return False + self.state = new_state + self.state_changed_at = time.time() + return True + + @property + def is_active(self) -> bool: + return self.state not in (SessionState.IDLE, SessionState.CLOSED) + + @property + def session_key(self) -> str: + return f"{self.platform}:{self.group_id}:{self.user_id}" + + +class SessionManager: + """Manages all customer service sessions (in-memory).""" + + def __init__(self) -> None: + self._sessions: dict[str, ServiceSession] = {} + + def _key(self, platform: str, group_id: str, user_id: str) -> str: + return f"{platform}:{group_id}:{user_id}" + + def get_session( + self, platform: str, group_id: str, user_id: str + ) -> ServiceSession | None: + return self._sessions.get(self._key(platform, group_id, user_id)) + + def get_or_create( + self, + platform: str, + group_id: str, + user_id: str, + user_name: str = "", + ) -> ServiceSession: + key = self._key(platform, group_id, user_id) + session = self._sessions.get(key) + if session is None or session.state == SessionState.CLOSED: + session = ServiceSession( + user_id=user_id, + user_name=user_name or user_id, + group_id=group_id, + platform=platform, + ) + self._sessions[key] = session + return session + + def start_ai_serving(self, session: ServiceSession) -> bool: + if session.state == SessionState.IDLE: + return session.transition(SessionState.AI_SERVING) + if session.state == SessionState.CLOSED: + return session.transition(SessionState.AI_SERVING) + return session.state == SessionState.AI_SERVING + + def request_human(self, session: ServiceSession) -> bool: + return session.transition(SessionState.WAITING_HUMAN) + + def connect_agent( + self, session: ServiceSession, agent_id: str, agent_name: str = "" + ) -> bool: + ok = session.transition(SessionState.HUMAN_CONNECTED) + if ok: + session.agent_id = agent_id + session.agent_name = agent_name or agent_id + return ok + + def close_session(self, session: ServiceSession) -> bool: + if session.state == SessionState.CLOSED: + return True + ok = session.transition(SessionState.CLOSED) + if ok: + session.agent_id = "" + session.agent_name = "" + return ok + + def back_to_ai(self, session: ServiceSession) -> bool: + if session.state in (SessionState.WAITING_HUMAN, SessionState.HUMAN_CONNECTED): + ok = session.transition(SessionState.AI_SERVING) + if ok: + session.agent_id = "" + session.agent_name = "" + return ok + return False + + def remove_session(self, platform: str, group_id: str, user_id: str) -> None: + key = self._key(platform, group_id, user_id) + self._sessions.pop(key, None) + + def get_waiting_sessions(self) -> list[ServiceSession]: + return [ + s for s in self._sessions.values() if s.state == SessionState.WAITING_HUMAN + ] + + def get_sessions_by_agent(self, agent_id: str) -> list[ServiceSession]: + return [ + s + for s in self._sessions.values() + if s.agent_id == agent_id and s.state == SessionState.HUMAN_CONNECTED + ] + + def is_agent_busy(self, agent_id: str) -> bool: + return any( + s.agent_id == agent_id and s.state == SessionState.HUMAN_CONNECTED + for s in self._sessions.values() + ) + + def get_all_active(self) -> list[ServiceSession]: + return [s for s in self._sessions.values() if s.is_active] diff --git a/astrbot_plugin_smart_customer_service/managers/timeout_manager.py b/astrbot_plugin_smart_customer_service/managers/timeout_manager.py new file mode 100644 index 0000000000..f182e951af --- /dev/null +++ b/astrbot_plugin_smart_customer_service/managers/timeout_manager.py @@ -0,0 +1,64 @@ +"""Conversation timeout tracking.""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field + +logger = logging.getLogger("astrbot") + + +@dataclass +class TimerEntry: + session_key: str + start_time: float = field(default_factory=time.time) + warned: bool = False + + +class TimeoutManager: + """Tracks conversation timeouts and warnings.""" + + def __init__( + self, + conversation_timeout: int = 0, + warning_seconds: int = 120, + ) -> None: + self.conversation_timeout = conversation_timeout + self.warning_seconds = warning_seconds + self._timers: dict[str, TimerEntry] = {} + + def start(self, session_key: str) -> None: + self._timers[session_key] = TimerEntry(session_key=session_key) + + def stop(self, session_key: str) -> None: + self._timers.pop(session_key, None) + + def get_remaining(self, session_key: str) -> int | None: + if self.conversation_timeout <= 0: + return None + timer = self._timers.get(session_key) + if timer is None: + return None + elapsed = time.time() - timer.start_time + remaining = self.conversation_timeout - elapsed + return max(0, int(remaining)) + + def check_timeouts(self) -> tuple[list[str], list[str]]: + """Returns (timed_out_keys, needs_warning_keys).""" + if self.conversation_timeout <= 0: + return [], [] + now = time.time() + timed_out = [] + needs_warning = [] + for key, timer in list(self._timers.items()): + elapsed = now - timer.start_time + remaining = self.conversation_timeout - elapsed + if remaining <= 0: + timed_out.append(key) + elif remaining <= self.warning_seconds and not timer.warned: + needs_warning.append(key) + timer.warned = True + for key in timed_out: + self._timers.pop(key, None) + return timed_out, needs_warning diff --git a/astrbot_plugin_smart_customer_service/metadata.yaml b/astrbot_plugin_smart_customer_service/metadata.yaml new file mode 100644 index 0000000000..1f424e641d --- /dev/null +++ b/astrbot_plugin_smart_customer_service/metadata.yaml @@ -0,0 +1,6 @@ +name: astrbot_plugin_smart_customer_service +display_name: Smart Customer Service +desc: AI-powered intelligent customer service plugin with sales skill integration, customer profiling, and human agent escalation support. +author: AstrBot +version: 0.1.0 +repo: "" diff --git a/astrbot_plugin_smart_customer_service/skills/sales_skill_template/SKILL.md b/astrbot_plugin_smart_customer_service/skills/sales_skill_template/SKILL.md new file mode 100644 index 0000000000..456b17c2d5 --- /dev/null +++ b/astrbot_plugin_smart_customer_service/skills/sales_skill_template/SKILL.md @@ -0,0 +1,59 @@ +--- +name: smart-customer-service-sales +description: Sales consultant skill that guides AI to handle customer inquiries professionally, recommend products, qualify leads, and escalate to human agents when needed. +--- + +## Role + +You are a professional sales consultant providing customer service. Be friendly, helpful, and focused on understanding customer needs. + +## Response Strategy + +### 1. Greeting +- Warmly greet the customer +- Quickly identify their needs with open-ended questions +- Example: "Welcome! How can I help you today?" + +### 2. Need Discovery +- Ask open-ended questions to understand pain points +- Listen actively and confirm understanding +- Example: "What specific features are most important to you?" + +### 3. Product Recommendation +- Use the `query_product` tool to find matching products +- Present products based on customer needs, not just features +- Highlight how the product solves their specific problem +- NEVER fabricate product information — always use the tool + +### 4. Objection Handling +- Acknowledge the customer's concern +- Provide factual information to address it +- Offer alternatives if the original suggestion doesn't fit + +### 5. Lead Capture +- When a customer shows purchase intent, use `create_lead` to record it +- Use `update_customer_profile` to tag customer interests +- Guide them toward next steps (purchase, demo, consultation) + +## Escalation Rules + +Use `escalate_to_human` when: +- Customer explicitly asks for a human agent +- The issue involves complaints, refunds, or returns +- AI fails to understand the customer's intent after 2-3 attempts +- The inquiry involves custom pricing or large-volume orders +- The customer is upset or frustrated + +## Prohibited Actions + +- Do NOT invent product details; always query via tools +- Do NOT promise discounts or services beyond your authority +- Do NOT argue with customers +- Do NOT share internal information or other customer data +- Do NOT ignore a customer's request to speak with a human + +## Tone + +- Professional but approachable +- Concise — avoid overly long responses +- Use the customer's language (Chinese if they write in Chinese, English if in English) diff --git a/astrbot_plugin_smart_customer_service/tools/__init__.py b/astrbot_plugin_smart_customer_service/tools/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/test_smart_customer_service.py b/tests/test_smart_customer_service.py new file mode 100644 index 0000000000..edd070a0f6 --- /dev/null +++ b/tests/test_smart_customer_service.py @@ -0,0 +1,305 @@ +"""Tests for the Smart Customer Service plugin managers.""" + +import tempfile +import time +from pathlib import Path + +from astrbot_plugin_smart_customer_service.managers.conversation_context_manager import ( + ConversationContextManager, +) +from astrbot_plugin_smart_customer_service.managers.customer_profile_manager import ( + CustomerProfileManager, +) +from astrbot_plugin_smart_customer_service.managers.escalation_manager import ( + EscalationManager, +) +from astrbot_plugin_smart_customer_service.managers.product_manager import ( + ProductManager, +) +from astrbot_plugin_smart_customer_service.managers.session_manager import ( + ServiceSession, + SessionManager, + SessionState, +) +from astrbot_plugin_smart_customer_service.managers.timeout_manager import ( + TimeoutManager, +) + + +class TestSessionManager: + def test_create_and_get_session(self): + mgr = SessionManager() + session = mgr.get_or_create("tg", "grp1", "user1", "Alice") + assert session.user_id == "user1" + assert session.user_name == "Alice" + assert session.state == SessionState.IDLE + + def test_start_ai_serving(self): + mgr = SessionManager() + session = mgr.get_or_create("tg", "grp1", "user1", "Alice") + assert mgr.start_ai_serving(session) is True + assert session.state == SessionState.AI_SERVING + + def test_full_lifecycle(self): + mgr = SessionManager() + session = mgr.get_or_create("tg", "grp1", "user1", "Alice") + + assert mgr.start_ai_serving(session) is True + assert session.state == SessionState.AI_SERVING + + assert mgr.request_human(session) is True + assert session.state == SessionState.WAITING_HUMAN + + assert mgr.connect_agent(session, "agent1", "Bob") is True + assert session.state == SessionState.HUMAN_CONNECTED + assert session.agent_id == "agent1" + + assert mgr.close_session(session) is True + assert session.state == SessionState.CLOSED + + def test_back_to_ai(self): + mgr = SessionManager() + session = mgr.get_or_create("tg", "grp1", "user1", "Alice") + mgr.start_ai_serving(session) + mgr.request_human(session) + + assert mgr.back_to_ai(session) is True + assert session.state == SessionState.AI_SERVING + assert session.agent_id == "" + + def test_invalid_transition(self): + mgr = SessionManager() + session = mgr.get_or_create("tg", "grp1", "user1", "Alice") + assert session.transition(SessionState.HUMAN_CONNECTED) is False + + def test_get_waiting_sessions(self): + mgr = SessionManager() + s1 = mgr.get_or_create("tg", "grp1", "user1", "Alice") + s2 = mgr.get_or_create("tg", "grp1", "user2", "Bob") + mgr.start_ai_serving(s1) + mgr.request_human(s1) + mgr.start_ai_serving(s2) + + waiting = mgr.get_waiting_sessions() + assert len(waiting) == 1 + assert waiting[0].user_id == "user1" + + def test_is_agent_busy(self): + mgr = SessionManager() + session = mgr.get_or_create("tg", "grp1", "user1", "Alice") + mgr.start_ai_serving(session) + mgr.request_human(session) + mgr.connect_agent(session, "agent1", "Bob") + + assert mgr.is_agent_busy("agent1") is True + assert mgr.is_agent_busy("agent2") is False + + +class TestConversationContextManager: + def test_append_and_get(self): + mgr = ConversationContextManager() + mgr.append("key1", "user", "Hello", "Alice") + mgr.append("key1", "assistant", "Hi there!", "Bot") + history = mgr.get_history("key1") + assert len(history) == 2 + assert history[0].content == "Hello" + + def test_max_messages(self): + mgr = ConversationContextManager(max_messages=3) + for i in range(5): + mgr.append("key1", "user", f"msg {i}") + history = mgr.get_history("key1") + assert len(history) == 3 + assert history[0].content == "msg 2" + + def test_lru_eviction(self): + mgr = ConversationContextManager(max_sessions=2) + mgr.append("key1", "user", "a") + mgr.append("key2", "user", "b") + mgr.append("key3", "user", "c") + assert mgr.get_history("key1") == [] + assert len(mgr.get_history("key2")) == 1 + + def test_get_history_text(self): + mgr = ConversationContextManager() + mgr.append("key1", "user", "Hello", "Alice") + mgr.append("key1", "assistant", "Hi!", "Bot") + text = mgr.get_history_text("key1") + assert "[Alice]: Hello" in text + assert "[Bot]: Hi!" in text + + def test_clear(self): + mgr = ConversationContextManager() + mgr.append("key1", "user", "Hello") + mgr.clear("key1") + assert mgr.get_history("key1") == [] + + +class TestCustomerProfileManager: + def test_create_and_get(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = CustomerProfileManager(Path(tmp) / "profiles.db") + profile = mgr.get_or_create("user1", "Alice") + assert profile.customer_id == "user1" + assert profile.name == "Alice" + assert profile.lead_status == "none" + + def test_add_tag(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = CustomerProfileManager(Path(tmp) / "profiles.db") + mgr.get_or_create("user1", "Alice") + mgr.add_tag("user1", "high_intent") + profile = mgr.get("user1") + assert "high_intent" in profile.tags + + def test_set_lead_status(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = CustomerProfileManager(Path(tmp) / "profiles.db") + mgr.get_or_create("user1", "Alice") + mgr.set_lead_status("user1", "interested") + profile = mgr.get("user1") + assert profile.lead_status == "interested" + + def test_record_interaction(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = CustomerProfileManager(Path(tmp) / "profiles.db") + mgr.record_interaction("user1", "Alice") + mgr.record_interaction("user1", "Alice") + profile = mgr.get("user1") + assert profile.interaction_count == 2 + + def test_search_by_tag(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = CustomerProfileManager(Path(tmp) / "profiles.db") + mgr.get_or_create("user1", "Alice") + mgr.add_tag("user1", "vip") + mgr.get_or_create("user2", "Bob") + results = mgr.search_by_tag("vip") + assert len(results) == 1 + assert results[0].customer_id == "user1" + + +class TestProductManager: + def test_load_sample_products(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = ProductManager(Path(tmp) / "products.json") + products = mgr.list_all() + assert len(products) >= 2 + + def test_search(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = ProductManager(Path(tmp) / "products.json") + results = mgr.search("Pro") + assert len(results) >= 1 + assert any("Pro" in p.name for p in results) + + def test_search_by_category(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = ProductManager(Path(tmp) / "products.json") + results = mgr.search("", "Software") + assert len(results) >= 1 + + def test_get_by_id(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = ProductManager(Path(tmp) / "products.json") + p = mgr.get("PROD-001") + assert p is not None + assert p.name == "AstrBot Pro License" + + def test_format_product_info(self): + with tempfile.TemporaryDirectory() as tmp: + mgr = ProductManager(Path(tmp) / "products.json") + p = mgr.get("PROD-001") + info = mgr.format_product_info(p) + assert "AstrBot Pro License" in info + assert "299" in info + + +class TestEscalationManager: + def _make_session(self, user_id="user1"): + return ServiceSession( + user_id=user_id, + user_name=user_id, + group_id="grp1", + platform="tg", + state=SessionState.WAITING_HUMAN, + ) + + def test_enqueue_and_dequeue(self): + mgr = EscalationManager() + s = self._make_session() + pos = mgr.enqueue(s, "test") + assert pos == 1 + assert mgr.get_queue_size() == 1 + + entry = mgr.dequeue(s) + assert entry is not None + assert mgr.get_queue_size() == 0 + + def test_pop_next(self): + mgr = EscalationManager() + s1 = self._make_session("user1") + s2 = self._make_session("user2") + mgr.enqueue(s1) + mgr.enqueue(s2) + + entry = mgr.pop_next() + assert entry.session.user_id == "user1" + assert mgr.get_queue_size() == 1 + + def test_agents(self): + mgr = EscalationManager() + mgr.set_agents(["a1", "a2"], ["Agent 1", "Agent 2"]) + assert mgr.is_agent("a1") is True + assert mgr.is_agent("random") is False + assert mgr.get_agent_name("a1") == "Agent 1" + + def test_queue_position(self): + mgr = EscalationManager() + s1 = self._make_session("user1") + s2 = self._make_session("user2") + mgr.enqueue(s1) + mgr.enqueue(s2) + assert mgr.get_position(s1) == 1 + assert mgr.get_position(s2) == 2 + + def test_check_timeouts(self): + mgr = EscalationManager(queue_timeout=1) + s = self._make_session() + mgr.enqueue(s) + s_entry = mgr._queue[s.session_key] + s_entry.enqueued_at = time.time() - 2 + timed_out = mgr.check_timeouts() + assert len(timed_out) == 1 + assert mgr.get_queue_size() == 0 + + +class TestTimeoutManager: + def test_start_and_remaining(self): + mgr = TimeoutManager(conversation_timeout=60) + mgr.start("key1") + remaining = mgr.get_remaining("key1") + assert remaining is not None + assert 58 <= remaining <= 60 + + def test_check_timeouts(self): + mgr = TimeoutManager(conversation_timeout=1, warning_seconds=0) + mgr.start("key1") + mgr._timers["key1"].start_time = time.time() - 2 + timed_out, warnings = mgr.check_timeouts() + assert "key1" in timed_out + + def test_warning(self): + mgr = TimeoutManager(conversation_timeout=10, warning_seconds=5) + mgr.start("key1") + mgr._timers["key1"].start_time = time.time() - 7 + _, warnings = mgr.check_timeouts() + assert "key1" in warnings + + def test_disabled(self): + mgr = TimeoutManager(conversation_timeout=0) + mgr.start("key1") + assert mgr.get_remaining("key1") is None + timed_out, warnings = mgr.check_timeouts() + assert timed_out == [] + assert warnings == []