81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
import pyaudio, wave, time, os, numpy as np, requests, subprocess, sys, json
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
|
|
ROOT_DIR = os.getenv('ROOT_DIR', os.path.dirname(__file__))
|
|
|
|
##########################################################################################
|
|
|
|
##### OLLAMA #####
|
|
def ping_api(url):
|
|
"""Ping the specified API URL and return the response status."""
|
|
try:
|
|
# Send a GET request to the API
|
|
response = requests.get(url)
|
|
|
|
# Check the response status code
|
|
if response.status_code == 200:
|
|
print("API is reachable.")
|
|
return True
|
|
else:
|
|
print(f"API responded with status code: {response.status_code}")
|
|
return False
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Error pinging API: {e}")
|
|
return False
|
|
|
|
def start_ollama_server(llm_model):
|
|
"""Start the Ollama server using subprocess."""
|
|
# Adjust the command as necessary for your setup
|
|
command = ["ollama", "run", "--keepalive", "24h", llm_model, 'Don\'t Say Anything'] # Replace with the correct command to start the server
|
|
|
|
# Start the server in a new process
|
|
subprocess.run(command)
|
|
|
|
|
|
|
|
def converse(input_text, ping_url, api_url, llm_model, llm_response_path, prompt, context_file, have_context = False):
|
|
"""
|
|
Send a prompt to the Ollama API and return the response.
|
|
input_text: The text to send to the API.
|
|
ping_url: The URL to ping to check if the API is running.
|
|
api_url: The URL of the Ollama API.
|
|
llm_model: The LLM model to use.
|
|
llm_response_path: The path to save the LLM responses and prompts.
|
|
prompt: The prompt to use for the conversation.
|
|
context_file: The path to the context file.
|
|
"""
|
|
# Ping the Llama
|
|
if not ping_api(ping_url):
|
|
try:
|
|
start_ollama_server(llm_model)
|
|
except Exception as e:
|
|
print(f"Error starting Ollama server: {e}. If you are using another Ollama server, please ensure you have correctly specified the BASE_URL and that the server is running and not firewalled off.")
|
|
sys.exit(1)
|
|
|
|
payload = { "model": llm_model, "prompt": f'{prompt} {input_text}', "stream": False, "keep_alive": "24h" }
|
|
|
|
if have_context:
|
|
# load json context file
|
|
with open(context_file, 'r') as f:
|
|
context = json.load(f).get('context')
|
|
payload.update({'context': context})
|
|
|
|
# Make the POST request
|
|
response = requests.post(api_url, json=payload)
|
|
|
|
# Check for errors and print the response
|
|
if not response.ok:
|
|
print("Error:", response.status_code, response.text)
|
|
|
|
# Save the context and all other responses of the API call to a file
|
|
with open(context_file, 'w') as f:
|
|
json.dump(response.json(), f)
|
|
|
|
# Save the conversations to a file
|
|
with open(llm_response_path, "a") as f:
|
|
f.write(f'[{datetime.now().isoformat()}] Prompt: {input_text}\n')
|
|
f.write(f'[{response.json().get('created_at')}] Response: {response.json().get('response')}\n')
|
|
|
|
return response.json().get('response') |