diff --git a/app/services/llm_service.py b/app/services/llm_service.py new file mode 100644 index 0000000..1835995 --- /dev/null +++ b/app/services/llm_service.py @@ -0,0 +1,67 @@ +from typing import Dict, Any, List +import vertexai +from vertexai.generative_models import GenerativeModel, Tool, FunctionDeclaration +from app.core.settings import settings +from app.models.tool_model import ToolDefinition + + +class LLMService: + + def __init__(self): + vertexai.init( + project=settings.google_project_id, + location=settings.google_location + ) + + self.model = GenerativeModel("gemini-1.5-pro") + + def build_vertex_tools(self, tools: List[ToolDefinition]): # Converte as Tools internas (ToolDefinition) para o formato que o Vertex AI entende. + + vertex_tools = [] + + for tool in tools: + vertex_tools.append( + Tool( + function_declarations=[ + FunctionDeclaration( + name=tool.name, + description=tool.description, + parameters=tool.parameters + ) + ] + ) + ) + + return vertex_tools + + async def generate_response( + self, + message: str, + tools: List[ToolDefinition], + history: List[Dict[str, Any]] = None + ) -> Dict[str, Any]: + + vertex_tools = self.build_vertex_tools(tools) # Convertendo tools para formato do Vertex + + chat = self.model.start_chat( + history=history or [], + tools=vertex_tools + ) + + response = chat.send_message(message) + + part = response.candidates[0].content.parts[0] + + if part.function_call: + return { + "response": None, + "tool_call": { + "name": part.function_call.name, + "arguments": dict(part.function_call.args) + } + } + + return { + "response": response.text, + "tool_call": None + } diff --git a/app/services/tool_registry.py b/app/services/tool_registry.py new file mode 100644 index 0000000..147a9ae --- /dev/null +++ b/app/services/tool_registry.py @@ -0,0 +1,49 @@ +from typing import List +from app.models.tool_model import ToolDefinition +from app.integrations.estoque_service import consultar_estoque + + +class ToolRegistry: + + def __init__(self): + self._tools = [] + + self.register_tool( + name="consultar_estoque", + description="Consulta veículos disponíveis até determinado preço", # Descrição que ajuda o modelo Gemini a entender quado selecionar essa Tool. + parameters={ + "type": "object", + "properties": { + "preco_max": { + "type": "number", + "description": "Preço máximo do veículo" + } + }, + "required": ["preco_max"] + }, + handler=consultar_estoque + ) + + + def register_tool(self, name, description, parameters, handler): + self._tools.append( + ToolDefinition( + name=name, + description=description, + parameters=parameters, + handler=handler + ) + ) + + + def get_tools(self) -> List[ToolDefinition]: + return self._tools + + + async def execute(self, name: str, arguments: dict): + tool = next((t for t in self._tools if t.name == name), None) # Procura dentro da lista de tools aquela que tem o mesmo nome + + if not tool: + raise Exception(f"Tool {name} não encontrada.") + + return await tool.handler(**arguments)