Add context-aware conversational AI assistant for Sunbird Ed (Issue #515)

sunbird-ai-assistant/DockerFile

@@ -0,0 +1,5 @@
+FROM python:3.10-slim
+WORKDIR /app
+COPY . .
+RUN pip install -r requirements.txt
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

sunbird-ai-assistant/README.MD

@@ -0,0 +1,84 @@
+# 🧠 Sunbird AI Assistant
+
+A context-aware conversational agent built on top of the **Model Context Protocol (MCP)** and powered by **OpenAI’s function-calling tools**. This assistant interacts with **Sunbird Ed APIs**, adapting its capabilities to different user roles like **learner**, **admin**, and **mentor** to enhance the digital public good ecosystem in education.
+
+---
+
+## 🌐 Overview
+
+Sunbird Ed is a modular Digital Public Good (DPG) designed to support learning and skilling platforms. However, it lacks intelligent conversational capabilities. This project integrates a role-aware AI assistant that:
+
+- Understands the **context** of a specific installation
+- Personalizes responses based on **user persona**
+- Interacts with key **Sunbird Ed APIs** (mocked or real)
+- Is modular and extensible across various deployments
+
+---
+
+## 📌 Features
+
+- 🔧 Tool-based LangChain agent integration with role-aware access
+- 🔑 Authentication layer for API authorization
+- 💬 Conversational interface (CLI or REST)
+- 🧠 Session context manager per installation and user
+- 🧰 Supports key Sunbird Ed APIs:
+  - Course metadata
+  - User enrollments
+  - Learning progress
+  - Admin controls
+- 📦 Easily extendable for additional DPGs and APIs
+
+---
+
+## 🧱 Architecture
+
+```txt
+                          +-------------------+
+                          |   CLI / Web UI    |
+                          +--------+----------+
+                                   |
+                                   v
+                        +----------------------+
+                        |   FastAPI Backend     |
+                        +----------+-----------+
+                                   |
+        +--------------------------+-------------------------+
+        |                        LangChain Agent              |
+        |       (Function Calling, Role-based Tool Access)    |
+        +----------+------------+------------+----------------+
+                   |            |            |
+         +---------+    +-------+-----+    +-+------------------+
+         | Course Tools | Enrollment Tools | Admin Tools (RBAC) |
+         +-------------+------------------+---------------------+
+
+
+## Project Structure :
+
+.
+├── app/
+│   ├── main.py                  
+│   ├── agent/
+│   │   ├── session_manager.py   # Manages session/user context
+│   ├── auth/
+│   │   └── auth.py             
+│   ├── tool_schemas/
+│   │   ├── course_tools.py      
+│   │   ├── enrollment_tools.py  
+│   │   └── admin_tools.py       
+│   ├── utils/
+│   │   └── logger.py           
+│   ├── tool_registry.py         # Registers tools based on role
+│   └── agent_setup.py           
+├── requirements.txt
+└── README.md
+
+
+## 🧪 API Simulation & Mocking
+
+All Sunbird Ed APIs like `/course/v1/search`, `/user/enrollment/list` are currently mocked.  
+You can easily replace their logic in the tool schema files using actual API calls like:
+
+```python
+requests.get("https://<your-sunbird-endpoint>/course/v1/search", headers={...})
+
+

sunbird-ai-assistant/app/agent/mcp_agent.py

@@ -0,0 +1,32 @@
+from langchain.agents import initialize_agent
+from langchain.chat_models import ChatOpenAI
+from tool_registry import register_tools
+from app.agent.session_manager import SessionManager
+from app.utils.loggers import get_logger
+
+logger = get_logger()
+
+def setup_agent(installation_id: str, user_id: str, persona: str):
+    """
+    Initialize the LangChain agent with tools registered based on user persona.
+    """
+    logger.info(f"Setting up agent for user '{user_id}' with role '{persona}' on installation '{installation_id}'")
+
+    # Register tools based on user role/persona (e.g., learner, admin)
+    tools = register_tools(persona)
+
+    # Setup LLM (OpenAI GPT-4)
+    llm = ChatOpenAI(temperature=0, model="gpt-4")
+
+    # Create a session manager with contextual memory
+    session = SessionManager(installation_id, user_id, persona)
+
+    # Initialize LangChain agent with tools and LLM
+    agent = initialize_agent(
+        tools=tools,
+        llm=llm,
+        agent_type="openai-tools",
+        verbose=True
+    )
+
+    return agent, session

sunbird-ai-assistant/app/agent/session_manager.py

@@ -0,0 +1,23 @@
+from app.services.context_loader import get_installation_context
+from app.utils.loggers import get_logger
+
+logger = get_logger()
+
+class SessionManager:
+    def __init__(self, installation_id: str, user_id: str, persona: str):
+        self.installation_id = installation_id
+        self.user_id = user_id
+        self.persona = persona
+        self.context = self.load_context()
+
+    def load_context(self):
+        context = get_installation_context(self.installation_id)
+        logger.info(f"Loaded context for {self.installation_id}")
+        return {
+            "persona": self.persona,
+            "installation_context": context,
+            "user_id": self.user_id
+        }
+
+    def inject_context(self, prompt: str) -> str:
+        return f"{self.persona.upper()} CONTEXT: {self.context}\n\n{prompt}"

sunbird-ai-assistant/app/agent/tool_registry.py

@@ -0,0 +1,13 @@
+from app.tool_schemas.course_tools import get_course_metadata
+from app.tool_schemas.enrollment_tools import get_user_enrollments, get_course_progress
+
+def register_tools(user_role: str):
+    """Dynamically register tools based on the user's role/persona."""
+    tools = [get_course_metadata, get_user_enrollments, get_course_progress]
+
+    if user_role == "admin":
+        # Simulate additional admin-only tools (add them to a separate schema file)
+        from app.tool_schemas.admin_tools import manage_users, list_all_courses
+        tools += [manage_users, list_all_courses]
+
+    return tools

sunbird-ai-assistant/app/config.py

@@ -0,0 +1,9 @@
+import os
+
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-3.5-turbo")
+OPENAI_TEMPERATURE = float(os.getenv("OPENAI_TEMPERATURE", 0.7))
+OPENAI_MAX_TOKENS = int(os.getenv("OPENAI_MAX_TOKENS", 4096))
+OPENAI_TOP_P = float(os.getenv("OPENAI_TOP_P", 1.0))
+OPENAI_FREQUENCY_PENALTY = float(os.getenv("OPENAI_FREQUENCY_PENALTY", 0.0))
+OPENAI_PRESENCE_PENALTY = float(os.getenv("OPENAI_PRESENCE_PENALTY", 0.0))

sunbird-ai-assistant/app/main.py

@@ -0,0 +1,21 @@
+from fastapi import FastAPI
+from app.agent.mcp_agent import setup_agent
+from app.services.mock_api import mock_router
+
+from utils.loggers import get_logger
+
+logger = get_logger()
+
+logger.info("Assistant initialized successfully.")
+logger.error("Failed to load user enrollments.")
+
+
+app = FastAPI()
+
+# Include mock API routes to simulate Sunbird Ed
+app.include_router(mock_router)
+
+
+@app.on_event("startup")
+async def startup_event():
+    setup_agent()

sunbird-ai-assistant/app/services/auth.py

@@ -0,0 +1,59 @@
+import os
+import time
+import hashlib
+from typing import Optional
+from utils.loggers import get_logger
+
+logger = get_logger()
+
+# Simulate a user/token database
+MOCK_USERS = {
+    "user_001": "admin",
+    "user_002": "learner",
+    "user_003": "mentor"
+}
+
+# Simulated secret key (could be replaced with JWT secret or OAuth config)
+SECRET_KEY = os.getenv("AUTH_SECRET_KEY", "sunbird_secret_key")
+
+# Token TTL in seconds (e.g., 1 hour)
+TOKEN_TTL = 3600
+
+
+def generate_token(user_id: str) -> str:
+    """Generate a simple hashed token with TTL."""
+    if user_id not in MOCK_USERS:
+        raise ValueError("Invalid user ID")
+
+    timestamp = str(int(time.time()) + TOKEN_TTL)
+    raw = f"{user_id}:{timestamp}:{SECRET_KEY}"
+    token = hashlib.sha256(raw.encode()).hexdigest()
+    logger.info(f"Generated token for {user_id}")
+    return f"{user_id}:{timestamp}:{token}"
+
+
+def validate_token(token: str) -> Optional[str]:
+    """Validate token and return user_id if valid, else None."""
+    try:
+        user_id, timestamp, token_hash = token.split(":")
+        if time.time() > int(timestamp):
+            logger.warning("Token expired")
+            return None
+
+        expected_raw = f"{user_id}:{timestamp}:{SECRET_KEY}"
+        expected_hash = hashlib.sha256(expected_raw.encode()).hexdigest()
+
+        if expected_hash == token_hash:
+            logger.info(f"Validated token for {user_id}")
+            return user_id
+        else:
+            logger.warning("Token hash mismatch")
+            return None
+    except Exception as e:
+        logger.error(f"Token validation failed: {e}")
+        return None
+
+
+def get_user_role(user_id: str) -> Optional[str]:
+    """Get the role of the user from mock DB."""
+    return MOCK_USERS.get(user_id)

sunbird-ai-assistant/app/services/context_loader.py

@@ -0,0 +1,13 @@
+import json
+from app.utils.redis_cache import get_cache, set_cache
+
+def get_installation_context(installation_id: str):
+    key = f"context:{installation_id}"
+    cached = get_cache(key)
+    if cached:
+        return json.loads(cached)
+
+    with open(f"data/installation_contexts/demo_deployment.json") as f:
+        context = json.load(f)
+        set_cache(key, json.dumps(context))
+        return context

sunbird-ai-assistant/app/services/mock_api.py

@@ -0,0 +1,21 @@
+from fastapi import APIRouter
+
+mock_router = APIRouter()
+
+@mock_router.get("/course/v1/search")
+def search_courses(query: str = ""):
+    return {
+        "courses": [
+            {"id": "c101", "title": "Python Basics"},
+            {"id": "c102", "title": "Data Science"},
+        ]
+    }
+
+@mock_router.get("/user/enrollment/list")
+def get_enrollments(user_id: str):
+    return {
+        "user_id": user_id,
+        "enrollments": [
+            {"course_id": "c101", "progress": "50%"},
+        ]
+    }

sunbird-ai-assistant/app/tool_schemas/admin_tools.py

@@ -0,0 +1,41 @@
+from langchain.tools import tool
+from typing import List, Dict
+
+# Simulated Course and User DB
+ALL_COURSES = {
+    "python101": "Intro to Python",
+    "ml202": "Machine Learning Fundamentals",
+    "dl301": "Deep Learning with PyTorch"
+}
+
+ALL_USERS = {
+    "user_001": "admin",
+    "user_002": "learner",
+    "user_003": "mentor"
+}
+
+@tool
+def list_all_courses() -> List[str]:
+    """
+    List all available courses in the platform (Admin only).
+    """
+    return list(ALL_COURSES.values())
+
+@tool
+def manage_users(action: str, target_user: str, new_role: str = "") -> Dict:
+    """
+    Simulate user management operations (Admin only).
+    Supported actions: 'view', 'update'
+    """
+    if target_user not in ALL_USERS:
+        return {"error": "User not found"}
+
+    if action == "view":
+        return {target_user: ALL_USERS[target_user]}
+    elif action == "update":
+        if not new_role:
+            return {"error": "New role must be specified for update"}
+        ALL_USERS[target_user] = new_role
+        return {"message": f"Role updated to '{new_role}' for user '{target_user}'"}
+    else:
+        return {"error": "Unsupported action. Use 'view' or 'update'"}

sunbird-ai-assistant/app/tool_schemas/course_tools.py

@@ -0,0 +1,23 @@
+from typing import Dict
+from langchain.tools import tool
+
+# Simulated Course Catalog
+COURSE_DB = {
+    "python101": {"title": "Intro to Python", "level": "Beginner", "duration": "4 weeks"},
+    "ml202": {"title": "Machine Learning Fundamentals", "level": "Intermediate", "duration": "6 weeks"}
+}
+
+@tool
+def get_course_metadata(course_id: str) -> Dict:
+    """
+    Fetch metadata about a specific course using its ID.
+    """
+    course = COURSE_DB.get(course_id)
+    if not course:
+        return {"error": f"No course found with ID '{course_id}'."}
+    return {
+        "course_id": course_id,
+        "title": course["title"],
+        "level": course["level"],
+        "duration": course["duration"]
+    }

sunbird-ai-assistant/app/tool_schemas/enrollment_tools.py

@@ -0,0 +1,30 @@
+from typing import List, Dict
+from langchain.tools import tool
+
+# Simulated User Enrollment Data
+USER_ENROLLMENTS = {
+    "user_001": ["python101"],
+    "user_002": ["ml202"],
+}
+
+PROGRESS_TRACKER = {
+    "user_001": {"python101": "80%"},
+    "user_002": {"ml202": "40%"},
+}
+
+@tool
+def get_user_enrollments(user_id: str) -> List[str]:
+    """
+    Retrieve a list of course IDs that the user is enrolled in.
+    """
+    return USER_ENROLLMENTS.get(user_id, [])
+
+@tool
+def get_course_progress(user_id: str, course_id: str) -> Dict:
+    """
+    Get progress data for a specific user and course.
+    """
+    progress = PROGRESS_TRACKER.get(user_id, {}).get(course_id)
+    if not progress:
+        return {"error": "Progress data not found for given user and course."}
+    return {"user_id": user_id, "course_id": course_id, "progress": progress}