changes

README.md

@@ -9,4 +9,71 @@ license: apache-2.0
 short_description: Email  ai agent project with memory.
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 📧 AI-Driven Email Agent 🧠
+
+A production-grade, multi-agent system built with LangGraph and FastAPI that automates email triage, context retrieval, and drafting. This project demonstrates advanced implementation of **Long-term Memory**, **State Persistence**, and **Human-in-the-Loop Interrupts** using LangGraph's Functional API Command pattern.
+
+---
+
+## 🚀 Key Features
+
+- **Advanced Learning Implementation**  
+  Implemented **Semantic Memory**, **Checkpointer Persistence**, and **Functional Interrupts**, enabling the agent to maintain state and handle user feedback reliably.
+
+- **Multi-Agent Workflow**  
+  Specialized agents for:
+  - Triage
+  - Context Synthesis
+  - Email Drafting
+
+- **Intelligent Triage**  
+  Automatically classifies emails, assigns priority, and determines if a reply is required.
+
+- **Semantic Memory**  
+  Uses `langmem` and `PostgresStore` to retrieve past interactions, allowing the agent to remember previous project details.
+
+- **Resource Management**  
+  A dedicated **Token Count Node** ensures large emails (e.g., deployment logs) are summarized before processing to optimize costs.
+
+- **Human-in-the-Loop**  
+  The graph pauses using `interrupt()` to allow users to:
+  - Review drafts  
+  - Approve responses  
+  - Provide feedback via `Command(resume=...)`
+
+- **Scalable Architecture**  
+  Built with **FastAPI**, **Docker**, and a modular structure for enterprise-grade deployment.
+
+---
+
+## 🛠️ Tech Stack
+
+- **Orchestration:** `langgraph` (Functional API), `langchain`  
+- **LLM Interface:** `langchain-groq`  
+- **Memory & Persistence:**  
+  - `langmem`  
+  - `PostgresCheckpoint`  
+  - `PostgresStore` (Neon/PostgreSQL)  
+- **Database ORM:** SQLAlchemy 2.0  
+- **Embeddings:** `langchain_huggingface` (DistilBERT)  
+- **Backend:** FastAPI + Uvicorn  
+- **Configuration:** `pydantic-settings` (.env management)  
+- **Authentication:** `google-auth` (Gmail API Integration)  
+
+---
+
+## 📂 Project Structure
+
+```bash
+app/
+├── agents/             # Brains: Specialized LLM logic (Triage, Writer, Context)
+├── database/           # Data: SQLAlchemy models and Connection Pooling
+├── nodes/              # Workflow: Functional steps of the graph (Safety, Tokens)
+├── persistance/        # Persistence: Postgres Checkpointer & Memory Store config
+├── state/              # Schema: Pydantic & TypedDict state definitions
+├── utils/              # Toolbox: Token counters, Embeddings, and Auth helpers
+├── graph.py            # Logic: StateGraph construction and compilation
+└── main.py             # Entry: FastAPI app and Controller logic
+
+
+

app/core/auth.py

@@ -0,0 +1,105 @@
+from datetime import datetime, timedelta, timezone
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from fastapi import HTTPException, Depends, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.orm import Session
+import hashlib
+import base64
+from app.core.config import settings
+from app.database.connection import get_session
+from app.database.models import User
+
+
+# CONFIG
+
+
+SECRET_KEY = settings.SECRET_KEY
+ALGORITHM = settings.ALGORITHM
+ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# auto_error=False → allows us to return 401 instead of FastAPI's 403
+security = HTTPBearer(auto_error=False)
+
+
+# TOKEN LOGIC
+
+
+def create_access_token(data: dict) -> str:
+    """
+    Create a JWT access token with UTC-aware expiration
+    """
+    to_encode = data.copy()
+
+    expire = datetime.now(timezone.utc) + timedelta(
+        minutes=ACCESS_TOKEN_EXPIRE_MINUTES
+    )
+
+    to_encode.update({
+        "exp": expire,
+        "sub": str(data.get("id"))  # JWT best practice
+    })
+
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+def verify_token(
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+) -> dict:
+    """
+    Verify JWT token and return payload
+    """
+    if credentials is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Not authenticated"
+        )
+
+    try:
+        token = credentials.credentials
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+
+        user_id = payload.get("id")
+        if user_id is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid token payload"
+            )
+
+        return payload
+
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token"
+        )
+
+# CURRENT USER DEPENDENCY
+
+
+async def get_current_user(
+    token_data: dict = Depends(verify_token),
+    db: Session = Depends(get_session)
+) -> User:
+    """
+    Fetch the logged-in user from DB using JWT payload
+    """
+    user_id = token_data.get("id")
+
+    user = (
+        db.query(User)
+        .filter(User.id == user_id)
+        .first()
+    )
+
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="User not found"
+        )
+
+    return user
+
+# PASSWORD LOGIC (bcrypt-safe, unlimited length)

app/core/config.py

@@ -11,6 +11,11 @@ class Settings(BaseSettings):
     GMAIL_CREDENTIALS_PATH: str = "credentials.json"
     GMAIL_TOKEN_PATH: str = "token.json"
 
+    SECRET_KEY: str
+    ALGORITHM: str
+
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440
+
     DB_URL_FOR_SQL_AL:str
  
     model_config = SettingsConfigDict(

app/gmail_auth.py

@@ -0,0 +1,20 @@
+from app.core.config import settings
+from langchain_google_community.gmail.utils import (
+    build_resource_service,
+    get_gmail_credentials,
+)
+
+from langchain_google_community import GmailToolkit
+credentials = get_gmail_credentials(
+    token_file=settings.GMAIL_TOKEN_PATH,
+    scopes=["https://mail.google.com/"],
+    client_sercret_file=settings.GMAIL_CREDENTIALS_PATH,
+)
+
+api_resource = build_resource_service(
+    credentials=credentials
+)
+
+gmail_toolkit = GmailToolkit(
+    api_resource=api_resource
+)

app/main.py

@@ -4,7 +4,6 @@ from typing import Optional, Dict, Any, TypedDict, Annotated, Sequence
 from langchain_core.messages import BaseMessage, HumanMessage
 from langgraph.graph import add_messages
 from langgraph.types import Command
-import uuid
 import logging
 from app.graph import graph
 from app.state.state import EmailAgentState
@@ -12,6 +11,24 @@ from app.database.connection import get_session
 from app.database.utils import get_or_create_user
 from sqlalchemy.orm import Session
 from app.database.connection import SessionLocal
+from app.core.config import settings
+from fastapi import Request
+import os
+from app.database.models import User
+from app.core.auth import create_access_token,get_current_user
+
+# CREATE GMAIL AUTH FILES FROM HF SECRETS
+
+
+if not os.path.exists(settings.GMAIL_CREDENTIALS_PATH):
+
+    with open(settings.GMAIL_CREDENTIALS_PATH, "w") as f:
+        f.write(os.environ["GOOGLE_CREDENTIALS"])
+
+if not os.path.exists(settings.GMAIL_TOKEN_PATH):
+
+    with open(settings.GMAIL_TOKEN_PATH, "w") as f:
+        f.write(os.environ["GOOGLE_TOKEN"])
 
 
 logger = logging.getLogger(__name__)
@@ -29,7 +46,6 @@ app = FastAPI(title="AI Email Agent API")
 
 class EmailProcessRequest(BaseModel):
     thread_id: str
-    user_email: EmailStr
     sender_email_id: EmailStr
     sender_subject: str
     sender_email_body: str
@@ -37,13 +53,12 @@ class EmailProcessRequest(BaseModel):
 
 class ReviewActionRequest(BaseModel):
     thread_id: str
-    user_id: str
+    
     status: str  # "approved" or "rejected"
     feedback: Optional[str] = None
 
 class SendEmailRequest(BaseModel):
     thread_id: str
-    user_id: str
     human_message: str
 # --- Helper Functions ---
 
@@ -67,24 +82,38 @@ def parse_interrupt(final_state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
 
 # --- Endpoints ---
 
+
+@app.post("/get-user-data")
+def get_user_data(user_email: EmailStr, db: Session = Depends(get_session)):
+    """Get user data by email."""
+    user = get_or_create_user(db, user_email)
+
+    token = create_access_token({
+            "id": user.id,
+            "email": user_email
+        })
+    return {"user_id": str(user.id), "email": user.email, "token": token}
+
+
+
 @app.post("/process-email")
-def process_email(request: EmailProcessRequest, db: Session = Depends(get_session)) -> Dict[str, Any]:
+def process_email(request: EmailProcessRequest, db: Session = Depends(get_session), current_user: User = Depends(get_current_user)) -> Dict[str, Any]:
     """Process email through the graph pipeline."""
     
     try:
-        user = get_or_create_user(db, request.user_email)
+        
         
         thread_id = request.thread_id
         config = {
             "configurable": {
                 "thread_id": thread_id,
-                "user_id": str(user.id)
+                "user_id": str(current_user.id)
             }
         }
         
         input_data = {
-            "user_email_id": request.user_email,
-            "user_id": user.id,
+            "user_email_id": current_user.email,
+            "user_id": current_user.id,
             "user_name": "Atharva",
             "sender_email_id": request.sender_email_id,
             "sender_subject": request.sender_subject,
@@ -123,14 +152,14 @@ def process_email(request: EmailProcessRequest, db: Session = Depends(get_sessio
 
 
 @app.post("/review-action")
-def review_action(request: ReviewActionRequest) -> Dict[str, Any]:
+def review_action(request: ReviewActionRequest,db: Session = Depends(get_session), current_user: User = Depends(get_current_user)) -> Dict[str, Any]:
     """Resume graph execution based on user review."""
     
     try:
         config = {
             "configurable": {
                 "thread_id": request.thread_id,
-                "user_id": request.user_id
+                "user_id": current_user.id
             }
         }
 
@@ -182,12 +211,12 @@ def review_action(request: ReviewActionRequest) -> Dict[str, Any]:
 
 
 @app.post("/send_email")
-def send_email(request: SendEmailRequest) -> Dict[str, Any]:
+def send_email(request: SendEmailRequest,db: Session = Depends(get_session),current_user: User = Depends(get_current_user)) -> Dict[str, Any]:
    
         config = {
             "configurable": {
                 "thread_id": request.thread_id,
-                "user_id": request.user_id
+                "user_id": current_user.id
             }
         }
 
@@ -195,7 +224,7 @@ def send_email(request: SendEmailRequest) -> Dict[str, Any]:
           config,
           {"messages": [HumanMessage(content=request.human_message)]},
        as_node="prepare_context_node" 
-     )  
+        )  
         final_state = graph.invoke(None, config=config)
 
         return {
@@ -210,4 +239,4 @@ def send_email(request: SendEmailRequest) -> Dict[str, Any]:
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="127.0.0.1", port=8000)
+    uvicorn.run(app, host="127.0.0.1", port=8080)

app/tools/email_writing_agent_tools.py

@@ -7,9 +7,10 @@ from typing import Annotated, Union
 from langchain_core.tools import InjectedToolCallId, tool
 from langchain.tools import ToolRuntime
 from langgraph.types import Command
-from langchain_core.messages import SystemMessage, HumanMessage,ToolMessage,AIMessage,BaseMessage
 from langgraph.graph import END
+from langchain_core.messages import ToolMessage,SystemMessage, HumanMessage
 
+from app.gmail_auth import gmail_toolkit ,api_resource
 @tool(args_schema=CreateDraftSchema)
 def create_gmail_draft(
     to: Union[str, list], 
@@ -31,8 +32,8 @@ def create_gmail_draft(
         "data": {"to": to, "subject": subject, "body": body}
     })
 
-    toolkit = GmailToolkit() 
-    draft_tool = [t for t in toolkit.get_tools() if t.name == "create_gmail_draft"][0]
+    
+    draft_tool = [t for t in gmail_toolkit.get_tools() if t.name == "create_gmail_draft"][0]
 
     # 2. Handle Logic
     if response.get("status") == "approved":
@@ -61,14 +62,15 @@ def create_gmail_draft(
 #---------------------------------------------------------------------------
 
 
+@tool
 def send_draft(
     tool_call_id: Annotated[str, InjectedToolCallId],runtime: ToolRuntime # Injected ID
 ):
     """Sends a finalized Gmail draft by its ID."""
 
     try:
-        toolkit = GmailToolkit()
-        result = toolkit.api_resource.users().drafts().send(
+    
+        result = gmail_toolkit.api_resource.users().drafts().send(
             userId="me", body={"id": runtime.state["draft_id"]}
         ).execute()
         

authenticate_gmail.py

@@ -0,0 +1,29 @@
+import os
+from google_auth_oauthlib.flow import InstalledAppFlow
+
+# Gmail API scope
+SCOPES = [
+    "https://mail.google.com/"
+]
+
+def run_authentication():
+    print("Starting Google OAuth Authentication Flow...")
+
+    if not os.path.exists("credentials.json"):
+        print("ERROR: Please place your 'credentials.json' file in this folder first!")
+        return
+
+    flow = InstalledAppFlow.from_client_secrets_file(
+        "credentials.json",
+        SCOPES
+    )
+
+    creds = flow.run_local_server(port=8080)
+
+    with open("token.json", "w") as token_file:
+        token_file.write(creds.to_json())
+
+    print("\nSUCCESS: 'token.json' has been created successfully!")
+
+if __name__ == "__main__":
+    run_authentication()

requirements.txt

@@ -14,3 +14,6 @@ psycopg-pool
 transformers
 pydantic-settings
 email-validator
+itsdangerous
+python-jose[cryptography]
+passlib[bcrypt]