""" MatchmakingAgent for Patent Wake-Up Scenario Matches patents with potential licensees, partners, and investors: - Semantic search in stakeholder database - Multi-dimensional match scoring - Geographic alignment (EU-Canada focus) - Generates match rationale and collaboration opportunities """ from typing import List, Optional from loguru import logger from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_core.messages import HumanMessage from ..base_agent import BaseAgent, Task from ...llm.langchain_ollama_client import LangChainOllamaClient from ...workflow.langgraph_state import ( PatentAnalysis, MarketAnalysis, StakeholderMatch ) class MatchmakingAgent(BaseAgent): """ Specialized agent for stakeholder matching. Uses semantic search and LLM reasoning to find best-fit partners. """ def __init__(self, llm_client: LangChainOllamaClient, memory_agent): """ Initialize MatchmakingAgent. Args: llm_client: LangChain Ollama client memory_agent: Memory agent (required for stakeholder search) """ # Note: MatchmakingAgent uses LangChain directly self.name = "MatchmakingAgent" self.description = "Stakeholder matching and partner identification" self.llm_client = llm_client self.memory_agent = memory_agent # Required if not memory_agent: raise ValueError("MatchmakingAgent requires memory_agent for stakeholder database") # Use complex reasoning for matching self.llm = llm_client.get_llm('complex') # qwen2.5:14b # Scoring chain self.scoring_chain = self._create_scoring_chain() # Ensure sample stakeholders exist self._stakeholders_initialized = False logger.info("Initialized MatchmakingAgent") def _create_scoring_chain(self): """Create chain for match scoring""" prompt = ChatPromptTemplate.from_messages([ ("system", "You are an expert in technology transfer and business development."), ("human", """ Evaluate the match quality between this patent and stakeholder: PATENT: - Title: {patent_title} - Technical Domains: {technical_domains} - Key Innovations: {key_innovations} - TRL: {trl_level} - Target Markets: {target_markets} STAKEHOLDER: - Name: {stakeholder_name} - Type: {stakeholder_type} - Expertise: {stakeholder_expertise} - Focus Sectors: {stakeholder_sectors} - Location: {stakeholder_location} Provide match assessment in JSON format: {{ "technical_fit": 0.85, "market_fit": 0.90, "geographic_fit": 1.0, "strategic_fit": 0.80, "overall_fit_score": 0.88, "match_rationale": "Detailed explanation of why this is a strong match", "collaboration_opportunities": ["Licensing", "Joint development", "Co-marketing"], "potential_value": "High/Medium/Low", "recommended_approach": "How to approach this stakeholder", "talking_points": ["Point 1", "Point 2", "Point 3"] }} Scoring guidelines: - technical_fit: Does stakeholder have expertise in this technology? - market_fit: Does stakeholder operate in target markets? - geographic_fit: Geographic alignment (EU/Canada priority) - strategic_fit: Overall strategic alignment - overall_fit_score: Weighted average (0-1) Return ONLY valid JSON. """) ]) parser = JsonOutputParser() return prompt | self.llm | parser async def find_matches( self, patent_analysis: PatentAnalysis, market_analysis: MarketAnalysis, max_matches: int = 10 ) -> List[StakeholderMatch]: """ Find best-fit stakeholders for patent commercialization. Args: patent_analysis: Patent technical details market_analysis: Market opportunities max_matches: Maximum number of matches to return Returns: List of StakeholderMatch objects ranked by fit score """ logger.info(f"🤝 Finding matches for: {patent_analysis.title}") # Ensure stakeholders are initialized if not self._stakeholders_initialized: await self._ensure_stakeholders() # Create search query from patent and market analysis query = self._create_search_query(patent_analysis, market_analysis) # Search stakeholder profiles in memory logger.info("Searching stakeholder database...") stakeholder_docs = await self.memory_agent.retrieve_relevant_context( query=query, context_type="stakeholders", top_k=max_matches * 2 # Get more for filtering ) logger.info(f"Found {len(stakeholder_docs)} potential stakeholders") # Score and rank matches matches = [] for doc in stakeholder_docs: try: stakeholder = self._parse_stakeholder(doc) match = await self._score_match( patent_analysis, market_analysis, stakeholder ) matches.append(match) except Exception as e: logger.warning(f"Failed to score match: {e}") continue # Sort by fit score and return top matches matches.sort(key=lambda x: x.overall_fit_score, reverse=True) logger.success(f"✅ Found {len(matches)} matches, returning top {max_matches}") return matches[:max_matches] def _create_search_query( self, patent: PatentAnalysis, market: MarketAnalysis ) -> str: """Create search query for stakeholder matching""" query_parts = [] # Add technical domains query_parts.extend(patent.technical_domains) # Add top market sectors query_parts.extend(market.top_sectors) # Add key innovations (first few words only) for innovation in patent.key_innovations[:2]: query_parts.append(innovation.split('.')[0]) return " ".join(query_parts) def _parse_stakeholder(self, doc) -> dict: """Parse stakeholder document into dict""" import json # Extract profile from metadata profile_json = doc.metadata.get('profile', '{}') profile = json.loads(profile_json) # Add page content for additional context profile['search_match_text'] = doc.page_content return profile async def _score_match( self, patent: PatentAnalysis, market: MarketAnalysis, stakeholder: dict ) -> StakeholderMatch: """ Score match quality using LLM reasoning. Args: patent: Patent analysis market: Market analysis stakeholder: Stakeholder profile dict Returns: StakeholderMatch with scores and rationale """ # Invoke scoring chain scoring = await self.scoring_chain.ainvoke({ "patent_title": patent.title, "technical_domains": ", ".join(patent.technical_domains), "key_innovations": ", ".join(patent.key_innovations[:3]), "trl_level": patent.trl_level, "target_markets": ", ".join(market.top_sectors), "stakeholder_name": stakeholder.get('name', 'Unknown'), "stakeholder_type": stakeholder.get('type', 'Unknown'), "stakeholder_expertise": ", ".join(stakeholder.get('expertise', [])), "stakeholder_sectors": ", ".join(stakeholder.get('focus_sectors', [])), "stakeholder_location": stakeholder.get('location', 'Unknown') }) # Build StakeholderMatch return StakeholderMatch( stakeholder_name=stakeholder.get('name', 'Unknown'), stakeholder_type=stakeholder.get('type', 'Unknown'), location=stakeholder.get('location', 'Unknown'), contact_info=stakeholder.get('contact_info'), overall_fit_score=scoring.get('overall_fit_score', 0.5), technical_fit=scoring.get('technical_fit', 0.5), market_fit=scoring.get('market_fit', 0.5), geographic_fit=scoring.get('geographic_fit', 0.5), strategic_fit=scoring.get('strategic_fit', 0.5), match_rationale=scoring.get('match_rationale', 'Match assessment'), collaboration_opportunities=scoring.get('collaboration_opportunities', []), potential_value=scoring.get('potential_value', 'Medium'), recommended_approach=scoring.get('recommended_approach', 'Professional outreach'), talking_points=scoring.get('talking_points', []) ) async def _ensure_stakeholders(self): """Ensure sample stakeholders exist in database""" # Check if stakeholders exist stats = self.memory_agent.get_collection_stats() if stats.get('stakeholders_count', 0) < 5: logger.info("Populating sample stakeholder database...") await self._populate_sample_stakeholders() self._stakeholders_initialized = True async def _populate_sample_stakeholders(self): """ Create sample stakeholder profiles for demonstration. In production, this would be populated from real databases. """ sample_stakeholders = [ { "name": "BioVentures Capital (Toronto)", "type": "Investor", "expertise": ["AI", "Machine Learning", "Drug Discovery", "Healthcare"], "focus_sectors": ["Pharmaceuticals", "Biotechnology", "Healthcare AI"], "location": "Toronto, Canada", "investment_stage": ["Seed", "Series A"], "description": "Early-stage deep tech venture capital focusing on AI-driven healthcare innovation" }, { "name": "EuroTech Licensing GmbH", "type": "Licensing Organization", "expertise": ["Materials Science", "Nanotechnology", "Energy", "Manufacturing"], "focus_sectors": ["Renewable Energy", "Advanced Materials", "Industrial IoT"], "location": "Munich, Germany", "description": "Technology licensing and commercialization across European markets" }, { "name": "McGill University Technology Transfer", "type": "University TTO", "expertise": ["Biomedical Engineering", "Software", "Clean Tech", "AI"], "focus_sectors": ["Healthcare", "Environmental Tech", "AI Applications"], "location": "Montreal, Canada", "description": "Academic technology transfer and industry partnerships" }, { "name": "PharmaTech Solutions Inc.", "type": "Company", "expertise": ["Drug Discovery", "Clinical Trials", "Regulatory Affairs"], "focus_sectors": ["Pharmaceuticals", "Biotechnology"], "location": "Basel, Switzerland", "description": "Pharmaceutical development and commercialization services" }, { "name": "Nordic Innovation Partners", "type": "Investor", "expertise": ["Clean Tech", "Sustainability", "Energy", "Manufacturing"], "focus_sectors": ["Renewable Energy", "Circular Economy", "Green Tech"], "location": "Stockholm, Sweden", "investment_stage": ["Series A", "Series B"], "description": "Impact investment in sustainable technologies" }, { "name": "Canadian AI Consortium", "type": "Industry Consortium", "expertise": ["AI", "Machine Learning", "Computer Vision", "NLP"], "focus_sectors": ["AI Applications", "Software", "Healthcare AI"], "location": "Vancouver, Canada", "description": "Collaborative AI research and commercialization network" }, { "name": "MedTech Innovators (Amsterdam)", "type": "Company", "expertise": ["Medical Devices", "Digital Health", "AI Diagnostics"], "focus_sectors": ["Healthcare", "Medical Technology"], "location": "Amsterdam, Netherlands", "description": "Medical technology development and distribution" }, { "name": "Quebec Pension Fund Technology", "type": "Investor", "expertise": ["Technology", "Healthcare", "Clean Tech", "AI"], "focus_sectors": ["Healthcare", "Clean Energy", "AI", "Manufacturing"], "location": "Montreal, Canada", "investment_stage": ["Series B", "Growth"], "description": "Large-scale technology investment fund" }, { "name": "European Patent Office Services", "type": "IP Services", "expertise": ["Patent Strategy", "IP Licensing", "Technology Transfer"], "focus_sectors": ["All Technology Sectors"], "location": "Munich, Germany", "description": "Patent commercialization and licensing support" }, { "name": "CleanTech Accelerator Berlin", "type": "Accelerator", "expertise": ["Clean Tech", "Sustainability", "Energy", "Materials"], "focus_sectors": ["Renewable Energy", "Environmental Tech", "Circular Economy"], "location": "Berlin, Germany", "description": "Accelerator program for sustainable technology startups" } ] # Store in memory for stakeholder in sample_stakeholders: try: await self.memory_agent.store_stakeholder_profile( name=stakeholder["name"], profile=stakeholder, categories=[stakeholder["type"]] + stakeholder["expertise"][:3] ) logger.debug(f"Stored stakeholder: {stakeholder['name']}") except Exception as e: logger.warning(f"Failed to store stakeholder {stakeholder['name']}: {e}") logger.success(f"✅ Populated {len(sample_stakeholders)} sample stakeholders") async def process_task(self, task: Task) -> Task: """ Process task using agent interface. Args: task: Task with patent_analysis and market_analysis in metadata Returns: Task with list of StakeholderMatch results """ task.status = "in_progress" try: patent_dict = task.metadata.get('patent_analysis') market_dict = task.metadata.get('market_analysis') if not patent_dict or not market_dict: raise ValueError("Both patent_analysis and market_analysis required") # Convert dicts to objects patent_analysis = PatentAnalysis(**patent_dict) market_analysis = MarketAnalysis(**market_dict) matches = await self.find_matches(patent_analysis, market_analysis) task.result = [m.model_dump() for m in matches] task.status = "completed" except Exception as e: logger.error(f"Matchmaking failed: {e}") task.status = "failed" task.error = str(e) return task