""" 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?" ]