github/gpt-pilot
1
1
Fork 0
mirror of https://github.com/Pythagora-io/gpt-pilot.git synced 2026-01-09 13:17:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

ARCHITECTURE function_calls works on meta-llama/codellama-34b-instruct

Browse Source
This commit is contained in:
Nicholas Albion
2023-09-21 23:05:37 +10:00
parent 07f9842312
commit 5853b95ab9
8 changed files with 298 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.github/workflows/ci.yml vendored
View File

@@ -13,7 +13,9 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.8', '3.9', '3.10', '3.11']
# 3.10 - 04 Oct 2021
# 3.11 - 24 Oct 2022
python-version: ['3.11']
steps:
- uses: actions/checkout@v4

2
README.md
View File

@@ -51,7 +51,7 @@ https://github.com/Pythagora-io/gpt-pilot/assets/10895136/0495631b-511e-451b-93d
# 🔌 Requirements
- **Python**
- **Python >= 3.11**
- **PostgreSQL** (optional, projects default is SQLite)
- DB is needed for multiple reasons like continuing app development if you had to stop at any point or app crashed, going back to specific step so you can change some later steps in development, easier debugging, for future we will add functionality to update project (change some things in existing project or add new features to the project and so on)...

14
pilot/helpers/AgentConvo.py
View File

@@ -1,11 +1,10 @@
import json
import re
import subprocess
import uuid
from utils.style import yellow, yellow_bold
from database.database import get_saved_development_step, save_development_step, delete_all_subsequent_steps
from helpers.files import get_files_content
from const.common import IGNORE_FOLDERS
from helpers.exceptions.TokenLimitError import TokenLimitError
from utils.utils import array_of_objects_to_string, get_prompt
from utils.llm_connection import create_gpt_chat_completion
@@ -188,10 +187,17 @@ class AgentConvo:
"""
if 'function_calls' in response and function_calls is not None:
if 'send_convo' in function_calls:
response['function_calls']['arguments']['convo'] = self
response['function_calls']['arguments']['convo'] = self
response = function_calls['functions'][response['function_calls']['name']](**response['function_calls']['arguments'])
elif 'text' in response:
response = response['text']
if function_calls:
values = list(json.loads(response['text']).values())
if len(values) == 1:
return values[0]
else:
return tuple(values)
else:
response = response['text']
return response

2
pilot/prompts/system_messages/architect.prompt
View File

@@ -1,7 +1,7 @@
You are an experienced software architect. Your expertise is in creating an architecture for an MVP (minimum viable products) for {{ app_type }}s that can be developed as fast as possible by using as many ready-made technologies as possible. The technologies that you prefer using when other technologies are not explicitly specified are:
**Scripts**: you prefer using Node.js for writing scripts that are meant to be ran just with the CLI.
**Backend**: you prefer using Node.js with Mongo database if not explicitely specified otherwise. When you're using Mongo, you always use Mongoose and when you're using Postgresql, you always use PeeWee as an ORM.
**Backend**: you prefer using Node.js with Mongo database if not explicitly specified otherwise. When you're using Mongo, you always use Mongoose and when you're using Postgresql, you always use PeeWee as an ORM.
**Testing**: To create unit and integration tests, you prefer using Jest for Node.js projects and pytest for Python projects. To create end-to-end tests, you prefer using Cypress.

169
pilot/utils/function_calling.py Normal file
View File

@@ -0,0 +1,169 @@
import json
# from local_llm_function_calling import Generator
# from local_llm_function_calling.model.llama import LlamaModel
# from local_llm_function_calling.model.huggingface import HuggingfaceModel
from local_llm_function_calling.prompter import FunctionType, CompletionModelPrompter, InstructModelPrompter
# from local_llm_function_calling.model.llama import LlamaInstructPrompter
from typing import Literal, NotRequired, Protocol, TypeVar, TypedDict, Callable
class FunctionCallSet(TypedDict):
definitions: list[FunctionType]
functions: dict[str, Callable]
def add_function_calls_to_request(gpt_data, function_calls: FunctionCallSet | None):
if function_calls is None:
return
if gpt_data['model'] == 'gpt-4':
gpt_data['functions'] = function_calls['definitions']
if len(function_calls['definitions']) > 1:
gpt_data['function_call'] = 'auto'
else:
gpt_data['function_call'] = {'name': function_calls['definitions'][0]['name']}
return
# prompter = CompletionModelPrompter()
# prompter = InstructModelPrompter()
prompter = LlamaInstructPrompter()
if len(function_calls['definitions']) > 1:
function_call = None
else:
function_call = function_calls['definitions'][0]['name']
gpt_data['messages'].append({
'role': 'user',
'content': prompter.prompt('', function_calls['definitions'], function_call)
})
class LlamaInstructPrompter:
"""
A prompter for Llama2 instruct models.
Adapted from local_llm_function_calling
"""
def function_descriptions(
self, functions: list[FunctionType], function_to_call: str
) -> list[str]:
"""Get the descriptions of the functions
Args:
functions (list[FunctionType]): The functions to get the descriptions of
function_to_call (str): The function to call
Returns:
list[str]: The descriptions of the functions
(empty if the function doesn't exist or has no description)
"""
return [
"Function description: " + function["description"]
for function in functions
if function["name"] == function_to_call and "description" in function
]
def function_parameters(
self, functions: list[FunctionType], function_to_call: str
) -> str:
"""Get the parameters of the function
Args:
functions (list[FunctionType]): The functions to get the parameters of
function_to_call (str): The function to call
Returns:
str: The parameters of the function as a JSON schema
"""
return next(
json.dumps(function["parameters"]["properties"], indent=4)
for function in functions
if function["name"] == function_to_call
)
def function_data(
self, functions: list[FunctionType], function_to_call: str
) -> str:
"""Get the data for the function
Args:
functions (list[FunctionType]): The functions to get the data for
function_to_call (str): The function to call
Returns:
str: The data necessary to generate the arguments for the function
"""
return "\n".join(
self.function_descriptions(functions, function_to_call)
+ [
"Function parameters should follow this schema:",
"```jsonschema",
self.function_parameters(functions, function_to_call),
"```",
]
)
def function_summary(self, function: FunctionType) -> str:
"""Get a summary of a function
Args:
function (FunctionType): The function to get the summary of
Returns:
str: The summary of the function, as a bullet point
"""
return f"- {function['name']}" + (
f" - {function['description']}" if "description" in function else ""
)
def functions_summary(self, functions: list[FunctionType]) -> str:
"""Get a summary of the functions
Args:
functions (list[FunctionType]): The functions to get the summary of
Returns:
str: The summary of the functions, as a bulleted list
"""
return "Available functions:\n" + "\n".join(
self.function_summary(function) for function in functions
)
def prompt(
self,
prompt: str,
functions: list[FunctionType],
function_to_call: str | None = None,
) -> str:
"""Generate the llama prompt
Args:
prompt (str): The prompt to generate the response to
functions (list[FunctionType]): The functions to generate the response from
function_to_call (str | None): The function to call. Defaults to None.
Returns:
list[bytes | int]: The llama prompt, a function selection prompt if no
function is specified, or a function argument prompt if a function is
specified
"""
system = (
"Help choose the appropriate function to call to answer the user's question."
if function_to_call is None
else f"Define the arguments for {function_to_call} to answer the user's question."
) + "In your response you must only use JSON output and provide no notes or commentary."
data = (
self.function_data(functions, function_to_call)
if function_to_call
else self.functions_summary(functions)
)
response_start = (
f"Here are the arguments for the `{function_to_call}` function: ```json\n"
if function_to_call
else "Here's the function the user should call: "
)
return f"[INST] <<SYS>>\n{system}\n\n{data}\n<</SYS>>\n\n{prompt} [/INST]"
# {response_start}"

33
pilot/utils/llm_connection.py
View File

@@ -7,16 +7,14 @@ import json
import tiktoken
import questionary
from utils.style import red
from typing import List
from const.llm import MIN_TOKENS_FOR_GPT_RESPONSE, MAX_GPT_MODEL_TOKENS
from logger.logger import logger
from helpers.exceptions.TokenLimitError import TokenLimitError
from utils.utils import fix_json
model = os.getenv('MODEL_NAME')
endpoint = os.getenv('ENDPOINT')
from utils.function_calling import add_function_calls_to_request
def get_tokens_in_messages(messages: List[str]) -> int:
tokenizer = tiktoken.get_encoding("cl100k_base") # GPT-4 tokenizer
@@ -24,7 +22,7 @@ def get_tokens_in_messages(messages: List[str]) -> int:
return sum(len(tokens) for tokens in tokenized_messages)
def num_tokens_from_functions(functions, model=model):
def num_tokens_from_functions(functions):
"""Return the number of tokens used by a list of functions."""
encoding = tiktoken.get_encoding("cl100k_base")
@@ -96,13 +94,7 @@ def create_gpt_chat_completion(messages: List[dict], req_type, min_tokens=MIN_TO
if key in gpt_data:
del gpt_data[key]
if function_calls is not None:
# Advise the LLM of the JSON response schema we are expecting
gpt_data['functions'] = function_calls['definitions']
if len(function_calls['definitions']) > 1:
gpt_data['function_call'] = 'auto'
else:
gpt_data['function_call'] = {'name': function_calls['definitions'][0]['name']}
add_function_calls_to_request(gpt_data, function_calls)
try:
response = stream_gpt_completion(gpt_data, req_type)
@@ -110,7 +102,7 @@ def create_gpt_chat_completion(messages: List[dict], req_type, min_tokens=MIN_TO
except TokenLimitError as e:
raise e
except Exception as e:
print('The request to OpenAI API failed. Here is the error message:')
print(f'The request to {os.getenv("ENDPOINT")} API failed. Here is the error message:')
print(e)
@@ -126,6 +118,7 @@ def count_lines_based_on_width(content, width):
lines_required = sum(len(line) // width + 1 for line in content.split('\n'))
return lines_required
def get_tokens_in_messages_from_openai_error(error_message):
"""
Extract the token count from a message.
@@ -208,7 +201,10 @@ def stream_gpt_completion(data, req_type):
logger.info(f'Request data: {data}')
# Check if the ENDPOINT is AZURE
# Configure for the selected ENDPOINT
model = os.getenv('MODEL_NAME')
endpoint = os.getenv('ENDPOINT')
if endpoint == 'AZURE':
# If yes, get the AZURE_ENDPOINT from .ENV file
endpoint_url = os.getenv('AZURE_ENDPOINT') + '/openai/deployments/' + model + '/chat/completions?api-version=2023-05-15'
@@ -239,10 +235,9 @@ def stream_gpt_completion(data, req_type):
gpt_response = ''
function_calls = {'name': '', 'arguments': ''}
for line in response.iter_lines():
# Ignore keep-alive new lines
if line:
if line and line != b': OPENROUTER PROCESSING':
line = line.decode("utf-8") # decode the bytes to string
if line.startswith('data: '):
@@ -262,11 +257,13 @@ def stream_gpt_completion(data, req_type):
logger.error(f'Error in LLM response: {json_line}')
raise ValueError(f'Error in LLM response: {json_line["error"]["message"]}')
if json_line['choices'][0]['finish_reason'] == 'function_call':
choice = json_line['choices'][0]
if 'finish_reason' in choice and choice['finish_reason'] == 'function_call':
function_calls['arguments'] = load_data_to_json(function_calls['arguments'])
return return_result({'function_calls': function_calls}, lines_printed)
json_line = json_line['choices'][0]['delta']
json_line = choice['delta']
except json.JSONDecodeError:
logger.error(f'Unable to decode line: {line}')

106
pilot/utils/test_llm_connection.py
View File

@@ -1,9 +1,14 @@
import builtins
import os
from dotenv import load_dotenv
from const.function_calls import ARCHITECTURE
from unittest.mock import patch
from local_llm_function_calling.prompter import CompletionModelPrompter, InstructModelPrompter
from const.function_calls import ARCHITECTURE, DEV_STEPS
from helpers.AgentConvo import AgentConvo
from helpers.Project import Project
from helpers.agents.Architect import Architect
from helpers.agents.Developer import Developer
from .llm_connection import create_gpt_chat_completion
from main import get_custom_print
@@ -16,7 +21,31 @@ class TestLlmConnection:
def setup_method(self):
builtins.print, ipc_client_instance = get_custom_print({})
def test_chat_completion_Architect(self):
# def test_break_down_development_task(self):
# # Given
# agent = Developer(project)
# convo = AgentConvo(agent)
# # convo.construct_and_add_message_from_prompt('architecture/technologies.prompt',
# # {
# # 'name': 'Test App',
# # 'prompt': '''
#
# messages = convo.messages
# function_calls = DEV_STEPS
#
# # When
# # response = create_gpt_chat_completion(messages, '', function_calls=function_calls)
# response = {'function_calls': {
# 'name': 'break_down_development_task',
# 'arguments': {'tasks': [{'type': 'command', 'description': 'Run the app'}]}
# }}
# response = convo.postprocess_response(response, function_calls)
#
# # Then
# # assert len(convo.messages) == 2
# assert response == ([{'type': 'command', 'description': 'Run the app'}], 'more_tasks')
def test_chat_completion_Architect(self, monkeypatch):
"""Test the chat completion method."""
# Given
agent = Architect(project)
@@ -49,19 +78,80 @@ class TestLlmConnection:
})
messages = convo.messages
function_calls = ARCHITECTURE
endpoint = 'OPENROUTER'
# monkeypatch.setattr('utils.llm_connection.endpoint', endpoint)
monkeypatch.setenv('ENDPOINT', endpoint)
monkeypatch.setenv('MODEL_NAME', 'meta-llama/codellama-34b-instruct')
# with patch('.llm_connection.endpoint', endpoint):
# When
response = create_gpt_chat_completion(messages, '', function_calls=ARCHITECTURE)
response = create_gpt_chat_completion(messages, '', function_calls=function_calls)
# Then
assert len(convo.messages) == 2
assert convo.messages[0]['content'].startswith('You are an experienced software architect')
assert convo.messages[1]['content'].startswith('You are working in a software development agency')
assert response is not None
assert len(response) > 0
technologies: list[str] = response['function_calls']['arguments']['technologies']
assert 'Node.js' in technologies
assert response is not None
response = convo.postprocess_response(response, function_calls)
# response = response['function_calls']['arguments']['technologies']
assert 'Node.js' in response
def test_completion_function_prompt(self):
# Given
prompter = CompletionModelPrompter()
# When
prompt = prompter.prompt('Create a web-based chat app', ARCHITECTURE['definitions']) # , 'process_technologies')
# Then
assert prompt == '''Create a web-based chat app
Available functions:
process_technologies - Print the list of technologies that are created.
```jsonschema
{
"technologies": {
"type": "array",
"description": "List of technologies that are created in a list.",
"items": {
"type": "string",
"description": "technology"
}
}
}
```
Function call:
Function call: '''
def test_instruct_function_prompter(self):
# Given
prompter = InstructModelPrompter()
# When
prompt = prompter.prompt('Create a web-based chat app', ARCHITECTURE['definitions']) # , 'process_technologies')
# Then
assert prompt == '''Your task is to call a function when needed. You will be provided with a list of functions. Available functions:
process_technologies - Print the list of technologies that are created.
```jsonschema
{
"technologies": {
"type": "array",
"description": "List of technologies that are created in a list.",
"items": {
"type": "string",
"description": "technology"
}
}
}
```
Create a web-based chat app
Function call: '''
def _create_convo(self, agent):
convo = AgentConvo(agent)

1
requirements.txt
View File

@@ -4,6 +4,7 @@ charset-normalizer==3.2.0
distro==1.8.0
idna==3.4
Jinja2==3.1.2
local_llm_function_calling==0.1.14
MarkupSafe==2.1.3
peewee==3.16.2
prompt-toolkit==3.0.39