Added python sample for assistant skills

samples/assistant/python/assistant_skills.py

@@ -0,0 +1,31 @@
+import json
+import logging
+import uuid
+import azure.functions as func
+
+from todo_manager import CreateTodoManager, TodoItem
+
+bp = func.Blueprint()
+
+todo_manager = CreateTodoManager()
+
+
+@bp.function_name("AddTodo")
+@bp.generic_trigger(arg_name="taskDescription", type="assistantSkillTrigger", data_type=func.DataType.STRING, functionDescription="Create a new todo task")
+def add_todo(taskDescription: str) -> None:
+    if not taskDescription:
+        raise ValueError("Task description cannot be empty")
+
+    logging.info(f"Adding todo: {taskDescription}")
+
+    todo_id = str(uuid.uuid4())[0:6]
+    todo_manager.add_todo(TodoItem(id=todo_id, task=taskDescription))
+    return
+
+
+@bp.function_name("GetTodos")
+@bp.generic_trigger(arg_name="inputIgnored", type="assistantSkillTrigger", data_type=func.DataType.STRING, functionDescription="Fetch the list of previously created todo tasks")
+def get_todos(inputIgnored: str) -> str:
+    logging.info("Fetching list of todos")
+    results = todo_manager.get_todos()
+    return json.dumps(results)

samples/assistant/python/demo.http

@@ -0,0 +1,28 @@
+### Create a new assistant - instructions are hardcoded in the function
+PUT http://localhost:7071/api/assistants/assistant123
+
+
+### Reminder #1
+POST http://localhost:7071/api/assistants/assistant123
+Content-Type: text/plain
+
+Remind me to call my dad
+
+
+### Reminder #2
+POST http://localhost:7071/api/assistants/assistant123
+Content-Type: text/plain
+
+Oh, and to take out the trash
+
+
+### Get the list of tasks
+POST http://localhost:7071/api/assistants/assistant123
+Content-Type: text/plain
+
+What do I need to do today?
+
+
+### Query the chat history
+GET http://localhost:7071/api/assistants/assistant123?timestampUTC=2023-01-01T00:00:00Z
+Accept: application/json

samples/assistant/python/extensions.csproj

@@ -0,0 +1,17 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFramework>net60</TargetFramework>
+	  <WarningsAsErrors></WarningsAsErrors>
+	  <DefaultItemExcludes>**</DefaultItemExcludes>
+    <OutDir>bin</OutDir>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Azure.WebJobs.Script.ExtensionsMetadataGenerator" Version="4.0.*" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <!-- Locally build and reference the extension -->
+    <ProjectReference Include="../../../src/WebJobs.Extensions.OpenAI/WebJobs.Extensions.OpenAI.csproj" />
+  </ItemGroup>
+</Project>

samples/assistant/python/function_app.py

@@ -0,0 +1,45 @@
+import json
+import azure.functions as func
+
+from assistant_skills import bp
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+app.register_functions(bp)
+
+
+@app.function_name("CreateAssistant")
+@app.route(route="assistants/{assistantId}", methods=["PUT"])
+@app.generic_output_binding(arg_name="requests", type="chatBotCreate", data_type=func.DataType.STRING)
+def create_assistant(req: func.HttpRequest, requests: func.Out[str]) -> func.HttpResponse:
+    assistantId = req.route_params.get("assistantId")
+    instructions = """
+            Don't make assumptions about what values to plug into functions.
+            Ask for clarification if a user request is ambiguous.
+            """
+    create_request = {
+        "id": assistantId,
+        "instructions": instructions
+    }
+    requests.set(json.dumps(create_request))
+    response_json = {"assistantId": assistantId}
+    return func.HttpResponse(json.dumps(response_json), status_code=202, mimetype="application/json")
+
+
+@app.function_name("PostUserQuery")
+@app.route(route="assistants/{assistantId}", methods=["POST"])
+@app.generic_output_binding(arg_name="requests", type="chatBotPost", data_type=func.DataType.STRING, id="{assistantId}", model="gpt-4")
+def post_user_query(req: func.HttpRequest, requests: func.Out[str]) -> func.HttpResponse:
+    userMessage = req.get_body().decode("utf-8")
+    if not userMessage:
+        return func.HttpResponse(json.dumps({"message": "Request body is empty"}), status_code=400, mimetype="application/json")
+
+    requests.set(json.dumps({"userMessage": userMessage}))
+    return func.HttpResponse(status_code=202)
+
+
+@app.function_name("GetChatState")
+@app.route(route="assistants/{assistantId}", methods=["GET"])
+@app.generic_input_binding(arg_name="state", type="chatBotQuery", data_type=func.DataType.STRING, id="{assistantId}", timestampUtc="{Query.timestampUTC}")
+def get_chat_state(req: func.HttpRequest, state: str) -> func.HttpResponse:
+    return func.HttpResponse(state, status_code=200, mimetype="application/json")

samples/assistant/python/host.json

@@ -0,0 +1,8 @@
+{
+  "version": "2.0",
+  "logging": {
+    "logLevel": {
+      "WebJobs.Extensions.OpenAI": "Information"
+    }
+  }
+}

samples/assistant/python/local.settings.json

@@ -0,0 +1,8 @@
+{
+  "IsEncrypted": false,
+  "Values": {
+    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
+    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
+    "FUNCTIONS_WORKER_RUNTIME": "python"
+  }
+}

samples/assistant/python/requirements.txt

@@ -0,0 +1,6 @@
+# DO NOT include azure-functions-worker in this file
+# The Python Worker is managed by Azure Functions platform
+# Manually managing azure-functions-worker may cause unexpected issues
+
+azure-functions
+azure-cosmos

samples/assistant/python/todo_manager.py

@@ -0,0 +1,62 @@
+import abc
+import logging
+import os
+
+from azure.cosmos import CosmosClient
+
+
+class TodoItem:
+    def __init__(self, id, task):
+        self.id = id
+        self.task = task
+
+
+class ITodoManager(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def add_todo(self, todo: TodoItem):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def get_todos(self):
+        raise NotImplementedError()
+
+
+class InMemoryTodoManager(ITodoManager):
+    def __init__(self):
+        self.todos = []
+
+    def add_todo(self, todo: TodoItem):
+        self.todos.append(todo)
+
+    def get_todos(self):
+        return [item.__dict__ for item in self.todos]
+
+
+class CosmosDbTodoManager(ITodoManager):
+    def __init__(self, cosmos_client: CosmosClient):
+        self.cosmos_client = cosmos_client
+        self.database = self.cosmos_client.get_database_client("testdb")
+        self.container = self.database.get_container_client("my-todos")
+
+    def add_todo(self, todo: TodoItem):
+        logging.info(
+            f"Adding todo ID = {todo.id} to container '{self.container.id}'.")
+        self.container.create_item(todo.__dict__)
+
+    def get_todos(self):
+        logging.info(
+            f"Getting all todos from container '{self.container.id}'.")
+        results = [item for item in self.container.query_items(
+            "SELECT * FROM c", enable_cross_partition_query=True)]
+        logging.info(
+            f"Found {len(results)} todos in container '{self.container.id}'.")
+        return results
+
+
+def CreateTodoManager() -> ITodoManager:
+    if not os.environ.get("CosmosDbConnectionString"):
+        return InMemoryTodoManager()
+    else:
+        cosmos_client = CosmosClient.from_connection_string(
+            os.environ["CosmosDbConnectionString"])
+        return CosmosDbTodoManager(cosmos_client)