Metadata-Version: 2.4
Name: cx-cloud-ai
Version: 0.1.1
Summary: Distributed AI orchestration toolkit for customer service platforms
Author: Bharath Janumpally
License-Expression: MIT
Project-URL: Homepage, https://github.com/bharathjanumpally/cx-cloud-ai
Project-URL: Repository, https://github.com/bharathjanumpally/cx-cloud-ai
Keywords: customer-service,contact-center,ai,routing,distributed-systems,audit,sla
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Provides-Extra: demo
Requires-Dist: fastapi>=0.110; extra == "demo"
Requires-Dist: uvicorn>=0.27; extra == "demo"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"

# cx-cloud-ai

**Distributed AI orchestration toolkit for customer service platforms.**

`cx-cloud-ai` is an open-source Python toolkit for building distributed, AI-powered customer service systems with intelligent routing, agent state management, SLA monitoring, and audit-ready AI workflows.

It is not a chatbot framework, a helpdesk product, or a thin LLM wrapper. It provides reusable orchestration primitives for contact-center workloads across voice, chat, email, and ticketing channels.

## Why It Exists

Modern customer service systems need to route requests from many channels, give agents real-time context, summarize conversations, preserve audit trails, and stay responsive during traffic spikes. `cx-cloud-ai` packages those concerns into small Python building blocks that can be used independently or composed into a larger distributed cloud architecture.

## Core Features

- **Smart routing**: Route by channel, intent, priority, customer tier, and sentiment.
- **Agent state management**: Track availability, skills, active channels, capacity, and load.
- **AI summarization interface**: Use a deterministic local fallback or plug in your own provider.
- **Distributed workload orchestration**: Assign AI and service tasks to healthy workers.
- **SLA monitoring**: Track p50, p90, and p99 latency and detect breaches.
- **Audit logging**: Hash customer IDs, redact common PII, and export decision traces.

## Install

```bash
pip install cx-cloud-ai
```

For local development:

```bash
python3 -m pip install -e ".[dev,demo]"
```

## Quick Start

```python
from cx_cloud_ai.routing import SmartRouter

router = SmartRouter()

router.add_rule(
    channel="voice",
    intent="billing_issue",
    priority="high",
    route_to="billing_specialist_queue",
)

decision = router.route({
    "channel": "voice",
    "customer_tier": "premium",
    "intent": "billing_issue",
    "sentiment": "negative",
    "priority": "high",
})

print(decision.queue)
```

## End-to-End Flow

```python
from cx_cloud_ai.agent_state import AgentStateManager
from cx_cloud_ai.ai import ConversationSummarizer, IntentClassifier
from cx_cloud_ai.audit import AuditLogger
from cx_cloud_ai.models import CustomerRequest
from cx_cloud_ai.routing import SmartRouter
from cx_cloud_ai.sla import SLAMonitor

request = CustomerRequest(
    channel="chat",
    customer_id="customer-123",
    message="I was charged twice and need this fixed urgently.",
    customer_tier="premium",
    priority="high",
)

classifier = IntentClassifier()
intent = classifier.classify(request.message)
payload = {**request.to_routing_payload(), "intent": intent, "sentiment": "negative"}

router = SmartRouter()
router.add_rule(
    channel="chat",
    intent="billing_issue",
    customer_tier="premium",
    route_to="premium_billing_queue",
)

agents = AgentStateManager()
agents.update_agent(
    "agent-123",
    state="AVAILABLE",
    skills=["billing_issue"],
    active_channels=["chat"],
)

decision = router.route(payload)
available_agents = agents.find_available_agents(skill=intent, channel="chat")

summary = ConversationSummarizer().summarize([
    f"Customer: {request.message}",
    "Agent: I can help check the duplicate charge.",
])

sla = SLAMonitor()
sla.define_sla(metric="routing_latency_ms", threshold=300, percentile="p99")
breach = sla.check("routing_latency_ms", 148)

audit = AuditLogger()
audit.log_event(
    event_type="ROUTE_DECISION_CREATED",
    customer_id=request.customer_id,
    agent_id=available_agents[0].agent_id if available_agents else None,
    decision=decision.queue,
    metadata={"summary": summary.short_summary, "sla_breached": breach.breached},
)
```

## Demo

Run the FastAPI demo:

```bash
python3 -m pip install -e ".[demo]"
uvicorn examples.fastapi_demo.main:app --reload
```

Run the script demos from the repository root:

```bash
python3 -m examples.contact_center_demo.demo
python3 -m examples.chat_support_demo.demo
```

Then submit a request:

```bash
curl -X POST http://127.0.0.1:8000/orchestrate \
  -H "Content-Type: application/json" \
  -d '{"channel":"chat","customer_id":"customer-123","message":"I was charged twice and this is urgent","customer_tier":"premium","priority":"high"}'
```

## Architecture

```mermaid
flowchart TD
    A["Customer Channel"] --> B["Request Ingestion"]
    B --> C["Intent Detection / AI Classifier"]
    C --> D["Smart Router"]
    D --> E["Agent State Manager"]
    E --> F["Agent or AI Assistant"]
    F --> G["Conversation Summary"]
    G --> H["Audit Log + SLA Metrics"]
```

## Roadmap

- Provider adapters for OpenAI and Bedrock summarization.
- Redis-backed agent state and task queues.
- Prometheus-compatible SLA metrics export.
- Contact-center event adapters.
- Benchmark suite for routing and orchestration latency.
