Store

Python module for dynamically updating tools and constructing AppletStore.

While we have existing functions in core for constructing dynamic tools from locally installed functions or API endpoints, we need to incorporate these utilities into functions compatible with our current GPT Function Calling framework. There are some problems complicating this matter, including:

Our llmcam.core.complete function operates with a tools instance (Python list) at the top level and does not pass on this Python pointer to utilities for function execution.
The tools management function should be dynamic and operate in real-time. This means that the interface for calling llmcam.core.complete should not change during application runs.
Pasting tools as text in messages for GPT to interpret is unreliable because generated tools schema may exceed maximum token and GPT may not reliably add the new tools.

Therefore, the goal in this notebook is to create a framework that allows a function to manage this top-level tools Python list via GPT Function calling. The function should extend the existing tools list or delete entries from this list. Considering that tools cannot be passed as Python instance in arguments of GPT tool call, we can utilize the fixup function to retrieve tools from global environment with appropriate metadata.

Dynamic function installation

Supposing that we have a Python function installed locally on our machine, it is possible to dynamically import with importlib.

source

add_function_tools

 add_function_tools (tools:list, service_name:str,
                     function_names:list[str])

Add function tools to the toolbox.

	Type	Details
tools	list	List of existing tools
service_name	str	Name of the service
function_names	list	List of function names (with module prefix)

Example usage with some functions from llmcam.vision modules:

tools = []
add_function_tools(
    tools, 
    "youtube_live", 
    [
        "llmcam.vision.ytlive.capture_youtube_live_frame", 
        "llmcam.vision.gpt4v.ask_gpt4v_about_image_file"
    ])
assert len(tools) == 2, "Tools should be updated with 2 new function."
tools

[{'type': 'function',
  'function': {'name': 'capture_youtube_live_frame',
   'description': 'Capture a jpeg image from YouTube Live and return the path to the saved image',
   'parameters': {'type': 'object',
    'properties': {'link': {'anyOf': [{'type': 'string',
        'description': 'YouTube Live link'},
       {'type': 'null',
        'description': 'A default value will be automatically used.'}]},
     'place': {'anyOf': [{'type': 'string',
        'description': 'Location of live image'},
       {'type': 'null',
        'description': 'A default value will be automatically used.'}]}},
    'required': []},
   'metadata': {'module': 'llmcam.vision.ytlive', 'service': 'youtube_live'}}},
 {'type': 'function',
  'function': {'name': 'ask_gpt4v_about_image_file',
   'description': 'Tell all about quantitative information from a given image file',
   'parameters': {'type': 'object',
    'properties': {'path': {'type': 'string',
      'description': 'Path to the image file'}},
    'required': ['path']},
   'metadata': {'module': 'llmcam.vision.gpt4v', 'service': 'youtube_live'}}}]

Dynamic API installation

With the functions in oas_to_schema, we can dynamically generate tools schema from API server with OpenAPI Specification file:

source

add_api_tools

 add_api_tools (tools:list, service_name:str, base_url:str,
                oas_url:Optional[str]=None,
                oas_destination:Optional[str]=None)

Add API tools to the toolbox.

	Type	Default	Details
tools	list		List of existing tools
service_name	str		Name of the API service
base_url	str		Base URL of the API service
oas_url	Optional	None	OpenAPI Specification URL
oas_destination	Optional	None	OpenAPI Specification destination file

Example usage with Road DigiTraffic API:

tools = []
add_api_tools(tools, "road_digitraffic", "https://tie.digitraffic.fi")
assert len(tools) > 0, "Tools should be updated with new API tools."
tools[0]

{'type': 'function',
 'function': {'name': 'tmsStationsDatex2Xml',
  'description': 'The static information of TMS stations in Datex2 format (Traffic Measurement System / LAM)',
  'parameters': {'type': 'object',
   'properties': {'query': {'type': 'object',
     'properties': {'state': {'type': 'string',
       'description': 'Road station state',
       'default': 'ACTIVE',
       'enum': ['ALL', 'REMOVED', 'ACTIVE']}},
     'required': []}},
   'required': []},
  'metadata': {'url': 'https://tie.digitraffic.fi/api/beta/tms-stations-datex2.xml',
   'method': 'get',
   'accepted_queries': ['state'],
   'service': 'road_digitraffic'},
  'fixup': 'llmcam.core.oas_to_schema.generate_request'}}

Removal of existing tools

Currently, there is an upper limit for the number of tools (around 128). Therefore, to better manage the toolbox, we can add an utility function to remove tools of via service name.

source

remove_tools

 remove_tools (tools:list, service_name:str)

Remove tools from the toolbox.

	Type	Details
tools	list	List of existing tools
service_name	str	Name of the service

Fixup setup and FC framework

As described previously, we can introduce a special fixup function is used for handlers which accesses the correct instance of tools. This function is to be created and attached to the tools instance at the moment of creation.

Because this fixup function should be attached to the tools instance during its creation, we should define a *_core function which operates the core operations for re-definition of execution function depending on the context rather than a fixed definition.

source

execute_handler_core

 execute_handler_core (tools:list, function_name:str, module:str,
                       **kwargs)

Execute the handler function by retrieving function with session ID.

	Type	Details
tools	list	Tools for each session
function_name	str	Name of the function to execute
module	str	Module of the function
kwargs	VAR_KEYWORD

source

handler_schema

 handler_schema (function:Callable, service_name:str='toolbox_handler',
                 fixup:Optional[Callable]=None, **kwargs)

Create a schema for handlers.

	Type	Default	Details
function	Callable		Handler function
service_name	str	toolbox_handler	Name of the service
fixup	Optional	None	Function to fixup function execution
kwargs	VAR_KEYWORD

Example usage with execution function adding tools from Road DigiTraffic:

tools = []

# Define the fixup function attached to this `tools` instance
def execute_handler(function_name, **kwargs):
    execute_handler_core(tools, function_name, **kwargs)

# Add the handler functions to the toolbox
tools.extend([handler_schema(function, service_name="toolbox_handler", fixup=execute_handler) for function in [
        add_api_tools,
        add_function_tools,
        remove_tools
    ]])
assert len(tools) == 3, "Tools should be updated with 3 new handler functions."
tools[0]

{'type': 'function',
 'function': {'name': 'add_api_tools',
  'description': 'Add API tools to the toolbox.',
  'parameters': {'type': 'object',
   'properties': {'service_name': {'type': 'string',
     'description': 'Name of the API service'},
    'base_url': {'type': 'string',
     'description': 'Base URL of the API service'},
    'oas_url': {'anyOf': [{'type': 'string',
       'description': 'OpenAPI Specification URL'},
      {'type': 'null',
       'description': 'A default value will be automatically used.'}]},
    'oas_destination': {'anyOf': [{'type': 'string',
       'description': 'OpenAPI Specification destination file'},
      {'type': 'null',
       'description': 'A default value will be automatically used.'}]}},
   'required': ['service_name', 'base_url']},
  'metadata': {'module': '__main__', 'service': 'toolbox_handler'},
  'fixup': '__main__.execute_handler'}}

execute_handler(
    "add_api_tools",
    module="__main__",
    service_name="road_digitraffic",
    base_url="https://tie.digitraffic.fi"
)
assert len(tools) > 3, "Tools should be updated with new API tools."
tools[4]

{'type': 'function',
 'function': {'name': 'tmsStationsDatex2Json',
  'description': 'The static information of TMS stations in Datex2 format (Traffic Measurement System / LAM)',
  'parameters': {'type': 'object',
   'properties': {'query': {'type': 'object',
     'properties': {'state': {'type': 'string',
       'description': 'Road station state',
       'default': 'ACTIVE',
       'enum': ['ALL', 'REMOVED', 'ACTIVE']}},
     'required': []}},
   'required': []},
  'metadata': {'url': 'https://tie.digitraffic.fi/api/beta/tms-stations-datex2.json',
   'method': 'get',
   'accepted_queries': ['state'],
   'service': 'road_digitraffic'},
  'fixup': 'llmcam.core.oas_to_schema.generate_request'}}

Simulated GPT workflow

Test integrating with our current GPT framework:

from llmcam.core.fc import *

tools = []
def execute_handler(function_name, **kwargs):
    execute_handler_core(tools, function_name, **kwargs)

tools.extend([handler_schema(function, service_name="toolbox_handler", fixup=execute_handler) for function in [
        add_api_tools,
        add_function_tools,
        remove_tools
    ]])
assert len(tools) == 3

messages = form_msgs([
    ("system", "You are a helpful system administrator. Use the supplied tools to assist the user."),
    ("user", "Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.")
])
complete(messages, tools=tools)
print_msgs(messages)

>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and
the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.
>> Assistant:
The API service 'road_digitraffic' has been successfully added.

len(tools)

messages.append(form_msg("user", "Get the weather camera information for the stations with ID C01503 and C01504."))
complete(messages, tools=tools)
print_msgs(messages)

>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and
the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.
>> Assistant:
The API service 'road_digitraffic' has been successfully added.
>> User:
Get the weather camera information for the stations with ID C01503 and C01504.
>> Assistant:
Here is the weather camera information for the stations with IDs C01503 and C01504:  ### Station
C01503 - kt51_Inkoo - **Coordinates**: [23.99616, 60.05374] - **Camera Type**: BOSCH - **Collection
Status**: GATHERING - **State**: FAULT_DOUBT - **Data Updated Time**: 2024-12-15T15:25:43Z -
**Municipality**: Inkoo - **Province**: Uusimaa - **Purpose**: Weather Observation - **Presets**:
- **Inkooseen**: ![View](https://weathercam.digitraffic.fi/C0150301.jpg)   - **Hankoon**:
![View](https://weathercam.digitraffic.fi/C0150302.jpg)   - **Tienpinta**:
![View](https://weathercam.digitraffic.fi/C0150309.jpg)  ### Station C01504 - vt2_Karkkila_Korpi -
**Coordinates**: [24.235601, 60.536727] - **Camera Type**: HIKVISION - **Collection Status**:
GATHERING - **Data Updated Time**: 2024-12-15T15:27:10Z - **Municipality**: Karkkila - **Province**:
Uusimaa - **Purpose**: Weather Observation - **Presets**:   - **Poriin**:
![View](https://weathercam.digitraffic.fi/C0150401.jpg)   - **Helsinkiin**:
![View](https://weathercam.digitraffic.fi/C0150402.jpg)   - **Tienpinta**:
![View](https://weathercam.digitraffic.fi/C0150409.jpg)  Feel free to click on the image links to
view the camera feed.

messages.append(form_msg("user", "Remove the 'road_digitraffic' service."))
complete(messages, tools=tools)
print_msgs(messages)

>> System:
You are a helpful system administrator. Use the supplied tools to assist the user.
>> User:
Add a new API service called 'road_digitraffic'. Use the base URL 'https://tie.digitraffic.fi', and
the OpenAPI Specification URL 'https://tie.digitraffic.fi/swagger/openapi.json'.
>> Assistant:
The API service 'road_digitraffic' has been successfully added.
>> User:
Get the weather camera information for the stations with ID C01503 and C01504.
>> Assistant:
Here is the weather camera information for the stations with IDs C01503 and C01504:  ### Station
C01503 - kt51_Inkoo - **Coordinates**: [23.99616, 60.05374] - **Camera Type**: BOSCH - **Collection
Status**: GATHERING - **State**: FAULT_DOUBT - **Data Updated Time**: 2024-12-15T15:25:43Z -
**Municipality**: Inkoo - **Province**: Uusimaa - **Purpose**: Weather Observation - **Presets**:
- **Inkooseen**: ![View](https://weathercam.digitraffic.fi/C0150301.jpg)   - **Hankoon**:
![View](https://weathercam.digitraffic.fi/C0150302.jpg)   - **Tienpinta**:
![View](https://weathercam.digitraffic.fi/C0150309.jpg)  ### Station C01504 - vt2_Karkkila_Korpi -
**Coordinates**: [24.235601, 60.536727] - **Camera Type**: HIKVISION - **Collection Status**:
GATHERING - **Data Updated Time**: 2024-12-15T15:27:10Z - **Municipality**: Karkkila - **Province**:
Uusimaa - **Purpose**: Weather Observation - **Presets**:   - **Poriin**:
![View](https://weathercam.digitraffic.fi/C0150401.jpg)   - **Helsinkiin**:
![View](https://weathercam.digitraffic.fi/C0150402.jpg)   - **Tienpinta**:
![View](https://weathercam.digitraffic.fi/C0150409.jpg)  Feel free to click on the image links to
view the camera feed.
>> User:
Remove the 'road_digitraffic' service.
>> Assistant:
The 'road_digitraffic' service has been successfully removed.

# After removing the 'road_digitraffic' service, the tools should only contain the initial handlers
len(tools)