Elrond MCP

""" Critique agents and synthesis logic using instructor and Google AI. This module implements the core thinking augmentation system with specialized agents for different critique perspectives and a synthesis agent that combines all perspectives into a coherent analysis. """ import asyncio import logging from datetime import UTC, datetime from pydantic import ValidationError from .client import get_critique_client, get_synthesis_client from .models import ( CritiquePerspective, CritiqueRequest, CritiqueResponse, SynthesisResponse, ThinkingAugmentationResult, ) logger = logging.getLogger(__name__) class CritiqueAgent: """Base class for critique agents with different perspectives.""" def __init__( self, perspective: CritiquePerspective, model_name: str = "gemini-2.5-flash", ): self.perspective = perspective self.model_name = model_name # Get instructor client with Google AI self.client = get_critique_client(model_name) def _get_system_prompt(self) -> str: """Get the system prompt for this critique perspective.""" base_prompt = f"""You are an expert analyst tasked with providing a {self.perspective} critique of proposals. Your role is to thoroughly analyze the proposal from a {self.perspective} perspective, focusing on: - Providing balanced, evidence-based analysis - Identifying key strengths, weaknesses, or neutral observations - Considering multiple stakeholders and scenarios - Being specific and actionable in your insights You must provide structured analysis covering feasibility, risks, benefits, implementation, stakeholder impact, and resource requirements. Your confidence level should reflect the certainty of your analysis based on the information provided.""" perspective_specific = { "positive": """ Focus on: - Strengths and advantages of the proposal - Potential for positive outcomes and success - Opportunities and benefits for stakeholders - Ways the proposal could exceed expectations - Innovative or particularly strong aspects While maintaining objectivity, emphasize the constructive and optimistic aspects of the analysis.""", "negative": """ Focus on: - Potential weaknesses and vulnerabilities - Risks, challenges, and obstacles - Unintended consequences and downsides - Resource constraints and limitations - Areas where the proposal might fall short While being constructively critical, identify genuine concerns and potential failure modes.""", "neutral": """ Focus on: - Objective, balanced assessment - Trade-offs and competing considerations - Factual analysis without bias toward success or failure - Realistic expectations and likely outcomes - Areas requiring more information or clarification Provide an impartial, evidence-based perspective that considers both positives and negatives.""", } return base_prompt + "\n\n" + perspective_specific[self.perspective] async def analyze_proposal(self, proposal: str) -> CritiqueResponse: """Analyze a proposal from this agent's perspective.""" try: system_prompt = self._get_system_prompt() user_prompt = f"Please analyze the following proposal:\n\n{proposal}" # Use instructor to get structured response response = await self.client.chat.completions.create( response_model=CritiqueResponse, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], generation_config={ "temperature": 0.7, "max_tokens": 2000, }, ) return response except ValidationError as e: logger.error(f"Validation error in {self.perspective} critique: {e}") raise except Exception as e: logger.error(f"Error in {self.perspective} critique: {e}") raise class SynthesisAgent: """Agent responsible for synthesizing multiple critiques into a coherent analysis.""" def __init__(self, model_name: str = "gemini-2.5-pro"): self.model_name = model_name # Get instructor client with Google AI self.client = get_synthesis_client(model_name) def _get_system_prompt(self) -> str: """Get the system prompt for synthesis.""" return """You are an expert synthesis analyst responsible for combining multiple critique perspectives into a coherent, actionable analysis. Your role is to: - Identify areas of consensus and disagreement among the critiques - Synthesize insights from positive, neutral, and negative perspectives - Provide a balanced, evidence-based recommendation - Highlight critical considerations and next steps - Flag areas of significant uncertainty You should: - Weight insights based on their validity and supporting evidence - Acknowledge trade-offs and competing priorities - Provide actionable recommendations and next steps - Be honest about limitations and uncertainties - Focus on practical outcomes and decision-making Your synthesis should help decision-makers understand the full picture and make informed choices.""" async def synthesize_critiques( self, original_proposal: str, critiques: list[CritiqueResponse] ) -> SynthesisResponse: """Synthesize multiple critiques into a comprehensive analysis.""" try: system_prompt = self._get_system_prompt() # Build critiques section critiques_text = "" for critique in critiques: key_insights_text = ", ".join(critique.key_insights) critiques_text += f""" **{critique.perspective.title()} Perspective:** Executive Summary: {critique.executive_summary} Analysis: - Feasibility: {critique.analysis.feasibility} - Risks: {critique.analysis.risks} - Benefits: {critique.analysis.benefits} - Implementation: {critique.analysis.implementation} - Stakeholder Impact: {critique.analysis.stakeholder_impact} - Resource Requirements: {critique.analysis.resource_requirements} Key Insights: {key_insights_text} Confidence: {critique.confidence_level} """ user_prompt = f""" Please synthesize the following critiques of this proposal: **Original Proposal:** {original_proposal} **Critiques:** {critiques_text} Provide a comprehensive synthesis that identifies consensus, disagreements, and actionable recommendations. """ # Use instructor to get structured response response = await self.client.chat.completions.create( response_model=SynthesisResponse, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], generation_config={ # Lower temperature for more consistent synthesis "temperature": 0.3, "max_tokens": 3000, }, ) return response except ValidationError as e: logger.error(f"Validation error in synthesis: {e}") raise except Exception as e: logger.error(f"Error in synthesis: {e}") raise async def process_proposal(request: CritiqueRequest) -> ThinkingAugmentationResult: """Process a proposal through the complete thinking augmentation pipeline.""" start_time = datetime.now(UTC) try: logger.info("Starting thinking augmentation process") # Create agents for this processing request positive_agent = CritiqueAgent("positive") neutral_agent = CritiqueAgent("neutral") negative_agent = CritiqueAgent("negative") synthesis_agent = SynthesisAgent() # Run all critiques in parallel critique_tasks = [ positive_agent.analyze_proposal(request.proposal), neutral_agent.analyze_proposal(request.proposal), negative_agent.analyze_proposal(request.proposal), ] logger.info("Running parallel critiques") critiques = await asyncio.gather(*critique_tasks) logger.info("Critiques completed, starting synthesis") # Synthesize all critiques synthesis = await synthesis_agent.synthesize_critiques( request.proposal, critiques ) end_time = datetime.now(UTC) processing_time = (end_time - start_time).total_seconds() logger.info(f"Thinking augmentation completed in {processing_time:.2f} seconds") # Create final result result = ThinkingAugmentationResult( original_proposal=request.proposal, critiques=critiques, synthesis=synthesis, processing_metadata={ "start_time": start_time.isoformat(), "end_time": end_time.isoformat(), "processing_time_seconds": processing_time, "critique_model": "gemini-2.5-flash", "synthesis_model": "gemini-2.5-pro", "num_critiques": len(critiques), }, ) return result except Exception as e: logger.error(f"Error in thinking augmentation process: {e}") raise # Convenience function for easy access async def consult_the_council(proposal: str) -> ThinkingAugmentationResult: """Convenience function to consult the council for thinking augmentation.""" request = CritiqueRequest(proposal=proposal) return await process_proposal(request)