Auto deploy Chat API

Dockerfile

@@ -0,0 +1,25 @@
+# Use an official lightweight Python image
+FROM python:3.11-slim
+
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y build-essential
+
+# Copy requirements and install Python dependencies
+COPY requirements.txt .
+RUN pip install --upgrade pip && pip install -r requirements.txt
+
+# Copy the source code
+COPY . .
+
+# Expose the FastAPI port
+EXPOSE 8000
+
+# Start FastAPI app using uvicorn
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,10 @@
+---
+title: Fourways Chat Api
+emoji: 🌖
+colorFrom: blue
+colorTo: yellow
+sdk: docker
+pinned: false
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

agent.py

@@ -0,0 +1,188 @@
+from pydantic import BaseModel, Field
+from typing import Optional, List, TypedDict
+from langchain_core.tools import tool, Tool
+from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langgraph.graph import END, StateGraph
+from langgraph.prebuilt import ToolNode
+from supabase import create_client
+from neighbourhoods import neighbourhood_lookup
+import os
+
+
+google_api_key = os.getenv("GOOGLE_API_KEY")
+supabase_url = os.getenv("SUPABASE_URL")
+supabase_key = os.getenv("SUPABASE_ANON_KEY")
+supabase = create_client(supabase_url, supabase_key)
+
+neighbourhoods = supabase.table('neighbourhoods').select('id,name').execute()
+
+class PropertySearchInput(BaseModel):
+  maxPrice: Optional[float] = None
+  minPrice: Optional[float] = None
+  minBedrooms: Optional[int] = None
+  maxBedrooms: Optional[int] = None
+  minBathrooms: Optional[int] = None
+  maxBathrooms: Optional[int] = None
+  propertyType: Optional[str] = None
+  neighbourhood: Optional[str] = None
+  status: Optional[str] = None
+
+
+@tool("search_for_properties", args_schema=PropertySearchInput)
+def search_for_properties_tool(maxPrice: int = None, minPrice: int = None,
+                               minBedrooms: int = None, maxBedrooms: int = None,
+                               minBathrooms: int = None, maxBathrooms: int = None,
+                               propertyType: str = None,
+                               neighbourhood: str = None, status: str = None
+                               ) -> List[dict]:
+  """
+  Search for properties using a combination of filters provided by the user.
+
+  This tool queries the 'properties' table in the database and returns all listings
+  that match the specified filters. All filters are optional and can be combined to
+  narrow down the results. It supports searching by location, price range, bedrooms,
+  bathrooms, property type, and status (for sale or rent).
+
+  Parameters (all optional):
+      - maxPrice: Maximum price of the property.
+      - minPrice: Minimum price of the property.
+      - minBedrooms: Minimum number of bedrooms required.
+      - maxBedrooms: Maximum number of bedrooms required.
+      - minBathrooms: Minimum number of bathrooms required.
+      - maxBathrooms: Maximum number of bathrooms required.
+      - propertyType: The type of property (e.g. house, apartment).
+      - neighbourhood: The neighbourhood or area to search in.
+      - status: Whether the property is for sale or for rent.
+
+  Returns:
+      A list of property records (dictionaries) that match the search criteria.
+
+  Usage example:
+      The assistant may call:
+      search_for_properties({
+          "bedrooms": 2,
+          "status": "for_rent",
+          "maxPrice": 800,
+          "neighbourhood": "Kacyiru"
+      })
+
+  Notes:
+      - The assistant should never guess or fabricate property data.
+      - Always use this tool when the user is asking to see or search for properties.
+  """
+
+  query = supabase.table("properties").select("*")
+
+  if neighbourhood:
+    neighbourhood_id = neighbourhood_lookup.get_neighbourhood_id(neighbourhood)
+
+    # If the neighbourhood id does not exist (i.e is None) then the database does not have
+    # properties in that neighbourhood, so we return an empty list
+    if neighbourhood_id is None:
+      return []
+
+    # If id exists, then search database using neighbourhood_id
+    query = query.eq("neighbourhood", neighbourhood_id)
+  if status:
+    query = query.eq("status", status)
+  if propertyType:
+    query = query.eq("property_type", propertyType)
+  if minBedrooms:
+    query = query.gte("bedrooms", minBedrooms)
+  if maxBedrooms:
+    query = query.gte("bedrooms", maxBedrooms)
+  if minBathrooms:
+    query = query.gte("bathrooms", minBathrooms)
+  if maxBathrooms:
+    query = query.gte("bathrooms", maxBathrooms)
+  if minPrice:
+    query = query.gte("price", minPrice)
+  if maxPrice:
+    query = query.lte("price", maxPrice)
+
+  try:
+    response = query.execute()
+  except e:
+    print(f"Error executing query: {e}")
+    return []
+
+  return response.data
+
+tools = [search_for_properties_tool]
+
+
+llm = ChatGoogleGenerativeAI(
+  model="gemini-2.5-flash-lite-preview-06-17",
+  temperature=0,
+  google_api_key=google_api_key
+)
+llm = llm.bind_tools(tools)
+
+    
+class AgentState(TypedDict):
+  messages: List[BaseMessage]
+
+
+def llm_node(state: AgentState) -> AgentState:
+  """Invokes the LLM to reason and decide on tool use."""
+  response = llm.invoke(state['messages'])
+
+  tool_calls = getattr(response, "tool_calls", None)
+
+  # If tool call was made, store it in AIMessage
+  if tool_calls:
+    # Always provide an empty string for content when there are tool calls
+    state['messages'].append(
+        AIMessage(content="", tool_calls=tool_calls)
+    )
+  else:
+    content = response.content or "I'm sorry, I wasn't able to generate a response."
+    state['messages'].append(AIMessage(content=content.strip()))
+
+  return state
+
+
+def should_continue(state: AgentState) -> str:
+  last_message = state['messages'][-1]
+  # Check for tool_calls attribute and if the list is not empty
+  if hasattr(last_message, "tool_calls") and last_message.tool_calls:
+      return "continue"
+  return "end"
+
+
+def tool_node(state: AgentState) -> AgentState:
+  # Get the most recent AIMessage that called the tool
+  last_message = state["messages"][-1]
+  tool_call = last_message.tool_calls[0]  # Only handling one tool call here for simplicity
+
+  tool_name = tool_call["name"]
+  tool_args = tool_call["args"]
+  tool_call_id = tool_call["id"]
+
+  # Find and execute the tool
+  tool: Tool = next(t for t in tools if t.name == tool_name)
+  tool_result = tool.invoke(tool_args)
+
+  # Add the result to the messages as a ToolMessage
+  state["messages"].append(
+      ToolMessage(content=str(tool_result), tool_call_id=tool_call_id)
+  )
+
+  return state
+
+
+graph = StateGraph(AgentState)
+graph.add_node("llm_node", llm_node)
+graph.add_node("tool_node", tool_node)
+graph.set_entry_point("llm_node")
+graph.add_conditional_edges(
+  "llm_node",
+  should_continue,
+  {
+      "continue": "tool_node",
+      "end": END
+  }
+)
+graph.add_edge("tool_node", "llm_node")
+agent = graph.compile()

main.py

@@ -0,0 +1,107 @@
+from fastapi import FastAPI, HTTPException, Header
+from pydantic import BaseModel
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
+from typing import Optional, Literal
+from agent import agent
+from supabase_client import client
+from neighbourhoods import neighbourhood_lookup
+from json import load
+import os
+
+# Load service URLs
+with open('services.json') as f:
+    services = load(f)
+
+SUPABASE_URL = services.get('SUPABASE', '')
+SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY")
+
+# Load system prompt content
+with open('system_prompt.txt') as f:
+    system_prompt_content = f.read()
+
+system_prompt = SystemMessage(content=system_prompt_content)
+
+app = FastAPI()
+
+
+class MessageRequest(BaseModel):
+    message: str
+    chat_id: str
+
+
+class WebhookRequest(BaseModel):
+    type: Literal["INSERT", "UPDATE", "DELETE"]
+    table: str
+    schema: str
+    record: dict
+    old_record: Optional[dict] = None
+
+
+@app.post("/chat")
+async def chat(request: MessageRequest, authorization: str = Header(...)):
+    access_token = authorization.replace("Bearer ", "")
+    user_response = client.auth.get_user(access_token)
+    user = user_response.user if user_response else None
+
+    if not user:
+        raise HTTPException(status_code=401, detail="Invalid or expired token.")
+
+    # Insert user message into chat_messages
+    client.table("chat_messages").insert({
+        "chat_session_id": request.chat_id,
+        "message": request.message,
+        "sent_by": user.id
+    }).execute()
+
+    # Retrieve chat session details
+    chat_session = client.table("chat_sessions").select("*").eq("id", request.chat_id).execute()
+    if not chat_session.data:
+        raise HTTPException(status_code=404, detail="Chat session not found")
+
+    is_ai = not chat_session.data[0].get("user_two")
+    if not is_ai:
+        return { "success": True, "message": "Message added to chat." }
+
+    # Fetch full chat history
+    messages = client.table("chat_messages")\
+        .select("*")\
+        .eq("chat_session_id", request.chat_id)\
+        .order("created_at", desc=False)\
+        .execute()
+
+    # Construct message history
+    history = [system_prompt]
+    for msg in messages.data:
+        if msg["sent_by"] == user.id:
+            history.append(HumanMessage(content=msg["message"]))
+        else:
+            history.append(AIMessage(content=msg["message"]))
+
+    # Generate AI response
+    state = { "messages": history }
+    result = agent.invoke(state)
+    final_message = result["messages"][-1].content
+
+    # Insert AI response into chat_messages
+    client.table("chat_messages").insert({
+        "chat_session_id": request.chat_id,
+        "message": final_message,
+        "sent_by": None
+    }).execute()
+
+    return { "success": True, "ai_response": final_message }
+
+
+@app.post("/on-neighbourhoods-change")
+def on_neighbourhoods_change(request: WebhookRequest):
+    id = request.record.get("id")
+    name = request.record.get("name")
+
+    if request.type == "INSERT":
+        neighbourhood_lookup.add_neighbourhood(name, id)
+    elif request.type == "UPDATE":
+        neighbourhood_lookup.update_neighbourhood(name, id)
+    elif request.type == "DELETE":
+        neighbourhood_lookup.delete_neighbourhood(id)
+
+    return { "success": True }

neighbourhoods.py

@@ -0,0 +1,28 @@
+from supabase_client import client
+
+
+class Neighbourhoods:
+  def __init__(self):
+    self.neighbourhoods = {}
+    neighbourhoods = client.table('neighbourhoods').select('id,name').execute().data
+
+    for neighbourhood in neighbourhoods:
+      id = neighbourhood['id']
+      name = neighbourhood['name']
+      self.add_neighourhood(name, id)
+  
+  def delete_neighbourhood(self, id):
+    self.neighbourhoods = { k: v for k, v in self.neighbourhoods if v != id }
+  
+  def update_neighourhood(self, name, id):
+    self.delete_neighbourhood(id)
+    self.add_neighourhood(name, id)
+  
+  def add_neighourhood(self, name, id):
+    self.neighbourhoods[name.lower()] = id
+  
+  def get_neighbourhood_id(self, name):
+    return self.neighbourhoods.get(name.lower())
+
+
+neighbourhood_lookup = Neighbourhoods()

requirements.txt

@@ -0,0 +1,11 @@
+fastapi
+uvicorn
+pydantic
+langgraph
+langchain
+langchain-core
+langchain-google-genai
+google-generativeai
+supabase
+python-dotenv
+pytest

services.json

@@ -0,0 +1,7 @@
+{
+  "EMBEDDINGS": "https://akafesu-fourwalls-embeddings-api.hf.space",
+  "RECOMMENDATIONS": "https://akafesu-fourwalls-recommendations-api.hf.space",
+  "CHAT": "https://akafesu-fourwalls-chat-api.hf.space",
+  "SUPABASE": "https://nhwwhpbxlqsdbirdrxmc.supabase.co",
+  "MIGRATIONS": "https://akafesu-fourwalls-migrations-api.hf.space"
+}

supabase_client.py

@@ -0,0 +1,11 @@
+from supabase import create_client
+from os import getenv
+from json import load
+
+with open('./services.json') as f:
+  services = load(f)
+
+SUPABASE_URL = services['SUPABASE']
+SUPABASE_ANON_KEY = getenv('SUPABASE_ANON_KEY')
+
+client = create_client(SUPABASE_URL, SUPABASE_ANON_KEY)

system_prompt.txt

@@ -0,0 +1,28 @@
+You are Scout, an AI assistant who helps users find houses on our real estate platform.
+Your job is to understand the user's needs and search for matching properties using the available filters.
+These filters include location, price, number of bedrooms, property type, and status (e.g. for sale or for rent).
+You should always use the available tools to perform searches, rather than making up information.
+If a user asks about what properties are available, or describes what they’re looking for, call the search tool with the appropriate filters.
+The search tool allows you to search for properties that are both for sale and for rent.
+When presenting search results, it is important that you embed the property ID in your responses using this format: <%propertyId%>.
+For example, if you’re referencing a house with ID `abc123`, embed it as <%abc123%>.
+The propertyId is contained in the id field of the property record.
+This format helps link the properties with our frontend display system.
+This helps the user view more details on the property, so it is vital that you embed property ids.
+When embedding properties, you should provide a short description of the property.
+The description of the property should be in natural language.
+Keep the description concise, informative and natural.
+You can embed multiple properties in the same message.
+You can use markdown to format messages in a more readable way.
+But, do not overuse markdown as the messages should feel like a natural conversation, not a blog post.
+Keep your responses friendly, helpful, and concise.
+If you are responding with search results, explain the filters you applied to the user.
+Do not answer questions unless you have used the tool to retrieve relevant results.
+If you cannot find any matching properties, kindly let the user know and suggest adjusting the search criteria.
+Always behave like a knowledgeable assistant focused on helping users find their ideal home.
+If a user is asking for a particular property type (e.g. apartment), and you cannot find any listings of that type in the database, broaden the search
+to include other property types but let them know in the final message that you had to adjust their query, and how you adjusted their query.
+Sometimes, house can be taken to refer to all property types.
+The database currency is RWF. If the user quotes prices in USD ask them to restate in RWF.
+If you cannot find properties in the user's budget, suggest that they should adjust their budget.
+If you cannot find properties within a particular neighbourhood, broaden your search to all neighbourhoods.

tests/test_agent.py

@@ -0,0 +1,32 @@
+import pytest
+from chat.agent import PropertySearchInput, search_for_properties_tool, AgentState, llm_node, should_continue, tool_node
+
+# Test PropertySearchInput model
+def test_property_search_input():
+    data = {'maxPrice': 1000, 'minPrice': 500, 'location': 'Harare'}
+    obj = PropertySearchInput(**data)
+    assert obj.maxPrice == 1000
+    assert obj.minPrice == 500
+    assert obj.location == 'Harare'
+
+# Test search_for_properties_tool function
+def test_search_for_properties_tool():
+    result = search_for_properties_tool(maxPrice=2000, minPrice=1000, location='Bulawayo')
+    assert isinstance(result, list) or result is None
+
+# Test AgentState and llm_node, should_continue, tool_node
+@pytest.fixture
+def agent_state():
+    return {'step': 1, 'history': []}
+
+def test_llm_node(agent_state):
+    result = llm_node(agent_state)
+    assert isinstance(result, dict)
+
+def test_should_continue(agent_state):
+    result = should_continue(agent_state)
+    assert isinstance(result, str)
+
+def test_tool_node(agent_state):
+    result = tool_node(agent_state)
+    assert isinstance(result, dict)

tests/test_main.py

@@ -0,0 +1,28 @@
+from chat.main import MessageRequest, WebhookRequest, chat, on_neighbourhoods_change
+import pytest
+from fastapi import Header
+import asyncio
+
+# Test MessageRequest and WebhookRequest models
+def test_message_request():
+    req = MessageRequest(message="Hello", session_id="abc123")
+    assert req.message == "Hello"
+    assert req.session_id == "abc123"
+
+def test_webhook_request():
+    req = WebhookRequest(event="update", data={"id": 1})
+    assert req.event == "update"
+    assert req.data == {"id": 1}
+
+# Test chat endpoint (sync call for test)
+@pytest.mark.asyncio
+def test_chat():
+    req = MessageRequest(message="Test", session_id="test1")
+    result = asyncio.run(chat(req, authorization="Bearer test"))
+    assert result is not None
+
+# Test on_neighbourhoods_change
+def test_on_neighbourhoods_change():
+    req = WebhookRequest(event="delete", data={"id": 2})
+    result = on_neighbourhoods_change(req)
+    assert result is not None

tests/test_neighbourhoods.py

@@ -0,0 +1,20 @@
+from chat.neighbourhoods import Neighbourhoods
+import pytest
+
+@pytest.fixture
+def neighbourhoods():
+    return Neighbourhoods()
+
+def test_add_and_get_neighbourhood(neighbourhoods):
+    neighbourhoods.add_neighourhood('Avondale', 1)
+    assert neighbourhoods.get_neighbourhood_id('Avondale') == 1
+
+def test_update_neighbourhood(neighbourhoods):
+    neighbourhoods.add_neighourhood('Borrowdale', 2)
+    neighbourhoods.update_neighourhood('Borrowdale Updated', 2)
+    # Depending on implementation, check if update is reflected
+
+def test_delete_neighbourhood(neighbourhoods):
+    neighbourhoods.add_neighourhood('Greendale', 3)
+    neighbourhoods.delete_neighbourhood(3)
+    # Depending on implementation, check if deletion is reflected