File size: 13,097 Bytes
e7586f8
 
 
 
 
 
 
 
 
 
 
 
 
6bc9349
 
799e82f
e7586f8
 
 
 
 
 
 
 
 
 
 
 
 
799e82f
ba8f1ce
e7586f8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ba8f1ce
e7586f8
11d48fe
6bc9349
11d48fe
 
e7586f8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3d20163
 
 
 
 
 
 
 
 
 
 
 
6bc9349
e7586f8
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
"""
FastAPI application for FinBot RAG system.
Exposes HTTP endpoints for chat, user management, and system diagnostics.
"""

import logging
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request
from typing import Optional
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel



from pipeline.rag_pipeline import get_rag_pipeline
from retrieval.user_auth import get_user_manager
from vector_store import get_vector_store
from ingestion.document_ingester import DocumentIngester
from config import DocumentCollection

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

# Logging setup remains as is


# ====================
# REQUEST/RESPONSE MODELS
# ====================

class ChatRequest(BaseModel):
    """Request model for chat endpoint."""
    user_role: str
    query: str
    user_id: str = None


class ChatResponse(BaseModel):
    """Response model for chat endpoint."""
    answer: str
    sources: list
    route: str
    user_role: str
    accessible_collections: list
    guardrail_flags: list = []
    guardrail_warnings: list = []
    rbac_denied: bool = False
    rbac_reason: Optional[str] = None


class UserInfo(BaseModel):
    """User information model."""
    username: str
    name: str
    role: str
    department: str
    accessible_collections: list[str] = []


class CollectionInfo(BaseModel):
    """Collection information model."""
    name: str
    description: str
    accessible_roles: list


# ====================
# INITIALIZATION
# ====================

async def startup_event():
    """Initialize application on startup."""
    logger.info("="*60)
    logger.info("FinBot RAG System Starting Up")
    logger.info("="*60)
    
    # Check for API key
    if not os.getenv("GROQ_API_KEY"):
        logger.warning("GROQ_API_KEY not set!  Chat functionality will fail.")
    
    # Initialize vector store and check collections
    vector_store = get_vector_store()
    collections = vector_store.list_collections()
    logger.info(f"Available collections: {collections if collections else 'None (ingestion pending)'}")
    
    logger.info("FinBot RAG System Ready")
    logger.info("="*60)


async def shutdown_event():
    """Cleanup on application shutdown."""
    logger.info("FinBot RAG System Shutting Down")


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifecycle."""
    await startup_event()
    yield
    await shutdown_event()


# ====================
# CREATE FASTAPI APP
# ====================

app = FastAPI(
    title="FinBot RAG API",
    description="Advanced RAG system with RBAC, hierarchical chunking, and guardrails",
    version="1.0.0",
    lifespan=lifespan,
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)







# ====================
# CHAT ENDPOINT
# ====================

@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """
    Process a user query through the RAG pipeline.
    
    Args:
        request: ChatRequest with user_role, query, and optional user_id
        
    Returns:
        ChatResponse with answer, sources, and metadata
    """
    try:
        # Validate user role
        valid_roles = ["employee", "finance", "engineering", "marketing", "c_level"]
        if request.user_role not in valid_roles:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid user role. Must be one of: {valid_roles}"
            )
        
        # Get RAG pipeline
        pipeline = get_rag_pipeline()
        
        # Process query
        rag_response = pipeline.answer_query(
            user_role=request.user_role,
            query_text=request.query,
            user_id=request.user_id,
        )
        print(rag_response)
        # Convert to response model
        return ChatResponse(
            answer=rag_response.answer,
            sources=rag_response.sources,
            route=rag_response.route,
            user_role=rag_response.user_role,
            accessible_collections=rag_response.accessible_collections,
            guardrail_flags=rag_response.guardrail_flags,
            guardrail_warnings=rag_response.guardrail_warnings,
            rbac_denied=rag_response.rbac_denied,
            rbac_reason=rag_response.rbac_reason,
        )
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error processing chat request: {str(e)}", exc_info=True)
        # Return a proper ChatResponse with error info instead of a 500,
        # so the frontend always has something to display.
        return ChatResponse(
            answer="I'm sorry, I encountered an unexpected error while processing your question. Please try again in a moment.",
            sources=[],
            route="error",
            user_role=request.user_role,
            accessible_collections=[],
            guardrail_flags=["server_error"],
            guardrail_warnings=[f"Internal error: {str(e)}"],
        )



# ====================
# USER MANAGEMENT ENDPOINTS
# ====================

@app.get("/api/users", response_model=list[UserInfo])
async def list_users():
    """Get list of demo users for login screen."""
    try:
        user_manager = get_user_manager()
        users = user_manager.list_users()
        return [
            UserInfo(
                username=u.username,
                name=u.name,
                role=u.role.value,  # Use .value to get "finance" not "UserRole.FINANCE"
                department=u.department,
                accessible_collections=user_manager.get_user_accessible_collections(
                    u.role.value
                ),
            )
            for u in users
        ]
    except Exception as e:
        logger.error(f"Error listing users: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")


@app.get("/api/users/{username}")
async def get_user(username: str):
    """Get specific user information."""
    try:
        user_manager = get_user_manager()
        user = user_manager.get_user(username)
        
        if not user:
            raise HTTPException(status_code=404, detail=f"User not found: {username}")
        
        return {
            "username": user.username,
            "name": user.name,
            "role": user.role,
            "department": user.department,
            "accessible_collections": user_manager.get_user_accessible_collections(user.role),
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error getting user: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")


# ====================
# COLLECTIONS ENDPOINTS
# ====================

@app.get("/api/collections", response_model=list[CollectionInfo])
async def list_collections():
    """Get list of document collections."""
    try:
        from config import COLLECTION_CONFIGS
        
        collections = []
        for coll_enum in DocumentCollection:
            config = COLLECTION_CONFIGS.get(coll_enum)
            if config:
                collections.append(
                    CollectionInfo(
                        name=coll_enum.value,
                        description=config.get("description", ""),
                        accessible_roles=config.get("access_roles", []),
                    )
                )
        
        return collections
    except Exception as e:
        logger.error(f"Error listing collections: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")


@app.get("/api/collections/{collection_name}")
async def get_collection_info(collection_name: str):
    """Get information about a specific collection."""
    try:
        from config import COLLECTION_CONFIGS
        
        # Find collection
        coll = None
        for c in DocumentCollection:
            if c.value == collection_name:
                coll = c
                break
        
        if not coll:
            raise HTTPException(status_code=404, detail=f"Collection not found: {collection_name}")
        
        config = COLLECTION_CONFIGS.get(coll)
        
        # Get vector store stats
        vector_store = get_vector_store()
        stats = vector_store.get_collection_stats(collection_name)
        
        return {
            "name": collection_name,
            "description": config.get("description", ""),
            "accessible_roles": config.get("access_roles", []),
            "chunks_count": stats.get("points_count", 0) if stats else 0,
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error getting collection info: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")


# ====================
# INGESTION ENDPOINT (Admin)
# ====================

@app.post("/api/admin/ingest")
async def ingest_documents():
    """
    Ingest all document collections.
    WARNING: Only use for demo/testing!
    """
    try:
        logger.info("Starting document ingestion...")
        
        ingester = DocumentIngester()
        results = ingester.ingest_all_collections()
        
        stats = ingester.verify_ingestion()
        
        return {
            "status": "success",
            "ingestion_results": results,
            "collection_stats": stats,
        }
    except Exception as e:
        logger.error(f"Error ingesting documents: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Ingestion failed: {str(e)}")


# ====================
# SYSTEM ENDPOINTS
# ====================

@app.get("/api/health")
async def health_check():
    """Health check endpoint."""
    try:
        vector_store = get_vector_store()
        collections = vector_store.list_collections()
        
        return {
            "status": "healthy",
            "collections_available": len(collections) > 0,
            "collections": collections,
        }
    except Exception as e:
        logger.error(f"Health check failed: {str(e)}")
        return JSONResponse(
            status_code=503,
            content={
                "status": "unhealthy",
                "error": str(e),
            },
        )


@app.get("/api/info")
async def system_info():
    """Get system information."""
    try:
        return {
            "name": "FinBot RAG System",
            "version": "1.0.0",
            "features": [
                "Role-Based Access Control (RBAC)",
                "Hierarchical Document Chunking",
                "Semantic Query Routing",
                "Input/Output Guardrails",
                "RAGAs Evaluation Support",
            ],
            "available_roles": ["employee", "finance", "engineering", "marketing", "c_level"],
        }
    except Exception as e:
        logger.error(f"Error getting system info: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")


# ====================
# ERROR HANDLERS
# ====================

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """Handle HTTP exceptions."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.detail,
            "status_code": exc.status_code,
        },
    )


@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Handle general exceptions."""
    logger.error(f"Unhandled exception: {str(exc)}")
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "detail": str(exc),
        },
    )


# ====================
# ROOT ENDPOINT
# ====================

@app.get("/")
async def root():
    """Root endpoint with API documentation."""
    return {
        "name": "FinBot RAG API",
        "version": "1.0.0",
        "description": "Advanced RAG system with RBAC, hierarchical chunking, and guardrails",
        "endpoints": {
            "chat": "POST /api/chat - Process a user query",
            "users": "GET /api/users - List demo users",
            "collections": "GET /api/collections - List document collections",
            "health": "GET /api/health - Health check",
            "info": "GET /api/info - System information",
            "ingest": "POST /api/admin/ingest - Ingest documents (admin only)",
        },
        "documentation": "/docs",
    }


if __name__ == "__main__":
    import uvicorn
    
    logger.info("Starting FinBot RAG API server...")
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info",
    )