Merging

__init__.py

@@ -1 +1 @@
-"""Meeting Assistant Multi-Agent System""" 
+"""Meeting Assistant Multi-Agent System"""

action_item_extraction_agent.py

@@ -1,5 +1,6 @@
 import json
 import re
+
 import openai
 
 
@@ -9,61 +10,55 @@ def __init__(self, api_key=None, model="gpt-3.5-turbo"):
         self.model = model
         # Regular expressions for simple action item extraction
         self.action_keywords = [
-            r'(?:need to|must|should|will|going to|have to|shall) ([^.!?]*)',
-            r'(?:action item|task|todo|to-do|to do|follow-up|followup)[:\s]* ([^.!?]*)',
-            r'(\w+)(?:\s*will|\s*is going to|\s*needs to|\s*must) ([^.!?]*)',
-            r'(?:by|before|due)(?:\s*the)?\s*(\d{1,2}(?:st|nd|rd|th)?\s+(?:of\s+)?(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)|tomorrow|next week|(?:this|next) month|(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday))',
+            r"(?:need to|must|should|will|going to|have to|shall) ([^.!?]*)",
+            r"(?:action item|task|todo|to-do|to do|follow-up|followup)[:\s]* ([^.!?]*)",
+            r"(\w+)(?:\s*will|\s*is going to|\s*needs to|\s*must) ([^.!?]*)",
+            r"(?:by|before|due)(?:\s*the)?\s*(\d{1,2}(?:st|nd|rd|th)?\s+(?:of\s+)?(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)|tomorrow|next week|(?:this|next) month|(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday))",
         ]
-    
+
     def extract_action_items(self, transcription, summary=None):
         """
         Extract action items from meeting transcription and/or summary.
-        
+
         Args:
             transcription (dict): Transcription data from the TranscriptionAgent
             summary (dict, optional): Summary data from the SummarizationAgent
-            
+
         Returns:
             dict: Extraction result with list of action items and metadata
         """
         print("ActionItemExtractionAgent: Extracting action items")
-        
+
         try:
             # Use both transcription and summary if available
             text = transcription.get("transcription", "")
-            
+
             if summary and "summary" in summary:
                 text += "\n\n" + summary["summary"]
-            
+
             if not text:
                 raise ValueError("No text provided for action item extraction")
-            
+
             # Extract action items using provided method
             action_items = self._extract_action_items(text)
-            
+
             return {
                 "action_items": action_items,
-                "metadata": {
-                    "items_found": len(action_items),
-                    "status": "completed"
-                }
+                "metadata": {"items_found": len(action_items), "status": "completed"},
             }
-            
+
         except Exception as e:
             print(f"Error during action item extraction: {str(e)}")
             return {
                 "action_items": [],
-                "metadata": {
-                    "status": "error",
-                    "error": str(e)
-                }
+                "metadata": {"status": "error", "error": str(e)},
             }
-    
+
     def _extract_action_items(self, text):
         """
         Extract action items from the provided text.
-        
-        This is a placeholder method. In a production environment, 
+
+        This is a placeholder method. In a production environment,
         this would use more sophisticated NLP techniques or an LLM.
         """
         # If API key is provided, use LLM for extraction
@@ -76,21 +71,27 @@ def _extract_action_items(self, text):
         else:
             # Fallback to regex-based extraction
             return self._extract_with_regex(text)
-    
+
     def _extract_with_llm(self, text):
         """Extract action items using an LLM"""
         client = openai.OpenAI(api_key=self.api_key)
-        
+
         response = client.chat.completions.create(
             model=self.model,
             messages=[
-                {"role": "system", "content": "You are a meeting assistant that extracts action items from meeting transcripts. Extract all tasks, responsibilities, and deadlines in a structured format."},
-                {"role": "user", "content": f"Extract all action items from this meeting transcript as a JSON array. Each action item should have 'task', 'assignee', and 'deadline' fields. Use null for missing information:\n\n{text}"}
+                {
+                    "role": "system",
+                    "content": "You are a meeting assistant that extracts action items from meeting transcripts. Extract all tasks, responsibilities, and deadlines in a structured format.",
+                },
+                {
+                    "role": "user",
+                    "content": f"Extract all action items from this meeting transcript as a JSON array. Each action item should have 'task', 'assignee', and 'deadline' fields. Use null for missing information:\n\n{text}",
+                },
             ],
             response_format={"type": "json_object"},
-            max_tokens=1000
+            max_tokens=1000,
         )
-        
+
         # Parse the response
         result_content = response.choices[0].message.content
         try:
@@ -99,53 +100,52 @@ def _extract_with_llm(self, text):
         except:
             # If parsing fails, return an empty list
             return []
-    
+
     def _extract_with_regex(self, text):
         """Extract action items using regular expressions"""
         action_items = []
-        sentences = re.split(r'[.!?]\s+', text)
-        
+        sentences = re.split(r"[.!?]\s+", text)
+
         for sentence in sentences:
             item = self._extract_from_sentence(sentence)
             if item and all(item != existing for existing in action_items):
                 action_items.append(item)
-        
+
         return action_items
-    
+
     def _extract_from_sentence(self, sentence):
         """Extract an action item from a single sentence using pattern matching"""
         sentence = sentence.strip()
         if not sentence:
             return None
-        
+
         # Look for action patterns
         task = None
         assignee = None
         deadline = None
-        
+
         # Find potential task
         for pattern in self.action_keywords:
             match = re.search(pattern, sentence, re.IGNORECASE)
             if match:
                 task = match.group(1).strip()
                 break
-        
+
         if not task:
             return None
-        
+
         # Try to find assignee - look for names followed by verbs
-        assignee_match = re.search(r'(\b[A-Z][a-z]+\b)(?:\s+will|\s+should|\s+is going to|\s+needs to)', sentence)
+        assignee_match = re.search(
+            r"(\b[A-Z][a-z]+\b)(?:\s+will|\s+should|\s+is going to|\s+needs to)",
+            sentence,
+        )
         if assignee_match:
             assignee = assignee_match.group(1)
-        
+
         # Try to find deadline
-        deadline_pattern = r'(?:by|before|due)(?:\s*the)?\s*(\d{1,2}(?:st|nd|rd|th)?\s+(?:of\s+)?(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)|tomorrow|next week|(?:this|next) month|(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday))'
+        deadline_pattern = r"(?:by|before|due)(?:\s*the)?\s*(\d{1,2}(?:st|nd|rd|th)?\s+(?:of\s+)?(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)|tomorrow|next week|(?:this|next) month|(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday))"
         deadline_match = re.search(deadline_pattern, sentence, re.IGNORECASE)
         if deadline_match:
             deadline = deadline_match.group(1)
-
-        return {
-            "task": task,
-            "assignee": assignee,
-            "deadline": deadline
-        } 
+
+        return {"task": task, "assignee": assignee, "deadline": deadline}

app.py

@@ -1,19 +1,26 @@
-import os
 import json
-from typing import Optional
-from fastapi import FastAPI, File, UploadFile, Form, Request
-from fastapi.responses import HTMLResponse, JSONResponse
-from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
+import os
 from pathlib import Path
+from typing import Any, Dict, Optional
+
 import uvicorn
 from dotenv import load_dotenv
+from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
 from orchestrator import MeetingAssistantOrchestrator
 
+from meeting_assistant.config import load_config
+
 # Load environment variables
 load_dotenv()
 
-app = FastAPI(title="Meeting Assistant")
+app = FastAPI(
+    title="Meeting Assistant API",
+    description="API for processing meeting recordings using multi-agent system",
+    version="1.0.0",
+)
 
 # Create directories for static files and templates
 BASE_DIR = Path(__file__).resolve().parent
@@ -32,71 +39,67 @@
 # Setup templates
 templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
 
+# Initialize the orchestrator
+config = load_config()
+orchestrator = MeetingAssistantOrchestrator(config)
+
+
 @app.get("/", response_class=HTMLResponse)
 async def home(request: Request):
     """Render the home page"""
-    return templates.TemplateResponse(
-        "index.html",
-        {"request": request}
-    )
-
-@app.post("/process")
-async def process_meeting(
-    file: UploadFile = File(...),
-    openai_api_key: Optional[str] = Form(None),
-    azure_speech_key: Optional[str] = Form(None)
-):
-    """Process a meeting recording"""
+    return templates.TemplateResponse("index.html", {"request": request})
+
+
+@app.post("/process-meeting")
+async def process_meeting(audio_file: UploadFile = File(...)) -> Dict[str, Any]:
+    """Process a meeting recording.
+
+    Args:
+        audio_file: The uploaded meeting audio file.
+
+    Returns:
+        Dict containing the processing results.
+
+    Raises:
+        HTTPException: If file upload or processing fails.
+    """
     try:
-        # Save the uploaded file
-        file_path = UPLOAD_DIR / file.filename
-        with open(file_path, "wb") as buffer:
-            content = await file.read()
-            buffer.write(content)
-
-        # Configure the orchestrator
-        config = {
-            "openai_api_key": openai_api_key or os.getenv("OPENAI_API_KEY"),
-            "azure_speech_key": azure_speech_key or os.getenv("AZURE_SPEECH_KEY")
-        }
-
+        # Save uploaded file
+        temp_path = f"temp/{audio_file.filename}"
+        with open(temp_path, "wb") as f:
+            content = await audio_file.read()
+            f.write(content)
+
         # Process the meeting
-        orchestrator = MeetingAssistantOrchestrator(config)
-        results = orchestrator.process_meeting(str(file_path))
-
-        # Generate report
-        report = orchestrator.generate_report(results)
-
-        # Save results and report with unique names based on timestamp
-        results_file = UPLOAD_DIR / f"results_{file.filename}.json"
-        report_file = UPLOAD_DIR / f"report_{file.filename}.md"
-
-        with open(results_file, "w") as f:
-            json.dump(results, f, indent=2)
-
-        with open(report_file, "w") as f:
-            f.write(report)
-
-        # Clean up the uploaded audio file
-        os.remove(file_path)
-
-        return JSONResponse({
-            "status": "success",
-            "message": "Meeting processed successfully",
-            "results": results,
-            "report": report
-        })
-
+        results = orchestrator.process_meeting(temp_path)
+
+        # Clean up
+        os.remove(temp_path)
+
+        return results
+
     except Exception as e:
-        return JSONResponse({
-            "status": "error",
-            "message": str(e)
-        }, status_code=500)
+        raise HTTPException(
+            status_code=500, detail=f"Failed to process meeting: {str(e)}"
+        )
+
 
 @app.get("/health")
-async def health_check():
-    """Health check endpoint"""
+async def health_check() -> Dict[str, str]:
+    """Check the health of the API.
+
+    Returns:
+        Dict containing status information.
+    """
     return {"status": "healthy"}
 
+
 if __name__ == "__main__":
-    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True) 
+    # Use environment variable for host in production, default to localhost
+    host = os.getenv("API_HOST", "127.0.0.1")
+    port = int(os.getenv("API_PORT", "8000"))
+
+    # Enable reload only in development
+    reload = os.getenv("API_ENV", "development").lower() == "development"
+
+    uvicorn.run("app:app", host=host, port=port, reload=reload)