File size: 4,046 Bytes
d57551e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
#converterPlugin.py
from typing import Annotated, Optional
from cosmosConnector import CosmosLampHandler
from semantic_kernel.functions import kernel_function
class ConverterPlugin:
def __init__(self, logger):
self.logger = logger
self.db = CosmosLampHandler(logger=logger)
@kernel_function(
name="query_converters",
description="Execute SQL query against Cosmos DB converters collection"
)
async def query_converters(self, query: str) -> str:
try:
items = await self.db.query_converters(query)
self.logger.info(f"Executed query: {query}")
if not items:
return "No items found for the given query."
return str(items)
except Exception as e:
return f"Query failed: {str(e)}"
@kernel_function
async def get_compatible_lamps(
self,
artnr: Annotated[int, "Converter ARTNR (partition key)"]
) -> str:
"""Get compatible lamps for a converter by ARTNR"""
try:
lamps = await self.db.get_compatible_lamps(artnr)
self.logger.info(f"Used get_compatible_lamps with ARTNR: {artnr}")
return f"Compatible lamps: {', '.join(lamps)}" if lamps else "No lamps found"
# return "\n".join([f"{c.model_dump()})" for c in lamps]) if lamps else "No lamps found"
except Exception as e:
return f"Error retrieving compatible lamps: {str(e)}"
@kernel_function(
name="get_converters_by_lamp_type",
description="Find converters compatible with a specific lamp type"
)
async def get_converters_by_lamp_type(
self,
lamp_type: Annotated[str, "Lamp model (e.g., Haloled, B4)"]
) -> str:
"""Find converters compatible with specific lamp type"""
try:
converters = await self.db.get_converters_by_lamp_type(lamp_type)
self.logger.info(f"Used get_converters_by_lamp_type with lamp_type: {lamp_type}")
if not converters:
return "No compatible converters found"
# return "\n".join([f"{c.name} (ARTNR: {c.artnr})\nTYPE: {c.type}\nMANUAL: {c.pdf_link}" for c in converters])
return "\n".join([f"{c.model_dump()})" for c in converters]) if converters else "No converters found"
except Exception as e:
return f"Error retrieving converters: {str(e)}"
@kernel_function(
name="get_lamp_limits",
description="Get min/max lamps for a converter by ARTNR and lamp type"
)
async def get_lamp_limits(
self,
artnr: Annotated[int, "Converter ARTNR"],
lamp_type: Annotated[str, "Lamp model (e.g., Haloled)"]
) -> str:
"""Get min/max lamps for a converter"""
try:
limits = await self.db.get_lamp_limits(artnr, lamp_type)
self.logger.info(f"Used get_lamp_limits with ARTNR: {artnr} and lamp_type: {lamp_type}")
return f"{lamp_type}: Min {limits['min']} - Max {limits['max']} lamps"
except Exception as e:
return f"Error retrieving lamp limits: {str(e)}"
@kernel_function(
name = "RAG_search",
description="Get data based on vector similarity scores for queries regarding dimmability or in cases where SQL generation fails"
)
async def RAG_search(
self,
query: Annotated[str, "User query"]
) -> str:
try:
converters = await self.db.RAG_search(query=query)
self.logger.info(f"Used RAG search for user query: {query}")
if not converters:
return "No converters found"
# return "\n".join([f"{c.name} (ARTNR: {c.artnr})\nTYPE: {c.type}\nMANUAL: {c.pdf_link}" for c in converters])
return "\n".join([f"{c.model_dump()})" for c in converters]) if converters else "No converters found"
except Exception as e:
return f"Error retrieving converters: {str(e)}"
|