Upload agentic_patterns.py with huggingface_hub

agentic_patterns.py

@@ -0,0 +1,631 @@
+"""
+Agentic AI Patterns Implementation for Scientific Q&A
+Based on 5 proven agentic patterns for better LLM performance
+"""
+
+import asyncio
+import json
+import re
+from typing import Dict, List, Any, Optional, Tuple
+from openai import OpenAI
+import requests
+import time
+from datetime import datetime
+
+class OpenAlexTool:
+    """OpenAlex API integration for scientific literature search"""
+    
+    def __init__(self):
+        self.base_url = "https://api.openalex.org"
+        self.session = requests.Session()
+        self.session.headers.update({
+            'User-Agent': 'AgenticAI-ScientificQA/1.0 (mailto:research@university.edu)'
+        })
+    
+    def search_papers(self, query: str, max_results: int = 10) -> List[Dict]:
+        """Search for scientific papers using OpenAlex"""
+        try:
+            url = f"{self.base_url}/works"
+            params = {
+                'search': query,
+                'per-page': min(max_results, 25),
+                'sort': 'relevance_score:desc'
+            }
+            
+            time.sleep(0.1)  # Rate limiting
+            response = self.session.get(url, params=params, timeout=30)
+            
+            if response.status_code == 200:
+                data = response.json()
+                results = []
+                
+                for work in data.get('results', []):
+                    processed_work = {
+                        'title': work.get('display_name', ''),
+                        'year': work.get('publication_year'),
+                        'citations': work.get('cited_by_count', 0),
+                        'doi': work.get('doi', ''),
+                        'authors': self._extract_authors(work.get('authorships', [])),
+                        'abstract': self._reconstruct_abstract(work.get('abstract_inverted_index', {})),
+                        'url': work.get('id', ''),
+                        'open_access': work.get('open_access', {}).get('is_oa', False)
+                    }
+                    results.append(processed_work)
+                
+                return results
+            else:
+                return []
+                
+        except Exception as e:
+            print(f"OpenAlex search error: {e}")
+            return []
+    
+    def _extract_authors(self, authorships: List[Dict]) -> List[str]:
+        """Extract author names from authorships"""
+        authors = []
+        for authorship in authorships[:3]:  # First 3 authors
+            author = authorship.get('author', {})
+            name = author.get('display_name', 'Unknown')
+            authors.append(name)
+        return authors
+    
+    def _reconstruct_abstract(self, inverted_index: Dict) -> str:
+        """Reconstruct abstract from inverted index"""
+        if not inverted_index:
+            return ""
+        
+        word_positions = []
+        for word, positions in inverted_index.items():
+            for pos in positions:
+                word_positions.append((pos, word))
+        
+        word_positions.sort()
+        abstract_words = [word for _, word in word_positions]
+        
+        return ' '.join(abstract_words)[:500]  # Limit length
+
+
+class AgenticPatterns:
+    """Implementation of 5 Agentic AI Patterns for Scientific Q&A"""
+    
+    def __init__(self, api_key: str):
+        try:
+            self.client = OpenAI(api_key=api_key)
+        except Exception as e:
+            # Fallback for older OpenAI versions
+            import openai
+            openai.api_key = api_key
+            self.client = openai
+        self.openalex = OpenAlexTool()
+    
+    async def pure_llm(self, question: str) -> Dict[str, Any]:
+        """Pattern 0: Pure LLM without any agentic enhancement"""
+        start_time = time.time()
+        
+        try:
+            response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a helpful scientific assistant. Answer questions clearly and accurately."},
+                    {"role": "user", "content": question}
+                ],
+                temperature=0.7,
+                max_tokens=1000
+            )
+            
+            answer = response.choices[0].message.content
+            processing_time = time.time() - start_time
+            
+            return {
+                "pattern": "Pure LLM",
+                "answer": answer,
+                "processing_time": f"{processing_time:.2f}s",
+                "steps": ["Direct LLM response"],
+                "confidence": "N/A",
+                "sources": []
+            }
+            
+        except Exception as e:
+            return {
+                "pattern": "Pure LLM",
+                "answer": f"Error: {str(e)}",
+                "processing_time": "0s",
+                "steps": ["Error occurred"],
+                "confidence": "N/A",
+                "sources": []
+            }
+    
+    async def reflection_pattern(self, question: str) -> Dict[str, Any]:
+        """Pattern 1: Reflection - Agent checks its own work"""
+        start_time = time.time()
+        steps = []
+        
+        try:
+            # Step 1: Initial response
+            steps.append("Generating initial response")
+            initial_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a scientific assistant. Provide a detailed answer to the question."},
+                    {"role": "user", "content": question}
+                ],
+                temperature=0.7,
+                max_tokens=800
+            )
+            
+            initial_answer = initial_response.choices[0].message.content
+            
+            # Step 2: Self-reflection and critique
+            steps.append("Reflecting on initial response")
+            reflection_prompt = f"""
+            Review this scientific answer for accuracy, completeness, and clarity:
+            
+            Question: {question}
+            Answer: {initial_answer}
+            
+            Provide:
+            1. What's good about this answer
+            2. What could be improved
+            3. Any potential errors or missing information
+            4. Confidence score (1-10)
+            """
+            
+            reflection_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a critical reviewer of scientific content."},
+                    {"role": "user", "content": reflection_prompt}
+                ],
+                temperature=0.3,
+                max_tokens=500
+            )
+            
+            reflection = reflection_response.choices[0].message.content
+            
+            # Step 3: Improved response based on reflection
+            steps.append("Generating improved response")
+            improvement_prompt = f"""
+            Based on this reflection, provide an improved answer:
+            
+            Original Question: {question}
+            Original Answer: {initial_answer}
+            Reflection: {reflection}
+            
+            Provide a refined, more accurate answer:
+            """
+            
+            final_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a scientific assistant providing refined answers."},
+                    {"role": "user", "content": improvement_prompt}
+                ],
+                temperature=0.5,
+                max_tokens=1000
+            )
+            
+            final_answer = final_response.choices[0].message.content
+            
+            # Extract confidence score
+            confidence_match = re.search(r'confidence.*?(\d+)', reflection.lower())
+            confidence = confidence_match.group(1) if confidence_match else "7"
+            
+            processing_time = time.time() - start_time
+            
+            return {
+                "pattern": "Reflection",
+                "answer": final_answer,
+                "processing_time": f"{processing_time:.2f}s",
+                "steps": steps,
+                "confidence": f"{confidence}/10",
+                "reflection": reflection,
+                "sources": []
+            }
+            
+        except Exception as e:
+            return {
+                "pattern": "Reflection",
+                "answer": f"Error: {str(e)}",
+                "processing_time": "0s",
+                "steps": steps,
+                "confidence": "N/A",
+                "sources": []
+            }
+    
+    async def planning_pattern(self, question: str) -> Dict[str, Any]:
+        """Pattern 2: Planning - Agent creates and follows a plan"""
+        start_time = time.time()
+        steps = []
+        
+        try:
+            # Step 1: Create a plan
+            steps.append("Creating research plan")
+            planning_prompt = f"""
+            Create a step-by-step research plan to answer this scientific question:
+            {question}
+            
+            Provide a numbered list of 3-5 specific steps to thoroughly research and answer this question.
+            Each step should be actionable and specific.
+            """
+            
+            plan_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a research planner. Create systematic research plans."},
+                    {"role": "user", "content": planning_prompt}
+                ],
+                temperature=0.3,
+                max_tokens=400
+            )
+            
+            plan = plan_response.choices[0].message.content
+            
+            # Step 2: Execute the plan
+            steps.append("Executing research plan")
+            execution_prompt = f"""
+            Follow this research plan to answer the question:
+            
+            Question: {question}
+            Plan: {plan}
+            
+            Execute each step systematically and provide a comprehensive answer based on your planned approach.
+            """
+            
+            execution_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a systematic researcher following a detailed plan."},
+                    {"role": "user", "content": execution_prompt}
+                ],
+                temperature=0.6,
+                max_tokens=1200
+            )
+            
+            answer = execution_response.choices[0].message.content
+            
+            processing_time = time.time() - start_time
+            
+            return {
+                "pattern": "Planning",
+                "answer": answer,
+                "processing_time": f"{processing_time:.2f}s",
+                "steps": steps,
+                "plan": plan,
+                "confidence": "8/10",
+                "sources": []
+            }
+            
+        except Exception as e:
+            return {
+                "pattern": "Planning",
+                "answer": f"Error: {str(e)}",
+                "processing_time": "0s",
+                "steps": steps,
+                "confidence": "N/A",
+                "sources": []
+            }
+    
+    async def tool_use_pattern(self, question: str) -> Dict[str, Any]:
+        """Pattern 3: Tool Use - Agent uses OpenAlex to find relevant papers"""
+        start_time = time.time()
+        steps = []
+        
+        try:
+            # Step 1: Analyze question for search terms
+            steps.append("Extracting search terms")
+            search_prompt = f"""
+            Extract 2-3 key scientific search terms from this question for academic paper search:
+            {question}
+            
+            Provide only the search terms, separated by spaces, optimized for finding relevant scientific papers.
+            """
+            
+            search_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You extract optimal search terms for scientific literature."},
+                    {"role": "user", "content": search_prompt}
+                ],
+                temperature=0.2,
+                max_tokens=50
+            )
+            
+            search_terms = search_response.choices[0].message.content.strip()
+            
+            # Step 2: Search for relevant papers
+            steps.append(f"Searching papers with: '{search_terms}'")
+            papers = self.openalex.search_papers(search_terms, max_results=5)
+            
+            # Step 3: Synthesize answer with paper evidence
+            steps.append("Synthesizing answer with literature evidence")
+            
+            papers_context = ""
+            if papers:
+                papers_context = "\n\nRelevant Research Papers:\n"
+                for i, paper in enumerate(papers[:3], 1):
+                    papers_context += f"{i}. {paper['title']} ({paper['year']}) - {paper['citations']} citations\n"
+                    if paper['abstract']:
+                        papers_context += f"   Abstract: {paper['abstract'][:200]}...\n"
+            
+            synthesis_prompt = f"""
+            Answer this scientific question using the provided research papers as evidence:
+            
+            Question: {question}
+            {papers_context}
+            
+            Provide a comprehensive answer that:
+            1. Directly addresses the question
+            2. References relevant findings from the papers
+            3. Acknowledges any limitations in current research
+            """
+            
+            synthesis_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a scientific researcher synthesizing evidence from literature."},
+                    {"role": "user", "content": synthesis_prompt}
+                ],
+                temperature=0.6,
+                max_tokens=1200
+            )
+            
+            answer = synthesis_response.choices[0].message.content
+            
+            processing_time = time.time() - start_time
+            
+            return {
+                "pattern": "Tool Use (OpenAlex)",
+                "answer": answer,
+                "processing_time": f"{processing_time:.2f}s",
+                "steps": steps,
+                "search_terms": search_terms,
+                "papers_found": len(papers),
+                "confidence": "9/10" if papers else "6/10",
+                "sources": [{"title": p["title"], "year": p["year"], "citations": p["citations"]} for p in papers[:3]]
+            }
+            
+        except Exception as e:
+            return {
+                "pattern": "Tool Use (OpenAlex)",
+                "answer": f"Error: {str(e)}",
+                "processing_time": "0s",
+                "steps": steps,
+                "confidence": "N/A",
+                "sources": []
+            }
+    
+    async def multi_agent_pattern(self, question: str) -> Dict[str, Any]:
+        """Pattern 4: Multi-Agent - Multiple specialized agents collaborate"""
+        start_time = time.time()
+        steps = []
+        
+        try:
+            # Agent 1: Research Specialist
+            steps.append("Research Specialist analyzing question")
+            research_prompt = f"""
+            As a Research Specialist, analyze this scientific question and provide:
+            1. Key research areas involved
+            2. Important concepts to address
+            3. Potential methodologies relevant to the question
+            
+            Question: {question}
+            """
+            
+            research_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a Research Specialist with expertise in scientific methodology."},
+                    {"role": "user", "content": research_prompt}
+                ],
+                temperature=0.4,
+                max_tokens=400
+            )
+            
+            research_analysis = research_response.choices[0].message.content
+            
+            # Agent 2: Domain Expert
+            steps.append("Domain Expert providing specialized knowledge")
+            expert_prompt = f"""
+            As a Domain Expert, provide specialized knowledge for this question:
+            {question}
+            
+            Research Analysis: {research_analysis}
+            
+            Focus on:
+            1. Current state of knowledge in this area
+            2. Key findings and established facts
+            3. Recent developments or controversies
+            """
+            
+            expert_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a Domain Expert with deep specialized knowledge."},
+                    {"role": "user", "content": expert_prompt}
+                ],
+                temperature=0.5,
+                max_tokens=500
+            )
+            
+            expert_knowledge = expert_response.choices[0].message.content
+            
+            # Agent 3: Critical Reviewer
+            steps.append("Critical Reviewer evaluating perspectives")
+            reviewer_prompt = f"""
+            As a Critical Reviewer, evaluate these perspectives on the question:
+            {question}
+            
+            Research Analysis: {research_analysis}
+            Expert Knowledge: {expert_knowledge}
+            
+            Provide:
+            1. Strengths of each perspective
+            2. Potential gaps or limitations
+            3. Areas of uncertainty
+            """
+            
+            reviewer_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a Critical Reviewer who evaluates scientific arguments."},
+                    {"role": "user", "content": reviewer_prompt}
+                ],
+                temperature=0.3,
+                max_tokens=400
+            )
+            
+            critical_review = reviewer_response.choices[0].message.content
+            
+            # Agent 4: Synthesizer
+            steps.append("Synthesizer creating final answer")
+            synthesis_prompt = f"""
+            Synthesize all agent perspectives into a comprehensive answer:
+            
+            Question: {question}
+            
+            Research Specialist: {research_analysis}
+            Domain Expert: {expert_knowledge}
+            Critical Reviewer: {critical_review}
+            
+            Provide a balanced, comprehensive answer that incorporates insights from all agents.
+            """
+            
+            synthesis_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a Synthesizer who combines multiple expert perspectives."},
+                    {"role": "user", "content": synthesis_prompt}
+                ],
+                temperature=0.6,
+                max_tokens=1000
+            )
+            
+            final_answer = synthesis_response.choices[0].message.content
+            
+            processing_time = time.time() - start_time
+            
+            return {
+                "pattern": "Multi-Agent",
+                "answer": final_answer,
+                "processing_time": f"{processing_time:.2f}s",
+                "steps": steps,
+                "agents_used": ["Research Specialist", "Domain Expert", "Critical Reviewer", "Synthesizer"],
+                "confidence": "9/10",
+                "sources": []
+            }
+            
+        except Exception as e:
+            return {
+                "pattern": "Multi-Agent",
+                "answer": f"Error: {str(e)}",
+                "processing_time": "0s",
+                "steps": steps,
+                "confidence": "N/A",
+                "sources": []
+            }
+    
+    async def chain_of_thought_pattern(self, question: str) -> Dict[str, Any]:
+        """Pattern 5: Chain of Thought - Step-by-step reasoning"""
+        start_time = time.time()
+        steps = []
+        
+        try:
+            steps.append("Breaking down question into reasoning steps")
+            
+            cot_prompt = f"""
+            Answer this scientific question using step-by-step reasoning. Show your thinking process clearly.
+            
+            Question: {question}
+            
+            Please think through this step by step:
+            
+            Step 1: Understand what the question is asking
+            Step 2: Identify the key scientific concepts involved
+            Step 3: Consider what we know about these concepts
+            Step 4: Apply logical reasoning to connect the concepts
+            Step 5: Draw conclusions based on the reasoning
+            Step 6: Consider any limitations or uncertainties
+            
+            Show your reasoning for each step, then provide a final answer.
+            """
+            
+            cot_response = self.client.chat.completions.create(
+                model="gpt-4o-mini",
+                messages=[
+                    {"role": "system", "content": "You are a scientific reasoner who thinks step-by-step and shows your reasoning process clearly."},
+                    {"role": "user", "content": cot_prompt}
+                ],
+                temperature=0.4,
+                max_tokens=1500
+            )
+            
+            answer = cot_response.choices[0].message.content
+            
+            processing_time = time.time() - start_time
+            
+            return {
+                "pattern": "Chain of Thought",
+                "answer": answer,
+                "processing_time": f"{processing_time:.2f}s",
+                "steps": steps,
+                "reasoning_approach": "Step-by-step logical reasoning",
+                "confidence": "8/10",
+                "sources": []
+            }
+            
+        except Exception as e:
+            return {
+                "pattern": "Chain of Thought",
+                "answer": f"Error: {str(e)}",
+                "processing_time": "0s",
+                "steps": steps,
+                "confidence": "N/A",
+                "sources": []
+            }
+    
+    async def run_all_patterns(self, question: str) -> Dict[str, Any]:
+        """Run all patterns simultaneously for comparison"""
+        start_time = time.time()
+        
+        # Run all patterns concurrently
+        tasks = [
+            self.pure_llm(question),
+            self.reflection_pattern(question),
+            self.planning_pattern(question),
+            self.tool_use_pattern(question),
+            self.multi_agent_pattern(question),
+            self.chain_of_thought_pattern(question)
+        ]
+        
+        results = await asyncio.gather(*tasks)
+        
+        total_time = time.time() - start_time
+        
+        return {
+            "comparison_mode": True,
+            "total_processing_time": f"{total_time:.2f}s",
+            "results": {
+                "pure_llm": results[0],
+                "reflection": results[1],
+                "planning": results[2],
+                "tool_use": results[3],
+                "multi_agent": results[4],
+                "chain_of_thought": results[5]
+            },
+            "summary": {
+                "fastest": min(results, key=lambda x: float(x["processing_time"].replace('s', '')) if x["processing_time"] != "0s" else float('inf'))["pattern"],
+                "highest_confidence": max(results, key=lambda x: int(x["confidence"].split('/')[0]) if x["confidence"] != "N/A" else 0)["pattern"],
+                "most_sources": max(results, key=lambda x: len(x.get("sources", [])))["pattern"]
+            }
+        }
+
+
+# Example questions for testing
+EXAMPLE_QUESTIONS = [
+    "What are the main mechanisms behind CRISPR-Cas9 gene editing and what are its current limitations?",
+    "How does climate change affect ocean acidification and marine ecosystems?",
+    "What is the relationship between gut microbiome diversity and human immune system function?",
+    "Explain the current understanding of dark matter and dark energy in cosmology.",
+    "What are the key differences between mRNA vaccines and traditional vaccines in terms of mechanism and efficacy?"
+]
+