Spaces:
Runtime error
Runtime error
| from langchain_openai import OpenAI | |
| import openai | |
| import sys | |
| import os | |
| import requests | |
| sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../'))) | |
| from config import * | |
| class LLM_API_Call: | |
| def __init__(self, type) -> None: | |
| if type == "openai": | |
| self.llm = OpenAI_API_Call(api_key = LLM_CONFIG[""], | |
| model = LLM_CONFIG["model"]) | |
| elif type == "gilas": | |
| self.llm = Gilas_API_Call(api_key = GILAS_CONFIG["api_key"], | |
| model = GILAS_CONFIG["model"], | |
| base_url=GILAS_CONFIG["base_url"]) | |
| else: | |
| self.llm = OpenAI( | |
| **LLM_CONFIG | |
| ) | |
| def get_LLM_response(self, prompt: str) -> str: | |
| return self.llm.invoke(prompt) | |
| class OpenAI_API_Call: | |
| def __init__(self, api_key, model="gpt-4"): | |
| self.api_key = api_key | |
| openai.api_key = api_key | |
| self.model = model | |
| self.conversation = [] | |
| def add_message(self, role, content): | |
| self.conversation.append({"role": role, "content": content}) | |
| def get_response(self): | |
| response = openai.ChatCompletion.create( | |
| model=self.model, | |
| messages=self.conversation | |
| ) | |
| return response['choices'][0]['message']['content'] | |
| def invoke(self, user_input): | |
| self.add_message("user", user_input) | |
| response = self.get_response() | |
| self.add_message("assistant", response) | |
| return response | |
| class Gilas_API_Call: | |
| def __init__(self, api_key, base_url, model="gpt-4o-mini"): | |
| self.api_key = api_key | |
| self.base_url = base_url | |
| self.model = model | |
| self.headers = { | |
| "Authorization": f"Bearer {self.api_key}", | |
| "Content-Type": "application/json" | |
| } | |
| self.conversation = [] | |
| def add_message(self, role, content): | |
| self.conversation.append({"role": role, "content": content}) | |
| def get_response(self): | |
| data = { | |
| "model": self.model, | |
| "messages": self.conversation | |
| } | |
| response = requests.post( | |
| url=f"{self.base_url}/chat/completions", | |
| headers=self.headers, | |
| json=data | |
| ) | |
| # print(f"Response status code: {response.status_code}") | |
| # print(f"Response content: {response.text}") | |
| if response.status_code == 200: | |
| try: | |
| return response.json()['choices'][0]['message']['content'] | |
| except (KeyError, IndexError, ValueError) as e: | |
| raise Exception(f"Unexpected API response format: {e}") | |
| else: | |
| raise Exception(f"Gilas API call failed: {response.status_code} - {response.text}") | |
| def invoke(self, user_input): | |
| self.add_message("user", user_input) | |
| response = self.get_response() | |
| self.add_message("assistant", response) | |
| return response | |
| # test = LLM_API_Call(type = "gilas") | |
| # print(test.get_LLM_response("سلام")) |