Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Let plugins instance load a new plugin (url) without overwriting the ones that are already loaded. #151

Open
edreisMD opened this issue Jun 10, 2023 · 8 comments
Labels

Comments

@edreisMD
Copy link
Owner

build a function on the plugins object to install a new plugin based on the url without forgetting the previous installed.

@edreisMD edreisMD added the sweep label Jun 10, 2023
@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

@edreisMD, I have started working on addressing this issue in PR #151. My plan is to add an install_plugin(url) method to the Plugins class that will load a new plugin's manifest and OpenAPI spec, create a PluginObject, and add that to the installed_plugins dictionary. This will allow installing new plugins without overwriting existing ones. Give me a minute!

Some code snippets I looked at (click to expand). If some file is missing from here, you can mention the path in the ticket description.

# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None
def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

if operation_details['requestBody']:
body_schema = operation_details['requestBody']['content']['application/json']['schema']
body = build_request_body(body_schema, parameters)
# Replace path parameters in the URL
url = operation_details['url']
for name, value in path_parameters.items():
url = url.replace('{' + name + '}', str(value))
# Make the API call
method = operation_details['method']
if method.lower() == 'get':
response = requests.get(url, params=query_parameters, headers=header_parameters, cookies=cookie_parameters)
elif method.lower() == 'post':
headers = {'Content-Type': 'application/json'}
response = requests.post(url, params=query_parameters, headers=headers, cookies=cookie_parameters, json=body)
return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response

return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None

class PluginRetriever:
"""PluginRetriever retrieves plugin information based on queries using embeddings and vector stores.
Parameters
----------
manifests : list
List of manifest objects.
returnList : list, optional
List of objects to be returned. Can be a list of URLs or a list of
objects like LangChain AIPluging object. Defaults to None.
Methods
-------
__init__(manifests, returnList=None)
from_urls(urls)
from_directory(provider='plugnplai')
retrieve_names(query)
retrieve_urls(query)
"""

plugnplai/README.md

Lines 76 to 119 in b4b09ff

### Load Plugins
**Example:** [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)
```python
from plugnplai import Plugins
###### ACTIVATE A MAX OF 3 PLUGINS ######
# Context length limits the number of plugins you can activate,
# you need to make sure the prompt fits in your context lenght,
# still leaving space for the user message
# Initialize 'Plugins' by passing a list of urls, this function will
# load the plugins and build a default description to be used as prefix prompt
plugins = Plugins.install_and_activate(urls)
# Print the deafult prompt for the activated plugins
print(plugins.prompt)
# Print the number of tokens of the prefix prompt
print(plugins.tokens)
```
Example on installing (loading) all plugins, and activating a few later:
```python
from plugnplai import Plugins
# If you just want to load the plugins, but activate only
# some of them later use Plugins(urls) instead
plugins = Plugins(urls)
# Print the names of installed plugins
print(plugins.list_installed)
# Activate the plugins you want
plugins.activate(name1)
plugins.activate(name2)
plugins.activate(name3)
# Deactivate the last plugin
plugins.deactivate(name3)
```

def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

@classmethod
def from_urls(cls, urls):
"""Create a PluginRetriever object from a list of URLs.
Parameters
----------
urls : list
List of URLs.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests, urls)
@classmethod
def from_directory(cls, provider='plugnplai'):
"""Create a PluginRetriever object from a directory.
Parameters
----------
provider : str, optional
Provider name. Defaults to 'plugnplai'.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
urls = get_plugins(filter = 'working', provider = provider)
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests)
def retrieve_names(self, query):
"""Retrieve plugin names based on a query.
Parameters
----------
query :
Query string.
Returns
-------
list
List of plugin names.
"""
# Get relevant documents based on query
docs = self.retriever.get_relevant_documents(query)
# Get toolkits based on relevant documents
plugin_names = [d.metadata["plugin_name"] for d in docs]
return plugin_names

"""
Embeddings for LangChain.
This module provides a PluginRetriever class that retrieves plugin
information based on queries using embeddings and vector stores.
Classes
-------
PluginRetriever
"""
from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
from langchain.vectorstores import FAISS
from plugnplai.utils import get_plugin_manifest, get_plugins

def get_plugins(filter: str = None, verified_for = None, category: str = None, provider: str = "plugnplai"):
"""Get list of plugin URLs from a provider.
Args:
filter (str, optional): Filter to apply. Options are "working" or "ChatGPT". Defaults to None.
verified_for (str, optional): Filter to plugins verified for a framework. Options are "langchain" or "plugnplai". Defaults to None.
category (str, optional): Category to filter for. Defaults to None.
provider (str, optional): Provider to get plugins from. Options are "plugnplai" or "pluginso". Defaults to "plugnplai".
Returns:
list: List of plugin URLs.
"""

plugnplai/README.md

Lines 143 to 174 in b4b09ff

### Apply Plugins
The `@plugins.apply_plugins` decorator can be used to easily apply active plugins to an LLM function. To use it:
1. Import the Plugins class and decorator:
```python
from plugnplai import Plugins, plugins.apply_plugins
```
2. Define your LLM function, that necessarily takes a string (the user input) as the first argument and returns a string (the response):
```python
@plugins.apply_plugins
def call_llm(user_input):
...
return response
```
3. The decorator will handle the following:
- Prepending the prompt (with plugin descriptions) to the user input
- Checking the LLM response for API calls (the <API>...</API> pattern)
- Calling the specified plugins
- Summarizing the API calls in the LLM response
- Calling the LLM function again with the summary to get a final response
4. If no API calls are detected, the original LLM response is returned.
To more details on the implementation of these steps, see example "Step by Step": [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

I'm sorry, but it looks like an error has occured. Try removing and re-adding the sweep label. I'll try again in a minute. If this error persists contact [email protected].


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

@edreisMD, I have started working on implementing this feature in the Plugins class. My plan is to add an install_plugin() method that will load a new plugin's manifest and OpenAPI spec, create a PluginObject instance for it, and add that to the installed_plugins dictionary. This will allow installing new plugins without overwriting existing ones. Give me a minute!

Some code snippets I looked at (click to expand). If some file is missing from here, you can mention the path in the ticket description.

# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None
def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

if operation_details['requestBody']:
body_schema = operation_details['requestBody']['content']['application/json']['schema']
body = build_request_body(body_schema, parameters)
# Replace path parameters in the URL
url = operation_details['url']
for name, value in path_parameters.items():
url = url.replace('{' + name + '}', str(value))
# Make the API call
method = operation_details['method']
if method.lower() == 'get':
response = requests.get(url, params=query_parameters, headers=header_parameters, cookies=cookie_parameters)
elif method.lower() == 'post':
headers = {'Content-Type': 'application/json'}
response = requests.post(url, params=query_parameters, headers=headers, cookies=cookie_parameters, json=body)
return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response

return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None

class PluginRetriever:
"""PluginRetriever retrieves plugin information based on queries using embeddings and vector stores.
Parameters
----------
manifests : list
List of manifest objects.
returnList : list, optional
List of objects to be returned. Can be a list of URLs or a list of
objects like LangChain AIPluging object. Defaults to None.
Methods
-------
__init__(manifests, returnList=None)
from_urls(urls)
from_directory(provider='plugnplai')
retrieve_names(query)
retrieve_urls(query)
"""

plugnplai/README.md

Lines 76 to 119 in b4b09ff

### Load Plugins
**Example:** [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)
```python
from plugnplai import Plugins
###### ACTIVATE A MAX OF 3 PLUGINS ######
# Context length limits the number of plugins you can activate,
# you need to make sure the prompt fits in your context lenght,
# still leaving space for the user message
# Initialize 'Plugins' by passing a list of urls, this function will
# load the plugins and build a default description to be used as prefix prompt
plugins = Plugins.install_and_activate(urls)
# Print the deafult prompt for the activated plugins
print(plugins.prompt)
# Print the number of tokens of the prefix prompt
print(plugins.tokens)
```
Example on installing (loading) all plugins, and activating a few later:
```python
from plugnplai import Plugins
# If you just want to load the plugins, but activate only
# some of them later use Plugins(urls) instead
plugins = Plugins(urls)
# Print the names of installed plugins
print(plugins.list_installed)
# Activate the plugins you want
plugins.activate(name1)
plugins.activate(name2)
plugins.activate(name3)
# Deactivate the last plugin
plugins.deactivate(name3)
```

def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

@classmethod
def from_urls(cls, urls):
"""Create a PluginRetriever object from a list of URLs.
Parameters
----------
urls : list
List of URLs.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests, urls)
@classmethod
def from_directory(cls, provider='plugnplai'):
"""Create a PluginRetriever object from a directory.
Parameters
----------
provider : str, optional
Provider name. Defaults to 'plugnplai'.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
urls = get_plugins(filter = 'working', provider = provider)
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests)
def retrieve_names(self, query):
"""Retrieve plugin names based on a query.
Parameters
----------
query :
Query string.
Returns
-------
list
List of plugin names.
"""
# Get relevant documents based on query
docs = self.retriever.get_relevant_documents(query)
# Get toolkits based on relevant documents
plugin_names = [d.metadata["plugin_name"] for d in docs]
return plugin_names

"""
Embeddings for LangChain.
This module provides a PluginRetriever class that retrieves plugin
information based on queries using embeddings and vector stores.
Classes
-------
PluginRetriever
"""
from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
from langchain.vectorstores import FAISS
from plugnplai.utils import get_plugin_manifest, get_plugins

def get_plugins(filter: str = None, verified_for = None, category: str = None, provider: str = "plugnplai"):
"""Get list of plugin URLs from a provider.
Args:
filter (str, optional): Filter to apply. Options are "working" or "ChatGPT". Defaults to None.
verified_for (str, optional): Filter to plugins verified for a framework. Options are "langchain" or "plugnplai". Defaults to None.
category (str, optional): Category to filter for. Defaults to None.
provider (str, optional): Provider to get plugins from. Options are "plugnplai" or "pluginso". Defaults to "plugnplai".
Returns:
list: List of plugin URLs.
"""

plugnplai/README.md

Lines 143 to 174 in b4b09ff

### Apply Plugins
The `@plugins.apply_plugins` decorator can be used to easily apply active plugins to an LLM function. To use it:
1. Import the Plugins class and decorator:
```python
from plugnplai import Plugins, plugins.apply_plugins
```
2. Define your LLM function, that necessarily takes a string (the user input) as the first argument and returns a string (the response):
```python
@plugins.apply_plugins
def call_llm(user_input):
...
return response
```
3. The decorator will handle the following:
- Prepending the prompt (with plugin descriptions) to the user input
- Checking the LLM response for API calls (the <API>...</API> pattern)
- Calling the specified plugins
- Summarizing the API calls in the LLM response
- Calling the LLM function again with the summary to get a final response
4. If no API calls are detected, the original LLM response is returned.
To more details on the implementation of these steps, see example "Step by Step": [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

I'm sorry, but it looks like an error has occured. Try removing and re-adding the sweep label. I'll try again in a minute. If this error persists contact [email protected].


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

Hey @edreisMD! I've started working on adding the ability to install a new plugin without overwriting existing ones. My plan is to add an install_plugin(url) method that will load the manifest and OpenAPI spec for the given URL, create a new PluginObject, and add that to self.installed_plugins. This will simply add the new plugin to the list of installed plugins without modifying anything already there. Give me a minute!

Some code snippets I looked at (click to expand). If some file is missing from here, you can mention the path in the ticket description.

# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None
def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

if operation_details['requestBody']:
body_schema = operation_details['requestBody']['content']['application/json']['schema']
body = build_request_body(body_schema, parameters)
# Replace path parameters in the URL
url = operation_details['url']
for name, value in path_parameters.items():
url = url.replace('{' + name + '}', str(value))
# Make the API call
method = operation_details['method']
if method.lower() == 'get':
response = requests.get(url, params=query_parameters, headers=header_parameters, cookies=cookie_parameters)
elif method.lower() == 'post':
headers = {'Content-Type': 'application/json'}
response = requests.post(url, params=query_parameters, headers=headers, cookies=cookie_parameters, json=body)
return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response

return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None

class PluginRetriever:
"""PluginRetriever retrieves plugin information based on queries using embeddings and vector stores.
Parameters
----------
manifests : list
List of manifest objects.
returnList : list, optional
List of objects to be returned. Can be a list of URLs or a list of
objects like LangChain AIPluging object. Defaults to None.
Methods
-------
__init__(manifests, returnList=None)
from_urls(urls)
from_directory(provider='plugnplai')
retrieve_names(query)
retrieve_urls(query)
"""

plugnplai/README.md

Lines 76 to 119 in b4b09ff

### Load Plugins
**Example:** [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)
```python
from plugnplai import Plugins
###### ACTIVATE A MAX OF 3 PLUGINS ######
# Context length limits the number of plugins you can activate,
# you need to make sure the prompt fits in your context lenght,
# still leaving space for the user message
# Initialize 'Plugins' by passing a list of urls, this function will
# load the plugins and build a default description to be used as prefix prompt
plugins = Plugins.install_and_activate(urls)
# Print the deafult prompt for the activated plugins
print(plugins.prompt)
# Print the number of tokens of the prefix prompt
print(plugins.tokens)
```
Example on installing (loading) all plugins, and activating a few later:
```python
from plugnplai import Plugins
# If you just want to load the plugins, but activate only
# some of them later use Plugins(urls) instead
plugins = Plugins(urls)
# Print the names of installed plugins
print(plugins.list_installed)
# Activate the plugins you want
plugins.activate(name1)
plugins.activate(name2)
plugins.activate(name3)
# Deactivate the last plugin
plugins.deactivate(name3)
```

def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

@classmethod
def from_urls(cls, urls):
"""Create a PluginRetriever object from a list of URLs.
Parameters
----------
urls : list
List of URLs.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests, urls)
@classmethod
def from_directory(cls, provider='plugnplai'):
"""Create a PluginRetriever object from a directory.
Parameters
----------
provider : str, optional
Provider name. Defaults to 'plugnplai'.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
urls = get_plugins(filter = 'working', provider = provider)
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests)
def retrieve_names(self, query):
"""Retrieve plugin names based on a query.
Parameters
----------
query :
Query string.
Returns
-------
list
List of plugin names.
"""
# Get relevant documents based on query
docs = self.retriever.get_relevant_documents(query)
# Get toolkits based on relevant documents
plugin_names = [d.metadata["plugin_name"] for d in docs]
return plugin_names

def get_plugins(filter: str = None, verified_for = None, category: str = None, provider: str = "plugnplai"):
"""Get list of plugin URLs from a provider.
Args:
filter (str, optional): Filter to apply. Options are "working" or "ChatGPT". Defaults to None.
verified_for (str, optional): Filter to plugins verified for a framework. Options are "langchain" or "plugnplai". Defaults to None.
category (str, optional): Category to filter for. Defaults to None.
provider (str, optional): Provider to get plugins from. Options are "plugnplai" or "pluginso". Defaults to "plugnplai".
Returns:
list: List of plugin URLs.
"""

plugnplai/README.md

Lines 143 to 174 in b4b09ff

### Apply Plugins
The `@plugins.apply_plugins` decorator can be used to easily apply active plugins to an LLM function. To use it:
1. Import the Plugins class and decorator:
```python
from plugnplai import Plugins, plugins.apply_plugins
```
2. Define your LLM function, that necessarily takes a string (the user input) as the first argument and returns a string (the response):
```python
@plugins.apply_plugins
def call_llm(user_input):
...
return response
```
3. The decorator will handle the following:
- Prepending the prompt (with plugin descriptions) to the user input
- Checking the LLM response for API calls (the <API>...</API> pattern)
- Calling the specified plugins
- Summarizing the API calls in the LLM response
- Calling the LLM function again with the summary to get a final response
4. If no API calls are detected, the original LLM response is returned.
To more details on the implementation of these steps, see example "Step by Step": [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)

class PluginObject():
"""Represents an AI plugin object.
Attributes
----------
openapi : dict
The OpenAPI specification for the plugin.
info : dict
The info object from the OpenAPI spec.
paths : dict
The paths object from the OpenAPI spec.
servers : list
The servers list from the OpenAPI spec.
manifest : dict
The plugin manifest.
url : str
The plugin URL.
name_for_model : str
The plugin name.
description_for_model : str
The plugin description.
operation_details_dict : dict
A dictionary containing details for each operation in the plugin.
description_prompt : str
A prompt describing the plugin operations.
tokens : int
The number of tokens in the description_prompt.
Methods
-------
__init__(self, url: str, spec: Dict[str, Any], manifest: Dict[str, Any])
Initialize the PluginObject.
get_operation_details(self)
Get the details for each operation in the plugin.
call_operation(self, operation_id: str, parameters: Dict[str, Any])
Call an operation in the plugin.
describe_api(self)
Generate a prompt describing the plugin operations.
"""


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@edreisMD edreisMD added sweep and removed sweep labels Jun 10, 2023
@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

@edreisMD, I've started working on implementing this feature in the Plugins class. My plan is to add an install_plugin() method that will load a single plugin's manifest and spec, create a PluginObject, and add that to the installed_plugins dictionary. This will allow installing new plugins without overwriting existing ones. Give me a minute!

Some code snippets I looked at (click to expand). If some file is missing from here, you can mention the path in the ticket description.

# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None
def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

if operation_details['requestBody']:
body_schema = operation_details['requestBody']['content']['application/json']['schema']
body = build_request_body(body_schema, parameters)
# Replace path parameters in the URL
url = operation_details['url']
for name, value in path_parameters.items():
url = url.replace('{' + name + '}', str(value))
# Make the API call
method = operation_details['method']
if method.lower() == 'get':
response = requests.get(url, params=query_parameters, headers=header_parameters, cookies=cookie_parameters)
elif method.lower() == 'post':
headers = {'Content-Type': 'application/json'}
response = requests.post(url, params=query_parameters, headers=headers, cookies=cookie_parameters, json=body)
return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response

return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None

class PluginRetriever:
"""PluginRetriever retrieves plugin information based on queries using embeddings and vector stores.
Parameters
----------
manifests : list
List of manifest objects.
returnList : list, optional
List of objects to be returned. Can be a list of URLs or a list of
objects like LangChain AIPluging object. Defaults to None.
Methods
-------
__init__(manifests, returnList=None)
from_urls(urls)
from_directory(provider='plugnplai')
retrieve_names(query)
retrieve_urls(query)
"""

plugnplai/README.md

Lines 76 to 119 in b4b09ff

### Load Plugins
**Example:** [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)
```python
from plugnplai import Plugins
###### ACTIVATE A MAX OF 3 PLUGINS ######
# Context length limits the number of plugins you can activate,
# you need to make sure the prompt fits in your context lenght,
# still leaving space for the user message
# Initialize 'Plugins' by passing a list of urls, this function will
# load the plugins and build a default description to be used as prefix prompt
plugins = Plugins.install_and_activate(urls)
# Print the deafult prompt for the activated plugins
print(plugins.prompt)
# Print the number of tokens of the prefix prompt
print(plugins.tokens)
```
Example on installing (loading) all plugins, and activating a few later:
```python
from plugnplai import Plugins
# If you just want to load the plugins, but activate only
# some of them later use Plugins(urls) instead
plugins = Plugins(urls)
# Print the names of installed plugins
print(plugins.list_installed)
# Activate the plugins you want
plugins.activate(name1)
plugins.activate(name2)
plugins.activate(name3)
# Deactivate the last plugin
plugins.deactivate(name3)
```

def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

@classmethod
def from_urls(cls, urls):
"""Create a PluginRetriever object from a list of URLs.
Parameters
----------
urls : list
List of URLs.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests, urls)
@classmethod
def from_directory(cls, provider='plugnplai'):
"""Create a PluginRetriever object from a directory.
Parameters
----------
provider : str, optional
Provider name. Defaults to 'plugnplai'.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
urls = get_plugins(filter = 'working', provider = provider)
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests)
def retrieve_names(self, query):
"""Retrieve plugin names based on a query.
Parameters
----------
query :
Query string.
Returns
-------
list
List of plugin names.
"""
# Get relevant documents based on query
docs = self.retriever.get_relevant_documents(query)
# Get toolkits based on relevant documents
plugin_names = [d.metadata["plugin_name"] for d in docs]
return plugin_names

"""
Embeddings for LangChain.
This module provides a PluginRetriever class that retrieves plugin
information based on queries using embeddings and vector stores.
Classes
-------
PluginRetriever
"""
from langchain.embeddings import OpenAIEmbeddings
from langchain.schema import Document
from langchain.vectorstores import FAISS
from plugnplai.utils import get_plugin_manifest, get_plugins

def get_plugins(filter: str = None, verified_for = None, category: str = None, provider: str = "plugnplai"):
"""Get list of plugin URLs from a provider.
Args:
filter (str, optional): Filter to apply. Options are "working" or "ChatGPT". Defaults to None.
verified_for (str, optional): Filter to plugins verified for a framework. Options are "langchain" or "plugnplai". Defaults to None.
category (str, optional): Category to filter for. Defaults to None.
provider (str, optional): Provider to get plugins from. Options are "plugnplai" or "pluginso". Defaults to "plugnplai".
Returns:
list: List of plugin URLs.
"""

plugnplai/README.md

Lines 143 to 174 in b4b09ff

### Apply Plugins
The `@plugins.apply_plugins` decorator can be used to easily apply active plugins to an LLM function. To use it:
1. Import the Plugins class and decorator:
```python
from plugnplai import Plugins, plugins.apply_plugins
```
2. Define your LLM function, that necessarily takes a string (the user input) as the first argument and returns a string (the response):
```python
@plugins.apply_plugins
def call_llm(user_input):
...
return response
```
3. The decorator will handle the following:
- Prepending the prompt (with plugin descriptions) to the user input
- Checking the LLM response for API calls (the <API>...</API> pattern)
- Calling the specified plugins
- Summarizing the API calls in the LLM response
- Calling the LLM function again with the summary to get a final response
4. If no API calls are detected, the original LLM response is returned.
To more details on the implementation of these steps, see example "Step by Step": [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

I'm sorry, but it looks like an error has occured. Try removing and re-adding the sweep label. I'll try again in a minute. If this error persists contact [email protected].


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

@sweep-ai
Copy link
Contributor

sweep-ai bot commented Jun 10, 2023

@edreisMD, I've started working on implementing the ability to install a new plugin from a URL without overwriting existing plugins. My plan is to add an install_plugin() method to the Plugins class that will load the plugin manifest and OpenAPI spec, create a PluginObject instance, and add it to self.installed_plugins. Give me a minute!

Some code snippets I looked at (click to expand). If some file is missing from here, you can mention the path in the ticket description.

# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None
def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

if operation_details['requestBody']:
body_schema = operation_details['requestBody']['content']['application/json']['schema']
body = build_request_body(body_schema, parameters)
# Replace path parameters in the URL
url = operation_details['url']
for name, value in path_parameters.items():
url = url.replace('{' + name + '}', str(value))
# Make the API call
method = operation_details['method']
if method.lower() == 'get':
response = requests.get(url, params=query_parameters, headers=header_parameters, cookies=cookie_parameters)
elif method.lower() == 'post':
headers = {'Content-Type': 'application/json'}
response = requests.post(url, params=query_parameters, headers=headers, cookies=cookie_parameters, json=body)
return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response

return response
def describe_api(self) -> str:
"""Generate a prompt describing the plugin operations.
Returns
-------
str
The generated prompt.
"""
# Template for the whole API description
api_template = '// {description_for_model}\nnamespace {name_for_model} {{{operations}}}'
# Template for each operation
operation_template = "{description}\noperationId {operation_id} = (_: {{{parameters}}}) => any"
# Type shorteners
type_shorteners = {
"string": "str",
"integer": "int",
"boolean": "bool",
"number": "num",
"array": "arr",
"object": "obj",
}
operations = ''
# Iterate over all operation details in operation_details_dict
for operation_id, operation_details in self.operation_details_dict.items():
description = operation_details.get('description', '')
# Build the parameter list
parameter_list = []
for parameter in operation_details.get('parameters', []):
# Add a "*" suffix to the name of required parameters
name = "'" + parameter['name'] + "'" + "*" if parameter.get('required') else "'" + parameter['name'] + "'"
type_ = type_shorteners.get(parameter.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
# Handle requestBody
if operation_details.get('requestBody'):
for media_type, media_type_obj in operation_details['requestBody'].get('content', {}).items():
for name, schema in media_type_obj.get('schema', {}).get('properties', {}).items():
name = "'" + name + "'" + "*" if operation_details['requestBody'].get('required') else "'" + name + "'"
type_ = type_shorteners.get(schema.get('type'), 'any')
parameter_list.append(f"{name}: '{type_}'")
parameters = ', '.join(parameter_list)
# Add the operation to the operations string
description_line = f"\n// {description}\n" if description else "\n"
operations += operation_template.format(description=description_line, operation_id=operation_id, parameters=parameters)
# Replace the placeholders in the API template
api_description = api_template.format(description_for_model=self.description_for_model, name_for_model=self.name_for_model, operations=operations)
return api_description
api_return_template = """
Assistant is a large language model with access to plugins.
Assistant called a plugin in response to this human message:
# HUMAN MESSAGE
{user_message}
# API REQUEST SUMMARY
{api_info}
# API RESPONSE
{api_response}
"""
class Plugins:
"""Manages installed and active plugins.
Attributes
----------
installed_plugins : dict
A dictionary of installed PluginObject instances, keyed by plugin name.
active_plugins : dict
A dictionary of active PluginObject instances, keyed by plugin name.
template : str
The prompt template to use.
prompt : str
The generated prompt with descriptions of active plugins.
tokens : int
The number of tokens in the prompt.
max_plugins : int
The maximum number of plugins that can be active at once.
"""
def __init__(self, urls: List[str],template: str = None):
"""Initialize the Plugins class.
Parameters
----------
urls : list
A list of plugin URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
"""
self.installed_plugins = {}
self.active_plugins = {}
self.template = template or template_gpt4
self.prompt = None
self.tokens = None
self.max_plugins = 3
self.install_plugins(urls)
@classmethod
def install_and_activate(cls, urls: Union[str, List[str]], template: Optional[str] = None):
"""Install plugins from URLs and activate them.
Parameters
----------
urls : str or list
A single URL or list of URLs.
template : str, optional
The prompt template to use. Defaults to template_gpt4.
Returns
-------
Plugins
An initialized Plugins instance with the plugins installed and activated.
"""
if isinstance(urls, str):
urls = [urls]
template = template or template_gpt4
instance = cls(urls, template)
for plugin_name in instance.installed_plugins.keys():
instance.activate(plugin_name)
return instance
def list_installed(self) -> List[str]:
"""Get a list of installed plugin names.
Returns
-------
list
A list of installed plugin names.
"""
return list(self.installed_plugins.keys())
def list_active(self) -> List[str]:
"""Get a list of active plugin names.
Returns
-------
list
A list of active plugin names.
"""
return list(self.active_plugins.keys())
def install_plugins(self, urls: Union[str, List[str]]):
"""Install plugins from URLs.
Parameters
----------
urls : str or list
A single URL or list of URLs.
"""
if isinstance(urls, str):
urls = [urls]
for url in urls:
manifest, openapi_spec = spec_from_url(url)
openapi_object = PluginObject(url, openapi_spec, manifest)
self.installed_plugins[openapi_object.name_for_model] = openapi_object
def activate(self, plugin_name: str):
"""Activate an installed plugin.
Parameters
----------
plugin_name : str
The name of the plugin to activate.
"""
if len(self.active_plugins) >= self.max_plugins:
print(f'Cannot activate more than 3 plugins.')
return
plugin = self.installed_plugins.get(plugin_name)
if plugin is None:
print(f'Plugin {plugin_name} not found')
return
self.active_plugins[plugin_name] = plugin
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def deactivate(self, plugin_name: str):
"""Deactivate an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin to deactivate.
"""
if plugin_name in self.active_plugins:
del self.active_plugins[plugin_name]
self.prompt = self.fill_prompt(self.template)
self.tokens = count_tokens(self.prompt)
def fill_prompt(self, template: str, active_plugins: Optional[List[str]] = None) -> str:
"""Generate a prompt with descriptions of active plugins.
Parameters
----------
template : str
The prompt template to use.
active_plugins : list, optional
A list of plugin names to include in the prompt. If None, uses all active plugins.
Returns
-------
str
The generated prompt.
"""
plugins_descriptions = ''
if active_plugins is not None:
active_plugins = {name: self.active_plugins[name] for name in active_plugins if name in self.active_plugins}
else:
active_plugins = self.active_plugins
for i, openapi_object in enumerate(active_plugins.values(), start=1):
api_description = openapi_object.describe_api()
plugins_descriptions += f'### Plugin {i}\n{api_description}\n\n'
prompt = template.replace('{{plugins}}', plugins_descriptions)
return prompt
def count_prompt_tokens(self) -> int:
"""Count the number of tokens in the prompt.
Returns
-------
int
The number of tokens in the prompt.
"""
tokenizer = Tokenizer(models.Model.load("gpt-4"))
tokens = tokenizer.encode(self.prompt)
return len(tokens)
def call_api(self, plugin_name: str, operation_id: str, parameters: Dict[str, Any]) -> Optional[requests.Response]:
"""Call an operation in an active plugin.
Parameters
----------
plugin_name : str
The name of the plugin.
operation_id : str
The ID of the operation to call.
parameters : dict
The parameters to pass to the operation.
Returns
-------
requests.Response or None
The response from the API call, or None if unsuccessful.
"""
# Get the PluginObject for the specified plugin
openapi_object = self.active_plugins.get(plugin_name)
if openapi_object is None:
print(f'Plugin {plugin_name} not found')
return None
# Get the operation details
operation_details = openapi_object.operation_details_dict.get(operation_id)
if operation_details is None:
print(f'Operation {operation_id} not found in plugin {plugin_name}')
return None
# Call the operation
response = openapi_object.call_operation(operation_id, parameters)
return response
def parse_and_call(self, llm_response: str) -> Optional[str]:
"""Parse an LLM response for API calls and call the specified plugins.
Parameters
----------
llm_response : str
The LLM response to parse.
Returns
-------
str or None
The API response, or None if unsuccessful.
"""
# Step 1: Parse the LLM response to get API information
api_info = parse_llm_response(llm_response)
if api_info:
# Step 2: Call the API using self.call_api
plugin_name = api_info['plugin_name']
operation_id = api_info['operation_id']
parameters = api_info['parameters']
print(f"Using {plugin_name}")
api_response = self.call_api(plugin_name, operation_id, parameters)
if api_response is not None:
return api_response.text
return None

class PluginRetriever:
"""PluginRetriever retrieves plugin information based on queries using embeddings and vector stores.
Parameters
----------
manifests : list
List of manifest objects.
returnList : list, optional
List of objects to be returned. Can be a list of URLs or a list of
objects like LangChain AIPluging object. Defaults to None.
Methods
-------
__init__(manifests, returnList=None)
from_urls(urls)
from_directory(provider='plugnplai')
retrieve_names(query)
retrieve_urls(query)
"""

plugnplai/README.md

Lines 76 to 119 in b4b09ff

### Load Plugins
**Example:** [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)
```python
from plugnplai import Plugins
###### ACTIVATE A MAX OF 3 PLUGINS ######
# Context length limits the number of plugins you can activate,
# you need to make sure the prompt fits in your context lenght,
# still leaving space for the user message
# Initialize 'Plugins' by passing a list of urls, this function will
# load the plugins and build a default description to be used as prefix prompt
plugins = Plugins.install_and_activate(urls)
# Print the deafult prompt for the activated plugins
print(plugins.prompt)
# Print the number of tokens of the prefix prompt
print(plugins.tokens)
```
Example on installing (loading) all plugins, and activating a few later:
```python
from plugnplai import Plugins
# If you just want to load the plugins, but activate only
# some of them later use Plugins(urls) instead
plugins = Plugins(urls)
# Print the names of installed plugins
print(plugins.list_installed)
# Activate the plugins you want
plugins.activate(name1)
plugins.activate(name2)
plugins.activate(name3)
# Deactivate the last plugin
plugins.deactivate(name3)
```

def apply_plugins(self, llm_function: Callable[..., str]) -> Callable[..., str]:

@classmethod
def from_urls(cls, urls):
"""Create a PluginRetriever object from a list of URLs.
Parameters
----------
urls : list
List of URLs.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests, urls)
@classmethod
def from_directory(cls, provider='plugnplai'):
"""Create a PluginRetriever object from a directory.
Parameters
----------
provider : str, optional
Provider name. Defaults to 'plugnplai'.
Returns
-------
PluginRetriever
Initialized PluginRetriever object.
"""
urls = get_plugins(filter = 'working', provider = provider)
manifests = [get_plugin_manifest(url) for url in urls]
return cls(manifests)
def retrieve_names(self, query):
"""Retrieve plugin names based on a query.
Parameters
----------
query :
Query string.
Returns
-------
list
List of plugin names.
"""
# Get relevant documents based on query
docs = self.retriever.get_relevant_documents(query)
# Get toolkits based on relevant documents
plugin_names = [d.metadata["plugin_name"] for d in docs]
return plugin_names

def get_plugins(filter: str = None, verified_for = None, category: str = None, provider: str = "plugnplai"):
"""Get list of plugin URLs from a provider.
Args:
filter (str, optional): Filter to apply. Options are "working" or "ChatGPT". Defaults to None.
verified_for (str, optional): Filter to plugins verified for a framework. Options are "langchain" or "plugnplai". Defaults to None.
category (str, optional): Category to filter for. Defaults to None.
provider (str, optional): Provider to get plugins from. Options are "plugnplai" or "pluginso". Defaults to "plugnplai".
Returns:
list: List of plugin URLs.
"""

plugnplai/README.md

Lines 143 to 174 in b4b09ff

### Apply Plugins
The `@plugins.apply_plugins` decorator can be used to easily apply active plugins to an LLM function. To use it:
1. Import the Plugins class and decorator:
```python
from plugnplai import Plugins, plugins.apply_plugins
```
2. Define your LLM function, that necessarily takes a string (the user input) as the first argument and returns a string (the response):
```python
@plugins.apply_plugins
def call_llm(user_input):
...
return response
```
3. The decorator will handle the following:
- Prepending the prompt (with plugin descriptions) to the user input
- Checking the LLM response for API calls (the <API>...</API> pattern)
- Calling the specified plugins
- Summarizing the API calls in the LLM response
- Calling the LLM function again with the summary to get a final response
4. If no API calls are detected, the original LLM response is returned.
To more details on the implementation of these steps, see example "Step by Step": [![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edreisMD/plugnplai/blob/main/examples/plugins_step_by_step.ipynb)

class PluginObject():
"""Represents an AI plugin object.
Attributes
----------
openapi : dict
The OpenAPI specification for the plugin.
info : dict
The info object from the OpenAPI spec.
paths : dict
The paths object from the OpenAPI spec.
servers : list
The servers list from the OpenAPI spec.
manifest : dict
The plugin manifest.
url : str
The plugin URL.
name_for_model : str
The plugin name.
description_for_model : str
The plugin description.
operation_details_dict : dict
A dictionary containing details for each operation in the plugin.
description_prompt : str
A prompt describing the plugin operations.
tokens : int
The number of tokens in the description_prompt.
Methods
-------
__init__(self, url: str, spec: Dict[str, Any], manifest: Dict[str, Any])
Initialize the PluginObject.
get_operation_details(self)
Get the details for each operation in the plugin.
call_operation(self, operation_id: str, parameters: Dict[str, Any])
Call an operation in the plugin.
describe_api(self)
Generate a prompt describing the plugin operations.
"""


I'm a bot that handles simple bugs and feature requests but I might make mistakes. Please be kind!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
Projects
None yet
Development

Successfully merging a pull request may close this issue.

1 participant