feat: add OpenAI assistant api integration (#63)

* feat: add openai assistant api integration

* add EventHandler

* change api

* fix hitl

Author: wenfengwang

examples/github_notifier/main.py

@@ -1,6 +1,6 @@
 from openai import OpenAI
 from npiai.core import Agent
-from npiai.core.hitl import ConsoleHITLHandler
+from npiai.tools.hitl.console import ConsoleHandler
 from npiai.app.github import GitHub
 from npiai.app.google import Gmail
 
@@ -11,7 +11,7 @@ def main():
         prompt='You are a Github Notifier that informs users when a new issue or pull request is open.',
         description='Github Notifier that informs users when a new issue or pull request is open',
         llm=OpenAI(),
-        hitl_handler=ConsoleHITLHandler()
+        hitl_handler=ConsoleHandler()
     )
 
     agent.use(GitHub(), Gmail())

examples/openai/assistant.py

@@ -0,0 +1,43 @@
+
+from openai import OpenAI
+from npiai.app import Gmail, GitHub
+from npiai.tools.hitl.console import ConsoleHandler
+from npiai.core.toolset import ToolSet
+from npiai.integration.oai import EventHandler
+
+
+if __name__ == "__main__":
+    client = OpenAI(api_key="xxxxx")
+    ts = ToolSet(
+        llm=client,
+        hitl_handler=ConsoleHandler(),
+        tools=[
+            GitHub(access_token="xxxxx"),
+        ],
+    )
+
+    assistant = client.beta.assistants.create(
+        name="GitHub Issue Assistant",
+        instructions="You are an Assistant can maintain issue comment for repo npi-ai/npi.",
+        model="gpt-4o",
+        tools=ts.openai(),
+    )
+
+    thread = client.beta.threads.create()
+    message = client.beta.threads.messages.create(
+        thread_id=thread.id,
+        role="user",
+        content="what's title of issue #27 of repo npi-ai/npi?",
+    )
+
+    def stream_handler(run_id: str, stream):
+        for text in stream.text_deltas:
+            print(text, end="", flush=True)
+
+    eh = EventHandler(toolset=ts, llm=client, thread_id=thread.id, stream_handler=stream_handler)
+    with client.beta.threads.runs.stream(
+            thread_id=thread.id,
+            assistant_id=assistant.id,
+            event_handler=eh,
+    ) as _stream:
+        _stream.until_done()

sdk/python/npiai/app/github.py

@@ -1,12 +1,18 @@
 from npiai.core.base import App
 from npiai_proto import api_pb2
+from typing_extensions import override
 
 
 class GitHub(App):
 
-    def __init__(self, npi_endpoint: str = None):
+    def __init__(self, access_token: str, npi_endpoint: str = None):
         super().__init__(
             app_name="github",
             app_type=api_pb2.GITHUB,
             endpoint=npi_endpoint,
         )
+        self.access_token = access_token
+
+    @override
+    def authorize(self):
+        super()._authorize(credentials={"access_token": self.access_token})

sdk/python/npiai/core/base.py

@@ -33,14 +33,17 @@ def __init__(
             endpoint: str = "localhost:9140",
             hitl_handler: hitl.HITLHandler = None,
             npi_token: str = None,
+            insecure: bool = True,
     ):
         self.__app_name = app_name
         if endpoint is None:
             endpoint = "localhost:9140"
         self.__npi_endpoint = endpoint
         self.__app_type = app_type
-
-        channel = grpc.secure_channel(target=self.__npi_endpoint, credentials=grpc.ssl_channel_credentials())
+        if insecure:
+            channel = grpc.insecure_channel(self.__npi_endpoint)
+        else:
+            channel = grpc.secure_channel(target=self.__npi_endpoint, credentials=grpc.ssl_channel_credentials())
         self.stub = api_pb2_grpc.AppServerStub(channel)
         self.hitl_handler = hitl_handler
         self.__npi_token = npi_token
@@ -104,15 +107,24 @@ def chat(self, msg: str) -> str:
                 case api_pb2.ResponseCode.ACTION_REQUIRED:
                     resp = self.stub.Chat(request=self.__call_human(resp), metadata=self.__get_metadata())
                 case _:
+                    logger.error(f'[{self.__app_name}]: Error: failed to call function, unknown response code {resp.code}')
                     raise Exception("Error: failed to call function")
 
     def hitl(self, handler: hitl.HITLHandler):
         self.hitl_handler = handler
 
-    # @abstractmethod
-    def authorize(self, **kwargs):
+    def authorize(self):
         pass
 
+    def _authorize(self, credentials: dict[str, str]):
+        self.stub.Authorize(
+            request=api_pb2.AuthorizeRequest(
+                type=self.__app_type,
+                credentials=credentials,
+            ),
+        )
+        logger.info(f'[{self.__app_name}]: Authorized')
+
     def __call_human(self, resp: api_pb2.Response) -> api_pb2.Request:
         human_resp = self.hitl_handler.handle(
             hitl.convert_to_hitl_request(

sdk/python/npiai/core/toolset.py

@@ -0,0 +1,39 @@
+from npiai.core.base import App
+from npiai.core.hitl import HITLHandler
+from openai import Client
+
+from npiai.tools.hitl import EmptyHandler
+import json
+from typing import List
+
+
+class ToolSet:
+
+    def __init__(self, tools: List[App], llm: Client, hitl_handler: HITLHandler = None, ):
+        self._llm = llm
+        self._hitl_handler = hitl_handler
+        if self._hitl_handler is None:
+            self._hitl_handler = EmptyHandler()
+
+        if len(tools) == 0:
+            raise Exception("At least one tool is required.")
+
+        self._tools = {}
+        for tool in tools:
+            tool.authorize()
+            tool.hitl(self._hitl_handler)
+            self._tools[tool.tool_name()] = tool
+
+    def openai(self):
+        tools = []
+        for tool in self._tools.values():
+            tools.append(tool.schema())
+        return tools
+
+    def call(self, body) -> str:
+        tool = self._tools.get(body.function.name)
+        if tool is None:
+            return "Tool not found"
+        params = json.loads(body.function.arguments)
+
+        return tool.chat(params['message'])

sdk/python/npiai/integration/oai.py

@@ -0,0 +1,43 @@
+from typing_extensions import override
+
+from openai import AssistantEventHandler, Client
+from npiai.core.toolset import ToolSet
+
+
+class EventHandler(AssistantEventHandler):
+    def __init__(self, toolset: ToolSet, llm: Client, thread_id: str, stream_handler=None):
+        super().__init__()
+        self.toolset = toolset
+        self.llm = llm
+        self.thread_id = thread_id
+        self.stream_handler = stream_handler
+
+    @override
+    def on_event(self, event):
+        # Retrieve events that are denoted with 'requires_action'
+        # since these will have our tool_calls
+        if event.event == 'thread.run.requires_action':
+            run_id = event.data.id  # Retrieve the run ID from the event data
+            self.handle_requires_action(event.data, run_id)
+
+    def handle_requires_action(self, data, run_id):
+        tool_outputs = []
+
+        for tool in data.required_action.submit_tool_outputs.tool_calls:
+            tool_outputs.append({
+                "tool_call_id": tool.id,
+                "output": self.toolset.call(tool),
+            })
+
+        # Submit all tool_outputs at the same time
+        self.submit_tool_outputs(tool_outputs, run_id)
+
+    def submit_tool_outputs(self, tool_outputs, run_id):
+        # Use the submit_tool_outputs_stream helper
+        with self.llm.beta.threads.runs.submit_tool_outputs_stream(
+                thread_id=self.thread_id,
+                run_id=run_id,
+                tool_outputs=tool_outputs,
+                event_handler=EventHandler(self.toolset, self.llm, self.thread_id),
+        ) as stream:
+            self.stream_handler(run_id, stream)

sdk/python/npiai/tools/hitl/__init__.py

@@ -0,0 +1,5 @@
+from .console import ConsoleHandler
+from .empty import EmptyHandler
+from .twilio import TwilioHandler
+
+__all__ = ['ConsoleHandler', 'EmptyHandler', 'TwilioHandler']

sdk/python/npiai/tools/hitl/empty.py

@@ -0,0 +1,23 @@
+from npiai.core.hitl import (
+    HITLHandler, HITLRequest, HITLResponse, ActionRequestCode, ActionResultCode, ACTION_APPROVED, ACTION_DENIED)
+
+from npiai_proto import api_pb2
+
+
+class EmptyHandler(HITLHandler):
+    def handle(self, req: HITLRequest) -> HITLResponse:
+        match req.code:
+            case ActionRequestCode.INFORMATION:
+                return HITLResponse(
+                    code=ActionResultCode.SUCCESS,
+                    msg="No human can response your help",
+                )
+            case ActionRequestCode.CONFIRMATION:
+                return HITLResponse(
+                    code=ActionResultCode.APPROVED,
+                    msg="automatically approved",
+                )
+        return ACTION_DENIED
+
+    def type(self) -> api_pb2.ActionType:
+        return api_pb2.ActionType.CONSOLE