Spaces:
Running
Running
File size: 2,866 Bytes
a55ea57 | 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 | """
File: /client_large.py
Created Date: Sunday October 19th 2025
Author: Christian Nonis <alch.infoemail@gmail.com>
-----
Last Modified: Sunday October 19th 2025 10:39:17 am
Modified By: the developer formerly known as Christian Nonis at <alch.infoemail@gmail.com>
-----
"""
import json
from openai import AzureOpenAI
from langchain_openai import AzureChatOpenAI
from src.adapters.interfaces.llm import LLM
from src.config import config
# https://console.cloud.google.com/vertex-ai/publishers/google/model-garden/gemini-2.5-pro?authuser=2&hl=en&project=virtual-assistant-2-474207
class LLMClientLarge(LLM):
"""
Large language model client, used for main agents and other high-level tasks.
"""
def __init__(self):
self.model = config.azure.large_llm_model
self.client = AzureOpenAI(
api_version=config.azure.llm_api_version,
azure_endpoint=config.azure.llm_endpoint,
api_key=config.azure.llm_subscription_key,
)
self.langchain_model = AzureChatOpenAI(
azure_deployment=config.azure.large_llm_model,
api_version=config.azure.llm_api_version,
azure_endpoint=config.azure.llm_endpoint,
api_key=config.azure.llm_subscription_key,
)
def generate_text(self, prompt: str, max_new_tokens: int = None) -> str:
"""
Generate a text response from the model and return it as a string.
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
**({"max_tokens": max_new_tokens} if max_new_tokens else {}),
)
return response.choices[0].message.content
def generate_json(
self, prompt: str, max_new_tokens: int = None, max_retries: int = 3
) -> dict:
"""
Generate a JSON response from the model and return it as a dictionary.
"""
response = None
while max_retries > 0 and response is None:
try:
_response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
**({"max_tokens": max_new_tokens} if max_new_tokens else {}),
response_format={"type": "json_object"},
)
_response = (
_response.choices[0].message.content.strip("```json").strip("```")
)
response = json.loads(_response)
except Exception as e: # pylint: disable=broad-exception-caught
max_retries -= 1
if max_retries <= 0:
print(f"Failed to generate JSON response: {e}")
raise e
return response
_llm_large_client = LLMClientLarge()
|