# **Kɑlimerɑ.ɑi [Backend]**
- Created at : 2025-03-31
- Created by : Manoj Vala
## **Introduction**
This document outlines the process of:
1. Creating an **AI Agent** and storing its configuration in the database.
2. Implementing **API endpoints for conversation statistics**, dynamically calculated from real conversations with agents.
>[!IMPORTANT]
>```python
>api.add_resource(kalimera, 'kalimera/')
>```
>All the methods and endpoints are called with prefix `kalimera`.
## **Part 1: AI Agent API**
### **Step 1: Define the AI Agent Model**
**Description:** To store the AI Agent configuration, we will create a database model where predefined fields will be stored separately, and the rest of the configuration will be stored in a `JSONB` column.
---
### **AI Agent Database Schema**
| **Field Name** | **Type** | **Attributes** | **Description** |
|-------------|-----------|-------------|--------------|
| `id` | `UUID` | `primary_key=True` | Unique identifier for the AI Agent. |
| `tenant_id` | `UUID` | `nullable=False` | For mapping of user. |
| `name` | `String(255)` | `nullable=False` | Name of the AI Agent (required). |
| `profile_type` | `String(50)` | `nullable=False` | Choice ['inbount', 'outbound'] |
| `configuration` | `JSONB` | `nullable=True` | Includes all the section |
| created_at | DateTime | server_default=db.func.now() | Timestamp of when the agent was created (auto-set).|
| updated_at | DateTime | onupdate=db.func.now() | Timestamp of the last update (auto-updated).|
---
### **Key Notes**
- **`JSONB`**: PostgreSQL’s binary JSON format for flexible, schema-less storage.
- **Timestamps**: `created_at` and `updated_at` are automated.
**Code Snippet:**
```python
class KalimeraAIAgent(db.Model):
id = db.Column(db.UUID, primary_key=True)
tenant_id = db.Column(db.UUID, nullable=False)
name = db.Column(db.String(255), nullable=False)
profile_type = db.Column(db.String(255), nullable=True)
configuration = db.Column(JSONB, nullable=True)
created_at = db.Column(db.DateTime, server_default=db.func.now())
updated_at = db.Column(db.DateTime, onupdate=db.func.now())
```
---
### **Step 2: Create the AI Agent API Endpoint**
**Description:** We will create a RESTful API using Flask-RESTful with three methods. All the database transactions are made in a seperate **Service** file. AI Agent are stored in database as well as in redis cache.:
- `GET`: Retrieve an AI Agent by ID. Some data will be **hidden**. it cannot be readable for anyone.
- `POST`: Create a new AI Agent.
- `PUT`: Update an existing AI Agent.
- `DELETE`: Delete an existing AI Agent.
**Code Snippet:**
```python
class ConfigurationSection(Enum):
AI_HUB = "AIHub"
AUDIO_SERVER = "AudioServer"
AZURE = "Azure"
ELEVEN_LABS_TTS = "ElevenLabsTTS"
CALL_PROCESSING = "CallProcessing"
REDIS = "Redis"
KNOWLEDGE_BASE = "KnowledgeBase"
BACKGROUND_AUDIO = "BackgroundAudio"
WEB_TALK = "WebTalk"
EXPERIMENTAL_FEATURES = "ExperimentalFeatures"
SERILOG = "Serilog"
LOGGING = "Logging"
ALLOWED_HOSTS = "AllowedHosts"
class KalimeraAgent(Resource):
def get(self, app_model, end_user):
"""
Get the agent profile by agent ID.\n
:param agent_id: The ID of the agent.\n
:return: The agent profile.
"""
if agent_id:
agent = db.session.query(KalimeraAIAgent).filter(
KalimeraAIAgent.tenant_id == tenant_id,
KalimeraAIAgent.id == agent_id).first()
if agent:
return agent
elif tenant_id and isinstance(args, dict):
search_value = args.get("search", "")
if search_value:
search_value = search_value.strip()
agents = db.session.query(KalimeraAIAgent).filter(
KalimeraAIAgent.tenant_id == tenant_id,
or_(
KalimeraAIAgent.name.ilike(f"%{search_value}%"),
KalimeraAIAgent.description.ilike(f"%{search_value}%"),
KalimeraAIAgent.profile_type.ilike(f"%{search_value}%")
)
).order_by(
KalimeraAIAgent.created_at.desc()
).limit(args.get("limit", 10)).offset((args.get("page", 1) - 1) * args.get("limit", 10)).all()
else:
agents = db.session.query(KalimeraAIAgent).filter(
KalimeraAIAgent.tenant_id == tenant_id
).order_by(
KalimeraAIAgent.created_at.desc()
).limit(args.get("limit", 10)).offset((args.get("page", 1) - 1) * args.get("limit", 10)).all()
if agents:
return agents
return False
def post(self, app_model, end_user):
"""Create a new AI Agent with the provided data."""
if "avatar" in args and not cls.is_valid_image_url(args["avatar"]):
return False, "Invalid image URL.", 400, None
if args["type"] not in ["inbound", "outbound"]:
return False, "Invalid profile type.", 400, None
agent = KalimeraAIAgent(
name = args["name"],
description = args["description"],
avatar = args["avatar"] if "avatar" in args else None,
profile_type = args["type"],
tenant_id = tenant_id
)
db.session.add(agent)
db.session.commit()
db.session.refresh(agent)
agent_id = agent.id
db.session.close()
return True, "Agent created successfully", 201, str(agent_id)
def put(self, app_model, end_user):
try:
ConfigurationSection(section)
agent = KalimeraAIAgent.query.get(agent_id)
if not agent:
return jsonify({'error': 'AI Agent not found'}), 404
try:
new_data = request.json
updated_config = update_configuration_section(agent, section, new_data)
return jsonify({'message': 'Configuration updated successfully', 'configuration': updated_config}), 200
except Exception as e:
return jsonify({'error': str(e)}), 400
except Exception as e:
return jsonify({'error': 'Invalid section name'}), 404
def delete(self, app_model, end_user):
agent = KalimeraAIAgent.query.get_or_404(agent_id)
db.session.delete(agent)
db.session.commit()
return {'message': 'Contact deleted successfully'}
api.add_resource(KalimeraAgent, 'kalimera/ai-agent', 'kalimera/ai-agent/<int:agent_id>', 'kalimera/ai-agent/<uuid:agent_id>/<section>')
```
## Update Agent configuration
```python
class UpdateKalimeraAgent(Resource):
def update_configuration_section(agent, section_name, new_data):
if not agent.configuration:
agent.configuration = {}
agent.configuration[section_name] = new_data
db.session.commit()
return agent.configuration
def put(agent_id, section):
if isinstance(configuration, dict):
agent = db.session.query(KalimeraAIAgent).filter(
KalimeraAIAgent.tenant_id == agent_id,
KalimeraAIAgent.id == agent_id).first()
if not agent:
db.session.close()
return False, "Agent profile not found.", 404
# Update the configuration section
if not agent.configuration:
agent.configuration = configuration
else:
agent.configuration.update(configuration)
db.session.commit()
db.session.refresh(agent)
db.session.close()
return True, "Configuration section updated successfully.", 200
api.add_resource(UpdateKalimeraAgent, 'kalimera/ai-agent/<uuid:agent_id>')
```
---
### **Step 3: Handle API Endpoint in service file**
**Description:** Every endpoints will call service file `KalimeraService`. Which includes whole logic of controller.
**Service** file. AI Agent are stored in database as well as in redis cache.:
## 📌 **1. Retrieve AI Agent**
>
> **Endpoint**
> ```
> /ai-agent/{agent_id}
> ```
>
> **Description**:
> Retrieve an AI Agent by ID. Some data will be **hidden**. it cannot be readable for anyone.
>
> ### **Method**
> `GET`
>
> ### 🔑 **Headers**
> | Header | Type | Required | Description |
> |----------------|--------|----------|-------------------------------|
> | `Content-Type` | String | ✅ | `application/json` |
> | `Authorization`| String | ✅ | `api-token-generated` |
>
> **cURL Example**
> ```bash
> curl --location 'https://api.my-buddy.ai/console/api/ai-agent/<uuid>' \
> --header 'Content-Type: application/json'
> ```
> **Response**
> ```json
> { "id": 1, "name": "AI Assistant", "description": "Virtual AI Agent", "ai_hub": {}, "audio_server": {}, "azure": {}, "eleven_labs_tts": {}, "call_processing": {}, "redis": {}, "knowledge_base": {}, "background_audio": {}, "web_talk": {}, "experimental_features": {}, "serilog": {}, "logging": {}, "allowed_hosts": {}}
> ```
>
> Service file:
> ```python
> class AIAgentService:
> @staticmethod
> def get_ai_agent(agent_id):
> """Fetch AI Agent from Redis or Database."""
> cache_key = f"kalimera:{agent_id}"
> cached_agent = redis_client.get(cache_key)
>
> if cached_agent:
> return json.loads(cached_agent) # Return cached data if available
>
> agent = KalimeraAIAgent.query.get(agent_id)
> if not agent:
> return None # Return None if agent not found
>
> agent_data = {
> "id": agent.id,
> "name": agent.name,
> "description": agent.description,
> "ai_hub": agent.ai_hub,
> "audio_server": agent.audio_server,
> "azure": agent.azure,
> "eleven_labs_tts": agent.eleven_labs_tts,
> "call_processing": agent.call_processing,
> "redis": agent.redis,
> "knowledge_base": agent.knowledge_base,
> "background_audio": agent.background_audio,
> "web_talk": agent.web_talk,
> "experimental_features": agent.experimental_features,
> "serilog": agent.serilog,
> "logging": agent.logging,
> "allowed_hosts": agent.allowed_hosts
> }
>
> redis_client.setex(cache_key, 3600, json.dumps(agent_data)) # Store in Redis for 1 hour
> return agent_data
>```
---
### 📌 **2. Create AI Agent**
>
> **Endpoint**
> ```
> /ai-agent
> ```
>
> **Description**:
> Creates a new AI Agent and stores it in the database. The agent data is also cached in Redis.
>
> ### **Method**
> `POST`
>
> ### 🔑 **Headers**
> | Header | Type | Required | Description |
> |---------------|--------|----------|-------------------------------|
> | `Content-Type` | String | ✅ | `application/json` |
> | `Authorization` | String | ✅ | `api-token-generated` |
>
> ### **Request Body**
> ```json
> { "name": "AI Assistant", "description": "Virtual AI Agent", "ai_hub": {}, "audio_server": {}, "azure": {}, "eleven_labs_tts": {}, "call_processing": {}, "redis": {}, "knowledge_base": {}, "background_audio": {}, "web_talk": {}, "experimental_features": {}, "serilog": {}, "logging": {}, "allowed_hosts": {}}
> ```
>
> ### **cURL Example**
> ```bash
> curl --location 'https://api.my-buddy.ai/console/api/ai-agent' \
> --header 'Content-Type: application/json' \
> --data '{ "name": "AI Assistant", "description": "Virtual AI Agent", "ai_hub": {}, "audio_server": {}, "azure": {}, "eleven_labs_tts": {}, "call_processing": {}, "redis": {}, "knowledge_base": {}, "background_audio": {}, "web_talk": {}, "experimental_features": {}, "serilog": {}, "logging": {}, "allowed_hosts": {}}'
> ```
>
> ### **Response**
> ```json
> {
> "id": 1,
> "message": "AI Agent created successfully"
> }
> ```
>
---
>
> ### **Service File**
> ```python
> class AIAgentService:
> @staticmethod
> def create_ai_agent(agent_data):
> """Create a new AI Agent, store it in DB, and cache it in Redis."""
> new_agent = KalimeraAIAgent(**agent_data)
> db.session.add(new_agent)
> db.session.commit()
>
> agent_data["id"] = new_agent.id
> cache_key = f"kalimera:{new_agent.id}"
> redis_client.setex(cache_key, 3600, json.dumps(agent_data)) # Cache in Redis for 1 hour
>
> return {"id": new_agent.id, "name": new_agent.name, "description": new_agent.description, "message": "AI Agent created successfully"}
> ```
>
---
## 📌 **3. Update AI Agent**
>
> **Endpoint**
> ```
> /ai-agent/{agent_id}
> ```
>
> **Description**:
> Updates an existing AI Agent's details. The updated data is stored in the database and refreshed in Redis.
>
> ### **Method**
> `PUT`
>
> ### 🔑 **Headers**
> | Header | Type | Required | Description |
> |---------------|--------|----------|-------------------------------|
> | `Content-Type` | String | ✅ | `application/json` |
> | `Authorization` | String | ✅ | `api-token-generated` |
>
> ### **Request Body**
> ```json
> { "name": "Updated AI Assistant", "description": "Updated Virtual AI Agent", "ai_hub": {}, "audio_server": {}, "azure": {}, "eleven_labs_tts": {}, "call_processing": {}, "redis": {}, "knowledge_base": {}, "background_audio": {}, "web_talk": {}, "experimental_features": {}, "serilog": {}, "logging": {}, "allowed_hosts": {}}
> ```
>
> ### **cURL Example**
> ```bash
> curl --location --request PUT 'https://api.my-buddy.ai/console/api/ai-agent/<uuid>' \
> --header 'Content-Type: application/json' \
> --data '{ "name": "Updated AI Assistant", "description": "Updated Virtual AI Agent", "ai_hub": {}, "audio_server": {}, "azure": {}, "eleven_labs_tts": {}, "call_processing": {}, "redis": {}, "knowledge_base": {}, "background_audio": {}, "web_talk": {}, "experimental_features": {}, "serilog": {}, "logging": {}, "allowed_hosts": {}}'
> ```
>
> ### **Response**
> ```json
> {
> "id": 1,
> "message": "AI Agent updated successfully"
> }
> ```
>
> ---
>
> ### **Service File**
> ```python
> class AIAgentService:
> @staticmethod
> def update_ai_agent(agent_id, agent_data):
> """Update an existing AI Agent, store it in DB, and refresh cache."""
> agent = KalimeraAIAgent.query.get(agent_id)
> if not agent:
> return None # Return None if agent does not exist
>
> for key, value in agent_data.items():
> setattr(agent, key, value) # Update agent attributes
>
> agent_data["id"] = agent_id
> cache_key = f"kalimera:{agent_id}"
> redis_client.setex(cache_key, 3600, json.dumps(agent_data)) # Refresh cache
>
> return {"id": agent_id, "message": "AI Agent updated successfully"}
> ```
---
## **Part 2: Voice APIs**
## 1. Contact Management:
| Field | Type | Nullable | Description |
|--------|------|----------|---------------|
| id | Integer | False| Primary key |
| first_name | String(50)| False| Contact's first name (required)|
| last_name | String(50) | False| Contact's last name (required)|
| contact_no | String(20) | True| Contact phone number |
| email | String(100) | True| Contact email address|
| start_datetime | String(50) | True| Start date/time of interaction|
| end_datetime | String(50) | True| End date/time of interaction |
| language| String(30) | True| Language preference (e.g., English) |
| voice | String(10) | True| Voice preference ('Male'/'Female') |
| status | String(20) | True| Interaction status ('Inbound'/'Outbound') |
```python
class KalimeraContact(db.Model):
id = db.Column(db.Integer, primary_key=True)
first_name = db.Column(db.String(50), nullable=False)
last_name = db.Column(db.String(50), nullable=False)
contact_no = db.Column(db.String(20))
email = db.Column(db.String(100))
start_datetime = db.Column(db.String(50))
end_datetime = db.Column(db.String(50))
language = db.Column(db.String(30))
voice = db.Column(db.String(10)) # 'Male' or 'Female'
status = db.Column(db.String(20)) # 'Inbound' or 'Outbound'
created_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))
updated_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))
def to_dict(self):
return { 'id': self.id, 'first_name': self.first_name, 'last_name': self.last_name, 'contact_no': self.contact_no, 'email': self.email, 'start_datetime': self.start_datetime, 'end_datetime': self.end_datetime, 'language': self.language, 'voice': self.voice, 'status': self.status}
```
---
### To manage above contact table we need controller which is as below:
```python
# RESTful Resource
class KalimeraContact(Resource):
def get(self, contact_id=None):
if contact_id:
contact = Contact.query.get_or_404(contact_id)
return contact.to_dict()
else:
# Pagination parameters
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 10, type=int)
contacts = Contact.query.paginate(page=page, per_page=per_page)
return {
'contacts': [contact.to_dict() for contact in contacts.items],
'total': contacts.total,
'pages': contacts.pages,
'current_page': contacts.page
}
def post(self):
data = request.get_json()
# Validate required fields
if not data.get('first_name') or not data.get('last_name'):
return {'message': 'First name and last name are required'}, 400
contact = KalimeraContact(
first_name=data['first_name'],
last_name=data['last_name'],
contact_no=data.get('contact_no'),
email=data.get('email'),
start_datetime=data.get('start_datetime', '24 Aug, 05:30 pm'), # Default from image
end_datetime=data.get('end_datetime', '24 Aug, 05:30 pm'), # Default from image
language=data.get('language', 'English'), # Default from image
voice=data.get('voice', 'Male'), # Default from image
status=data.get('status', 'Inbound') # Default from image
)
db.session.add(contact)
db.session.commit()
return contact.to_dict(), 201
def put(self, contact_id):
contact = KalimeraContact.query.get_or_404(contact_id)
data = request.get_json()
# Update fields if they exist in the request
if 'first_name' in data:
contact.first_name = data['first_name']
if 'last_name' in data:
contact.last_name = data['last_name']
if 'contact_no' in data:
contact.contact_no = data['contact_no']
if 'email' in data:
contact.email = data['email']
if 'start_datetime' in data:
contact.start_datetime = data['start_datetime']
if 'end_datetime' in data:
contact.end_datetime = data['end_datetime']
if 'language' in data:
contact.language = data['language']
if 'voice' in data:
contact.voice = data['voice']
if 'status' in data:
contact.status = data['status']
db.session.commit()
return contact.to_dict()
def delete(self, contact_id):
contact = Contact.query.get_or_404(contact_id)
db.session.delete(contact)
db.session.commit()
return {'message': 'Contact deleted successfully'}
api.add_resource(KalimeraContact, '/kalimera/contact', '/kalimera/contact/<int:contact_id>')
```
### Note:
> Above endpoint will manage whole contacts by create, update, read and delete methods.
**For storing conversation records need to create table `KalimeraConversation`**
### Schema:
Here is the updated **table view** for the `KalimeraConversation` model based on the latest fields:
| **Column Name** | **Data Type**| **Constraints** | **Description** |
|-----------------|--------------|---------------------|----------------|
| `conversation_guid` | UUID | Primary Key, Default: `uuid_generate_v4()` | Unique identifier for the conversation. |
| `agent_id`| UUID | Not Null | Identifier for the agent handling the conversation. |
| `tenant_id` | UUID | Not Null | Identifier for the tenant. |
| `title` | VARCHAR(255) | Not Null | Title of the conversation. |
| `status` | VARCHAR(255) | Not Null | Status of the conversation (Active, Completed, Failed, etc.). |
| `profile_type` | ENUM(`inbound`, `outbound`) | Not Null | Defines whether the conversation is **inbound** or **outbound**. |
| `instance_id` | UUID | Nullable | Identifier of the user who initiated the conversation. |
| `from_number` | VARCHAR(255) | Nullable | When agent receive the call |
| `to_number` | VARCHAR(255) | Nullable | When agent calls |
| `created_at` | TIMESTAMP | Not Null, Default: `CURRENT_TIMESTAMP(0)` | Creation timestamp. |
| `updated_at` | TIMESTAMP | Not Null, Default: `CURRENT_TIMESTAMP(0)` | Last updated timestamp. |
| `input_file_url` | TEXT| Nullable | URL of the conversation's associated file in Azure Blob Storage. |
| `output_file_url` | TEXT| Nullable | URL of the conversation's associated file in Azure Blob Storage. |
| `conversation_url` | TEXT| Nullable | URL of the conversation's associated file in Azure Blob Storage. |
| `language` | VARCHAR(50) | Nullable | Language used in the conversation. |
| `duration` | INTEGER | Nullable | Duration of the conversation (in seconds). |
```python
class KalimeraConversation(db.Model):
__tablename__ = "kalimera_conversations"
__table_args__ = (
db.PrimaryKeyConstraint("id", name="kalimera_conversation_pkey"),
db.Index("kalimera_conversation_idx", "agent_id", "from_source", "end_user_id", "title"),
)
id = db.Column(StringUUID, server_default=db.text("uuid_generate_v4()"))
agent_id = db.Column(StringUUID, nullable=False)
tenant_id = db.Column(StringUUID, nullable=False)
title = db.Column(db.String(255), nullable=False)
status = db.Column(db.String(255), nullable=False) # Active, Completed, Failed, etc.
profile_type = db.Column(db.Enum("inbound", "outbound", name="profile_type_enum"), nullable=False)
instance_id = db.Column(StringUUID, nullable=True) # User who initiated
from_number = db.Column(db.String(255), nullable=False)
to_number = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))
updated_at = db.Column(db.DateTime, nullable=False, server_default=db.text("CURRENT_TIMESTAMP(0)"))
# Azure Blob Storage URLs
input_file_url = db.Column(db.Text, nullable=True)
output_file_url = db.Column(db.Text, nullable=True)
# Additional metadata
language = db.Column(db.String(50), nullable=True)
duration = db.Column(db.Integer, nullable=True)
```
### Note:
> - ~~Files are stored in Azure-blob and it's URl will be in every message which is stored in invidual conversation.~~
> - ~~All conversations are stored in azure-blob and URL of that will be stored in `KalimeraConversation.blob_url`.~~
> - Instead of the above machenism, we will have `input_file_url` and `output_file_url`
```python
ALLOWED_EXTENSIONS = {'mp3', 'wav', 'ogg', 'flac'}
class KalimeraVoice(Resource):
def get(self, app_model, end_user):
params = request.args.get_json()
if "conversation_id" in params:
conversation_id = request.args.get_json()["conversation_id"]
return file_helpers.get_signed_file_url(app_model + conversation)
# List all blobs under the tenant_id/end_user_id/ path
prefix = f"searchguru/{agent_id}/{end_user_id}/"
print("Prefix:", prefix)
blobs = widget_storage.get_list(starts_with=prefix)
# Download and parse the JSON content
conversations = []
for blob in blobs:
content = widget_storage.load(blob.name)
conversations.append(json.loads(content))
print(len(conversations))
return conversations
def post(self, app_model, end_user):
if 'voice_file' not in request.files:
return jsonify({"error": "No file part"}), 400
file = request.files['voice_file']
json_body = request.get_json()
if file.filename == '':
return jsonify({"error": "No selected file"}), 400
if file and allowed_file(file.filename):
# Retrieve additional fields
conversation_id = request.form.get('conversation_id', 'Unknown')
language = request.form.get('language', 'en')
# Store/Append metadata JSON in asynchonous call
cache_key = f"Kalimera_voice_{agentid}_{conversation_id}"
store_data = {
"input_file_url": file_url,
"output_file_url": output_file_url,
"conversation_id": conversation_id,
"language": language,
"profile_type": "inbound/outbound"
"created_at": current_timestamp - TimeDelta(len(file))
"updated_at": current_timestamp
}
stored_data = store_voice_conversation(cache_key, json_body)
store_voice_conversation.apply_async([agentid, end_user.id, conversation_id])
# Return WAV file as response
return {
"status": True,
"message": "Successfully stored."
}
@classmethod
def allowed_file(filename):
"""Check if the uploaded file has a valid extension."""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@classmethod
def upload_to_blob(file_content, filename):
"""Upload content to Azure Blob Storage and return the URL."""
blob_client = container_client.get_blob_client(filename)
blob_client.upload_blob(file_content, overwrite=True)
return blob_client.url
@classmethod
def store_voice_conversation(data, cache_key):
stored_data = redis_client.get(cache_key)
if not stored_data:
# Create conversation
conversation = KalimeraConversation(
name = data["name"] or "",
blob_url = 'kalimera/data["agent_id"]/data["conversation_id"]' + '.json',
created_at = str(datetime.now()),
updated_at = str(datetime.now()),
# Add more which are needed
)
redis_conversation = {
"name":data["name"] or "",
"history": [data]
"created_at":str(datetime.now()),
"updated_at":str(datetime.now()),
}
serialized_results = json.dumps(data_to_store, ensure_ascii=False)
cache_key = f"sg_conversation_{agentid}_{guid}"
redis_client.setex(cache_key, cls.CACHE_TTL, serialized_results)
# If exist then append last message.
stored_data["history"].append(data)
redis_client.set(cache_key, store_data)
api.add_resource(KalimeraVoice, 'kalimera/ai-agent/<uuid:agent_id>/voice')
```
> The above endpoint will take
> Above endpoint will return `.wav` file.
> To Store in azure blob we will use celery task `kalimera_store_conversation`. as below:
```python
def kalimera_store_conversation(agent_id, end_user_id, conversation_id):
blob_path = f'searchguru/{agent_id}/{end_user_id}/{conversation_id}.json'
is_exists = widget_storage.exists(blob_path)
conversation = None
if is_exists:
file_content = widget_storage.load(blob_path)
data = json.loads(file_content)
else:
data = {
"conversation_id": conversation_id,
"title": "New Conversation",
"date_created": str(datetime.now()),
"date_updated": str(datetime.now()),
"messages": []
}
logging.info(f'agent_id, {agent_id}')
if validate_uuid(agent_id):
agent = db.session.query(SearchGuruAgent).filter(SearchGuruAgent.agent_id == agent_id).first()
conversation = SearchGuruConversation(
# Store necessory data
)
db.session.add(conversation)
db.session.commit()
if data:
cache_key = f"Kalimera_voice_{agentid}_{conversation_id}"
new_data = redis_client.get(cache_key)
data.append(new_data)
file_content = json.dumps(data, indent=4)
content_settings = ContentSettings(
content_type="application/json"
)
widget_storage.save(blob_path, file_content, content_settings, overwrite=True)
else:
logging.info("No data to store")
```
---
---
## **Part 3: Conversation Statistics APIs**
### **Overview**
The Conversation Statistics API is designed to provide **Scheduled** insights into conversations between users and AI agents. These statistics are dynamically generated from actual conversations stored in the **Azure-blob**. The endpoints offer data such as sentiment analysis, call breakdowns, and historical trends. (**`This all APIs and data are dynamically generated from the conversation table and data(azure-blob)`**)
### **Key Features**
- **Real-time Statistics**: Tracks active inbound and outbound conversations.
- **Historical Data**: Provides monthly and weekly statistics of interactions (**`If we exclude the current day it will benefits to make it run once a day. Otherwise we need to make it updated by new conversation.`**)
- **Sentiment Analysis**: Categorizes conversations based on user sentiment.
- **Call Breakdown**: Analyzes conversation patterns and provides insights into engagement trends.
- **Recent Logs**: Displays the latest conversation data for quick monitoring.
### **Implementation Approach**
1. **Database Design**: The `Conversation` model stores key details such as agent name, sentiment, start time, and end time.
2. **Query Optimization**: Efficiently fetch large datasets using aggregation functions to compute statistics.
3. **Flask Endpoints**: API routes retrieve real-time and historical statistics for reporting dashboards. But data are calculated by worker asynchronously.
4. **JSON Response Formatting**: Ensures the data is structured for easy visualization and integration.
### **Expected Outcome**
With these APIs, users will be able to monitor AI-driven conversations effectively, gain insights into trends, and optimize interactions based on sentiment analysis and engagement patterns.
Sign in with Wallet
Connect another wallet