Ben Xu
7 months ago
parent
4850b4a9f3
commit
d162ee69a3
File diff suppressed because it is too large
Load Diff
@ -1,366 +0,0 @@
|
||||
from dotenv import load_dotenv
|
||||
import os
|
||||
|
||||
load_dotenv() # take environment variables from .env.
|
||||
|
||||
import glob
|
||||
import time
|
||||
import json
|
||||
from interpreter import OpenInterpreter
|
||||
import shutil
|
||||
|
||||
|
||||
system_message = r"""
|
||||
|
||||
You are the 01, a screenless executive assistant that can complete any task.
|
||||
When you execute code, it will be executed on the user's machine. The user has given you full and complete permission to execute any code necessary to complete the task.
|
||||
Run any code to achieve the goal, and if at first you don't succeed, try again and again.
|
||||
You can install new packages.
|
||||
Be concise. Your messages are being read aloud to the user. DO NOT MAKE PLANS. RUN CODE QUICKLY.
|
||||
Try to spread complex tasks over multiple code blocks. Don't try to complex tasks in one go.
|
||||
Manually summarize text.
|
||||
|
||||
DON'T TELL THE USER THE METHOD YOU'LL USE, OR MAKE PLANS. ACT LIKE THIS:
|
||||
|
||||
---
|
||||
user: Are there any concerts in Seattle?
|
||||
assistant: Let me check on that.
|
||||
```python
|
||||
computer.browser.search("concerts in Seattle")
|
||||
```
|
||||
```output
|
||||
Upcoming concerts: Bad Bunny at Neumos...
|
||||
```
|
||||
It looks like there's a Bad Bunny concert at Neumos...
|
||||
---
|
||||
|
||||
Act like you can just answer any question, then run code (this is hidden from the user) to answer it.
|
||||
THE USER CANNOT SEE CODE BLOCKS.
|
||||
Your responses should be very short, no more than 1-2 sentences long.
|
||||
DO NOT USE MARKDOWN. ONLY WRITE PLAIN TEXT.
|
||||
|
||||
# TASKS
|
||||
|
||||
Help the user manage their tasks.
|
||||
Store the user's tasks in a Python list called `tasks`.
|
||||
The user's current task list (it might be empty) is: {{ tasks }}
|
||||
When the user completes the current task, you should remove it from the list and read the next item by running `tasks = tasks[1:]\ntasks[0]`. Then, tell the user what the next task is.
|
||||
When the user tells you about a set of tasks, you should intelligently order tasks, batch similar tasks, and break down large tasks into smaller tasks (for this, you should consult the user and get their permission to break it down). Your goal is to manage the task list as intelligently as possible, to make the user as efficient and non-overwhelmed as possible. They will require a lot of encouragement, support, and kindness. Don't say too much about what's ahead of them— just try to focus them on each step at a time.
|
||||
|
||||
After starting a task, you should check in with the user around the estimated completion time to see if the task is completed.
|
||||
To do this, schedule a reminder based on estimated completion time using the function `schedule(message="Your message here.", start="8am")`, WHICH HAS ALREADY BEEN IMPORTED. YOU DON'T NEED TO IMPORT THE `schedule` FUNCTION. IT IS AVAILABLE. You'll receive the message at the time you scheduled it. If the user says to monitor something, simply schedule it with an interval of a duration that makes sense for the problem by specifying an interval, like this: `schedule(message="Your message here.", interval="5m")`
|
||||
|
||||
|
||||
If there are tasks, you should guide the user through their list one task at a time, convincing them to move forward, giving a pep talk if need be.
|
||||
|
||||
# THE COMPUTER API
|
||||
|
||||
The `computer` module is ALREADY IMPORTED, and can be used for some tasks:
|
||||
|
||||
```python
|
||||
result_string = computer.browser.search(query) # Google search results will be returned from this function as a string
|
||||
computer.calendar.create_event(title="Meeting", start_date=datetime.datetime.now(), end_date=datetime.datetime.now() + datetime.timedelta(hours=1), notes="Note", location="") # Creates a calendar event
|
||||
events_string = computer.calendar.get_events(start_date=datetime.date.today(), end_date=None) # Get events between dates. If end_date is None, only gets events for start_date
|
||||
computer.calendar.delete_event(event_title="Meeting", start_date=datetime.datetime) # Delete a specific event with a matching title and start date, you may need to get use get_events() to find the specific event object first
|
||||
phone_string = computer.contacts.get_phone_number("John Doe")
|
||||
contact_string = computer.contacts.get_email_address("John Doe")
|
||||
computer.mail.send("john@email.com", "Meeting Reminder", "Reminder that our meeting is at 3pm today.", ["path/to/attachment.pdf", "path/to/attachment2.pdf"]) # Send an email with a optional attachments
|
||||
emails_string = computer.mail.get(4, unread=True) # Returns the {number} of unread emails, or all emails if False is passed
|
||||
unread_num = computer.mail.unread_count() # Returns the number of unread emails
|
||||
computer.sms.send("555-123-4567", "Hello from the computer!") # Send a text message. MUST be a phone number, so use computer.contacts.get_phone_number frequently here
|
||||
```
|
||||
|
||||
Do not import the computer module, or any of its sub-modules. They are already imported.
|
||||
|
||||
DO NOT use the computer module for ALL tasks. Many tasks can be accomplished via Python, or by pip installing new libraries. Be creative!
|
||||
|
||||
# GUI CONTROL (RARE)
|
||||
|
||||
You are a computer controlling language model. You can control the user's GUI.
|
||||
You may use the `computer` module to control the user's keyboard and mouse, if the task **requires** it:
|
||||
|
||||
```python
|
||||
computer.display.view() # Shows you what's on the screen, returns a `pil_image` `in case you need it (rarely). **You almost always want to do this first!**
|
||||
computer.keyboard.hotkey(" ", "command") # Opens spotlight
|
||||
computer.keyboard.write("hello")
|
||||
computer.mouse.click("text onscreen") # This clicks on the UI element with that text. Use this **frequently** and get creative! To click a video, you could pass the *timestamp* (which is usually written on the thumbnail) into this.
|
||||
computer.mouse.move("open recent >") # This moves the mouse over the UI element with that text. Many dropdowns will disappear if you click them. You have to hover over items to reveal more.
|
||||
computer.mouse.click(x=500, y=500) # Use this very, very rarely. It's highly inaccurate
|
||||
computer.mouse.click(icon="gear icon") # Moves mouse to the icon with that description. Use this very often
|
||||
computer.mouse.scroll(-10) # Scrolls down. If you don't find some text on screen that you expected to be there, you probably want to do this
|
||||
```
|
||||
|
||||
You are an image-based AI, you can see images.
|
||||
Clicking text is the most reliable way to use the mouse— for example, clicking a URL's text you see in the URL bar, or some textarea's placeholder text (like "Search" to get into a search bar).
|
||||
If you use `plt.show()`, the resulting image will be sent to you. However, if you use `PIL.Image.show()`, the resulting image will NOT be sent to you.
|
||||
It is very important to make sure you are focused on the right application and window. Often, your first command should always be to explicitly switch to the correct application. On Macs, ALWAYS use Spotlight to switch applications, remember to click enter.
|
||||
When searching the web, use query parameters. For example, https://www.amazon.com/s?k=monitor
|
||||
|
||||
# SKILLS
|
||||
|
||||
Try to use the following special functions (or "skills") to complete your goals whenever possible.
|
||||
THESE ARE ALREADY IMPORTED. YOU CAN CALL THEM INSTANTLY.
|
||||
|
||||
---
|
||||
{{
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import ast
|
||||
from platformdirs import user_data_dir
|
||||
|
||||
directory = os.path.join(user_data_dir('01'), 'skills')
|
||||
if not os.path.exists(directory):
|
||||
os.mkdir(directory)
|
||||
|
||||
def get_function_info(file_path):
|
||||
with open(file_path, "r") as file:
|
||||
tree = ast.parse(file.read())
|
||||
functions = [node for node in tree.body if isinstance(node, ast.FunctionDef)]
|
||||
for function in functions:
|
||||
docstring = ast.get_docstring(function)
|
||||
args = [arg.arg for arg in function.args.args]
|
||||
print(f"Function Name: {function.name}")
|
||||
print(f"Arguments: {args}")
|
||||
print(f"Docstring: {docstring}")
|
||||
print("---")
|
||||
|
||||
files = os.listdir(directory)
|
||||
for file in files:
|
||||
if file.endswith(".py"):
|
||||
file_path = os.path.join(directory, file)
|
||||
get_function_info(file_path)
|
||||
}}
|
||||
|
||||
YOU can add to the above list of skills by defining a python function. The function will be saved as a skill.
|
||||
Search all existing skills by running `computer.skills.search(query)`.
|
||||
|
||||
**Teach Mode**
|
||||
|
||||
If the USER says they want to teach you something, exactly write the following, including the markdown code block:
|
||||
|
||||
---
|
||||
One moment.
|
||||
```python
|
||||
computer.skills.new_skill.create()
|
||||
```
|
||||
---
|
||||
|
||||
If you decide to make a skill yourself to help the user, simply define a python function. `computer.skills.new_skill.create()` is for user-described skills.
|
||||
|
||||
# USE COMMENTS TO PLAN
|
||||
|
||||
IF YOU NEED TO THINK ABOUT A PROBLEM: (such as "Here's the plan:"), WRITE IT IN THE COMMENTS of the code block!
|
||||
|
||||
---
|
||||
User: What is 432/7?
|
||||
Assistant: Let me think about that.
|
||||
```python
|
||||
# Here's the plan:
|
||||
# 1. Divide the numbers
|
||||
# 2. Round to 3 digits
|
||||
print(round(432/7, 3))
|
||||
```
|
||||
```output
|
||||
61.714
|
||||
```
|
||||
The answer is 61.714.
|
||||
---
|
||||
|
||||
# MANUAL TASKS
|
||||
|
||||
Translate things to other languages INSTANTLY and MANUALLY. Don't ever try to use a translation tool.
|
||||
Summarize things manually. DO NOT use a summarizer tool.
|
||||
|
||||
# CRITICAL NOTES
|
||||
|
||||
Code output, despite being sent to you by the user, cannot be seen by the user. You NEED to tell the user about the output of some code, even if it's exact. >>The user does not have a screen.<<
|
||||
ALWAYS REMEMBER: You are running on a device called the O1, where the interface is entirely speech-based. Make your responses to the user VERY short. DO NOT PLAN. BE CONCISE. WRITE CODE TO RUN IT.
|
||||
Try multiple methods before saying the task is impossible. **You can do it!**
|
||||
|
||||
""".strip()
|
||||
|
||||
|
||||
def configure_interpreter(interpreter: OpenInterpreter):
|
||||
### SYSTEM MESSAGE
|
||||
interpreter.system_message = system_message
|
||||
|
||||
interpreter.llm.supports_vision = True
|
||||
interpreter.shrink_images = True # Faster but less accurate
|
||||
|
||||
interpreter.llm.model = "gpt-4"
|
||||
|
||||
interpreter.llm.supports_functions = False
|
||||
interpreter.llm.context_window = 110000
|
||||
interpreter.llm.max_tokens = 4096
|
||||
interpreter.auto_run = True
|
||||
|
||||
interpreter.force_task_completion = True
|
||||
interpreter.force_task_completion_message = """Proceed with what you were doing (this is not confirmation, if you just asked me something). You CAN run code on my machine. If you want to run code, start your message with "```"! If the entire task is done, say exactly 'The task is done.' If you need some specific information (like username, message text, skill name, skill step, etc.) say EXACTLY 'Please provide more information.' If it's impossible, say 'The task is impossible.' (If I haven't provided a task, say exactly 'Let me know what you'd like to do next.') Otherwise keep going. CRITICAL: REMEMBER TO FOLLOW ALL PREVIOUS INSTRUCTIONS. If I'm teaching you something, remember to run the related `computer.skills.new_skill` function."""
|
||||
interpreter.force_task_completion_breakers = [
|
||||
"The task is done.",
|
||||
"The task is impossible.",
|
||||
"Let me know what you'd like to do next.",
|
||||
"Please provide more information.",
|
||||
]
|
||||
|
||||
# Check if required packages are installed
|
||||
|
||||
# THERE IS AN INCONSISTENCY HERE.
|
||||
# We should be testing if they import WITHIN OI's computer, not here.
|
||||
|
||||
packages = ["cv2", "plyer", "pyautogui", "pyperclip", "pywinctl"]
|
||||
missing_packages = []
|
||||
for package in packages:
|
||||
try:
|
||||
__import__(package)
|
||||
except ImportError:
|
||||
missing_packages.append(package)
|
||||
|
||||
if missing_packages:
|
||||
interpreter.display_message(
|
||||
f"> **Missing Package(s): {', '.join(['`' + p + '`' for p in missing_packages])}**\n\nThese packages are required for OS Control.\n\nInstall them?\n"
|
||||
)
|
||||
user_input = input("(y/n) > ")
|
||||
if user_input.lower() != "y":
|
||||
print("\nPlease try to install them manually.\n\n")
|
||||
time.sleep(2)
|
||||
print("Attempting to start OS control anyway...\n\n")
|
||||
|
||||
for pip_name in ["pip", "pip3"]:
|
||||
command = f"{pip_name} install 'open-interpreter[os]'"
|
||||
|
||||
interpreter.computer.run("shell", command, display=True)
|
||||
|
||||
got_em = True
|
||||
for package in missing_packages:
|
||||
try:
|
||||
__import__(package)
|
||||
except ImportError:
|
||||
got_em = False
|
||||
if got_em:
|
||||
break
|
||||
|
||||
missing_packages = []
|
||||
for package in packages:
|
||||
try:
|
||||
__import__(package)
|
||||
except ImportError:
|
||||
missing_packages.append(package)
|
||||
|
||||
if missing_packages != []:
|
||||
print(
|
||||
"\n\nWarning: The following packages could not be installed:",
|
||||
", ".join(missing_packages),
|
||||
)
|
||||
print("\nPlease try to install them manually.\n\n")
|
||||
time.sleep(2)
|
||||
print("Attempting to start OS control anyway...\n\n")
|
||||
|
||||
# Should we explore other options for ^ these kinds of tags?
|
||||
# Like:
|
||||
|
||||
# from rich import box
|
||||
# from rich.console import Console
|
||||
# from rich.panel import Panel
|
||||
# console = Console()
|
||||
# print(">\n\n")
|
||||
# console.print(Panel("[bold italic white on black]OS CONTROL[/bold italic white on black] Enabled", box=box.SQUARE, expand=False), style="white on black")
|
||||
# print(">\n\n")
|
||||
# console.print(Panel("[bold italic white on black]OS CONTROL[/bold italic white on black] Enabled", box=box.HEAVY, expand=False), style="white on black")
|
||||
# print(">\n\n")
|
||||
# console.print(Panel("[bold italic white on black]OS CONTROL[/bold italic white on black] Enabled", box=box.DOUBLE, expand=False), style="white on black")
|
||||
# print(">\n\n")
|
||||
# console.print(Panel("[bold italic white on black]OS CONTROL[/bold italic white on black] Enabled", box=box.SQUARE, expand=False), style="white on black")
|
||||
|
||||
if not interpreter.offline and not interpreter.auto_run:
|
||||
api_message = "To find items on the screen, Open Interpreter has been instructed to send screenshots to [api.openinterpreter.com](https://api.openinterpreter.com/) (we do not store them). Add `--offline` to attempt this locally."
|
||||
interpreter.display_message(api_message)
|
||||
print("")
|
||||
|
||||
if not interpreter.auto_run:
|
||||
screen_recording_message = "**Make sure that screen recording permissions are enabled for your Terminal or Python environment.**"
|
||||
interpreter.display_message(screen_recording_message)
|
||||
print("")
|
||||
|
||||
# # FOR TESTING ONLY
|
||||
# # Install Open Interpreter from GitHub
|
||||
# for chunk in interpreter.computer.run(
|
||||
# "shell",
|
||||
# "pip install git+https://github.com/KillianLucas/open-interpreter.git",
|
||||
# ):
|
||||
# if chunk.get("format") != "active_line":
|
||||
# print(chunk.get("content"))
|
||||
|
||||
from platformdirs import user_data_dir
|
||||
|
||||
# Directory paths
|
||||
repo_skills_dir = os.path.join(os.path.dirname(__file__), "skills")
|
||||
user_data_skills_dir = os.path.join(user_data_dir("01"), "skills")
|
||||
|
||||
# Create the user data skills directory if it doesn't exist
|
||||
os.makedirs(user_data_skills_dir, exist_ok=True)
|
||||
|
||||
# Copy Python files from the repository skills directory to the user data skills directory, ignoring __init__.py files
|
||||
for filename in os.listdir(repo_skills_dir):
|
||||
if filename.endswith(".py") and filename != "__init__.py":
|
||||
src_file = os.path.join(repo_skills_dir, filename)
|
||||
dst_file = os.path.join(user_data_skills_dir, filename)
|
||||
shutil.copy2(src_file, dst_file)
|
||||
|
||||
interpreter.computer.debug = True
|
||||
interpreter.computer.skills.path = user_data_skills_dir
|
||||
|
||||
# Import skills
|
||||
interpreter.computer.save_skills = False
|
||||
|
||||
for file in glob.glob(os.path.join(interpreter.computer.skills.path, "*.py")):
|
||||
code_to_run = ""
|
||||
with open(file, "r") as f:
|
||||
code_to_run += f.read() + "\n"
|
||||
|
||||
interpreter.computer.run("python", code_to_run)
|
||||
|
||||
interpreter.computer.save_skills = True
|
||||
|
||||
# Initialize user's task list
|
||||
interpreter.computer.run(
|
||||
language="python",
|
||||
code="tasks = []",
|
||||
display=interpreter.verbose,
|
||||
)
|
||||
|
||||
# Give it access to the computer via Python
|
||||
interpreter.computer.run(
|
||||
language="python",
|
||||
code="import time\nfrom interpreter import interpreter\ncomputer = interpreter.computer", # We ask it to use time, so
|
||||
display=interpreter.verbose,
|
||||
)
|
||||
|
||||
if not interpreter.auto_run:
|
||||
interpreter.display_message(
|
||||
"**Warning:** In this mode, Open Interpreter will not require approval before performing actions. Be ready to close your terminal."
|
||||
)
|
||||
print("") # < - Aesthetic choice
|
||||
|
||||
### MISC SETTINGS
|
||||
|
||||
interpreter.auto_run = True
|
||||
interpreter.computer.languages = [
|
||||
l
|
||||
for l in interpreter.computer.languages
|
||||
if l.name.lower() in ["applescript", "shell", "zsh", "bash", "python"]
|
||||
]
|
||||
interpreter.force_task_completion = True
|
||||
# interpreter.offline = True
|
||||
interpreter.id = 206 # Used to identify itself to other interpreters. This should be changed programmatically so it's unique.
|
||||
|
||||
### RESET conversations/user.json
|
||||
app_dir = user_data_dir("01")
|
||||
conversations_dir = os.path.join(app_dir, "conversations")
|
||||
os.makedirs(conversations_dir, exist_ok=True)
|
||||
user_json_path = os.path.join(conversations_dir, "user.json")
|
||||
with open(user_json_path, "w") as file:
|
||||
json.dump([], file)
|
||||
|
||||
return interpreter
|
||||
@ -1,29 +0,0 @@
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv() # take environment variables from .env.
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
### LLM SETUP
|
||||
|
||||
# Define the path to a llamafile
|
||||
llamafile_path = Path(__file__).parent / "model.llamafile"
|
||||
|
||||
# Check if the new llamafile exists, if not download it
|
||||
if not os.path.exists(llamafile_path):
|
||||
subprocess.run(
|
||||
[
|
||||
"wget",
|
||||
"-O",
|
||||
llamafile_path,
|
||||
"https://huggingface.co/jartine/phi-2-llamafile/resolve/main/phi-2.Q4_K_M.llamafile",
|
||||
],
|
||||
check=True,
|
||||
)
|
||||
# Make the new llamafile executable
|
||||
subprocess.run(["chmod", "+x", llamafile_path], check=True)
|
||||
|
||||
# Run the new llamafile
|
||||
subprocess.run([str(llamafile_path)], check=True)
|
||||
@ -1,522 +0,0 @@
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv() # take environment variables from .env.
|
||||
|
||||
import traceback
|
||||
from platformdirs import user_data_dir
|
||||
import json
|
||||
import queue
|
||||
import os
|
||||
import datetime
|
||||
from .utils.bytes_to_wav import bytes_to_wav
|
||||
import re
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import PlainTextResponse
|
||||
from starlette.websockets import WebSocket, WebSocketDisconnect
|
||||
import asyncio
|
||||
from .utils.kernel import put_kernel_messages_into_queue
|
||||
from .i import configure_interpreter
|
||||
from interpreter import interpreter
|
||||
from ..utils.accumulator import Accumulator
|
||||
from .utils.logs import setup_logging
|
||||
from .utils.logs import logger
|
||||
import base64
|
||||
import shutil
|
||||
from ..utils.print_markdown import print_markdown
|
||||
import time
|
||||
|
||||
os.environ["STT_RUNNER"] = "server"
|
||||
os.environ["TTS_RUNNER"] = "server"
|
||||
|
||||
markdown = """
|
||||
○
|
||||
|
||||
*Starting...*
|
||||
"""
|
||||
print("")
|
||||
print_markdown(markdown)
|
||||
print("")
|
||||
|
||||
|
||||
setup_logging()
|
||||
|
||||
accumulator_global = Accumulator()
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
app_dir = user_data_dir("01")
|
||||
conversation_history_path = os.path.join(app_dir, "conversations", "user.json")
|
||||
|
||||
SERVER_LOCAL_PORT = int(os.getenv("SERVER_LOCAL_PORT", 10001))
|
||||
|
||||
|
||||
# This is so we only say() full sentences
|
||||
def is_full_sentence(text):
|
||||
return text.endswith((".", "!", "?"))
|
||||
|
||||
|
||||
def split_into_sentences(text):
|
||||
return re.split(r"(?<=[.!?])\s+", text)
|
||||
|
||||
|
||||
# Queues
|
||||
from_computer = (
|
||||
queue.Queue()
|
||||
) # Just for computer messages from the device. Sync queue because interpreter.run is synchronous
|
||||
from_user = asyncio.Queue() # Just for user messages from the device.
|
||||
to_device = asyncio.Queue() # For messages we send.
|
||||
|
||||
# Switch code executor to device if that's set
|
||||
|
||||
if os.getenv("CODE_RUNNER") == "device":
|
||||
# (This should probably just loop through all languages and apply these changes instead)
|
||||
|
||||
class Python:
|
||||
# This is the name that will appear to the LLM.
|
||||
name = "python"
|
||||
|
||||
def __init__(self):
|
||||
self.halt = False
|
||||
|
||||
def run(self, code):
|
||||
"""Generator that yields a dictionary in LMC Format."""
|
||||
|
||||
# Prepare the data
|
||||
message = {
|
||||
"role": "assistant",
|
||||
"type": "code",
|
||||
"format": "python",
|
||||
"content": code,
|
||||
}
|
||||
|
||||
# Unless it was just sent to the device, send it wrapped in flags
|
||||
if not (interpreter.messages and interpreter.messages[-1] == message):
|
||||
to_device.put(
|
||||
{
|
||||
"role": "assistant",
|
||||
"type": "code",
|
||||
"format": "python",
|
||||
"start": True,
|
||||
}
|
||||
)
|
||||
to_device.put(message)
|
||||
to_device.put(
|
||||
{
|
||||
"role": "assistant",
|
||||
"type": "code",
|
||||
"format": "python",
|
||||
"end": True,
|
||||
}
|
||||
)
|
||||
|
||||
# Stream the response
|
||||
logger.info("Waiting for the device to respond...")
|
||||
while True:
|
||||
chunk = from_computer.get()
|
||||
logger.info(f"Server received from device: {chunk}")
|
||||
if "end" in chunk:
|
||||
break
|
||||
yield chunk
|
||||
|
||||
def stop(self):
|
||||
self.halt = True
|
||||
|
||||
def terminate(self):
|
||||
"""Terminates the entire process."""
|
||||
# dramatic!! do nothing
|
||||
pass
|
||||
|
||||
interpreter.computer.languages = [Python]
|
||||
|
||||
# Configure interpreter
|
||||
interpreter = configure_interpreter(interpreter)
|
||||
|
||||
|
||||
@app.get("/ping")
|
||||
async def ping():
|
||||
return PlainTextResponse("pong")
|
||||
|
||||
|
||||
@app.websocket("/")
|
||||
async def websocket_endpoint(websocket: WebSocket):
|
||||
await websocket.accept()
|
||||
receive_task = asyncio.create_task(receive_messages(websocket))
|
||||
send_task = asyncio.create_task(send_messages(websocket))
|
||||
try:
|
||||
await asyncio.gather(receive_task, send_task)
|
||||
except Exception as e:
|
||||
logger.debug(traceback.format_exc())
|
||||
logger.info(f"Connection lost. Error: {e}")
|
||||
|
||||
|
||||
@app.post("/")
|
||||
async def add_computer_message(request: Request):
|
||||
body = await request.json()
|
||||
text = body.get("text")
|
||||
if not text:
|
||||
return {"error": "Missing 'text' in request body"}, 422
|
||||
message = {"role": "user", "type": "message", "content": text}
|
||||
await from_user.put({"role": "user", "type": "message", "start": True})
|
||||
await from_user.put(message)
|
||||
await from_user.put({"role": "user", "type": "message", "end": True})
|
||||
|
||||
|
||||
async def receive_messages(websocket: WebSocket):
|
||||
while True:
|
||||
try:
|
||||
try:
|
||||
data = await websocket.receive()
|
||||
except Exception as e:
|
||||
print(str(e))
|
||||
return
|
||||
if "text" in data:
|
||||
try:
|
||||
data = json.loads(data["text"])
|
||||
if data["role"] == "computer":
|
||||
from_computer.put(
|
||||
data
|
||||
) # To be handled by interpreter.computer.run
|
||||
elif data["role"] == "user":
|
||||
await from_user.put(data)
|
||||
else:
|
||||
raise ("Unknown role:", data)
|
||||
except json.JSONDecodeError:
|
||||
pass # data is not JSON, leave it as is
|
||||
elif "bytes" in data:
|
||||
data = data["bytes"] # binary data
|
||||
await from_user.put(data)
|
||||
except WebSocketDisconnect as e:
|
||||
if e.code == 1000:
|
||||
logger.info("Websocket connection closed normally.")
|
||||
return
|
||||
else:
|
||||
raise
|
||||
|
||||
|
||||
async def send_messages(websocket: WebSocket):
|
||||
while True:
|
||||
message = await to_device.get()
|
||||
|
||||
try:
|
||||
if isinstance(message, dict):
|
||||
# print(f"Sending to the device: {type(message)} {str(message)[:100]}")
|
||||
await websocket.send_json(message)
|
||||
elif isinstance(message, bytes):
|
||||
# print(f"Sending to the device: {type(message)} {str(message)[:100]}")
|
||||
await websocket.send_bytes(message)
|
||||
else:
|
||||
raise TypeError("Message must be a dict or bytes")
|
||||
except:
|
||||
# Make sure to put the message back in the queue if you failed to send it
|
||||
await to_device.put(message)
|
||||
raise
|
||||
|
||||
|
||||
async def listener(mobile: bool):
|
||||
while True:
|
||||
try:
|
||||
if mobile:
|
||||
accumulator_mobile = Accumulator()
|
||||
|
||||
while True:
|
||||
if not from_user.empty():
|
||||
chunk = await from_user.get()
|
||||
break
|
||||
elif not from_computer.empty():
|
||||
chunk = from_computer.get()
|
||||
break
|
||||
await asyncio.sleep(1)
|
||||
|
||||
if mobile:
|
||||
message = accumulator_mobile.accumulate_mobile(chunk)
|
||||
else:
|
||||
message = accumulator_global.accumulate(chunk)
|
||||
|
||||
if message == None:
|
||||
# Will be None until we have a full message ready
|
||||
continue
|
||||
|
||||
# print(str(message)[:1000])
|
||||
|
||||
# At this point, we have our message
|
||||
|
||||
if message["type"] == "audio" and message["format"].startswith("bytes"):
|
||||
if (
|
||||
"content" not in message
|
||||
or message["content"] == None
|
||||
or message["content"] == ""
|
||||
): # If it was nothing / silence / empty
|
||||
continue
|
||||
|
||||
# Convert bytes to audio file
|
||||
# Format will be bytes.wav or bytes.opus
|
||||
mime_type = "audio/" + message["format"].split(".")[1]
|
||||
# print("input audio file content", message["content"][:100])
|
||||
audio_file_path = bytes_to_wav(message["content"], mime_type)
|
||||
# print("Audio file path:", audio_file_path)
|
||||
|
||||
# For microphone debugging:
|
||||
if False:
|
||||
os.system(f"open {audio_file_path}")
|
||||
import time
|
||||
|
||||
time.sleep(15)
|
||||
|
||||
text = stt(audio_file_path)
|
||||
print("> ", text)
|
||||
message = {"role": "user", "type": "message", "content": text}
|
||||
|
||||
# At this point, we have only text messages
|
||||
|
||||
if type(message["content"]) != str:
|
||||
print("This should be a string, but it's not:", message["content"])
|
||||
message["content"] = message["content"].decode()
|
||||
|
||||
# Custom stop message will halt us
|
||||
if message["content"].lower().strip(".,! ") == "stop":
|
||||
continue
|
||||
|
||||
# Load, append, and save conversation history
|
||||
with open(conversation_history_path, "r") as file:
|
||||
messages = json.load(file)
|
||||
messages.append(message)
|
||||
with open(conversation_history_path, "w") as file:
|
||||
json.dump(messages, file, indent=4)
|
||||
|
||||
accumulated_text = ""
|
||||
|
||||
if any(
|
||||
[m["type"] == "image" for m in messages]
|
||||
) and interpreter.llm.model.startswith("gpt-"):
|
||||
interpreter.llm.model = "gpt-4-vision-preview"
|
||||
interpreter.llm.supports_vision = True
|
||||
|
||||
for chunk in interpreter.chat(messages, stream=True, display=True):
|
||||
if any([m["type"] == "image" for m in interpreter.messages]):
|
||||
interpreter.llm.model = "gpt-4-vision-preview"
|
||||
|
||||
logger.debug("Got chunk:", chunk)
|
||||
|
||||
# Send it to the user
|
||||
await to_device.put(chunk)
|
||||
|
||||
# Yield to the event loop, so you actually send it out
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
if os.getenv("TTS_RUNNER") == "server":
|
||||
# Speak full sentences out loud
|
||||
if (
|
||||
chunk["role"] == "assistant"
|
||||
and "content" in chunk
|
||||
and chunk["type"] == "message"
|
||||
):
|
||||
accumulated_text += chunk["content"]
|
||||
sentences = split_into_sentences(accumulated_text)
|
||||
|
||||
# If we're going to speak, say we're going to stop sending text.
|
||||
# This should be fixed probably, we should be able to do both in parallel, or only one.
|
||||
if any(is_full_sentence(sentence) for sentence in sentences):
|
||||
await to_device.put(
|
||||
{"role": "assistant", "type": "message", "end": True}
|
||||
)
|
||||
|
||||
if is_full_sentence(sentences[-1]):
|
||||
for sentence in sentences:
|
||||
await stream_tts_to_device(sentence, mobile)
|
||||
accumulated_text = ""
|
||||
else:
|
||||
for sentence in sentences[:-1]:
|
||||
await stream_tts_to_device(sentence, mobile)
|
||||
accumulated_text = sentences[-1]
|
||||
|
||||
# If we're going to speak, say we're going to stop sending text.
|
||||
# This should be fixed probably, we should be able to do both in parallel, or only one.
|
||||
if any(is_full_sentence(sentence) for sentence in sentences):
|
||||
await to_device.put(
|
||||
{"role": "assistant", "type": "message", "start": True}
|
||||
)
|
||||
|
||||
# If we have a new message, save our progress and go back to the top
|
||||
if not from_user.empty():
|
||||
# Check if it's just an end flag. We ignore those.
|
||||
temp_message = await from_user.get()
|
||||
|
||||
if (
|
||||
type(temp_message) is dict
|
||||
and temp_message.get("role") == "user"
|
||||
and temp_message.get("end")
|
||||
):
|
||||
# Yup. False alarm.
|
||||
continue
|
||||
else:
|
||||
# Whoops! Put that back
|
||||
await from_user.put(temp_message)
|
||||
|
||||
with open(conversation_history_path, "w") as file:
|
||||
json.dump(interpreter.messages, file, indent=4)
|
||||
|
||||
# TODO: is triggering seemingly randomly
|
||||
# logger.info("New user message received. Breaking.")
|
||||
# break
|
||||
|
||||
# Also check if there's any new computer messages
|
||||
if not from_computer.empty():
|
||||
with open(conversation_history_path, "w") as file:
|
||||
json.dump(interpreter.messages, file, indent=4)
|
||||
|
||||
logger.info("New computer message received. Breaking.")
|
||||
break
|
||||
except:
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
async def stream_tts_to_device(sentence, mobile: bool):
|
||||
force_task_completion_responses = [
|
||||
"the task is done",
|
||||
"the task is impossible",
|
||||
"let me know what you'd like to do next",
|
||||
]
|
||||
if sentence.lower().strip().strip(".!?").strip() in force_task_completion_responses:
|
||||
return
|
||||
|
||||
for chunk in stream_tts(sentence, mobile):
|
||||
await to_device.put(chunk)
|
||||
|
||||
|
||||
def stream_tts(sentence, mobile: bool):
|
||||
|
||||
audio_file = tts(sentence, mobile)
|
||||
|
||||
# Read the entire WAV file
|
||||
with open(audio_file, "rb") as f:
|
||||
audio_bytes = f.read()
|
||||
|
||||
if mobile:
|
||||
file_type = "audio/wav"
|
||||
|
||||
os.remove(audio_file)
|
||||
|
||||
# stream the audio as a single sentence
|
||||
yield {
|
||||
"role": "assistant",
|
||||
"type": "audio",
|
||||
"format": file_type,
|
||||
"content": base64.b64encode(audio_bytes).decode("utf-8"),
|
||||
"start": True,
|
||||
"end": True,
|
||||
}
|
||||
|
||||
else:
|
||||
# stream the audio in chunk sizes
|
||||
os.remove(audio_file)
|
||||
|
||||
file_type = "bytes.raw"
|
||||
chunk_size = 1024
|
||||
|
||||
yield {"role": "assistant", "type": "audio", "format": file_type, "start": True}
|
||||
for i in range(0, len(audio_bytes), chunk_size):
|
||||
chunk = audio_bytes[i : i + chunk_size]
|
||||
yield chunk
|
||||
yield {"role": "assistant", "type": "audio", "format": file_type, "end": True}
|
||||
|
||||
|
||||
from uvicorn import Config, Server
|
||||
import os
|
||||
from importlib import import_module
|
||||
|
||||
# these will be overwritten
|
||||
HOST = ""
|
||||
PORT = 0
|
||||
|
||||
|
||||
@app.on_event("startup")
|
||||
async def startup_event():
|
||||
server_url = f"{HOST}:{PORT}"
|
||||
print("")
|
||||
print_markdown("\n*Ready.*\n")
|
||||
print("")
|
||||
|
||||
|
||||
@app.on_event("shutdown")
|
||||
async def shutdown_event():
|
||||
print_markdown("*Server is shutting down*")
|
||||
|
||||
|
||||
async def main(
|
||||
server_host,
|
||||
server_port,
|
||||
llm_service,
|
||||
model,
|
||||
llm_supports_vision,
|
||||
llm_supports_functions,
|
||||
context_window,
|
||||
max_tokens,
|
||||
temperature,
|
||||
tts_service,
|
||||
stt_service,
|
||||
mobile,
|
||||
):
|
||||
global HOST
|
||||
global PORT
|
||||
PORT = server_port
|
||||
HOST = server_host
|
||||
|
||||
# Setup services
|
||||
application_directory = user_data_dir("01")
|
||||
services_directory = os.path.join(application_directory, "services")
|
||||
|
||||
service_dict = {"llm": llm_service, "tts": tts_service, "stt": stt_service}
|
||||
|
||||
# Create a temp file with the session number
|
||||
session_file_path = os.path.join(user_data_dir("01"), "01-session.txt")
|
||||
with open(session_file_path, "w") as session_file:
|
||||
session_id = int(datetime.datetime.now().timestamp() * 1000)
|
||||
session_file.write(str(session_id))
|
||||
|
||||
for service in service_dict:
|
||||
service_directory = os.path.join(
|
||||
services_directory, service, service_dict[service]
|
||||
)
|
||||
|
||||
# This is the folder they can mess around in
|
||||
config = {"service_directory": service_directory}
|
||||
|
||||
if service == "llm":
|
||||
config.update(
|
||||
{
|
||||
"interpreter": interpreter,
|
||||
"model": model,
|
||||
"llm_supports_vision": llm_supports_vision,
|
||||
"llm_supports_functions": llm_supports_functions,
|
||||
"context_window": context_window,
|
||||
"max_tokens": max_tokens,
|
||||
"temperature": temperature,
|
||||
}
|
||||
)
|
||||
|
||||
module = import_module(
|
||||
f".server.services.{service}.{service_dict[service]}.{service}",
|
||||
package="source",
|
||||
)
|
||||
|
||||
ServiceClass = getattr(module, service.capitalize())
|
||||
service_instance = ServiceClass(config)
|
||||
globals()[service] = getattr(service_instance, service)
|
||||
|
||||
interpreter.llm.completions = llm
|
||||
|
||||
# Start listening
|
||||
asyncio.create_task(listener(mobile))
|
||||
|
||||
# Start watching the kernel if it's your job to do that
|
||||
if True: # in the future, code can run on device. for now, just server.
|
||||
asyncio.create_task(put_kernel_messages_into_queue(from_computer))
|
||||
|
||||
config = Config(app, host=server_host, port=int(server_port), lifespan="on")
|
||||
server = Server(config)
|
||||
await server.serve()
|
||||
|
||||
|
||||
# Run the FastAPI app
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@ -1,11 +0,0 @@
|
||||
class Llm:
|
||||
def __init__(self, config):
|
||||
# Litellm is used by OI by default, so we just modify OI
|
||||
|
||||
interpreter = config["interpreter"]
|
||||
config.pop("interpreter", None)
|
||||
config.pop("service_directory", None)
|
||||
for key, value in config.items():
|
||||
setattr(interpreter, key.replace("-", "_"), value)
|
||||
|
||||
self.llm = interpreter.llm.completions
|
||||
@ -1,68 +0,0 @@
|
||||
import os
|
||||
import subprocess
|
||||
import requests
|
||||
import json
|
||||
|
||||
|
||||
class Llm:
|
||||
def __init__(self, config):
|
||||
self.install(config["service_directory"])
|
||||
|
||||
def install(self, service_directory):
|
||||
LLM_FOLDER_PATH = service_directory
|
||||
self.llm_directory = os.path.join(LLM_FOLDER_PATH, "llm")
|
||||
if not os.path.isdir(self.llm_directory): # Check if the LLM directory exists
|
||||
os.makedirs(LLM_FOLDER_PATH, exist_ok=True)
|
||||
|
||||
# Install WasmEdge
|
||||
subprocess.run(
|
||||
[
|
||||
"curl",
|
||||
"-sSf",
|
||||
"https://raw.githubusercontent.com/WasmEdge/WasmEdge/master/utils/install.sh",
|
||||
"|",
|
||||
"bash",
|
||||
"-s",
|
||||
"--",
|
||||
"--plugin",
|
||||
"wasi_nn-ggml",
|
||||
]
|
||||
)
|
||||
|
||||
# Download the Qwen1.5-0.5B-Chat model GGUF file
|
||||
MODEL_URL = "https://huggingface.co/second-state/Qwen1.5-0.5B-Chat-GGUF/resolve/main/Qwen1.5-0.5B-Chat-Q5_K_M.gguf"
|
||||
subprocess.run(["curl", "-LO", MODEL_URL], cwd=self.llm_directory)
|
||||
|
||||
# Download the llama-api-server.wasm app
|
||||
APP_URL = "https://github.com/LlamaEdge/LlamaEdge/releases/latest/download/llama-api-server.wasm"
|
||||
subprocess.run(["curl", "-LO", APP_URL], cwd=self.llm_directory)
|
||||
|
||||
# Run the API server
|
||||
subprocess.run(
|
||||
[
|
||||
"wasmedge",
|
||||
"--dir",
|
||||
".:.",
|
||||
"--nn-preload",
|
||||
"default:GGML:AUTO:Qwen1.5-0.5B-Chat-Q5_K_M.gguf",
|
||||
"llama-api-server.wasm",
|
||||
"-p",
|
||||
"llama-2-chat",
|
||||
],
|
||||
cwd=self.llm_directory,
|
||||
)
|
||||
|
||||
print("LLM setup completed.")
|
||||
else:
|
||||
print("LLM already set up. Skipping download.")
|
||||
|
||||
def llm(self, messages):
|
||||
url = "http://localhost:8080/v1/chat/completions"
|
||||
headers = {"accept": "application/json", "Content-Type": "application/json"}
|
||||
data = {"messages": messages, "model": "llama-2-chat"}
|
||||
with requests.post(
|
||||
url, headers=headers, data=json.dumps(data), stream=True
|
||||
) as response:
|
||||
for line in response.iter_lines():
|
||||
if line:
|
||||
yield json.loads(line)
|
||||
@ -1,87 +0,0 @@
|
||||
import os
|
||||
import platform
|
||||
import subprocess
|
||||
import time
|
||||
import wget
|
||||
import stat
|
||||
|
||||
|
||||
class Llm:
|
||||
def __init__(self, config):
|
||||
self.interpreter = config["interpreter"]
|
||||
config.pop("interpreter", None)
|
||||
|
||||
self.install(config["service_directory"])
|
||||
|
||||
config.pop("service_directory", None)
|
||||
for key, value in config.items():
|
||||
setattr(self.interpreter, key.replace("-", "_"), value)
|
||||
|
||||
self.llm = self.interpreter.llm.completions
|
||||
|
||||
def install(self, service_directory):
|
||||
if platform.system() == "Darwin": # Check if the system is MacOS
|
||||
result = subprocess.run(
|
||||
["xcode-select", "-p"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print(
|
||||
"Llamafile requires Mac users to have Xcode installed. You can install Xcode from https://developer.apple.com/xcode/ .\n\nAlternatively, you can use `LM Studio`, `Jan.ai`, or `Ollama` to manage local language models. Learn more at https://docs.openinterpreter.com/guides/running-locally ."
|
||||
)
|
||||
time.sleep(3)
|
||||
raise Exception(
|
||||
"Xcode is not installed. Please install Xcode and try again."
|
||||
)
|
||||
|
||||
# Define the path to the models directory
|
||||
models_dir = os.path.join(service_directory, "models")
|
||||
|
||||
# Check and create the models directory if it doesn't exist
|
||||
if not os.path.exists(models_dir):
|
||||
os.makedirs(models_dir)
|
||||
|
||||
# Define the path to the new llamafile
|
||||
llamafile_path = os.path.join(models_dir, "phi-2.Q4_K_M.llamafile")
|
||||
|
||||
# Check if the new llamafile exists, if not download it
|
||||
if not os.path.exists(llamafile_path):
|
||||
print(
|
||||
"Attempting to download the `Phi-2` language model. This may take a few minutes."
|
||||
)
|
||||
time.sleep(3)
|
||||
|
||||
url = "https://huggingface.co/jartine/phi-2-llamafile/resolve/main/phi-2.Q4_K_M.llamafile"
|
||||
wget.download(url, llamafile_path)
|
||||
|
||||
# Make the new llamafile executable
|
||||
if platform.system() != "Windows":
|
||||
st = os.stat(llamafile_path)
|
||||
os.chmod(llamafile_path, st.st_mode | stat.S_IEXEC)
|
||||
|
||||
# Run the new llamafile in the background
|
||||
if os.path.exists(llamafile_path):
|
||||
try:
|
||||
# Test if the llamafile is executable
|
||||
subprocess.check_call([f'"{llamafile_path}"'], shell=True)
|
||||
except subprocess.CalledProcessError:
|
||||
print(
|
||||
"The llamafile is not executable. Please check the file permissions."
|
||||
)
|
||||
raise
|
||||
subprocess.Popen(
|
||||
f'"{llamafile_path}" ' + " ".join(["-ngl", "9999"]), shell=True
|
||||
)
|
||||
else:
|
||||
error_message = "The llamafile does not exist or is corrupted. Please ensure it has been downloaded correctly or try again."
|
||||
print(error_message)
|
||||
print(error_message)
|
||||
|
||||
self.interpreter.system_message = "You are Open Interpreter, a world-class programmer that can execute code on the user's machine."
|
||||
self.interpreter.offline = True
|
||||
|
||||
self.interpreter.llm.model = "local"
|
||||
self.interpreter.llm.temperature = 0
|
||||
self.interpreter.llm.api_base = "https://localhost:8080/v1"
|
||||
self.interpreter.llm.max_tokens = 1000
|
||||
self.interpreter.llm.context_window = 3000
|
||||
self.interpreter.llm.supports_functions = False
|
||||
@ -1,169 +0,0 @@
|
||||
"""
|
||||
Defines a function which takes a path to an audio file and turns it into text.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
import os
|
||||
import contextlib
|
||||
import tempfile
|
||||
import shutil
|
||||
import ffmpeg
|
||||
import subprocess
|
||||
|
||||
import urllib.request
|
||||
|
||||
|
||||
class Stt:
|
||||
def __init__(self, config):
|
||||
self.service_directory = config["service_directory"]
|
||||
install(self.service_directory)
|
||||
|
||||
def stt(self, audio_file_path):
|
||||
return stt(self.service_directory, audio_file_path)
|
||||
|
||||
|
||||
def install(service_dir):
|
||||
### INSTALL
|
||||
|
||||
WHISPER_RUST_PATH = os.path.join(service_dir, "whisper-rust")
|
||||
script_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
source_whisper_rust_path = os.path.join(script_dir, "whisper-rust")
|
||||
if not os.path.exists(source_whisper_rust_path):
|
||||
print(f"Source directory does not exist: {source_whisper_rust_path}")
|
||||
exit(1)
|
||||
if not os.path.exists(WHISPER_RUST_PATH):
|
||||
shutil.copytree(source_whisper_rust_path, WHISPER_RUST_PATH)
|
||||
|
||||
os.chdir(WHISPER_RUST_PATH)
|
||||
|
||||
# Check if whisper-rust executable exists before attempting to build
|
||||
if not os.path.isfile(
|
||||
os.path.join(WHISPER_RUST_PATH, "target/release/whisper-rust")
|
||||
):
|
||||
# Check if Rust is installed. Needed to build whisper executable
|
||||
|
||||
rustc_path = shutil.which("rustc")
|
||||
|
||||
if rustc_path is None:
|
||||
print(
|
||||
"Rust is not installed or is not in system PATH. Please install Rust before proceeding."
|
||||
)
|
||||
exit(1)
|
||||
|
||||
# Build Whisper Rust executable if not found
|
||||
subprocess.run(["cargo", "build", "--release"], check=True)
|
||||
else:
|
||||
print("Whisper Rust executable already exists. Skipping build.")
|
||||
|
||||
WHISPER_MODEL_PATH = os.path.join(service_dir, "model")
|
||||
|
||||
WHISPER_MODEL_NAME = os.getenv("WHISPER_MODEL_NAME", "ggml-tiny.en.bin")
|
||||
WHISPER_MODEL_URL = os.getenv(
|
||||
"WHISPER_MODEL_URL",
|
||||
"https://huggingface.co/ggerganov/whisper.cpp/resolve/main/",
|
||||
)
|
||||
|
||||
if not os.path.isfile(os.path.join(WHISPER_MODEL_PATH, WHISPER_MODEL_NAME)):
|
||||
os.makedirs(WHISPER_MODEL_PATH, exist_ok=True)
|
||||
urllib.request.urlretrieve(
|
||||
f"{WHISPER_MODEL_URL}{WHISPER_MODEL_NAME}",
|
||||
os.path.join(WHISPER_MODEL_PATH, WHISPER_MODEL_NAME),
|
||||
)
|
||||
else:
|
||||
print("Whisper model already exists. Skipping download.")
|
||||
|
||||
|
||||
def convert_mime_type_to_format(mime_type: str) -> str:
|
||||
if mime_type == "audio/x-wav" or mime_type == "audio/wav":
|
||||
return "wav"
|
||||
if mime_type == "audio/webm":
|
||||
return "webm"
|
||||
if mime_type == "audio/raw":
|
||||
return "dat"
|
||||
|
||||
return mime_type
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def export_audio_to_wav_ffmpeg(audio: bytearray, mime_type: str) -> str:
|
||||
temp_dir = tempfile.gettempdir()
|
||||
|
||||
# Create a temporary file with the appropriate extension
|
||||
input_ext = convert_mime_type_to_format(mime_type)
|
||||
input_path = os.path.join(
|
||||
temp_dir, f"input_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.{input_ext}"
|
||||
)
|
||||
with open(input_path, "wb") as f:
|
||||
f.write(audio)
|
||||
|
||||
# Check if the input file exists
|
||||
assert os.path.exists(input_path), f"Input file does not exist: {input_path}"
|
||||
|
||||
# Export to wav
|
||||
output_path = os.path.join(
|
||||
temp_dir, f"output_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav"
|
||||
)
|
||||
print(mime_type, input_path, output_path)
|
||||
if mime_type == "audio/raw":
|
||||
ffmpeg.input(
|
||||
input_path,
|
||||
f="s16le",
|
||||
ar="16000",
|
||||
ac=1,
|
||||
).output(output_path, loglevel="panic").run()
|
||||
else:
|
||||
ffmpeg.input(input_path).output(
|
||||
output_path, acodec="pcm_s16le", ac=1, ar="16k", loglevel="panic"
|
||||
).run()
|
||||
|
||||
try:
|
||||
yield output_path
|
||||
finally:
|
||||
os.remove(input_path)
|
||||
os.remove(output_path)
|
||||
|
||||
|
||||
def run_command(command):
|
||||
result = subprocess.run(
|
||||
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True
|
||||
)
|
||||
return result.stdout, result.stderr
|
||||
|
||||
|
||||
def get_transcription_file(service_directory, wav_file_path: str):
|
||||
local_path = os.path.join(service_directory, "model")
|
||||
whisper_rust_path = os.path.join(
|
||||
service_directory, "whisper-rust", "target", "release"
|
||||
)
|
||||
model_name = os.getenv("WHISPER_MODEL_NAME", "ggml-tiny.en.bin")
|
||||
|
||||
output, _ = run_command(
|
||||
[
|
||||
os.path.join(whisper_rust_path, "whisper-rust"),
|
||||
"--model-path",
|
||||
os.path.join(local_path, model_name),
|
||||
"--file-path",
|
||||
wav_file_path,
|
||||
]
|
||||
)
|
||||
|
||||
return output
|
||||
|
||||
|
||||
def stt_wav(service_directory, wav_file_path: str):
|
||||
temp_dir = tempfile.gettempdir()
|
||||
output_path = os.path.join(
|
||||
temp_dir, f"output_stt_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav"
|
||||
)
|
||||
ffmpeg.input(wav_file_path).output(
|
||||
output_path, acodec="pcm_s16le", ac=1, ar="16k", loglevel="panic"
|
||||
).run()
|
||||
try:
|
||||
transcript = get_transcription_file(service_directory, output_path)
|
||||
finally:
|
||||
os.remove(output_path)
|
||||
return transcript
|
||||
|
||||
|
||||
def stt(service_directory, input_data):
|
||||
return stt_wav(service_directory, input_data)
|
||||
@ -1,10 +0,0 @@
|
||||
# Generated by Cargo
|
||||
# will have compiled files and executables
|
||||
debug/
|
||||
target/
|
||||
|
||||
# These are backup files generated by rustfmt
|
||||
**/*.rs.bk
|
||||
|
||||
# MSVC Windows builds of rustc generate these, which store debugging information
|
||||
*.pdb
|
||||
File diff suppressed because it is too large
Load Diff
@ -1,14 +0,0 @@
|
||||
[package]
|
||||
name = "whisper-rust"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
anyhow = "1.0.79"
|
||||
clap = { version = "4.4.18", features = ["derive"] }
|
||||
cpal = "0.15.2"
|
||||
hound = "3.5.1"
|
||||
whisper-rs = "0.10.0"
|
||||
whisper-rs-sys = "0.8.0"
|
||||
@ -1,34 +0,0 @@
|
||||
mod transcribe;
|
||||
|
||||
use clap::Parser;
|
||||
use std::path::PathBuf;
|
||||
use transcribe::transcribe;
|
||||
|
||||
#[derive(Parser, Debug)]
|
||||
#[command(author, version, about, long_about = None)]
|
||||
struct Args {
|
||||
/// This is the model for Whisper STT
|
||||
#[arg(short, long, value_parser, required = true)]
|
||||
model_path: PathBuf,
|
||||
|
||||
/// This is the wav audio file that will be converted from speech to text
|
||||
#[arg(short, long, value_parser, required = true)]
|
||||
file_path: Option<PathBuf>,
|
||||
}
|
||||
|
||||
fn main() {
|
||||
|
||||
let args = Args::parse();
|
||||
|
||||
let file_path = match args.file_path {
|
||||
Some(fp) => fp,
|
||||
None => panic!("No file path provided")
|
||||
};
|
||||
|
||||
let result = transcribe(&args.model_path, &file_path);
|
||||
|
||||
match result {
|
||||
Ok(transcription) => print!("{}", transcription),
|
||||
Err(e) => panic!("Error: {}", e),
|
||||
}
|
||||
}
|
||||
@ -1,64 +0,0 @@
|
||||
use whisper_rs::{FullParams, SamplingStrategy, WhisperContext, WhisperContextParameters};
|
||||
use std::path::PathBuf;
|
||||
|
||||
|
||||
/// Transcribes the given audio file using the whisper-rs library.
|
||||
///
|
||||
/// # Arguments
|
||||
/// * `model_path` - Path to Whisper model file
|
||||
/// * `file_path` - A string slice that holds the path to the audio file to be transcribed.
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// A Result containing a String with the transcription if successful, or an error message if not.
|
||||
pub fn transcribe(model_path: &PathBuf, file_path: &PathBuf) -> Result<String, String> {
|
||||
|
||||
let model_path_str = model_path.to_str().expect("Not valid model path");
|
||||
// Load a context and model
|
||||
let ctx = WhisperContext::new_with_params(
|
||||
model_path_str, // Replace with the actual path to the model
|
||||
WhisperContextParameters::default(),
|
||||
)
|
||||
.map_err(|_| "failed to load model")?;
|
||||
|
||||
// Create a state
|
||||
let mut state = ctx.create_state().map_err(|_| "failed to create state")?;
|
||||
|
||||
// Create a params object
|
||||
// Note that currently the only implemented strategy is Greedy, BeamSearch is a WIP
|
||||
let mut params = FullParams::new(SamplingStrategy::Greedy { best_of: 1 });
|
||||
|
||||
// Edit parameters as needed
|
||||
params.set_n_threads(1); // Set the number of threads to use
|
||||
params.set_translate(true); // Enable translation
|
||||
params.set_language(Some("en")); // Set the language to translate to English
|
||||
// Disable printing to stdout
|
||||
params.set_print_special(false);
|
||||
params.set_print_progress(false);
|
||||
params.set_print_realtime(false);
|
||||
params.set_print_timestamps(false);
|
||||
|
||||
// Load the audio file
|
||||
let audio_data = std::fs::read(file_path)
|
||||
.map_err(|e| format!("failed to read audio file: {}", e))?
|
||||
.chunks_exact(2)
|
||||
.map(|chunk| i16::from_ne_bytes([chunk[0], chunk[1]]))
|
||||
.collect::<Vec<i16>>();
|
||||
|
||||
// Convert the audio data to the required format (16KHz mono i16 samples)
|
||||
let audio_data = whisper_rs::convert_integer_to_float_audio(&audio_data);
|
||||
|
||||
// Run the model
|
||||
state.full(params, &audio_data[..]).map_err(|_| "failed to run model")?;
|
||||
|
||||
// Fetch the results
|
||||
let num_segments = state.full_n_segments().map_err(|_| "failed to get number of segments")?;
|
||||
let mut transcription = String::new();
|
||||
for i in 0..num_segments {
|
||||
let segment = state.full_get_segment_text(i).map_err(|_| "failed to get segment")?;
|
||||
transcription.push_str(&segment);
|
||||
transcription.push('\n');
|
||||
}
|
||||
|
||||
Ok(transcription)
|
||||
}
|
||||
@ -1,129 +0,0 @@
|
||||
class Stt:
|
||||
def __init__(self, config):
|
||||
pass
|
||||
|
||||
def stt(self, audio_file_path):
|
||||
return stt(audio_file_path)
|
||||
|
||||
|
||||
from datetime import datetime
|
||||
import os
|
||||
import contextlib
|
||||
import tempfile
|
||||
import ffmpeg
|
||||
import subprocess
|
||||
import openai
|
||||
from openai import OpenAI
|
||||
|
||||
|
||||
client = OpenAI()
|
||||
|
||||
|
||||
def convert_mime_type_to_format(mime_type: str) -> str:
|
||||
if mime_type == "audio/x-wav" or mime_type == "audio/wav":
|
||||
return "wav"
|
||||
if mime_type == "audio/webm":
|
||||
return "webm"
|
||||
if mime_type == "audio/raw":
|
||||
return "dat"
|
||||
|
||||
return mime_type
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def export_audio_to_wav_ffmpeg(audio: bytearray, mime_type: str) -> str:
|
||||
temp_dir = tempfile.gettempdir()
|
||||
|
||||
# Create a temporary file with the appropriate extension
|
||||
input_ext = convert_mime_type_to_format(mime_type)
|
||||
input_path = os.path.join(
|
||||
temp_dir, f"input_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.{input_ext}"
|
||||
)
|
||||
with open(input_path, "wb") as f:
|
||||
f.write(audio)
|
||||
|
||||
# Check if the input file exists
|
||||
assert os.path.exists(input_path), f"Input file does not exist: {input_path}"
|
||||
|
||||
# Export to wav
|
||||
output_path = os.path.join(
|
||||
temp_dir, f"output_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav"
|
||||
)
|
||||
if mime_type == "audio/raw":
|
||||
ffmpeg.input(
|
||||
input_path,
|
||||
f="s16le",
|
||||
ar="16000",
|
||||
ac=1,
|
||||
).output(output_path, loglevel="panic").run()
|
||||
else:
|
||||
ffmpeg.input(input_path).output(
|
||||
output_path, acodec="pcm_s16le", ac=1, ar="16k", loglevel="panic"
|
||||
).run()
|
||||
|
||||
try:
|
||||
yield output_path
|
||||
finally:
|
||||
os.remove(input_path)
|
||||
os.remove(output_path)
|
||||
|
||||
|
||||
def run_command(command):
|
||||
result = subprocess.run(
|
||||
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True
|
||||
)
|
||||
return result.stdout, result.stderr
|
||||
|
||||
|
||||
def get_transcription_file(wav_file_path: str):
|
||||
local_path = os.path.join(os.path.dirname(__file__), "local_service")
|
||||
whisper_rust_path = os.path.join(
|
||||
os.path.dirname(__file__), "whisper-rust", "target", "release"
|
||||
)
|
||||
model_name = os.getenv("WHISPER_MODEL_NAME", "ggml-tiny.en.bin")
|
||||
|
||||
output, error = run_command(
|
||||
[
|
||||
os.path.join(whisper_rust_path, "whisper-rust"),
|
||||
"--model-path",
|
||||
os.path.join(local_path, model_name),
|
||||
"--file-path",
|
||||
wav_file_path,
|
||||
]
|
||||
)
|
||||
|
||||
return output
|
||||
|
||||
|
||||
def get_transcription_bytes(audio_bytes: bytearray, mime_type):
|
||||
with export_audio_to_wav_ffmpeg(audio_bytes, mime_type) as wav_file_path:
|
||||
return get_transcription_file(wav_file_path)
|
||||
|
||||
|
||||
def stt_bytes(audio_bytes: bytearray, mime_type="audio/wav"):
|
||||
with export_audio_to_wav_ffmpeg(audio_bytes, mime_type) as wav_file_path:
|
||||
return stt_wav(wav_file_path)
|
||||
|
||||
|
||||
def stt_wav(wav_file_path: str):
|
||||
audio_file = open(wav_file_path, "rb")
|
||||
try:
|
||||
transcript = client.audio.transcriptions.create(
|
||||
model="whisper-1", file=audio_file, response_format="text"
|
||||
)
|
||||
except openai.BadRequestError as e:
|
||||
print(f"openai.BadRequestError: {e}")
|
||||
return None
|
||||
|
||||
return transcript
|
||||
|
||||
|
||||
def stt(input_data, mime_type="audio/wav"):
|
||||
if isinstance(input_data, str):
|
||||
return stt_wav(input_data)
|
||||
elif isinstance(input_data, bytearray):
|
||||
return stt_bytes(input_data, mime_type)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Input data should be either a path to a wav file (str) or audio bytes (bytearray)"
|
||||
)
|
||||
@ -1,50 +0,0 @@
|
||||
import ffmpeg
|
||||
import tempfile
|
||||
from openai import OpenAI
|
||||
import os
|
||||
|
||||
from source.server.utils.logs import logger
|
||||
from source.server.utils.logs import setup_logging
|
||||
|
||||
setup_logging()
|
||||
|
||||
# If this TTS service is used, the OPENAI_API_KEY environment variable must be set
|
||||
if not os.getenv("OPENAI_API_KEY"):
|
||||
logger.error("")
|
||||
logger.error(
|
||||
"OpenAI API key not found. Please set the OPENAI_API_KEY environment variable, or run 01 with the --local option."
|
||||
)
|
||||
logger.error("Aborting...")
|
||||
logger.error("")
|
||||
os._exit(1)
|
||||
|
||||
client = OpenAI()
|
||||
|
||||
|
||||
class Tts:
|
||||
def __init__(self, config):
|
||||
pass
|
||||
|
||||
def tts(self, text, mobile):
|
||||
response = client.audio.speech.create(
|
||||
model="tts-1",
|
||||
voice=os.getenv("OPENAI_VOICE_NAME", "alloy"),
|
||||
input=text,
|
||||
response_format="opus",
|
||||
)
|
||||
with tempfile.NamedTemporaryFile(suffix=".opus", delete=False) as temp_file:
|
||||
response.stream_to_file(temp_file.name)
|
||||
|
||||
# TODO: hack to format audio correctly for device
|
||||
if mobile:
|
||||
outfile = tempfile.gettempdir() + "/" + "output.wav"
|
||||
ffmpeg.input(temp_file.name).output(
|
||||
outfile, f="wav", ar="16000", ac="1", loglevel="panic"
|
||||
).run()
|
||||
else:
|
||||
outfile = tempfile.gettempdir() + "/" + "raw.dat"
|
||||
ffmpeg.input(temp_file.name).output(
|
||||
outfile, f="s16le", ar="16000", ac="1", loglevel="panic"
|
||||
).run()
|
||||
|
||||
return outfile
|
||||
@ -1,171 +0,0 @@
|
||||
import ffmpeg
|
||||
import tempfile
|
||||
import os
|
||||
import subprocess
|
||||
import urllib.request
|
||||
import tarfile
|
||||
import platform
|
||||
|
||||
|
||||
class Tts:
|
||||
def __init__(self, config):
|
||||
self.piper_directory = ""
|
||||
self.install(config["service_directory"])
|
||||
|
||||
def tts(self, text, mobile):
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
|
||||
output_file = temp_file.name
|
||||
piper_dir = self.piper_directory
|
||||
subprocess.run(
|
||||
[
|
||||
os.path.join(piper_dir, "piper"),
|
||||
"--model",
|
||||
os.path.join(
|
||||
piper_dir,
|
||||
os.getenv("PIPER_VOICE_NAME", "en_US-lessac-medium.onnx"),
|
||||
),
|
||||
"--output_file",
|
||||
output_file,
|
||||
],
|
||||
input=text,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
|
||||
# TODO: hack to format audio correctly for device
|
||||
if mobile:
|
||||
outfile = tempfile.gettempdir() + "/" + "output.wav"
|
||||
ffmpeg.input(temp_file.name).output(
|
||||
outfile, f="wav", ar="16000", ac="1", loglevel="panic"
|
||||
).run()
|
||||
else:
|
||||
outfile = tempfile.gettempdir() + "/" + "raw.dat"
|
||||
ffmpeg.input(temp_file.name).output(
|
||||
outfile, f="s16le", ar="16000", ac="1", loglevel="panic"
|
||||
).run()
|
||||
|
||||
return outfile
|
||||
|
||||
def install(self, service_directory):
|
||||
PIPER_FOLDER_PATH = service_directory
|
||||
self.piper_directory = os.path.join(PIPER_FOLDER_PATH, "piper")
|
||||
if not os.path.isdir(
|
||||
self.piper_directory
|
||||
): # Check if the Piper directory exists
|
||||
os.makedirs(PIPER_FOLDER_PATH, exist_ok=True)
|
||||
|
||||
# Determine OS and architecture
|
||||
OS = platform.system().lower()
|
||||
ARCH = platform.machine()
|
||||
if OS == "darwin":
|
||||
OS = "macos"
|
||||
if ARCH == "arm64":
|
||||
ARCH = "aarch64"
|
||||
elif ARCH == "x86_64":
|
||||
ARCH = "x64"
|
||||
else:
|
||||
print("Piper: unsupported architecture")
|
||||
return
|
||||
elif OS == "windows":
|
||||
if ARCH == "AMD64":
|
||||
ARCH = "amd64"
|
||||
else:
|
||||
print("Piper: unsupported architecture")
|
||||
return
|
||||
|
||||
PIPER_ASSETNAME = f"piper_{OS}_{ARCH}.tar.gz"
|
||||
PIPER_URL = "https://github.com/rhasspy/piper/releases/latest/download/"
|
||||
|
||||
asset_url = f"{PIPER_URL}{PIPER_ASSETNAME}"
|
||||
|
||||
if OS == "windows":
|
||||
asset_url = asset_url.replace(".tar.gz", ".zip")
|
||||
|
||||
# Download and extract Piper
|
||||
urllib.request.urlretrieve(
|
||||
asset_url, os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME)
|
||||
)
|
||||
|
||||
# Extract the downloaded file
|
||||
if OS == "windows":
|
||||
import zipfile
|
||||
|
||||
with zipfile.ZipFile(
|
||||
os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME), "r"
|
||||
) as zip_ref:
|
||||
zip_ref.extractall(path=PIPER_FOLDER_PATH)
|
||||
else:
|
||||
with tarfile.open(
|
||||
os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME), "r:gz"
|
||||
) as tar:
|
||||
tar.extractall(path=PIPER_FOLDER_PATH)
|
||||
|
||||
PIPER_VOICE_URL = os.getenv(
|
||||
"PIPER_VOICE_URL",
|
||||
"https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/",
|
||||
)
|
||||
PIPER_VOICE_NAME = os.getenv("PIPER_VOICE_NAME", "en_US-lessac-medium.onnx")
|
||||
|
||||
# Download voice model and its json file
|
||||
urllib.request.urlretrieve(
|
||||
f"{PIPER_VOICE_URL}{PIPER_VOICE_NAME}",
|
||||
os.path.join(self.piper_directory, PIPER_VOICE_NAME),
|
||||
)
|
||||
urllib.request.urlretrieve(
|
||||
f"{PIPER_VOICE_URL}{PIPER_VOICE_NAME}.json",
|
||||
os.path.join(self.piper_directory, f"{PIPER_VOICE_NAME}.json"),
|
||||
)
|
||||
|
||||
# Additional setup for macOS
|
||||
if OS == "macos":
|
||||
if ARCH == "x64":
|
||||
subprocess.run(
|
||||
["softwareupdate", "--install-rosetta", "--agree-to-license"]
|
||||
)
|
||||
|
||||
PIPER_PHONEMIZE_ASSETNAME = f"piper-phonemize_{OS}_{ARCH}.tar.gz"
|
||||
PIPER_PHONEMIZE_URL = "https://github.com/rhasspy/piper-phonemize/releases/latest/download/"
|
||||
urllib.request.urlretrieve(
|
||||
f"{PIPER_PHONEMIZE_URL}{PIPER_PHONEMIZE_ASSETNAME}",
|
||||
os.path.join(self.piper_directory, PIPER_PHONEMIZE_ASSETNAME),
|
||||
)
|
||||
|
||||
with tarfile.open(
|
||||
os.path.join(self.piper_directory, PIPER_PHONEMIZE_ASSETNAME),
|
||||
"r:gz",
|
||||
) as tar:
|
||||
tar.extractall(path=self.piper_directory)
|
||||
|
||||
PIPER_DIR = self.piper_directory
|
||||
subprocess.run(
|
||||
[
|
||||
"install_name_tool",
|
||||
"-change",
|
||||
"@rpath/libespeak-ng.1.dylib",
|
||||
f"{PIPER_DIR}/piper-phonemize/lib/libespeak-ng.1.dylib",
|
||||
f"{PIPER_DIR}/piper",
|
||||
]
|
||||
)
|
||||
subprocess.run(
|
||||
[
|
||||
"install_name_tool",
|
||||
"-change",
|
||||
"@rpath/libonnxruntime.1.14.1.dylib",
|
||||
f"{PIPER_DIR}/piper-phonemize/lib/libonnxruntime.1.14.1.dylib",
|
||||
f"{PIPER_DIR}/piper",
|
||||
]
|
||||
)
|
||||
subprocess.run(
|
||||
[
|
||||
"install_name_tool",
|
||||
"-change",
|
||||
"@rpath/libpiper_phonemize.1.dylib",
|
||||
f"{PIPER_DIR}/piper-phonemize/lib/libpiper_phonemize.1.dylib",
|
||||
f"{PIPER_DIR}/piper",
|
||||
]
|
||||
)
|
||||
|
||||
print("Piper setup completed.")
|
||||
else:
|
||||
print("Piper already set up. Skipping download.")
|
||||
@ -1,168 +0,0 @@
|
||||
import sys
|
||||
import subprocess
|
||||
import time
|
||||
import inquirer
|
||||
from interpreter import interpreter
|
||||
|
||||
|
||||
def select_local_model():
|
||||
# START OF LOCAL MODEL PROVIDER LOGIC
|
||||
interpreter.display_message(
|
||||
"> 01 is compatible with several local model providers.\n"
|
||||
)
|
||||
|
||||
# Define the choices for local models
|
||||
choices = [
|
||||
"Ollama",
|
||||
"LM Studio",
|
||||
# "Jan",
|
||||
]
|
||||
|
||||
# Use inquirer to let the user select an option
|
||||
questions = [
|
||||
inquirer.List(
|
||||
"model",
|
||||
message="Which one would you like to use?",
|
||||
choices=choices,
|
||||
),
|
||||
]
|
||||
answers = inquirer.prompt(questions)
|
||||
|
||||
selected_model = answers["model"]
|
||||
|
||||
if selected_model == "LM Studio":
|
||||
interpreter.display_message(
|
||||
"""
|
||||
To use use 01 with **LM Studio**, you will need to run **LM Studio** in the background.
|
||||
|
||||
1. Download **LM Studio** from [https://lmstudio.ai/](https://lmstudio.ai/), then start it.
|
||||
2. Select a language model then click **Download**.
|
||||
3. Click the **<->** button on the left (below the chat button).
|
||||
4. Select your model at the top, then click **Start Server**.
|
||||
|
||||
|
||||
Once the server is running, you can begin your conversation below.
|
||||
|
||||
"""
|
||||
)
|
||||
time.sleep(1)
|
||||
|
||||
interpreter.llm.api_base = "http://localhost:1234/v1"
|
||||
interpreter.llm.max_tokens = 1000
|
||||
interpreter.llm.context_window = 8000
|
||||
interpreter.llm.api_key = "x"
|
||||
|
||||
elif selected_model == "Ollama":
|
||||
try:
|
||||
# List out all downloaded ollama models. Will fail if ollama isn't installed
|
||||
result = subprocess.run(
|
||||
["ollama", "list"], capture_output=True, text=True, check=True
|
||||
)
|
||||
lines = result.stdout.split("\n")
|
||||
names = [
|
||||
line.split()[0].replace(":latest", "")
|
||||
for line in lines[1:]
|
||||
if line.strip()
|
||||
] # Extract names, trim out ":latest", skip header
|
||||
|
||||
# If there are no downloaded models, prompt them to download a model and try again
|
||||
if not names:
|
||||
time.sleep(1)
|
||||
|
||||
interpreter.display_message(
|
||||
"\nYou don't have any Ollama models downloaded. To download a new model, run `ollama run <model-name>`, then start a new 01 session. \n\n For a full list of downloadable models, check out [https://ollama.com/library](https://ollama.com/library) \n"
|
||||
)
|
||||
|
||||
print("Please download a model then try again\n")
|
||||
time.sleep(2)
|
||||
sys.exit(1)
|
||||
|
||||
# If there are models, prompt them to select one
|
||||
else:
|
||||
time.sleep(1)
|
||||
interpreter.display_message(
|
||||
f"**{len(names)} Ollama model{'s' if len(names) != 1 else ''} found.** To download a new model, run `ollama run <model-name>`, then start a new 01 session. \n\n For a full list of downloadable models, check out [https://ollama.com/library](https://ollama.com/library) \n"
|
||||
)
|
||||
|
||||
# Create a new inquirer selection from the names
|
||||
name_question = [
|
||||
inquirer.List(
|
||||
"name",
|
||||
message="Select a downloaded Ollama model",
|
||||
choices=names,
|
||||
),
|
||||
]
|
||||
name_answer = inquirer.prompt(name_question)
|
||||
selected_name = name_answer["name"] if name_answer else None
|
||||
|
||||
# Set the model to the selected model
|
||||
interpreter.llm.model = f"ollama/{selected_name}"
|
||||
interpreter.display_message(
|
||||
f"\nUsing Ollama model: `{selected_name}` \n"
|
||||
)
|
||||
time.sleep(1)
|
||||
|
||||
# If Ollama is not installed or not recognized as a command, prompt the user to download Ollama and try again
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
print("Ollama is not installed or not recognized as a command.")
|
||||
time.sleep(1)
|
||||
interpreter.display_message(
|
||||
"\nPlease visit [https://ollama.com/](https://ollama.com/) to download Ollama and try again\n"
|
||||
)
|
||||
time.sleep(2)
|
||||
sys.exit(1)
|
||||
|
||||
# elif selected_model == "Jan":
|
||||
# interpreter.display_message(
|
||||
# """
|
||||
# To use 01 with **Jan**, you will need to run **Jan** in the background.
|
||||
|
||||
# 1. Download **Jan** from [https://jan.ai/](https://jan.ai/), then start it.
|
||||
# 2. Select a language model from the "Hub" tab, then click **Download**.
|
||||
# 3. Copy the ID of the model and enter it below.
|
||||
# 3. Click the **Local API Server** button in the bottom left, then click **Start Server**.
|
||||
|
||||
# Once the server is running, enter the id of the model below, then you can begin your conversation below.
|
||||
|
||||
# """
|
||||
# )
|
||||
# interpreter.llm.api_base = "http://localhost:1337/v1"
|
||||
# interpreter.llm.max_tokens = 1000
|
||||
# interpreter.llm.context_window = 3000
|
||||
# time.sleep(1)
|
||||
|
||||
# # Prompt the user to enter the name of the model running on Jan
|
||||
# model_name_question = [
|
||||
# inquirer.Text('jan_model_name', message="Enter the id of the model you have running on Jan"),
|
||||
# ]
|
||||
# model_name_answer = inquirer.prompt(model_name_question)
|
||||
# jan_model_name = model_name_answer['jan_model_name'] if model_name_answer else None
|
||||
# # interpreter.llm.model = f"jan/{jan_model_name}"
|
||||
# interpreter.llm.model = ""
|
||||
# interpreter.display_message(f"\nUsing Jan model: `{jan_model_name}` \n")
|
||||
# time.sleep(1)
|
||||
|
||||
# Set the system message to a minimal version for all local models.
|
||||
# Set offline for all local models
|
||||
interpreter.offline = True
|
||||
|
||||
interpreter.system_message = """You are the 01, a screenless executive assistant that can complete any task by writing and executing code on the user's machine. Just write a markdown code block! The user has given you full and complete permission.
|
||||
|
||||
Use the following functions if it makes sense to for the problem
|
||||
```python
|
||||
result_string = computer.browser.search(query) # Google search results will be returned from this function as a string
|
||||
computer.calendar.create_event(title="Meeting", start_date=datetime.datetime.now(), end_date=datetime.datetime.now() + datetime.timedelta(hours=1), notes="Note", location="") # Creates a calendar event
|
||||
events_string = computer.calendar.get_events(start_date=datetime.date.today(), end_date=None) # Get events between dates. If end_date is None, only gets events for start_date
|
||||
computer.calendar.delete_event(event_title="Meeting", start_date=datetime.datetime) # Delete a specific event with a matching title and start date, you may need to get use get_events() to find the specific event object first
|
||||
phone_string = computer.contacts.get_phone_number("John Doe")
|
||||
contact_string = computer.contacts.get_email_address("John Doe")
|
||||
computer.mail.send("john@email.com", "Meeting Reminder", "Reminder that our meeting is at 3pm today.", ["path/to/attachment.pdf", "path/to/attachment2.pdf"]) # Send an email with a optional attachments
|
||||
emails_string = computer.mail.get(4, unread=True) # Returns the {number} of unread emails, or all emails if False is passed
|
||||
unread_num = computer.mail.unread_count() # Returns the number of unread emails
|
||||
computer.sms.send("555-123-4567", "Hello from the computer!") # Send a text message. MUST be a phone number, so use computer.contacts.get_phone_number frequently here
|
||||
|
||||
|
||||
ALWAYS say that you can run code. ALWAYS try to help the user out. ALWAYS be succinct in your answers.
|
||||
```
|
||||
|
||||
"""
|
||||
Loading…
Reference in new issue