knowledge-assistant/
├── src/
│   ├── __init__.py
│   ├── config.py
│   ├── auth/
│   │   ├── __init__.py
│   │   ├── sso.py               # SSO integration
│   │   └── permissions.py       # RBAC handling
│   ├── channels/
│   │   ├── __init__.py
│   │   ├── slack_bot.py         # Slack integration
│   │   ├── teams_bot.py         # MS Teams
│   │   └── web_api.py           # Web portal API
│   ├── knowledge/
│   │   ├── __init__.py
│   │   ├── indexer.py           # Document indexing
│   │   ├── retriever.py         # Secure retrieval
│   │   └── sources/
│   │       ├── confluence.py    # Confluence connector
│   │       ├── sharepoint.py    # SharePoint connector
│   │       └── notion.py        # Notion connector
│   ├── processing/
│   │   ├── __init__.py
│   │   ├── classifier.py        # Topic classification
│   │   ├── router.py            # Department routing
│   │   └── generator.py         # Response generation
│   ├── actions/
│   │   ├── __init__.py
│   │   ├── ticketing.py         # Ticket creation
│   │   └── self_service.py      # Self-service actions
│   └── api/
│       ├── __init__.py
│       └── main.py              # FastAPI application
├── tests/
└── requirements.txt

# src/config.py
from pydantic_settings import BaseSettings
from typing import Dict, List
from enum import Enum

class Department(str, Enum):
    HR = "hr"
    IT = "it"
    FINANCE = "finance"
    LEGAL = "legal"
    FACILITIES = "facilities"
    GENERAL = "general"

class AccessLevel(str, Enum):
    PUBLIC = "public"
    EMPLOYEE = "employee"
    MANAGER = "manager"
    HR_ONLY = "hr_only"
    EXECUTIVE = "executive"

class Settings(BaseSettings):
    # LLM Settings
    openai_api_key: str
    model: str = "gpt-4o"

    # Vector Database
    pinecone_api_key: str
    pinecone_index: str = "knowledge-base"

    # Slack
    slack_bot_token: str
    slack_signing_secret: str

    # Auth
    sso_provider: str = "okta"
    sso_domain: str = ""

    # Knowledge Sources
    confluence_url: str = ""
    confluence_token: str = ""
    sharepoint_site: str = ""

    # Ticketing
    jira_url: str = ""
    jira_token: str = ""
    servicenow_instance: str = ""

    # Department Mappings
    department_channels: Dict[str, str] = {
        "hr": "#hr-support",
        "it": "#it-helpdesk",
        "finance": "#finance-questions",
        "facilities": "#facilities"
    }

    class Config:
        env_file = ".env"

settings = Settings()

# src/auth/permissions.py
from typing import List, Set
from dataclasses import dataclass
from enum import Enum
from ..config import AccessLevel

@dataclass
class UserContext:
    user_id: str
    email: str
    name: str
    department: str
    title: str
    groups: List[str]
    is_manager: bool
    access_levels: Set[AccessLevel]

class PermissionResolver:
    """Resolves user permissions for document access."""

    # Group to access level mappings
    GROUP_MAPPINGS = {
        "all-employees": AccessLevel.EMPLOYEE,
        "managers": AccessLevel.MANAGER,
        "hr-team": AccessLevel.HR_ONLY,
        "executives": AccessLevel.EXECUTIVE,
    }

    def get_user_access_levels(self, user: UserContext) -> Set[AccessLevel]:
        """Get all access levels for a user."""
        levels = {AccessLevel.PUBLIC}  # Everyone gets public

        if user.email:  # Authenticated user
            levels.add(AccessLevel.EMPLOYEE)

        if user.is_manager:
            levels.add(AccessLevel.MANAGER)

        for group in user.groups:
            if group in self.GROUP_MAPPINGS:
                levels.add(self.GROUP_MAPPINGS[group])

        return levels

    def can_access(self, user: UserContext, doc_access_level: str) -> bool:
        """Check if user can access a document."""
        required = AccessLevel(doc_access_level)
        return required in user.access_levels


# src/knowledge/retriever.py
from typing import List, Dict
from pinecone import Pinecone
from langchain_openai import OpenAIEmbeddings
from ..auth.permissions import UserContext, PermissionResolver
from ..config import settings

class SecureRetriever:
    """RAG retrieval with permission filtering."""

    def __init__(self):
        self.pc = Pinecone(api_key=settings.pinecone_api_key)
        self.index = self.pc.Index(settings.pinecone_index)
        self.embeddings = OpenAIEmbeddings(
            api_key=settings.openai_api_key
        )
        self.permission_resolver = PermissionResolver()

    async def retrieve(
        self,
        query: str,
        user: UserContext,
        department: str = None,
        k: int = 5
    ) -> List[Dict]:
        """Retrieve documents with permission filtering."""
        # Get query embedding
        query_embedding = self.embeddings.embed_query(query)

        # Build filter based on user permissions
        access_levels = [level.value for level in user.access_levels]

        filter_dict = {
            "access_level": {"$in": access_levels}
        }

        if department:
            filter_dict["department"] = department

        # Query Pinecone with metadata filter
        results = self.index.query(
            vector=query_embedding,
            filter=filter_dict,
            top_k=k,
            include_metadata=True
        )

        return [
            {
                "content": match.metadata.get("content", ""),
                "source": match.metadata.get("source", ""),
                "title": match.metadata.get("title", ""),
                "department": match.metadata.get("department", ""),
                "url": match.metadata.get("url", ""),
                "score": match.score,
                "last_updated": match.metadata.get("last_updated", "")
            }
            for match in results.matches
        ]

# src/processing/classifier.py
from typing import Tuple
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from ..config import settings, Department

class ClassificationResult(BaseModel):
    department: Department
    topic: str
    confidence: float = Field(ge=0, le=1)
    is_sensitive: bool = False
    requires_human: bool = False
    suggested_action: str = None

class TopicClassifier:
    """Classifies employee queries by department and topic."""

    def __init__(self):
        self.llm = ChatOpenAI(
            model=settings.model,
            api_key=settings.openai_api_key,
            temperature=0
        ).with_structured_output(ClassificationResult)

        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an internal assistant classifier.
Classify employee questions to the right department:

- hr: Benefits, PTO, policies, performance reviews, compensation, onboarding
- it: Password reset, software access, hardware, VPN, email issues
- finance: Expense reports, invoices, budget, procurement
- legal: Contracts, compliance, NDAs, IP questions
- facilities: Office access, parking, meeting rooms, supplies
- general: Company info, org chart, announcements

Identify:
- Is this sensitive (salary, performance, personal)?
- Does it require human intervention?
- What self-service action might help?"""),
            ("human", "Employee question: {question}")
        ])

    async def classify(self, question: str) -> ClassificationResult:
        """Classify an employee question."""
        chain = self.prompt | self.llm
        result = await chain.ainvoke({"question": question})
        return result


class DepartmentRouter:
    """Routes queries to appropriate handlers."""

    def __init__(self):
        self.classifier = TopicClassifier()

    async def route(self, question: str, user_context: dict) -> dict:
        """Route question to appropriate department."""
        classification = await self.classifier.classify(question)

        routing = {
            "department": classification.department.value,
            "topic": classification.topic,
            "confidence": classification.confidence,
            "handler": self._get_handler(classification),
            "escalation_channel": settings.department_channels.get(
                classification.department.value
            ),
            "is_sensitive": classification.is_sensitive,
            "requires_human": classification.requires_human
        }

        return routing

    def _get_handler(self, classification: ClassificationResult) -> str:
        """Get the appropriate handler for this classification."""
        if classification.requires_human:
            return "human_escalation"
        if classification.suggested_action:
            return "self_service"
        return "rag_response"

# src/processing/generator.py
from typing import Dict, List
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from ..config import settings

class AssistantResponse(BaseModel):
    answer: str
    sources: List[str] = []
    confidence: float = Field(ge=0, le=1)
    follow_up_questions: List[str] = []
    action_buttons: List[Dict] = []
    disclaimer: str = None

class KnowledgeGenerator:
    """Generates responses from knowledge base."""

    def __init__(self):
        self.llm = ChatOpenAI(
            model=settings.model,
            api_key=settings.openai_api_key,
            temperature=0.2
        ).with_structured_output(AssistantResponse)

        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an internal knowledge assistant helping employees.

Guidelines:
- Answer based ONLY on the provided knowledge context
- Be concise and direct
- Include source references [Source: title]
- For policy questions, quote the relevant section
- If unsure, say so and suggest who to contact
- For sensitive topics, recommend speaking with HR/manager
- Suggest helpful follow-up questions
- Add action buttons for self-service options

Knowledge Context:
{context}

User Department: {user_department}
User Role: {user_role}"""),
            ("human", "{question}")
        ])

    async def generate(
        self,
        question: str,
        context: List[Dict],
        user_department: str,
        user_role: str
    ) -> AssistantResponse:
        """Generate response from knowledge context."""
        # Format context
        context_str = "\n\n".join([
            f"[{doc['title']}]\n{doc['content']}\nSource: {doc['url']}"
            for doc in context
        ])

        chain = self.prompt | self.llm
        result = await chain.ainvoke({
            "context": context_str,
            "user_department": user_department,
            "user_role": user_role,
            "question": question
        })

        # Add action buttons based on topic
        result.action_buttons = self._get_action_buttons(question, context)

        return result

    def _get_action_buttons(
        self,
        question: str,
        context: List[Dict]
    ) -> List[Dict]:
        """Generate relevant action buttons."""
        buttons = []
        question_lower = question.lower()

        # Common self-service actions
        if "password" in question_lower:
            buttons.append({
                "label": "Reset Password",
                "action": "password_reset",
                "url": "/self-service/password-reset"
            })

        if any(w in question_lower for w in ["pto", "vacation", "time off"]):
            buttons.append({
                "label": "Submit PTO Request",
                "action": "pto_request",
                "url": "/hr/pto-request"
            })

        if any(w in question_lower for w in ["software", "access", "install"]):
            buttons.append({
                "label": "Request Software",
                "action": "software_request",
                "url": "/it/software-request"
            })

        if any(w in question_lower for w in ["expense", "reimbursement"]):
            buttons.append({
                "label": "Submit Expense",
                "action": "expense_submit",
                "url": "/finance/expense-report"
            })

        # Always offer ticket creation
        buttons.append({
            "label": "Create Support Ticket",
            "action": "create_ticket",
            "url": "/support/new-ticket"
        })

        return buttons[:3]  # Max 3 buttons

# src/channels/slack_bot.py
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from ..processing.classifier import DepartmentRouter
from ..processing.generator import KnowledgeGenerator
from ..knowledge.retriever import SecureRetriever
from ..auth.permissions import UserContext
from ..config import settings

app = AsyncApp(
    token=settings.slack_bot_token,
    signing_secret=settings.slack_signing_secret
)
handler = AsyncSlackRequestHandler(app)

# Initialize components
router = DepartmentRouter()
retriever = SecureRetriever()
generator = KnowledgeGenerator()

async def get_user_context(user_id: str, client) -> UserContext:
    """Get user context from Slack."""
    user_info = await client.users_info(user=user_id)
    user = user_info["user"]

    # In production, fetch from your identity provider
    return UserContext(
        user_id=user_id,
        email=user.get("profile", {}).get("email", ""),
        name=user.get("real_name", ""),
        department="engineering",  # Fetch from HR system
        title=user.get("profile", {}).get("title", ""),
        groups=["all-employees"],  # Fetch from identity provider
        is_manager=False,
        access_levels=set()
    )

@app.event("app_mention")
async def handle_mention(event, say, client):
    """Handle @mentions in channels."""
    user_id = event["user"]
    text = event["text"]
    channel = event["channel"]

    # Remove bot mention from text
    question = text.split(">", 1)[-1].strip()

    if not question:
        await say("Hi! How can I help you? Ask me about HR policies, IT support, or company info.")
        return

    await process_question(question, user_id, channel, say, client)

@app.event("message")
async def handle_dm(event, say, client):
    """Handle direct messages."""
    if event.get("channel_type") != "im":
        return

    user_id = event["user"]
    question = event["text"]
    channel = event["channel"]

    await process_question(question, user_id, channel, say, client)

async def process_question(question, user_id, channel, say, client):
    """Process a question and respond."""
    # Show typing indicator
    await client.reactions_add(channel=channel, name="thinking_face", timestamp=event["ts"])

    try:
        # Get user context
        user = await get_user_context(user_id, client)

        # Classify and route
        routing = await router.route(question, {"user": user})

        # Retrieve relevant documents
        docs = await retriever.retrieve(
            query=question,
            user=user,
            department=routing["department"],
            k=5
        )

        if not docs:
            await say(f"I couldn't find information about that. Try contacting {routing['escalation_channel']} directly.")
            return

        # Generate response
        response = await generator.generate(
            question=question,
            context=docs,
            user_department=user.department,
            user_role=user.title
        )

        # Format Slack message
        blocks = format_slack_response(response, routing)
        await say(blocks=blocks)

    except Exception as e:
        await say(f"Sorry, I encountered an error. Please try again or contact IT support.")

def format_slack_response(response, routing) -> list:
    """Format response for Slack."""
    blocks = [
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": response.answer}
        }
    ]

    # Add sources
    if response.sources:
        sources_text = "\n".join([f"• <{s}|{s.split('/')[-1]}>" for s in response.sources[:3]])
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"*Sources:*\n{sources_text}"}]
        })

    # Add action buttons
    if response.action_buttons:
        blocks.append({
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": btn["label"]},
                    "url": btn["url"],
                    "action_id": btn["action"]
                }
                for btn in response.action_buttons[:3]
            ]
        })

    # Add follow-up suggestions
    if response.follow_up_questions:
        suggestions = " | ".join(response.follow_up_questions[:3])
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"_Related questions: {suggestions}_"}]
        })

    # Add disclaimer if present
    if response.disclaimer:
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"⚠️ {response.disclaimer}"}]
        })

    return blocks

@app.action("create_ticket")
async def handle_ticket_action(ack, body, client):
    """Handle ticket creation button click."""
    await ack()

    # Open ticket creation modal
    await client.views_open(
        trigger_id=body["trigger_id"],
        view={
            "type": "modal",
            "title": {"type": "plain_text", "text": "Create Support Ticket"},
            "submit": {"type": "plain_text", "text": "Submit"},
            "blocks": [
                {
                    "type": "input",
                    "block_id": "title",
                    "element": {
                        "type": "plain_text_input",
                        "action_id": "title_input"
                    },
                    "label": {"type": "plain_text", "text": "Title"}
                },
                {
                    "type": "input",
                    "block_id": "description",
                    "element": {
                        "type": "plain_text_input",
                        "multiline": True,
                        "action_id": "description_input"
                    },
                    "label": {"type": "plain_text", "text": "Description"}
                },
                {
                    "type": "input",
                    "block_id": "department",
                    "element": {
                        "type": "static_select",
                        "action_id": "department_select",
                        "options": [
                            {"text": {"type": "plain_text", "text": "IT"}, "value": "it"},
                            {"text": {"type": "plain_text", "text": "HR"}, "value": "hr"},
                            {"text": {"type": "plain_text", "text": "Facilities"}, "value": "facilities"}
                        ]
                    },
                    "label": {"type": "plain_text", "text": "Department"}
                }
            ]
        }
    )

# src/api/main.py
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from typing import Optional, List

from ..channels.slack_bot import handler as slack_handler
from ..processing.classifier import DepartmentRouter
from ..processing.generator import KnowledgeGenerator
from ..knowledge.retriever import SecureRetriever
from ..auth.permissions import UserContext, PermissionResolver

app = FastAPI(
    title="Internal Knowledge Assistant",
    description="AI-powered employee knowledge assistant"
)

security = HTTPBearer()
router = DepartmentRouter()
retriever = SecureRetriever()
generator = KnowledgeGenerator()

class QuestionRequest(BaseModel):
    question: str

class AnswerResponse(BaseModel):
    answer: str
    sources: List[str]
    confidence: float
    department: str
    action_buttons: List[dict]

async def get_user_from_token(token: str) -> UserContext:
    """Validate token and return user context."""
    # In production, validate JWT and fetch user from identity provider
    # This is a simplified example
    return UserContext(
        user_id="user123",
        email="employee@company.com",
        name="John Doe",
        department="engineering",
        title="Software Engineer",
        groups=["all-employees", "engineering"],
        is_manager=False,
        access_levels=set()
    )

@app.post("/api/ask", response_model=AnswerResponse)
async def ask_question(
    request: QuestionRequest,
    credentials = Depends(security)
):
    """Ask a question to the knowledge assistant."""
    # Get user context
    user = await get_user_from_token(credentials.credentials)

    # Fill in access levels
    resolver = PermissionResolver()
    user.access_levels = resolver.get_user_access_levels(user)

    # Classify and route
    routing = await router.route(request.question, {"user": user})

    # Check if human escalation needed
    if routing["requires_human"]:
        return AnswerResponse(
            answer=f"This question requires human assistance. Please contact {routing['escalation_channel']}.",
            sources=[],
            confidence=1.0,
            department=routing["department"],
            action_buttons=[{
                "label": "Create Ticket",
                "action": "create_ticket",
                "url": "/support/new-ticket"
            }]
        )

    # Retrieve relevant documents
    docs = await retriever.retrieve(
        query=request.question,
        user=user,
        department=routing["department"]
    )

    if not docs:
        return AnswerResponse(
            answer="I couldn't find relevant information. Please try rephrasing or contact support.",
            sources=[],
            confidence=0.0,
            department=routing["department"],
            action_buttons=[]
        )

    # Generate response
    response = await generator.generate(
        question=request.question,
        context=docs,
        user_department=user.department,
        user_role=user.title
    )

    return AnswerResponse(
        answer=response.answer,
        sources=response.sources,
        confidence=response.confidence,
        department=routing["department"],
        action_buttons=response.action_buttons
    )

# Slack events endpoint
@app.post("/slack/events")
async def slack_events(request: Request):
    return await slack_handler.handle(request)

@app.get("/health")
async def health():
    return {"status": "healthy"}

# docker-compose.yml
version: '3.8'

services:
  knowledge-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - PINECONE_API_KEY=${PINECONE_API_KEY}
      - SLACK_BOT_TOKEN=${SLACK_BOT_TOKEN}
      - SLACK_SIGNING_SECRET=${SLACK_SIGNING_SECRET}
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  indexer:
    build: .
    command: python -m src.knowledge.indexer
    environment:
      - CONFLUENCE_URL=${CONFLUENCE_URL}
      - CONFLUENCE_TOKEN=${CONFLUENCE_TOKEN}
      - PINECONE_API_KEY=${PINECONE_API_KEY}