file added

Dockerfile.client

@@ -0,0 +1,14 @@
+FROM python:3.12.3-slim
+
+WORKDIR /app
+
+COPY requirements_client.txt ./
+RUN pip install --no-cache-dir -r requirements_client.txt
+
+COPY . .
+
+EXPOSE 8088
+
+HEALTHCHECK CMD curl --fail http://localhost:8088/_stcore/health
+
+ENTRYPOINT ["streamlit", "run", "src/interface.py", "--server.port=8088", "--server.address=0.0.0.0"]

Dockerfile.server

@@ -0,0 +1,17 @@
+FROM python:3.12.3-slim
+
+# Set environment variables
+ENV PYTHONUNBUFFERED 1
+WORKDIR /app
+
+COPY requirements_server.txt ./
+RUN pip install --no-cache-dir -r requirements_server.txt
+
+# Copy the application code
+COPY . .
+
+# Expose the port
+EXPOSE 8080
+
+# Start the backend app
+CMD ["python", "src/main.py"]

README.md

@@ -1,11 +1,111 @@
+# DAPA (Digital Arrest Protection App)
+
+**Inspired by the Hindi word 'धप्पा' (meaning ‘busted’), A tool to combat digital scams.**
+
+---
+
+## Overview
+DAPA enables users to report financial scams and frauds efficiently via a user-friendly platform. The system facilitates the following:
+
+1. **Report a Scam**: Users can provide details such as the scammer’s phone number and a brief description of the fraud. DAPA categorizes the scam and records the report.
+2. **Search a Phone Number**: Users can check whether a specific number has been previously flagged for fraudulent activities.
+
+With its streamlined architecture powered by AI, DAPA aims to mitigate the growing epidemic of financial scams by creating a comprehensive record of scam incidents.
+
+---
+
+## Why DAPA Matters
+### The Growing Threat of Financial Scams:
+- **India**:
+  - According to reports, the annual financial loss due to digital scams and frauds in India amounts to **₹60,000 crore**.
+  - A surge in digital adoption has inadvertently made India a hotspot for cybercriminal activities, including the **Digital Arrest Scam** where scammers impersonate law enforcement officials to defraud victims of significant amounts.
+
+- **Global**:
+  - Financial scams cost individuals and businesses **over $5 trillion annually worldwide**.
+  - An increasing number of scams now exploit vulnerabilities in mobile and social media platforms.
+
+### Spotlight on the Digital Arrest Scam:
+The Digital Arrest Scam, a notable threat in India, sees fraudsters pretending to be law enforcement officials to extort money by intimidation. Victims are coerced into paying under the pretense of avoiding legal trouble. (Sources: **The Hindu**)
+
+Prime Minister Modi have highlighted the urgent need to address these issues through education and vigilance.
+
+---
+
+## Why an AI Chatbot Instead of Traditional Form-Based Tools?
+DAPA leverages an AI chatbot over traditional form-based tools for the following reasons:
+
+1. **Localized Language Support**:
+   - Users can narrate their experiences in their own local language, including dialects, through platforms like WhatsApp. The AI chatbot interprets, summarizes, and processes this information seamlessly.
+
+2. **Ease of Use**:
+   - Unlike rigid forms, a conversational AI provides a more intuitive, human-like interface, enabling users to express their ordeal naturally without worrying about form structure or specific formats.
+
+3. **Advanced Understanding**:
+   - The AI model not only captures and categorizes scam details but also creates concise summaries, ensuring that valuable information is accurately stored in a centralized database for combating scams.
+
+4. **Centralized Scam Pattern Analysis**:
+   - The summaries generated by the chatbot help uncover common scam patterns. By analyzing these trends, DAPA contributes to developing preventive measures and combating scams more effectively at scale.
+
+5. **Accessibility**:
+   - By integrating with popular platforms like WhatsApp, the AI chatbot increases accessibility, making it easier for users in remote or underserved areas to report scams.
+
+6. **Insights**:
+   - With AI, identifying patterns and emerging threats faster than traditional tools, thereby enhancing preventative measures.
+
 ---
-title: Dapa
-emoji: 🐢
-colorFrom: indigo
-colorTo: indigo
-sdk: docker
-pinned: false
-license: mit
+
+## Features
+### Core Features:
+- **Fraud Reporting**:
+  - Easy submission of scammer details.
+  - Categorization of the scam using AI-based inference.
+  - Centralized record maintenance for tracking scam patterns.
+
+- **Fraud Search**:
+  - Quick lookup of phone numbers to check for prior scam reports.
+
+### Technology Stack:
+- **Backend**: Python, FastAPI, LangGraph
+- **Frontend**: Streamlit
+- **Database**: PostgreSQL
+- **AI Model**: GROQ LLaMA 3 (Configurable)
+
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+## Setup Instructions
+### Environment Configuration
+Create the environment file:
+```bash
+cp example.env .env
+```
+- Add Groq token
+
+### Build and Run the Application
+Build the application container using Docker:
+```bash
+docker compose up --build -d
+```
+
+### Access the App:
+- Chatbot: [http://localhost:8088/](http://localhost:8088/)
+- API Documentation: [http://localhost:8080/docs](http://localhost:8080/docs)
+
+---
+
+## Architecture Diagram
+![flow](https://github.com/user-attachments/assets/c51bd311-7b9b-4e9c-888b-190fc08e4da0)
+
+---
+
+## Future Roadmap
+- **WhatsApp Integration**: Users will be able to report and identify scams via WhatsApp for enhanced convenience. This will be achieved through a webhook system.
+- **Enhanced AI Capabilities**: Continuous improvement of the scam categorization model to ensure accuracy and adaptability to emerging scam types.
+---
+
+## Contact
+For improvements or collaboration, feel free to connect:  
+[Raushan's LinkedIn Profile](https://www.linkedin.com/in/raushan-in/)
+
+Together, let’s combat financial fraud and make the digital world safer for everyone. 
+
+**DAPA – Digital Arrest Protection App**

compose.yml

@@ -0,0 +1,65 @@
+services:
+  dapa_be:
+    build:
+      context: .
+      dockerfile: Dockerfile.server
+    container_name: dapa_backend
+    depends_on:
+        - dapa_pg
+    ports:
+      - "8080:8080"
+    env_file:
+      - .env
+    environment:
+      - DATABASE_URL=postgresql+asyncpg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@dapa_pg:${POSTGRES_PORT}/${POSTGRES_DB}
+    networks:
+      - dapa-network
+
+
+  dapa_streamlit_fe:
+      build:
+        context: .
+        dockerfile: Dockerfile.client
+      container_name: dapa_streamlit
+      ports:
+        - "8088:8088"
+      depends_on:
+        - dapa_be
+      environment:
+        - BACKEND_URL=http://dapa_backend:8080
+      networks:
+        - dapa-network
+      develop:
+        watch:
+          - path: src/client/
+            action: sync+restart
+            target: /app/client/
+          - path: src/schema/
+            action: sync+restart
+            target: /app/schema/
+          - path: src/interface.py
+            action: sync+restart
+            target: /app/interface.py
+  
+  dapa_pg:
+    image: postgres:latest
+    container_name: dapa_pg
+    ports:
+      - "5432:5432"
+    env_file:
+      - .env
+    networks:
+      - dapa-network
+    volumes:
+      - pg_db_1:/var/lib/postgresql/data
+
+volumes:
+  pg_db_1:
+
+networks:
+  dapa-network:
+
+# To build and run the app:
+# docker compose up --build -d
+# or for dev easy
+# docker compose watch

example.env

@@ -0,0 +1,13 @@
+MODE=dev
+
+GROQ_API_KEY=
+GROQ_MODEL=llama-3.3-70b-versatile
+
+LANGCHAIN_TRACING_V2=false
+LANGCHAIN_API_KEY=
+
+# POSTGRES_DB
+POSTGRES_USER=username
+POSTGRES_PASSWORD=password
+POSTGRES_DB=pgdb_name
+POSTGRES_PORT=5432

requirements_client.txt

@@ -0,0 +1,4 @@
+httpx
+pydantic
+python-dotenv
+streamlit

requirements_server.txt

@@ -0,0 +1,45 @@
+annotated-types==0.7.0
+anyio==4.8.0
+asyncpg==0.30.0
+certifi==2024.12.14
+charset-normalizer==3.4.1
+click==8.1.8
+distro==1.9.0
+fastapi==0.115.6
+greenlet==3.1.1
+groq==0.15.0
+h11==0.14.0
+httpcore==1.0.7
+httpx==0.28.1
+idna==3.10
+jsonpatch==1.33
+jsonpointer==3.0.0
+langchain-core==0.3.29
+langchain-groq==0.2.3
+langgraph==0.2.62
+langgraph-checkpoint==2.0.9
+langgraph-sdk==0.1.51
+langsmith==0.2.10
+msgpack==1.1.0
+numexpr==2.10.2
+numpy==2.2.1
+orjson==3.10.14
+packaging==24.2
+psycopg2-binary==2.9.10
+pydantic==2.10.5
+pydantic-settings==2.7.1
+pydantic_core==2.27.2
+python-dotenv==1.0.1
+PyYAML==6.0.2
+requests==2.32.3
+requests-toolbelt==1.0.0
+setuptools==69.5.1
+sniffio==1.3.1
+SQLAlchemy==2.0.37
+sqlmodel==0.0.22
+starlette==0.41.3
+tenacity==9.0.0
+typing_extensions==4.12.2
+urllib3==2.3.0
+uvicorn==0.34.0
+wheel==0.43.0

src/agent.py

@@ -0,0 +1,61 @@
+"""AI agents"""
+
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import AIMessage, SystemMessage
+from langchain_core.runnables import (
+    RunnableConfig,
+    RunnableLambda,
+    RunnableSerializable,
+)
+from langchain_groq import ChatGroq
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import END, MessagesState, StateGraph
+from langgraph.prebuilt import ToolNode
+
+from prompts import instructions
+from settings import settings
+from tools import register_scam, search_scam
+
+llm = ChatGroq(
+    model=settings.GROQ_MODEL, temperature=settings.GROQ_MODEL_TEMP, streaming=False
+)
+
+
+tools = [register_scam, search_scam]
+
+
+class AgentState(MessagesState, total=False):
+    """`total=False` is PEP589 specs.
+
+    documentation: https://typing.readthedocs.io/en/latest/spec/typeddict.html#totality
+    """
+
+
+def wrap_model(model: BaseChatModel) -> RunnableSerializable[AgentState, AIMessage]:
+    model_with_tools = model.bind_tools(tools)
+    preprocessor = RunnableLambda(
+        lambda state: [SystemMessage(content=instructions)] + state["messages"],
+        name="StateModifier",
+    )
+    return preprocessor | model_with_tools
+
+
+async def acall_model(state: AgentState, config: RunnableConfig) -> AgentState:
+    model_runnable = wrap_model(llm)
+    response = await model_runnable.ainvoke(state, config)
+    # We return a list, because this will get added to the existing list
+    return {"messages": [response]}
+
+
+# Define the graph
+agent = StateGraph(AgentState)
+agent.add_node("model", acall_model)
+agent.add_node("tools", ToolNode(tools))
+
+agent.set_entry_point("model")
+
+# Add edges (transitions)
+agent.add_edge("model", "tools")
+agent.add_edge("tools", END)
+
+cyber_guard = agent.compile(checkpointer=MemorySaver())

src/auth.py

@@ -0,0 +1,23 @@
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from settings import settings
+
+
+def verify_bearer(
+    http_auth: Annotated[
+        HTTPAuthorizationCredentials | None,
+        Depends(
+            HTTPBearer(
+                description="Please provide AUTH_SECRET api key.", auto_error=False
+            )
+        ),
+    ],
+) -> None:
+    if not settings.AUTH_SECRET:
+        return
+    auth_secret = settings.AUTH_SECRET.get_secret_value()
+    if not http_auth or http_auth.credentials != auth_secret:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

src/database.py

@@ -0,0 +1,67 @@
+import re
+from contextlib import asynccontextmanager
+from datetime import datetime
+
+from fastapi import Depends
+from pydantic import validator
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.orm import sessionmaker
+from sqlmodel import Field, SQLModel
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from scams import scam_categories
+from settings import settings
+
+database_url = settings.DATABASE_URL.get_secret_value()
+
+engine = create_async_engine(database_url, echo=settings.is_dev())
+
+async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+
+async def create_db_and_tables():
+    async with engine.begin() as conn:
+        await conn.run_sync(SQLModel.metadata.create_all)
+
+
+@asynccontextmanager
+async def get_session() -> AsyncSession:
+    """
+    Dependency function to provide an async database session.
+    Ensures proper cleanup after use.
+    """
+    async with async_session() as session:
+        try:
+            yield session
+        finally:
+            await session.close()
+
+
+class Scammer(SQLModel, table=True):
+    """Scammer ORM Model."""
+
+    id: int = Field(default=None, primary_key=True)
+    scammer_mobile: str = Field(index=True, description="Scammer mobile number")
+    scam_id: int = Field(description="Scam ID of the scam type")
+    reporter_ordeal: str = Field(description="Summary of the scam")
+    reporter_mobile: str = Field(description="Reporter mobile number")
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow, description="Timestamp of report creation"
+    )
+
+    @validator("scammer_mobile", "reporter_mobile", pre=True)
+    def validate_mobile_number(cls, value: str) -> str:
+        """Validate mobile numbers using a regex."""
+        pattern = r"^\+\d{1,3}-?\d{6,14}$"  # E.164 format
+        if not re.match(pattern, value):
+            raise ValueError(f"Invalid mobile number: {value}")
+        return value
+
+    @validator("scam_id")
+    def validate_scam_id(cls, value: int) -> int:
+        """Validate if scam_id exists in scam_categories."""
+        if value not in scam_categories.keys():
+            raise ValueError(
+                f"Invalid scam_id: {value}. Must be one of {list(scam_categories.keys())}."
+            )
+        return value

src/interface.py

@@ -0,0 +1,88 @@
+import asyncio
+import os
+
+import httpx
+import streamlit as st
+from dotenv import load_dotenv
+
+APP_TITLE = "DAPA"
+APP_ICON = "🛡️"
+
+# Load environment variables
+load_dotenv()
+BACKEND_URL = os.getenv("BACKEND_URL")
+CHAT_API = BACKEND_URL + "/chat"
+
+
+async def get_response(user_message: str, thread_id: str | None = None) -> dict:
+    payload = {"user_message": user_message, "thread_id": thread_id}
+    async with httpx.AsyncClient(timeout=10.0) as client:
+        try:
+            response = await client.post(CHAT_API, json=payload)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPError as e:
+            return {"error": f"Error connecting to backend: {str(e)}"}
+
+
+async def main():
+    st.set_page_config(page_title=APP_TITLE, page_icon=APP_ICON)
+
+    if "thread_id" not in st.session_state:
+        st.session_state.thread_id = None
+    if "messages" not in st.session_state:
+        st.session_state.messages = []
+
+    st.title(f"{APP_ICON} DAPA AI Assistant")
+    st.write(
+        "This bot helps you identify or report phone numbers involved in financial fraud or cyber scams."
+        "Please describe your incident below."
+    )
+
+    for message in st.session_state.messages:
+        responder_type = message["responder"]
+        if responder_type == "tool":
+            st.chat_message("tool", avatar="🛡️").write(message["content"])
+        else:
+            st.chat_message(responder_type).write(message["content"])
+
+    # User input
+    if user_input := st.chat_input("Type your message here..."):
+        st.session_state.messages.append({"responder": "human", "content": user_input})
+        st.chat_message("human").write(user_input)
+
+        response = await get_response(user_input, st.session_state.thread_id)
+
+        if "error" in response:
+            st.error(response["error"])
+        else:
+            response_message = response["response_message"]
+            responder = response["responder"]
+            st.session_state.thread_id = response["thread_id"]
+
+            # Append the response to session state
+            st.session_state.messages.append(
+                {"responder": responder, "content": response_message}
+            )
+            if responder == "tool":
+                st.chat_message("tool", avatar="🛡️").write(response_message)
+            else:
+                st.chat_message(responder).write(response_message)
+
+    with st.sidebar:
+        st.header(f"{APP_ICON} {APP_TITLE}")
+        st.write("DAPA chatbot for secure reporting of scams.")
+
+        # Privacy Section
+        with st.expander("🔒 Privacy"):
+            st.write(
+                "Query and response in this app are anonymously recorded and saved to LangSmith for product evaluation and improvement purposes."
+            )
+
+        st.markdown(
+            "Made with ❤️ by [Raushan](https://www.linkedin.com/in/raushan-in/) in Trier"
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/main.py

@@ -0,0 +1,23 @@
+import uvicorn
+from fastapi import FastAPI
+
+from database import create_db_and_tables
+from routes import bot_router
+from settings import settings
+
+
+async def lifespan(app):
+    await create_db_and_tables()
+    yield
+
+
+app = FastAPI(title="DAPA", summary="Digital Arrest Protection App", lifespan=lifespan)
+
+# Endpoint router
+app.include_router(bot_router)
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "main:app", host=settings.HOST, port=settings.PORT, reload=settings.is_dev()
+    )

src/prompts.py

@@ -0,0 +1,56 @@
+from scams import scam_categories_str
+
+instructions = f"""
+    You are an AI bot named DAPA. Your job is to assist users in reporting potential digital financial scams or fraud via mobile communication. 
+    DAPA stands for Digital Arrest Protection App.
+
+    # Follow these instructions strictly:
+    - Concise Responses: Keep your replies clear and short.
+    - Only register a scam if the description indicates financial fraud or money involved; otherwise, guide the user to report the issue to appropriate authorities.
+    - Language Adaptability: Respond in the user’s preferred language but use English for tool inputs.
+    - Validate Inputs: Collect all required details from the user before using any tool.
+    - Scammer’s Mobile Number: Must be in +XX-<mobile_number> format.
+    - Scam Type: Identify Scam Type from reporter’s ordeal. Show the scam name (e.g., "Fake Job Scam") to the user, but pass the corresponding ID (e.g., 9) to the tool.
+    - Display Scam Name only instead of Scam ID for human user understanding.
+    - Reporter’s Mobile Number: Must also be in +XX-<mobile_number> format.
+    - Only respond to cases involving cyber scams that are financial in nature and connected to a mobile number.
+    - Avoid assisting with unrelated queries (e.g., general protection tips, general knowledge, mathematical, language or programming questions).
+    - Confirm Before Registering: Always confirm the scammer’s mobile number before registering. Register only if the user explicitly agrees.
+    - If the user provide 0 as the country code for the scammer's mobile number, even after explicitly being asked, use the reporter's country code as a fallback.
+    - Prioritize Scammer Search When Only Mobile Number is Provided.
+    - Before searching for a scammer's mobile number, format it into the standard format with country code (+XX-<mobile_number>).
+    - Keep responses concise and ask for one piece of information at a time to avoid overwhelming the user.
+    - Validate that all required details (scammer's number, description, and reporter's number) are provided before attempting to register the report.
+    - If the country code is not provided, do not make assumptions. Always ask the user to provide.
+    - In case of a ValueError when using the tool, Correct the parameters or missing value and try again.
+
+    Tool Usage: Use tools only after collecting and validating all inputs.
+    Pass scam ID to the tool but show scam name to the user for clarity.
+    Use the Register Scam tool only after explicit user confirmation.
+    Use the Search Scam tool only if the user requests to check a specific number.
+
+    # Predefined Scam Categories: Only use the following scam types:
+
+    {scam_categories_str}
+
+    # Example Scenarios:
+
+    1. Reporting a Scam
+    User: Hi.
+    DAPA: Hi there! 😊 I’m here to assist you.
+
+        I can help you in two ways:
+
+        1. **Report a Scam:** Please provide the following details:  
+        - Scammer’s mobile number (with country code).  
+        - A brief description of your experience (up to 50 words).  
+        - Your mobile number.  
+
+        2. **Identify a Suspicious Number:**  
+        Provide the mobile number and type "search". I’ll check if the number has been reported before.
+
+    2. What is Digital Arrest?
+    DAPA: A digital arrest scam is an online scam that defrauds victims of their hard-earned money.
+          The scammers intimidate the victims and falsely accuse them of illegal activities.
+          They later demand money and puts them under pressure for making the payment.
+    """

src/routes.py

@@ -0,0 +1,32 @@
+"""
+Endpoints defination
+"""
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from auth import verify_bearer
+from schema import SingleResponse, UserInput
+from utils import get_llm_response, infer_chat_message
+
+bot_router = APIRouter(tags=["bot"], dependencies=[Depends(verify_bearer)])
+
+
+@bot_router.post("/chat")
+async def invoke(user_input: UserInput) -> SingleResponse:
+    """
+    Invoke an agent with user input to retrieve a final response.
+
+    Use thread_id to persist and continue a multi-turn conversation.
+    """
+    try:
+        response, thread_id = await get_llm_response(user_input)
+        last_message = infer_chat_message(response["messages"][-1])
+        response = {
+            "response_message": last_message.content,
+            "responder": last_message.type,
+            "thread_id": thread_id,
+        }
+        return SingleResponse(**response)
+    except Exception as e:
+        print(e)
+        raise HTTPException(status_code=500, detail="Unexpected error")

src/scams.py

@@ -0,0 +1,51 @@
+scam_categories = {
+    1: [
+        "Fake Authority Call",
+        "Scammers impersonating law enforcement officials (e.g., CBI, customs, police) or service agents, coercing victims into making payments or money.",
+    ],
+    2: [
+        "Service Disconnection Scam",
+        "Threats of service disconnection unless immediate verification or payment is made.",
+    ],
+    3: [
+        "UPI Scam",
+        "Scammers claim accidental UPI payments and request refunds, or attempt scams related to UPI, PhonePe, Google Pay, or any quick payment interface.",
+    ],
+    4: ["OTP Scam", "Scammers request OTPs to gain unauthorized access to accounts."],
+    5: [
+        "Fake Buyer/Seller Scam",
+        "Scammers pose as buyers requesting refunds or as fraudulent sellers asking for advance payments.",
+    ],
+    6: [
+        "Phishing or Link Scam",
+        "Fraudulent SMS or calls designed to gain unauthorized access to banking platforms by sending malicious links or asking for sensitive details.",
+    ],
+    7: [
+        "Video Call Scam",
+        "Blackmail involving compromising video calls, or screenshots, used to demand money.",
+    ],
+    8: [
+        "Fake Bank Staff Scam",
+        "Calls from scammers posing as bank officials, requesting sensitive banking details.",
+    ],
+    9: [
+        "Fake Job Scam",
+        "Scammers posing as recruiters, demanding service or registration fees for fake job offers.",
+    ],
+    10: [
+        "Lottery Scam",
+        "Messages claiming lottery wins, lucky draws, or prizes, and requesting fees for processing.",
+    ],
+    11: [
+        "Fake Identity Scam",
+        "Scammers imitate known individuals and request money transfers.",
+    ],
+    12: [
+        "Other Cyber Scam",
+        "Any other scam conducted via phone that involves monetary fraud.",
+    ],
+}
+
+scam_categories_str = "\n".join(
+    [f"{k}: {v[0]}-{v[1]}" for k, v in scam_categories.items()]
+)

src/schema.py

@@ -0,0 +1,83 @@
+from typing import Any, Literal, NotRequired
+
+from pydantic import BaseModel, Field
+from typing_extensions import TypedDict
+
+
+class UserInput(BaseModel):
+    """Basic user input for the agent."""
+
+    user_message: str = Field(
+        description="User message to the AI agent.",
+        examples=["Hello, I want to report."],
+    )
+    thread_id: str | None = Field(
+        description="Thread ID to persist and continue a multi-turn conversation.",
+        default=None,
+        examples=["847c6285-8fc9-4560-a83f-4e628xx09254"],
+    )
+
+
+class SingleResponse(BaseModel):
+    """Basic user input for the agent."""
+
+    response_message: str = Field(
+        description="Response content based on user input.",
+        examples=["Hello, I am a bot."],
+    )
+    responder: Literal["human", "ai", "tool", "custom"] = Field(
+        description="Generator of response.",
+        examples=["human", "ai", "tool", "custom"],
+    )
+    thread_id: str | None = Field(
+        description="Thread ID to persist and continue a multi-turn conversation.",
+        default=None,
+        examples=["847c6285-8fc9-4560-a83f-4e628xx09254"],
+    )
+
+
+class ToolCall(TypedDict):
+    """Represents a request to call a tool."""
+
+    name: str
+    """The name of the tool to be called."""
+    args: dict[str, Any]
+    """The arguments to the tool call."""
+    id: str | None
+    """An identifier associated with the tool call."""
+    type: NotRequired[Literal["tool_call"]]
+
+
+class Chat(BaseModel):
+    """Message in a chat."""
+
+    type: Literal["human", "ai", "tool", "custom"] = Field(
+        description="Role of the message.",
+        examples=["human", "ai", "tool", "custom"],
+    )
+    content: str = Field(
+        description="Content of the message.",
+        examples=["Hello, world!"],
+    )
+    tool_calls: list[ToolCall] = Field(
+        description="Tool calls in the message.",
+        default=[],
+    )
+    tool_call_id: str | None = Field(
+        description="Tool call that this message is responding to.",
+        default=None,
+        examples=["call_Jja7J89XsjrOLA5r!MEOW!SL"],
+    )
+    run_id: str | None = Field(
+        description="Run ID of the message.",
+        default=None,
+        examples=["847c6285-8fc9-4560-a83f-4e6285809254"],
+    )
+    response_metadata: dict[str, Any] = Field(
+        description="Response metadata. For example: response headers, logprobs, token counts.",
+        default={},
+    )
+    custom_data: dict[str, Any] = Field(
+        description="Custom message data.",
+        default={},
+    )

src/settings.py

@@ -0,0 +1,66 @@
+from typing import Any
+
+from dotenv import find_dotenv
+from pydantic import SecretStr, computed_field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """
+    Environment specific configuration
+    """
+
+    model_config = SettingsConfigDict(
+        env_file=find_dotenv(),
+        env_file_encoding="utf-8",
+        env_ignore_empty=True,
+        extra="ignore",
+        validate_default=False,
+    )
+    # path
+    MODE: str | None = None  # dev, prod
+    HOST: str = "0.0.0.0"
+    PORT: int = 8080
+
+    # Secret keys
+    AUTH_SECRET: SecretStr | None = None
+    JWT_ALGORITHM: str = "HS256"
+
+    # LLM API keys
+    GROQ_API_KEY: SecretStr
+    GROQ_MODEL: str = "llama3-8b-8192"
+    GROQ_MODEL_TEMP: float = 0.5
+
+    # Tracing for Langchain
+    LANGCHAIN_TRACING_V2: bool = (
+        False  # Data will be sent to Langchain for monitering and improvement, If enabled.
+    )
+    LANGCHAIN_PROJECT: str = "default"
+    LANGCHAIN_ENDPOINT: str = "https://api.smith.langchain.com"
+    LANGCHAIN_API_KEY: SecretStr | None = None  # LangSmith API key
+
+    # DB
+    DATABASE_URL: SecretStr
+
+    def model_post_init(self, __context: Any) -> None:
+        """
+        Validate the settings after initialization
+        """
+        if self.LANGCHAIN_TRACING_V2 and self.LANGCHAIN_API_KEY is None:
+            raise ValueError("Tracing is enabled, but LANGCHAIN_API_KEY is missing!")
+
+        if self.GROQ_API_KEY is None:
+            raise ValueError(
+                "GROQ_API_KEY is required! This key enables the application to connect with an advanced language model that understands queries and provides intelligent responses. You can generate your API at https://console.groq.com/keys ."
+            )
+
+    @computed_field
+    @property
+    def BASE_URL(self) -> str:
+        return f"http://{self.HOST}:{self.PORT}"
+
+    def is_dev(self) -> bool:
+        return self.MODE == "dev"
+
+
+settings = Settings()

src/tools.py

@@ -0,0 +1,76 @@
+from langchain_core.tools import tool
+from sqlmodel import select
+
+from database import Scammer, get_session
+
+
+@tool
+async def register_scam(
+    scammer_mobile: str, scam_id: int, reporter_ordeal: str, reporter_mobile: str
+) -> str:
+    """
+    Registers a report of a scam incident into the database.
+
+    Parameters:
+            - scammer_mobile (str): The mobile_number of the alleged scammer.
+              Must be formatted as "+XX-<mobile_number>", where "+XX" is the country code.
+            - scam_id (int): The unique identifier for the type of scam.
+            - reporter_ordeal (str): A summary of the ordeal narrated by the reporter.
+              Should not exceed 50 words.
+            - reporter_mobile (str): The mobile_number of the person reporting the scam.
+              Must be formatted as "+XX-<mobile_number>", where "+XX" is the country code.
+
+    Returns:
+        str: A confirmation message if the report is registered successfully, or an error
+        message if an exception occurs during the registration process.
+    """
+    try:
+        scammer = Scammer(
+            scammer_mobile=scammer_mobile,
+            scam_id=scam_id,
+            reporter_ordeal=reporter_ordeal,
+            reporter_mobile=reporter_mobile,
+        )
+        async with get_session() as session:
+            session.add(scammer)
+            await session.commit()
+            return f"{scammer_mobile} has been registered as a scammer. ✅ Thank you for combating scams! 🥇"
+    except ValueError as e:
+        print(repr(exc))
+        return f"ValueError: {repr(exc)}"
+    except Exception as exc:
+        print(repr(exc))
+        return f"An error occurred in registering a report for {scammer_mobile}."
+
+
+register_scam.name = "Register Scam"
+
+
+@tool
+async def search_scam(scammer_mobile: str) -> str:
+    """
+    Searches the database for scam reports associated with the provided mobile number.
+
+    Parameters:
+        scammer_mobile (str): The mobile number of the alleged scammer, formatted as "+XX-<mobile_number>",
+        where "+XX" is the country code.
+
+    Returns:
+        str: If a scam report is found, returns a string representation of the scam count.
+        If no scams are found, returns a message indicating that the mobile number is not reported.
+        If an error occurs during the search process, returns an error message.
+    """
+    try:
+        async with get_session() as session:
+            statement = select(Scammer).where(Scammer.scammer_mobile == scammer_mobile)
+            result = await session.exec(statement)
+            scams = result.all()
+            if not scams:
+                return f"{scammer_mobile} has never been reported for scams or fraudulent activity."
+            return f"{scammer_mobile} has been reported as a scammer {len(scams)} times in the past. 🚨 Be alert! ⚠️"
+    except Exception as exc:
+        print(repr(exc))
+        return f"An error occurred while searching scam for {scammer_mobile}."
+
+
+search_scam.name = "Search Scam"

src/utils.py

@@ -0,0 +1,79 @@
+from uuid import uuid4
+
+from langchain_core.messages import (
+    AIMessage,
+    BaseMessage,
+    ChatMessage,
+    HumanMessage,
+    ToolMessage,
+)
+from langchain_core.runnables import RunnableConfig
+
+from agent import cyber_guard
+from schema import Chat
+from settings import settings
+
+
+async def get_llm_response(user_input):
+    thread_id = user_input.thread_id or str(uuid4())
+    kwargs = {
+        "input": {"messages": [HumanMessage(content=user_input.user_message)]},
+        "config": RunnableConfig(
+            configurable={"thread_id": thread_id, "model": settings.GROQ_MODEL}
+        ),
+    }
+    response = await cyber_guard.ainvoke(**kwargs)
+    return response, thread_id
+
+
+def convert_message_content_to_string(content: str | list[str | dict]) -> str:
+    if isinstance(content, str):
+        return content
+    text: list[str] = []
+    for content_item in content:
+        if isinstance(content_item, str):
+            text.append(content_item)
+            continue
+        if content_item["type"] == "text":
+            text.append(content_item["text"])
+    return "".join(text)
+
+
+def infer_chat_message(message: BaseMessage) -> Chat:
+    """Create a Chat from a LangChain message."""
+    match message:
+        case HumanMessage():
+            human_message = Chat(
+                type="human",
+                content=convert_message_content_to_string(message.content),
+            )
+            return human_message
+        case AIMessage():
+            ai_message = Chat(
+                type="ai",
+                content=convert_message_content_to_string(message.content),
+            )
+            if message.tool_calls:
+                ai_message.tool_calls = message.tool_calls
+            if message.response_metadata:
+                ai_message.response_metadata = message.response_metadata
+            return ai_message
+        case ToolMessage():
+            tool_message = Chat(
+                type="tool",
+                content=convert_message_content_to_string(message.content),
+                tool_call_id=message.tool_call_id,
+            )
+            return tool_message
+        case ChatMessage():
+            if message.role == "custom":
+                custom_message = Chat(
+                    type="custom",
+                    content="",
+                    custom_data=message.content[0],
+                )
+                return custom_message
+            else:
+                raise ValueError(f"Unsupported chat message role: {message.role}")
+        case _:
+            raise ValueError(f"Unsupported message type: {message.__class__.__name__}")