Spaces:
Sleeping
Sleeping
| """ | |
| src/ai/llm_service.py - Enhanced LLM service for interaction analysis | |
| """ | |
| from typing import Dict, Any, List | |
| from langchain_openai import ChatOpenAI | |
| from langchain_anthropic import ChatAnthropic | |
| from langchain.chains import create_extraction_chain, LLMChain | |
| from langchain.prompts import ChatPromptTemplate | |
| class LLMService: | |
| def __init__(self, openai_api_key: str, anthropic_api_key: str): | |
| """Initialize LLM service with required models""" | |
| self.setup_models(openai_api_key, anthropic_api_key) | |
| self.setup_prompts() | |
| self.setup_chains() | |
| def setup_models(self, openai_api_key: str, anthropic_api_key: str): | |
| """Setup LLM models""" | |
| self.summary_model = ChatAnthropic( | |
| api_key=anthropic_api_key, | |
| model="claude-3-haiku-20240307", | |
| temperature=0.7 | |
| ) | |
| self.extraction_model = ChatOpenAI( | |
| api_key=openai_api_key, | |
| model="gpt-4-turbo-preview" | |
| ) | |
| def setup_prompts(self): | |
| """Setup prompt templates for different analysis tasks""" | |
| self.entity_schema = { | |
| "properties": { | |
| "people": {"type": "array", "items": {"type": "string"}}, | |
| "organizations": {"type": "array", "items": {"type": "string"}}, | |
| "action_items": {"type": "array", "items": {"type": "string"}}, | |
| "key_points": {"type": "array", "items": {"type": "string"}}, | |
| "next_steps": {"type": "array", "items": {"type": "string"}}, | |
| "sentiment_indicators": {"type": "array", "items": {"type": "string"}} | |
| }, | |
| "required": ["action_items", "key_points", "next_steps"] | |
| } | |
| self.summary_template = """ | |
| Analyze this interaction and provide: | |
| 1. A concise summary of the key points discussed | |
| 2. Specific action items and their owners | |
| 3. Clear next steps and timeline | |
| 4. Notable concerns or opportunities mentioned | |
| 5. Overall sentiment of the interaction | |
| Focus on being specific and actionable. Include names and dates where mentioned. | |
| Interaction Content: | |
| {text} | |
| """ | |
| self.summary_prompt = ChatPromptTemplate.from_template(self.summary_template) | |
| def setup_chains(self): | |
| """Setup extraction and analysis chains""" | |
| self.extraction_chain = create_extraction_chain( | |
| self.entity_schema, | |
| self.extraction_model | |
| ) | |
| self.summary_chain = LLMChain( | |
| llm=self.summary_model, | |
| prompt=self.summary_prompt | |
| ) | |
| def analyze_interaction(self, text: str) -> Dict[str, Any]: | |
| """ | |
| Analyze interaction text to extract insights | |
| Args: | |
| text: Interaction text content | |
| Returns: | |
| Dictionary containing extracted insights and analysis | |
| """ | |
| # Extract entities and relationships | |
| extracted_info = self.extraction_chain.run(text) | |
| # Generate summary | |
| summary_result = self.summary_chain.run(text=text) | |
| # Calculate sentiment score based on indicators | |
| sentiment_indicators = extracted_info[0].get('sentiment_indicators', []) | |
| sentiment_score = self._calculate_sentiment(sentiment_indicators) | |
| return { | |
| 'summary': summary_result, | |
| 'key_points': extracted_info[0].get('key_points', []), | |
| 'action_items': extracted_info[0].get('action_items', []), | |
| 'next_steps': extracted_info[0].get('next_steps', []), | |
| 'entities': { | |
| 'people': extracted_info[0].get('people', []), | |
| 'organizations': extracted_info[0].get('organizations', []) | |
| }, | |
| 'sentiment_score': sentiment_score | |
| } | |
| def _calculate_sentiment(self, indicators: List[str]) -> float: | |
| """Calculate normalized sentiment score from indicators""" | |
| if not indicators: | |
| return 0.5 | |
| # Simple sentiment calculation | |
| positive_words = {'positive', 'excited', 'interested', 'agreed', 'satisfied'} | |
| negative_words = {'negative', 'concerned', 'unsure', 'disagreed', 'dissatisfied'} | |
| positive_count = sum(1 for ind in indicators if any(word in ind.lower() for word in positive_words)) | |
| negative_count = sum(1 for ind in indicators if any(word in ind.lower() for word in negative_words)) | |
| total = len(indicators) | |
| if total == 0: | |
| return 0.5 | |
| # Calculate normalized score between 0 and 1 | |
| score = (positive_count - negative_count) / total | |
| return (score + 1) / 2 # Normalize to 0-1 range |