---
# System prepended metadata

title: 'Kɑlimerɑ.ɑi [Backend]'

---

# **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.

