Merge pull request #2 from ric9176/add-whatsapp-interface

Add whatsapp interface, refactor project structure, simplify graph state

agent/utils/nodes.py

@@ -175,7 +175,7 @@ async def write_memory(state: AgentState, config: RunnableConfig, store: BaseSto
     # Store the updated memory using async interface
     await store.aput(namespace, "user_memory", {"memory": new_memory.content})
     
-    
+    # Update the state with the new memory
     return state
 
 # Initialize tool node

agent/utils/state.py

@@ -1,4 +1,4 @@
-from typing import Annotated, TypedDict
+from typing import Annotated, TypedDict, Optional
 from langgraph.graph.message import add_messages
 
 class AgentState(TypedDict):

interfaces/__init__.py

@@ -0,0 +1 @@
+"""Interfaces package for Chief Joy Officer.""" 

interfaces/chainlit/__init__.py

@@ -0,0 +1 @@
+"""Chainlit interface for Chief Joy Officer.""" 

app.py → interfaces/chainlit/app.py

@@ -1,11 +1,19 @@
 import uuid
+import os
+import json
 from langchain_core.messages import HumanMessage, AIMessage, AIMessageChunk
 from langchain.schema.runnable.config import RunnableConfig
 import chainlit as cl
+import sys
+from pathlib import Path
+
+# Add the project root to the Python path
+project_root = str(Path(__file__).parent.parent.parent)
+if project_root not in sys.path:
+    sys.path.append(project_root)
+
 from agent import create_agent_graph, get_checkpointer
 from agent.utils.state import AgentState
-import os
-import json
 
 SHORT_TERM_MEMORY_DB_PATH = "data/short_term.db"
 
@@ -19,9 +27,6 @@ async def on_chat_start():
         session_id = str(uuid.uuid4())
         cl.user_session.set("session_id", session_id)
         
-        # Initialize empty message history
-        cl.user_session.set("message_history", [])
-        
         welcome_message = cl.Message(
             content="Hello! I'm your chief joy officer, here to help you with finding fun things to do in London!",
             author="Assistant"
@@ -30,21 +35,14 @@ async def on_chat_start():
         
     except Exception as e:
         print(f"Error in chat initialization: {str(e)}")
-        error_message = cl.Message(
+        await cl.Message(
             content="I apologize, but I encountered an error during initialization. Please try refreshing the page.",
             author="System"
-        )
-        await error_message.send()
+        ).send()
 
 @cl.on_message
 async def on_message(message: cl.Message):
     """Handle incoming messages and stream responses"""
-    # Get or create session ID
-    session_id = cl.user_session.get("session_id")
-    if not session_id:
-        session_id = str(uuid.uuid4())
-        cl.user_session.set("session_id", session_id)
-    
     # Initialize response message
     msg = cl.Message(content="")
     
@@ -54,39 +52,24 @@ async def on_message(message: cl.Message):
                 # Create graph with memory
                 graph = await create_agent_graph(saver)
                 
-                # Get message history and add current message
-                message_history = cl.user_session.get("message_history", [])
-                current_message = HumanMessage(content=message.content)
-                message_history.append(current_message)
+                # Get session ID
+                session_id = cl.user_session.get("session_id")
                 
-                # Create current state
-                current_state = AgentState(
-                    messages=message_history,
-                    context=cl.user_session.get("last_context", [])
-                )
-                
-                # Stream the response
+                # Process through graph with current message
                 async for chunk in graph.astream(
-                    current_state,
-                    config={"configurable": {"thread_id": session_id}},
+                    {"messages": [HumanMessage(content=message.content)]},
+                    {"configurable": {"thread_id": session_id}},
                     stream_mode="messages"
                 ):
-                    # Handle different node outputs
-                    if isinstance(chunk[0], AIMessageChunk):
+                    if chunk[1]["langgraph_node"] == "agent" and isinstance(
+                        chunk[0], (AIMessageChunk, AIMessage)
+                    ):
                         await msg.stream_token(chunk[0].content)
-                    elif isinstance(chunk[0], AIMessage):
-                        if chunk[0] not in message_history:
-                            message_history.append(chunk[0])
                 
                 # Get final state
                 final_state = await graph.aget_state(
                     config={"configurable": {"thread_id": session_id}}
                 )
-                
-                # Update session state
-                if final_state:
-                    cl.user_session.set("message_history", message_history)
-                    cl.user_session.set("last_context", final_state.values.get("context", []))
         
         # Send the final message
         await msg.send()

interfaces/whatsapp/webhook_endpoint.py

@@ -0,0 +1,5 @@
+from fastapi import FastAPI
+from ai_companion.interfaces.whatsapp.whatsapp_response import whatsapp_router
+
+app = FastAPI()
+app.include_router(whatsapp_router) 

interfaces/whatsapp/whatsapp_response.py

@@ -0,0 +1,169 @@
+import logging
+import os
+from io import BytesIO
+from typing import Dict
+
+import httpx
+from fastapi import APIRouter, Request, Response
+from langchain_core.messages import HumanMessage
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+from agent import create_agent_graph
+from agent.utils.state import AgentState
+
+logger = logging.getLogger(__name__)
+
+# Global module instances
+speech_to_text = SpeechToText()
+text_to_speech = TextToSpeech()
+image_to_text = ImageToText()
+
+# Router for WhatsApp respo
+whatsapp_router = APIRouter()
+
+# WhatsApp API credentials
+WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN")
+WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
+
+
+@whatsapp_router.api_route("/whatsapp_response", methods=["GET", "POST"])
+async def whatsapp_handler(request: Request) -> Response:
+    """Handles incoming messages and status updates from the WhatsApp Cloud API."""
+
+    if request.method == "GET":
+        params = request.query_params
+        if params.get("hub.verify_token") == os.getenv("WHATSAPP_VERIFY_TOKEN"):
+            return Response(content=params.get("hub.challenge"), status_code=200)
+        return Response(content="Verification token mismatch", status_code=403)
+
+    try:
+        data = await request.json()
+        change_value = data["entry"][0]["changes"][0]["value"]
+        if "messages" in change_value:
+            message = change_value["messages"][0]
+            from_number = message["from"]
+            session_id = from_number
+
+            # Get user message (text only for now)
+            if message["type"] != "text":
+                await send_response(from_number, "Sorry, I can only process text messages at the moment.")
+                return Response(content="Non-text message received", status_code=200)
+            
+            content = message["text"]["body"]
+
+            # Process message through the graph agent
+            async with AsyncSqliteSaver.from_conn_string("data/short_term.db") as short_term_memory:
+                graph = await create_agent_graph(short_term_memory)
+                current_state = AgentState(
+                    messages=[HumanMessage(content=content)],
+                    context=[]
+                )
+                
+                # Get the response from the graph
+                output_state = await graph.ainvoke(
+                    current_state,
+                    {"configurable": {"thread_id": session_id}},
+                )
+
+                response_message = output_state["messages"][-1].content
+                success = await send_response(from_number, response_message)
+
+                if not success:
+                    return Response(content="Failed to send message", status_code=500)
+
+            return Response(content="Message processed", status_code=200)
+
+        elif "statuses" in change_value:
+            return Response(content="Status update received", status_code=200)
+
+        else:
+            return Response(content="Unknown event type", status_code=400)
+
+    except Exception as e:
+        logger.error(f"Error processing message: {e}", exc_info=True)
+        return Response(content="Internal server error", status_code=500)
+
+
+async def download_media(media_id: str) -> bytes:
+    """Download media from WhatsApp."""
+    media_metadata_url = f"https://graph.facebook.com/v21.0/{media_id}"
+    headers = {"Authorization": f"Bearer {WHATSAPP_TOKEN}"}
+
+    async with httpx.AsyncClient() as client:
+        metadata_response = await client.get(media_metadata_url, headers=headers)
+        metadata_response.raise_for_status()
+        metadata = metadata_response.json()
+        download_url = metadata.get("url")
+
+        media_response = await client.get(download_url, headers=headers)
+        media_response.raise_for_status()
+        return media_response.content
+
+
+async def process_audio_message(message: Dict) -> str:
+    """Download and transcribe audio message."""
+    audio_id = message["audio"]["id"]
+    media_metadata_url = f"https://graph.facebook.com/v21.0/{audio_id}"
+    headers = {"Authorization": f"Bearer {WHATSAPP_TOKEN}"}
+
+    async with httpx.AsyncClient() as client:
+        metadata_response = await client.get(media_metadata_url, headers=headers)
+        metadata_response.raise_for_status()
+        metadata = metadata_response.json()
+        download_url = metadata.get("url")
+
+    # Download the audio file
+    async with httpx.AsyncClient() as client:
+        audio_response = await client.get(download_url, headers=headers)
+        audio_response.raise_for_status()
+
+    # Prepare for transcription
+    audio_buffer = BytesIO(audio_response.content)
+    audio_buffer.seek(0)
+    audio_data = audio_buffer.read()
+
+    return await speech_to_text.transcribe(audio_data)
+
+
+async def send_response(from_number: str, response_text: str) -> bool:
+    """Send text response to user via WhatsApp API."""
+    headers = {
+        "Authorization": f"Bearer {WHATSAPP_TOKEN}",
+        "Content-Type": "application/json",
+    }
+
+    json_data = {
+        "messaging_product": "whatsapp",
+        "to": from_number,
+        "type": "text",
+        "text": {"body": response_text},
+    }
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            f"https://graph.facebook.com/v21.0/{WHATSAPP_PHONE_NUMBER_ID}/messages",
+            headers=headers,
+            json=json_data,
+        )
+
+    return response.status_code == 200
+
+
+async def upload_media(media_content: BytesIO, mime_type: str) -> str:
+    """Upload media to WhatsApp servers."""
+    headers = {"Authorization": f"Bearer {WHATSAPP_TOKEN}"}
+    files = {"file": ("response.mp3", media_content, mime_type)}
+    data = {"messaging_product": "whatsapp", "type": mime_type}
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            f"https://graph.facebook.com/v21.0/{WHATSAPP_PHONE_NUMBER_ID}/media",
+            headers=headers,
+            files=files,
+            data=data,
+        )
+        result = response.json()
+
+    if "id" not in result:
+        raise Exception("Failed to upload media")
+    return result["id"] 

pyproject.toml

@@ -1,29 +1,31 @@
 [project]
-name = "aie5-deploypythonicrag"
+name = "chief-joy-officer"
 version = "0.1.0"
-description = "Simple Pythonic RAG App"
+description = "AI-powered Chief Joy Officer for social activity recommendations"
 readme = "README.md"
-requires-python = ">=3.11,<3.12"  
+requires-python = ">=3.12"
 dependencies = [
-    "chainlit>=2.2.1",
-    "numpy>=2.2.2",
-    "openai>=1.59.9",
-    "pydantic==2.10.1",
-    "pypdf2>=3.0.1",
-    "websockets>=14.2",
-    "langchain-openai>=0.0.5",
-    "langgraph>=0.0.19",
-    "langchain>=0.1.8",
-    "langchain-core>=0.1.23",
-    "langchain-community>=0.0.19",
-    "langchain-qdrant>=0.2.0",
-    "langgraph-checkpoint-duckdb>=2.0.1",
+    "chainlit>=1.3.2",
+    "fastapi[standard]>=0.115.6",
+    "httpx>=0.27.0",
+    "langchain>=0.3.13",
+    "langchain-community>=0.3.13",
+    "langchain-core>=0.1.27",
+    "langgraph>=0.2.60",
+    "langchain-openai>=0.2.14",
+    "pydantic>=2.10.0",
     "langgraph-checkpoint-sqlite>=2.0.1",
-    "duckdb>=1.1.3",
     "aiosqlite>=0.20.0",
-    "tavily-python>=0.3.1",
-    "typing-extensions>=4.9.0",
-    "beautifulsoup4==4.13.3",
-    "sentence-transformers==3.4.1",
-    "uuid==1.30"
+    "python-dotenv>=1.0.1",
+    "elevenlabs>=1.50.3",
+    "groq>=0.13.1",
+    "langchain-groq>=0.2.2",
+    "pydantic-settings>=2.7.0",
+    "pre-commit>=4.0.1",
+    "supabase>=2.11.0",
+    "langgraph-checkpoint-duckdb>=2.0.1",
+    "duckdb>=1.1.3",
+    "qdrant-client>=1.12.1",
+    "sentence-transformers>=3.3.1",
+    "together>=1.3.10"
 ]