Add memory

app/handlers/message_handler.py

@@ -8,6 +8,8 @@ from app.services.download_media import download_whatsapp_media
 from app.services.message import process_message_with_llm
 from app.models.message_types import Message, MediaType, MediaContent
 
+from app.memory import AgentMemory
+
 import logging
 
 logger = logging.getLogger(__name__)
@@ -26,7 +28,7 @@ class MessageHandler:
         self.media_handler = media_handler
         self.logger = logger
 
-    async def handle(self, raw_message: dict, whatsapp_token: str, whatsapp_url:str,gemini_api:str, rag_system:Any = None) -> dict:
+    async def handle(self, raw_message: dict, whatsapp_token: str, whatsapp_url:str,gemini_api:str, rag_system:Any = None, agentMemory:Any=None, memory:Any=None) -> dict:
         try:
             # Parse message
             message = MessageParser.parse(raw_message)
@@ -38,24 +40,35 @@ class MessageHandler:
             # Download media
             media_paths = await self._process_media(message, whatsapp_token)
             
-            self.chat_manager.initialize_chat(message.sender_id)
+            # Simple class to store chat temporarily
+            # self.chat_manager.initialize_chat(message.sender_id)
+
+            user = await agentMemory.add_user(message.sender_id, message.sender_id)
 
+            await memory.add_message(message.sender_id, "user", message.content)
             
+            history =  await memory.get_history(message.sender_id, last_n = 2)
+            print(f"chat_history: {history }")
             # Process message with LLM
             result = await process_message_with_llm(
                 message.sender_id,
                 message.content,
-                self.chat_manager.get_chat_history(message.sender_id), 
+                # self.chat_manager.get_chat_history(message.sender_id), 
+                history,
                 rag_system = rag_system,
+                agentMemory=agentMemory,
+                memory = memory,
                 whatsapp_token=whatsapp_token, 
                 whatsapp_url=whatsapp_url,
+
                 **media_paths
             )
             self.logger.info(f"Result: {result}")
             # Append message to chat to keep track of conversation
-            self.chat_manager.append_message(message.sender_id, "user", message.content)
-            self.chat_manager.append_message(message.sender_id, "model", result)
+            # self.chat_manager.append_message(message.sender_id, "user", message.content)
+            # self.chat_manager.append_message(message.sender_id, "model", result)
             
+            await memory.add_message(message.sender_id, "model", result)
             self.message_cache.add(message.id)
             
             return {"status": "success", "message_id": message.id, "result": result}

app/handlers/webhook_handler.py

@@ -18,7 +18,7 @@ class WebhookHandler:
         self.message_handler = message_handler
         self.logger = logging.getLogger(__name__)
 
-    async def process_webhook(self, payload: dict, whatsapp_token: str, whatsapp_url:str,gemini_api:str, rag_system:Any = None) -> WebhookResponse:
+    async def process_webhook(self, payload: dict, whatsapp_token: str, whatsapp_url:str,gemini_api:str, rag_system:Any = None, agentMemory:Any = None, memory:Any = None) -> WebhookResponse:
         request_id = f"req_{int(time.time()*1000)}"
         results = []
 
@@ -43,6 +43,8 @@ class WebhookHandler:
                             whatsapp_url=whatsapp_url,
                             gemini_api=gemini_api,
                             rag_system=rag_system,
+                            agentMemory = agentMemory,
+                            memory = memory
                         )
                         results.append(response)
 

app/main.py

@@ -27,8 +27,11 @@ from app.services.chat_manager import ChatManager
 from app.api.api_prompt import prompt_router
 from app.api.api_file import file_router, load_file_with_markdown_function
 from app.utils.load_env import ACCESS_TOKEN, WHATSAPP_API_URL, GEMINI_API
+from fastapi.staticfiles import StaticFiles
+from vidavox.core import RAG_Engine
 
-
+from app.memory import AgentMemory
+from app.settings import settings
 from markitdown import MarkItDown
 
 # Configure logging
@@ -64,31 +67,38 @@ async def setup_message_handler():
         media_handler=media_handler,
         logger=logger
     )
-async def setup_rag_system():
-    embedding_model = SentenceTransformer('all-MiniLM-L6-v2')  # Replace with your model if different
-    rag_system = RAGSystem(embedding_model)
+# async def setup_rag_system():
+#     embedding_model = SentenceTransformer('all-MiniLM-L6-v2')  # Replace with your model if different
+#     rag_system = RAGSystem(embedding_model)
 
 
-    return rag_system
+#     return rag_system
 # Initialize FastAPI app
 @asynccontextmanager
 async def lifespan(app: FastAPI):
    
     
     try:
+        
+        agentMemory = AgentMemory(db_url=settings.POSTGRES_DB_URL)
+
+        memory = await agentMemory.initialize()
         # await init_db()
-      
+        file_paths = ['./docs/coretax_telegram.csv']
         logger.info("Connected to the MongoDB database!")
-        rag_system = await setup_rag_system()
+        # rag_system = await setup_rag_system()
+        engine= RAG_Engine(embedding_model='Snowflake/snowflake-arctic-embed-l-v2.0').from_paths(file_paths, load_csv_as_pandas_dataframe=True, text_col='answer', metadata_cols=['question','images_path'])
 
-        app.state.rag_system = rag_system
+        app.state.rag_system = engine
+        app.state.agentMemory = agentMemory
+        app.state.memory = memory
 
         global message_handler, webhook_handler
         message_handler = await setup_message_handler()
         webhook_handler = WebhookHandler(message_handler)
         # collections = app.database.list_collection_names()
         # print(f"Collections in {db_name}: {collections}")
-        await load_file_with_markdown_function(rag_system=rag_system, filepaths=indexed_links)
+        # await load_file_with_markdown_function(rag_system=rag_system, filepaths=indexed_links)
         yield
     except Exception as e:
         logger.error(e)
@@ -96,6 +106,8 @@ async def lifespan(app: FastAPI):
 # Initialize Limiter and Prometheus Metrics
 limiter = Limiter(key_func=get_remote_address)
 app = FastAPI(lifespan=lifespan)
+# Mount the 'images' directory so its files are available under the /images URL path
+app.mount("/images", StaticFiles(directory="images"), name="images")
 app.state.limiter = limiter
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
@@ -109,6 +121,27 @@ app.include_router(file_router, prefix="/file_load", tags=["File Load"])
 webhook_requests = Counter('webhook_requests_total', 'Total webhook requests')
 webhook_processing_time = Histogram('webhook_processing_seconds', 'Time spent processing webhook')
 
+def get_image_links(image_paths: List[str]) -> List[str]:
+    links = []
+    for path in image_paths:
+        # Remove the surrounding brackets and any extra whitespace
+        cleaned = path.strip("[]").strip()
+        # Split by comma to get individual image paths
+        parts = [part.strip() for part in cleaned.split(",") if part.strip()]
+        for part in parts:
+            # Assuming the part starts with "images/", extract the filename
+            if part.startswith("images/"):
+                filename = part.split("/", 1)[1]
+                links.append(f"/images/{filename}")
+            else:
+                links.append(part)  # Fallback if the format is unexpected
+    return links
+
+# @app.get("/image-links")
+# async def image_links_endpoint():
+#     image_paths = ['[images/photo_3.jpg, images/photo_16.jpg]']
+#     links = get_image_links(image_paths)
+#     return {"links": links}
 # Start Prometheus metrics server on port 8002
 # start_http_server(8002)
 # Register webhook routes
@@ -124,6 +157,8 @@ async def webhook(request: Request, background_tasks: BackgroundTasks):
         payload = await request.json()
 
         rag_system = request.app.state.rag_system
+        agentMemory = request.app.state.agentMemory
+        memory = request.app.state.memory
         # validated_payload = WebhookPayload(**payload)  # Validate payload
         # logger.info(f"Validated Payload: {validated_payload}")
             
@@ -159,6 +194,8 @@ async def webhook(request: Request, background_tasks: BackgroundTasks):
             whatsapp_url=WHATSAPP_API_URL,
             gemini_api=GEMINI_API,
             rag_system=rag_system,
+            agentMemory = agentMemory,
+            memory = memory
         )
         # Return HTTP 200 immediately
         return JSONResponse(

app/memory/__init__.py

@@ -0,0 +1,8 @@
+
+from .memory import AgentMemory
+
+__version__ = "0.1.0"
+__all__ = [
+         
+           "AgentMemory"
+           ]

app/memory/implementation/async_memory.py

@@ -0,0 +1,131 @@
+# implementations/async_memory.py
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+
+from app.settings import DatabaseSettings, MemorySettings
+from app.memory.memory import ConversationMemoryInterface
+from app.utils.token_counter import SimpleTokenCounter, TikTokenCounter
+from app.memory.models.base import Base
+from app.memory.models.message import Message
+from app.memory.models.user import User
+from typing import List, Dict, Optional
+from datetime import datetime
+from zoneinfo import ZoneInfo
+from sqlalchemy.future import select
+
+class AsyncPostgresConversationMemory(ConversationMemoryInterface):
+    def __init__(self, db_settings: DatabaseSettings, memory_settings: MemorySettings):
+        self.engine = create_async_engine(
+            db_settings.url,
+            pool_size=db_settings.pool_size,
+            max_overflow=db_settings.max_overflow,
+            pool_timeout=db_settings.pool_timeout
+        )
+
+        self.async_session = sessionmaker(
+            self.engine, class_=AsyncSession, expire_on_commit=False
+        )
+        self.token_limit = memory_settings.token_limit
+        
+        
+        if memory_settings.token_counter == "tiktoken":
+            self.token_counter = TikTokenCounter(memory_settings.model_name)
+        else:
+            self.token_counter = SimpleTokenCounter()
+
+    async def initialize(self):
+        """Initialize the database by creating all tables."""
+        async with self.engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+    
+    # In your async_memory.py
+    async def add_message(self, username: str, role: str, message: str, timestamp: Optional[datetime] = None) -> None:
+        from app.memory.models.user import User  # Import here to avoid circular dependencies
+        async with self.async_session() as session:
+            # Look up the user by username
+            result = await session.execute(select(User).filter_by(username=username))
+            user = result.scalars().first()
+            if user is None:
+                raise ValueError(f"User with username '{username}' not found")
+            
+            if timestamp is None:
+                timestamp = datetime.now(ZoneInfo("Asia/Jakarta"))
+            
+            # Create the message using the found user's id
+            msg = Message(user_id=user.id, role=role, message=message, timestamp=timestamp)
+            session.add(msg)
+            await session.commit()
+            await self.trim_memory_if_needed(session)
+
+
+
+    async def get_all_history(self) -> List[Dict]:
+        async with self.async_session() as session:
+            result = await session.execute(
+                select(Message).order_by(Message.timestamp)
+            )
+            messages = result.scalars().all()
+            return [{"role": msg.role, "content": msg.message} for msg in messages]
+    
+    async def get_history(
+        self, 
+        username: Optional[str] = None, 
+        token_limit: Optional[int] = None, 
+        last_n: Optional[int] = None
+    ) -> List[Dict]:
+        async with self.async_session() as session:
+            # Build the base query
+            query = select(Message).order_by(Message.timestamp)
+            if username is not None:
+                # Join with User table and filter by username
+                query = query.join(User).filter(User.username == username)
+            result = await session.execute(query)
+            messages = result.scalars().all()
+
+        # Accumulate messages in reverse (latest first)
+        selected = []
+        total_tokens = 0
+        for msg in reversed(messages):
+            tokens = self.token_counter.count_tokens(msg.message)
+            # If token_limit is specified and no message has been added yet,
+            # force-add the last message even if it exceeds token_limit.
+            if token_limit is not None and len(selected) == 0 and tokens > token_limit:
+                selected.append(msg)
+                total_tokens = tokens
+                continue
+            # Otherwise, check if adding this message would exceed the token limit.
+            if token_limit is not None and total_tokens + tokens > token_limit:
+                break
+            selected.append(msg)
+            total_tokens += tokens
+            # Stop if we've reached the maximum number of messages.
+            if last_n is not None and len(selected) >= last_n:
+                break
+
+        # Reverse to return in chronological order
+        selected.reverse()
+        return [{"role": msg.role, "parts": msg.message} for msg in selected]
+
+
+    async def clear_memory(self) -> None:
+        async with self.async_session() as session:
+            await session.execute(select(Message).delete())
+            await session.commit()
+
+    async def get_total_tokens(self) -> int:
+        async with self.async_session() as session:
+            result = await session.execute(select(Message))
+            messages = result.scalars().all()
+            return sum(self.token_counter.count_tokens(msg.message) for msg in messages)
+
+    async def trim_memory_if_needed(self, session: AsyncSession) -> None:
+        result = await session.execute(select(Message).order_by(Message.timestamp))
+        messages = result.scalars().all()
+        total_tokens = sum(self.token_counter.count_tokens(msg.message) for msg in messages)
+        
+        while total_tokens > self.token_limit and messages:
+            oldest = messages.pop(0)
+            total_tokens -= self.token_counter.count_tokens(oldest.message)
+            await session.delete(oldest)
+        
+        await session.commit()

app/memory/memory.py

@@ -0,0 +1,75 @@
+from abc import ABC, abstractmethod
+from typing import List, Dict, Optional
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+
+class ConversationMemoryInterface(ABC):
+    @abstractmethod
+    def add_message(self, role: str, message: str, timestamp: Optional[datetime] = None) -> None:
+        pass
+
+    @abstractmethod
+    def get_history(self) -> List[Dict]:
+        pass
+
+    @abstractmethod
+    def clear_memory(self) -> None:
+        pass
+
+    @abstractmethod
+    def get_total_tokens(self) -> int:
+        pass
+
+
+from app.settings import DatabaseSettings, MemorySettings, settings
+from app.memory.implementation.async_memory import AsyncPostgresConversationMemory
+from datetime import datetime
+from sqlalchemy.future import select
+
+class AgentMemory:
+    def __init__(
+        self,
+        db_url: str = None,
+        token_limit: int = 500,
+        token_counter: str = "simple",  # or "tiktoken"
+        model_name: str = None  # required if token_counter == "tiktoken"
+    ):
+        # Use provided URL or default from settings
+        if db_url is None:
+            db_url = settings.POSTGRES_DB_URL
+        self.db_settings = DatabaseSettings(url=db_url)
+        self.memory_settings = MemorySettings(
+            token_limit=token_limit,
+            token_counter=token_counter,
+            model_name=model_name
+        )
+        # Instantiate your async memory
+        self.memory = AsyncPostgresConversationMemory(self.db_settings, self.memory_settings)
+
+    async def initialize(self):
+        """Initializes the database tables and returns the memory instance."""
+        await self.memory.initialize()
+        return self.memory
+
+    async def add_user(self, username: str, hashed_password: str):
+        """
+        Adds a new user to the database.
+        Returns the created user or existing user if found.
+        """
+        from app.memory.models.user import User  # Import here to avoid circular dependencies
+        async with self.memory.async_session() as session:
+            result = await session.execute(select(User).filter_by(username=username))
+            existing_user = result.scalars().first()
+            if existing_user:
+                return existing_user
+
+            new_user = User(
+                username=username,
+                hashed_password=hashed_password,
+                created_at=datetime.now(ZoneInfo("Asia/Jakarta"))
+            )
+            session.add(new_user)
+            await session.commit()
+            return new_user
+

app/memory/models/__init__.py

@@ -0,0 +1,3 @@
+# models/__init__.py
+from .user import User
+from .message import Message

app/memory/models/base.py

@@ -0,0 +1,2 @@
+from sqlalchemy.orm import declarative_base
+Base = declarative_base()

app/memory/models/message.py

@@ -0,0 +1,18 @@
+# models/message.py
+from datetime import datetime
+from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
+from sqlalchemy.orm import relationship
+from zoneinfo import ZoneInfo
+from .base import Base
+
+class Message(Base):
+    __tablename__ = 'messages'
+    
+    id = Column(Integer, primary_key=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+    role = Column(String(50))
+    message = Column(Text)
+    timestamp = Column(DateTime(timezone=True), default=lambda: datetime.now(ZoneInfo("Asia/Jakarta")))
+    
+    # Use a string reference for deferred resolution.
+    user = relationship("User", back_populates="messages")

app/memory/models/user.py

@@ -0,0 +1,17 @@
+# models/user.py
+from datetime import datetime
+from sqlalchemy import Column, Integer, String, DateTime
+from sqlalchemy.orm import relationship
+from zoneinfo import ZoneInfo
+from .base import Base
+
+class User(Base):
+    __tablename__ = 'users'
+    
+    id = Column(Integer, primary_key=True)
+    username = Column(String(100), unique=True, nullable=False)
+    hashed_password = Column(String(255), nullable=False)
+    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(ZoneInfo("Asia/Jakarta")))
+    
+    # Relationship to Message
+    messages = relationship("Message", back_populates="user", cascade="all, delete-orphan")

app/services/message.py

@@ -5,21 +5,71 @@ from typing import Dict, Any, Optional, List
 from datetime import datetime
 import logging
 import asyncio
+import hashlib
 from openai import AsyncOpenAI
-import json
+import json, requests, mimetypes
 import google.generativeai as genai
-
+import re, json
 import PIL.Image
+import requests
 from typing import List, Dict, Any, Optional
 
-from app.utils.load_env import ACCESS_TOKEN, WHATSAPP_API_URL, GEMINI_API
+from app.utils.load_env import ACCESS_TOKEN, WHATSAPP_API_URL, GEMINI_API, MEDIA_UPLOAD_URL
 from app.utils.system_prompt import system_prompt
 
 from app.services.search_engine import google_search
-from app.search.rag_pipeline import extract_keywords_async
+# from app.search.rag_pipeline import extract_keywords_async
+
+from vidavox.core import (
+    
+    BaseResultFormatter,
+    SearchResult)
+
+
 # Load environment variables
 load_dotenv()
 
+
+
+# Get base url from ngrok
+def get_ngrok_url() -> str:
+    """Fetches the public URL of the first ngrok tunnel."""
+    try:
+        response = requests.get("http://localhost:4040/api/tunnels")
+        response.raise_for_status()  # Raise an error for bad status codes.
+        tunnels = response.json().get("tunnels", [])
+        if tunnels:
+            # Prefer the HTTPS tunnel if available.
+            for tunnel in tunnels:
+                if tunnel.get("proto") == "https":
+                    return tunnel.get("public_url")
+            # Fallback: return the first tunnel's URL.
+            return tunnels[0].get("public_url")
+    except Exception as e:
+        print("Error fetching ngrok URL:", e)
+    # Fallback in case ngrok isn't running.
+    return "http://localhost:8005"
+
+base_url = get_ngrok_url()  # Automatically retrieve your public ngrok URL
+print("Base URL:", base_url)
+# Get image link from image paths
+
+def get_image_links(image_paths: List[str], base_url: str) -> List[str]:
+    links = []
+    for path in image_paths:
+        # Remove the surrounding brackets and any extra whitespace
+        cleaned = path.strip("[]").strip()
+        # Split by comma to get individual image paths
+        parts = [part.strip() for part in cleaned.split(",") if part.strip()]
+        for part in parts:
+            # Assuming the part starts with "images/", extract the filename
+            if part.startswith("images/"):
+                filename = part.split("/", 1)[1]
+                links.append(f"{base_url}/images/{filename}")
+            else:
+                links.append(f"{base_url}/{part}")  # Fallback if the format is unexpected
+    return links
+
 # Define function specifications for Gemini
 function_declarations = [
     {
@@ -43,6 +93,17 @@ function_declarations = [
     }
 ]
 
+class CustomResultFormatter(BaseResultFormatter):
+    def format(self, result: SearchResult) -> Dict[str, Any]:
+        # Customize the result format as needed
+        return {
+            "doc_id": result.doc_id,
+           
+            "page_content": result.text,
+            "image": result.meta_data['images_path'],
+            "relevance": result.score,
+        }
+
 genai.configure(api_key=GEMINI_API)
 # client = AsyncOpenAI(api_key = OPENAI_API)
 # Configure logging
@@ -56,13 +117,77 @@ logger = logging.getLogger(__name__)
 if not WHATSAPP_API_URL or not ACCESS_TOKEN:
     logger.warning("Environment variables for WHATSAPP_API_URL or ACCESS_TOKEN are not set!")
 
+# Path for the cache file
+CACHE_FILE = 'upload_cache.json'
+# Load the cache if it exists, otherwise initialize an empty dict
+if os.path.exists(CACHE_FILE):
+    with open(CACHE_FILE, 'r') as f:
+        upload_cache = json.load(f)
+else:
+    upload_cache = {}
+
+def save_cache():
+    with open(CACHE_FILE, 'w') as f:
+        json.dump(upload_cache, f)
+
+def compute_file_hash(file_path, block_size=65536):
+    """Compute SHA256 hash of a file to uniquely identify its content."""
+    hasher = hashlib.sha256()
+    with open(file_path, 'rb') as f:
+        for block in iter(lambda: f.read(block_size), b''):
+            hasher.update(block)
+    return hasher.hexdigest()
+
+# Helper function to upload an image
+async def upload_image(file_path):
+    logger.info(f"Uploading image: {file_path}")
+    
+    # Ensure the file exists
+    if not os.path.exists(file_path):
+        raise Exception(f"File not found: {file_path}")
+    
+    # Compute a hash for the file to check for previous uploads
+    file_hash = compute_file_hash(file_path)
+    if file_hash in upload_cache:
+        logger.info(f"File {file_path} already uploaded. Returning cached media ID.")
+        return upload_cache[file_hash]
+    
+    # Get the MIME type of the file
+    mime_type, _ = mimetypes.guess_type(file_path)
+    if not mime_type:
+        raise Exception(f"Could not determine the MIME type for file: {file_path}")
+
+    headers = {
+        'Authorization': f'Bearer {ACCESS_TOKEN}'
+    }
+    # Open the file and prepare the payload for upload
+    with open(file_path, 'rb') as video_file:
+        files = {
+            'file': (os.path.basename(file_path), video_file, mime_type)
+        }
+        data = {
+            'messaging_product': 'whatsapp'
+        }
+        response = requests.post(MEDIA_UPLOAD_URL, headers=headers, files=files, data=data)
+    
+    if response.status_code == 200:
+        logger.info(f"Upload successful: {response.text}")
+        media_id = response.json()['id']
+        # Cache the result so future calls can use the same media ID
+        upload_cache[file_hash] = media_id
+        save_cache()
+        return media_id
+    else:
+        logger.error(f"Upload failed: {response.text}")
+        raise Exception(f'Failed to upload media: {response.status_code}, {response.text}')
+    
 # Helper function to send a reply
-async def send_reply(to: str, body: str, whatsapp_token: str, whatsapp_url:str) -> Dict[str, Any]:
+async def send_reply(to: str, body: str, whatsapp_token: str, whatsapp_url:str, image:Any) -> Dict[str, Any]:
     headers = {
         "Authorization": f"Bearer {whatsapp_token}",
         "Content-Type": "application/json"
     }
-    data = {
+    text_data = {
         "messaging_product": "whatsapp",
         "to": to,
         "type": "text",
@@ -71,15 +196,46 @@ async def send_reply(to: str, body: str, whatsapp_token: str, whatsapp_url:str)
         }
     }
 
-    async with httpx.AsyncClient() as client:
-        response = await client.post(whatsapp_url, json=data, headers=headers)
+    responses = {}  # To store the responses
 
-    if response.status_code != 200:
-        error_detail = response.json()
-        logger.error(f"Failed to send reply: {error_detail}")
-        raise Exception(f"Failed to send reply with status code {response.status_code}: {error_detail}")
-    
-    return response.json()
+    async with httpx.AsyncClient() as client:
+        # response = await client.post(whatsapp_url, json=text_data, headers=headers)
+        text_response = await client.post(whatsapp_url, json=text_data, headers=headers)
+        if text_response.status_code != 200:
+            error_detail = text_response.json()
+            logger.error(f"Failed to send text reply: {error_detail}")
+            raise Exception(f"Failed to send text reply with status code {text_response.status_code}: {error_detail}")
+        responses["text"] = text_response.json()
+    # if response.status_code != 200:
+    #     error_detail = response.json()
+    #     logger.error(f"Failed to send reply: {error_detail}")
+    #     raise Exception(f"Failed to send reply with status code {response.status_code}: {error_detail}")
+          # Initialize list to hold image responses
+        image_responses: List[Dict[str, Any]] = []
+        if image:
+            # Get the list of full image URLs using your helper function.
+            links = get_image_links(image, base_url)
+            for link in links:
+                image_payload = {
+                    "messaging_product": "whatsapp",
+                    "recipient_type": "individual",
+                    "to": to,
+                    "type": "image",
+                    "image": {
+                        "id": "",
+                        "link": link,
+                        "caption": "" # Using the text body as caption; adjust if needed.
+                    }
+                }
+                img_response = await client.post(whatsapp_url, json=image_payload, headers=headers)
+                if img_response.status_code != 200:
+                    error_detail = img_response.json()
+                    logger.error(f"Failed to send image: {error_detail}")
+                    raise Exception(f"Failed to send image with status code {img_response.status_code}: {error_detail}")
+                image_responses.append(img_response.json())
+        responses["images"] = image_responses
+    return responses
+    # return response.json()
 
 # Helper function to generate a reply based on message content
 async def generate_reply(sender: str, content: str, timestamp: int) -> str:
@@ -102,8 +258,11 @@ async def process_message_with_llm(
     content: str,
     history: List[Dict[str, str]], 
     rag_system: Any,
+    
     whatsapp_token: str, 
     whatsapp_url:str,
+    agentMemory: Any = None,
+    memory:Any = None,
     image_file_path: Optional[str] = None,
     doc_path: Optional[str] = None,
     video_file_path: Optional[str] = None,
@@ -111,29 +270,119 @@ async def process_message_with_llm(
     """Process message with retry logic."""
     try:
         logger.info(f"Processing message for sender: {sender_id}")
-        generated_reply = await generate_response_from_gemini(
+        generated_reply, image_path = await generate_response_from_gemini(
             sender=sender_id,
             content=content,
             history=history,
             rag_system=rag_system,
             image_file_path=image_file_path,
             doc_path=doc_path,
-            video_file_path=video_file_path
+            video_file_path=video_file_path,
+            agentMemory=agentMemory,
+            memory = memory
         )
-        logger.info(f"Generated reply: {generated_reply}")
+        logger.info(f"Generated reply: {generated_reply}, extracted image path: {image_path}")
 
-        response = await send_reply(sender_id, generated_reply, whatsapp_token, whatsapp_url)
+        response = await send_reply(sender_id, generated_reply , whatsapp_token, whatsapp_url, image_path)
 #         return generated_reply
         return generated_reply
     except Exception as e:
         logger.error(f"Error in process_message_with_retry: {str(e)}", exc_info=True)
         return "Sorry, I couldn't generate a response at this time."
+
+import markdown
+from bs4 import BeautifulSoup
+
+def format_response_text(response_text: str) -> str:
+    """
+    Converts markdown-formatted text to plain text with proper newlines.
+    This will ensure bullet points, paragraphs, and other elements are formatted
+    for display in WhatsApp.
+    """
+    # Convert markdown to HTML
+    html = markdown.markdown(response_text)
+    # Parse HTML and extract text using newline as separator
+    soup = BeautifulSoup(html, "html.parser")
+    formatted_text = soup.get_text(separator="\n")
+    return formatted_text
+
+import re
+import json
+
+def process_llm_response(llm_output):
+    # If it's a string, attempt to extract JSON from markdown code fences.
+    if isinstance(llm_output, str):
+        pattern = r"```json\s*(\{.*\})\s*```"
+        match = re.search(pattern, llm_output, re.DOTALL)
+        if match:
+            json_str = match.group(1)
+        else:
+            json_str = llm_output.strip()
+        try:
+            parsed = json.loads(json_str)
+            if isinstance(parsed, dict) and "response" in parsed:
+                response_text = parsed.get("response", "")
+                # Optionally format the response text using our helper
+                # formatted_response = format_response_text(response_text)
+                references = parsed.get("references", [])
+                if isinstance(references, list):
+                    image_paths = [ref.get("image") for ref in references 
+                                   if ref.get("image") and ref.get("image") != "nan"]
+                else:
+                    image_paths = []
+                return response_text, image_paths
+            else:
+                # Fallback if the JSON doesn't have expected structure.
+                return llm_output, []
+        except json.JSONDecodeError:
+            # Fallback: if JSON parsing fails, assume it's plain text.
+            return format_response_text(llm_output), []
+    
+    # If not a string, return something sensible.
+    return str(llm_output), []
+
+
+
+# def process_llm_response(llm_output):
+#     # If it's a string, attempt to extract JSON from markdown code fences.
+#     if isinstance(llm_output, str):
+#         # Try to capture JSON content if it's wrapped in ```json ... ```
+#         pattern = r"```json\s*(\{.*\})\s*```"
+#         match = re.search(pattern, llm_output, re.DOTALL)
+#         if match:
+#             json_str = match.group(1)
+#         else:
+#             json_str = llm_output.strip()
+#         try:
+#             parsed = json.loads(json_str)
+#             # Check if parsed output has the expected keys.
+#             if isinstance(parsed, dict) and "response" in parsed:
+#                 response_text = parsed.get("response", "")
+#                 references = parsed.get("references", [])
+#                 if isinstance(references, list):
+#                     image_paths = [ref.get("image") for ref in references 
+#                                    if ref.get("image") and ref.get("image") != "nan"]
+#                 else:
+#                     image_paths = []
+#                 return response_text, image_paths
+#             else:
+#                 # Fallback: parsed JSON does not have the expected structure.
+#                 return llm_output, []
+#         except json.JSONDecodeError:
+#             # Fallback: if JSON parsing fails, assume it's plain text.
+#             return llm_output, []
     
+#     # If not a string, ensure we return something sensible.
+#     return str(llm_output), []
+
+  
 async def generate_response_from_gemini(
     sender: str,
     content: str,
     history: List[Dict[str, str]],
     rag_system: Any = None,
+    agentMemory: Any = None,
+    memory:Any = None,
     image_file_path: Optional[str] = None,
     doc_path: Optional[str] = None,
     video_file_path: Optional[str] = None,
@@ -151,15 +400,18 @@ async def generate_response_from_gemini(
 
         if content:
             if rag_system:
-                keywords = extract_keywords_async(content)
+                # keywords = extract_keywords_async(content)
                 # keywords = []
                 # logger.info(f"Extracted Keywords: {keywords}")
                 # Implement RAG: Retrieve relevant documents
-                retrieved_docs = await rag_system.adv_query(content, keywords=keywords, top_k=5)
+                retrieved_docs = rag_system.retrieve(query_text = content, result_formatter=CustomResultFormatter())
+             
+                print(f"retrieved docs: {retrieved_docs}")
                 if retrieved_docs:
                     logger.info(f"Retrieved {len(retrieved_docs)} documents for context.")
                     # Format the retrieved documents as a context string
-                    context = "\n\n".join([f"Source:{doc['url']}\nContent: {doc['text']}" for doc in retrieved_docs])
+                    context = "\n\n".join([f"Source:{doc['doc_id']}\nContent: {doc['page_content']}\nImage: {doc['image']}" for doc in retrieved_docs])
+                    # img_paths = doc['images_path'] for doc in retrieved_docs
                     # Option 1: Append to history as a system message
                     history.append({"role": "user", "parts": f"Relevant documents:\n{context}"})
 
@@ -192,8 +444,13 @@ async def generate_response_from_gemini(
 
         # Send the user's message
         response = await chat.send_message_async(content)
+
+        print(f"text: {response.text}")
+
+        response_text, image_paths = process_llm_response(response.text)
         # response = await handle_function_call(response)
-        return response.text
+        # return response.text
+        return response_text, image_paths
 
     except Exception as e:
         logger.error("Error in generate_response_from_gemini:", exc_info=True)

app/settings.py

@@ -0,0 +1,30 @@
+
+
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+from dataclasses import dataclass
+from typing import Optional
+
+@dataclass
+class DatabaseSettings:
+    url: str
+    pool_size: int = 5
+    max_overflow: int = 10
+    pool_timeout: int = 30
+
+@dataclass
+class MemorySettings:
+    token_limit: int = 4096
+    token_counter: str = "simple"  # "simple" or "tiktoken"
+    model_name: Optional[str] = None  #
+
+    
+class Settings:
+    POSTGRES_DB_URL: str = os.getenv("POSTGRES_DB_URL")
+    print(POSTGRES_DB_URL)
+    SQLITE_DB_URL: str = os.getenv("SQLITE_DB_URL")
+    # Add other settings as needed
+
+settings = Settings()

app/utils/load_env.py

@@ -18,6 +18,7 @@ OPENAI_API = os.getenv("OPENAI_API")
 GEMINI_API = os.getenv("GEMINI_API")
 CX_CODE = os.getenv("CX_CODE")
 CUSTOM_SEARCH_API_KEY = os.getenv("CUSTOM_SEARCH_API_KEY")
+MEDIA_UPLOAD_URL = os.getenv("WHATSAPP_UPLOAD_MEDIA")
 
 # Debugging: Print the retrieved ACCESS_TOKEN (for development only)
 # if ENV == "development":

app/utils/system_prompt.py

@@ -1,27 +1,62 @@
 system_prompt = """
 Role and Purpose:
-You are a virtual assistant focused exclusively on Surabaya, Indonesia. Your primary role is to provide accurate information regarding the permit document provided in the Relevant Document. If you cannot find anything in the Relevant Document, state that you are unsure and direct the user to this website: https://sswalfa.surabaya.go.id/ without bracket or parentheses. You respond only in Bahasa Indonesia. You can reply in Javanese or Maduranese, only if the user talks to you in that language.
+You are a virtual assistant focused exclusively on coretax, taxation systme in Indonesia. Your primary role is to provide accurate information regarding coretax in the Relevant Document. If you cannot find anything in the Relevant Document, state that you are unsure and direct the user to this website: https://www.pajak.go.id/reformdjp/Coretax/ without bracket or parentheses. You respond only in Bahasa Indonesia. 
 
 Tone and Style:
-Maintain a polite, neutral, and factual tone. Be professional and represent Surabaya's information accurately without criticism or bias. Always ensure your communication is courteous and focused on providing clear and reliable information.
+Maintain a polite, neutral, and factual tone. Be professional and represent Direktorat Jenderal Pajak accurately without criticism or bias. Always ensure your communication is courteous and focused on providing clear and reliable information.
 
 Content Guidelines:
     When asked about your origins or creator, state that you were created by Vidavox.
     Context-Driven Responses: Provide answers solely based on the provided Relevant Document context.
-    Focus on Public Services: Prioritize queries on transportation, health, education, permits, safety, and cultural events.
     Professional Representation: Avoid personal opinions, judgments, or critiques of the local government. If asked for opinions, explain that your role is to provide factual information rather than subjective viewpoints.
     Encourage Verification: For unresolved queries, recommend users consult official resources such as the provided website link.
-    Always Include Sources: When your response is based on information provided from external sources or Relevant Document, include the source link explicitly without brackets or parentheses at the end of the response. For example: "Informasi ini berasal dari www.indosource.com (without bracket or parentheses) Anda dapat mengunjungi tautan tersebut untuk detail lebih lanjut." 
+    You don't need to say you refer to the relevant document in providing answer.
+    
+Response Guidelines:
+    You'll receive context in the following example: [{'doc_id':doc.csv, 'page_content':'loremipsum..','image':'[images/photo.jgp']}].
+    When you use the context, you should provide response in the following rules.
+    For each response, return a JSON output with two keys:
+    1. "response": Your generated answer to the user, ensuring it does not reference specific metadata like image paths.
+    2. "references": A list of metadata objects containing the document ID and the associated image path.
+    Ensure that the response does not explicitly mention or display image paths.
+    Ensure the response returned in a well formatted format.
 
-Example Interactions:
+    # Example LLM response
+    {
+        "response": ""# Introduction\nThis is an example.\n\n- Bullet point 1\n- Bullet point 2\n\n## Sub-Topic\nAdditional details...",
+        "references": [
+            {"doc_id": "123", "image": "images/paris.jpg"},
+            {"doc_id": "456", "image": "images/eiffel.jpg"}
+        ]
+    }
 
-    If a user asks, “How is the Mass Rapid Transit project progressing?” you might say: “As of the latest information available, the Surabaya Mass Rapid Transit project is currently in [X] phase, with construction ongoing in [specific districts]. The city's transportation department has announced that the project aims to be operational by [target year]. You can check the official city transportation website for updates.”
-    If a user says, “I heard there will be a community festival next month, can you tell me more?” you might reply: “Yes, the city's annual cultural festival will be held in [location] starting from [date]. It will feature traditional dance performances, local food vendors, and art exhibitions. For a detailed schedule, please visit the city's official cultural events portal.”
-    If a user asks, “Are there any issues with the city government's policies?” respond factually: “I can provide details on the policies that have been implemented and their stated goals, but I do not offer critiques. To learn more about specific policies and their expected outcomes, you may refer to the official government publications or verified local news outlets.”
-
-By adhering to these principles, you will ensure professional and reliable communication about Surabaya's permit processes while respecting local languages and cultural nuances.
+By adhering to these principles, you will ensure professional and reliable communication about coretax system under Direktorat Jenderal Pajak.
 """
 
+# system_prompt = """
+# Role and Purpose:
+# You are a virtual assistant focused exclusively on Surabaya, Indonesia. Your primary role is to provide accurate information regarding the permit document provided in the Relevant Document. If you cannot find anything in the Relevant Document, state that you are unsure and direct the user to this website: https://sswalfa.surabaya.go.id/ without bracket or parentheses. You respond only in Bahasa Indonesia. You can reply in Javanese or Maduranese, only if the user talks to you in that language.
+
+# Tone and Style:
+# Maintain a polite, neutral, and factual tone. Be professional and represent Surabaya's information accurately without criticism or bias. Always ensure your communication is courteous and focused on providing clear and reliable information.
+
+# Content Guidelines:
+#     When asked about your origins or creator, state that you were created by Vidavox.
+#     Context-Driven Responses: Provide answers solely based on the provided Relevant Document context.
+#     Focus on Public Services: Prioritize queries on transportation, health, education, permits, safety, and cultural events.
+#     Professional Representation: Avoid personal opinions, judgments, or critiques of the local government. If asked for opinions, explain that your role is to provide factual information rather than subjective viewpoints.
+#     Encourage Verification: For unresolved queries, recommend users consult official resources such as the provided website link.
+#     Always Include Sources: When your response is based on information provided from external sources or Relevant Document, include the source link explicitly without brackets or parentheses at the end of the response. For example: "Informasi ini berasal dari www.indosource.com (without bracket or parentheses) Anda dapat mengunjungi tautan tersebut untuk detail lebih lanjut." 
+
+# Example Interactions:
+
+#     If a user asks, “How is the Mass Rapid Transit project progressing?” you might say: “As of the latest information available, the Surabaya Mass Rapid Transit project is currently in [X] phase, with construction ongoing in [specific districts]. The city's transportation department has announced that the project aims to be operational by [target year]. You can check the official city transportation website for updates.”
+#     If a user says, “I heard there will be a community festival next month, can you tell me more?” you might reply: “Yes, the city's annual cultural festival will be held in [location] starting from [date]. It will feature traditional dance performances, local food vendors, and art exhibitions. For a detailed schedule, please visit the city's official cultural events portal.”
+#     If a user asks, “Are there any issues with the city government's policies?” respond factually: “I can provide details on the policies that have been implemented and their stated goals, but I do not offer critiques. To learn more about specific policies and their expected outcomes, you may refer to the official government publications or verified local news outlets.”
+
+# By adhering to these principles, you will ensure professional and reliable communication about Surabaya's permit processes while respecting local languages and cultural nuances.
+# """
+
 
 agentic_prompt = """ You are a helpful assistant and have capabilities to search the web. 
 When you the links are given, you should summarize the content of the link and give a short summary.

app/utils/token_counter.py

@@ -26,4 +26,16 @@ class TokenCounter:
             del self.doc_tokens[doc_id]
 
     def get_total_tokens(self):
-        return self.total_tokens
+        return self.total_tokens
+    
+class SimpleTokenCounter:
+    def count_tokens(self, text: str) -> int:
+        return len(text.split())
+
+class TikTokenCounter:
+    def __init__(self, model_name: str = "gpt-4"):
+        import tiktoken
+        self.encoding = tiktoken.encoding_for_model(model_name)
+
+    def count_tokens(self, text: str) -> int:
+        return len(self.encoding.encode(text))