cx-cloud-ai 0.1.1

Distributed AI orchestration toolkit for customer service platforms

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

pip install cx-cloud-ai

For local development:

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

Quick Start

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

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:

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

Run the script demos from the repository root:

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

Then submit a request:

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

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.